diff --git a/.config/nextest.toml b/.config/nextest.toml index 65282c997..fd4faaf7a 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -3,6 +3,11 @@ slow-timeout = { period = "1s", terminate-after = 3 } leak-timeout = "500ms" + # Real Docker worker crashes wait for supervisor cleanup and retain files. + [[profile.default.overrides]] + filter = "package(fabro-cli) & test(worker_crash_honors_the_docker_stop_policy_and_retains_files)" + slow-timeout = { period = "30s", terminate-after = 3 } + [[profile.default.overrides]] filter = "package(fabro-cli)" slow-timeout = { period = "6s", terminate-after = 4 } diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs b/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs index 108c0c13c..4d13401c7 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs @@ -509,3 +509,87 @@ async fn docker_failed_cleanup_preserves_success_across_server_restarts() { .await; server.shutdown(); } + +/// Remove the disposable run's container even if its regression fails. +struct WorkerCrashContainer(String); + +impl Drop for WorkerCrashContainer { + fn drop(&mut self) { + cleanup(&self.0); + } +} + +/// A worker-only crash leaves the server responsible for terminal cleanup. +/// The keep-running control must not be stopped by that supervisor backstop. +#[tokio::test(flavor = "multi_thread")] +async fn worker_crash_honors_the_docker_stop_policy_and_retains_files() { + if !fabro_test::docker_available() { + return; + } + for stop_on_terminal in [true, false] { + let context = test_context!(); + let server = docker_server().await; + let workspace = write_petri_workflow( + &context, + "digraph Cleanup { start [shape=Mdiamond]; exit [shape=Msquare]; \ + work [shape=parallelogram, script=\"printf retained > retained.txt\"]; \ + gate [shape=hexagon, label=\"Continue?\"]; \ + start -> work -> gate; gate -> exit [label=\"[A] Continue\"]; }", + ); + // Workflow configuration is public input, above the catalog default. + let config = workspace.join("workflow.toml"); + let text = std::fs::read_to_string(&config).unwrap(); + std::fs::write( + &config, + format!("{text}\n[run.environment.lifecycle]\nstop_on_terminal = {stop_on_terminal}\n"), + ) + .unwrap(); + let run_id = run_detached_in(&context, &server, &workspace, ENVIRONMENT, &[]); + let _container_cleanup = WorkerCrashContainer(run_id.clone()); + let gate_deadline = std::time::Instant::now() + std::time::Duration::from_mins(1); + while super::petri::questions(&server, &run_id).await.is_empty() { + let run = run_json(&server, &format!("runs/{run_id}")).await; + assert!( + run["lifecycle"]["status"]["kind"] != "failed" + && run["lifecycle"]["status"]["kind"] != "succeeded" + && std::time::Instant::now() < gate_deadline, + "the fixture did not reach its human gate: {run}" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + let container = container_of(&run_id).expect("the gated run has a container"); + let worker = wait_for_worker(&run_id); + fabro_proc::sigkill(worker); + wait_for_status(&server, &run_id, &["failed"]).await; + + // Usage is counted after the supervisor finishes handling the exit, + // so a keep-running assertion cannot race a pending cleanup task. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(35); + while run_json(&server, "usage").await["totals"]["runs"].as_u64() != Some(1) { + assert!( + std::time::Instant::now() < deadline, + "worker exit did not settle" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + let output = Command::new("docker") + .args(["inspect", "--format", "{{.State.Running}}", &container]) + .output() + .expect("inspect runs"); + assert!(output.status.success(), "the container is retained"); + let running = String::from_utf8_lossy(&output.stdout).trim() == "true"; + // docker cp also reads a stopped container without activating it. + let retained = context.temp_dir.join("retained.txt"); + let copied = Command::new("docker") + .args(["cp", &format!("{container}:/workspace/retained.txt")]) + .arg(&retained) + .status() + .expect("docker cp runs"); + let contents = std::fs::read_to_string(&retained); + cleanup(&run_id); + server.shutdown(); + assert_eq!(running, !stop_on_terminal); + assert!(copied.success()); + assert_eq!(contents.unwrap(), "retained"); + } +} diff --git a/lib/apps/fabro-server/src/sandbox_access.rs b/lib/apps/fabro-server/src/sandbox_access.rs index 9c07adcaf..fe82c7ad6 100644 --- a/lib/apps/fabro-server/src/sandbox_access.rs +++ b/lib/apps/fabro-server/src/sandbox_access.rs @@ -34,6 +34,7 @@ use anyhow::Context as _; use fabro_config::Storage; use fabro_petri::providers; pub(crate) use fabro_petri::providers::DaytonaCredentials; +use fabro_types::settings::run::RunEnvironmentSettings; use fabro_types::settings::server::{ SandboxPluginSettings, ServerSandboxProviderSettings, ServerSandboxProvidersSettings, }; @@ -270,6 +271,60 @@ async fn attach_run_sandbox_on( .with_context(|| format!("Failed to reconnect {kind} sandbox '{sandbox_id}'")) } +/// Stop the sandboxes an exited worker can no longer release. This is a +/// supervisor backstop, not a deletion: the workspace and Petri's lease +/// records remain available for inspection, recovery and later pruning. +/// Local process cleanup belongs to the Host provider's worker guardian. +pub(crate) async fn stop_run_sandboxes( + access: &ProviderAccess, + environment: &RunEnvironmentSettings, + run_id: RunId, +) -> anyhow::Result<()> { + if !environment.lifecycle.stop_on_terminal + || environment.provider.bundled() == Some(BundledProvider::Local) + { + return Ok(()); + } + let provider = connect_provider(&environment.provider, access).await?; + stop_run_sandboxes_on(provider, run_id).await +} + +async fn stop_run_sandboxes_on( + provider: Arc, + run_id: RunId, +) -> anyhow::Result<()> { + // Listing also finds a container whose create completed just before the + // worker died without recording its id, and every nested scope. Both + // list and attach verify ownership, so unrelated resources stay alone. + let provider = OwnedProvider::new(provider, run_ownership(run_id)); + let sandboxes = provider + .list(&SandboxFilter::default()) + .await + .context("Failed to list the exited worker's sandboxes")?; + let results = + join_all(sandboxes.into_iter().map(|status| { + let provider = &provider; + async move { + let result = match provider.attach(&status.id, None).await { + Ok(sandbox) => sandbox.stop().await, + Err(error) => Err(error), + }; + match result { + // A concurrent run deletion already cleaned it up. + Ok(()) | Err(DriverError::NotFound { .. }) => Ok(()), + Err(error) => Err(anyhow::Error::new(error) + .context("Failed to stop an exited worker's sandbox")), + } + } + })) + .await; + // Every owned sandbox gets its stop attempt even if one fails. + for result in results { + result?; + } + Ok(()) +} + /// Attaches to a run's host sandbox: the managed directory the record /// names, with the record the run's worker kept for it. /// @@ -861,6 +916,54 @@ mod tests { assert!(ownership.owns(&labels)); } + #[tokio::test] + async fn worker_cleanup_stops_all_owned_sandboxes_without_deleting_their_data() { + let run_id = RunId::new(); + let mine = petri_scripted_sandbox("root", &run_id.to_string()); + let nested = petri_scripted_sandbox("nested", &run_id.to_string()); + let theirs = petri_scripted_sandbox("foreign", &RunId::new().to_string()); + let unlabelled = Arc::new(ScriptedSandbox::new()); + mine.fs().write("retained", b"keep me").await.unwrap(); + let provider = scripted_provider("docker", vec![ + Arc::clone(&mine), + Arc::clone(&nested), + Arc::clone(&theirs), + Arc::clone(&unlabelled), + ]); + + stop_run_sandboxes_on(provider, run_id).await.unwrap(); + + for sandbox in [&mine, &nested] { + assert_eq!(sandbox.current_state(), SandboxState::Stopped); + assert_eq!(sandbox.delete_count(), 0); + assert_eq!(sandbox.start_count(), 0); + } + assert_eq!(mine.fs().read("retained").await.unwrap(), b"keep me"); + for sandbox in [&theirs, &unlabelled] { + assert_eq!(sandbox.current_state(), SandboxState::Running); + assert_eq!(sandbox.stop_count(), 0); + assert_eq!(sandbox.delete_count(), 0); + } + } + + #[tokio::test] + async fn worker_cleanup_skips_keep_running_and_host_environments() { + // This plugin cannot connect: skipping the policy must happen first. + let provider = kind("unreachable-cleanup"); + let access = unreachable_plugin_access(provider.as_str()); + let mut environment = RunEnvironmentSettings { + provider, + ..RunEnvironmentSettings::default() + }; + environment.lifecycle.stop_on_terminal = false; + stop_run_sandboxes(&access, &environment, RunId::new()) + .await + .unwrap(); + stop_run_sandboxes(&access, &RunEnvironmentSettings::default(), RunId::new()) + .await + .unwrap(); + } + #[tokio::test] async fn a_scoped_provider_attaches_only_to_the_runs_sandbox() { let run_id = RunId::new(); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 27164f760..413c8c608 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -4417,6 +4417,32 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { } }; + // Normal release is Petri's. An abnormal exit can leave its scopes + // running forever; the surviving supervisor must enforce the terminal + // stop policy after deciding that this execution will not be resumed. + if !worker_exit.succeeded() && final_state.status.is_terminal() { + let cleanup = async { + let access = state.provider_access().await?; + sandbox_access::stop_run_sandboxes( + &access, + &final_state.spec.settings.run.environment, + run_id, + ) + .await + }; + match timeout(Duration::from_secs(30), cleanup).await { + Ok(Ok(())) => {} + Ok(Err(error)) => tracing::warn!( + run_id = %run_id, error = ?error, + "Failed to stop sandboxes after the worker exited" + ), + Err(error) => tracing::warn!( + run_id = %run_id, error = ?error, + "Timed out stopping sandboxes after the worker exited" + ), + } + } + accumulate_concluded_run_usage(&state, &final_state); let mut runs = state.runs.lock().expect("runs lock poisoned");