diff --git a/crates/ark/src/console/console_repl.rs b/crates/ark/src/console/console_repl.rs index b5e52af889..4e2fbd347a 100644 --- a/crates/ark/src/console/console_repl.rs +++ b/crates/ark/src/console/console_repl.rs @@ -2509,6 +2509,9 @@ impl Console { return; } + // Interrupt any in-flight evaluations that have outlived their timeout + crate::timeout::check_timeout(); + // Skip running tasks if we don't have 128KB of stack space available. // This is 1/8th of the typical Windows stack space (1MB, whereas macOS // and Linux have 8MB). diff --git a/crates/ark/src/dap/dap_jupyter_handler.rs b/crates/ark/src/dap/dap_jupyter_handler.rs index 2ac5dc05d5..bacdbe3aef 100644 --- a/crates/ark/src/dap/dap_jupyter_handler.rs +++ b/crates/ark/src/dap/dap_jupyter_handler.rs @@ -32,6 +32,7 @@ use stdext::result::ResultExt; use crate::dap::dap_notebook; use crate::dap::dap_server::DapConsoleEvent; use crate::dap::dap_server::DapHandler; +use crate::dap::dap_server::EvaluateOutcome; use crate::dap::dap_state::Breakpoint; use crate::dap::dap_state::BreakpointState; use crate::dap::dap_state::Dap; @@ -138,24 +139,37 @@ impl DapJupyterHandler { let expression = args.expression; let frame_id = args.frame_id; - // This only returns `Some()` if R is idle. Even though we stopped at - // some point, causing the Evaluate request, R could be busy evaluating - // the next expression already. - let result = r_task::try_idle_task(move |capture| { + // This runs only if R is idle. Even though we stopped at some point, + // causing the Evaluate request, R could be busy evaluating the next + // expression already, in which case `try_idle_task()` returns `None`. + // `DapHandler::evaluate()` runs the eval under a timeout so a runaway + // expression is interrupted and doesn't freeze the kernel. + let outcome = r_task::try_idle_task(move |capture| { DapHandler::evaluate(&state, &expression, frame_id, capture) }); - match result { - None => Ok(self.error_response(seq, "evaluate", "R is busy")), - // The task raised an R error, caught by the sandbox on the R thread. - Some(Err(err)) => Ok(self.error_response(seq, "evaluate", &format!("{err}"))), - Some(Ok(Ok(body))) => { - let ResponseBody::Evaluate(eval) = body else { - return Err(anyhow::anyhow!("Unexpected response body from evaluate")); - }; + let Some(outcome) = outcome else { + return Ok(self.error_response(seq, "evaluate", "R is busy")); + }; + + let outcome = match outcome { + Ok(outcome) => outcome, + // A longjump that escaped `DapHandler::evaluate()`'s own `try_catch` + // (e.g. while building the response), caught by the try-idle sandbox. + Err(err) => return Ok(self.error_response(seq, "evaluate", &err.to_string())), + }; + + match outcome { + EvaluateOutcome::Ok(ResponseBody::Evaluate(eval)) => { Ok(self.success_response(seq, "evaluate", serde_json::to_value(eval)?)) }, - Some(Ok(Err(err))) => Ok(self.error_response(seq, "evaluate", &format!("{err}"))), + EvaluateOutcome::Ok(_) => { + Err(anyhow::anyhow!("Unexpected response body from evaluate")) + }, + EvaluateOutcome::TimedOut => { + Ok(self.error_response(seq, "evaluate", "Evaluation timed out")) + }, + EvaluateOutcome::Error(err) => Ok(self.error_response(seq, "evaluate", &err)), } } diff --git a/crates/ark/src/dap/dap_server.rs b/crates/ark/src/dap/dap_server.rs index 3abb28734b..fbd24eb95a 100644 --- a/crates/ark/src/dap/dap_server.rs +++ b/crates/ark/src/dap/dap_server.rs @@ -53,6 +53,8 @@ use crate::r_task::RTask; use crate::request::debug_request_command; use crate::request::DebugRequest; use crate::request::RRequest; +use crate::timeout::with_timeout; +use crate::timeout::EVAL_TIMEOUT; /// Sentinel expression sent by the frontend to notify the kernel that the user /// selected a different stack frame in the debugger UI. Subsequent console @@ -115,6 +117,14 @@ pub struct DapHandler { pub(crate) r_request_tx: Sender, } +/// Outcome of a debugger `evaluate` request, produced on the R thread by +/// [`DapHandler::evaluate()`] and turned into a DAP response by each transport. +pub(crate) enum EvaluateOutcome { + Ok(ResponseBody), + TimedOut, + Error(String), +} + impl DapHandler { pub fn new(state: Arc>, r_request_tx: Sender) -> Self { Self { @@ -125,10 +135,11 @@ impl DapHandler { /// Dispatch a parsed DAP request to the appropriate handler. /// - /// `Evaluate` is not handled here because `dispatch()` returns - /// synchronously, while Evaluate requires an async round-trip through the - /// R thread. Each transport calls [`DapHandler::evaluate()`] directly - /// inside its own async mechanism. + /// `Evaluate` is not handled here because it runs user code, which must + /// happen on the R thread. Each transport calls [`DapHandler::evaluate()`] + /// through its own mechanism (an idle task on the console, a try-idle one + /// on the notebook). The other commands run directly here without touching + /// the R thread. pub fn dispatch(&self, req: Request) -> DapOutput { let cmd = req.command.clone(); @@ -680,11 +691,11 @@ impl DapHandler { expression: &str, frame_id: Option, capture: &mut ConsoleOutputCapture, - ) -> anyhow::Result { + ) -> EvaluateOutcome { if expression == SELECTED_FRAME_EXPRESSION { log::trace!("DAP: Received frame selection sentinel, frame_id: {frame_id:?}"); Console::get_mut().set_debug_selected_frame_id(frame_id); - return Ok(ResponseBody::Evaluate(EvaluateResponse { + return EvaluateOutcome::Ok(ResponseBody::Evaluate(EvaluateResponse { result: String::new(), type_field: None, presentation_hint: None, @@ -706,18 +717,31 @@ impl DapHandler { // would leave the mutex locked forever. The env stays valid without the // lock because its owning `RObject` remains in the state's map, which // only the R thread mutates. - let env = state - .lock() - .unwrap() - .frame_env(frame_id) - .map_err(|err| anyhow::anyhow!("{err}"))?; + let env = match state.lock().unwrap().frame_env(frame_id) { + Ok(env) => env, + Err(err) => return EvaluateOutcome::Error(err.to_string()), + }; let capture = if print { Some(capture) } else { None }; - let variable = Dap::evaluate(expr, env, capture)?; + + // Run under `with_timeout()` so a long-running or inflooping Watch Pane + // expression is interrupted instead of freezing the kernel. + let (res, timed_out) = with_timeout(EVAL_TIMEOUT, || Dap::evaluate(expr, env, capture)); + + if timed_out { + return EvaluateOutcome::TimedOut; + } + + let variable = match res { + Ok(Ok(variable)) => variable, + Ok(Err(err)) => return EvaluateOutcome::Error(format!("{err}")), + // An R longjump that escaped `Dap::evaluate()`'s own `try_catch`. + Err(err) => return EvaluateOutcome::Error(err.to_string()), + }; log::trace!("DAP: Evaluate completed"); let response = state.lock().unwrap().into_evaluate_response(variable); - Ok(ResponseBody::Evaluate(response)) + EvaluateOutcome::Ok(ResponseBody::Evaluate(response)) } } @@ -918,8 +942,9 @@ impl DapServer { }; log::trace!("DAP: Got request: {:#?}", req); - // Evaluate is async: the response is sent later via `responses_tx`. - // It is the only command that needs transport-specific handling. + // Evaluate runs user code on the R thread, so it's handled off on an + // idle task to keep `serve()` responsive to other requests. Its + // response is delivered asynchronously via `responses_tx`. if let Command::Evaluate(args) = &req.command { let args = args.clone(); if let Err(err) = self.handle_evaluate(req, args) { @@ -995,8 +1020,12 @@ impl DapServer { self.server.send_event(event) } - // Tied to the TCP transport for now. For Jupyter we need to figure out how - // to do async responses with Jupyter's Control channel. + /// Evaluate an expression on the R thread, with a timeout. + /// + /// Spawns an idle task to handle the evaluation request asynchronously + /// without blocking the DAP service while the evaluation is in flight on + /// the R thread. The eval runs under `timeout::with_timeout()`, which + /// guards against a runaway expression (e.g. an infloop in the watch pane). fn handle_evaluate( &mut self, req: Request, @@ -1007,13 +1036,11 @@ impl DapServer { let state = self.handler.state.clone(); let responses_tx = self.responses_tx.clone(); - log::trace!("DAP: Spawning idle task for evaluate"); r_task::spawn(RTask::send_idle_any_prompt(async move |mut capture| { - log::trace!("DAP: Idle task started for evaluate"); - let rsp = match DapHandler::evaluate(&state, &expression, frame_id, &mut capture) { - Ok(body) => req.success(body), - Err(err) => req.error(&format!("Error: {err}")), + EvaluateOutcome::Ok(body) => req.success(body), + EvaluateOutcome::TimedOut => req.error("Evaluation timed out"), + EvaluateOutcome::Error(err) => req.error(&format!("Error: {err}")), }; responses_tx.send(rsp).log_err(); diff --git a/crates/ark/src/lib.rs b/crates/ark/src/lib.rs index 7178caa34b..bf7a8eaaa1 100644 --- a/crates/ark/src/lib.rs +++ b/crates/ark/src/lib.rs @@ -44,6 +44,7 @@ pub mod startup; pub mod strings; pub mod sys; pub mod thread; +pub mod timeout; pub mod traps; pub mod treesitter; pub mod ui; diff --git a/crates/ark/src/r_task.rs b/crates/ark/src/r_task.rs index b8ea10d2a9..ee1d27a39b 100644 --- a/crates/ark/src/r_task.rs +++ b/crates/ark/src/r_task.rs @@ -20,7 +20,6 @@ use crossbeam::channel::Sender; use harp::exec::r_sandbox; #[cfg(debug_assertions)] use libr::SEXP; -use stdext::result::ResultExt; use uuid::Uuid; use crate::console::Console; @@ -113,12 +112,9 @@ const TRY_IDLE_TIMEOUT: Duration = Duration::from_millis(200); /// Attempt to run `f` on the R thread if it is currently idle at a prompt. /// -/// Returns `None` if R was busy. Otherwise returns `Some` with the outcome: +/// Returns `None` if R was busy, i.e. the Console event loop did not accept the +/// task within `TRY_IDLE_TIMEOUT`. Otherwise returns `Some` with the outcome: /// `Ok` if `f` ran cleanly, `Err` if it raised an R error. -/// -/// On the bounded(0) channel, `send_timeout` only hands off the task when the -/// event loop's `Select` is parked receiving, so the task runs iff R is idle. -/// Waiting up to `TRY_IDLE_TIMEOUT`. pub(crate) fn try_idle_task(f: F) -> Option> where F: FnOnce(&mut ConsoleOutputCapture) -> T + Send + 'static, @@ -152,7 +148,10 @@ where }, } - done_rx.recv().log_err()?; + if done_rx.recv().is_err() { + return None; + } + let out = result.lock().unwrap().take(); out } diff --git a/crates/ark/src/timeout.rs b/crates/ark/src/timeout.rs new file mode 100644 index 0000000000..064447fdde --- /dev/null +++ b/crates/ark/src/timeout.rs @@ -0,0 +1,132 @@ +// +// timeout.rs +// +// Copyright (C) 2026 Posit Software, PBC. All rights reserved. +// +// + +//! A timeout that breaks runaway R-thread work with an interrupt. +//! +//! R runs on a single thread, so work handed to it (a debugger `evaluate`, say) +//! can loop forever and freeze the kernel. There's no way to cancel R code from +//! the outside other than an interrupt, which R polls for at loop iterations and +//! other check points. +//! +//! Instead of watching the clock from another thread, we piggyback on R's own +//! polled-events handler and call [`check_timeout()`] there. The interrupt +//! unwinds R back to the nearest `try_catch`, which surfaces it as an error. +//! +//! FIXME: This is unix-only: on Windows R's process-events callback is a no-op +//! (see `sys::windows::console`), so the handler never fires mid-computation +//! and the deadline is never checked. Fixed by https://github.com/posit-dev/ark/pull/1222. + +use std::cell::Cell; +use std::time::Duration; +use std::time::Instant; + +use harp::raii::RLocalInterruptsSuspended; + +/// How long an evaluation in `with_timeout()` may run before we interrupt it. +pub(crate) const EVAL_TIMEOUT: Duration = Duration::from_secs(1); + +// These variables are thread-local to provide safe lock-free interior +// mutability on the R thread +thread_local! { + /// When the in-flight evaluation should be interrupted, if it's still + /// running. Set by `with_timeout()`, read by `check_timeout()`. + static DEADLINE: Cell> = const { Cell::new(None) }; + + /// Set by `check_timeout()` when it trips the interrupt, read by + /// `with_timeout()` which reports the timeout. + static TIMED_OUT: Cell = const { Cell::new(false) }; +} + +/// Run `f` on the R thread, interrupting it if it outlasts `timeout`. +/// +/// `f` runs in `try_catch()` with interrupts and polled events enabled so +/// `check_timeout()` can fire at interrupt time and the resulting interrupt can +/// longjump. +/// +/// Note that the longjump is either caught by our `try_catch()`, or a +/// `try_catch()` inside `f`. An alternative approach would be to try and +/// propagate the interrupt with a Rust panic, but that needs to be carefully +/// engineered. Be aware that because of the approach taken here, it's possible +/// for code inside `f` to try and recover from Rust errors +/// (`harp::Error::TopLevelExecError`) caused by the cancellation interrupt. +/// +/// Returns `f`'s value wrapped in the `try_catch()` result and a boolean +/// indicating whether the timeout fired. +/// +/// Must run on the R thread. +pub(crate) fn with_timeout(timeout: Duration, f: F) -> (harp::Result, bool) +where + F: FnOnce() -> T, +{ + // Save and restore so a nested evaluation doesn't clobber an outer deadline. + let old_deadline = DEADLINE.replace(Some(Instant::now() + timeout)); + let old_timed_out = TIMED_OUT.replace(false); + + let res = try_catch_with_timeout(f); + + let timed_out = TIMED_OUT.get(); + + // If `check_timeout()` tripped the interrupt but `f` finished before R + // acted on it, a stray interrupt is left pending. Clear it here so it can't + // fire on a later evaluation. + if timed_out { + crate::signals::set_interrupts_pending(false); + } + + DEADLINE.set(old_deadline); + TIMED_OUT.set(old_timed_out); + + (res, timed_out) +} + +/// Called from the polled-events handler at R's interrupt check points. If the +/// in-flight evaluation has outlived its deadline, ask R to interrupt itself. +/// Runs on the R thread. +pub(crate) fn check_timeout() { + if TIMED_OUT.get() { + // Already tripped, don't ask twice + return; + } + let Some(deadline) = DEADLINE.get() else { + // No evaluation under a timeout + return; + }; + if Instant::now() >= deadline { + TIMED_OUT.set(true); + crate::signals::set_interrupts_pending(true); + } +} + +/// Run `f` in a `try_catch` with interrupts and polled events live. +/// +/// An outer sandbox (ReadConsole, or the task executor) suspends both; we +/// re-enable them so the watchdog runs during `f` and the interrupt it trips can +/// longjump. On Windows there's no polled-events handler to re-enable, so `f` +/// runs interruptible but unwatched. +#[cfg(unix)] +fn try_catch_with_timeout(f: F) -> harp::Result +where + F: FnOnce() -> T, +{ + let _interrupts = RLocalInterruptsSuspended::new(false); + // Re-enable Ark's polled-events handler so `check_timeout()` runs during `f`. + let _polled = harp::raii::RLocal::new( + Some(crate::console::r_polled_events as unsafe extern "C-unwind" fn()), + unsafe { libr::R_PolledEvents }, + ); + harp::try_catch(f) +} + +#[cfg(not(unix))] +fn try_catch_with_timeout(f: F) -> harp::Result +where + F: FnOnce() -> T, +{ + // TODO: Update after https://github.com/posit-dev/ark/pull/1222 is merged + let _interrupts = RLocalInterruptsSuspended::new(false); + harp::try_catch(f) +} diff --git a/crates/ark/tests/integration/dap_evaluate.rs b/crates/ark/tests/integration/dap_evaluate.rs index 12781ea36f..dbf671f757 100644 --- a/crates/ark/tests/integration/dap_evaluate.rs +++ b/crates/ark/tests/integration/dap_evaluate.rs @@ -428,6 +428,43 @@ fn test_dap_evaluate_erroring_print_does_not_deadlock() { dap.recv_continued(); } +/// The watch pane evaluates expressions while stopped in a frame. A runaway +/// expression there (`repeat 1`) must time out and leave the debug session +/// intact rather than freezing the kernel. This is the console (TCP) path; the +/// notebook path is checked in `test_notebook_watch_pane_infloop_times_out`. +/// https://github.com/posit-dev/positron/issues/14481 +#[test] +#[cfg_attr(target_os = "windows", ignore)] +fn test_dap_evaluate_infloop_times_out() { + let frontend = DummyArkFrontend::lock(); + let mut dap = frontend.start_dap(); + + let _file = frontend.send_source( + " +local({ + x <- 42 + browser() +}) +", + ); + dap.recv_stopped(); + + let stack = dap.stack_trace(); + let frame_id = stack[0].id; + + // `repeat 1` never returns. The kernel must interrupt it after the timeout. + let err = dap.evaluate_error("repeat 1", Some(frame_id)); + assert!(err.contains("timed out"), "got: {err}"); + + // The session survived the interrupt: a normal watch expression in the same + // frame still resolves. + let result = dap.evaluate("x", Some(frame_id)); + assert_eq!(result, "42"); + + frontend.debug_send_quit(); + dap.recv_continued(); +} + #[test] fn test_dap_evaluate_unknown_frame_id() { let frontend = DummyArkFrontend::lock(); diff --git a/crates/ark/tests/integration/dap_notebook.rs b/crates/ark/tests/integration/dap_notebook.rs index aceb005f83..22b9ecb5ed 100644 --- a/crates/ark/tests/integration/dap_notebook.rs +++ b/crates/ark/tests/integration/dap_notebook.rs @@ -1182,6 +1182,178 @@ fn test_notebook_evaluate_error() { assert_eq!(reply["body"]["result"], "2"); } +/// A runaway expression in the watch pane must not freeze the kernel. The +/// evaluation runs on the R main thread, so `try_idle_task` interrupts it after +/// a timeout and reports an error rather than blocking forever. The follow-up +/// evaluate checks the interrupt left no stray pending interrupt behind. +/// https://github.com/posit-dev/positron/issues/14481 +#[test] +#[cfg_attr(target_os = "windows", ignore)] +fn test_notebook_evaluate_timeout() { + let frontend = DummyArkFrontendNotebook::lock(); + + // `repeat 1` never returns. The kernel must interrupt it after the timeout. + let reply = notebook_evaluate(&frontend, 1, "repeat 1"); + assert_eq!(reply["success"], false); + let message = reply["message"].as_str().unwrap(); + assert!(message.contains("timed out"), "got: {message}"); + + // The kernel is still responsive and not stuck on a leftover interrupt. + let reply = notebook_evaluate(&frontend, 2, "1 + 1"); + assert_eq!(reply["success"], true); + assert_eq!(reply["body"]["result"], "2"); +} + +/// The watch pane evaluates expressions while stopped in a frame. A runaway +/// expression there (the canonical bug: `repeat 1`) must time out and leave the +/// debug session intact rather than freezing the kernel: the stopped frame stays +/// usable and execution can still be continued. +/// https://github.com/posit-dev/positron/issues/14481 +#[test] +#[cfg_attr(target_os = "windows", ignore)] +fn test_notebook_watch_pane_infloop_times_out() { + let frontend = DummyArkFrontendNotebook::lock(); + + let fn_code = "fn_watch <- function() {\n x <- 42\n x\n}"; + + // Dump the cell and set a breakpoint on the return expression (line 3). + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 1, + "command": "dumpCell", + "arguments": { "code": fn_code } + })); + frontend.recv_iopub_busy(); + let dump_reply = frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + let source_path = dump_reply["body"]["sourcePath"] + .as_str() + .unwrap() + .to_string(); + + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 2, + "command": "setBreakpoints", + "arguments": { + "source": { "path": &source_path }, + "breakpoints": [{ "line": 3 }] + } + })); + frontend.recv_iopub_busy(); + frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + + // Attach so the breakpoint fires. + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 3, + "command": "attach", + "arguments": { "request": "attach", "type": "notebook" } + })); + frontend.recv_iopub_busy(); + frontend.recv_iopub_debug_event(); // thread started + frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + + // Define the function. + frontend.send_execute_request_with_metadata( + fn_code, + ExecuteRequestOptions::default(), + serde_json::json!({ "cellId": "watch-def" }), + ); + frontend.recv_iopub_busy(); + frontend.recv_iopub_execute_input(); + let bp_event = frontend.recv_iopub_debug_event(); + assert_eq!(bp_event["event"], "breakpoint"); + frontend.recv_iopub_idle(); + frontend.recv_shell_execute_reply(); + + // Call it: stops at the breakpoint, kernel stays busy. + frontend.send_execute_request_with_metadata( + "fn_watch()", + ExecuteRequestOptions::default(), + serde_json::json!({ "cellId": "watch-call" }), + ); + frontend.recv_iopub_busy(); + frontend.recv_iopub_execute_input(); + let stopped = frontend.recv_iopub_debug_event(); + assert_eq!(stopped["event"], "stopped"); + + // The frame we're stopped in (what the watch pane targets). + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 4, + "command": "stackTrace", + "arguments": { "threadId": -1 } + })); + frontend.recv_iopub_busy(); + let stack_reply = frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + let frame_id = stack_reply["body"]["stackFrames"][0]["id"] + .as_i64() + .unwrap(); + + // The watch pane evaluates `repeat 1` in the stopped frame. It never returns, + // so the kernel must interrupt it after the timeout. + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 5, + "command": "evaluate", + "arguments": { "expression": "repeat 1", "frameId": frame_id } + })); + frontend.recv_iopub_busy(); + let eval_reply = frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + assert_eq!(eval_reply["success"], false); + let message = eval_reply["message"].as_str().unwrap(); + assert!(message.contains("timed out"), "got: {message}"); + + // The session survived the interrupt: a normal watch expression in the same + // frame still resolves. + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 6, + "command": "evaluate", + "arguments": { "expression": "x", "frameId": frame_id } + })); + frontend.recv_iopub_busy(); + let eval_reply = frontend.recv_debug_reply(); + frontend.recv_iopub_idle(); + assert_eq!(eval_reply["success"], true); + assert_eq!(eval_reply["body"]["result"], "42"); + + // And execution can still be continued to completion. + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 7, + "command": "continue", + "arguments": { "threadId": -1 } + })); + frontend.recv_debug_reply(); + frontend.recv_shell_execute_reply(); + let msgs = frontend.recv_iopub_interleaved(&[ + &[IopubExpectation::BusyControl, IopubExpectation::IdleControl], + &[ + IopubExpectation::DebugEvent, + IopubExpectation::ExecuteResult, + IopubExpectation::IdleShell, + ], + ]); + find_debug_event(&msgs, "continued"); + + // Disconnect to reset is_connected for other tests. + frontend.send_debug_request(serde_json::json!({ + "type": "request", + "seq": 8, + "command": "disconnect", + "arguments": { "restart": false } + })); + frontend.recv_debug_reply(); + frontend.recv_iopub_busy(); + frontend.recv_iopub_idle(); +} + /// A print method that raises an R error longjumps out of `Rf_PrintValue`. The /// evaluation must surface that as an error and leave both the try-idle /// handshake and the `Dap` state intact, so a follow-up evaluate still succeeds.