From b18be4dacc4dfc4f9635c0c9743a2fff5f37d4db Mon Sep 17 00:00:00 2001 From: Sebastian <106036+schoeller@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:10:56 +0200 Subject: [PATCH] Adding timeout protection to host from stuck runner --- crates/ocs_plugin_api/src/process.rs | 438 +++++++++++++++++++++++++-- docs/plugin-architecture.md | 13 + 2 files changed, 431 insertions(+), 20 deletions(-) diff --git a/crates/ocs_plugin_api/src/process.rs b/crates/ocs_plugin_api/src/process.rs index 32a71a5e..05b5b11d 100644 --- a/crates/ocs_plugin_api/src/process.rs +++ b/crates/ocs_plugin_api/src/process.rs @@ -5,7 +5,7 @@ use std::process::{Child, Command, Stdio}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc; use std::sync::Mutex; -use std::time::Duration; +use std::time::{Duration, Instant}; use interprocess::local_socket::traits::Listener; use interprocess::local_socket::{GenericNamespaced, ListenerOptions, Stream, ToNsName}; @@ -34,6 +34,42 @@ fn spawn_timeout() -> Duration { .unwrap_or(SPAWN_TIMEOUT) } +/// Default maximum time to wait for a plugin call to respond. +const CALL_TIMEOUT_DEFAULT: Duration = Duration::from_secs(30); + +fn call_timeout() -> Duration { + std::env::var("OCS_PLUGIN_CALL_TIMEOUT_SECS") + .ok() + .and_then(|s| s.parse().ok()) + .map(Duration::from_secs) + .unwrap_or(CALL_TIMEOUT_DEFAULT) +} + +/// Per-request-kind timeout floors. The user-configured default is raised to +/// these minima so that no request kind can be configured into an unsafe value. +fn request_timeout(kind: &'static str) -> Duration { + let base = call_timeout(); + let floor = match kind { + "GetManifest" | "GetRibbon" => Duration::from_secs(5), + "Dispatch" => Duration::from_secs(10), + "InteractiveEvent" | "GetPrompt" | "NeedsEntityPick" => Duration::from_secs(2), + _ => Duration::from_secs(1), + }; + base.max(floor) +} + +fn request_kind(req: &HostRequest) -> &'static str { + match req { + HostRequest::GetManifest => "GetManifest", + HostRequest::GetRibbon => "GetRibbon", + HostRequest::Dispatch { .. } => "Dispatch", + HostRequest::InteractiveEvent { .. } => "InteractiveEvent", + HostRequest::GetPrompt { .. } => "GetPrompt", + HostRequest::NeedsEntityPick { .. } => "NeedsEntityPick", + HostRequest::Shutdown => "Shutdown", + } +} + #[derive(Debug, thiserror::Error)] pub enum PluginError { #[error("IO error: {0}")] @@ -44,6 +80,11 @@ pub enum PluginError { Runner(String), #[error("spawn timeout: runner did not connect within {0:?}")] SpawnTimeout(Duration), + #[error("call timeout: {request} did not respond within {duration:?}")] + CallTimeout { + request: &'static str, + duration: Duration, + }, #[error("runner exited before connecting")] RunnerExited, #[error("unexpected response: {0:?}")] @@ -85,6 +126,7 @@ impl PluginProcess { .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn()?; + let child = Mutex::new(Some(child)); // Accept the runner connection with a timeout so a hung/crashed runner // does not block the host indefinitely. @@ -99,11 +141,15 @@ impl PluginProcess { } Ok(Err(e)) => return Err(e.into()), Err(mpsc::RecvTimeoutError::Timeout) => { - reap(child); + if let Some(child) = child.lock().unwrap_or_else(|e| e.into_inner()).take() { + reap(child); + } return Err(PluginError::SpawnTimeout(spawn_timeout())); } Err(mpsc::RecvTimeoutError::Disconnected) => { - reap(child); + if let Some(child) = child.lock().unwrap_or_else(|e| e.into_inner()).take() { + reap(child); + } return Err(PluginError::RunnerExited); } }; @@ -111,11 +157,11 @@ impl PluginProcess { // The runner first answers GetManifest and GetRibbon so the host can // build the UI without keeping the plugin object alive. let no_op = &mut |_| {}; - let manifest = match call(&stream, host, HostRequest::GetManifest, no_op)? { + let manifest = match call(&stream, &child, host, HostRequest::GetManifest, no_op)? { HostResponse::Manifest(m) => m, other => return Err(PluginError::UnexpectedResponse(other)), }; - let ribbon = match call(&stream, host, HostRequest::GetRibbon, no_op)? { + let ribbon = match call(&stream, &child, host, HostRequest::GetRibbon, no_op)? { HostResponse::Ribbon(r) => r, other => return Err(PluginError::UnexpectedResponse(other)), }; @@ -123,7 +169,7 @@ impl PluginProcess { let id = manifest.id.clone(); Ok(Self { stream, - child: Mutex::new(Some(child)), + child, id, manifest, ribbon, @@ -151,6 +197,7 @@ impl PluginProcess { eprintln!("[plugin] dispatching {cmd}"); let result = match call( &self.stream, + &self.child, host, HostRequest::Dispatch { cmd: cmd.to_string(), @@ -173,8 +220,17 @@ impl PluginProcess { event: InteractiveEvent, ) -> Result { self.send_request(HostRequest::InteractiveEvent { command_id, event })?; + let kind = "InteractiveEvent"; + let timeout = request_timeout(kind); + let deadline = Instant::now() + timeout; loop { - match self.recv_response::()? { + match recv_with_deadline::( + &self.stream, + &self.child, + deadline, + timeout, + kind, + )? { PluginToHost::Response(HostResponse::CommandStep(s)) => return Ok(s), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -192,8 +248,17 @@ impl PluginProcess { /// Ask the plugin process for the current prompt of an interactive command. pub fn get_prompt(&self, command_id: u64) -> Result { self.send_request(HostRequest::GetPrompt { command_id })?; + let kind = "GetPrompt"; + let timeout = request_timeout(kind); + let deadline = Instant::now() + timeout; loop { - match self.recv_response::()? { + match recv_with_deadline::( + &self.stream, + &self.child, + deadline, + timeout, + kind, + )? { PluginToHost::Response(HostResponse::Text(s)) => return Ok(s), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -211,8 +276,17 @@ impl PluginProcess { /// Ask the plugin process whether an interactive command wants object picks. pub fn needs_entity_pick(&self, command_id: u64) -> Result { self.send_request(HostRequest::NeedsEntityPick { command_id })?; + let kind = "NeedsEntityPick"; + let timeout = request_timeout(kind); + let deadline = Instant::now() + timeout; loop { - match self.recv_response::()? { + match recv_with_deadline::( + &self.stream, + &self.child, + deadline, + timeout, + kind, + )? { PluginToHost::Response(HostResponse::Bool(b)) => return Ok(b), PluginToHost::Response(other) => { return Err(PluginError::UnexpectedResponse(other)) @@ -276,12 +350,6 @@ impl PluginProcess { let stream = guard.as_mut().ok_or_else(shutdown_error)?; send(stream, &HostToPlugin::Response(resp)).map_err(Into::into) } - - fn recv_response(&self) -> Result { - let mut guard = self.stream.lock().unwrap_or_else(|e| e.into_inner()); - let stream = guard.as_mut().ok_or_else(shutdown_error)?; - recv(stream).map_err(Into::into) - } } /// Kill a child process and reap it without blocking the caller. The blocking @@ -301,14 +369,84 @@ fn shutdown_error() -> PluginError { )) } +/// Take the stream and child away from a process and kill the child without +/// blocking the caller. After this the process is considered dead and any +/// further IPC will fail. +fn mark_dead(stream: &Mutex>, child: &Mutex>) { + let _ = stream.lock().unwrap_or_else(|e| e.into_inner()).take(); + if let Some(child) = child.lock().unwrap_or_else(|e| e.into_inner()).take() { + reap(child); + } +} + +/// Receive one message from the plugin runner with a deadline. +/// +/// A short-lived reader thread performs the blocking `recv` so that the main +/// thread can time it out. If the deadline passes, the process is marked dead +/// (stream closed, child killed) so that subsequent dispatch attempts are +/// skipped. +fn recv_with_deadline( + stream: &Mutex>, + child: &Mutex>, + deadline: Instant, + timeout: Duration, + request: &'static str, +) -> Result { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + mark_dead(stream, child); + return Err(PluginError::CallTimeout { + request, + duration: timeout, + }); + } + + let (tx, rx) = mpsc::channel::<(Result, Option)>(); + let stream_to_thread = stream.lock().unwrap_or_else(|e| e.into_inner()).take(); + + std::thread::spawn(move || { + let result = match stream_to_thread { + Some(mut stream) => match recv::(&mut stream) { + Ok(msg) => (Ok(msg), Some(stream)), + Err(e) => (Err(PluginError::from(e)), Some(stream)), + }, + None => (Err(shutdown_error()), None), + }; + let _ = tx.send(result); + }); + + match rx.recv_timeout(remaining) { + Ok((Ok(msg), stream_opt)) => { + *stream.lock().unwrap_or_else(|e| e.into_inner()) = stream_opt; + Ok(msg) + } + Ok((Err(e), stream_opt)) => { + *stream.lock().unwrap_or_else(|e| e.into_inner()) = stream_opt; + Err(e) + } + Err(mpsc::RecvTimeoutError::Timeout) => { + mark_dead(stream, child); + Err(PluginError::CallTimeout { + request, + duration: timeout, + }) + } + Err(mpsc::RecvTimeoutError::Disconnected) => Err(shutdown_error()), + } +} + /// Send a host request and wait for the response, handling any nested plugin /// requests inline using the supplied `HostApi`. fn call( stream: &Mutex>, + child: &Mutex>, host: &mut dyn HostApi, req: HostRequest, on_start_interactive: &mut dyn FnMut(u64), ) -> Result { + let kind = request_kind(&req); + let timeout = request_timeout(kind); + let deadline = Instant::now() + timeout; eprintln!("[plugin] host -> runner: {req:?}"); { let mut guard = stream.lock().unwrap_or_else(|e| e.into_inner()); @@ -316,11 +454,7 @@ fn call( send(stream, &HostToPlugin::Request(req))?; } loop { - let msg = { - let mut guard = stream.lock().unwrap_or_else(|e| e.into_inner()); - let stream = guard.as_mut().ok_or_else(shutdown_error)?; - recv::(stream)? - }; + let msg = recv_with_deadline::(stream, child, deadline, timeout, kind)?; eprintln!("[plugin] runner -> host: {msg:?}"); match msg { PluginToHost::Response(resp) => return Ok(resp), @@ -434,3 +568,267 @@ mod tests { assert_eq!(runner, PathBuf::from("/app/OpenCADStudio-plugin-runner")); } } + +#[cfg(all(test, feature = "host"))] +mod timeout_tests { + use super::*; + use crate::host::{DocumentReader, HostApi, ReaderEntity}; + use crate::ipc::protocol::{ + HostRequest, HostResponse, HostToPlugin, PluginRequest, PluginResponse, PluginToHost, + }; + use crate::ipc::transport::{recv, send}; + use crate::ribbon::owned::OwnedPluginManifest; + use acadrust::xdata::ExtendedDataRecord; + use acadrust::{CadDocument, EntityType, Handle}; + use interprocess::local_socket::{ + traits::{Listener, Stream as StreamTrait}, + GenericNamespaced, ListenerOptions, Stream, ToNsName, + }; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::Mutex as StdMutex; + use std::thread; + use std::time::Instant; + + static ENV_LOCK: StdMutex<()> = StdMutex::new(()); + + struct EmptyReader; + + impl DocumentReader for EmptyReader { + fn entity_count(&self) -> usize { + 0 + } + fn for_each_entity(&self, _f: &mut dyn FnMut(ReaderEntity<'_>)) {} + fn layer_name(&self, _handle: Handle) -> Option<&str> { + None + } + fn app_id_name(&self, _handle: Handle) -> Option<&str> { + None + } + } + + struct DummyHost { + doc: CadDocument, + } + + impl HostApi for DummyHost { + fn tab_index(&self) -> usize { + 0 + } + fn document(&self) -> &CadDocument { + &self.doc + } + fn document_mut(&mut self) -> &mut CadDocument { + &mut self.doc + } + fn document_reader(&self) -> Box { + Box::new(EmptyReader) + } + fn add_entity(&mut self, _entity: EntityType) -> Handle { + panic!("not used") + } + fn bump_geometry(&mut self) {} + fn read_record(&self, _handle: Handle, _app_name: &str) -> Option<&ExtendedDataRecord> { + None + } + fn write_record(&mut self, _handle: Handle, _record: ExtendedDataRecord) -> bool { + false + } + fn remove_record(&mut self, _handle: Handle, _app_name: &str) -> bool { + false + } + fn push_undo(&mut self, _label: &str) {} + fn set_dirty(&mut self) {} + fn push_info(&mut self, _msg: &str) {} + fn push_output(&mut self, _msg: &str) {} + fn push_error(&mut self, _msg: &str) {} + fn start_interactive(&mut self, _command: Box) {} + fn plugin_state_any(&self, _plugin_id: &str) -> Option<&(dyn std::any::Any + Send + Sync)> { + None + } + fn plugin_state_any_mut( + &mut self, + _plugin_id: &str, + ) -> Option<&mut (dyn std::any::Any + Send + Sync)> { + None + } + fn ensure_plugin_state_any( + &mut self, + _plugin_id: &'static str, + _init: &mut dyn FnMut() -> Box, + ) -> &mut (dyn std::any::Any + Send + Sync) { + panic!("not used") + } + } + + fn unique_socket_name() -> String { + static COUNTER: AtomicU64 = AtomicU64::new(0); + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + format!("ocs_plugin_timeout_test_{}_{}", std::process::id(), n) + } + + fn connected_pair() -> (Stream, Stream) { + let name = unique_socket_name(); + let name_ref = name + .clone() + .to_ns_name::() + .expect("valid name"); + let listener = ListenerOptions::new() + .name(name_ref) + .create_sync() + .expect("listener"); + let client_name = name.clone(); + let client_thread = thread::spawn(move || { + StreamTrait::connect(client_name.to_ns_name::().unwrap()) + .expect("connect") + }); + let server = listener.accept().expect("accept"); + let client = client_thread.join().expect("client thread"); + (server, client) + } + + fn sleepy_child() -> Child { + #[cfg(windows)] + { + std::process::Command::new("cmd") + .arg("/c") + .arg("ping -n 30 127.0.0.1") + .stdout(std::process::Stdio::null()) + .spawn() + .expect("spawn sleep") + } + #[cfg(not(windows))] + { + std::process::Command::new("sleep") + .arg("30") + .spawn() + .expect("spawn sleep") + } + } + + fn fake_manifest() -> OwnedPluginManifest { + OwnedPluginManifest { + id: "test.plugin".to_string(), + name: "Test Plugin".to_string(), + version: "0.1.0".to_string(), + description: "test".to_string(), + api_version: 1, + ribbon_order: 0, + xdata_apps: vec![], + command_prefixes: vec![], + } + } + + fn fake_process() -> (PluginProcess, Stream) { + let (host_stream, runner_stream) = connected_pair(); + let process = PluginProcess { + stream: Mutex::new(Some(host_stream)), + child: Mutex::new(Some(sleepy_child())), + id: "test.plugin".to_string(), + manifest: fake_manifest(), + ribbon: vec![], + }; + (process, runner_stream) + } + + #[test] + fn dispatch_call_timeout_marks_process_dead() { + let _env_guard = ENV_LOCK.lock().expect("env lock"); + let prev = std::env::var("OCS_PLUGIN_CALL_TIMEOUT_SECS").ok(); + std::env::set_var("OCS_PLUGIN_CALL_TIMEOUT_SECS", "1"); + let (process, runner_stream) = fake_process(); + + let _runner = thread::spawn(move || { + let mut peer = runner_stream; + let req = recv::(&mut peer).expect("read dispatch"); + assert!( + matches!(req, HostToPlugin::Request(HostRequest::Dispatch { ref cmd }) if cmd == "HANG") + ); + // Block until the host closes the connection after the timeout. + let _ = recv::(&mut peer); + }); + + let mut host = DummyHost { + doc: CadDocument::default(), + }; + let start = Instant::now(); + let result = process.dispatch(&mut host, "HANG", &mut |_| {}); + let elapsed = start.elapsed(); + + assert!( + matches!( + result, + Err(PluginError::CallTimeout { + request: "Dispatch", + .. + }) + ), + "expected Dispatch timeout, got {result:?}" + ); + assert!( + elapsed >= Duration::from_secs(10), + "timeout should respect the 10 s Dispatch floor: {elapsed:?}" + ); + assert!( + elapsed < Duration::from_secs(12), + "timed out too slowly: {elapsed:?}" + ); + assert!(!process.is_alive(), "process should be marked dead"); + + // Do not join the fake runner thread: it blocks until the host closes + // the socket. In production the killed child process closes its end of + // the socket and the reader thread exits; this test uses a local thread + // instead, so we let it be reaped with the test process. + match prev { + Some(v) => std::env::set_var("OCS_PLUGIN_CALL_TIMEOUT_SECS", v), + None => std::env::remove_var("OCS_PLUGIN_CALL_TIMEOUT_SECS"), + } + } + + #[test] + fn dispatch_succeeds_with_nested_request_within_deadline() { + let _env_guard = ENV_LOCK.lock().expect("env lock"); + let prev = std::env::var("OCS_PLUGIN_CALL_TIMEOUT_SECS").ok(); + std::env::set_var("OCS_PLUGIN_CALL_TIMEOUT_SECS", "2"); + let (process, runner_stream) = fake_process(); + + let runner = thread::spawn(move || { + let mut peer = runner_stream; + let req = recv::(&mut peer).expect("read dispatch"); + assert!( + matches!(req, HostToPlugin::Request(HostRequest::Dispatch { ref cmd }) if cmd == "NESTED") + ); + send( + &mut peer, + &PluginToHost::Request(PluginRequest::PushInfo("hello".to_string())), + ) + .expect("send nested request"); + let resp = recv::(&mut peer).expect("read nested response"); + assert!(matches!(resp, HostToPlugin::Response(PluginResponse::Ok))); + send(&mut peer, &PluginToHost::Response(HostResponse::Bool(true))) + .expect("send final response"); + }); + + let mut host = DummyHost { + doc: CadDocument::default(), + }; + let result = process.dispatch(&mut host, "NESTED", &mut |_| {}); + assert!(result.expect("dispatch succeeds")); + assert!(process.is_alive(), "process should still be alive"); + + // Clean up the helper child so it does not outlive the test. + if let Some(mut child) = process + .child + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take() + { + let _ = child.kill(); + } + + runner.join().expect("runner thread"); + match prev { + Some(v) => std::env::set_var("OCS_PLUGIN_CALL_TIMEOUT_SECS", v), + None => std::env::remove_var("OCS_PLUGIN_CALL_TIMEOUT_SECS"), + } + } +} diff --git a/docs/plugin-architecture.md b/docs/plugin-architecture.md index d4e3d692..84295787 100644 --- a/docs/plugin-architecture.md +++ b/docs/plugin-architecture.md @@ -305,6 +305,17 @@ corruption cannot affect the host or other plugins. Plugin processes stay `try_dispatch` path the host uses and honour the enable/disable set (`disabled_plugins` in `settings.txt`). +Two timeouts protect the host from a stuck runner: + +| Timeout | Env var | Default | Floor | +|---|---|---|---| +| Spawn (connection) | `OCS_PLUGIN_SPAWN_TIMEOUT_SECS` | 10 s | — | +| Per-call | `OCS_PLUGIN_CALL_TIMEOUT_SECS` | 30 s | `GetManifest`/`GetRibbon` ≥ 5 s, `Dispatch` ≥ 10 s, interactive events/prompt/pick ≥ 2 s | + +A call timeout covers the full round-trip, including any nested plugin→host +requests handled inline. When it fires the host kills the runner, marks the +plugin dead, and reports a `CallTimeout` error via the normal plugin error path. + `` is `%APPDATA%` (Windows), `~/Library/Application Support` (macOS), or `$XDG_CONFIG_HOME` / `~/.config` (Linux). @@ -314,6 +325,7 @@ corruption cannot affect the host or other plugins. Plugin processes stay |---|---| | Plugin panics | Caught inside the plugin runner child; an error response is returned to the host and stays alive. | | Plugin crash / hang / malformed message | The host detects a dead process via `try_wait` on the next dispatch or ribbon rebuild; the tab is dropped and an error is logged. | +| Slow or non-responsive call | The per-call timeout (`OCS_PLUGIN_CALL_TIMEOUT_SECS`) fires, kills the runner, and surfaces a `CallTimeout` error. | | Spawn failure | Reported per-plugin during startup and surfaced in the Plugin Manager / command line. | | Oversized message | The length-framed transport rejects messages larger than 64 MiB. | @@ -371,6 +383,7 @@ Done: - [x] Marketplace — curated registry + manual repo link, install / upgrade / reinstall / uninstall, enable/disable. - [x] Interactive command round-trip over IPC (prompt, point/enter/object-pick). +- [x] Spawn and per-call IPC timeouts so a stuck runner cannot freeze the host. Next: