From a21ab245210227b997b6a48bfa2c52dd8c30dad3 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 19 Apr 2026 18:40:21 -0400 Subject: [PATCH] fix(store): harden archive/unarchive projection apply against corrupt logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review surfaced that both RunArchived and RunUnarchived apply arms were naive — any out-of-spec event in the log (concurrent double archive, tampered import, replayed retry) would permanently corrupt the projection: - RunArchived unconditionally captured current status into prior_status. A second RunArchived would set prior_status=Some(Archived). Unarchive would then emit restored_status=Archived, the apply arm would set status=Archived and clear prior_status, and the run would be unrecoverable. - RunUnarchived trusted restored_status unconditionally. An imported event with restored_status=Running produced a projection reporting status=Running with no RunRunning event in the log — breaking is_active/is_terminal invariants. Both arms now require a sensible pre-state before mutating: - RunArchived only transitions from Succeeded|Failed|Dead. - RunUnarchived only runs from Archived with a terminal restored_status. Adds three regression tests: - double_archive_preserves_prior_status - run_unarchived_with_non_terminal_restored_status_is_ignored - run_archived_on_non_terminal_projection_is_ignored The operations layer (archive/unarchive in fabro-workflow) still validates at emit time; the projection guards are a defensive second line for replay, imports, and any future code path that double-writes. --- lib/crates/fabro-store/src/run_state.rs | 164 +++++++++++++++++++++++- 1 file changed, 161 insertions(+), 3 deletions(-) diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index da05d3ac3..4d205613b 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -215,10 +215,38 @@ impl RunProjection { self.reset_for_rewind(); } EventBody::RunArchived(_props) => { - self.prior_status = self.status.as_ref().map(|record| record.status); - self.status = Some(run_status_record(RunStatus::Archived, None, ts)); + // Idempotent on replay: a second RunArchived (from a concurrent + // archive race or a tampered log) must not overwrite + // prior_status — doing so would permanently corrupt the run + // (unarchive would then restore status=Archived and never + // recover). We also ignore the archive attempt when current + // status is not a terminal non-archived — the operations + // layer is the only legitimate emitter and it validates + // there, but replay must be defensive. + let current = self.status.as_ref().map(|record| record.status); + if matches!( + current, + Some(RunStatus::Succeeded | RunStatus::Failed | RunStatus::Dead) + ) { + self.prior_status = current; + self.status = Some(run_status_record(RunStatus::Archived, None, ts)); + } } - EventBody::RunUnarchived(props) => { + EventBody::RunUnarchived(props) + if matches!( + self.status.as_ref().map(|record| record.status), + Some(RunStatus::Archived) + ) && matches!( + props.restored_status, + RunStatus::Succeeded | RunStatus::Failed | RunStatus::Dead + ) => + { + // Defensive: restored_status must be a real terminal status. + // Accepting e.g. Running here would produce a projection + // whose status is Running without any RunRunning event in + // the log, breaking the is_active / is_terminal invariants. + // The operations layer only ever writes validated terminals, + // but replay must enforce it. self.status = Some(run_status_record(props.restored_status, None, ts)); self.prior_status = None; } @@ -1246,4 +1274,134 @@ mod tests { ); assert_eq!(state.prior_status, None); } + + #[test] + fn double_archive_preserves_prior_status() { + // Two RunArchived events in sequence — possible under a concurrent + // archive race or a replayed/retried write — must not overwrite + // prior_status with Archived. Otherwise unarchive would emit + // restored_status=Archived, the apply arm would set status=Archived + + // prior_status=None, and the run would be permanently unrecoverable. + use fabro_types::RunStatus; + use fabro_types::run_event::{RunArchivedProps, RunCompletedProps}; + + let mut state = RunProjection::default(); + state + .apply_event(&test_event( + 1, + EventBody::RunCompleted(RunCompletedProps { + duration_ms: 10, + artifact_count: 0, + status: "success".to_string(), + reason: None, + total_usd_micros: None, + final_git_commit_sha: None, + final_patch: None, + billing: None, + }), + None, + )) + .unwrap(); + state + .apply_event(&test_event( + 2, + EventBody::RunArchived(RunArchivedProps { actor: None }), + None, + )) + .unwrap(); + state + .apply_event(&test_event( + 3, + EventBody::RunArchived(RunArchivedProps { actor: None }), + None, + )) + .unwrap(); + + assert_eq!( + state.status.as_ref().map(|record| record.status), + Some(RunStatus::Archived) + ); + assert_eq!(state.prior_status, Some(RunStatus::Succeeded)); + } + + #[test] + fn run_unarchived_with_non_terminal_restored_status_is_ignored() { + // The operations layer never emits this, but a corrupt log or an + // imported event must not drag the projection into a nonsense state + // (status=Running without any RunRunning event preceding it). + use fabro_types::RunStatus; + use fabro_types::run_event::{RunArchivedProps, RunCompletedProps, RunUnarchivedProps}; + + let mut state = RunProjection::default(); + state + .apply_event(&test_event( + 1, + EventBody::RunCompleted(RunCompletedProps { + duration_ms: 10, + artifact_count: 0, + status: "success".to_string(), + reason: None, + total_usd_micros: None, + final_git_commit_sha: None, + final_patch: None, + billing: None, + }), + None, + )) + .unwrap(); + state + .apply_event(&test_event( + 2, + EventBody::RunArchived(RunArchivedProps { actor: None }), + None, + )) + .unwrap(); + state + .apply_event(&test_event( + 3, + EventBody::RunUnarchived(RunUnarchivedProps { + actor: None, + restored_status: RunStatus::Running, + }), + None, + )) + .unwrap(); + + assert_eq!( + state.status.as_ref().map(|record| record.status), + Some(RunStatus::Archived) + ); + assert_eq!(state.prior_status, Some(RunStatus::Succeeded)); + } + + #[test] + fn run_archived_on_non_terminal_projection_is_ignored() { + // Same defensive posture on the other side: an event log that + // somehow contains RunArchived on a Running run must not produce + // a projection with status=Archived and prior_status=Some(Running). + use fabro_types::RunStatus; + use fabro_types::run_event::{RunArchivedProps, RunStatusTransitionProps}; + + let mut state = RunProjection::default(); + state + .apply_event(&test_event( + 1, + EventBody::RunRunning(RunStatusTransitionProps { reason: None }), + None, + )) + .unwrap(); + state + .apply_event(&test_event( + 2, + EventBody::RunArchived(RunArchivedProps { actor: None }), + None, + )) + .unwrap(); + + assert_eq!( + state.status.as_ref().map(|record| record.status), + Some(RunStatus::Running) + ); + assert_eq!(state.prior_status, None); + } }