mod decode;
use super::super::restore_checkpoint_history::{HistoryCodecError, HistoryDecoder};
use super::super::restore_checkpoint_history_key::{CheckpointHistoryKey, HistoryEntryKey};
use super::{
ExpectedSpecificFate, FateOccurrenceClass, FateOccurrenceConflict, FateOccurrenceKey,
FateOccurrenceState, FatePresentationOwner, PrimaryFate,
};
const MAGIC: &[u8; 4] = b"LSHV";
const VERSION: u16 = 1;
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub(in super::super) enum FateHistoryError {
#[error(transparent)]
Codec(#[from] HistoryCodecError),
#[error(transparent)]
Invariant(#[from] FateOccurrenceConflict),
#[error("fate value does not belong to its complete history key")]
KeyAssociation,
#[error("fate source {sequence} exceeds saved covered sequence {covered}")]
SourceBeyondCovered { sequence: u64, covered: u64 },
}
impl FateOccurrenceKey {
pub(in super::super) const fn history_key(self) -> CheckpointHistoryKey {
CheckpointHistoryKey::new(
self.conversation_id,
HistoryEntryKey::FateOccurrence {
participant_id: self.participant_id,
binding_epoch: self.binding_epoch,
},
)
}
}
impl FateOccurrenceState {
pub(in super::super) fn encode_history_value(
&self,
key: FateOccurrenceKey,
covered_sequence: u64,
) -> Result<Vec<u8>, FateHistoryError> {
self.validate_history_value(covered_sequence)?;
let key_bytes = key.history_key().encode_canonical();
let key_length = u16::try_from(key_bytes.len()).map_err(|_| HistoryCodecError::Length)?;
let mut bytes = Vec::new();
bytes.extend_from_slice(MAGIC);
bytes.extend_from_slice(&VERSION.to_be_bytes());
bytes.extend_from_slice(&key_length.to_be_bytes());
bytes.extend_from_slice(&key_bytes);
encode_primary(&mut bytes, self.primary);
bytes.push(match self.specific {
None => 0,
Some(FateOccurrenceClass::Ordinary) => 1,
Some(FateOccurrenceClass::Recovered) => 2,
Some(FateOccurrenceClass::Died | FateOccurrenceClass::Detached) => {
return Err(FateOccurrenceConflict::SpecificClass.into());
}
});
bytes.push(match self.presentation_owner {
None => 0,
Some(FatePresentationOwner::Died) => 1,
Some(FatePresentationOwner::Detached) => 2,
Some(FatePresentationOwner::Recovered) => 3,
Some(FatePresentationOwner::Finalizer) => 4,
});
bytes.push(u8::from(self.recovered_reservation.is_some()));
if let Some(sequence) = self.recovered_reservation {
bytes.extend_from_slice(&sequence.to_be_bytes());
}
bytes.push(u8::from(self.reservation_consumed));
Ok(bytes)
}
pub(in super::super) fn decode_history_value(
bytes: &[u8],
expected_key: FateOccurrenceKey,
covered_sequence: u64,
) -> Result<Self, FateHistoryError> {
let mut input = HistoryDecoder::new(bytes);
if input.take::<4>()? != *MAGIC {
return Err(HistoryCodecError::Magic("history value").into());
}
let version = input.u16()?;
if version != VERSION {
return Err(HistoryCodecError::Version {
field: "history value",
version,
}
.into());
}
let key_length = usize::from(input.u16()?);
let key = CheckpointHistoryKey::decode_canonical(
input.slice(key_length)?,
expected_key.conversation_id,
)?;
if key != expected_key.history_key() {
return Err(FateHistoryError::KeyAssociation);
}
let primary = decode_primary(&mut input)?;
let specific = match input.byte()? {
0 => None,
1 => Some(FateOccurrenceClass::Ordinary),
2 => Some(FateOccurrenceClass::Recovered),
tag => return Err(tag_error("specific fate", tag)),
};
let presentation_owner = match input.byte()? {
0 => None,
1 => Some(FatePresentationOwner::Died),
2 => Some(FatePresentationOwner::Detached),
3 => Some(FatePresentationOwner::Recovered),
4 => Some(FatePresentationOwner::Finalizer),
tag => return Err(tag_error("fate presentation", tag)),
};
let recovered_reservation = if read_flag(&mut input)? {
Some(input.u64()?)
} else {
None
};
let reservation_consumed = read_flag(&mut input)?;
input.finish()?;
let value = Self {
primary,
specific,
presentation_owner,
recovered_reservation,
reservation_consumed,
};
value.validate_history_value(covered_sequence)?;
Ok(value)
}
fn validate_history_value(&self, covered: u64) -> Result<(), FateHistoryError> {
let source = match self.primary {
PrimaryFate::Died {
source_sequence, ..
}
| PrimaryFate::Detached {
source_sequence, ..
} => source_sequence,
};
if source > covered {
return Err(FateHistoryError::SourceBeyondCovered {
sequence: source,
covered,
});
}
if let Some(sequence) = self.recovered_reservation {
if sequence > covered {
return Err(FateHistoryError::SourceBeyondCovered { sequence, covered });
}
if sequence <= source {
return Err(FateOccurrenceConflict::RecoveredReservation.into());
}
} else if self.reservation_consumed {
return Err(FateOccurrenceConflict::RecoveredReservation.into());
}
self.validate_presentation()?;
Ok(())
}
fn validate_presentation(&self) -> Result<(), FateOccurrenceConflict> {
match self.primary {
PrimaryFate::Died {
committed,
expected_specific,
..
} => {
if let Some(specific) = self.specific {
let expected = match expected_specific {
Some(ExpectedSpecificFate::Ordinary) => FateOccurrenceClass::Ordinary,
Some(ExpectedSpecificFate::Recovered) => FateOccurrenceClass::Recovered,
None => return Err(FateOccurrenceConflict::SpecificClass),
};
if specific != expected {
return Err(FateOccurrenceConflict::SpecificClass);
}
}
if committed {
self.validate_fixed_owner(Some(FatePresentationOwner::Died))
} else if self.specific == Some(FateOccurrenceClass::Recovered) {
if self.presentation_owner != Some(FatePresentationOwner::Recovered)
|| self.recovered_reservation.is_none()
{
return Err(FateOccurrenceConflict::RecoveredReservation);
}
Ok(())
} else {
self.validate_pending_owner()
}
}
PrimaryFate::Detached { committed, .. } => {
if self.specific.is_some() {
return Err(FateOccurrenceConflict::SpecificClass);
}
if committed {
self.validate_fixed_owner(Some(FatePresentationOwner::Detached))
} else {
self.validate_pending_owner()
}
}
}
}
fn validate_fixed_owner(
&self,
owner: Option<FatePresentationOwner>,
) -> Result<(), FateOccurrenceConflict> {
if self.presentation_owner != owner {
return Err(FateOccurrenceConflict::PresentationOwner);
}
if self.recovered_reservation.is_some() || self.reservation_consumed {
return Err(FateOccurrenceConflict::RecoveredReservation);
}
Ok(())
}
const fn validate_pending_owner(&self) -> Result<(), FateOccurrenceConflict> {
if !matches!(
self.presentation_owner,
None | Some(FatePresentationOwner::Finalizer)
) {
return Err(FateOccurrenceConflict::PresentationOwner);
}
if self.recovered_reservation.is_some() || self.reservation_consumed {
return Err(FateOccurrenceConflict::RecoveredReservation);
}
Ok(())
}
}
fn encode_primary(bytes: &mut Vec<u8>, primary: PrimaryFate) {
let (tag, source, committed) = match primary {
PrimaryFate::Died {
source_sequence,
committed,
..
} => (1, source_sequence, committed),
PrimaryFate::Detached {
source_sequence,
committed,
} => (2, source_sequence, committed),
};
bytes.push(tag);
bytes.extend_from_slice(&source.to_be_bytes());
bytes.push(u8::from(committed));
if let PrimaryFate::Died {
expected_specific, ..
} = primary
{
bytes.push(match expected_specific {
None => 0,
Some(ExpectedSpecificFate::Ordinary) => 1,
Some(ExpectedSpecificFate::Recovered) => 2,
});
}
}
fn decode_primary(input: &mut HistoryDecoder<'_>) -> Result<PrimaryFate, FateHistoryError> {
let tag = input.byte()?;
let source_sequence = input.u64()?;
let committed = read_flag(input)?;
match tag {
1 => {
let expected_specific = match input.byte()? {
0 => None,
1 => Some(ExpectedSpecificFate::Ordinary),
2 => Some(ExpectedSpecificFate::Recovered),
tag => return Err(tag_error("expected specific fate", tag)),
};
Ok(PrimaryFate::Died {
source_sequence,
committed,
expected_specific,
})
}
2 => Ok(PrimaryFate::Detached {
source_sequence,
committed,
}),
tag => Err(tag_error("primary fate", tag)),
}
}
fn read_flag(input: &mut HistoryDecoder<'_>) -> Result<bool, FateHistoryError> {
match input.byte()? {
0 => Ok(false),
1 => Ok(true),
tag => Err(tag_error("fate flag", tag)),
}
}
fn tag_error(field: &'static str, tag: u8) -> FateHistoryError {
HistoryCodecError::Tag { field, tag }.into()
}
#[cfg(test)]
mod tests;
impl super::FateOccurrenceRouter {
pub(in super::super) fn encode_checkpoint(
&self,
covered_sequence: u64,
) -> Result<Vec<u8>, super::super::restore_checkpoint::CheckpointError> {
use super::super::state::checkpoint_body::CheckpointBodyEncoder;
let mut encoder = CheckpointBodyEncoder::new();
encoder.count("fate occurrences", self.occurrences.len())?;
for (key, value) in &self.occurrences {
encoder.bytes("fate occurrence key", &key.history_key().encode_canonical())?;
let bytes = value
.encode_history_value(*key, covered_sequence)
.map_err(
|source| super::super::restore_checkpoint::CheckpointError::LiveState {
source: Box::new(source),
},
)?;
encoder.bytes("fate occurrence value", &bytes)?;
}
Ok(encoder.finish())
}
}