From 88c006ee37de6ce9314ca9e38e2ac3f55f207941 Mon Sep 17 00:00:00 2001 From: jearhe <0verme7753@gmail.com> Date: Sun, 4 Oct 2026 10:37:44 +0800 Subject: [PATCH] fix(plugins): terminate Windows sidecar process trees --- Cargo.lock | 1 + agents/drivers/duckdb/Cargo.lock | 1 + .../dbx-core/examples/plugin_package_smoke.rs | 2 +- crates/dbx-core/src/ai/agent_explain.rs | 2 +- .../dbx-core/src/connection/h2_jdbc_tests.rs | 32 +- crates/dbx-core/src/connection/mod.rs | 18 +- crates/dbx-core/src/query/mod.rs | 6 +- crates/dbx-platform/Cargo.toml | 11 +- crates/dbx-platform/src/process.rs | 465 ++++++++++++++++++ crates/dbx-plugin-runtime/src/plugins.rs | 22 +- crates/dbx-plugin-runtime/src/plugins/host.rs | 21 +- .../src/plugins/jdbc_sessions.rs | 24 +- .../dbx-plugin-runtime/src/plugins/runtime.rs | 135 +++-- crates/dbx-web/src/routes/plugins.rs | 8 +- src-tauri/src/commands/plugins.rs | 14 +- 15 files changed, 656 insertions(+), 106 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index efb999949b..13e22a95c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2561,6 +2561,7 @@ dependencies = [ "serde", "tokio", "uuid", + "windows-sys 0.61.2", ] [[package]] diff --git a/agents/drivers/duckdb/Cargo.lock b/agents/drivers/duckdb/Cargo.lock index 5d1bd2c2d7..614d19a8be 100644 --- a/agents/drivers/duckdb/Cargo.lock +++ b/agents/drivers/duckdb/Cargo.lock @@ -1577,6 +1577,7 @@ dependencies = [ "serde", "tokio", "uuid", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/dbx-core/examples/plugin_package_smoke.rs b/crates/dbx-core/examples/plugin_package_smoke.rs index 17cf18c916..340d3c5f96 100644 --- a/crates/dbx-core/examples/plugin_package_smoke.rs +++ b/crates/dbx-core/examples/plugin_package_smoke.rs @@ -149,7 +149,7 @@ async fn run() -> Result<(), String> { } handle.disconnect().await?; - host.stop_all().await; + host.stop_all().await?; installer.uninstall(PLUGIN_ID)?; if !PluginRegistry::new_with_app_version(store.path().to_path_buf(), APP_VERSION).list_installed()?.is_empty() { return Err("Plugin store is not empty after uninstall".to_string()); diff --git a/crates/dbx-core/src/ai/agent_explain.rs b/crates/dbx-core/src/ai/agent_explain.rs index 5386655a76..ea287c2e61 100644 --- a/crates/dbx-core/src/ai/agent_explain.rs +++ b/crates/dbx-core/src/ai/agent_explain.rs @@ -305,7 +305,7 @@ mod tests { .unwrap(); assert_eq!(plan, "TABLE ACCESS FULL DUAL"); - session.shutdown().await; + session.shutdown().await.unwrap(); drop(state); std::fs::remove_dir_all(dir).unwrap(); } diff --git a/crates/dbx-core/src/connection/h2_jdbc_tests.rs b/crates/dbx-core/src/connection/h2_jdbc_tests.rs index 5dad5aa84e..8f256c5477 100644 --- a/crates/dbx-core/src/connection/h2_jdbc_tests.rs +++ b/crates/dbx-core/src/connection/h2_jdbc_tests.rs @@ -163,7 +163,7 @@ async fn embedded_h2_jdbc_failed_connect_cancel_drop_and_last_close_are_isolated panic!() }; drop(dropped); - good.shutdown().await; + good.shutdown().await.unwrap(); tokio::time::timeout(Duration::from_secs(5), async { while good.pid().await.is_some() { tokio::task::yield_now().await; @@ -176,7 +176,7 @@ async fn embedded_h2_jdbc_failed_connect_cancel_drop_and_last_close_are_isolated panic!() }; assert_ne!(reopened.pid().await, pid); - reopened.shutdown().await; + reopened.shutdown().await.unwrap(); state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); } @@ -201,8 +201,8 @@ async fn embedded_h2_jdbc_keeps_remote_memory_and_unrelated_drivers_dedicated() }; assert_ne!(first.pid().await, second.pid().await); assert_eq!(identity(&first).await["session"], "legacy"); - first.shutdown().await; - second.shutdown().await; + first.shutdown().await.unwrap(); + second.shutdown().await.unwrap(); } state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); @@ -268,9 +268,9 @@ async fn embedded_h2_jdbc_rejects_old_plugins_and_preserves_selected_driver_runt panic!() }; assert_ne!(first.pid().await, different.pid().await); - different.shutdown().await; - first.shutdown().await; - second.shutdown().await; + different.shutdown().await.unwrap(); + first.shutdown().await.unwrap(); + second.shutdown().await.unwrap(); state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); } @@ -293,7 +293,7 @@ async fn embedded_h2_jdbc_last_owner_close_can_race_a_new_session() { let PoolKind::ExternalDriver { session: replacement, .. } = replacement.unwrap() else { panic!() }; assert!(replacement.is_available(), "cycle {cycle_index}"); identity(&replacement).await; - replacement.shutdown().await; + replacement.shutdown().await.unwrap(); assert!(replacement.pid().await.is_none()); } state.shutdown(Duration::from_secs(5)).await; @@ -332,7 +332,7 @@ async fn embedded_h2_jdbc_cancelled_open_releases_its_member_and_server_state() }) .await .unwrap(); - survivor.shutdown().await; + survivor.shutdown().await.unwrap(); assert!(survivor.pid().await.is_none()); state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); @@ -366,9 +366,11 @@ async fn embedded_h2_jdbc_replaces_dead_runtime_with_sibling_handles_still_alive }; assert_ne!(replacement.pid().await, original_pid); assert!(replacement.pid().await.is_some()); - tokio::join!(first.shutdown(), sibling.shutdown()); + let (first_closed, sibling_closed) = tokio::join!(first.shutdown(), sibling.shutdown()); + first_closed.unwrap(); + sibling_closed.unwrap(); identity(&replacement).await; - replacement.shutdown().await; + replacement.shutdown().await.unwrap(); assert!(replacement.pid().await.is_none()); state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); @@ -392,7 +394,7 @@ async fn embedded_h2_jdbc_timeout_and_interrupted_shutdown_finish_session_cleanu .unwrap_err(); assert!(error.contains("timed out")); assert!(!timed.is_available()); - timed.shutdown().await; + timed.shutdown().await.unwrap(); identity(&survivor).await; let mut delayed = config.clone(); delayed.username = "delay-close".into(); @@ -402,7 +404,7 @@ async fn embedded_h2_jdbc_timeout_and_interrupted_shutdown_finish_session_cleanu }; let close_session = closing.clone(); let close = tokio::spawn(async move { - close_session.shutdown().await; + close_session.shutdown().await.unwrap(); }); tokio::time::timeout(Duration::from_secs(5), async { while closing.is_available() { @@ -414,7 +416,7 @@ async fn embedded_h2_jdbc_timeout_and_interrupted_shutdown_finish_session_cleanu close.abort(); assert!(close.await.unwrap_err().is_cancelled()); identity(&survivor).await; - survivor.shutdown().await; + survivor.shutdown().await.unwrap(); tokio::time::timeout(Duration::from_secs(5), async { while survivor.pid().await.is_some() { tokio::task::yield_now().await; @@ -422,7 +424,7 @@ async fn embedded_h2_jdbc_timeout_and_interrupted_shutdown_finish_session_cleanu }) .await .unwrap(); - closing.shutdown().await; + closing.shutdown().await.unwrap(); state.shutdown(Duration::from_secs(5)).await; std::fs::remove_dir_all(directory).unwrap(); } diff --git a/crates/dbx-core/src/connection/mod.rs b/crates/dbx-core/src/connection/mod.rs index 91207b9bd0..ebbc613263 100644 --- a/crates/dbx-core/src/connection/mod.rs +++ b/crates/dbx-core/src/connection/mod.rs @@ -1871,7 +1871,9 @@ impl AppState { Some(external_driver_connect_timeout(config)), ) .await; - session.shutdown().await; + if let Err(error) = session.shutdown().await { + log::warn!("Failed to stop the plugin runtime after testing connection '{}': {error}", config.name); + } return result.map(|response| { ConnectionTestResult::success("Connection successful") .with_database_info(database_info_from_protocol_value(&response)) @@ -1916,7 +1918,12 @@ impl AppState { session, }), Err(error) => { - session.shutdown().await; + if let Err(shutdown_error) = session.shutdown().await { + log::warn!( + "Failed to stop the plugin runtime after failing to connect '{}': {shutdown_error}", + config.name + ); + } Err(error) } } @@ -2429,7 +2436,7 @@ impl AppState { let shutdown = async { let routing = self.pool_routing_control(); - tokio::join!( + let (_, _, _, _, _, _, plugin_shutdown) = tokio::join!( self.task_supervisor.shutdown(deadline), routing.close_removed(removed_pools), self.tunnels.stop_all_tunnels(), @@ -2438,6 +2445,9 @@ impl AppState { self.agent_manager.stop_daemons(), self.plugin_host.stop_all(), ); + if let Err(error) = plugin_shutdown { + log::warn!("Failed to stop plugin runtimes during shutdown: {error}"); + } }; if tokio::time::timeout(deadline, shutdown).await.is_err() { log::warn!("Timed out shutting down DBX runtime resources after {}ms", deadline.as_millis()); @@ -6847,7 +6857,7 @@ async fn close_pool_kind(pool: PoolKind) -> Result<(), String> { client.disconnect().await?; } PoolKind::ExternalDriver { session, .. } => { - session.shutdown().await; + session.shutdown().await?; } PoolKind::PluginConnection(handle) => { if let Err(error) = handle.disconnect().await { diff --git a/crates/dbx-core/src/query/mod.rs b/crates/dbx-core/src/query/mod.rs index e2d5cd8a3e..59b39ec288 100644 --- a/crates/dbx-core/src/query/mod.rs +++ b/crates/dbx-core/src/query/mod.rs @@ -10194,7 +10194,7 @@ for line in sys.stdin: assert_eq!(cells.len(), 1); assert_eq!(std::fs::read_to_string(&calls).unwrap().lines().count(), 4); - session.shutdown().await; + session.shutdown().await.unwrap(); let _ = std::fs::remove_dir_all(dir); } @@ -10379,7 +10379,7 @@ for line in sys.stdin: assert_eq!(result.rows, vec![vec![serde_json::json!(42)]]); assert_eq!(std::fs::read_to_string(&calls).unwrap(), "executeQueryPage\nexecuteQuery\n"); - session.shutdown().await; + session.shutdown().await.unwrap(); let _ = std::fs::remove_dir_all(dir); } @@ -10443,7 +10443,7 @@ for line in sys.stdin: assert_eq!(error, "Incorrect syntax near SELECT"); assert_eq!(std::fs::read_to_string(&calls).unwrap(), "request\n"); - session.shutdown().await; + session.shutdown().await.unwrap(); let _ = std::fs::remove_dir_all(dir); } diff --git a/crates/dbx-platform/Cargo.toml b/crates/dbx-platform/Cargo.toml index 40d8d37b05..b732191807 100644 --- a/crates/dbx-platform/Cargo.toml +++ b/crates/dbx-platform/Cargo.toml @@ -19,7 +19,16 @@ serde = { version = "1.0", features = ["derive"], optional = true } uuid = { version = "1", features = ["v4"], optional = true } [dev-dependencies] -tokio = { version = "1", features = ["rt"] } +tokio = { version = "1", features = ["rt", "macros"] } [target.'cfg(all(unix, not(target_os = "redox")))'.dependencies] nix = { version = "0.31.3", features = ["user"] } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_System_Diagnostics_ToolHelp", + "Win32_System_JobObjects", + "Win32_System_Threading", +] } diff --git a/crates/dbx-platform/src/process.rs b/crates/dbx-platform/src/process.rs index 6df6cf7df5..2c05dd287f 100644 --- a/crates/dbx-platform/src/process.rs +++ b/crates/dbx-platform/src/process.rs @@ -1,7 +1,353 @@ use std::ffi::OsStr; +use std::io; +use std::process::ExitStatus; + +use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command}; #[cfg(windows)] const CREATE_NO_WINDOW: u32 = 0x08000000; +#[cfg(windows)] +const CREATE_SUSPENDED: u32 = 0x00000004; + +/// A Tokio child together with the process-tree ownership required to stop it safely. +/// +/// Windows children are created suspended, assigned to a private Job Object, and only then +/// resumed. This ensures descendants inherit the Job before any plugin code can run. Other +/// platforms intentionally keep Tokio's existing direct-child behavior. +pub struct ManagedChild { + child: Child, + #[cfg(windows)] + job: windows::ProcessTreeJob, +} + +impl ManagedChild { + pub fn id(&self) -> Option { + self.child.id() + } + + pub fn take_stdin(&mut self) -> Option { + self.child.stdin.take() + } + + pub fn take_stdout(&mut self) -> Option { + self.child.stdout.take() + } + + pub fn take_stderr(&mut self) -> Option { + self.child.stderr.take() + } + + pub fn try_wait(&mut self) -> io::Result> { + self.child.try_wait() + } + + /// Force the owned process tree to exit and wait until the termination is observable. + /// + /// On Windows this terminates the Job Object and confirms it has no active processes before + /// reaping the launcher. On Unix and other platforms it preserves Tokio's direct-child kill + /// semantics. + pub async fn terminate(&mut self) -> io::Result<()> { + #[cfg(windows)] + { + self.job.terminate_and_wait().await?; + self.child.wait().await?; + Ok(()) + } + + #[cfg(not(windows))] + { + if self.child.try_wait()?.is_none() { + self.child.kill().await?; + } + Ok(()) + } + } +} + +/// Spawn a process with explicit ownership of its lifecycle. +/// +/// On Windows, failure to create or assign the private Job Object is returned as a spawn error; +/// the process is never resumed without ownership. Dropping the returned child closes its Job +/// Object, whose `KILL_ON_JOB_CLOSE` limit terminates any remaining members. +pub async fn spawn_managed_child(command: &mut Command) -> io::Result { + command.kill_on_drop(true); + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + + command.as_std_mut().creation_flags(CREATE_NO_WINDOW | CREATE_SUSPENDED); + let mut job = windows::ProcessTreeJob::new()?; + let mut child = command.spawn()?; + let Some(pid) = child.id() else { + let cleanup = kill_child_and_wait(&mut child).await; + return Err(spawn_cleanup_error( + io::Error::other("Windows returned a child process without a PID"), + cleanup.err(), + )); + }; + + if let Err(error) = job.assign(&child) { + let cleanup = kill_child_and_wait(&mut child).await; + return Err(spawn_cleanup_error(error, cleanup.err())); + } + if let Err(error) = job.resume_child(pid) { + if let Err(termination_error) = job.terminate_and_wait().await { + // terminate_and_wait closes the job on failure, triggering its kill-on-close limit. + let cleanup_error = match wait_for_child_exit(&mut child).await { + Ok(()) => termination_error, + Err(wait_error) => io::Error::new( + termination_error.kind(), + format!("{termination_error}; additionally failed to wait for launcher: {wait_error}"), + ), + }; + return Err(spawn_cleanup_error(error, Some(cleanup_error))); + } + let cleanup = wait_for_child_exit(&mut child).await; + return Err(spawn_cleanup_error(error, cleanup.err())); + } + + Ok(ManagedChild { child, job }) + } + + #[cfg(not(windows))] + { + command.spawn().map(|child| ManagedChild { child }) + } +} + +#[cfg(windows)] +async fn kill_child_and_wait(child: &mut Child) -> io::Result<()> { + if child.try_wait()?.is_none() { + child.kill().await?; + } + Ok(()) +} + +#[cfg(windows)] +async fn wait_for_child_exit(child: &mut Child) -> io::Result<()> { + child.wait().await.map(|_| ()) +} + +#[cfg(windows)] +fn spawn_cleanup_error(error: io::Error, cleanup_error: Option) -> io::Error { + match cleanup_error { + Some(cleanup_error) => io::Error::new( + error.kind(), + format!("{error}; additionally failed to confirm child cleanup: {cleanup_error}"), + ), + None => error, + } +} + +#[cfg(windows)] +mod windows { + use std::io; + use std::mem::size_of; + use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle, RawHandle}; + use std::time::Duration; + + use tokio::process::Child; + use tokio::time::{sleep, Instant}; + use windows_sys::Win32::Foundation::{HANDLE, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, Thread32First, Thread32Next, TH32CS_SNAPTHREAD, THREADENTRY32, + }; + use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JobObjectBasicAccountingInformation, + JobObjectExtendedLimitInformation, QueryInformationJobObject, SetInformationJobObject, TerminateJobObject, + JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + }; + use windows_sys::Win32::System::Threading::{OpenThread, ResumeThread, THREAD_SUSPEND_RESUME}; + + const PROCESS_TREE_TERMINATION_TIMEOUT: Duration = Duration::from_secs(10); + const PROCESS_TREE_POLL_INTERVAL: Duration = Duration::from_millis(10); + + pub(super) struct ProcessTreeJob { + handle: Option, + terminated: bool, + } + + impl ProcessTreeJob { + pub(super) fn new() -> io::Result { + let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) }; + if handle.is_null() { + let error = io::Error::last_os_error(); + return Err(io::Error::new(error.kind(), format!("CreateJobObjectW failed: {error}"))); + } + let handle = unsafe { OwnedHandle::from_raw_handle(handle as RawHandle) }; + let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default(); + limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + let configured = unsafe { + SetInformationJobObject( + handle.as_raw_handle() as HANDLE, + JobObjectExtendedLimitInformation, + (&limits as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(), + size_of::() as u32, + ) + }; + if configured == 0 { + let error = io::Error::last_os_error(); + return Err(io::Error::new( + error.kind(), + format!("failed to enable kill-on-close for plugin Job Object: {error}"), + )); + } + Ok(Self { handle: Some(handle), terminated: false }) + } + + pub(super) fn assign(&self, child: &Child) -> io::Result<()> { + let job = self.handle()?; + let process = child.raw_handle().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "suspended plugin child exited before Job assignment") + })?; + let assigned = unsafe { AssignProcessToJobObject(job, process as HANDLE) }; + if assigned == 0 { + let error = io::Error::last_os_error(); + return Err(io::Error::new( + error.kind(), + format!( + "failed to assign suspended plugin process to its Windows Job Object (the host job may not allow nesting): {error}" + ), + )); + } + Ok(()) + } + + pub(super) fn resume_child(&self, pid: u32) -> io::Result<()> { + let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; + if snapshot == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + let snapshot = unsafe { OwnedHandle::from_raw_handle(snapshot as RawHandle) }; + let mut entry = THREADENTRY32 { dwSize: size_of::() as u32, ..Default::default() }; + let mut found_thread = None; + let mut has_entry = unsafe { Thread32First(snapshot.as_raw_handle() as HANDLE, &mut entry) != 0 }; + while has_entry { + if entry.th32OwnerProcessID == pid && found_thread.replace(entry.th32ThreadID).is_some() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("suspended plugin process {pid} unexpectedly has multiple threads"), + )); + } + has_entry = unsafe { Thread32Next(snapshot.as_raw_handle() as HANDLE, &mut entry) != 0 }; + } + let thread_id = found_thread.ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("primary thread for suspended process {pid} was not found"), + ) + })?; + let thread = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, thread_id) }; + if thread.is_null() { + return Err(io::Error::last_os_error()); + } + let thread = unsafe { OwnedHandle::from_raw_handle(thread as RawHandle) }; + let previous_suspend_count = unsafe { ResumeThread(thread.as_raw_handle() as HANDLE) }; + if previous_suspend_count == u32::MAX { + return Err(io::Error::last_os_error()); + } + if previous_suspend_count != 1 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "suspended plugin process {pid} had unexpected primary-thread suspend count {previous_suspend_count}" + ), + )); + } + Ok(()) + } + + pub(super) async fn terminate_and_wait(&mut self) -> io::Result<()> { + if self.terminated { + return Ok(()); + } + + let active_processes = match self.active_processes() { + Ok(active_processes) => active_processes, + Err(error) => { + self.close(); + return Err(error); + } + }; + if active_processes > 0 { + let terminated = unsafe { TerminateJobObject(self.handle()?, 1) }; + if terminated == 0 { + let error = io::Error::last_os_error(); + match self.active_processes() { + Ok(0) => {} + Ok(_) => { + self.close(); + return Err(io::Error::new( + error.kind(), + format!("failed to terminate Windows plugin process tree: {error}"), + )); + } + Err(query_error) => { + self.close(); + return Err(io::Error::new( + query_error.kind(), + format!("failed to verify Windows plugin process-tree termination: {query_error}"), + )); + } + } + } + } + + let deadline = Instant::now() + PROCESS_TREE_TERMINATION_TIMEOUT; + loop { + match self.active_processes() { + Ok(0) => { + self.terminated = true; + self.close(); + return Ok(()); + } + Ok(_) if Instant::now() < deadline => sleep(PROCESS_TREE_POLL_INTERVAL).await, + Ok(_) => { + self.close(); + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out waiting for the Windows plugin process tree to exit", + )); + } + Err(error) => { + self.close(); + return Err(error); + } + } + } + } + + fn active_processes(&self) -> io::Result { + let mut accounting = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default(); + let queried = unsafe { + QueryInformationJobObject( + self.handle()?, + JobObjectBasicAccountingInformation, + (&mut accounting as *mut JOBOBJECT_BASIC_ACCOUNTING_INFORMATION).cast(), + size_of::() as u32, + std::ptr::null_mut(), + ) + }; + if queried == 0 { + return Err(io::Error::last_os_error()); + } + Ok(accounting.ActiveProcesses) + } + + fn handle(&self) -> io::Result { + self.handle + .as_ref() + .map(|handle| handle.as_raw_handle() as HANDLE) + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "Windows process-tree Job Object is closed")) + } + + fn close(&mut self) { + drop(self.handle.take()); + } + } +} pub fn new_std_command(program: impl AsRef) -> std::process::Command { let mut command = std::process::Command::new(program); @@ -35,3 +381,122 @@ pub fn hide_tokio_console_window(command: &mut tokio::process::Command) { command.creation_flags(CREATE_NO_WINDOW); } } + +#[cfg(all(test, windows))] +mod tests { + use std::fs; + use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle, RawHandle}; + use std::path::PathBuf; + use std::process::Stdio; + use std::thread; + use std::time::{Duration, SystemTime, UNIX_EPOCH}; + + use super::{new_tokio_command, spawn_managed_child, ManagedChild}; + use windows_sys::Win32::Foundation::{HANDLE, WAIT_FAILED, WAIT_OBJECT_0, WAIT_TIMEOUT}; + use windows_sys::Win32::System::Threading::{OpenProcess, WaitForSingleObject, PROCESS_SYNCHRONIZE}; + + struct TestDirectory(PathBuf); + + impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + #[tokio::test] + async fn windows_batch_launcher_shutdown_and_drop_terminate_descendants_across_lifecycles() { + let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); + let directory = TestDirectory( + std::env::temp_dir().join(format!("dbx-plugin-process-tree-{}-{timestamp}", std::process::id())), + ); + fs::create_dir_all(&directory.0).unwrap(); + let executable = std::env::current_exe().unwrap(); + let launcher = directory.0.join("launcher.bat"); + fs::write(&launcher, b"@echo off\r\n\"%DBX_PROCESS_TREE_TEST_EXE%\" windows_process_tree_test_descendant\r\n") + .unwrap(); + + for lifecycle in 0..3 { + let pid_file = directory.0.join(format!("descendant-{lifecycle}.pid")); + let mut command = new_tokio_command(&launcher); + command + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .env("DBX_PROCESS_TREE_TEST_EXE", &executable) + .env("DBX_PROCESS_TREE_TEST_PID_FILE", &pid_file); + let mut child = + spawn_managed_child(&mut command).await.expect("launcher should be assigned before it runs"); + let launcher_pid = child.id().expect("launcher should have a PID"); + let descendant_pid = wait_for_descendant_pid(&mut child, &pid_file).await; + let launcher_handle = open_process_handle(launcher_pid); + let descendant_handle = open_process_handle(descendant_pid); + assert!( + process_is_running(&launcher_handle), + "batch launcher {launcher_pid} should be alive before shutdown" + ); + assert!( + process_is_running(&descendant_handle), + "descendant {descendant_pid} should be alive before shutdown" + ); + + if lifecycle == 2 { + drop(child); + } else { + tokio::time::timeout(Duration::from_secs(15), child.terminate()) + .await + .expect("process-tree shutdown must be bounded") + .expect("Job Object termination should succeed"); + } + + assert_process_exited(&launcher_handle, launcher_pid); + assert_process_exited(&descendant_handle, descendant_pid); + } + } + + async fn wait_for_descendant_pid(child: &mut ManagedChild, pid_file: &std::path::Path) -> u32 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if let Ok(contents) = fs::read_to_string(pid_file) { + if let Ok(pid) = contents.trim().parse::() { + return pid; + } + } + if let Some(status) = child.try_wait().expect("launcher status should be readable") { + panic!("batch launcher exited before descendant startup with {status}"); + } + assert!(tokio::time::Instant::now() < deadline, "timed out waiting for the long-running descendant PID"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + fn open_process_handle(pid: u32) -> OwnedHandle { + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, pid) }; + assert!(!handle.is_null(), "failed to open process {pid}: {}", std::io::Error::last_os_error()); + unsafe { OwnedHandle::from_raw_handle(handle as RawHandle) } + } + + fn process_is_running(handle: &OwnedHandle) -> bool { + match unsafe { WaitForSingleObject(handle.as_raw_handle() as HANDLE, 0) } { + WAIT_TIMEOUT => true, + WAIT_OBJECT_0 => false, + WAIT_FAILED => panic!("failed to query process handle: {}", std::io::Error::last_os_error()), + result => panic!("unexpected WaitForSingleObject result: {result}"), + } + } + + fn assert_process_exited(handle: &OwnedHandle, pid: u32) { + let result = unsafe { WaitForSingleObject(handle.as_raw_handle() as HANDLE, 5_000) }; + assert_eq!(result, WAIT_OBJECT_0, "process {pid} must have exited before shutdown returns"); + } + + #[test] + fn windows_process_tree_test_descendant() { + let Some(pid_file) = std::env::var_os("DBX_PROCESS_TREE_TEST_PID_FILE") else { + return; + }; + fs::write(pid_file, std::process::id().to_string()).expect("write descendant PID"); + loop { + thread::sleep(Duration::from_secs(60)); + } + } +} diff --git a/crates/dbx-plugin-runtime/src/plugins.rs b/crates/dbx-plugin-runtime/src/plugins.rs index c3552cefaf..ed06c84da9 100644 --- a/crates/dbx-plugin-runtime/src/plugins.rs +++ b/crates/dbx-plugin-runtime/src/plugins.rs @@ -313,8 +313,16 @@ impl PluginRegistry { let env = env.with_plugin_data_dir(&self.plugin_data_dir(&plugin.manifest.id)); let session = PluginSidecarSession::start(plugin, self.app_version.clone(), env).await?; let result = session.invoke_with_timeout(method, params, Some(driver_id), timeout_duration).await; - session.shutdown().await; - result + match (result, session.shutdown().await) { + (Ok(value), Ok(())) => Ok(value), + (Err(invoke_error), Ok(())) => Err(invoke_error), + (Ok(_), Err(shutdown_error)) => { + Err(format!("Plugin invocation completed but shutdown failed: {shutdown_error}")) + } + (Err(invoke_error), Err(shutdown_error)) => { + Err(format!("{invoke_error}; additionally failed to stop plugin: {shutdown_error}")) + } + } } pub async fn start_driver_session(&self, driver_id: &str) -> Result, String> { @@ -426,12 +434,12 @@ impl PluginDriverSession { self.sidecar.invoke_with_timeout(method, params, Some(&self.driver_id), timeout_duration).await } - pub async fn shutdown(&self) { + pub async fn shutdown(&self) -> Result<(), String> { if let Some(logical) = &self.logical { logical.close().await; - return; + return Ok(()); } - self.sidecar.shutdown().await; + self.sidecar.shutdown().await } pub fn is_available(&self) -> bool { @@ -584,7 +592,7 @@ sleep 30 pending.abort(); assert!(pending.await.unwrap_err().is_cancelled()); assert!(lifecycle.begin_update("sample.sidecar").is_ok()); - host.stop_all().await; + host.stop_all().await.unwrap(); } #[cfg(unix)] @@ -627,7 +635,7 @@ sleep 30 .expect("session should start"); let pid = session.pid().await.expect("child should have a pid"); - session.shutdown().await; + session.shutdown().await.unwrap(); tokio::time::sleep(std::time::Duration::from_millis(50)).await; assert!(!process_exists(pid)); diff --git a/crates/dbx-plugin-runtime/src/plugins/host.rs b/crates/dbx-plugin-runtime/src/plugins/host.rs index 0fe4b6ecbc..7917236a81 100644 --- a/crates/dbx-plugin-runtime/src/plugins/host.rs +++ b/crates/dbx-plugin-runtime/src/plugins/host.rs @@ -339,16 +339,25 @@ impl PluginHost { active } - pub async fn stop(&self, plugin_id: &str) { + pub async fn stop(&self, plugin_id: &str) -> Result<(), String> { if let Some(session) = self.inner.sessions.write().await.remove(plugin_id) { - session.shutdown().await; + session.shutdown().await?; } + Ok(()) } - pub async fn stop_all(&self) { + pub async fn stop_all(&self) -> Result<(), String> { let sessions = std::mem::take(&mut *self.inner.sessions.write().await); - for (_, session) in sessions { - session.shutdown().await; + let mut errors = Vec::new(); + for (plugin_id, session) in sessions { + if let Err(error) = session.shutdown().await { + errors.push(format!("{plugin_id}: {error}")); + } + } + if errors.is_empty() { + Ok(()) + } else { + Err(format!("Failed to stop plugin sessions: {}", errors.join("; "))) } } @@ -362,7 +371,7 @@ impl PluginHost { /// refused while the plugin still has an active connection or operation. pub async fn uninstall_plugin(&self, plugin_id: &str) -> Result<(), String> { let _update = self.inner.registry.lifecycle.begin_update(plugin_id)?; - self.stop(plugin_id).await; + self.stop(plugin_id).await?; let root_dir = self.inner.registry.root_dir().to_path_buf(); let app_version = self.inner.registry.app_version().to_string(); // Uninstall never validates a package signature, so it must not load the user trust store. diff --git a/crates/dbx-plugin-runtime/src/plugins/jdbc_sessions.rs b/crates/dbx-plugin-runtime/src/plugins/jdbc_sessions.rs index 542222ea34..fe1f30e998 100644 --- a/crates/dbx-plugin-runtime/src/plugins/jdbc_sessions.rs +++ b/crates/dbx-plugin-runtime/src/plugins/jdbc_sessions.rs @@ -45,7 +45,9 @@ impl PluginRegistry { if !members.1 && existing.process.sidecar.status().state == super::PluginSessionState::Running { members.0 += 1; } else { - existing.process.sidecar.shutdown().await; + if let Err(error) = existing.process.sidecar.shutdown().await { + log::warn!("Failed to stop the stale shared JDBC plugin runtime: {error}"); + } members.1 = true; drop(members); runtime = None; @@ -58,9 +60,17 @@ impl PluginRegistry { let mut startup = StartupGuard(Some(process.clone())); let protocol = process.invoke::("jdbcSessionProtocol", serde_json::json!({})).await; if !matches!(protocol, Ok(ref value) if value["version"] == 2) { - process.shutdown().await; + let shutdown = process.shutdown().await; startup.0 = None; - return Err("Update the JDBC plugin to a version supporting embedded H2 logical sessions".into()); + return Err(match shutdown { + Ok(()) => { + "Update the JDBC plugin to a version supporting embedded H2 logical sessions".to_string() + } + Err(error) => format!( + "Update the JDBC plugin to a version supporting embedded H2 logical sessions; additionally \ + failed to stop its runtime: {error}" + ), + }); } startup.0 = None; let runtime = Arc::new(JdbcRuntime { process, members: Mutex::new((1, false)) }); @@ -149,7 +159,9 @@ impl JdbcLogicalSession { members.0 -= 1; if members.0 == 0 { members.1 = true; - runtime.process.sidecar.shutdown().await; + if let Err(error) = runtime.process.sidecar.shutdown().await { + log::warn!("Failed to stop the shared JDBC plugin runtime: {error}"); + } } session.close_done.send_replace(true); }); @@ -166,7 +178,9 @@ impl Drop for StartupGuard { if let Some(process) = self.0.take() { if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { - process.sidecar.shutdown().await; + if let Err(error) = process.sidecar.shutdown().await { + log::warn!("Failed to stop the discarded JDBC plugin runtime: {error}"); + } }); } } diff --git a/crates/dbx-plugin-runtime/src/plugins/runtime.rs b/crates/dbx-plugin-runtime/src/plugins/runtime.rs index b6c4bedbd5..40dc6b9cf8 100644 --- a/crates/dbx-plugin-runtime/src/plugins/runtime.rs +++ b/crates/dbx-plugin-runtime/src/plugins/runtime.rs @@ -9,7 +9,7 @@ use bytes::Bytes; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; -use tokio::process::{Child, ChildStdin, ChildStdout}; +use tokio::process::{ChildStdin, ChildStdout}; use tokio::sync::{broadcast, oneshot, watch, Mutex, RwLock}; use tokio::time::Instant; @@ -158,7 +158,8 @@ pub struct PluginSidecarSession { plugin: InstalledPlugin, app_version: String, transport: PluginBackendTransport, - child: Arc>, + child: Arc>, + shutdown_lock: Mutex<()>, stdin: Mutex, pending: Arc>>, next_request_id: AtomicU64, @@ -216,10 +217,10 @@ impl PluginSidecarSession { ensure_plugin_backend(&plugin)?; let transport = plugin.manifest.backend_entrypoint().map(|backend| backend.transport).unwrap_or_default(); let app_version = app_version.into(); - let mut child = spawn_plugin_child(&plugin, &app_version, &env)?; - let stdin = child.stdin.take().ok_or("Plugin stdin unavailable")?; - let stdout = child.stdout.take().ok_or("Plugin stdout unavailable")?; - let stderr = child.stderr.take(); + let mut child = spawn_plugin_child(&plugin, &app_version, &env).await?; + let stdin = child.take_stdin().ok_or("Plugin stdin unavailable")?; + let stdout = child.take_stdout().ok_or("Plugin stdout unavailable")?; + let stderr = child.take_stderr(); let child = Arc::new(Mutex::new(child)); let pending = Arc::new(Mutex::new(HashMap::new())); let (events, _) = broadcast::channel(256); @@ -230,6 +231,7 @@ impl PluginSidecarSession { app_version, transport, child, + shutdown_lock: Mutex::new(()), stdin: Mutex::new(stdin), pending, next_request_id: AtomicU64::new(1), @@ -249,8 +251,12 @@ impl PluginSidecarSession { let handshake = match session.initialize().await { Ok(handshake) => handshake, Err(error) => { - session.shutdown().await; - return Err(format!("Plugin '{}' initialization failed: {error}", session.plugin.manifest.id)); + let shutdown_error = session.shutdown().await.err(); + let message = match shutdown_error { + Some(shutdown_error) => format!("{error}; additionally failed to shut down: {shutdown_error}"), + None => error, + }; + return Err(format!("Plugin '{}' initialization failed: {message}", session.plugin.manifest.id)); } }; *session.handshake.write().await = Some(handshake); @@ -264,11 +270,12 @@ impl PluginSidecarSession { }); if !transitioned { let status = session.status(); - session.shutdown().await; + let shutdown_error = session.shutdown().await.err(); + let message = status.message.map_or_else(String::new, |message| format!(": {message}")); + let shutdown_error = shutdown_error.map_or_else(String::new, |error| format!("; shutdown failed: {error}")); return Err(format!( - "Plugin '{}' stopped during initialization{}", - session.plugin.manifest.id, - status.message.map_or_else(String::new, |message| format!(": {message}")) + "Plugin '{}' stopped during initialization{message}{shutdown_error}", + session.plugin.manifest.id )); } Ok(session) @@ -354,22 +361,42 @@ impl PluginSidecarSession { stdin.flush().await.map_err(|error| self.write_error(error)) } - pub async fn shutdown(&self) { + pub async fn shutdown(&self) -> Result<(), String> { + let _shutdown = self.shutdown_lock.lock().await; + if self.status().state == PluginSessionState::Stopped { + return Ok(()); + } + self.status.send_replace(PluginSessionStatus::new(PluginSessionState::Stopping, None)); // Let any open user prompt resolve and close its dialog. self.prompts.close(); - let mut child = self.child.lock().await; - let kill_result = child.kill().await; - // tokio's kill() is start_kill() + wait(), so a successful kill has - // already reaped the child; this bounded wait only covers the rare - // kill-failure path where the process is still alive. The OS releases - // the executable image lock when the session (holding kill_on_drop - // Child) drops, which uninstall performs before renaming the plugin - // directory (fixes os error 5 / access denied on uninstall). - let _ = tokio::time::timeout(Duration::from_secs(5), child.wait()).await; - let message = kill_result.err().map(|error| error.to_string()); + // Stop the whole sidecar process tree, not just the tracked child. Script launchers can + // leave descendants that keep using the plugin container, which makes the rename-first + // uninstall fail with ERROR_ACCESS_DENIED even though the launcher itself is gone + // (#10601). `terminate()` is bounded by construction, so the app-exit path (`stop_all`) + // can never wait on it indefinitely: on Windows it terminates the private Job Object and + // confirms the tree has no active processes before reaping the launcher, and on other + // platforms it keeps tokio's direct-child kill semantics. The executable image lock is + // released when the session holding the `kill_on_drop` child drops, which uninstall + // performs before renaming the plugin directory (#10943). + let termination = self + .child + .lock() + .await + .terminate() + .await + .map_err(|error| format!("Failed to stop plugin '{}' process tree: {error}", self.plugin.manifest.id)); fail_pending(&self.pending, "Plugin session stopped").await; - self.status.send_replace(PluginSessionStatus::new(PluginSessionState::Stopped, message)); + match termination { + Ok(()) => { + self.status.send_replace(PluginSessionStatus::new(PluginSessionState::Stopped, None)); + Ok(()) + } + Err(error) => { + self.status.send_replace(PluginSessionStatus::new(PluginSessionState::Stopping, Some(error.clone()))); + Err(error) + } + } } pub async fn pid(&self) -> Option { @@ -602,14 +629,9 @@ impl PluginSidecarSession { } async fn terminate_after_output_end(&self, message: String) -> String { - let mut child = self.child.lock().await; - match child.try_wait() { - Ok(Some(_)) => message, - Ok(None) => match child.kill().await { - Ok(()) => format!("{message}; process terminated by host"), - Err(error) => format!("{message}; failed to terminate process: {error}"), - }, - Err(error) => format!("{message}; failed to inspect process: {error}"), + match self.child.lock().await.terminate().await { + Ok(()) => format!("{message}; process tree terminated by host"), + Err(error) => format!("{message}; failed to terminate process tree: {error}"), } } @@ -1056,7 +1078,11 @@ fn ensure_plugin_backend(plugin: &InstalledPlugin) -> Result<(), String> { Ok(()) } -fn spawn_plugin_child(plugin: &InstalledPlugin, app_version: &str, env: &PluginRuntimeEnv) -> Result { +async fn spawn_plugin_child( + plugin: &InstalledPlugin, + app_version: &str, + env: &PluginRuntimeEnv, +) -> Result { let executable_path = plugin.compatibility.backend_executable.as_ref().ok_or_else(|| { format!("Plugin '{}' does not provide a compatible backend executable", plugin.manifest.id) @@ -1075,17 +1101,19 @@ fn spawn_plugin_child(plugin: &InstalledPlugin, app_version: &str, env: &PluginR .env("DBX_HOST_API_VERSION", SUPPORTED_PLUGIN_HOST_API_VERSION) .env("DBX_PLUGIN_PROTOCOL_VERSION", SUPPORTED_PLUGIN_PROTOCOL_VERSION.to_string()); env.apply_to(&mut command); - command.spawn().map_err(|error| format!("Failed to start plugin '{}': {error}", plugin.manifest.id)) + crate::process::spawn_managed_child(&mut command) + .await + .map_err(|error| format!("Failed to start plugin '{}': {error}", plugin.manifest.id)) } -fn ensure_executable_permission(path: &Path) -> Result<(), String> { +fn ensure_executable_permission(_path: &Path) -> Result<(), String> { #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; - let mode = std::fs::metadata(path).map_err(|error| error.to_string())?.permissions().mode(); + let mode = std::fs::metadata(_path).map_err(|error| error.to_string())?.permissions().mode(); if mode & 0o111 == 0 { - return Err(format!("Plugin backend is not executable: {}", path.display())); + return Err(format!("Plugin backend is not executable: {}", _path.display())); } } Ok(()) @@ -1096,10 +1124,13 @@ mod tests { use std::time::Duration; use super::{ - decode_response_value, read_limited_line, trim_ascii_whitespace, user_input_result, PluginSessionState, - PluginSidecarSession, PromptActivity, UserInputSpec, USER_INPUT_DEFAULT_TIMEOUT, USER_INPUT_MAX_OPTIONS, - USER_INPUT_MAX_TIMEOUT, USER_INPUT_MIN_TIMEOUT, + decode_response_value, read_limited_line, trim_ascii_whitespace, user_input_result, PromptActivity, + UserInputSpec, USER_INPUT_DEFAULT_TIMEOUT, USER_INPUT_MAX_OPTIONS, USER_INPUT_MAX_TIMEOUT, + USER_INPUT_MIN_TIMEOUT, }; + #[cfg(unix)] + use super::{PluginSessionState, PluginSidecarSession}; + #[cfg(unix)] use crate::plugins::{InstalledPlugin, PluginManifest, PluginRuntimeEnv}; use tokio::io::BufReader; @@ -1310,7 +1341,7 @@ sleep 30 .invoke("sample/echo", serde_json::Value::Null) .await .expect("host should answer every plugin-initiated request"); - session.shutdown().await; + session.shutdown().await.unwrap(); // A headless host (tests, MCP) must refuse instead of hanging the // plugin until the prompt timeout expires. @@ -1335,7 +1366,7 @@ sleep 30 .invoke("sample/echo", serde_json::Value::Null) .await .expect("host should answer every plugin-initiated request"); - session.shutdown().await; + session.shutdown().await.unwrap(); ssh_prompt::clear_ssh_prompt_gateway(); let prompt = echoed_answer(&echoed, "prompt-1"); @@ -1367,7 +1398,7 @@ sleep 30 .invoke("sample/echo", serde_json::Value::Null) .await .expect("host should answer every plugin-initiated request"); - session.shutdown().await; + session.shutdown().await.unwrap(); ssh_prompt::clear_ssh_prompt_gateway(); let prompt = echoed_answer(&echoed, "prompt-1"); @@ -1394,7 +1425,7 @@ sleep 30 .invoke_with_timeout("sample/echo", serde_json::Value::Null, None, Some(Duration::from_millis(300))) .await .expect("a prompt must pause the request deadline instead of failing it"); - session.shutdown().await; + session.shutdown().await.unwrap(); ssh_prompt::clear_ssh_prompt_gateway(); assert_eq!(echoed_answer(&echoed, "prompt-1")["result"]["value"], "123456"); @@ -1421,7 +1452,7 @@ sleep 30 assert_eq!(request.kind, dbx_platform::ssh_prompt::SshPromptKind::UserInput); assert!(!responder.is_closed()); - session.shutdown().await; + session.shutdown().await.unwrap(); ssh_prompt::clear_ssh_prompt_gateway(); // The request fails (the session is gone) and the prompt's responder is @@ -1455,7 +1486,7 @@ sleep 30 assert!(!second_responder.is_closed()); assert_eq!(session.prompts.state.borrow().open, 2); - session.shutdown().await; + session.shutdown().await.unwrap(); let (first_answer, second_answer) = tokio::time::timeout(Duration::from_secs(1), async { tokio::join!(first, second) }) .await @@ -1520,7 +1551,7 @@ sleep 30 assert!(result.unwrap_err().contains("timed out after 30 seconds")); } tokio::time::resume(); - session.shutdown().await; + session.shutdown().await.unwrap(); } #[cfg(unix)] @@ -1542,7 +1573,7 @@ sleep 30 assert_eq!(first.await.unwrap(), serde_json::json!({ "ok": true })); assert!(second.await.unwrap_err().contains("response channel closed")); session.prompts.end(); - session.shutdown().await; + session.shutdown().await.unwrap(); } #[test] @@ -1689,7 +1720,7 @@ sleep 30 let event = events.recv().await.unwrap(); assert_eq!(event.method, "sample/progress"); assert_eq!(event.params["value"], 50); - session.shutdown().await; + session.shutdown().await.unwrap(); assert_eq!(session.status().state, PluginSessionState::Stopped); } @@ -1738,7 +1769,7 @@ sleep 30 let plugin = InstalledPlugin::new(manifest, dir.path().to_path_buf(), "0.5.67"); let error = match PluginSidecarSession::start(plugin, "0.5.67", PluginRuntimeEnv::default()).await { Ok(session) => { - session.shutdown().await; + session.shutdown().await.unwrap(); panic!("mismatched backend identity should fail initialization") } Err(error) => error, @@ -1787,7 +1818,7 @@ sleep 30 let session = PluginSidecarSession::start(plugin, "0.5.67", PluginRuntimeEnv::default()).await.unwrap(); let result: serde_json::Value = session.invoke("ping", serde_json::Value::Null).await.unwrap(); assert_eq!(result["ok"], true); - session.shutdown().await; + session.shutdown().await.unwrap(); } /// A legacy `stdio-jsonl` plugin answers with one line per response, so a whole @@ -1831,7 +1862,7 @@ sleep 30 .await .expect("a 12 MB legacy JSONL response must be read instead of failing the query"); assert_eq!(result["page"].as_str().map(str::len), Some(12_000_000)); - session.shutdown().await; + session.shutdown().await.unwrap(); } #[cfg(unix)] diff --git a/crates/dbx-web/src/routes/plugins.rs b/crates/dbx-web/src/routes/plugins.rs index 05693e4856..3705956e92 100644 --- a/crates/dbx-web/src/routes/plugins.rs +++ b/crates/dbx-web/src/routes/plugins.rs @@ -239,7 +239,7 @@ pub async fn install_marketplace_plugin( let result = marketplace.install(request).await.map_err(AppError::bad_request)?; state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await; stop_external_driver_pools(&state, &result.plugin).await; - state.app.plugin_host.stop(&result.plugin.manifest.id).await; + state.app.plugin_host.stop(&result.plugin.manifest.id).await.map_err(AppError::internal)?; Ok(Json(result.response())) } @@ -322,7 +322,7 @@ pub async fn install_plugin( .map_err(AppError::bad_request)?; state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await; stop_external_driver_pools(&state, &result.plugin).await; - state.app.plugin_host.stop(&result.plugin.manifest.id).await; + state.app.plugin_host.stop(&result.plugin.manifest.id).await.map_err(AppError::internal)?; Ok(Json(result.response())) } @@ -341,7 +341,7 @@ pub async fn rollback_plugin( .map_err(AppError::bad_request)?; state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await; stop_external_driver_pools(&state, &result.plugin).await; - state.app.plugin_host.stop(&result.plugin.manifest.id).await; + state.app.plugin_host.stop(&result.plugin.manifest.id).await.map_err(AppError::internal)?; Ok(Json(result.response())) } @@ -402,7 +402,7 @@ pub async fn stop_plugin( State(state): State>, Json(request): Json, ) -> Result, AppError> { - state.app.plugin_host.stop(&request.plugin_id).await; + state.app.plugin_host.stop(&request.plugin_id).await.map_err(AppError::internal)?; Ok(Json(serde_json::json!({ "ok": true }))) } diff --git a/src-tauri/src/commands/plugins.rs b/src-tauri/src/commands/plugins.rs index f782478b9a..db6cf46e2c 100644 --- a/src-tauri/src/commands/plugins.rs +++ b/src-tauri/src/commands/plugins.rs @@ -123,7 +123,7 @@ pub async fn install_marketplace_plugin( PluginMarketplace::new(state.plugins.root_dir().to_path_buf(), state.plugins.app_version().to_string())? .with_lifecycle(state.plugins.lifecycle()); let result = marketplace.install(request).await?; - stop_replaced_plugin_runtime(&state, &result.plugin).await; + stop_replaced_plugin_runtime(&state, &result.plugin).await?; emit_plugin_runtime_replaced(&app, &result.plugin); Ok(result.response()) } @@ -200,7 +200,7 @@ pub async fn install_plugin_package( }) .await .map_err(|error| error.to_string())??; - stop_replaced_plugin_runtime(&state, &result.plugin).await; + stop_replaced_plugin_runtime(&state, &result.plugin).await?; emit_plugin_runtime_replaced(&app, &result.plugin); Ok(result.response()) } @@ -228,7 +228,7 @@ pub async fn install_plugin_package_from_url( ); }) .await?; - stop_replaced_plugin_runtime(&state, &result.plugin).await; + stop_replaced_plugin_runtime(&state, &result.plugin).await?; emit_plugin_runtime_replaced(&app, &result.plugin); Ok(result.response()) } @@ -247,7 +247,7 @@ pub async fn rollback_plugin( }) .await .map_err(|error| error.to_string())??; - stop_replaced_plugin_runtime(&state, &result.plugin).await; + stop_replaced_plugin_runtime(&state, &result.plugin).await?; emit_plugin_runtime_replaced(&app, &result.plugin); Ok(result.response()) } @@ -318,7 +318,7 @@ pub async fn stop_plugin( // The plugin's workbenches are going away with its runtime; without this // sweep, handles it never closed would pin the shared registry quota. crate::commands::plugin_file::close_all_plugin_files(&file_state, &plugin_id); - state.plugin_host.stop(&plugin_id).await; + state.plugin_host.stop(&plugin_id).await?; Ok(()) } @@ -533,10 +533,10 @@ fn emit_plugin_runtime_replaced(app: &AppHandle, plugin: &InstalledPlugin) { ); } -async fn stop_replaced_plugin_runtime(state: &Arc, plugin: &InstalledPlugin) { +async fn stop_replaced_plugin_runtime(state: &Arc, plugin: &InstalledPlugin) -> Result<(), String> { state.remove_plugin_connection_pools(&plugin.manifest.id).await; stop_external_driver_pools(state, plugin).await; - state.plugin_host.stop(&plugin.manifest.id).await; + state.plugin_host.stop(&plugin.manifest.id).await } async fn stop_external_driver_pools(state: &Arc, plugin: &InstalledPlugin) {