use crate::state_hash::CanonicalState;
use std::collections::HashSet;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CrashBoundaryPoint {
BeforeStateMutation,
AfterStateMutationBeforeDedupPersist,
AfterDedupPersist,
}
#[derive(Clone, Debug)]
pub struct CrashAtomicExecutor {
state: CanonicalState,
dedup_set: HashSet<String>, }
impl CrashAtomicExecutor {
pub fn new() -> Self {
Self {
state: CanonicalState::new(),
dedup_set: HashSet::new(),
}
}
pub fn apply_with_crash(
&mut self,
operation_id: &str,
record_id: &str,
value: i32,
) -> Result<(), String> {
if self.dedup_set.contains(operation_id) {
return Err(format!(
"Operation {} already applied (dedup hit)",
operation_id
));
}
self.mutate_state(record_id, value);
self.dedup_set.insert(operation_id.to_string());
Ok(())
}
pub fn apply_with_simulated_crash(
&mut self,
operation_id: &str,
record_id: &str,
value: i32,
crash_point: CrashBoundaryPoint,
) -> CrashRecoveryState {
match crash_point {
CrashBoundaryPoint::BeforeStateMutation => {
CrashRecoveryState {
state_mutated: false,
dedup_persisted: false,
operation_id: operation_id.to_string(),
}
}
CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist => {
self.mutate_state(record_id, value);
CrashRecoveryState {
state_mutated: true,
dedup_persisted: false,
operation_id: operation_id.to_string(),
}
}
CrashBoundaryPoint::AfterDedupPersist => {
self.mutate_state(record_id, value);
self.dedup_set.insert(operation_id.to_string());
CrashRecoveryState {
state_mutated: true,
dedup_persisted: true,
operation_id: operation_id.to_string(),
}
}
}
}
pub fn recover(&self, crash_state: &CrashRecoveryState) -> RecoveryDecision {
let in_dedup = self.dedup_set.contains(&crash_state.operation_id);
match (
crash_state.state_mutated,
crash_state.dedup_persisted,
in_dedup,
) {
(true, true, true) => RecoveryDecision::AlreadyApplied,
(true, false, false) => RecoveryDecision::PartiallyApplied {
reason: "State mutated but dedup lost in crash".to_string(),
},
(true, false, true) => RecoveryDecision::InconsistentState {
reason: "State mutated but dedup signal false, yet in dedup set".to_string(),
},
(false, true, true) => RecoveryDecision::PhantomOperation {
reason: "Dedup persisted but state mutation didn't complete".to_string(),
},
(false, false, false) => RecoveryDecision::SafeToRetry,
(true, true, false) => RecoveryDecision::InconsistentState {
reason: "State shows applied but dedup missing".to_string(),
},
(false, false, true) => RecoveryDecision::InconsistentState {
reason: "Dedup has operation but state not applied".to_string(),
},
(false, true, false) => RecoveryDecision::InconsistentState {
reason: "Dedup persisted signal true but not in set".to_string(),
},
}
}
pub fn verify_invariant(&self, crash_state: &CrashRecoveryState) -> InvariantCheckResult {
let in_dedup = self.dedup_set.contains(&crash_state.operation_id);
if crash_state.state_mutated == in_dedup {
InvariantCheckResult::Satisfied
} else {
InvariantCheckResult::Violated {
state_mutated: crash_state.state_mutated,
dedup_persisted: in_dedup,
}
}
}
fn mutate_state(&mut self, record_id: &str, value: i32) {
self.state.set_record(
"test_collection".to_string(),
record_id.to_string(),
serde_json::json!({ "value": value }),
);
}
pub fn state(&self) -> &CanonicalState {
&self.state
}
pub fn dedup_set(&self) -> &HashSet<String> {
&self.dedup_set
}
}
#[derive(Clone, Debug)]
pub struct CrashRecoveryState {
pub state_mutated: bool,
pub dedup_persisted: bool,
pub operation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RecoveryDecision {
AlreadyApplied,
PartiallyApplied { reason: String },
PhantomOperation { reason: String },
SafeToRetry,
InconsistentState { reason: String },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InvariantCheckResult {
Satisfied,
Violated {
state_mutated: bool,
dedup_persisted: bool,
},
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_crash_before_state_mutation() {
let mut executor = CrashAtomicExecutor::new();
let crash_state = executor.apply_with_simulated_crash(
"op1",
"rec1",
10,
CrashBoundaryPoint::BeforeStateMutation,
);
assert!(!crash_state.state_mutated);
assert!(!crash_state.dedup_persisted);
let decision = executor.recover(&crash_state);
assert_eq!(decision, RecoveryDecision::SafeToRetry);
assert_eq!(
executor.verify_invariant(&crash_state),
InvariantCheckResult::Satisfied
);
}
#[test]
fn test_crash_after_state_mutation_before_dedup() {
let mut executor = CrashAtomicExecutor::new();
let crash_state = executor.apply_with_simulated_crash(
"op1",
"rec1",
10,
CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
);
assert!(crash_state.state_mutated);
assert!(!crash_state.dedup_persisted);
let invariant_result = executor.verify_invariant(&crash_state);
assert!(matches!(
invariant_result,
InvariantCheckResult::Violated { .. }
));
let decision = executor.recover(&crash_state);
assert!(matches!(
decision,
RecoveryDecision::PartiallyApplied { .. }
));
}
#[test]
fn test_crash_after_dedup_persist() {
let mut executor = CrashAtomicExecutor::new();
let crash_state = executor.apply_with_simulated_crash(
"op1",
"rec1",
10,
CrashBoundaryPoint::AfterDedupPersist,
);
assert!(crash_state.state_mutated);
assert!(crash_state.dedup_persisted);
let decision = executor.recover(&crash_state);
assert_eq!(decision, RecoveryDecision::AlreadyApplied);
assert_eq!(
executor.verify_invariant(&crash_state),
InvariantCheckResult::Satisfied
);
}
#[test]
fn test_duplicate_prevention_after_recovery() {
let mut executor = CrashAtomicExecutor::new();
let crash_state = executor.apply_with_simulated_crash(
"op1",
"rec1",
10,
CrashBoundaryPoint::AfterDedupPersist,
);
assert!(executor.dedup_set().contains("op1"));
let result = executor.apply_with_crash("op1", "rec1", 20);
assert!(result.is_err());
assert!(result.unwrap_err().contains("already applied"));
}
#[test]
fn test_exactly_once_under_all_crash_points() {
let crash_points = vec![
CrashBoundaryPoint::BeforeStateMutation,
CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
CrashBoundaryPoint::AfterDedupPersist,
];
for crash_point in crash_points {
let mut executor = CrashAtomicExecutor::new();
let crash_state =
executor.apply_with_simulated_crash("op1", "rec1", 10, crash_point.clone());
let decision = executor.recover(&crash_state);
match decision {
RecoveryDecision::AlreadyApplied => {
}
RecoveryDecision::SafeToRetry => {
}
RecoveryDecision::PartiallyApplied { .. } => {
}
_ => panic!("Unexpected recovery decision: {:?}", decision),
}
}
}
#[test]
fn test_invariant_satisfaction_across_crash_points() {
let crash_points = vec![
CrashBoundaryPoint::BeforeStateMutation,
CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
CrashBoundaryPoint::AfterDedupPersist,
];
for crash_point in crash_points {
let mut executor = CrashAtomicExecutor::new();
let crash_state =
executor.apply_with_simulated_crash("op1", "rec1", 10, crash_point.clone());
let in_dedup = executor.dedup_set().contains("op1");
if crash_point == CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist {
assert_ne!(crash_state.state_mutated, in_dedup);
} else {
assert_eq!(crash_state.state_mutated, in_dedup);
}
}
}
#[test]
fn test_recovery_state_tracking() {
let mut executor = CrashAtomicExecutor::new();
let _ = executor.apply_with_crash("op1", "rec1", 10);
assert!(executor.dedup_set().contains("op1"));
let crash_state = CrashRecoveryState {
state_mutated: true,
dedup_persisted: true,
operation_id: "op1".to_string(),
};
let decision = executor.recover(&crash_state);
assert_eq!(decision, RecoveryDecision::AlreadyApplied);
}
#[test]
fn test_multiple_operations_crash_atomic() {
let mut executor = CrashAtomicExecutor::new();
let _ = executor.apply_with_crash("op1", "rec1", 10);
assert_eq!(executor.dedup_set().len(), 1);
let crash_state = executor.apply_with_simulated_crash(
"op2",
"rec2",
20,
CrashBoundaryPoint::AfterStateMutationBeforeDedupPersist,
);
assert!(executor.dedup_set().contains("op1"));
assert!(!executor.dedup_set().contains("op2"));
let invariant = executor.verify_invariant(&crash_state);
assert!(matches!(invariant, InvariantCheckResult::Violated { .. }));
}
}