diff --git a/lib/apps/fabro-cli/src/commands/run/attach.rs b/lib/apps/fabro-cli/src/commands/run/attach.rs index e8e608a60..40a0b4402 100644 --- a/lib/apps/fabro-cli/src/commands/run/attach.rs +++ b/lib/apps/fabro-cli/src/commands/run/attach.rs @@ -1034,6 +1034,7 @@ mod tests { let server = MockServer::start_async().await; let mut state = terminal_run_state_response(run_id); state["status"] = serde_json::json!({"kind": "running"}); + state["conclusion"] = serde_json::Value::Null; let state: server_client::RunProjection = serde_json::from_value(state).unwrap(); let executed = serde_json::json!({ "run_id": run_id, "stream_seq": 1, "kind": "petri", "id": "executed", @@ -1097,7 +1098,6 @@ mod tests { !task.is_finished(), "successful execution does not end attach while publication is pending" ); - waiting.delete_async().await; let finished = serde_json::json!({ "run_id": run_id, "stream_seq": 2, "kind": "petri", "id": "finished", "recorded_at": 2000, @@ -1114,6 +1114,7 @@ mod tests { .body(format!("data: {finished}\n\n")); }) .await; + waiting.delete_async().await; assert_eq!( tokio::time::timeout(Duration::from_secs(5), task) .await diff --git a/lib/apps/fabro-server/src/petri_runs.rs b/lib/apps/fabro-server/src/petri_runs.rs index eb3e19e93..c1eb6d018 100644 --- a/lib/apps/fabro-server/src/petri_runs.rs +++ b/lib/apps/fabro-server/src/petri_runs.rs @@ -1086,6 +1086,51 @@ mod tests { } } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_rejected_platform_failure_does_not_settle_the_run() { + let runtime = Arc::new(HeldWorkerRuntime::default()); + let (state, app, run_id, token) = held_worker_run(&runtime).await; + run_to_running_as_worker(&app, run_id, &token).await; + let pool = state.stores.run_summaries.pool(); + sqlx::query( + "CREATE TRIGGER reject_platform_record BEFORE INSERT ON platform_records \ + BEGIN SELECT RAISE(FAIL, 'scripted append failure'); END", + ) + .execute(&pool) + .await + .unwrap(); + let response = append_lifecycle_as_worker( + &app, + run_id, + &token, + RunLifecycleKind::Failed, + RunStatus::Failed { + reason: FailureReason::LaunchFailed, + }, + ) + .await; + fabro_test::assert_axum_status( + response, + StatusCode::INTERNAL_SERVER_ERROR, + "rejected platform terminal append", + ) + .await; + assert_public_result(&state, &app, run_id, RunStatus::Running, None).await; + crate::server::persist_run_failure( + &state, + run_id, + FailureReason::LaunchFailed, + "worker launch failed".into(), + ) + .await; + assert_public_result(&state, &app, run_id, RunStatus::Running, None).await; + sqlx::query("DROP TRIGGER reject_platform_record") + .execute(&pool) + .await + .unwrap(); + runtime.end_worker(); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn required_publication_result_agrees_across_worker_api_projection_and_cleanup() { use fabro_petri::petri::LogId; @@ -1177,6 +1222,29 @@ mod tests { } assert_eq!(concluded_runs(&app).await, 1); assert_public_result(&state, &app, run_id, expected, rejection).await; + // A host error arriving after cleanup must neither append a + // competing terminal record nor replace the useful failure. + let platform_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM platform_records WHERE run_id = ?") + .bind(run_id.to_string()) + .fetch_one(&state.stores.run_summaries.pool()) + .await + .unwrap(); + crate::server::persist_run_failure( + &state, + run_id, + FailureReason::Terminated, + "worker wait failed during teardown".into(), + ) + .await; + assert_public_result(&state, &app, run_id, expected, rejection).await; + let after_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM platform_records WHERE run_id = ?") + .bind(run_id.to_string()) + .fetch_one(&state.stores.run_summaries.pool()) + .await + .unwrap(); + assert_eq!(after_count, platform_count); let (rebuilt, _, _) = fabro_petri::test_support::rebuild(&state.db_pool, &state.db_pool, run_id) .await diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index dbd5db12f..e36a73efd 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -3503,13 +3503,55 @@ async fn fail_run_before_execution( reason: FailureReason, message: String, ) { - if let Err(err) = - run_records::lifecycle(state, run_id, run_records::failed(reason, message.clone())).await - { - error!(run_id = %run_id, error = %err, "Failed to persist run failure status"); - } + persist_run_failure(state, run_id, reason, message).await; +} - fail_managed_run(state, run_id, reason, message); +/// Record a host failure only while no terminal result is committed. This +/// also handles a worker wait/launch error racing its durable Petri finish. +pub(crate) async fn persist_run_failure( + state: &Arc, + run_id: RunId, + reason: FailureReason, + message: String, +) { + match run_records::projection(state, run_id).await { + Ok(Some(projection)) if projection.status.is_terminal() => { + let failure = projection + .conclusion + .as_ref() + .and_then(|conclusion| conclusion.failure.as_ref()) + .map(|failure| failure.detail.message.clone()); + settle_managed_run_at_finish(state, run_id, projection.status, failure); + } + Ok(Some(_)) => { + match run_records::lifecycle( + state, + run_id, + run_records::failed(reason, message.clone()), + ) + .await + { + Ok(_) => fail_managed_run(state, run_id, reason, message), + Err(err) => { + error!(run_id = %run_id, error = %err, "Failed to persist run failure status"); + } + } + } + Ok(None) => { + error!(run_id = %run_id, "Run missing when recording a host failure"); + } + Err(err) => { + error!(run_id = %run_id, error = %err, "Failed to read the committed run result"); + } + } + // Resource cleanup is operational; it cannot substitute for a committed + // outcome if storage is unavailable. + let mut runs = state.runs.lock().expect("runs lock poisoned"); + if let Some(managed_run) = runs.get_mut(&run_id) { + clear_live_run_state(managed_run); + } + drop(runs); + cleanup_worker_control_bus_for_run(state, run_id); state.scheduler_notify.notify_one(); } @@ -3764,26 +3806,13 @@ async fn fail_worker_launch(state: &Arc, run_id: RunId, err: anyhow::E None } }; - let launch_message = format!("Failed to spawn worker: {err}"); let (error, reason) = failure_honoring_pending_cancel(pending_control, || { ( WorkflowError::engine_with_anyhow("Failed to spawn worker", err), FailureReason::LaunchFailed, ) }); - let message = if reason == FailureReason::Cancelled { - "Run cancelled before worker launch completed".to_string() - } else { - launch_message - }; - let _ = run_records::lifecycle( - state, - run_id, - run_records::failed(reason, error.to_string()), - ) - .await; - fail_managed_run(state, run_id, reason, message); - state.scheduler_notify.notify_one(); + persist_run_failure(state, run_id, reason, error.to_string()).await; } /// A worker that exited without recording the run's end left it failed. @@ -4249,15 +4278,17 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { Err(err) => { tracing::error!(run_id = %run_id, error = %err, "Failed while waiting on worker"); let message = format!("Worker wait failed: {err}"); + let superseded = { + let runs = state.runs.lock().expect("runs lock poisoned"); + runs.get(&run_id) + .is_some_and(|run| run.worker_ref.as_ref() != Some(&worker_ref)) + }; + if superseded { + return; + } state.worker_runtime.force_stop(&worker_ref).await; - let _ = run_records::lifecycle( - &state, - run_id, - run_records::failed(FailureReason::Terminated, message.clone()), - ) - .await; - fail_managed_run(&state, run_id, FailureReason::Terminated, message); - state.scheduler_notify.notify_one(); + state.petri_runs.worker_exited(run_id); + persist_run_failure(&state, run_id, FailureReason::Terminated, message).await; return; } }; diff --git a/lib/apps/fabro-server/src/server/petri_runs.rs b/lib/apps/fabro-server/src/server/petri_runs.rs index 122f575dd..25e9e931e 100644 --- a/lib/apps/fabro-server/src/server/petri_runs.rs +++ b/lib/apps/fabro-server/src/server/petri_runs.rs @@ -488,8 +488,8 @@ pub(crate) async fn execute(state: Arc, run_id: RunId) { run_id: run_id.to_string(), run_dir: run_dir.join("petri"), execution, - // The projector's signal follows each durable append; the managed - // run's settle at Petri's finish precedes it. + // The coordinator finish is stored before managed status settles; + // the projector also reads only durable records. store: Arc::new(SettlingStore { inner: state .petri_projector @@ -534,7 +534,7 @@ pub(crate) async fn execute(state: Arc, run_id: RunId) { } }; match run_records::lifecycle(&state, run_id, record).await { - Ok(()) => finish(&state, run_id, status, error), + Ok(_) => finish(&state, run_id, status, error), Err(err) => { error!(run_id = %run_id, error = %err, "Failed to persist run outcome"); release_live_state(&state, run_id); @@ -669,7 +669,7 @@ async fn fail_before_execution(state: &Arc, run_id: RunId, message: &s error!(run_id = %run_id, error = message, "Petri run cannot start"); let (status, error, record) = failed(FailureReason::WorkflowError, message.to_string()); match run_records::lifecycle(state, run_id, record).await { - Ok(()) => finish(state, run_id, status, error), + Ok(_) => finish(state, run_id, status, error), Err(err) => { error!(run_id = %run_id, error = %err, "Failed to persist run failure status"); release_live_state(state, run_id); @@ -679,9 +679,9 @@ async fn fail_before_execution(state: &Arc, run_id: RunId, message: &s /// Settle the managed run at its terminal record and release its /// scheduler slot. A run that Petri finished settled already, at the -/// `run.finished` record ([`SettlingStore`]); this refines its status and -/// error and ends its live state. A run deleted since is gone from the map -/// and stays gone. +/// `run.finished` record ([`SettlingStore`]); this preserves that status, +/// fills any missing failure detail and ends its live state. A run deleted +/// since is gone from the map and stays gone. fn finish(state: &Arc, run_id: RunId, status: RunStatus, error: Option) { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { diff --git a/lib/components/fabro-petri/src/engine.rs b/lib/components/fabro-petri/src/engine.rs index 3035b8892..d620af722 100644 --- a/lib/components/fabro-petri/src/engine.rs +++ b/lib/components/fabro-petri/src/engine.rs @@ -137,7 +137,7 @@ pub enum RunStatus { #[derive(Clone, Debug)] pub struct RunOutcome { pub status: RunStatus, - /// The root invocation's failure message, when it failed. + /// Required-finalization failure detail, or the root execution's failure. pub failure: Option, /// Whether the record is whole: the run recorded its finish and every /// log replays byte for byte. @@ -365,8 +365,8 @@ pub async fn outcome_of(store: &dyn RunStore, run_id: &str) -> Result) -> Conclusion { match result { diff --git a/lib/components/fabro-petri/src/hooks.rs b/lib/components/fabro-petri/src/hooks.rs index 8148ce7d9..9cedffee4 100644 --- a/lib/components/fabro-petri/src/hooks.rs +++ b/lib/components/fabro-petri/src/hooks.rs @@ -53,16 +53,18 @@ //! //! # Operation identities //! -//! Every external effect here is keyed on `(run key, execution, DecisionId, -//! effect kind)` from the hook context and deduplicated on retry: the -//! checkpoint's key is the attempt's decision in its execution, effect +//! Checkpoint and artifact effects are keyed on `(run key, execution, +//! DecisionId, effect kind)` from the hook context and deduplicated on retry: +//! the checkpoint's key is the attempt's decision in its execution, effect //! `checkpoint`; an artifact's is the same decision, effect `artifact`, with //! the file's path and content digest as the identity within it. A //! re-dispatched attempt whose commit already landed reuses it when the //! workspace still sits on it unchanged (see [`RunWorkspaces::commit`]); a //! reissued routing decision finds the record, or the commit by its //! trailers, and writes nothing twice; a file already collected under the -//! same path and digest is not collected again. +//! same path and digest is not collected again. Publication keeps the +//! publisher's reconciliation policy; required finalization adds no independent +//! effect ledger or guarantee of deduplication across every external crash. //! //! # Where the workspace is //! diff --git a/lib/components/fabro-petri/src/test_support/finalization.rs b/lib/components/fabro-petri/src/test_support/finalization.rs index d21a4332b..b396c9342 100644 --- a/lib/components/fabro-petri/src/test_support/finalization.rs +++ b/lib/components/fabro-petri/src/test_support/finalization.rs @@ -10,6 +10,7 @@ use petri_runtime::frontend::CompileInputs; use petri_runtime::ir::FinalizationFailure; use petri_runtime::{RunOptions, Runtime}; use petri_store::{Access, LogId, MemoryRunStore, Record, RunKey, RunStore as _}; +use tokio::fs; use tokio::sync::{Notify, Semaphore}; use crate::providers::{self, SandboxProviderConfig}; @@ -76,7 +77,7 @@ pub async fn test_run_records( ) -> TestRunRecords { let root = tempfile::tempdir().expect("the fixture has an isolated directory"); let workflow = root.path().join("workflow.fabro"); - tokio::fs::write( + fs::write( &workflow, r#"digraph Finalization { graph [goal="Check required publication"] @@ -88,7 +89,7 @@ pub async fn test_run_records( ) .await .expect("the fixture workflow writes"); - tokio::fs::write( + fs::write( root.path().join("workflow.toml"), "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n", ) @@ -129,7 +130,7 @@ pub async fn test_run_records( )]; for execution in inspection.executions { let id = LogId::Execution(execution.execution); - records.push((id.clone(), logs.read(&id).await.expect("execution reads"))); + records.push((id, logs.read(&id).await.expect("execution reads"))); } let mut blobs = Vec::new(); for graph in inspection.graphs {