use super::super::facts::Digest;
use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};
use super::{AttachProvenanceRecord, AttachReceiptState, RetainedSource, Slot};
use liminal_protocol::lifecycle::{
BindingOrigin, BindingState, CredentialAttachLiveReceipt, DetachCell, EnrollmentLiveReceipt,
LiveMember,
};
use liminal_protocol::wire::{
AttachAttemptToken, AttachSecret, DetachAttemptToken, Generation, ReceiptExpiryReason,
};
use std::collections::BTreeMap;
#[derive(Debug, thiserror::Error)]
#[error("slot receipt codec refused: {0:?}")]
struct ReceiptCodecError(liminal_protocol::wire::CodecError);
fn wire(error: liminal_protocol::wire::CodecError) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(ReceiptCodecError(error)),
}
}
pub(super) struct SlotRestoreParts {
pub(super) retained_source: RetainedSource,
enrollment_receipt: EnrollmentLiveReceipt,
enrollment_receipt_expires_at: u128,
enrollment_provenance_expires_at: u128,
enrollment_receipt_ended: Option<ReceiptExpiryReason>,
enrollment_provenance_displaced: bool,
attach: Option<AttachReceiptState>,
attach_provenance: BTreeMap<[u8; 16], AttachProvenanceRecord>,
attach_secret: AttachSecret,
exact_detach_token: Option<DetachAttemptToken>,
binding_fate: Option<Vec<u8>>,
}
impl Slot {
pub(super) fn encode_checkpoint(&self) -> Result<Vec<u8>, CheckpointError> {
if self.enrollment_receipt.committed() != &self.enrollment_outcome {
return Err(invalid("enrollment receipt outcome association"));
}
let mut encoder = CheckpointBodyEncoder::new();
encoder.u64(self.checkpoint_source().log_sequence);
encoder.bytes(
"binding origin source",
&self.checkpoint_source().binding_origin,
)?;
encoder.bytes(
"enrollment receipt",
&self.enrollment_receipt.encode_committed_body(),
)?;
encoder.raw(&self.enrollment_receipt_expires_at.to_le_bytes());
encoder.raw(&self.enrollment_provenance_expires_at.to_le_bytes());
encoder.u8(reason(self.enrollment_receipt_ended));
encoder.u8(u8::from(self.enrollment_provenance_displaced));
encode_attach(&mut encoder, self.attach.as_ref())?;
encoder.count("attach provenance", self.attach_provenance.len())?;
for (token, record) in &self.attach_provenance {
encoder.raw(token);
encoder.u64(record.result_generation.get());
encoder.u8(reason(Some(record.reason)));
encoder.raw(&record.provenance_expires_at.to_le_bytes());
}
encoder.raw(self.attach_secret.as_bytes());
encoder.u8(u8::from(self.exact_detach_token.is_some()));
if let Some(token) = self.exact_detach_token.as_ref() {
encoder.raw(token.as_bytes());
}
let fate = self
.binding_fate
.as_ref()
.map(super::checkpoint_binding_fate::encode_pending)
.transpose()?;
encoder.bytes("slot sealed fate", fate.as_deref().unwrap_or(&[]))?;
Ok(encoder.finish())
}
}
impl SlotRestoreParts {
pub(super) fn stage_fenced_source(
&self,
participant_id: u64,
positions: &CheckpointPositions,
lifecycle: &mut liminal_protocol::lifecycle::RestoredLifecycle,
) -> Result<Option<liminal_protocol::lifecycle::FencedAttachCommit>, CheckpointError> {
self.binding_fate
.as_deref()
.map(|bytes| {
super::checkpoint_binding_fate::stage_native_source(
bytes,
positions,
participant_id,
lifecycle,
)
})
.transpose()
.map(Option::flatten)
}
pub(super) fn decode_checkpoint(
bytes: &[u8],
participant_id: u64,
positions: &CheckpointPositions,
) -> Result<Self, CheckpointError> {
let mut decoder = CheckpointBodyDecoder::new(bytes);
let log_sequence = decoder.u64("slot source sequence")?;
if log_sequence == 0 || log_sequence > positions.covered_log_sequence {
return Err(invalid("slot source coverage"));
}
let binding_origin = decoder.bytes("binding origin source")?.to_vec();
let enrollment_receipt =
EnrollmentLiveReceipt::decode_committed_body(decoder.bytes("enrollment receipt")?)
.map_err(wire)?;
let outcome = enrollment_receipt.committed();
if outcome.conversation_id() != positions.conversation_id
|| outcome.participant_id() != participant_id
{
return Err(invalid("enrollment receipt identity"));
}
let enrollment_receipt_expires_at = wide(&mut decoder)?;
let enrollment_provenance_expires_at = wide(&mut decoder)?;
if enrollment_receipt_expires_at != outcome.receipt_expires_at()
|| enrollment_provenance_expires_at != outcome.provenance_expires_at()
{
return Err(invalid("enrollment receipt deadlines"));
}
let enrollment_receipt_ended = decode_reason(decoder.u8("enrollment receipt reason")?)?;
let enrollment_provenance_displaced = boolean(&mut decoder)?;
let attach = decode_attach(&mut decoder, participant_id, positions.conversation_id)?;
let count = decoder.count("attach provenance", 41)?;
let mut attach_provenance = BTreeMap::new();
let mut previous = None;
for _ in 0..count {
let token = fixed::<16>(&mut decoder)?;
if previous.is_some_and(|value| value >= token) {
return Err(invalid("attach provenance order"));
}
previous = Some(token);
let result_generation = Generation::new(decoder.u64("provenance generation")?)
.ok_or_else(|| invalid("provenance generation"))?;
let reason = decode_reason(decoder.u8("attach provenance reason")?)?
.ok_or_else(|| invalid("attach provenance reason"))?;
let provenance_expires_at = wide(&mut decoder)?;
attach_provenance.insert(
token,
AttachProvenanceRecord {
result_generation,
reason,
provenance_expires_at,
},
);
}
let attach_secret = AttachSecret::new(fixed::<32>(&mut decoder)?);
let exact_detach_token = if boolean(&mut decoder)? {
Some(DetachAttemptToken::new(fixed::<16>(&mut decoder)?))
} else {
None
};
let fate = decoder.bytes("slot sealed fate")?;
let binding_fate = (!fate.is_empty()).then(|| fate.to_vec());
decoder.finish()?;
Ok(Self {
retained_source: RetainedSource {
log_sequence,
binding_origin,
},
enrollment_receipt,
enrollment_receipt_expires_at,
enrollment_provenance_expires_at,
enrollment_receipt_ended,
enrollment_provenance_displaced,
attach,
attach_provenance,
attach_secret,
exact_detach_token,
binding_fate,
})
}
pub(super) fn install(
self,
member: LiveMember<Digest>,
binding: BindingState,
cell: DetachCell<Digest>,
origin: Option<&BindingOrigin>,
positions: &CheckpointPositions,
recovered: Option<liminal_protocol::lifecycle::FencedAttachCommit>,
) -> Result<Slot, CheckpointError> {
if self.enrollment_receipt.committed().participant_id() != member.participant_id()
|| self.enrollment_receipt.committed().conversation_id() != member.conversation_id()
|| self.attach_secret != member.attach_secret()
{
return Err(invalid("slot member association"));
}
let binding_fate = self
.binding_fate
.as_deref()
.map(|bytes| {
super::checkpoint_binding_fate::decode_pending(
bytes,
origin,
member.cursor(),
positions,
super::checkpoint_binding_fate::FateRestoreIdentity::Slot {
participant_id: member.participant_id(),
},
recovered,
)
})
.transpose()?;
if binding_fate
.as_ref()
.is_some_and(|fate| fate.attached_source_sequence != self.retained_source.log_sequence)
{
return Err(invalid("slot fate binding-origin source association"));
}
let enrollment_outcome = self.enrollment_receipt.clone().into_committed();
Ok(Slot {
retained_source: self.retained_source,
member,
binding,
binding_fate,
cell,
enrollment_receipt: self.enrollment_receipt,
enrollment_outcome,
enrollment_receipt_expires_at: self.enrollment_receipt_expires_at,
enrollment_provenance_expires_at: self.enrollment_provenance_expires_at,
enrollment_receipt_ended: self.enrollment_receipt_ended,
enrollment_provenance_displaced: self.enrollment_provenance_displaced,
attach: self.attach,
attach_provenance: self.attach_provenance,
attach_secret: self.attach_secret,
exact_detach_token: self.exact_detach_token,
})
}
}
fn encode_attach(
encoder: &mut CheckpointBodyEncoder,
attach: Option<&AttachReceiptState>,
) -> Result<(), CheckpointError> {
encoder.u8(u8::from(attach.is_some()));
let Some(attach) = attach else { return Ok(()) };
if attach.receipt.committed() != &attach.outcome {
return Err(invalid("attach receipt outcome association"));
}
encoder.bytes("attach receipt", &attach.receipt.encode_committed_body())?;
encoder.raw(attach.token.as_bytes());
encoder.raw(&attach.verifier);
encoder.u64(attach.result_generation.get());
encoder.raw(&attach.receipt_expires_at.to_le_bytes());
encoder.raw(&attach.provenance_expires_at.to_le_bytes());
Ok(())
}
fn decode_attach(
decoder: &mut CheckpointBodyDecoder<'_>,
participant_id: u64,
conversation_id: u64,
) -> Result<Option<AttachReceiptState>, CheckpointError> {
if !boolean(decoder)? {
return Ok(None);
}
let receipt =
CredentialAttachLiveReceipt::decode_committed_body(decoder.bytes("attach receipt")?)
.map_err(wire)?;
let token = AttachAttemptToken::new(fixed::<16>(decoder)?);
let verifier = fixed::<32>(decoder)?;
let result_generation = Generation::new(decoder.u64("attach generation")?)
.ok_or_else(|| invalid("attach generation"))?;
let receipt_expires_at = wide(decoder)?;
let provenance_expires_at = wide(decoder)?;
let outcome = receipt.committed();
if outcome.conversation_id() != conversation_id
|| outcome.participant_id() != participant_id
|| outcome.token() != token
|| outcome.capability_generation() != result_generation
|| outcome.receipt_expires_at() != receipt_expires_at
|| outcome.provenance_expires_at() != provenance_expires_at
{
return Err(invalid("attach receipt association"));
}
let outcome = receipt.clone().into_committed();
Ok(Some(AttachReceiptState {
token,
receipt,
outcome,
verifier,
result_generation,
receipt_expires_at,
provenance_expires_at,
}))
}
const fn reason(reason: Option<ReceiptExpiryReason>) -> u8 {
match reason {
None => 0,
Some(ReceiptExpiryReason::Deadline) => 1,
Some(ReceiptExpiryReason::Superseded) => 2,
}
}
fn decode_reason(tag: u8) -> Result<Option<ReceiptExpiryReason>, CheckpointError> {
match tag {
0 => Ok(None),
1 => Ok(Some(ReceiptExpiryReason::Deadline)),
2 => Ok(Some(ReceiptExpiryReason::Superseded)),
_ => Err(invalid("receipt reason")),
}
}
fn boolean(decoder: &mut CheckpointBodyDecoder<'_>) -> Result<bool, CheckpointError> {
match decoder.u8("slot flag")? {
0 => Ok(false),
1 => Ok(true),
_ => Err(invalid("slot flag")),
}
}
fn fixed<const N: usize>(
decoder: &mut CheckpointBodyDecoder<'_>,
) -> Result<[u8; N], CheckpointError> {
decoder
.take("slot fixed field", N)?
.try_into()
.map_err(|_| invalid("slot fixed field"))
}
fn wide(decoder: &mut CheckpointBodyDecoder<'_>) -> Result<u128, CheckpointError> {
Ok(u128::from_le_bytes(fixed::<16>(decoder)?))
}
#[cfg(test)]
mod tests {
use super::*;
use liminal_protocol::wire::{
BindingEpoch, ConnectionIncarnation, EnrollBound, EnrollmentToken,
};
#[test]
fn borrowed_enrollment_receipt_matches_the_existing_move_writer() -> Result<(), CheckpointError>
{
let outcome = EnrollBound::new(
7,
EnrollmentToken::new([2; 16]),
0,
AttachSecret::new([3; 32]),
BindingEpoch::new(ConnectionIncarnation::new(1, 1), Generation::ONE),
5,
8,
)
.ok_or_else(|| invalid("fixture enrollment"))?;
let receipt = EnrollmentLiveReceipt::from_commit(outcome);
let borrowed = receipt.encode_committed_body();
let moved =
EnrollmentLiveReceipt::from_commit(receipt.into_committed()).encode_committed_body();
assert_eq!(borrowed, moved);
let restored = EnrollmentLiveReceipt::decode_committed_body(&borrowed).map_err(wire)?;
assert_eq!(restored.encode_committed_body(), borrowed);
Ok(())
}
}