From e21e6bcdf92db669a85271539f2c84eddff4dee7 Mon Sep 17 00:00:00 2001 From: Release Repro Date: Tue, 28 Jul 2026 15:38:35 -0400 Subject: [PATCH] refactor(sandbox): harden activation recovery --- .../src/server/handler/sessions.rs | 4 + .../fabro-sandbox/src/daytona/mod.rs | 92 +++++- lib/components/fabro-sandbox/src/docker.rs | 267 ++++++++++-------- lib/components/fabro-sandbox/src/local.rs | 3 + lib/components/fabro-sandbox/src/sandbox.rs | 5 +- .../fabro-sandbox/src/test_support.rs | 37 +++ .../fabro-workflow/src/lifecycle/mod.rs | 21 +- .../fabro-workflow/src/pipeline/execute.rs | 7 + .../src/pipeline/execute/tests.rs | 46 ++- .../fabro-workflow/src/pipeline/initialize.rs | 2 + lib/foundation/fabro-core/src/error.rs | 32 +++ 11 files changed, 375 insertions(+), 141 deletions(-) diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index c25d7ca27..d571078b1 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -730,6 +730,10 @@ async fn build_agent_session( let sandbox = reconnect_for_run(sandbox_instance, daytona_api_key, Some(run_id)) .await .map_err(AskFabroBuildError::SandboxUnavailable)?; + sandbox + .activate() + .await + .map_err(|err| AskFabroBuildError::SandboxUnavailable(anyhow::Error::new(err)))?; let sandbox: Arc = Arc::from(sandbox); // No optional web-tool dependencies: `AskFabroToolAccessPolicy` denies // `web_search` and `web_fetch`, and both `tools()` and the prompt are diff --git a/lib/components/fabro-sandbox/src/daytona/mod.rs b/lib/components/fabro-sandbox/src/daytona/mod.rs index c4e01e6ff..8de651f54 100644 --- a/lib/components/fabro-sandbox/src/daytona/mod.rs +++ b/lib/components/fabro-sandbox/src/daytona/mod.rs @@ -62,6 +62,7 @@ pub(crate) const DAYTONA_DASHBOARD_SANDBOXES_URL: &str = "https://app.daytona.io/dashboard/sandboxes"; const FABRO_SANDBOX_USER_AGENT: &str = concat!("fabro-sandbox/", env!("CARGO_PKG_VERSION")); const DAYTONA_PROBE_TIMEOUT: Duration = Duration::from_secs(20); +const DAYTONA_START_TIMEOUT: Duration = Duration::from_mins(1); /// Upper bound on explicit and Drop-triggered Daytona session deletion so a /// stalled REST call cannot block cancellation/timeout paths indefinitely. const DAYTONA_SESSION_CLOSE_TIMEOUT: Duration = Duration::from_secs(10); @@ -1376,6 +1377,14 @@ impl Sandbox for DaytonaSandbox { if current.state == Some(SandboxState::Started) { return Ok(()); } + if current.state == Some(SandboxState::Starting) { + return current + .wait_for_start(Some(DAYTONA_START_TIMEOUT)) + .await + .map_err(|e| { + crate::Error::context("Failed to wait for Daytona sandbox activation", e) + }); + } self.start().await } @@ -2534,10 +2543,12 @@ fn build_bash_session_command( #[cfg(test)] mod tests { + use std::sync::atomic::AtomicU32; + use daytona_api_client::models::api_key_list::Permissions; use fabro_util::error::collect_chain; - use httpmock::Method::GET; - use httpmock::MockServer; + use httpmock::Method::{GET, POST}; + use httpmock::{HttpMockResponse, MockServer}; use super::*; use crate::sandbox::BASH_PROBE_MARKER; @@ -2640,7 +2651,7 @@ mod tests { }) } - fn sandbox_body(name: &str, state: &str) -> serde_json::Value { + fn sandbox_body(name: &str, state: SandboxState) -> serde_json::Value { serde_json::json!({ "id": name, "organizationId": "org-1", @@ -2655,7 +2666,7 @@ mod tests { "gpu": 0.0, "memory": 4.0, "disk": 20.0, - "state": state + "state": state.to_string() }) } @@ -2824,7 +2835,17 @@ mod tests { .header("authorization", "Bearer dtn_test"); then.status(200) .header("content-type", "application/json") - .json_body(sandbox_body("test-sandbox", "started")); + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); + }) + .await; + let start_sandbox = server + .mock_async(|when, then| { + when.method(POST) + .path("/sandbox/test-sandbox/start") + .header("authorization", "Bearer dtn_test"); + then.status(200) + .header("content-type", "application/json") + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); }) .await; let sandbox = mock_daytona_sandbox(&server, "dtn_test", DaytonaConfig::default()).await; @@ -2838,12 +2859,71 @@ mod tests { .set(sdk_sandbox) .expect("test sandbox should initialize once"); + let get_calls_before = get_sandbox.calls_async().await; sandbox .activate() .await .expect("an active sandbox should require no restart"); - get_sandbox.assert_calls_async(2).await; + assert_eq!(get_sandbox.calls_async().await, get_calls_before + 1); + start_sandbox.assert_calls_async(0).await; + } + + #[tokio::test] + async fn activate_waits_for_a_daytona_start_already_in_progress() { + let server = MockServer::start_async().await; + let response_count = Arc::new(AtomicU32::new(0)); + let get_sandbox = server + .mock_async({ + let response_count = Arc::clone(&response_count); + move |when, then| { + when.method(GET) + .path("/sandbox/test-sandbox") + .header("authorization", "Bearer dtn_test"); + then.respond_with(move |_| { + let state = if response_count.fetch_add(1, Ordering::Relaxed) == 1 { + SandboxState::Starting + } else { + SandboxState::Started + }; + HttpMockResponse::builder() + .status(200) + .header("content-type", "application/json") + .body(sandbox_body("test-sandbox", state).to_string()) + .build() + }); + } + }) + .await; + let start_sandbox = server + .mock_async(|when, then| { + when.method(POST) + .path("/sandbox/test-sandbox/start") + .header("authorization", "Bearer dtn_test"); + then.status(200) + .header("content-type", "application/json") + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); + }) + .await; + let sandbox = mock_daytona_sandbox(&server, "dtn_test", DaytonaConfig::default()).await; + let sdk_sandbox = sandbox + .client + .get("test-sandbox") + .await + .expect("test sandbox should load"); + sandbox + .sandbox + .set(sdk_sandbox) + .expect("test sandbox should initialize once"); + + let get_calls_before = get_sandbox.calls_async().await; + sandbox + .activate() + .await + .expect("an in-progress start should be awaited"); + + assert_eq!(get_sandbox.calls_async().await, get_calls_before + 2); + start_sandbox.assert_calls_async(0).await; } #[tokio::test] diff --git a/lib/components/fabro-sandbox/src/docker.rs b/lib/components/fabro-sandbox/src/docker.rs index 50556550c..6236d425d 100644 --- a/lib/components/fabro-sandbox/src/docker.rs +++ b/lib/components/fabro-sandbox/src/docker.rs @@ -15,7 +15,7 @@ use bollard::container::{ use bollard::errors::Error as DockerError; use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults}; use bollard::image::CreateImageOptions; -use bollard::models::HostConfig; +use bollard::models::{ContainerInspectResponse, HostConfig}; use fabro_github::GitHubCredentials; use fabro_types::{CommandOutputStream, CommandTermination, RunId}; use fabro_util::time::elapsed_ms; @@ -145,6 +145,13 @@ enum EnsureImageOutcome { Pulled, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display)] +#[strum(serialize_all = "lowercase")] +enum ContainerStartAction { + Start, + Unpause, +} + impl DockerSandbox { pub fn new( config: DockerSandboxOptions, @@ -154,7 +161,25 @@ impl DockerSandbox { clone_branch: Option, ) -> crate::Result { let docker = Docker::connect_with_local_defaults().map_err(crate::Error::docker_connect)?; - Ok(Self { + Ok(Self::with_docker_client( + docker, + config, + github_app, + run_id, + clone_origin_url, + clone_branch, + )) + } + + fn with_docker_client( + docker: Docker, + config: DockerSandboxOptions, + github_app: Option, + run_id: Option, + clone_origin_url: Option, + clone_branch: Option, + ) -> Self { + Self { docker, config, github_app, @@ -169,7 +194,7 @@ impl DockerSandbox { cached_os_version: std::sync::OnceLock::new(), rg_available: OnceCell::const_new(), event_callback: None, - }) + } } pub async fn reconnect( @@ -755,24 +780,26 @@ impl DockerSandbox { verify_managed_labels(container_id, &labels, self.run_id.as_ref()) } - async fn inspect_labels(&self, container_id: &str) -> crate::Result> { - let inspect = self - .docker + async fn inspect_container( + &self, + container_id: &str, + ) -> crate::Result { + self.docker .inspect_container(container_id, None::) .await - .map_err(|e| { - if docker_not_found(&e) { - crate::Error::message(format!("Docker container '{container_id}' is gone")) + .map_err(|source| { + let message = if docker_not_found(&source) { + format!("Docker container '{container_id}' is gone") } else { - crate::Error::message(format!( - "Failed to inspect Docker container '{container_id}': {e}" - )) - } - })?; - Ok(inspect - .config - .and_then(|config| config.labels) - .unwrap_or_default()) + format!("Failed to inspect Docker container '{container_id}'") + }; + crate::Error::context(message, source) + }) + } + + async fn inspect_labels(&self, container_id: &str) -> crate::Result> { + let inspect = self.inspect_container(container_id).await?; + Ok(container_labels(&inspect)) } async fn ensure_name_available(&self) -> crate::Result> { @@ -835,6 +862,67 @@ impl DockerSandbox { .map_err(|e| crate::Error::context("Failed to upload file to container", e)) } + fn begin_start(&self) -> Instant { + self.emit(SandboxEvent::StartStarted { + provider: "docker".into(), + }); + Instant::now() + } + + async fn set_container_running( + &self, + container_id: &str, + labels: &HashMap, + action: ContainerStartAction, + ) -> crate::Result<()> { + let result = match action { + ContainerStartAction::Start => { + self.docker + .start_container(container_id, None::>) + .await + } + ContainerStartAction::Unpause => self.docker.unpause_container(container_id).await, + }; + if let Err(source) = result { + if !docker_not_modified(&source) { + return Err(crate::Error::context( + format!( + "Failed to {action} Docker container '{container_id}' with labels {labels:?}" + ), + source, + )); + } + } + Ok(()) + } + + async fn complete_start( + &self, + started: Instant, + container_id: &str, + labels: &HashMap, + action: ContainerStartAction, + ) -> crate::Result<()> { + if let Err(error) = self + .set_container_running(container_id, labels, action) + .await + { + return self.start_error(error); + } + if let Err(error) = self.probe_bash(None).await { + return self.start_error(crate::Error::context( + format!("Docker container '{container_id}' health check"), + error, + )); + } + + self.emit(SandboxEvent::StartCompleted { + provider: "docker".into(), + duration_ms: elapsed_ms(started), + }); + Ok(()) + } + fn start_error(&self, error: crate::Error) -> crate::Result<()> { self.emit(SandboxEvent::StartFailed { provider: "docker".into(), @@ -1230,6 +1318,24 @@ fn verify_managed_labels( Ok(()) } +fn container_labels(inspect: &ContainerInspectResponse) -> HashMap { + inspect + .config + .as_ref() + .and_then(|config| config.labels.clone()) + .unwrap_or_default() +} + +fn activation_action(inspect: &ContainerInspectResponse) -> Option { + let Some(state) = inspect.state.as_ref() else { + return Some(ContainerStartAction::Start); + }; + if state.running != Some(true) { + return Some(ContainerStartAction::Start); + } + (state.paused == Some(true)).then_some(ContainerStartAction::Unpause) +} + fn docker_not_found(error: &DockerError) -> bool { matches!(error, DockerError::DockerResponseServerError { status_code: 404, @@ -1418,93 +1524,40 @@ impl Sandbox for DockerSandbox { } async fn start(&self) -> crate::Result<()> { - self.emit(SandboxEvent::StartStarted { - provider: "docker".into(), - }); - let start = Instant::now(); + let started = self.begin_start(); let container_id = self.container_id()?.to_string(); - let labels = match self.inspect_labels(&container_id).await { - Ok(labels) => labels, - Err(e) => return self.start_error(e), + let inspect = match self.inspect_container(&container_id).await { + Ok(inspect) => inspect, + Err(error) => return self.start_error(error), }; - if let Err(e) = verify_managed_labels(&container_id, &labels, self.run_id.as_ref()) { - return self.start_error(e); + let labels = container_labels(&inspect); + if let Err(error) = verify_managed_labels(&container_id, &labels, self.run_id.as_ref()) { + return self.start_error(error); } - - if let Err(e) = self - .docker - .start_container(&container_id, None::>) + let action = activation_action(&inspect).unwrap_or(ContainerStartAction::Start); + self.complete_start(started, &container_id, &labels, action) .await - { - if !docker_not_modified(&e) { - return self.start_error(crate::Error::context( - format!( - "Failed to start Docker container '{container_id}' with labels {labels:?}" - ), - e, - )); - } - } - - if let Err(e) = self.probe_bash(None).await { - return self.start_error(crate::Error::context( - format!("Docker container '{container_id}' health check"), - e, - )); - } - - let duration_ms = elapsed_ms(start); - self.emit(SandboxEvent::StartCompleted { - provider: "docker".into(), - duration_ms, - }); - Ok(()) } async fn activate(&self) -> crate::Result<()> { let container_id = self.container_id()?.to_string(); - let inspect = self - .docker - .inspect_container(&container_id, None::) - .await - .map_err(|e| { - if docker_not_found(&e) { - crate::Error::message(format!("Docker container '{container_id}' is gone")) - } else { - crate::Error::message(format!( - "Failed to inspect Docker container '{container_id}': {e}" - )) - } - })?; - let labels = inspect - .config - .and_then(|config| config.labels) - .unwrap_or_default(); + let inspect = self.inspect_container(&container_id).await?; + let labels = container_labels(&inspect); verify_managed_labels(&container_id, &labels, self.run_id.as_ref())?; - if inspect - .state - .as_ref() - .is_some_and(|state| state.running == Some(true)) - { - if inspect - .state - .is_some_and(|state| state.paused != Some(true)) - { - return Ok(()); - } - if let Err(e) = self.docker.unpause_container(&container_id).await { - if !docker_not_modified(&e) { - return Err(crate::Error::context( - format!( - "Failed to unpause Docker container '{container_id}' with labels {labels:?}" - ), - e, - )); - } - } + let Some(action) = activation_action(&inspect) else { return Ok(()); + }; + match action { + ContainerStartAction::Unpause => { + self.set_container_running(&container_id, &labels, action) + .await + } + ContainerStartAction::Start => { + let started = self.begin_start(); + self.complete_start(started, &container_id, &labels, action) + .await + } } - self.start().await } async fn stop(&self) -> crate::Result<()> { @@ -2195,9 +2248,7 @@ mod tests { .header("content-type", "application/json") .json_body(serde_json::json!({ "Config": { - "Labels": { - "sh.fabro.managed": "true" - } + "Labels": managed_labels::for_run(None) }, "State": { "Running": true, @@ -2548,22 +2599,14 @@ mod tests { } fn test_docker_sandbox(docker: Docker, container_id: &str) -> DockerSandbox { - let sandbox = DockerSandbox { + let sandbox = DockerSandbox::with_docker_client( docker, - config: DockerSandboxOptions::default(), - github_app: None, - run_id: None, - clone_origin_url: None, - clone_branch: None, - container_id: OnceCell::new(), - repo_cloned: OnceCell::new(), - working_directory: OnceCell::new(), - origin_url: OnceCell::new(), - cached_platform: std::sync::OnceLock::new(), - cached_os_version: std::sync::OnceLock::new(), - rg_available: OnceCell::new(), - event_callback: None, - }; + DockerSandboxOptions::default(), + None, + None, + None, + None, + ); sandbox .container_id .set(container_id.to_string()) diff --git a/lib/components/fabro-sandbox/src/local.rs b/lib/components/fabro-sandbox/src/local.rs index 6af6c4e8d..937abdc64 100644 --- a/lib/components/fabro-sandbox/src/local.rs +++ b/lib/components/fabro-sandbox/src/local.rs @@ -836,6 +836,9 @@ impl Sandbox for LocalSandbox { } async fn activate(&self) -> crate::Result<()> { + // Local sandboxes have no provider resource that can stop or pause. + // Resume paths still call `start()` to recreate the directory and + // verify Bash. Ok(()) } diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index 90fdd9bcb..b29ccea39 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/lib/components/fabro-sandbox/src/sandbox.rs @@ -1134,11 +1134,12 @@ pub trait Sandbox: Send + Sync { remote_path: &str, ) -> crate::Result<()>; async fn initialize(&self) -> crate::Result<()>; - /// Ensure the sandbox is active and ready for ordinary operations. + /// Ensure the provider resource is running and not paused before access. /// /// This access-time operation must be idempotent. Providers that can stop /// independently should avoid restarting an already-active sandbox. This - /// method does not keep a sandbox active between calls. + /// lightweight check does not require the full health verification done by + /// [`Sandbox::start`], and it does not keep a sandbox active between calls. async fn activate(&self) -> crate::Result<()> { self.start().await } diff --git a/lib/components/fabro-sandbox/src/test_support.rs b/lib/components/fabro-sandbox/src/test_support.rs index a1f9e102b..65ff4af25 100644 --- a/lib/components/fabro-sandbox/src/test_support.rs +++ b/lib/components/fabro-sandbox/src/test_support.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use async_trait::async_trait; @@ -38,6 +39,8 @@ pub struct MockSandbox { pub captured_working_dirs: Mutex>>, /// Captures the `env_vars` argument from `exec_command` calls. pub captured_env_vars: Mutex>>, + pub active: AtomicBool, + pub activate_error: Option, pub activate_calls: Mutex, pub start_calls: Mutex, pub stop_calls: Mutex, @@ -52,6 +55,8 @@ pub struct MockSandbox { /// filtering. pub walk_files: Vec, pub walk_files_error: Option, + pub walk_files_called: AtomicBool, + pub walked_while_inactive: AtomicBool, /// Reported by `exec_command_streaming`. Set to `false` to model a /// provider that cannot separate stdout from stderr. pub streams_separated: bool, @@ -82,6 +87,14 @@ impl MockSandbox { *self.stop_calls.lock().expect("stop_calls lock poisoned") } + pub fn walk_files_was_called(&self) -> bool { + self.walk_files_called.load(Ordering::Relaxed) + } + + pub fn walked_while_inactive(&self) -> bool { + self.walked_while_inactive.load(Ordering::Relaxed) + } + pub fn delete_count(&self) -> u32 { *self .delete_calls @@ -107,6 +120,12 @@ impl MockSandbox { self.walk_files_error = Some(error.into()); self } + + #[must_use] + pub fn with_activate_error(mut self, error: impl Into) -> Self { + self.activate_error = Some(error.into()); + self + } } impl MockSandbox { @@ -140,6 +159,8 @@ impl Default for MockSandbox { captured_commands: Mutex::new(Vec::new()), captured_working_dirs: Mutex::new(Vec::new()), captured_env_vars: Mutex::new(None), + active: AtomicBool::new(true), + activate_error: None, activate_calls: Mutex::new(0), start_calls: Mutex::new(0), stop_calls: Mutex::new(0), @@ -150,6 +171,8 @@ impl Default for MockSandbox { exec_error: None, walk_files: Vec::new(), walk_files_error: None, + walk_files_called: AtomicBool::new(false), + walked_while_inactive: AtomicBool::new(false), streams_separated: true, } } @@ -376,6 +399,11 @@ impl Sandbox for MockSandbox { relative_start: &str, options: &WalkOptions, ) -> crate::Result> { + self.walk_files_called.store(true, Ordering::Relaxed); + if !self.active.load(Ordering::Relaxed) { + self.walked_while_inactive.store(true, Ordering::Relaxed); + return Err(crate::Error::message("Sandbox is stopped")); + } if let Some(error) = &self.walk_files_error { return Err(crate::Error::message(error.clone())); } @@ -439,6 +467,7 @@ impl Sandbox for MockSandbox { } async fn initialize(&self) -> crate::Result<()> { + self.active.store(true, Ordering::Relaxed); self.emit(SandboxEvent::Initializing { provider: "mock".into(), }); @@ -458,16 +487,24 @@ impl Sandbox for MockSandbox { .activate_calls .lock() .expect("activate_calls lock poisoned") += 1; + if let Some(error) = &self.activate_error { + return Err(crate::Error::context( + "Mock sandbox activation failed", + std::io::Error::other(error.clone()), + )); + } self.start().await } async fn start(&self) -> crate::Result<()> { *self.start_calls.lock().expect("start_calls lock poisoned") += 1; + self.active.store(true, Ordering::Relaxed); Ok(()) } async fn stop(&self) -> crate::Result<()> { *self.stop_calls.lock().expect("stop_calls lock poisoned") += 1; + self.active.store(false, Ordering::Relaxed); Ok(()) } diff --git a/lib/components/fabro-workflow/src/lifecycle/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index cf442f497..64a6323f2 100644 --- a/lib/components/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/components/fabro-workflow/src/lifecycle/mod.rs @@ -284,11 +284,10 @@ impl RunLifecycle for WorkflowLifecycle { } // A provider may auto-stop while the run is paused between nodes. self.sandbox.activate().await.map_err(|err| { - CoreError::Other(format!( - "failed to activate sandbox before node {:?}: {}", - node.id(), - err.display_with_causes() - )) + CoreError::context( + format!("failed to activate sandbox before node {}", node.id()), + err, + ) })?; if let Some(on_node) = &self.on_node { on_node(node.id()); @@ -343,11 +342,13 @@ impl RunLifecycle for WorkflowLifecycle { // Human, wait, and paused stages can return after a long period with // no sandbox traffic. Reactivate before artifact and checkpoint work. self.sandbox.activate().await.map_err(|err| { - CoreError::Other(format!( - "failed to activate sandbox after node attempt {:?}: {}", - ctx.node.id(), - err.display_with_causes() - )) + CoreError::context( + format!( + "failed to activate sandbox after node attempt {}", + ctx.node.id() + ), + err, + ) })?; self.artifact.after_attempt(ctx, state).await?; self.event.after_attempt(ctx, state).await?; diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index e89006ab0..a82fd7c8a 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -276,6 +276,13 @@ pub async fn execute(init: Initialized) -> Executed { Err(fabro_core::Error::Blocked { message }) => { (Err(Error::engine(message)), initial_context) } + Err(error @ fabro_core::Error::Context { .. }) => ( + Err(Error::engine_with_source( + "Pipeline lifecycle operation failed", + error, + )), + initial_context, + ), Err(e) => (Err(Error::engine(e.to_string())), initial_context), }; diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 6807277e8..5faa9c1ac 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -649,8 +649,7 @@ impl HandlerTrait for SlowHandler { } struct StopsSandboxHandler { - sandbox: Arc, - observed_activation_count: Arc, + sandbox: Arc, } #[async_trait] @@ -663,8 +662,6 @@ impl HandlerTrait for StopsSandboxHandler { _run_dir: &Path, _services: &crate::handler::EngineServices, ) -> std::result::Result { - self.observed_activation_count - .store(self.sandbox.activate_count(), Ordering::Relaxed); self.sandbox .stop() .await @@ -828,36 +825,63 @@ async fn execute_runs_simple_workflow() { assert_eq!(outcome.status, StageOutcome::Succeeded); } +#[tokio::test] +async fn execute_preserves_sandbox_activation_error_chain() { + let dir = tempfile::tempdir().unwrap(); + let sandbox: Arc = + Arc::new(MockSandbox::linux().with_activate_error("provider unavailable")); + + let error = run_graph( + make_registry(), + test_emitter_arc("test-run"), + sandbox, + &simple_graph(), + &test_run_options(dir.path(), "test-run"), + ) + .await + .expect_err("sandbox activation should fail"); + + assert_eq!(error.causes(), vec![ + "failed to activate sandbox before node start", + "Mock sandbox activation failed", + "provider unavailable", + ]); +} + #[tokio::test] async fn execute_reactivates_sandbox_after_a_stage_can_leave_it_stopped() { let dir = tempfile::tempdir().unwrap(); let sandbox = Arc::new(MockSandbox::linux()); - let observed_activation_count = Arc::new(AtomicU32::new(0)); let mut registry = make_registry(); registry.register( "start", Box::new(StopsSandboxHandler { - sandbox: Arc::clone(&sandbox), - observed_activation_count: Arc::clone(&observed_activation_count), + sandbox: Arc::clone(&sandbox), }), ); let sandbox_for_run: Arc = sandbox.clone(); + let mut run_options = test_run_options(dir.path(), "test-run"); + run_options + .settings + .run + .artifacts + .include + .push("**/*".to_string()); let outcome = run_graph( registry, test_emitter_arc("test-run"), sandbox_for_run, &simple_graph(), - &test_run_options(dir.path(), "test-run"), + &run_options, ) .await .unwrap(); assert_eq!(outcome.status, StageOutcome::Succeeded); - assert_eq!(observed_activation_count.load(Ordering::Relaxed), 1); assert_eq!(sandbox.stop_count(), 1); - assert_eq!(sandbox.activate_count(), 2); - assert_eq!(sandbox.start_count(), 2); + assert!(sandbox.walk_files_was_called()); + assert!(!sandbox.walked_while_inactive()); } #[tokio::test] diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 149a9ee59..0f02db0c2 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -392,6 +392,8 @@ pub async fn initialize( }); if attach_existing { + // Resume needs the full provider health check. `activate()` is the + // lighter access-time operation used after a run is already active. sandbox .start() .await diff --git a/lib/foundation/fabro-core/src/error.rs b/lib/foundation/fabro-core/src/error.rs index ea99e87d9..fc4bfbdeb 100644 --- a/lib/foundation/fabro-core/src/error.rs +++ b/lib/foundation/fabro-core/src/error.rs @@ -55,6 +55,12 @@ pub enum Error { StallTimeout { node_id: String }, #[error("{detail}")] Handler { detail: Box }, + #[error("{message}")] + Context { + message: String, + #[source] + source: Box, + }, #[error("{0}")] Other(String), } @@ -72,6 +78,16 @@ impl Error { } } + pub fn context( + message: impl Into, + source: impl std::error::Error + Send + Sync + 'static, + ) -> Self { + Self::Context { + message: message.into(), + source: Box::new(source), + } + } + pub fn is_retryable(&self) -> bool { matches!(self, Self::Handler { detail } if detail.retryable) } @@ -94,6 +110,8 @@ pub type Result = std::result::Result; #[cfg(test)] mod tests { + use std::error::Error as _; + use super::*; use crate::outcome::FailureCategory; @@ -153,6 +171,20 @@ mod tests { assert!(!not_retryable.is_retryable()); } + #[test] + fn core_error_context_preserves_source() { + let error = Error::context( + "failed to activate sandbox", + std::io::Error::other("provider unavailable"), + ); + + assert_eq!(error.to_string(), "failed to activate sandbox"); + assert_eq!( + error.source().map(ToString::to_string).as_deref(), + Some("provider unavailable") + ); + } + #[test] fn core_error_handler_to_fail_outcome() { let err = Error::handler(HandlerErrorDetail {