Merge pull request #1038 from fabro-sh/codex/docker-worker-cleanup

Stop retained sandboxes after abnormal worker exit

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

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");
+    }
+}

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<dyn SandboxProvider>,
+    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();

lib/apps/fabro-server/src/server.rs

@@ -4417,6 +4417,32 @@ async fn execute_run_subprocess(state: Arc<AppState>, 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");