diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri.rs b/lib/apps/fabro-cli/tests/it/scenario/petri.rs index 0662cf70f..8e5fc4337 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri.rs @@ -130,6 +130,12 @@ impl RunningServer { /// Start the server process over this storage; the same call brings /// it back after a kill. pub(super) async fn launch(&mut self) { + self.launch_with_env(&[]).await; + } + + /// Relaunch with overrides applied only to this server process. The next + /// ordinary launch returns to the original environment. + pub(super) async fn launch_with_env(&mut self, env: &[(&str, &str)]) { assert!(self.child.is_none(), "the server is already running"); let mut cmd = Command::new(env!("CARGO_BIN_EXE_fabro")); apply_test_isolation(&mut cmd, self.home_root.path()); @@ -144,6 +150,9 @@ impl RunningServer { for (name, value) in &self.env { cmd.env(name, value); } + for (name, value) in env { + cmd.env(name, value); + } cmd.args(["server", "start", "--foreground"]) .arg("--storage-dir") .arg(&self.storage_dir) 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 a27da97dd..108c0c13c 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs @@ -26,8 +26,8 @@ use fabro_test::{ use serde_json::json; use super::petri::{ - RunningServer, crash, run_detached_in, wait_for_status, wait_for_success, wait_for_worker, - write_petri_workflow, + RunningServer, crash, run_detached_in, run_json, wait_for_status, wait_for_success, + wait_for_worker, write_petri_workflow, }; use crate::support::TEST_DEV_TOKEN; @@ -424,3 +424,88 @@ async fn an_ask_fabro_turn_reads_a_file_inside_the_runs_container() { cleanup(&run_id); server.shutdown(); } + +/// An unavailable daemon refuses cleanup without changing a completed run, +/// even after repeated requests and server restarts. `DOCKER_HOST` is +/// overridden only on the disposable server, so other tests and the real daemon +/// are intact. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "requires Docker; creates a container and restarts a disposable server"] +async fn docker_failed_cleanup_preserves_success_across_server_restarts() { + assert!( + fabro_test::docker_available(), + "this explicit test requires Docker" + ); + let context = test_context!(); + let mut server = docker_server().await; + let workspace = hello_file_bundle(&context); + let run_id = run_detached_in(&context, &server, &workspace, ENVIRONMENT, &[ + "--auto-approve", + ]); + let _cleanup = scopeguard::guard(run_id.clone(), |id| cleanup(&id)); + wait_for_success(&server, &run_id).await; + let before = run_json(&server, &format!("runs/{run_id}/state")).await; + let container = container_of(&run_id).expect("the completed run retains its container"); + let absent_socket = format!("unix://{}", context.temp_dir.join("absent.sock").display()); + server.kill(); + server + .launch_with_env(&[(EnvVars::DOCKER_HOST, &absent_socket)]) + .await; + let client = fabro_test::test_http_client(); + for _ in 0..2 { + let response = client + .delete(format!("{}/api/v1/runs/{run_id}", server.api_base_url)) + .bearer_auth(TEST_DEV_TOKEN) + .send() + .await + .expect("the delete sends"); + fabro_test::expect_reqwest_status( + response, + fabro_http::StatusCode::CONFLICT, + "DELETE /api/v1/runs/{id} with unavailable Docker", + ) + .await; + let after = run_json(&server, &format!("runs/{run_id}/state")).await; + for field in ["status", "status_updated_at", "conclusion"] { + assert_eq!(after[field], before[field], "cleanup changed {field}"); + } + assert_eq!(container_of(&run_id).as_deref(), Some(container.as_str())); + } + server.kill(); + server.launch().await; + let after = run_json(&server, &format!("runs/{run_id}/state")).await; + for field in ["status", "status_updated_at", "conclusion"] { + assert_eq!(after[field], before[field], "restart changed {field}"); + } + assert_eq!(container_of(&run_id).as_deref(), Some(container.as_str())); + let response = client + .delete(format!("{}/api/v1/runs/{run_id}", server.api_base_url)) + .bearer_auth(TEST_DEV_TOKEN) + .send() + .await + .expect("the retry sends"); + fabro_test::expect_reqwest_status( + response, + fabro_http::StatusCode::NO_CONTENT, + "DELETE /api/v1/runs/{id} after restoring Docker", + ) + .await; + assert_eq!( + container_of(&run_id), + None, + "successful retry deletes the container" + ); + let response = client + .get(format!("{}/api/v1/runs/{run_id}", server.api_base_url)) + .bearer_auth(TEST_DEV_TOKEN) + .send() + .await + .expect("the run read sends"); + fabro_test::expect_reqwest_status( + response, + fabro_http::StatusCode::NOT_FOUND, + "GET /api/v1/runs/{id} after successful deletion", + ) + .await; + server.shutdown(); +} diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index c0069200d..e1dbdf4f5 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -2885,9 +2885,10 @@ async fn delete_run_sandbox_resource( )); } }; - let delete_started = matches!(projection.status, RunStatus::Removing); - let can_mark_removing = projection.status.can_transition_to(RunStatus::Removing); - if !delete_started && can_mark_removing { + // Deleting resources must not reopen a completed workflow. A failed prune + // leaves its terminal result intact, including across a server restart. + if !projection.status.is_terminal() && projection.status.can_transition_to(RunStatus::Removing) + { run_records::lifecycle( state, id, @@ -2934,9 +2935,9 @@ async fn delete_run_sandbox_resource( .join("petri"); let sandbox = match state.load_sandbox_provider_config(&record.provider).await { Ok(sandbox) => sandbox, - // A forced or restarted delete goes on without the sandboxes, as it + // An explicitly forced delete goes on without the sandboxes, as it // does for any other prune failure below. - Err(err) if force || delete_started => { + Err(err) if force => { tracing::warn!( run_id = %id, provider = %record.provider, @@ -2972,7 +2973,7 @@ async fn delete_run_sandbox_resource( Ok(SandboxDeleteOutcome::Cleaned) } // A lease Petri could not prune keeps its pending intent, so a - // later prune tries again; a forced or restarted delete goes on + // later prune tries again; an explicitly forced delete goes on // without it. Ok(report) => { let problems = report @@ -2981,7 +2982,7 @@ async fn delete_run_sandbox_resource( .map(|(lease, problem)| format!("lease {lease}: {problem}")) .collect::>() .join("; "); - if force || delete_started { + if force { tracing::warn!( run_id = %id, provider = %record.provider, @@ -3012,7 +3013,7 @@ async fn delete_run_sandbox_resource( } Err(error) => { let message = collect_chain(&error).join(": "); - if force || delete_started { + if force { tracing::warn!( run_id = %id, error = %message, diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index baaa5351e..a1f978590 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -10900,3 +10900,5 @@ fn an_unsettled_run_takes_the_stores_status_or_a_termination_at_worker_exit() { } mod artifact_storage; + +mod deletion; diff --git a/lib/apps/fabro-server/src/server/tests/deletion.rs b/lib/apps/fabro-server/src/server/tests/deletion.rs new file mode 100644 index 000000000..66c53fb9e --- /dev/null +++ b/lib/apps/fabro-server/src/server/tests/deletion.rs @@ -0,0 +1,310 @@ +//! Cleanup failures leave completed workflow results intact. These tests run +//! real Host scopes and temporarily obstruct the provider registry to exercise +//! Petri's prune error path without Docker or process-wide environment changes. + +use fabro_petri::petri::RunKey; +use fabro_types::RunProjection; + +use super::*; + +struct DeletionRun { + state: Arc, + app: Router, + id: RunId, + vault_path: PathBuf, + _source: tempfile::TempDir, +} + +fn deletion_state_builder(vault_path: PathBuf) -> TestAppStateBuilder { + TestAppStateBuilder::new() + .vault_path(vault_path) + .default_environment_provider(Some(SandboxProviderKind::LOCAL)) + .runtime_settings( + default_test_server_settings(), + manifest_run_defaults_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n"), + ) +} + +async fn completed_deletion_run(script: &str, cancel: bool) -> DeletionRun { + let vault_path = test_secret_store_path(); + let state = deletion_state_builder(vault_path.clone()) + .in_process_execution() + .build(); + let app = build_test_router(Arc::clone(&state)); + spawn_scheduler(Arc::clone(&state)); + let workflow = format!( + r#"digraph Cleanup {{ + graph [goal="Finish the required command", default_max_retries=0] + start [shape=Mdiamond] + work [shape=parallelogram, script="{script}", goal_gate=true, max_retries=0] + exit [shape=Msquare] + start -> work -> exit +}}"#, + ); + let source = tempfile::tempdir().unwrap(); + let mut intent = test_intent(&app, &workflow).await; + intent["target"] = json!({ "kind": "folder", "path": source.path() }); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(api("/runs")) + .header("content-type", "application/json") + .body(Body::from(intent.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let created = response_json!(response, StatusCode::CREATED).await; + let id = created["id"].as_str().unwrap().parse::().unwrap(); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{id}/start"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(response, StatusCode::OK).await; + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let projection = run_records::require_projection(&state, id).await.unwrap(); + if projection + .sandbox + .as_ref() + .and_then(fabro_types::RunSandbox::instance) + .is_some() + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the run acquires a Host scope"); + if cancel { + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{id}/cancel"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(response, StatusCode::ACCEPTED).await; + } + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let projection = run_records::require_projection(&state, id).await.unwrap(); + let key = RunKey::new(id.to_string()); + if projection.status.is_terminal() + && state + .petri_runs + .store() + .owner(&key) + .await + .unwrap() + .is_none() + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the run finishes and releases its lease"); + DeletionRun { + state, + app, + id, + vault_path, + _source: source, + } +} + +fn obstruct_host_registry(run: &DeletionRun) -> (PathBuf, PathBuf) { + let petri_dir = Storage::new(run.state.server_storage_dir()) + .run_scratch(&run.id) + .root() + .join("petri"); + let registry = petri_dir.join("host-registry"); + let saved = petri_dir.join("saved-host-registry"); + // Fault injection belongs in this server unit test: all workflow state was + // created through the API, and only the temporary provider filesystem is + // obstructed. The managed workspace remains available for a later retry. + std::fs::rename(®istry, &saved).unwrap(); + std::fs::write(®istry, "provider unavailable").unwrap(); + (registry, saved) +} + +async fn deletion_response(app: &Router, id: RunId, force: bool) -> Response { + app.clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(api(&format!("/runs/{id}?force={force}"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap() +} + +async fn assert_completed_result(app: &Router, id: RunId, before: &RunProjection) { + let response = app + .clone() + .oneshot( + Request::builder() + .uri(api(&format!("/runs/{id}/state"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let after = response_json!(response, StatusCode::OK).await; + assert_eq!( + after["status"], + serde_json::to_value(before.status).unwrap() + ); + assert_eq!( + after["status_updated_at"], + serde_json::to_value(before.status_updated_at).unwrap() + ); + assert_eq!( + after["conclusion"], + serde_json::to_value(&before.conclusion).unwrap() + ); + let response = app + .clone() + .oneshot( + Request::builder() + .uri(api("/runs")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let list = response_json!(response, StatusCode::OK).await; + assert!( + list["data"] + .as_array() + .unwrap() + .iter() + .any(|run| run_json_id(run) == Some(id.to_string().as_str())), + "failed cleanup must not hide the run from the default list" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_cleanup_preserves_completed_results_across_retries_and_restart() { + for (script, cancel, expected) in [ + ("echo completed", false, RunStatus::Succeeded { + reason: SuccessReason::Completed, + }), + ("exit 1", false, RunStatus::Failed { + reason: FailureReason::WorkflowError, + }), + ("sleep 30", true, RunStatus::Failed { + reason: FailureReason::Cancelled, + }), + ] { + let run = completed_deletion_run(script, cancel).await; + let before = run_records::require_projection(&run.state, run.id) + .await + .unwrap(); + assert_eq!(before.status, expected); + let workspace = PathBuf::from( + &before + .sandbox + .as_ref() + .unwrap() + .instance() + .unwrap() + .runtime + .working_directory, + ); + let (registry, saved) = obstruct_host_registry(&run); + for _ in 0..2 { + let response = deletion_response(&run.app, run.id, false).await; + let error = response_json!(response, StatusCode::CONFLICT).await; + assert!( + error["errors"][0]["detail"] + .as_str() + .unwrap() + .contains("Failed to delete the run's sandboxes") + ); + assert!(workspace.is_dir()); + assert_completed_result(&run.app, run.id, &before).await; + } + run.state.shutdown.cancel(); + let after = deletion_state_builder(run.vault_path.clone()) + .store_bundle( + Arc::clone(&run.state.stores.runs), + run.state.artifact_store.clone(), + ) + .build(); + let app = build_test_router(Arc::clone(&after)); + assert_eq!( + reconcile_incomplete_runs_on_startup(&after).await.unwrap(), + 0 + ); + assert_completed_result(&app, run.id, &before).await; + let response = deletion_response(&app, run.id, false).await; + assert_status!(response, StatusCode::CONFLICT).await; + assert_completed_result(&app, run.id, &before).await; + std::fs::remove_file(®istry).unwrap(); + std::fs::rename(saved, registry).unwrap(); + let response = deletion_response(&app, run.id, false).await; + assert_status!(response, StatusCode::NO_CONTENT).await; + assert!(!workspace.exists()); + let response = app + .oneshot( + Request::builder() + .uri(api(&format!("/runs/{}", run.id))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(response, StatusCode::NOT_FOUND).await; + after.shutdown.cancel(); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn only_explicit_force_bypasses_cleanup_failure_for_a_removing_run() { + let run = completed_deletion_run("echo completed", false).await; + obstruct_host_registry(&run); + // Exercise the existing Removing path independently: it must not make an + // ordinary delete bypass a provider failure. No historical repair is done. + run_records::lifecycle( + &run.state, + run.id, + run_records::transition(RunLifecycleKind::Removing, RunStatus::Removing), + ) + .await + .unwrap(); + let response = deletion_response(&run.app, run.id, false).await; + assert_status!(response, StatusCode::CONFLICT).await; + let response = deletion_response(&run.app, run.id, true).await; + assert_status!(response, StatusCode::NO_CONTENT).await; + let response = run + .app + .oneshot( + Request::builder() + .uri(api(&format!("/runs/{}", run.id))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(response, StatusCode::NOT_FOUND).await; + run.state.shutdown.cancel(); +} diff --git a/lib/apps/fabro-server/tests/it/scenario/petri.rs b/lib/apps/fabro-server/tests/it/scenario/petri.rs index 03b28cbeb..026491890 100644 --- a/lib/apps/fabro-server/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-server/tests/it/scenario/petri.rs @@ -753,6 +753,13 @@ async fn deleting_a_run_prunes_its_host_workspace_through_petri() { working_directory.is_dir(), "the refused delete left the workspace" ); + let after_refusal = settled_state(&state, &app, &run_id).await; + for field in ["status", "status_updated_at", "conclusion"] { + assert_eq!( + after_refusal[field], projection[field], + "refused cleanup changed {field}" + ); + } drop(held); wait_for_free_lease(store, &key).await;