use std::sync::OnceLock;
use liminal_protocol::wire::{BindingEpoch, ConnectionIncarnation, Generation};
use super::super::super::log::{
StoredBindingEpoch, StoredDied, StoredDiedCause, StoredRecoveredPresentation,
StoredSpecificFateIntent, StoredTerminalDisposition,
};
use super::super::{
ExpectedSpecificFate, FateOccurrenceClass, FateOccurrenceRouter, FatePresentationOwner,
PrimaryFate,
};
use super::{FateHistoryError, FateOccurrenceKey, FateOccurrenceState};
type States = Vec<(FateOccurrenceKey, FateOccurrenceState)>;
static FIXTURE: OnceLock<Result<States, FateHistoryError>> = OnceLock::new();
fn key(participant_id: u64) -> FateOccurrenceKey {
FateOccurrenceKey {
conversation_id: 43,
participant_id,
binding_epoch: BindingEpoch::new(
ConnectionIncarnation::new(7, participant_id),
Generation::ONE,
),
}
}
fn died_state(
committed: bool,
expected: Option<ExpectedSpecificFate>,
completed: bool,
finalized: bool,
) -> Result<(FateOccurrenceKey, FateOccurrenceState), FateHistoryError> {
let key = key(1);
let intent = expected.map(|expected| match expected {
ExpectedSpecificFate::Ordinary => StoredSpecificFateIntent::Ordinary {
attached_source_sequence: 2,
},
ExpectedSpecificFate::Recovered => StoredSpecificFateIntent::Recovered {
attached_source_sequence: 2,
prior_binding_epoch: StoredBindingEpoch::from(key.binding_epoch),
marker_delivery_seq: 31,
},
});
let mut router = FateOccurrenceRouter::new();
router.route_died(
43,
&StoredDied {
participant_id: key.participant_id,
binding_epoch: StoredBindingEpoch::from(key.binding_epoch),
cause: StoredDiedCause::ConnectionLost,
terminal_order: 11,
disposition: if committed {
StoredTerminalDisposition::Committed { terminal_seq: 19 }
} else {
StoredTerminalDisposition::Pending
},
connection_intent_sequence: Some(3),
specific_fate_intent: intent,
drained: None,
},
10,
)?;
if completed {
let expected = expected.ok_or(super::super::FateOccurrenceConflict::SpecificClass)?;
let class = match expected {
ExpectedSpecificFate::Ordinary => FateOccurrenceClass::Ordinary,
ExpectedSpecificFate::Recovered => FateOccurrenceClass::Recovered,
};
let presentation = (class == FateOccurrenceClass::Recovered).then_some(if committed {
StoredRecoveredPresentation::DiedCommittedOwns
} else {
StoredRecoveredPresentation::RecoveredOwnsAndReservesFinalizer
});
router.route_specific(key, 10, 11, class, presentation)?;
}
if finalized {
router.select_finalizer(key)?;
}
let state = router
.state(key)
.ok_or(super::super::FateOccurrenceConflict::MissingDied)?;
Ok((key, state))
}
fn build_states() -> Result<States, FateHistoryError> {
let mut states = Vec::new();
for committed in [false, true] {
for expected in [
None,
Some(ExpectedSpecificFate::Ordinary),
Some(ExpectedSpecificFate::Recovered),
] {
for completed in [false, true] {
if completed && expected.is_none() {
continue;
}
states.push(died_state(committed, expected, completed, false)?);
if !committed {
states.push(died_state(committed, expected, completed, true)?);
}
}
}
}
let key = key(2);
for committed in [false, true] {
let state = FateOccurrenceState {
primary: PrimaryFate::Detached {
source_sequence: 10,
committed,
},
specific: None,
presentation_owner: committed.then_some(FatePresentationOwner::Detached),
recovered_reservation: None,
reservation_consumed: false,
};
let mut router = FateOccurrenceRouter::new();
router.insert_primary(key, state, FateOccurrenceClass::Detached)?;
states.push((key, state));
if !committed {
router.select_finalizer(key)?;
states.push((
key,
router
.state(key)
.ok_or(super::super::FateOccurrenceConflict::MissingDied)?,
));
}
}
Ok(states)
}
fn fixture() -> Result<&'static [(FateOccurrenceKey, FateOccurrenceState)], FateHistoryError> {
match FIXTURE.get_or_init(build_states) {
Ok(states) => Ok(states),
Err(error) => Err(*error),
}
}
#[test]
fn every_router_reachable_fate_state_retains_complete_presentation_and_reservation()
-> Result<(), FateHistoryError> {
assert_eq!(fixture()?.len(), 18);
for (key, state) in fixture()? {
let bytes = state.encode_history_value(*key, 11)?;
let restored = FateOccurrenceState::decode_history_value(&bytes, *key, 11)?;
assert_eq!(restored, *state);
assert_eq!(restored.encode_history_value(*key, 11)?, bytes);
}
Ok(())
}
#[test]
fn fate_history_refuses_key_substitution_every_partial_record_and_excess_source()
-> Result<(), FateHistoryError> {
let (key, state) = fixture()?[0];
let bytes = state.encode_history_value(key, 11)?;
for wrong in [
FateOccurrenceKey {
conversation_id: 44,
..key
},
FateOccurrenceKey {
participant_id: 2,
..key
},
FateOccurrenceKey {
binding_epoch: BindingEpoch::new(ConnectionIncarnation::new(8, 1), Generation::ONE),
..key
},
] {
assert!(FateOccurrenceState::decode_history_value(&bytes, wrong, 11).is_err());
}
for end in 0..bytes.len() {
assert!(FateOccurrenceState::decode_history_value(&bytes[..end], key, 11).is_err());
}
let mut trailing = bytes.clone();
trailing.push(0);
assert!(FateOccurrenceState::decode_history_value(&trailing, key, 11).is_err());
assert!(matches!(
FateOccurrenceState::decode_history_value(&bytes, key, 9),
Err(FateHistoryError::SourceBeyondCovered {
sequence: 10,
covered: 9
}),
));
Ok(())
}
#[test]
fn fate_history_refuses_unsupported_presentation_and_reservation_states()
-> Result<(), FateHistoryError> {
let (key, pending) = fixture()?
.iter()
.find(|(_, state)| {
state.specific == Some(FateOccurrenceClass::Recovered)
&& state.presentation_owner == Some(FatePresentationOwner::Recovered)
&& !state.reservation_consumed
})
.copied()
.ok_or(super::super::FateOccurrenceConflict::RecoveredReservation)?;
let mut consumed_without_reservation = pending;
consumed_without_reservation.recovered_reservation = None;
consumed_without_reservation.reservation_consumed = true;
assert!(
consumed_without_reservation
.encode_history_value(key, 11)
.is_err()
);
let mut stale_reservation = pending;
stale_reservation.recovered_reservation = Some(10);
assert!(stale_reservation.encode_history_value(key, 11).is_err());
let mut wrong_owner = pending;
wrong_owner.presentation_owner = Some(FatePresentationOwner::Finalizer);
assert!(wrong_owner.encode_history_value(key, 11).is_err());
let mut wrong_class = pending;
wrong_class.specific = Some(FateOccurrenceClass::Ordinary);
assert!(wrong_class.encode_history_value(key, 11).is_err());
let bytes = pending.encode_history_value(key, 11)?;
assert!(matches!(
FateOccurrenceState::decode_history_value(&bytes, key, 10),
Err(FateHistoryError::SourceBeyondCovered {
sequence: 11,
covered: 10
}),
));
Ok(())
}
#[test]
fn fate_history_refuses_unknown_tags_and_noncanonical_flags() -> Result<(), FateHistoryError> {
let (key, state) = fixture()?
.iter()
.find(|(_, state)| state.recovered_reservation.is_some() && !state.reservation_consumed)
.copied()
.ok_or(super::super::FateOccurrenceConflict::RecoveredReservation)?;
let bytes = state.encode_history_value(key, 11)?;
let body = 8 + key.history_key().encode_canonical().len();
for (at, tag) in [
(body, 3),
(body + 9, 2),
(body + 10, 3),
(body + 11, 3),
(body + 12, 5),
(body + 13, 2),
(bytes.len() - 1, 2),
] {
let mut changed = bytes.clone();
changed[at] = tag;
assert!(FateOccurrenceState::decode_history_value(&changed, key, 11).is_err());
}
Ok(())
}
fn checkpoint_positions() -> super::super::super::restore_checkpoint::CheckpointPositions {
use super::super::super::restore_checkpoint::{CheckpointPositions, CheckpointSchema};
CheckpointPositions {
conversation_id: 43,
covered_log_sequence: 11,
next_event_ordinal: 12,
next_log_sequence: 12,
next_seq: 32,
next_order: 12,
next_participant: 3,
extension_sequence: 0,
highest_delivery_seq: 31,
observer_progress: 0,
merged_base_sequence: 12,
merged_extension_sequence: 0,
schema: CheckpointSchema::V4Suffix,
}
}
fn owned_history_checkpoint(
states: impl ExactSizeIterator<Item = (FateOccurrenceKey, FateOccurrenceState)>,
) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
use super::super::super::state::checkpoint_body::CheckpointBodyEncoder;
let mut encoder = CheckpointBodyEncoder::new();
encoder.count("fate occurrences", states.len())?;
for (key, state) in states {
encoder.bytes("fate occurrence key", &key.history_key().encode_canonical())?;
encoder.bytes(
"fate occurrence value",
&state.encode_history_value(key, 11)?,
)?;
}
Ok(encoder.finish())
}
#[test]
fn borrowed_fate_table_matches_owned_native_history_values()
-> Result<(), Box<dyn std::error::Error>> {
assert_eq!(fixture()?.len(), 18);
for (key, state) in fixture()? {
let router = FateOccurrenceRouter {
occurrences: std::collections::BTreeMap::from([(*key, *state)]),
};
let borrowed = router.encode_checkpoint(11)?;
let moved = owned_history_checkpoint(router.occurrences.into_iter())?;
assert_eq!(borrowed, moved);
}
Ok(())
}
#[test]
fn every_active_fate_checkpoint_restores_the_same_routing_authority()
-> Result<(), Box<dyn std::error::Error>> {
let positions = checkpoint_positions();
positions.validate()?;
assert_eq!(fixture()?.len(), 18);
for (key, state) in fixture()? {
let router = FateOccurrenceRouter {
occurrences: std::collections::BTreeMap::from([(*key, *state)]),
};
let bytes = router.encode_checkpoint(11)?;
let restored = FateOccurrenceRouter::decode_checkpoint(&bytes, &positions)?;
assert_eq!(restored.occurrences, router.occurrences);
assert_eq!(restored.encode_checkpoint(11)?, bytes);
}
Ok(())
}
#[test]
fn active_fate_checkpoint_refuses_foreign_allocator_order_and_partial_bodies()
-> Result<(), Box<dyn std::error::Error>> {
let first = fixture()?[0];
let second = *fixture()?
.iter()
.find(|(key, _)| key.participant_id == 2)
.ok_or(super::super::FateOccurrenceConflict::MissingDied)?;
let router = FateOccurrenceRouter {
occurrences: std::collections::BTreeMap::from([first, second]),
};
let positions = checkpoint_positions();
let bytes = router.encode_checkpoint(11)?;
assert_eq!(router.occurrences.len(), 2);
for end in 0..bytes.len() {
assert!(FateOccurrenceRouter::decode_checkpoint(&bytes[..end], &positions).is_err());
}
let mut trailing = bytes.clone();
trailing.push(0);
assert!(FateOccurrenceRouter::decode_checkpoint(&trailing, &positions).is_err());
for changed in [
super::super::super::restore_checkpoint::CheckpointPositions {
conversation_id: 44,
..positions
},
super::super::super::restore_checkpoint::CheckpointPositions {
next_participant: 2,
..positions
},
super::super::super::restore_checkpoint::CheckpointPositions {
covered_log_sequence: 9,
next_log_sequence: 10,
merged_base_sequence: 10,
..positions
},
] {
assert!(FateOccurrenceRouter::decode_checkpoint(&bytes, &changed).is_err());
}
for values in [[second, first], [first, first]] {
let invalid = owned_history_checkpoint(values.into_iter())?;
assert!(FateOccurrenceRouter::decode_checkpoint(&invalid, &positions).is_err());
}
assert_eq!(router.encode_checkpoint(11)?, bytes);
Ok(())
}