use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};
use super::{PendingBindingFate, RetainedMarkerSource};
use liminal_protocol::lifecycle::{
BindingOrigin, FencedAttachCommit, RestoredLifecycle, SealedBindingFateIntent,
SealedBindingFateToken,
};
#[derive(Debug, thiserror::Error)]
#[error("fenced predecessor codec refused: {0:?}")]
struct PredecessorCodecError(super::super::log::FencedAttachProofRefusal);
impl RetainedMarkerSource {
fn encode_checkpoint(&self) -> Result<Vec<u8>, CheckpointError> {
let mut body = CheckpointBodyEncoder::new();
body.u64(self.log_sequence);
body.bytes("validated marker record", &self.canonical_record)?;
body.bytes(
"validated fenced predecessor",
&self.canonical_fenced_predecessor,
)?;
Ok(body.finish())
}
fn decode_checkpoint(
bytes: &[u8],
attached: u64,
positions: &CheckpointPositions,
) -> Result<Self, CheckpointError> {
let mut body = CheckpointBodyDecoder::new(bytes);
let log_sequence = body.u64("validated marker source sequence")?;
if log_sequence == 0
|| log_sequence >= attached
|| attached > positions.covered_log_sequence
{
return Err(invalid("fenced marker source coverage"));
}
let canonical_record = body.bytes("validated marker record")?.to_vec();
let canonical_fenced_predecessor = body.bytes("validated fenced predecessor")?.to_vec();
body.finish()?;
Ok(Self {
log_sequence,
canonical_record,
canonical_fenced_predecessor,
})
}
pub(super) fn predecessor(
&self,
conversation_id: u64,
participant_id: u64,
) -> Result<liminal_protocol::lifecycle::FencedAttachCommitRestore, CheckpointError> {
super::super::log::StoredFencedAttachProof::decode_checkpoint_source(
&self.canonical_fenced_predecessor,
conversation_id,
participant_id,
)
.map_err(|error| CheckpointError::LiveState {
source: Box::new(PredecessorCodecError(error)),
})
}
pub(super) fn restore_native(
&self,
lifecycle: &mut RestoredLifecycle,
conversation_id: u64,
participant_id: u64,
) -> Result<FencedAttachCommit, CheckpointError> {
let predecessor = self.predecessor(conversation_id, participant_id)?;
lifecycle
.restore_checkpoint_fenced_attach(predecessor, &self.canonical_record)
.map_err(native)
}
}
#[derive(Debug, thiserror::Error)]
#[error("sealed fate native codec refused: {0:?}")]
struct FateCodecError(liminal_protocol::lifecycle::OwnerCheckpointRestoreError);
fn native(error: liminal_protocol::lifecycle::OwnerCheckpointRestoreError) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(FateCodecError(error)),
}
}
pub(super) fn encode_pending(pending: &PendingBindingFate) -> Result<Vec<u8>, CheckpointError> {
if pending.token.is_recovered() && pending.marker_source.is_none() {
return Err(CheckpointError::Position {
field: "recovered fate validated predecessor checkpoint",
});
}
if pending.token.intent().is_none() || pending.attached_source_sequence == 0 {
return Err(invalid("ordinary sealed fate source"));
}
let token = pending.token.encode_checkpoint().map_err(native)?;
let mut body = CheckpointBodyEncoder::new();
body.u64(pending.attached_source_sequence);
body.bytes("sealed fate token", &token)?;
match (&pending.marker_source, pending.token.is_recovered()) {
(Some(source), true) => {
body.bytes("fenced predecessor source", &source.encode_checkpoint()?)?;
}
(None, false) => {}
_ => return Err(invalid("sealed fate predecessor class association")),
}
Ok(body.finish())
}
pub(super) fn validate_retained_source(
pending: &PendingBindingFate,
conversation_id: u64,
participant_id: u64,
covered_log_sequence: u64,
) -> Result<(), CheckpointError> {
let source = pending
.marker_source
.as_ref()
.ok_or_else(|| invalid("absent fenced predecessor source"))?;
if source.log_sequence == 0
|| source.log_sequence >= pending.attached_source_sequence
|| pending.attached_source_sequence > covered_log_sequence
{
return Err(invalid("fenced predecessor source coverage"));
}
let predecessor = source.predecessor(conversation_id, participant_id)?;
if pending.token.intent()
!= Some(SealedBindingFateIntent::Recovered {
prior_binding_epoch: predecessor.prior_binding_epoch,
marker_delivery_seq: predecessor.marker_delivery_seq,
})
{
return Err(invalid("fenced predecessor token association"));
}
Ok(())
}
pub(super) fn stage_native_source(
bytes: &[u8],
positions: &CheckpointPositions,
participant_id: u64,
lifecycle: &mut RestoredLifecycle,
) -> Result<Option<FencedAttachCommit>, CheckpointError> {
let mut body = CheckpointBodyDecoder::new(bytes);
let attached = body.u64("sealed fate Attached source")?;
body.bytes("sealed fate token")?;
if body.is_empty() {
return Ok(None);
}
let source = RetainedMarkerSource::decode_checkpoint(
body.bytes("fenced predecessor source")?,
attached,
positions,
)?;
body.finish()?;
source
.restore_native(lifecycle, positions.conversation_id, participant_id)
.map(Some)
}
pub(super) fn validate_capture(
pending: &PendingBindingFate,
origin: &BindingOrigin,
cursor: u64,
recovered: Option<FencedAttachCommit>,
) -> Result<(), CheckpointError> {
let body = pending.token.encode_checkpoint().map_err(native)?;
let restored = SealedBindingFateToken::restore_checkpoint(&body, origin, cursor, recovered)
.map_err(native)?;
if restored != pending.token {
return Err(invalid("sealed fate native capture association"));
}
Ok(())
}
#[derive(Clone, Copy)]
pub(super) enum FateRestoreIdentity {
Slot {
participant_id: u64,
},
Pending {
participant_id: u64,
died_source_sequence: u64,
},
}
impl FateRestoreIdentity {
const fn missing_commit(
self,
conversation_id: u64,
attached_source_sequence: u64,
) -> CheckpointError {
match self {
Self::Slot { participant_id } => CheckpointError::MissingSlotFencedAttachCommit {
conversation_id,
participant_id,
attached_source_sequence,
},
Self::Pending {
participant_id,
died_source_sequence,
} => CheckpointError::MissingPendingFencedAttachCommit {
conversation_id,
participant_id,
attached_source_sequence,
died_source_sequence,
},
}
}
}
pub(super) fn decode_pending(
bytes: &[u8],
origin: Option<&BindingOrigin>,
cursor: u64,
positions: &CheckpointPositions,
identity: FateRestoreIdentity,
recovered: Option<FencedAttachCommit>,
) -> Result<PendingBindingFate, CheckpointError> {
let mut body = CheckpointBodyDecoder::new(bytes);
let attached_source_sequence = body.u64("sealed fate Attached source")?;
if attached_source_sequence == 0 || attached_source_sequence > positions.covered_log_sequence {
return Err(invalid("sealed fate Attached source coverage"));
}
let token = body.bytes("sealed fate token")?;
let marker_source = if body.is_empty() {
None
} else {
Some(RetainedMarkerSource::decode_checkpoint(
body.bytes("fenced predecessor source")?,
attached_source_sequence,
positions,
)?)
};
body.finish()?;
let origin = origin.ok_or(CheckpointError::Position {
field: "sealed fate validated binding origin",
})?;
let token = SealedBindingFateToken::restore_checkpoint(token, origin, cursor, recovered)
.map_err(|error| match error {
liminal_protocol::lifecycle::OwnerCheckpointRestoreError::Field(
"sealed_fate.missing_validated_fenced_proof",
) => identity.missing_commit(positions.conversation_id, attached_source_sequence),
error => native(error),
})?;
if token.is_recovered() != marker_source.is_some() {
return Err(invalid("sealed fate predecessor class association"));
}
Ok(PendingBindingFate {
marker_source,
attached_source_sequence,
token,
})
}
#[cfg(test)]
mod source_tests;
#[cfg(test)]
mod tests {
use super::super::super::restore_checkpoint::CheckpointSchema;
use super::*;
pub(super) fn positions() -> CheckpointPositions {
CheckpointPositions {
conversation_id: 7,
covered_log_sequence: 4,
next_event_ordinal: 1,
next_log_sequence: 5,
next_seq: 1,
next_order: 0,
next_participant: 1,
extension_sequence: 0,
highest_delivery_seq: 0,
observer_progress: 0,
merged_base_sequence: 4,
merged_extension_sequence: 0,
schema: CheckpointSchema::V4Suffix,
}
}
#[test]
fn missing_fenced_commit_refusals_retain_exact_owner_and_source_identities() {
assert!(matches!(
FateRestoreIdentity::Slot { participant_id: 17 }.missing_commit(59, 23),
CheckpointError::MissingSlotFencedAttachCommit {
conversation_id: 59,
participant_id: 17,
attached_source_sequence: 23,
}
));
assert!(matches!(
FateRestoreIdentity::Pending {
participant_id: 17,
died_source_sequence: 31,
}
.missing_commit(59, 23),
CheckpointError::MissingPendingFencedAttachCommit {
conversation_id: 59,
participant_id: 17,
attached_source_sequence: 23,
died_source_sequence: 31,
}
));
}
#[test]
fn fate_body_refuses_omitted_validated_origin() -> Result<(), CheckpointError> {
let mut body = CheckpointBodyEncoder::new();
body.u64(3);
body.bytes("sealed fate token", &[])?;
assert!(matches!(
decode_pending(
&body.finish(),
None,
0,
&positions(),
FateRestoreIdentity::Slot { participant_id: 0 },
None,
),
Err(CheckpointError::Position {
field: "sealed fate validated binding origin"
})
));
Ok(())
}
#[test]
fn fate_body_refuses_uncovered_source_before_native_restore() -> Result<(), CheckpointError> {
for sequence in [0, 5] {
let mut body = CheckpointBodyEncoder::new();
body.u64(sequence);
body.bytes("sealed fate token", &[])?;
assert!(
decode_pending(
&body.finish(),
None,
0,
&positions(),
FateRestoreIdentity::Slot { participant_id: 0 },
None,
)
.is_err()
);
}
Ok(())
}
#[test]
fn fate_body_refuses_truncated_or_trailing_server_framing() -> Result<(), CheckpointError> {
assert!(
decode_pending(
&[0; 7],
None,
0,
&positions(),
FateRestoreIdentity::Slot { participant_id: 0 },
None,
)
.is_err()
);
let mut body = CheckpointBodyEncoder::new();
body.u64(3);
body.bytes("sealed fate token", &[])?;
body.u8(1);
assert!(
decode_pending(
&body.finish(),
None,
0,
&positions(),
FateRestoreIdentity::Slot { participant_id: 0 },
None,
)
.is_err()
);
Ok(())
}
#[test]
fn recovered_fate_without_predecessor_authority_is_never_encoded_as_ordinary()
-> Result<(), Box<dyn std::error::Error>> {
let fixture =
liminal_protocol::lifecycle::test_support_external::executable_recovered_attach()
.map_err(std::io::Error::other)?;
let pending = PendingBindingFate {
marker_source: None,
attached_source_sequence: 3,
token: fixture.fate_token,
};
assert!(matches!(
encode_pending(&pending),
Err(CheckpointError::Position {
field: "recovered fate validated predecessor checkpoint"
})
));
Ok(())
}
}