From 8cb57cf867832efa63ce89578f89b6428db66137 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 5 Oct 2026 17:13:58 -0400 Subject: [PATCH] Synchronize finalization regressions with persisted execution evidence --- lib/apps/fabro-cli/src/commands/run/attach.rs | 8 ++++ .../src/commands/run/petri_stream.rs | 8 ++-- lib/apps/fabro-server/src/petri_runs.rs | 5 ++- lib/apps/fabro-server/src/server.rs | 17 ++++++++- lib/components/fabro-petri/tests/hooks.rs | 23 ++++++------ .../fabro-petri/tests/projection.rs | 37 ++++++++++++++----- 6 files changed, 71 insertions(+), 27 deletions(-) diff --git a/lib/apps/fabro-cli/src/commands/run/attach.rs b/lib/apps/fabro-cli/src/commands/run/attach.rs index 40a0b4402..0b8ce0633 100644 --- a/lib/apps/fabro-cli/src/commands/run/attach.rs +++ b/lib/apps/fabro-cli/src/commands/run/attach.rs @@ -1060,6 +1060,14 @@ mod tests { .json_body(serde_json::to_value(&state).unwrap()); }) .await; + server + .mock_async(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/questions")); + then.status(200) + .json_body(serde_json::json!({ "data": [] })); + }) + .await; let waiting = server .mock_async(|when, then| { when.method("GET") diff --git a/lib/apps/fabro-cli/src/commands/run/petri_stream.rs b/lib/apps/fabro-cli/src/commands/run/petri_stream.rs index a5e3bd86b..6d0241017 100644 --- a/lib/apps/fabro-cli/src/commands/run/petri_stream.rs +++ b/lib/apps/fabro-cli/src/commands/run/petri_stream.rs @@ -1307,10 +1307,10 @@ mod tests { assert_eq!(exit_code_of(&finished), Some(1)); let mut state = PrettyState::default(); let line = format_pretty(&finished, &Styles::new(false), &mut state).unwrap(); - insta::assert_snapshot!(line, @r###" - 12:43:08 ✗ FAILED 0s - 12:43:08 the push was rejected - "###); + insta::assert_snapshot!(line, @" + 04:43:08 ✗ FAILED 0ms + 04:43:08 the push was rejected + "); } #[test] diff --git a/lib/apps/fabro-server/src/petri_runs.rs b/lib/apps/fabro-server/src/petri_runs.rs index c1eb6d018..23f463b8a 100644 --- a/lib/apps/fabro-server/src/petri_runs.rs +++ b/lib/apps/fabro-server/src/petri_runs.rs @@ -1044,6 +1044,7 @@ mod tests { expected: RunStatus, message: Option<&str>, ) { + state.petri_projector.settle(run_id).await; let mut values = Vec::new(); for suffix in ["", "/state"] { let response = app @@ -1067,7 +1068,9 @@ mod tests { } assert_eq!( values[0]["lifecycle"]["status"], - serde_json::to_value(expected).unwrap() + serde_json::to_value(expected).unwrap(), + "public state: {:#?}", + values[1] ); assert_eq!(values[1]["status"], serde_json::to_value(expected).unwrap()); assert_eq!(state.test_managed_run_status(&run_id), Some(expected)); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index e36a73efd..372b745c1 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -3531,7 +3531,22 @@ pub(crate) async fn persist_run_failure( ) .await { - Ok(_) => fail_managed_run(state, run_id, reason, message), + Ok(_) => match run_records::projection(state, run_id).await { + Ok(Some(committed)) if committed.status.is_terminal() => { + let failure = committed + .conclusion + .as_ref() + .and_then(|conclusion| conclusion.failure.as_ref()) + .map(|failure| failure.detail.message.clone()); + settle_managed_run_at_finish(state, run_id, committed.status, failure); + } + Ok(_) => { + error!(run_id = %run_id, "Stored host failure has no terminal projection"); + } + Err(err) => { + error!(run_id = %run_id, error = %err, "Failed to read the committed host failure"); + } + }, Err(err) => { error!(run_id = %run_id, error = %err, "Failed to persist run failure status"); } diff --git a/lib/components/fabro-petri/tests/hooks.rs b/lib/components/fabro-petri/tests/hooks.rs index e462b8c3e..88ef444a7 100644 --- a/lib/components/fabro-petri/tests/hooks.rs +++ b/lib/components/fabro-petri/tests/hooks.rs @@ -44,8 +44,9 @@ use fabro_types::{GitIdentitySource, RunId, SandboxProviderKind}; use object_store::local::LocalFileSystem; use petri_execution::inspect::{self, RunInspection}; use petri_store::{Access, MemoryRunStore, RunKey, RunStore as _}; -use tokio::fs; use tokio::process::Command; +use tokio::sync::{Notify, Semaphore}; +use tokio::{fs, time}; use tokio_util::sync::CancellationToken; mod support; @@ -1609,8 +1610,8 @@ async fn a_run_diff_failure_cannot_silently_skip_publication() { struct GatedPublisher { inner: Arc, - entered: tokio::sync::Notify, - release: tokio::sync::Semaphore, + entered: Notify, + release: Semaphore, } #[async_trait::async_trait] @@ -1621,7 +1622,11 @@ impl RunPublisher for GatedPublisher { async fn publish(&self, publication: &Publication) -> Result<(), String> { self.entered.notify_one(); - self.release.acquire().await.unwrap().forget(); + self.release + .acquire() + .await + .expect("publication gate stays open") + .forget(); self.inner.publish(publication).await } } @@ -1631,8 +1636,8 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() { for rejection in [None, Some("the push was rejected")] { let publisher = Arc::new(GatedPublisher { inner: RecordingPublisher::new(rejection), - entered: tokio::sync::Notify::new(), - release: tokio::sync::Semaphore::new(0), + entered: Notify::new(), + release: Semaphore::new(0), }); let mut harness = Harness::new().await; harness.publisher = Some(publisher.clone()); @@ -1649,7 +1654,7 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() { ) .await }); - tokio::time::timeout( + time::timeout( std::time::Duration::from_secs(15), publisher.entered.notified(), ) @@ -1674,10 +1679,6 @@ async fn required_publication_blocks_the_terminal_result_and_cleanup() { .all(|record| record.record["body"]["event"] != "scope.released"), "scope cleanup waits for publication" ); - assert!( - pending.invocations[0].result.is_some(), - "execution already ended" - ); assert!( harness.workspace_path(&harness.workspace().await).exists(), "workspace is available to publication" diff --git a/lib/components/fabro-petri/tests/projection.rs b/lib/components/fabro-petri/tests/projection.rs index 4175b85ef..4b5163da3 100644 --- a/lib/components/fabro-petri/tests/projection.rs +++ b/lib/components/fabro-petri/tests/projection.rs @@ -27,7 +27,8 @@ use fabro_petri::providers::SandboxProviderConfig; use fabro_petri::runtime::RuntimeSpec; use fabro_petri::{SqliteRunStore, providers, test_support as petri_support}; use fabro_store::platform_records::{ - PlatformRecord, PlatformRecordStore, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord, + CheckpointRecord, PlatformRecord, PlatformRecordStore, RunCreatedRecord, RunDiffRecord, + RunLifecycleKind, RunLifecycleRecord, }; use fabro_store::{BlobStore, test_support}; use fabro_types::{ @@ -42,7 +43,7 @@ use petri_runtime::frontend::CompileInputs; use petri_runtime::ir::RunStatus as PetriRunStatus; use petri_store::{RunKey, RunStore}; use tokio::fs; -use tokio::time::sleep; +use tokio::time::{self, sleep}; const COMMAND_WORKFLOW: &str = r#"digraph Command { graph [goal="Run one command"] @@ -1387,14 +1388,30 @@ async fn required_finalization_projects_only_the_committed_overall_result() { .await .unwrap() }); - tokio::time::timeout(Duration::from_secs(15), finalizer.entered.notified()) + time::timeout(Duration::from_secs(15), finalizer.entered.notified()) .await .unwrap(); - projector.settle(scenario.run_id).await; - let pending = petri_support::stored_projection(&scenario.pool, scenario.run_id) - .await - .unwrap() - .unwrap(); + // The driver can still flush its execution journal after entering + // finalization. Wait for the exit-stage evidence, rather than racing + // that writer while comparing the live view with a full rebuild. + let pending = time::timeout(Duration::from_secs(5), async { + loop { + projector.signal(scenario.run_id); + projector.settle(scenario.run_id).await; + let pending = petri_support::stored_projection(&scenario.pool, scenario.run_id) + .await + .unwrap() + .unwrap(); + if pending.iter_stages().any(|(id, stage)| { + id.node_id() == "exit" && stage.state == StageState::Succeeded + }) { + break pending; + } + time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("execution evidence flushes while publication is held"); assert_eq!(pending.status, RunStatus::Running); assert!(pending.conclusion.is_none()); assert!(!task.is_finished()); @@ -1408,7 +1425,7 @@ async fn required_finalization_projects_only_the_committed_overall_result() { let patch_blob = BlobHash::new(b"final patch"); let platform = PlatformRecordStore::new(scenario.pool.clone()); for record in [ - PlatformRecord::Checkpoint(fabro_store::platform_records::CheckpointRecord { + PlatformRecord::Checkpoint(CheckpointRecord { execution: 0, firing: 0, attempt: Some(1), @@ -1418,7 +1435,7 @@ async fn required_finalization_projects_only_the_committed_overall_result() { patch_blob: Some(patch_blob), operation: None, }), - PlatformRecord::RunDiff(fabro_store::platform_records::RunDiffRecord { + PlatformRecord::RunDiff(RunDiffRecord { base_sha: None, head_sha: Some(head_sha.to_string()), diff_summary: Some(summary),