use liminal_protocol::lifecycle::{
BindingOrigin, ConversationLifecycleView, FrontierCheckpointSource, LifecycleCapsule,
LifecycleView, LiveFrontierOwner, ParticipantLifecycleView, RestoredCheckpointConversation,
RestoredParticipantLifecycle, SealedLifecycleView,
};
use super::super::barrier::ReceiptCapacityLimits;
use super::super::fate_occurrence::FateOccurrenceRouter;
use super::super::fenced_attach_terminal::ComposedTerminalValidation;
use super::super::outbox::{ConversationOutbox, ConversationOutboxLimits};
use super::super::restore_checkpoint::{CheckpointError, CheckpointLiveState, CheckpointPositions};
use super::super::restore_checkpoint_history::CheckpointHistoryRoot;
use super::ConversationAuthority;
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointOwnerBody, invalid};
use super::checkpoint_decode::{decode_admissions, decode_retired, decode_tokens};
use super::checkpoint_slots::SlotRestoreParts;
use crate::config::types::ParticipantConfig;
#[derive(Debug, thiserror::Error)]
enum ProtocolCheckpointError {
#[error("lifecycle decode refused: {0:?}")]
Decode(liminal_protocol::lifecycle::CapsuleDecodeError),
#[error("binding source capture refused: {0:?}")]
BindingCapture(liminal_protocol::lifecycle::BindingOriginCaptureError),
#[error("binding source restore refused: {0:?}")]
BindingRestore(liminal_protocol::lifecycle::BindingOriginSourceError),
#[error("frontier restore refused: {0:?}")]
Frontier(liminal_protocol::lifecycle::LiveFrontierRestoreError),
}
const fn refused(field: &'static str) -> CheckpointError {
CheckpointError::Position { field }
}
fn protocol(error: ProtocolCheckpointError) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(error),
}
}
pub(in crate::server::participant::production) struct ConversationAuthorityRestoreContext {
positions: CheckpointPositions,
history_root: CheckpointHistoryRoot,
receipt_limits: ReceiptCapacityLimits,
outbox_limits: ConversationOutboxLimits,
retained_record_limit: u64,
marker_max: liminal_protocol::algebra::ResourceVector,
identity_slots: u64,
}
impl ConversationAuthorityRestoreContext {
pub(in crate::server::participant::production) fn new(
positions: CheckpointPositions,
history_root: &[u8],
config: &ParticipantConfig,
) -> Result<Self, CheckpointError> {
positions.validate()?;
let history_root =
CheckpointHistoryRoot::decode_canonical(history_root, positions.conversation_id)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
let outbox_limits = ConversationOutboxLimits::try_new(
config.max_retained_record_rows,
config.identity_slots,
)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
Ok(Self {
positions,
history_root,
outbox_limits,
retained_record_limit: config.max_retained_record_rows,
marker_max: liminal_protocol::algebra::ResourceVector::new(
config.max_generated_marker_entries,
config.max_generated_marker_bytes,
),
identity_slots: config.identity_slots,
receipt_limits: ReceiptCapacityLimits::from_config(config),
})
}
fn validate_association(
&self,
positions: &CheckpointPositions,
history_root: &[u8],
) -> Result<(), CheckpointError> {
if self.positions != *positions {
return Err(CheckpointError::Association {
field: "owner context positions",
});
}
let root = CheckpointHistoryRoot::decode_canonical(history_root, positions.conversation_id)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
if root != self.history_root {
return Err(CheckpointError::Association {
field: "owner context history root",
});
}
Ok(())
}
}
impl ConversationAuthority {
pub(in crate::server::participant::production) fn retain_checkpoint_metadata(
&mut self,
root: CheckpointHistoryRoot,
audit: &ComposedTerminalValidation,
) -> Result<(), CheckpointError> {
if root.conversation_id() != self.conversation_id {
return Err(invalid("retained root conversation"));
}
self.checkpoint_audit_source = Some(audit.encode_checkpoint()?);
self.checkpoint_history_root = Some(root);
Ok(())
}
fn validate_checkpoint_fenced_sources(&self) -> Result<(), CheckpointError> {
let covered = self
.next_log_sequence
.checked_sub(1)
.ok_or_else(|| invalid("uncommitted checkpoint source"))?;
for (&participant_id, slot) in &self.slots {
if let Some(fate) = slot
.binding_fate
.as_ref()
.filter(|fate| fate.token.is_recovered())
{
if fate.marker_source.is_none() {
return Err(CheckpointError::MissingSlotFencedAttachCommit {
conversation_id: self.conversation_id,
participant_id,
attached_source_sequence: fate.attached_source_sequence,
});
}
super::checkpoint_binding_fate::validate_retained_source(
fate,
self.conversation_id,
participant_id,
covered,
)?;
}
}
for (&participant_id, fate) in self.checkpoint_specific_fates() {
if fate.binding_fate.token.is_recovered() {
if fate.binding_fate.marker_source.is_none() {
return Err(CheckpointError::MissingPendingFencedAttachCommit {
conversation_id: self.conversation_id,
participant_id,
attached_source_sequence: fate.binding_fate.attached_source_sequence,
died_source_sequence: fate.died_source_sequence,
});
}
super::checkpoint_binding_fate::validate_retained_source(
&fate.binding_fate,
self.conversation_id,
participant_id,
covered,
)?;
}
}
if self.slots.values().any(|slot| {
slot.binding_fate
.as_ref()
.is_some_and(|fate| fate.token.is_recovered())
}) || self
.pending_specific_fates
.values()
.any(|fate| fate.binding_fate.token.is_recovered())
{
let captured = self.capture_checkpoint_lifecycle()?;
self.validate_checkpoint_lifecycle_capture(
captured.capsule,
&captured.sources,
captured.covered_log_sequence,
)?;
}
Ok(())
}
fn checkpoint_capture_supported(&self) -> Result<(), CheckpointError> {
if self.checkpoint_transition_pending() {
return Err(refused("pending debt dispatch transition"));
}
self.validate_checkpoint_fenced_sources()?;
if self.obligation_debt_dispatch.is_some()
&& (self.checkpoint_audit_source.is_none() || self.checkpoint_history_root.is_none())
{
return Err(refused("retained audit and history-root authority"));
}
if let Some(refusal) = self
.checkpoint_observer_witnesses()
.checkpoint_source_refusal(self.conversation_id)
{
return Err(refusal);
}
let shell = self
.shell
.as_ref()
.ok_or_else(|| refused("absent conversation shell"))?;
if self.obligation_debt_dispatch.is_none()
&& (!self.slots.is_empty() || (!self.retired.is_empty() && !self.closed))
{
return Err(invalid("omitted frontier owner"));
}
if self.closed && self.obligation_debt_dispatch.is_some() {
return Err(invalid("sealed frontier owner"));
}
if self.closed != self.sealed_ledger_high_watermark.is_some() {
return Err(invalid("sealed watermark presence"));
}
let needs_metadata = self.obligation_debt_dispatch.is_some()
|| !shell.genesis_validated()
|| shell.next_event_ordinal() != 1
|| self.next_log_sequence != 1
|| self.next_participant != 0
|| self.next_order != 0
|| self.next_seq != 1
|| self.observer_progress != 0
|| self.closed
|| !self.tokens.is_empty()
|| !self.committed_admissions.is_empty();
if needs_metadata
&& (self.checkpoint_audit_source.is_none() || self.checkpoint_history_root.is_none())
{
return Err(refused("retained audit and history-root authority"));
}
Ok(())
}
}
impl CheckpointLiveState for ConversationAuthority {
type RestoreContext = ConversationAuthorityRestoreContext;
fn encode_live(&self) -> Result<Vec<u8>, CheckpointError> {
self.checkpoint_capture_supported()?;
let CapturedCheckpointLifecycle {
capsule,
sources,
covered_log_sequence,
} = self.capture_checkpoint_lifecycle()?;
let capsule_bytes = capsule.encode_canonical();
self.validate_checkpoint_lifecycle_capture(capsule, &sources, covered_log_sequence)?;
let capsule = capsule_bytes;
let mut slots = super::checkpoint_body::CheckpointBodyEncoder::new();
if !self.slots.is_empty() {
slots.count("slot count", self.slots.len())?;
for (&id, slot) in &self.slots {
slots.u64(id);
slots.bytes("slot", &slot.encode_checkpoint()?)?;
}
}
let slots = slots.finish();
let outbox = self
.outbox
.as_ref()
.ok_or_else(|| refused("absent outbox"))?
.encode_checkpoint()?;
let identities = self.encode_checkpoint_identities()?;
let debt = self.encode_checkpoint_debt()?;
let fates = self
.fate_occurrences
.encode_checkpoint(covered_log_sequence)?;
let finalizers = super::checkpoint_finalizers::encode(self.checkpoint_finalizers())?;
let pending_fates = super::checkpoint_specific_fates::encode(
self.checkpoint_specific_fates(),
&self.slots,
)?;
let observer_witnesses = self
.checkpoint_observer_witnesses()
.encode_checkpoint(self.conversation_id, self.observer_progress)?;
CheckpointOwnerBody {
conversation_id: self.conversation_id,
next_order: self.next_order,
next_seq: self.next_seq,
next_participant: self.next_participant,
next_log_sequence: self.next_log_sequence,
observer_progress: self.observer_progress,
closed: self.closed,
lifecycle_capsule: &capsule,
debt_dispatch: &debt,
outbox: &outbox,
slots: &slots,
pending_specific_fates: &pending_fates,
prepared_ordinary_finalizers: &finalizers,
fate_occurrences: &fates,
retired: &identities.retired,
tokens: &identities.tokens,
committed_admissions: &identities.committed_admissions,
observer_progress_witnesses: &observer_witnesses,
terminal_validation: self.checkpoint_audit_source.as_deref().unwrap_or(&[]),
}
.encode_canonical()
}
fn decode_live(
bytes: &[u8],
positions: &CheckpointPositions,
history_root: &[u8],
context: &Self::RestoreContext,
) -> Result<Self, CheckpointError> {
context.validate_association(positions, history_root)?;
let body = CheckpointOwnerBody::decode_canonical(bytes)?;
body.validate_positions(positions)?;
let (mut owner, origins, mut fenced) =
Self::restore_checkpoint_frontier(&body, positions, context)?;
if !body.terminal_validation.is_empty() {
let audit = ComposedTerminalValidation::decode_checkpoint(
body.terminal_validation,
context.identity_slots,
positions.covered_log_sequence,
)?;
owner.retain_checkpoint_metadata(context.history_root, &audit)?;
}
owner.outbox = Some(ConversationOutbox::decode_checkpoint(
body.outbox,
positions.conversation_id,
positions.extension_sequence,
context.outbox_limits,
)?);
owner.tokens = decode_tokens(body.tokens, positions)?;
owner.committed_admissions = decode_admissions(body.committed_admissions, positions)?;
let retired = decode_retired(body.retired, positions)?;
if (owner.obligation_debt_dispatch.is_some()
|| owner.sealed_ledger_high_watermark.is_some())
&& owner.retired != retired
{
return Err(invalid("retired capsule association"));
}
owner.retired = retired;
for (&id, slot) in &owner.slots {
if owner
.tokens
.get(slot.enrollment_receipt.committed().token().as_bytes())
!= Some(&id)
{
return Err(invalid("slot enrollment-token association"));
}
}
owner.fate_occurrences =
FateOccurrenceRouter::decode_checkpoint(body.fate_occurrences, positions)?;
owner.prepared_ordinary_finalizers = super::checkpoint_finalizers::decode(
body.prepared_ordinary_finalizers,
&origins,
positions,
)?;
owner.pending_specific_fates = super::checkpoint_specific_fates::decode(
body.pending_specific_fates,
&owner.slots,
&origins,
positions,
&mut fenced,
)?;
if !fenced.is_empty() {
return Err(invalid("unconsumed fenced source"));
}
if owner
.pending_specific_fates
.keys()
.any(|id| owner.prepared_ordinary_finalizers.contains_key(id))
{
return Err(invalid("duplicate pending and prepared fate authority"));
}
owner.closed = body.closed;
owner.next_order = body.next_order;
owner.next_seq = body.next_seq;
owner.next_participant = body.next_participant;
owner.next_log_sequence = body.next_log_sequence;
owner.observer_progress = body.observer_progress;
owner.validate_live(positions, history_root)?;
if owner.encode_live()? != bytes {
return Err(invalid("canonical live owner"));
}
Ok(owner)
}
fn validate_live(
&self,
positions: &CheckpointPositions,
history_root: &[u8],
) -> Result<(), CheckpointError> {
self.checkpoint_capture_supported()?;
let actual = self.checkpoint_positions(
positions.schema,
positions.merged_base_sequence,
positions.merged_extension_sequence,
)?;
if actual != *positions {
return Err(CheckpointError::Association {
field: "live owner positions",
});
}
let root = CheckpointHistoryRoot::decode_canonical(history_root, self.conversation_id)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
let expected_root = self
.checkpoint_history_root
.unwrap_or_else(|| CheckpointHistoryRoot::empty(self.conversation_id));
if root != expected_root {
return Err(CheckpointError::Association {
field: "uncaptured history index authority",
});
}
Ok(())
}
}
type CheckpointOrigins = std::collections::BTreeMap<u64, (BindingOrigin, u64)>;
pub(super) type CheckpointFencedSources =
std::collections::BTreeMap<u64, liminal_protocol::lifecycle::FencedAttachCommit>;
struct CapturedCheckpointLifecycle<'a> {
capsule: LifecycleCapsule,
sources: Vec<(u64, &'a [u8])>,
covered_log_sequence: u64,
}
fn decode_checkpoint_slots(
bytes: &[u8],
positions: &CheckpointPositions,
) -> Result<std::collections::BTreeMap<u64, SlotRestoreParts>, CheckpointError> {
let mut slots = std::collections::BTreeMap::new();
if !bytes.is_empty() {
let mut decoder = CheckpointBodyDecoder::new(bytes);
let count = decoder.count("slot count", 12)?;
let mut previous = None;
for _ in 0..count {
let id = decoder.u64("slot participant")?;
if id >= positions.next_participant || previous.is_some_and(|value| value >= id) {
return Err(invalid("slot participant order"));
}
previous = Some(id);
slots.insert(
id,
SlotRestoreParts::decode_checkpoint(decoder.bytes("slot")?, id, positions)?,
);
}
decoder.finish()?;
}
Ok(slots)
}
impl ConversationAuthority {
pub(in crate::server::participant::production) fn checkpoint_lifecycle_capturable(
&self,
) -> Result<(), CheckpointError> {
let captured = self.capture_checkpoint_lifecycle()?;
self.validate_checkpoint_lifecycle_capture(
captured.capsule,
&captured.sources,
captured.covered_log_sequence,
)
}
fn capture_checkpoint_lifecycle(
&self,
) -> Result<CapturedCheckpointLifecycle<'_>, CheckpointError> {
let shell = self
.shell
.as_ref()
.ok_or_else(|| refused("absent conversation shell"))?;
let covered_log_sequence = self
.next_log_sequence
.checked_sub(1)
.ok_or_else(|| refused("uncommitted conversation"))?;
let frontier = self
.checkpoint_debt_owner()
.map(liminal_protocol::lifecycle::ObligationDebtDispatchState::frontier);
let closure = frontier.map(|owner| owner.closure_accounting().state());
let mut participants: Vec<_> = self
.slots
.values()
.map(|slot| ParticipantLifecycleView::Live {
member: &slot.member,
binding: &slot.binding,
binding_origin: None,
detach_cell: &slot.cell,
})
.chain(self.retired.values().map(ParticipantLifecycleView::Retired))
.collect();
participants.sort_unstable_by_key(ParticipantLifecycleView::participant_id);
let sources: Vec<_> = self
.slots
.iter()
.map(|(&id, slot)| (id, slot.retained_source.binding_origin.as_slice()))
.collect();
let (lifecycle, sealed) = match (
frontier,
closure.as_ref(),
self.closed,
self.sealed_ledger_high_watermark,
) {
(Some(frontier), Some(closure), false, None) => (
Some(ConversationLifecycleView {
participants,
frontiers: frontier.frontiers(),
retained_record_limit: frontier.retained_record_limit(),
closure,
}),
None,
),
(None, None, false, None) => (None, None),
(None, None, true, Some(ledger_high_watermark)) => (
None,
Some(SealedLifecycleView {
participants,
ledger_high_watermark,
}),
),
_ => return Err(invalid("frontier closure presence")),
};
let capsule = LifecycleCapsule::capture_with_binding_sources(
&LifecycleView {
conversation: shell,
lifecycle,
sealed,
covered_log_sequence,
},
&sources,
)
.map_err(|error| protocol(ProtocolCheckpointError::BindingCapture(error)))?;
Ok(CapturedCheckpointLifecycle {
capsule,
sources,
covered_log_sequence,
})
}
fn validate_checkpoint_lifecycle_capture(
&self,
capsule: LifecycleCapsule,
sources: &[(u64, &[u8])],
covered_log_sequence: u64,
) -> Result<(), CheckpointError> {
if !self.prepared_ordinary_finalizers.is_empty()
|| !self.pending_specific_fates.is_empty()
|| self.slots.values().any(|slot| slot.binding_fate.is_some())
|| !self
.checkpoint_observer_witnesses()
.checkpoint_is_pristine()
{
let mut restored = capsule
.restore_with_binding_sources(sources)
.map_err(|error| protocol(ProtocolCheckpointError::BindingRestore(error)))?;
let mut fenced = self.stage_capture_fenced_sources(&mut restored)?;
self.checkpoint_observer_witnesses()
.validate_checkpoint_lifecycle(&restored)?;
let restored = restored.into_parts();
let mut origins = std::collections::BTreeMap::new();
if let Some(lifecycle) = restored.lifecycle {
for participant in lifecycle.participants() {
if let RestoredParticipantLifecycle::Live {
member,
binding_origin: Some(origin),
..
} = participant
{
let id = member.participant_id();
let slot = self
.slots
.get(&id)
.ok_or_else(|| invalid("missing source slot"))?;
origins.insert(id, (*origin, slot.retained_source.log_sequence));
}
}
}
for (&id, slot) in &self.slots {
if let Some(fate) = slot.binding_fate.as_ref() {
let (origin, sequence) = origins
.get(&id)
.ok_or_else(|| refused("sealed fate validated binding origin"))?;
if fate.attached_source_sequence != *sequence {
return Err(invalid("slot fate binding-origin source association"));
}
super::checkpoint_binding_fate::validate_capture(
fate,
origin,
slot.member.cursor(),
fenced.remove(&id),
)?;
}
}
super::checkpoint_finalizers::validate_capture(self.checkpoint_finalizers(), &origins)?;
super::checkpoint_specific_fates::validate_capture(
self.checkpoint_specific_fates(),
&self.slots,
&origins,
covered_log_sequence,
&mut fenced,
)?;
if !fenced.is_empty() {
return Err(invalid("unconsumed capture fenced source"));
}
}
Ok(())
}
fn stage_capture_fenced_sources(
&self,
lifecycle: &mut liminal_protocol::lifecycle::RestoredLifecycle,
) -> Result<CheckpointFencedSources, CheckpointError> {
let mut fenced = CheckpointFencedSources::new();
let fates = self
.slots
.iter()
.filter_map(|(&id, slot)| slot.binding_fate.as_ref().map(|fate| (id, fate)))
.chain(
self.pending_specific_fates
.iter()
.map(|(&id, fate)| (id, &fate.binding_fate)),
);
for (id, fate) in fates {
if let Some(source) = &fate.marker_source {
let proof = source.restore_native(lifecycle, self.conversation_id, id)?;
if fenced.insert(id, proof).is_some() {
return Err(invalid("duplicate capture fenced source"));
}
}
}
Ok(fenced)
}
fn restore_checkpoint_frontier(
body: &CheckpointOwnerBody<'_>,
positions: &CheckpointPositions,
context: &ConversationAuthorityRestoreContext,
) -> Result<(Self, CheckpointOrigins, CheckpointFencedSources), CheckpointError> {
let capsule = LifecycleCapsule::decode_canonical(body.lifecycle_capsule)
.map_err(|error| protocol(ProtocolCheckpointError::Decode(error)))?;
if capsule.conversation_id() != positions.conversation_id
|| capsule.covered_log_sequence() != positions.covered_log_sequence
{
return Err(CheckpointError::Association {
field: "lifecycle capsule positions",
});
}
let slots = decode_checkpoint_slots(body.slots, positions)?;
let sources: Vec<_> = slots
.iter()
.map(|(&id, slot)| (id, slot.retained_source.binding_origin.as_slice()))
.collect();
let mut restored = capsule
.restore_with_binding_sources(&sources)
.map_err(|error| protocol(ProtocolCheckpointError::BindingRestore(error)))?;
let mut fenced = CheckpointFencedSources::new();
for (&id, slot) in &slots {
if let Some(source) = slot.stage_fenced_source(id, positions, &mut restored)? {
fenced.insert(id, source);
}
}
super::checkpoint_specific_fates::stage_sources(
body.pending_specific_fates,
positions,
&mut restored,
&mut fenced,
)?;
let observer_witnesses =
super::super::observer_progress::ObserverProgressWitnessState::decode_checkpoint(
body.observer_progress_witnesses,
positions,
&restored,
)?;
let restored = restored.into_parts();
let mut debt = CheckpointBodyDecoder::new(body.debt_dispatch);
let present = debt.u8("debt dispatch presence")?;
let mut owner = Self::empty(positions.conversation_id, context.receipt_limits);
owner.observer_progress_witnesses = observer_witnesses;
let mut origins = std::collections::BTreeMap::new();
match present {
0 => {
debt.finish()?;
if restored.lifecycle.is_some() || !slots.is_empty() {
return Err(invalid("omitted frontier owner"));
}
if restored.sealed.is_some() != body.closed {
return Err(invalid("sealed capsule association"));
}
owner.shell = Some(restored.conversation);
if let Some(sealed) = restored.sealed {
let (retired, ledger_high_watermark) = sealed.into_parts();
for identity in retired {
owner.retired.insert(identity.participant_id(), identity);
}
owner.sealed_ledger_high_watermark = Some(ledger_high_watermark);
}
}
1 => {
if restored.sealed.is_some() {
return Err(invalid("sealed beside frontier owner"));
}
let mut saved = CheckpointBodyDecoder::new(debt.bytes("debt dispatch")?);
debt.finish()?;
let cursor_episode = match saved.u8("episode presence")? {
0 => None,
1 => Some(saved.bytes("debt episode")?),
_ => return Err(invalid("episode presence")),
};
let closure_accounting = saved.bytes("closure accounting")?;
let retained_charges = saved.bytes("retained charges")?;
saved.finish()?;
let restored = LiveFrontierOwner::restore_checkpoint(
restored,
FrontierCheckpointSource {
closure_accounting: Some(closure_accounting),
retained_charges: Some(retained_charges),
cursor_episode,
observer_progress: positions.observer_progress,
retained_record_limit: context.retained_record_limit,
marker_max: context.marker_max,
},
)
.map_err(|error| protocol(ProtocolCheckpointError::Frontier(error)))?
.into_debt_dispatch()
.map_err(|error| protocol(ProtocolCheckpointError::Frontier(error)))?;
origins = owner.install_checkpoint_participants(
restored,
slots,
positions,
&mut fenced,
)?;
}
_ => return Err(invalid("debt dispatch presence")),
}
Ok((owner, origins, fenced))
}
fn install_checkpoint_participants(
&mut self,
restored: RestoredCheckpointConversation,
mut slots: std::collections::BTreeMap<u64, SlotRestoreParts>,
positions: &CheckpointPositions,
fenced: &mut CheckpointFencedSources,
) -> Result<CheckpointOrigins, CheckpointError> {
let mut origins = std::collections::BTreeMap::new();
self.shell = Some(restored.conversation);
self.obligation_debt_dispatch = Some(restored.dispatch);
for participant in restored.participants {
match participant {
RestoredParticipantLifecycle::Retired(identity) => {
self.retired.insert(identity.participant_id(), identity);
}
RestoredParticipantLifecycle::Live {
member,
binding,
detach_cell,
binding_origin,
} => {
let id = member.participant_id();
let parts = slots.remove(&id).ok_or_else(|| invalid("missing slot"))?;
if let Some(origin) = binding_origin {
origins.insert(id, (origin, parts.retained_source.log_sequence));
}
self.slots.insert(
id,
parts.install(
member,
binding,
detach_cell,
binding_origin.as_ref(),
positions,
fenced.remove(&id),
)?,
);
}
}
}
if !slots.is_empty() {
return Err(invalid("foreign slot"));
}
Ok(origins)
}
}
#[cfg(test)]
mod tests {
use super::super::super::restore_checkpoint::{CheckpointSchema, RestoreCheckpoint};
use super::super::super::restore_checkpoint_history::ROOT_ENCODED_BYTES;
use super::*;
use liminal_protocol::lifecycle::{
ConversationDecision, ConversationGenesis, ParticipantConversation,
};
fn genesis_owner() -> Result<ConversationAuthority, CheckpointError> {
let config = crate::server::participant::test_participant_config();
let mut owner =
ConversationAuthority::empty(7, ReceiptCapacityLimits::from_config(&config));
let fresh = ParticipantConversation::from_genesis(ConversationGenesis::new(7));
let ConversationDecision::Commit(commit) = fresh.decide_genesis_validation() else {
return Err(refused("genesis validation"));
};
owner.shell = Some(commit.commit());
owner.next_log_sequence = 1;
let limits = ConversationOutboxLimits::try_new(
config.max_retained_record_rows,
config.identity_slots,
)
.map_err(|source| CheckpointError::LiveState {
source: Box::new(source),
})?;
owner.outbox = Some(ConversationOutbox::restore(7, Vec::new(), limits).map_err(
|source| CheckpointError::LiveState {
source: Box::new(source),
},
)?);
Ok(owner)
}
#[test]
fn concrete_genesis_owner_restores_through_the_checkpoint_envelope()
-> Result<(), CheckpointError> {
let owner = genesis_owner()?;
let positions = owner.checkpoint_positions(CheckpointSchema::V4Suffix, 1, 0)?;
let root = CheckpointHistoryRoot::empty(7).encode_canonical();
let config = crate::server::participant::test_participant_config();
let context = ConversationAuthorityRestoreContext::new(positions, &root, &config)?;
let checkpoint = RestoreCheckpoint::<_, ROOT_ENCODED_BYTES>::new(positions, root, owner)?;
let bytes = checkpoint.encode_canonical()?;
let restored =
RestoreCheckpoint::<ConversationAuthority, ROOT_ENCODED_BYTES>::decode_canonical(
&bytes, 7, &context,
)?;
assert_eq!(restored.encode_canonical()?, bytes);
assert_eq!(restored.positions(), positions);
assert_eq!(
restored
.live()
.outbox
.as_ref()
.map(ConversationOutbox::checkpoint_boundaries),
Some((0, 0))
);
Ok(())
}
#[test]
fn owner_association_is_checked_before_any_body_decode() -> Result<(), CheckpointError> {
let owner = genesis_owner()?;
let mut positions = owner.checkpoint_positions(CheckpointSchema::V4Suffix, 1, 0)?;
let root = CheckpointHistoryRoot::empty(7).encode_canonical();
let config = crate::server::participant::test_participant_config();
let context = ConversationAuthorityRestoreContext::new(positions, &root, &config)?;
positions.next_order = 1;
assert!(matches!(
ConversationAuthority::decode_live(&[], &positions, &root, &context),
Err(CheckpointError::Association {
field: "owner context positions"
})
));
Ok(())
}
#[test]
fn unencoded_authority_is_refused_and_never_replaced_with_empty_state()
-> Result<(), CheckpointError> {
let mut owner = genesis_owner()?;
owner.tokens.insert([1; 16], 0);
owner.next_participant = 1;
assert!(matches!(
owner.encode_live(),
Err(CheckpointError::Position {
field: "retained audit and history-root authority"
})
));
assert_eq!(owner.tokens.get(&[1; 16]), Some(&0));
Ok(())
}
#[test]
fn saved_audit_and_root_restore_an_advanced_log_without_a_genesis_only_limit()
-> Result<(), CheckpointError> {
let config = crate::server::participant::test_participant_config();
let mut owner = genesis_owner()?;
owner.next_log_sequence = 2;
let root = CheckpointHistoryRoot::empty(7);
let audit = ComposedTerminalValidation::for_identity_slots(config.identity_slots).map_err(
|source| CheckpointError::LiveState {
source: Box::new(source),
},
)?;
owner.retain_checkpoint_metadata(root, &audit)?;
let positions = owner.checkpoint_positions(CheckpointSchema::V4Suffix, 2, 0)?;
let root_bytes = root.encode_canonical();
let context = ConversationAuthorityRestoreContext::new(positions, &root_bytes, &config)?;
let bytes = owner.encode_live()?;
let restored =
ConversationAuthority::decode_live(&bytes, &positions, &root_bytes, &context)?;
assert_eq!(restored.next_log_sequence, 2);
assert_eq!(restored.checkpoint_history_root, Some(root));
assert_eq!(
restored.checkpoint_audit_source,
Some(audit.encode_checkpoint()?)
);
assert_eq!(restored.encode_live()?, bytes);
Ok(())
}
#[test]
fn saved_audit_is_validated_even_without_a_held_frontier() -> Result<(), CheckpointError> {
let config = crate::server::participant::test_participant_config();
let mut owner = genesis_owner()?;
owner.next_log_sequence = 2;
let root = CheckpointHistoryRoot::empty(7);
let audit = ComposedTerminalValidation::for_identity_slots(config.identity_slots).map_err(
|source| CheckpointError::LiveState {
source: Box::new(source),
},
)?;
owner.retain_checkpoint_metadata(root, &audit)?;
let positions = owner.checkpoint_positions(CheckpointSchema::V4Suffix, 2, 0)?;
let root_bytes = root.encode_canonical();
let context = ConversationAuthorityRestoreContext::new(positions, &root_bytes, &config)?;
let bytes = owner.encode_live()?;
let mut body = CheckpointOwnerBody::decode_canonical(&bytes)?;
body.terminal_validation = &[1, 0, 0, 0];
let corrupt = body.encode_canonical()?;
assert!(
ConversationAuthority::decode_live(&corrupt, &positions, &root_bytes, &context)
.is_err()
);
body.terminal_validation = &[];
let omitted = body.encode_canonical()?;
assert!(matches!(
ConversationAuthority::decode_live(&omitted, &positions, &root_bytes, &context),
Err(CheckpointError::Position {
field: "retained audit and history-root authority"
})
));
Ok(())
}
#[test]
fn genesis_with_retained_canonical_audit_preserves_all_four_count_bytes()
-> Result<(), CheckpointError> {
let config = crate::server::participant::test_participant_config();
let mut owner = genesis_owner()?;
let root = CheckpointHistoryRoot::empty(7);
let audit = ComposedTerminalValidation::for_identity_slots(config.identity_slots).map_err(
|source| CheckpointError::LiveState {
source: Box::new(source),
},
)?;
owner.retain_checkpoint_metadata(root, &audit)?;
let positions = owner.checkpoint_positions(CheckpointSchema::V4Suffix, 1, 0)?;
let root_bytes = root.encode_canonical();
let context = ConversationAuthorityRestoreContext::new(positions, &root_bytes, &config)?;
let bytes = owner.encode_live()?;
let body = CheckpointOwnerBody::decode_canonical(&bytes)?;
assert_eq!(body.terminal_validation, &[0; 4]);
let restored =
ConversationAuthority::decode_live(&bytes, &positions, &root_bytes, &context)?;
assert_eq!(restored.checkpoint_history_root, Some(root));
assert_eq!(
restored.checkpoint_audit_source.as_deref(),
Some([0; 4].as_slice())
);
assert_eq!(restored.encode_live()?, bytes);
Ok(())
}
}