Preserve completed run results when sandbox cleanup fails

This commit is contained in:
Scott Werner 2026-10-12 11:39:52 -04:00
parent 857ea59cca
commit a2ca338514
6 changed files with 424 additions and 10 deletions

View file

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

View file

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

View file

@ -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::<Vec<_>>()
.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,

View file

@ -10900,3 +10900,5 @@ fn an_unsettled_run_takes_the_stores_status_or_a_termination_at_worker_exit() {
}
mod artifact_storage;
mod deletion;

View file

@ -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<AppState>,
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::<RunId>().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(&registry, &saved).unwrap();
std::fs::write(&registry, "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(&registry).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();
}

View file

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