github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-19 16:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugins): terminate Windows sidecar process trees

Browse Source
This commit is contained in:
jearhe
2026-10-04 10:37:44 +08:00
committed by GitHub
parent c13af204dd
commit 88c006ee37
15 changed files with 656 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -2561,6 +2561,7 @@ dependencies = [
"serde",
"tokio",
"uuid",
"windows-sys 0.61.2",
]
[[package]]
agents/drivers/duckdb/Cargo.lock
Generated
+1
View File
@@ -1577,6 +1577,7 @@ dependencies = [
"serde",
"tokio",
"uuid",
"windows-sys 0.61.2",
]
[[package]]
crates/dbx-core/examples/plugin_package_smoke.rs
+1 -1
View File
@@ -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());
crates/dbx-core/src/ai/agent_explain.rs
+1 -1
View File
@@ -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();
}
crates/dbx-core/src/connection/h2_jdbc_tests.rs
+17 -15
View File
@@ -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();
}
crates/dbx-core/src/connection/mod.rs
+14 -4
View File
@@ -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 {
crates/dbx-core/src/query/mod.rs
+3 -3
View File
@@ -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);
}
crates/dbx-platform/Cargo.toml
+10 -1
View File
@@ -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",
] }
crates/dbx-platform/src/process.rs
+465
View File
@@ -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<u32> {
self.child.id()
}
pub fn take_stdin(&mut self) -> Option<ChildStdin> {
self.child.stdin.take()
}
pub fn take_stdout(&mut self) -> Option<ChildStdout> {
self.child.stdout.take()
}
pub fn take_stderr(&mut self) -> Option<ChildStderr> {
self.child.stderr.take()
}
pub fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
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<ManagedChild> {
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>) -> 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<OwnedHandle>,
terminated: bool,
}
impl ProcessTreeJob {
pub(super) fn new() -> io::Result<Self> {
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::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() 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::<THREADENTRY32>() 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<u32> {
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::<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION>() as u32,
std::ptr::null_mut(),
)
};
if queried == 0 {
return Err(io::Error::last_os_error());
}
Ok(accounting.ActiveProcesses)
}
fn handle(&self) -> io::Result<HANDLE> {
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<OsStr>) -> 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::<u32>() {
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));
}
}
}
crates/dbx-plugin-runtime/src/plugins.rs
+15 -7
View File
@@ -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<Arc<PluginDriverSession>, 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));
crates/dbx-plugin-runtime/src/plugins/host.rs
+15 -6
View File
@@ -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.
crates/dbx-plugin-runtime/src/plugins/jdbc_sessions.rs
+19 -5
View File
@@ -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::<serde_json::Value>("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}");
}
});
}
}
crates/dbx-plugin-runtime/src/plugins/runtime.rs
+83 -52
View File
@@ -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<Mutex<Child>>,
child: Arc<Mutex<crate::process::ManagedChild>>,
shutdown_lock: Mutex<()>,
stdin: Mutex<ChildStdin>,
pending: Arc<Mutex<HashMap<u64, PendingResponse>>>,
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<u32> {
@@ -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<Child, String> {
async fn spawn_plugin_child(
plugin: &InstalledPlugin,
app_version: &str,
env: &PluginRuntimeEnv,
) -> Result<crate::process::ManagedChild, String> {
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)]
crates/dbx-web/src/routes/plugins.rs
+4 -4
View File
@@ -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<Arc<WebState>>,
Json(request): Json<PluginIdRequest>,
) -> Result<Json<serde_json::Value>, 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 })))
}
src-tauri/src/commands/plugins.rs
+7 -7
View File
@@ -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<AppState>, plugin: &InstalledPlugin) {
async fn stop_replaced_plugin_runtime(state: &Arc<AppState>, 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<AppState>, plugin: &InstalledPlugin) {