use super::super::binding_fate_completion::stored_died_cause;
use super::super::log::{
StoredCommittedTerminalAudit, StoredDiedCause, StoredOrdinaryFate,
StoredOrdinaryTerminalSource, StoredPendingDiedFinalizer,
};
use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::PreparedOrdinaryFinalizer;
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};
use liminal_protocol::lifecycle::{
ActiveBinding, BindingOrigin, CommittedBindingTerminal, CommittedBindingTerminalRestore,
OrdinaryBindingAuthorityRestore, OrdinaryBindingFateRestore, PendingDiedOrdinaryFinalizer,
};
use liminal_protocol::wire::CloseCause;
use std::collections::BTreeMap;
#[derive(Debug, thiserror::Error)]
enum FinalizerCodecError {
#[error("prepared finalizer native codec refused: {0:?}")]
Native(liminal_protocol::lifecycle::OwnerCheckpointRestoreError),
#[error("prepared finalizer terminal refused: {0:?}")]
Terminal(liminal_protocol::lifecycle::StorageRestoreError),
#[error("prepared finalizer durable fields refused: {0}")]
Fields(#[source] serde_json::Error),
}
fn failure(error: FinalizerCodecError) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(error),
}
}
const fn raw(value: &PreparedOrdinaryFinalizer) -> OrdinaryBindingFateRestore {
let terminal = value.terminal;
let binding = ActiveBinding {
conversation_id: terminal.conversation_id(),
participant_id: terminal.participant_id(),
binding_epoch: terminal.binding_epoch(),
};
OrdinaryBindingFateRestore {
authority: OrdinaryBindingAuthorityRestore {
binding,
through_seq: value.fate.through_seq(),
},
terminal: CommittedBindingTerminalRestore {
binding,
cause: terminal.cause().close_cause(),
transaction_order: terminal.admission_order().transaction_order(),
delivery_seq: terminal.delivery_seq(),
},
resulting_floor: value.fate.resulting_floor(),
}
}
pub(super) fn validate_capture(
values: &BTreeMap<u64, PreparedOrdinaryFinalizer>,
origins: &BTreeMap<u64, (BindingOrigin, u64)>,
) -> Result<(), CheckpointError> {
for (&id, value) in values {
let (origin, sequence) = origins.get(&id).ok_or(CheckpointError::Position {
field: "prepared finalizer validated binding origin",
})?;
if value.attached_source_sequence != *sequence {
return Err(invalid(
"prepared finalizer binding-origin source association",
));
}
let fate =
OrdinaryBindingFateRestore::restore_checkpoint(&raw(value).encode_checkpoint(), origin)
.map_err(|error| failure(FinalizerCodecError::Native(error)))?;
if fate != value.fate {
return Err(invalid("prepared finalizer native fate association"));
}
PendingDiedOrdinaryFinalizer::restore_checkpoint(
&value.finalizer.encode_checkpoint(),
&fate,
)
.map_err(|error| failure(FinalizerCodecError::Native(error)))?;
}
Ok(())
}
pub(super) fn encode(
values: &BTreeMap<u64, PreparedOrdinaryFinalizer>,
) -> Result<Vec<u8>, CheckpointError> {
if values.is_empty() {
return Ok(Vec::new());
}
let mut body = CheckpointBodyEncoder::new();
body.count("prepared finalizers", values.len())?;
for (&id, value) in values {
if value.fate.participant_id() != id
|| value.terminal.participant_id() != id
|| value.fate.conversation_id() != value.terminal.conversation_id()
|| value.fate.last_dead_binding_epoch() != value.terminal.binding_epoch()
{
return Err(invalid("prepared finalizer terminal association"));
}
let row = StoredOrdinaryFate {
participant_id: id,
last_dead_binding_epoch: value.fate.last_dead_binding_epoch().into(),
ordinary_attached_source_sequence: value.attached_source_sequence,
terminal_source: value.terminal_source,
committed_terminal_audit: StoredCommittedTerminalAudit {
cause: stored_died_cause(value.terminal.cause()),
transaction_order: value.terminal.admission_order().transaction_order(),
terminal_seq: value.terminal.delivery_seq(),
},
resulting_floor: value.fate.resulting_floor(),
};
body.u64(id);
body.bytes("ordinary fate", &raw(value).encode_checkpoint())?;
body.bytes(
"ordinary durable source",
&serde_json::to_vec(&row)
.map_err(|error| failure(FinalizerCodecError::Fields(error)))?,
)?;
body.bytes("ordinary finalizer", &value.finalizer.encode_checkpoint())?;
}
Ok(body.finish())
}
pub(super) fn decode(
bytes: &[u8],
origins: &BTreeMap<u64, (BindingOrigin, u64)>,
positions: &CheckpointPositions,
) -> Result<BTreeMap<u64, PreparedOrdinaryFinalizer>, CheckpointError> {
let mut values = BTreeMap::new();
if bytes.is_empty() {
return Ok(values);
}
let mut body = CheckpointBodyDecoder::new(bytes);
let count = body.count("prepared finalizers", 20)?;
let mut previous = None;
for _ in 0..count {
let id = body.u64("prepared finalizer participant")?;
if id >= positions.next_participant || previous.is_some_and(|last| last >= id) {
return Err(invalid("prepared finalizer participant order"));
}
previous = Some(id);
let (origin, source_sequence) = origins.get(&id).ok_or(CheckpointError::Position {
field: "prepared finalizer validated binding origin",
})?;
let fate_body = body.bytes("ordinary fate")?;
let fate = OrdinaryBindingFateRestore::restore_checkpoint(fate_body, origin)
.map_err(|error| failure(FinalizerCodecError::Native(error)))?;
let source_body = body.bytes("ordinary durable source")?;
let row: StoredOrdinaryFate = serde_json::from_slice(source_body)
.map_err(|error| failure(FinalizerCodecError::Fields(error)))?;
validate_source(&row, *source_sequence, id, positions)?;
if fate.participant_id() != id
|| fate.conversation_id() != positions.conversation_id
|| fate.last_dead_binding_epoch()
!= row.last_dead_binding_epoch.to_epoch().map_err(|source| {
CheckpointError::LiveState {
source: Box::new(source),
}
})?
|| fate.resulting_floor() != row.resulting_floor
{
return Err(invalid("prepared finalizer fate association"));
}
let terminal = CommittedBindingTerminalRestore {
binding: ActiveBinding {
conversation_id: positions.conversation_id,
participant_id: id,
binding_epoch: fate.last_dead_binding_epoch(),
},
cause: cause(row.committed_terminal_audit.cause),
transaction_order: row.committed_terminal_audit.transaction_order,
delivery_seq: row.committed_terminal_audit.terminal_seq,
}
.restore()
.map_err(|error| failure(FinalizerCodecError::Terminal(error)))?;
let CommittedBindingTerminal::Died(terminal) = terminal else {
return Err(invalid("prepared finalizer death class"));
};
let finalizer = PendingDiedOrdinaryFinalizer::restore_checkpoint(
body.bytes("ordinary finalizer")?,
&fate,
)
.map_err(|error| failure(FinalizerCodecError::Native(error)))?;
let value = PreparedOrdinaryFinalizer {
attached_source_sequence: row.ordinary_attached_source_sequence,
terminal,
terminal_source: row.terminal_source,
fate,
finalizer,
};
if raw(&value).encode_checkpoint() != fate_body
|| serde_json::to_vec(&row)
.map_err(|error| failure(FinalizerCodecError::Fields(error)))?
!= source_body
{
return Err(invalid("prepared finalizer canonical source"));
}
values.insert(id, value);
}
body.finish()?;
Ok(values)
}
fn validate_source(
row: &StoredOrdinaryFate,
source_sequence: u64,
id: u64,
positions: &CheckpointPositions,
) -> Result<(), CheckpointError> {
covered(row.ordinary_attached_source_sequence, positions)?;
if row.participant_id != id
|| row.ordinary_attached_source_sequence != source_sequence
|| row.committed_terminal_audit.terminal_seq >= positions.next_seq
|| row.committed_terminal_audit.transaction_order >= positions.next_order
|| row.resulting_floor > positions.next_seq
{
return Err(invalid("prepared finalizer source association"));
}
match row.terminal_source {
StoredOrdinaryTerminalSource::DiedCommitted {
died_source_sequence,
} => covered(died_source_sequence, positions),
StoredOrdinaryTerminalSource::PendingDiedFinalized {
died_source_sequence,
finalizer,
} => {
covered(died_source_sequence, positions)?;
let (StoredPendingDiedFinalizer::Left { source_sequence }
| StoredPendingDiedFinalizer::FencedAttached { source_sequence }
| StoredPendingDiedFinalizer::Drained { source_sequence }) = finalizer;
covered(source_sequence, positions)
}
}
}
fn covered(sequence: u64, positions: &CheckpointPositions) -> Result<(), CheckpointError> {
if sequence == 0 || sequence > positions.covered_log_sequence {
return Err(invalid("prepared finalizer source coverage"));
}
Ok(())
}
pub(super) const fn cause(cause: StoredDiedCause) -> CloseCause {
match cause {
StoredDiedCause::ConnectionLost => CloseCause::ConnectionLost,
StoredDiedCause::ProcessKilled => CloseCause::ProcessKilled,
StoredDiedCause::ProtocolError => CloseCause::ProtocolError,
StoredDiedCause::UncleanServerRestart {
prior_server_incarnation,
} => CloseCause::UncleanServerRestart {
prior_server_incarnation,
},
}
}
#[cfg(test)]
mod tests {
use super::super::super::restore_checkpoint::CheckpointSchema;
use super::*;
fn positions() -> CheckpointPositions {
CheckpointPositions {
conversation_id: 7,
covered_log_sequence: 4,
next_event_ordinal: 1,
next_log_sequence: 5,
next_seq: 10,
next_order: 10,
next_participant: 1,
extension_sequence: 0,
highest_delivery_seq: 9,
observer_progress: 0,
merged_base_sequence: 4,
merged_extension_sequence: 0,
schema: CheckpointSchema::V4Suffix,
}
}
#[test]
fn empty_finalizers_preserve_the_existing_empty_body() -> Result<(), CheckpointError> {
let values = BTreeMap::new();
assert!(encode(&values)?.is_empty());
assert!(decode(&[], &BTreeMap::new(), &positions())?.is_empty());
Ok(())
}
#[test]
fn finalizer_decoder_refuses_an_unclosed_participant_frame() {
assert!(decode(&[1, 0, 0, 0], &BTreeMap::new(), &positions()).is_err());
}
#[test]
fn finalizer_source_must_be_a_covered_durable_row() {
assert!(covered(0, &positions()).is_err());
assert!(covered(5, &positions()).is_err());
assert!(covered(4, &positions()).is_ok());
}
}