mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-11 03:40:05 +00:00
fix(store): harden archive/unarchive projection apply against corrupt logs
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.
This commit is contained in:
Bryan Helmkamp
parent
48eb5efed6
commit
a21ab24521
1 changed files with 161 additions and 3 deletions
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue