use super::super::log::{
StoredBindingEpoch, StoredCommittedTerminalAudit, StoredOrdinaryTerminalSource,
StoredPendingDiedFinalizer, StoredSpecificFateIntent,
};
use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};
use super::{PendingSpecificFate, PendingSpecificFateTerminal, Slot};
use liminal_protocol::lifecycle::{
ActiveBinding, BindingOrigin, CommittedBindingTerminal, CommittedBindingTerminalRestore,
};
use std::collections::BTreeMap;
#[derive(Debug, thiserror::Error)]
#[error("pending fate terminal source refused: {0}")]
struct SourceError(#[source] serde_json::Error);
#[derive(Clone, Copy, serde::Serialize, serde::Deserialize)]
struct PendingTerminalSource {
source: StoredOrdinaryTerminalSource,
binding_epoch: StoredBindingEpoch,
audit: StoredCommittedTerminalAudit,
}
#[derive(Debug, thiserror::Error)]
#[error("pending fate native terminal refused: {0:?}")]
struct TerminalError(liminal_protocol::lifecycle::StorageRestoreError);
fn source(error: serde_json::Error) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(SourceError(error)),
}
}
pub(super) fn encode(
values: &BTreeMap<u64, PendingSpecificFate>,
slots: &BTreeMap<u64, Slot>,
) -> Result<Vec<u8>, CheckpointError> {
if values.is_empty() {
return Ok(Vec::new());
}
let mut body = CheckpointBodyEncoder::new();
body.count("pending specific fates", values.len())?;
for (&id, value) in values {
let slot = slots.get(&id).ok_or_else(|| invalid("pending fate slot"))?;
let attached_source_sequence = match value.intent {
StoredSpecificFateIntent::Ordinary {
attached_source_sequence,
}
| StoredSpecificFateIntent::Recovered {
attached_source_sequence,
..
} => attached_source_sequence,
};
if restored_intent(&value.binding_fate)? != value.intent {
return Err(invalid("pending fate intent association"));
}
if attached_source_sequence != value.binding_fate.attached_source_sequence
|| attached_source_sequence != slot.retained_source.log_sequence
|| slot.binding_fate.is_some()
|| value.died_source_sequence == 0
{
return Err(invalid("pending fate sole authority association"));
}
let terminal_source = value.terminal.map(|value| PendingTerminalSource {
source: value.source,
binding_epoch: value.terminal.binding_epoch().into(),
audit: StoredCommittedTerminalAudit {
cause: super::super::binding_fate_completion::stored_died_cause(
value.terminal.cause(),
),
transaction_order: value.terminal.admission_order().transaction_order(),
terminal_seq: value.terminal.delivery_seq(),
},
});
if let Some(terminal) = value.terminal {
if slot.member.latest_terminal()
!= Some(CommittedBindingTerminal::Died(terminal.terminal))
{
return Err(invalid("pending fate native terminal association"));
}
}
body.u64(id);
body.u64(value.died_source_sequence);
body.bytes(
"pending fate token",
&super::checkpoint_binding_fate::encode_pending(&value.binding_fate)?,
)?;
body.bytes(
"pending fate terminal source",
&serde_json::to_vec(&terminal_source).map_err(source)?,
)?;
}
Ok(body.finish())
}
pub(super) fn validate_capture(
values: &BTreeMap<u64, PendingSpecificFate>,
slots: &BTreeMap<u64, Slot>,
origins: &BTreeMap<u64, (BindingOrigin, u64)>,
covered_log_sequence: u64,
fenced: &mut super::checkpoint_owner::CheckpointFencedSources,
) -> Result<(), CheckpointError> {
for (&id, value) in values {
let (origin, _) = origins.get(&id).ok_or(CheckpointError::Position {
field: "pending fate validated binding origin",
})?;
let slot = slots.get(&id).ok_or_else(|| invalid("pending fate slot"))?;
super::checkpoint_binding_fate::validate_capture(
&value.binding_fate,
origin,
slot.member.cursor(),
fenced.remove(&id),
)?;
covered(value.died_source_sequence, covered_log_sequence)?;
match value.terminal {
Some(terminal) => validate_terminal_source(
terminal.source,
value.died_source_sequence,
covered_log_sequence,
)?,
None => {
if !matches!(
slot.binding,
liminal_protocol::lifecycle::BindingState::PendingFinalization(
liminal_protocol::lifecycle::PendingFinalization::Died(_)
)
) {
return Err(invalid("pending fate omitted committed terminal"));
}
}
}
}
Ok(())
}
pub(super) fn decode(
bytes: &[u8],
slots: &BTreeMap<u64, Slot>,
origins: &BTreeMap<u64, (BindingOrigin, u64)>,
positions: &CheckpointPositions,
fenced: &mut super::checkpoint_owner::CheckpointFencedSources,
) -> Result<BTreeMap<u64, PendingSpecificFate>, CheckpointError> {
let mut values = BTreeMap::new();
if bytes.is_empty() {
return Ok(values);
}
let mut body = CheckpointBodyDecoder::new(bytes);
let count = body.count("pending specific fates", 24)?;
let mut previous = None;
for _ in 0..count {
let id = body.u64("pending fate participant")?;
if id >= positions.next_participant || previous.is_some_and(|last| last >= id) {
return Err(invalid("pending fate participant order"));
}
previous = Some(id);
let slot = slots.get(&id).ok_or_else(|| invalid("pending fate slot"))?;
let (origin, source_sequence) = origins.get(&id).ok_or(CheckpointError::Position {
field: "pending fate validated binding origin",
})?;
let died_source_sequence = body.u64("pending fate Died source")?;
covered(died_source_sequence, positions.covered_log_sequence)?;
let binding_fate = super::checkpoint_binding_fate::decode_pending(
body.bytes("pending fate token")?,
Some(origin),
slot.member.cursor(),
positions,
super::checkpoint_binding_fate::FateRestoreIdentity::Pending {
participant_id: id,
died_source_sequence,
},
fenced.remove(&id),
)?;
if slot.binding_fate.is_some()
|| binding_fate.attached_source_sequence != *source_sequence
|| binding_fate.attached_source_sequence >= died_source_sequence
{
return Err(invalid("pending fate sole authority source association"));
}
let terminal_source_body = body.bytes("pending fate terminal source")?;
let terminal_source: Option<PendingTerminalSource> =
serde_json::from_slice(terminal_source_body).map_err(source)?;
if serde_json::to_vec(&terminal_source).map_err(source)? != terminal_source_body {
return Err(invalid("pending fate canonical terminal source"));
}
let terminal = match terminal_source {
None => {
if !matches!(
slot.binding,
liminal_protocol::lifecycle::BindingState::PendingFinalization(
liminal_protocol::lifecycle::PendingFinalization::Died(_)
)
) {
return Err(invalid("pending fate omitted committed terminal"));
}
None
}
Some(saved) => {
let terminal = restore_terminal(saved, id, slot, positions)?;
validate_terminal_source(
saved.source,
died_source_sequence,
positions.covered_log_sequence,
)?;
Some(PendingSpecificFateTerminal {
terminal,
source: saved.source,
})
}
};
let intent = restored_intent(&binding_fate)?;
values.insert(
id,
PendingSpecificFate {
died_source_sequence,
intent,
terminal,
binding_fate,
},
);
}
body.finish()?;
Ok(values)
}
fn restored_intent(
fate: &super::PendingBindingFate,
) -> Result<StoredSpecificFateIntent, CheckpointError> {
match fate.token.intent() {
Some(liminal_protocol::lifecycle::SealedBindingFateIntent::Ordinary) => {
Ok(StoredSpecificFateIntent::Ordinary {
attached_source_sequence: fate.attached_source_sequence,
})
}
Some(liminal_protocol::lifecycle::SealedBindingFateIntent::Recovered {
prior_binding_epoch,
marker_delivery_seq,
}) => Ok(StoredSpecificFateIntent::Recovered {
attached_source_sequence: fate.attached_source_sequence,
prior_binding_epoch: prior_binding_epoch.into(),
marker_delivery_seq,
}),
None => Err(invalid("pending fate intent class")),
}
}
pub(super) fn stage_sources(
bytes: &[u8],
positions: &CheckpointPositions,
lifecycle: &mut liminal_protocol::lifecycle::RestoredLifecycle,
fenced: &mut super::checkpoint_owner::CheckpointFencedSources,
) -> Result<(), CheckpointError> {
if bytes.is_empty() {
return Ok(());
}
let mut body = CheckpointBodyDecoder::new(bytes);
let count = body.count("pending specific fates", 24)?;
for _ in 0..count {
let id = body.u64("pending fate participant")?;
body.u64("pending fate Died source")?;
let token = body.bytes("pending fate token")?;
body.bytes("pending fate terminal source")?;
if let Some(source) =
super::checkpoint_binding_fate::stage_native_source(token, positions, id, lifecycle)?
{
if fenced.insert(id, source).is_some() {
return Err(invalid("duplicate pending fate source"));
}
}
}
body.finish()
}
fn restore_terminal(
saved: PendingTerminalSource,
id: u64,
slot: &Slot,
positions: &CheckpointPositions,
) -> Result<liminal_protocol::lifecycle::CommittedDiedTerminal, CheckpointError> {
if saved.audit.terminal_seq >= positions.next_seq
|| saved.audit.transaction_order >= positions.next_order
{
return Err(invalid("pending fate native terminal allocation"));
}
let terminal = CommittedBindingTerminalRestore {
binding: ActiveBinding {
conversation_id: positions.conversation_id,
participant_id: id,
binding_epoch: saved.binding_epoch.to_epoch().map_err(|source| {
CheckpointError::LiveState {
source: Box::new(source),
}
})?,
},
cause: super::checkpoint_finalizers::cause(saved.audit.cause),
transaction_order: saved.audit.transaction_order,
delivery_seq: saved.audit.terminal_seq,
}
.restore()
.map_err(|source| CheckpointError::LiveState {
source: Box::new(TerminalError(source)),
})?;
if slot.member.latest_terminal() != Some(terminal) {
return Err(invalid("pending fate native terminal association"));
}
let CommittedBindingTerminal::Died(terminal) = terminal else {
return Err(invalid("pending fate native death class"));
};
Ok(terminal)
}
fn validate_terminal_source(
source: StoredOrdinaryTerminalSource,
died_source: u64,
last: u64,
) -> Result<(), CheckpointError> {
let (sequence, finalizer) = match source {
StoredOrdinaryTerminalSource::DiedCommitted {
died_source_sequence,
} => (died_source_sequence, None),
StoredOrdinaryTerminalSource::PendingDiedFinalized {
died_source_sequence,
finalizer,
} => (died_source_sequence, Some(finalizer)),
};
if sequence != died_source {
return Err(invalid("pending fate exact Died source"));
}
covered(sequence, last)?;
if let Some(finalizer) = finalizer {
let (StoredPendingDiedFinalizer::Left { source_sequence }
| StoredPendingDiedFinalizer::FencedAttached { source_sequence }
| StoredPendingDiedFinalizer::Drained { source_sequence }) = finalizer;
covered(source_sequence, last)?;
if source_sequence <= died_source {
return Err(invalid("pending fate finalizer source order"));
}
}
Ok(())
}
fn covered(sequence: u64, last: u64) -> Result<(), CheckpointError> {
if sequence == 0 || sequence > last {
return Err(invalid("pending fate source coverage"));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn no_pending_fates_keep_the_canonical_empty_section() -> Result<(), CheckpointError> {
assert!(encode(&BTreeMap::new(), &BTreeMap::new())?.is_empty());
Ok(())
}
#[test]
fn durable_death_source_never_uses_zero_or_an_uncovered_row() {
assert!(covered(0, 4).is_err());
assert!(covered(5, 4).is_err());
assert!(covered(4, 4).is_ok());
}
}