scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(sandbox): harden activation recovery

Browse source
This commit is contained in:
Release Repro 2026-07-28 15:38:35 -04:00
parent 80f321073f
commit e21e6bcdf9
No known key found for this signature in database
11 changed files with 375 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
lib/apps/fabro-server/src/server/handler/sessions.rs
View file

@ -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<dyn fabro_agent::Sandbox> = Arc::from(sandbox);
// No optional web-tool dependencies: `AskFabroToolAccessPolicy` denies
// `web_search` and `web_fetch`, and both `tools()` and the prompt are

92
lib/components/fabro-sandbox/src/daytona/mod.rs
View file

@ -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]

267
lib/components/fabro-sandbox/src/docker.rs
View file

@ -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<String>,
) -> crate::Result<Self> {
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<GitHubCredentials>,
run_id: Option<RunId>,
clone_origin_url: Option<String>,
clone_branch: Option<String>,
) -> 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<HashMap<String, String>> {
let inspect = self
.docker
async fn inspect_container(
&self,
container_id: &str,
) -> crate::Result<ContainerInspectResponse> {
self.docker
.inspect_container(container_id, None::<InspectContainerOptions>)
.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<HashMap<String, String>> {
let inspect = self.inspect_container(container_id).await?;
Ok(container_labels(&inspect))
}
async fn ensure_name_available(&self) -> crate::Result<Option<String>> {
@ -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<String, String>,
action: ContainerStartAction,
) -> crate::Result<()> {
let result = match action {
ContainerStartAction::Start => {
self.docker
.start_container(container_id, None::<StartContainerOptions<String>>)
.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<String, String>,
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<String, String> {
inspect
.config
.as_ref()
.and_then(|config| config.labels.clone())
.unwrap_or_default()
}
fn activation_action(inspect: &ContainerInspectResponse) -> Option<ContainerStartAction> {
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::<StartContainerOptions<String>>)
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::<InspectContainerOptions>)
.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())

3
lib/components/fabro-sandbox/src/local.rs
View file

@ -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(())
}

5
lib/components/fabro-sandbox/src/sandbox.rs
View file

@ -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
}

37
lib/components/fabro-sandbox/src/test_support.rs
View file

@ -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<Vec<Option<String>>>,
/// Captures the `env_vars` argument from `exec_command` calls.
pub captured_env_vars: Mutex<Option<HashMap<String, String>>>,
pub active: AtomicBool,
pub activate_error: Option<String>,
pub activate_calls: Mutex<u32>,
pub start_calls: Mutex<u32>,
pub stop_calls: Mutex<u32>,
@ -52,6 +55,8 @@ pub struct MockSandbox {
/// filtering.
pub walk_files: Vec<SandboxFile>,
pub walk_files_error: Option<String>,
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<String>) -> 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<Vec<SandboxFile>> {
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(())
}

21
lib/components/fabro-workflow/src/lifecycle/mod.rs
View file

@ -284,11 +284,10 @@ impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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?;

7
lib/components/fabro-workflow/src/pipeline/execute.rs
View file

@ -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),
};

46
lib/components/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -649,8 +649,7 @@ impl HandlerTrait for SlowHandler {
}
struct StopsSandboxHandler {
sandbox: Arc<MockSandbox>,
observed_activation_count: Arc<AtomicU32>,
sandbox: Arc<MockSandbox>,
}
#[async_trait]
@ -663,8 +662,6 @@ impl HandlerTrait for StopsSandboxHandler {
_run_dir: &Path,
_services: &crate::handler::EngineServices,
) -> std::result::Result<Outcome, Error> {
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<dyn Sandbox> =
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<dyn Sandbox> = 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]

2
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

@ -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

32
lib/foundation/fabro-core/src/error.rs
View file

@ -55,6 +55,12 @@ pub enum Error {
StallTimeout { node_id: String },
#[error("{detail}")]
Handler { detail: Box<HandlerErrorDetail> },
#[error("{message}")]
Context {
message: String,
#[source]
source: Box<dyn std::error::Error + Send + Sync + 'static>,
},
#[error("{0}")]
Other(String),
}
@ -72,6 +78,16 @@ impl Error {
}
}
pub fn context(
message: impl Into<String>,
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<T> = std::result::Result<T, Error>;
#[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 {