liminal_protocol/lifecycle/

checkpoint_codec.rs

//! Canonical bytes for the lifecycle capsule's described sections.
//!
//! Big-endian throughout, exact lengths, every count a `u32`, every `Option`
//! one `0`/`1` byte, every enum a one-byte variant tag counted from `1`, and
//! each generic digest its raw fixed-width bytes. There is no serde: each type
//! has one hand-written writer and one reader, and a reader refuses every byte
//! its writer would not have produced. A reader never preallocates from a
//! count it has not yet proven, so a fabricated count costs at most the bytes
//! actually supplied.

mod frontier_accounting;
mod observer_sources;
mod owner_tokens;
pub use frontier_accounting::FrontierAccountingDecodeError;
pub use observer_sources::ObserverProjectionCheckpointKind;
pub(in crate::lifecycle) use observer_sources::{
    ObserverProjectionSource, decode_observer_source, encode_observer_source,
};
pub use owner_tokens::OwnerCheckpointRestoreError;
pub(in crate::lifecycle) use owner_tokens::{
    decode_ordinary_fate, encode_fate_token, encode_ordinary_fate, restore_fate_token,
};

use alloc::vec::Vec;

use crate::algebra::WideResourceVector;
use crate::outcome::CandidatePhase;
use crate::wire::{
    AttachSecret, BindingEpoch, CloseCause, ConnectionIncarnation, DetachAttemptToken, Generation,
    LeaveAttemptToken, ParticipantId,
};

use super::checkpoint::CapsuleDecodeError;
use super::storage::{
    BindingFateTerminalRestore, BindingStateRestore, ClosureStateRestore,
    CommittedBindingTerminalRestore, DebtCompletionRestore, DetachCellRestore,
    DetachedCredentialRecoveryRestore, DetachedCursorReleaseProvenanceRestore,
    DetachedMarkerReleaseProvenanceRestore, DetachedMarkerReleaseRestore,
    FencedAttachCommitRestore, LeaveCommittedRestore, LiveIdentityRestore,
    MarkerCursorProgressRestore, MarkerDeliveryRestore, OrdinaryBindingAuthorityRestore,
    OrdinaryBindingFateRestore, ParticipantConversationDescription, ParticipantConversationRestore,
    ParticipantLifecycleRestore, PendingFinalizationRestore, PendingRecoveredCursorReleaseRestore,
    RecoveredBindingFateRestore, RecoveredStorageCompletionRestore, RetiredIdentityRestore,
    StoredEdgeRestore,
};
use super::{
    ActiveBinding, AdmissionOrder, AttachedLifecycleRecord, AttachedRecordPosition, BindingOrigin,
    BindingTerminalOwner, ClaimFrontiersRestore, EnrollmentFingerprint, ExitProductRangeRestore,
    FrontierBinding, FrontierParticipant, HistoricalCausalFactRestore,
    HistoricalMarkerDeliveryFactRestore, ImmutableOrderCandidateMajorRestore,
    ImmutableSequenceCandidate, LeaveFingerprint, MarkerCandidateAuthority, MarkerProvenance,
    MarkerSequenceOwner, MovableOrderClaim, MovableSequenceClaim, OrderClaimFrontierRestore,
    OrderClaims, OrderDirectOwner, OrderHigh, OrderLedger, RecoveryOrderActiveBindingRestore,
    RecoveryOrderBlockRestore, RecoverySequenceBlockRestore, RecoverySequenceReserve,
    RecoverySequenceTerminalRestore, ReplacementTerminalProductRangeRestore, RetainedCausalRecord,
    RetainedCausalRecordKind, SequenceClaimFrontierRestore, SequenceClaims, SequenceDirectOwner,
    SequenceLedger, SequenceProductClass, SequenceProductRangesRestore,
    TerminalProductRangeRestore, TerminalProductSource,
};

/// Every live and retired participant, ascending by participant id.
pub(super) const PARTICIPANTS_SECTION_TAG: u16 = 2;
/// Both claim frontiers, both aggregate ledgers, and the issuance latch.
pub(super) const FRONTIERS_SECTION_TAG: u16 = 3;
/// The current closure state with its edge's predecessor provenance.
pub(super) const CLOSURE_SECTION_TAG: u16 = 4;
/// A sealed conversation's final sequence-ledger high watermark. It follows
/// the participants section in place of the frontiers and closure sections:
/// the seal retired the frontier owner, and what the observer sources still
/// need of it is this one number beside the retired generations.
pub(super) const SEALED_SECTION_TAG: u16 = 5;

/// The width of every digest a server's checkpoint stores.
pub const CANONICAL_DIGEST_LEN: usize = 32;

/// A digest a checkpoint stores as its canonical `N` bytes.
///
/// The raw bytes, or a [`CredentialDigest`](crate::wire::CredentialDigest)
/// that wraps 32 of them (PROTO-007 R1). Either encodes to the same bytes, so
/// the capsule format is unchanged.
pub trait CanonicalDigest<const N: usize>: Copy {
    /// The canonical bytes.
    fn to_canonical(self) -> [u8; N];
    /// The digest the canonical bytes name.
    fn from_canonical(bytes: [u8; N]) -> Self;
}

impl<const N: usize> CanonicalDigest<N> for [u8; N] {
    fn to_canonical(self) -> [u8; N] {
        self
    }
    fn from_canonical(bytes: [u8; N]) -> Self {
        bytes
    }
}

impl CanonicalDigest<CANONICAL_DIGEST_LEN> for crate::wire::CredentialDigest {
    fn to_canonical(self) -> [u8; CANONICAL_DIGEST_LEN] {
        self.into_bytes()
    }
    fn from_canonical(bytes: [u8; CANONICAL_DIGEST_LEN]) -> Self {
        Self::new(bytes)
    }
}

/// One conversation description, with every digest its canonical 32 bytes.
pub(super) type DigestDescription<D> = ParticipantConversationDescription<D, D, D, D>;

type DigestParticipant<D> = ParticipantLifecycleRestore<D, D, D, D>;

/// Encodes the participants, frontiers and closure sections, each with its
/// tag and exact length, in that fixed order.
///
/// Returns `None` when a count or a section length does not fit its `u32`.
pub(super) fn encode_description<const N: usize, D: CanonicalDigest<N>>(
    description: &DigestDescription<D>,
) -> Option<Vec<u8>> {
    let restore = &description.restore;
    let mut participants = Writer::default();
    write_list(
        &mut participants,
        &restore.participants,
        write_participant::<N, D>,
    );
    let mut frontiers = Writer::default();
    write_frontiers(&mut frontiers, &restore.frontiers);
    write_sequence_ledger(&mut frontiers, restore.sequence_ledger);
    write_order_ledger(&mut frontiers, restore.order_ledger);
    write_option(
        &mut frontiers,
        description.fenced_marker_issued.as_ref(),
        |w, &(participant_id, marker_delivery_seq, binding_epoch)| {
            w.put_u64(participant_id);
            w.put_u64(marker_delivery_seq);
            write_epoch(w, binding_epoch);
        },
    );
    let mut closure = Writer::default();
    write_closure(&mut closure, &restore.closure);
    let mut sections = Writer::default();
    sections.section(PARTICIPANTS_SECTION_TAG, participants);
    sections.section(FRONTIERS_SECTION_TAG, frontiers);
    sections.section(CLOSURE_SECTION_TAG, closure);
    sections.finish()
}

/// Encodes a sealed conversation's participants and sealed sections, each
/// with its tag and exact length, in that fixed order. Every participant is
/// retired: a seal leaves no live slot.
///
/// Returns `None` when a count or a section length does not fit its `u32`.
pub(super) fn encode_sealed_description<const N: usize, D: CanonicalDigest<N>>(
    participants: &[DigestParticipant<D>],
    ledger_high_watermark: u64,
) -> Option<Vec<u8>> {
    let mut retired = Writer::default();
    write_list(&mut retired, participants, write_participant::<N, D>);
    let mut sealed = Writer::default();
    sealed.put_u64(ledger_high_watermark);
    let mut sections = Writer::default();
    sections.section(PARTICIPANTS_SECTION_TAG, retired);
    sections.section(SEALED_SECTION_TAG, sealed);
    sections.finish()
}

/// Decodes the two section bodies [`encode_sealed_description`] wrote.
///
/// # Errors
///
/// Returns the errors [`decode_description`] returns for the participants
/// section, and [`CapsuleDecodeError::NonCanonicalSection`] for a live
/// participant in a sealed conversation.
pub(super) fn decode_sealed_description<const N: usize, D: CanonicalDigest<N>>(
    participants: &[u8],
    sealed: &[u8],
) -> Result<(Vec<DigestParticipant<D>>, u64), CapsuleDecodeError> {
    let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
    let participants = read_participants::<N, D>(&mut reader)?;
    reader.finish()?;
    if participants
        .iter()
        .any(|participant| matches!(participant, ParticipantLifecycleRestore::Live { .. }))
    {
        return Err(reader.invalid());
    }
    let mut reader = Reader::new(sealed, SEALED_SECTION_TAG);
    let ledger_high_watermark = reader.take_u64()?;
    reader.finish()?;
    Ok((participants, ledger_high_watermark))
}

/// Decodes the three section bodies [`encode_description`] wrote.
///
/// # Errors
///
/// Returns [`CapsuleDecodeError::NonCanonicalSectionLength`] for a body that
/// ends early or runs long, [`CapsuleDecodeError::NonCanonicalSection`] for a
/// byte no writer produces or a ledger that breaks its invariant, and
/// [`CapsuleDecodeError::ParticipantOutOfOrder`] for participants that are not
/// strictly ascending.
pub(super) fn decode_description<const N: usize, D: CanonicalDigest<N>>(
    participants: &[u8],
    frontiers: &[u8],
    closure: &[u8],
) -> Result<DigestDescription<D>, CapsuleDecodeError> {
    let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
    let participants = read_participants::<N, D>(&mut reader)?;
    reader.finish()?;

    let mut reader = Reader::new(frontiers, FRONTIERS_SECTION_TAG);
    let frontier_restore = read_frontiers(&mut reader)?;
    let sequence_ledger = read_sequence_ledger(&mut reader)?;
    let order_ledger = read_order_ledger(&mut reader)?;
    let fenced_marker_issued = read_option(&mut reader, |r| {
        Ok((r.take_u64()?, r.take_u64()?, read_epoch(r)?))
    })?;
    reader.finish()?;

    let mut reader = Reader::new(closure, CLOSURE_SECTION_TAG);
    let closure = read_closure(&mut reader)?;
    reader.finish()?;

    Ok(ParticipantConversationDescription {
        restore: ParticipantConversationRestore {
            participants,
            frontiers: frontier_restore,
            sequence_ledger,
            order_ledger,
            closure,
        },
        fenced_marker_issued,
    })
}

/// The permanent participant id one participant capsule belongs to.
pub(super) const fn participant_id_of<EF, V, LF, D>(
    participant: &ParticipantLifecycleRestore<EF, V, LF, D>,
) -> ParticipantId {
    match participant {
        ParticipantLifecycleRestore::Live { identity, .. } => identity.participant_id,
        ParticipantLifecycleRestore::Retired(retired) => retired.participant_id,
    }
}

// ---------------------------------------------------------------------------
// Writer and reader
// ---------------------------------------------------------------------------

/// Appends canonical bytes; a count or length that does not fit `u32` marks
/// the whole encoding as failed rather than writing a wrong value silently.
#[derive(Debug, Default)]
struct Writer {
    bytes: Vec<u8>,
    overflow: bool,
}

impl Writer {
    fn put_u8(&mut self, value: u8) {
        self.bytes.push(value);
    }

    fn put_u16(&mut self, value: u16) {
        self.bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn put_u32(&mut self, value: u32) {
        self.bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn put_u64(&mut self, value: u64) {
        self.bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn put_u128(&mut self, value: u128) {
        self.bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn put_bytes(&mut self, value: &[u8]) {
        self.bytes.extend_from_slice(value);
    }

    fn put_flag(&mut self, value: bool) {
        self.put_u8(u8::from(value));
    }

    fn put_count(&mut self, len: usize) {
        let count = u32::try_from(len);
        self.overflow |= count.is_err();
        self.put_u32(count.unwrap_or(u32::MAX));
    }

    /// Frames one finished section body behind its tag and exact length.
    fn section(&mut self, tag: u16, body: Self) {
        self.overflow |= body.overflow;
        self.put_u16(tag);
        self.put_count(body.bytes.len());
        self.bytes.extend(body.bytes);
    }

    fn finish(self) -> Option<Vec<u8>> {
        if self.overflow {
            None
        } else {
            Some(self.bytes)
        }
    }
}

/// Reads one section body; every refusal names that section.
#[derive(Debug)]
struct Reader<'a> {
    body: &'a [u8],
    at: usize,
    tag: u16,
}

impl<'a> Reader<'a> {
    const fn new(body: &'a [u8], tag: u16) -> Self {
        Self { body, at: 0, tag }
    }

    /// The body ended early or ran long.
    fn wrong_length(&self) -> CapsuleDecodeError {
        CapsuleDecodeError::NonCanonicalSectionLength {
            tag: self.tag,
            declared: u32::try_from(self.body.len()).unwrap_or(u32::MAX),
        }
    }

    /// A complete field holds a value no writer produces.
    const fn invalid(&self) -> CapsuleDecodeError {
        CapsuleDecodeError::NonCanonicalSection { tag: self.tag }
    }

    fn take<const W: usize>(&mut self) -> Result<[u8; W], CapsuleDecodeError> {
        let body = self.body;
        let end = self
            .at
            .checked_add(W)
            .ok_or(CapsuleDecodeError::LengthOverflow)?;
        let taken = body.get(self.at..end).ok_or_else(|| self.wrong_length())?;
        let array = <[u8; W]>::try_from(taken).map_err(|_| CapsuleDecodeError::LengthOverflow)?;
        self.at = end;
        Ok(array)
    }

    fn take_u8(&mut self) -> Result<u8, CapsuleDecodeError> {
        self.take::<1>().map(|[value]| value)
    }

    fn take_u32(&mut self) -> Result<u32, CapsuleDecodeError> {
        self.take().map(u32::from_be_bytes)
    }

    fn take_u64(&mut self) -> Result<u64, CapsuleDecodeError> {
        self.take().map(u64::from_be_bytes)
    }

    fn take_u128(&mut self) -> Result<u128, CapsuleDecodeError> {
        self.take().map(u128::from_be_bytes)
    }

    fn take_flag(&mut self) -> Result<bool, CapsuleDecodeError> {
        match self.take_u8()? {
            0 => Ok(false),
            1 => Ok(true),
            _ => Err(self.invalid()),
        }
    }

    fn take_count(&mut self) -> Result<usize, CapsuleDecodeError> {
        usize::try_from(self.take_u32()?).map_err(|_| CapsuleDecodeError::LengthOverflow)
    }

    fn finish(&self) -> Result<(), CapsuleDecodeError> {
        if self.at == self.body.len() {
            Ok(())
        } else {
            Err(self.wrong_length())
        }
    }
}

fn write_list<T>(w: &mut Writer, items: &[T], mut write: impl FnMut(&mut Writer, &T)) {
    w.put_count(items.len());
    for item in items {
        write(w, item);
    }
}

fn read_list<'a, T>(
    r: &mut Reader<'a>,
    mut read: impl FnMut(&mut Reader<'a>) -> Result<T, CapsuleDecodeError>,
) -> Result<Vec<T>, CapsuleDecodeError> {
    let count = r.take_count()?;
    let mut items = Vec::new();
    for _ in 0..count {
        items.push(read(r)?);
    }
    Ok(items)
}

fn write_option<T>(w: &mut Writer, item: Option<&T>, write: impl FnOnce(&mut Writer, &T)) {
    w.put_flag(item.is_some());
    if let Some(item) = item {
        write(w, item);
    }
}

fn read_option<'a, T>(
    r: &mut Reader<'a>,
    read: impl FnOnce(&mut Reader<'a>) -> Result<T, CapsuleDecodeError>,
) -> Result<Option<T>, CapsuleDecodeError> {
    if r.take_flag()? {
        read(r).map(Some)
    } else {
        Ok(None)
    }
}

// ---------------------------------------------------------------------------
// Shared value types
// ---------------------------------------------------------------------------

fn write_generation(w: &mut Writer, generation: Generation) {
    w.put_u64(generation.get());
}

fn read_generation(r: &mut Reader<'_>) -> Result<Generation, CapsuleDecodeError> {
    let value = r.take_u64()?;
    Generation::new(value).ok_or_else(|| r.invalid())
}

fn write_epoch(w: &mut Writer, epoch: BindingEpoch) {
    w.put_u64(epoch.connection_incarnation.server_incarnation);
    w.put_u64(epoch.connection_incarnation.connection_ordinal);
    write_generation(w, epoch.capability_generation);
}

fn read_epoch(r: &mut Reader<'_>) -> Result<BindingEpoch, CapsuleDecodeError> {
    let server_incarnation = r.take_u64()?;
    let connection_ordinal = r.take_u64()?;
    Ok(BindingEpoch::new(
        ConnectionIncarnation::new(server_incarnation, connection_ordinal),
        read_generation(r)?,
    ))
}

fn write_optional_epoch(w: &mut Writer, epoch: Option<BindingEpoch>) {
    write_option(w, epoch.as_ref(), |w, epoch| write_epoch(w, *epoch));
}

fn read_optional_epoch(r: &mut Reader<'_>) -> Result<Option<BindingEpoch>, CapsuleDecodeError> {
    read_option(r, read_epoch)
}

fn write_optional_u64(w: &mut Writer, value: Option<u64>) {
    write_option(w, value.as_ref(), |w, value| w.put_u64(*value));
}

fn read_optional_u64(r: &mut Reader<'_>) -> Result<Option<u64>, CapsuleDecodeError> {
    read_option(r, Reader::take_u64)
}

fn write_wide(w: &mut Writer, value: WideResourceVector) {
    w.put_u128(value.entries);
    w.put_u128(value.bytes);
}

fn read_wide(r: &mut Reader<'_>) -> Result<WideResourceVector, CapsuleDecodeError> {
    let entries = r.take_u128()?;
    Ok(WideResourceVector::new(entries, r.take_u128()?))
}

fn write_optional_wide(w: &mut Writer, value: Option<WideResourceVector>) {
    write_option(w, value.as_ref(), |w, value| write_wide(w, *value));
}

fn read_optional_wide(
    r: &mut Reader<'_>,
) -> Result<Option<WideResourceVector>, CapsuleDecodeError> {
    read_option(r, read_wide)
}

fn write_digest<const N: usize, D: CanonicalDigest<N>>(w: &mut Writer, digest: &D) {
    w.put_bytes(&digest.to_canonical());
}

fn read_digest<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<D, CapsuleDecodeError> {
    r.take::<N>().map(D::from_canonical)
}

fn write_active_binding(w: &mut Writer, binding: ActiveBinding) {
    w.put_u64(binding.participant_id);
    w.put_u64(binding.conversation_id);
    write_epoch(w, binding.binding_epoch);
}

fn read_active_binding(r: &mut Reader<'_>) -> Result<ActiveBinding, CapsuleDecodeError> {
    let participant_id = r.take_u64()?;
    let conversation_id = r.take_u64()?;
    Ok(ActiveBinding {
        participant_id,
        conversation_id,
        binding_epoch: read_epoch(r)?,
    })
}

fn write_close_cause(w: &mut Writer, cause: CloseCause) {
    match cause {
        CloseCause::CleanDeregister => w.put_u8(1),
        CloseCause::ConnectionLost => w.put_u8(2),
        CloseCause::ProcessKilled => w.put_u8(3),
        CloseCause::ProtocolError => w.put_u8(4),
        CloseCause::Superseded => w.put_u8(5),
        CloseCause::ServerShutdown => w.put_u8(6),
        CloseCause::UncleanServerRestart {
            prior_server_incarnation,
        } => {
            w.put_u8(7);
            w.put_u64(prior_server_incarnation);
        }
    }
}

fn read_close_cause(r: &mut Reader<'_>) -> Result<CloseCause, CapsuleDecodeError> {
    Ok(match r.take_u8()? {
        1 => CloseCause::CleanDeregister,
        2 => CloseCause::ConnectionLost,
        3 => CloseCause::ProcessKilled,
        4 => CloseCause::ProtocolError,
        5 => CloseCause::Superseded,
        6 => CloseCause::ServerShutdown,
        7 => CloseCause::UncleanServerRestart {
            prior_server_incarnation: r.take_u64()?,
        },
        _ => return Err(r.invalid()),
    })
}

fn write_admission_order(w: &mut Writer, order: AdmissionOrder) {
    w.put_u64(order.transaction_order());
    w.put_u8(match order.candidate_phase() {
        CandidatePhase::BindingTerminal => 1,
        CandidatePhase::MembershipExit => 2,
        CandidatePhase::AttachLifecycle => 3,
        CandidatePhase::OrdinaryRecord => 4,
        CandidatePhase::CompactionMarker => 5,
    });
    w.put_u64(order.participant_index());
}

fn read_admission_order(r: &mut Reader<'_>) -> Result<AdmissionOrder, CapsuleDecodeError> {
    let transaction_order = r.take_u64()?;
    let phase = match r.take_u8()? {
        1 => CandidatePhase::BindingTerminal,
        2 => CandidatePhase::MembershipExit,
        3 => CandidatePhase::AttachLifecycle,
        4 => CandidatePhase::OrdinaryRecord,
        5 => CandidatePhase::CompactionMarker,
        _ => return Err(r.invalid()),
    };
    Ok(AdmissionOrder::new(transaction_order, phase, r.take_u64()?))
}

fn write_frontier_binding(w: &mut Writer, binding: FrontierBinding) {
    match binding {
        FrontierBinding::Bound(epoch) => {
            w.put_u8(1);
            write_epoch(w, epoch);
        }
        FrontierBinding::Detached(epoch) => {
            w.put_u8(2);
            write_epoch(w, epoch);
        }
    }
}

fn read_frontier_binding(r: &mut Reader<'_>) -> Result<FrontierBinding, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => read_epoch(r).map(FrontierBinding::Bound),
        2 => read_epoch(r).map(FrontierBinding::Detached),
        _ => Err(r.invalid()),
    }
}

fn write_terminal_owner(w: &mut Writer, owner: BindingTerminalOwner) {
    w.put_u64(owner.participant_index);
    write_epoch(w, owner.binding_epoch);
}

fn read_terminal_owner(r: &mut Reader<'_>) -> Result<BindingTerminalOwner, CapsuleDecodeError> {
    let participant_index = r.take_u64()?;
    Ok(BindingTerminalOwner {
        participant_index,
        binding_epoch: read_epoch(r)?,
    })
}

// ---------------------------------------------------------------------------
// Participants section
// ---------------------------------------------------------------------------

fn read_participants<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<Vec<DigestParticipant<D>>, CapsuleDecodeError> {
    let participants = read_list(r, read_participant::<N, D>)?;
    for pair in participants.windows(2) {
        if let [earlier, later] = pair
            && participant_id_of(earlier) >= participant_id_of(later)
        {
            return Err(CapsuleDecodeError::ParticipantOutOfOrder {
                participant_id: participant_id_of(later),
            });
        }
    }
    Ok(participants)
}

fn write_participant<const N: usize, D: CanonicalDigest<N>>(
    w: &mut Writer,
    participant: &DigestParticipant<D>,
) {
    match participant {
        ParticipantLifecycleRestore::Live {
            identity,
            binding,
            binding_origin,
            detach_cell,
        } => {
            w.put_u8(1);
            write_live_identity(w, identity);
            write_binding_state(w, *binding);
            write_option(w, binding_origin.as_ref(), |w, origin| {
                write_binding_origin(w, *origin);
            });
            write_detach_cell(w, detach_cell);
        }
        ParticipantLifecycleRestore::Retired(retired) => {
            w.put_u8(2);
            write_retired(w, retired);
        }
    }
}

fn read_participant<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<DigestParticipant<D>, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => {
            let identity = read_live_identity::<N, D>(r)?;
            let binding = read_binding_state(r)?;
            let binding_origin = read_option(r, read_binding_origin)?;
            Ok(ParticipantLifecycleRestore::Live {
                identity,
                binding,
                binding_origin,
                detach_cell: read_detach_cell::<N, D>(r)?,
            })
        }
        2 => read_retired::<N, D>(r).map(ParticipantLifecycleRestore::Retired),
        _ => Err(r.invalid()),
    }
}

fn write_live_identity<const N: usize, D: CanonicalDigest<N>>(
    w: &mut Writer,
    identity: &LiveIdentityRestore<D>,
) {
    w.put_u64(identity.participant_id);
    w.put_u64(identity.conversation_id);
    write_generation(w, identity.generation);
    w.put_bytes(identity.attach_secret.as_bytes());
    w.put_u64(identity.cursor);
    write_digest(w, identity.enrollment_fingerprint.value());
    write_option(w, identity.latest_terminal.as_ref(), |w, terminal| {
        write_committed_terminal(w, *terminal);
    });
}

fn read_live_identity<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<LiveIdentityRestore<D>, CapsuleDecodeError> {
    let participant_id = r.take_u64()?;
    let conversation_id = r.take_u64()?;
    let generation = read_generation(r)?;
    let attach_secret = AttachSecret::new(r.take()?);
    let cursor = r.take_u64()?;
    let enrollment_fingerprint = EnrollmentFingerprint::new(read_digest::<N, D>(r)?);
    Ok(LiveIdentityRestore {
        participant_id,
        conversation_id,
        generation,
        attach_secret,
        cursor,
        enrollment_fingerprint,
        latest_terminal: read_option(r, read_committed_terminal)?,
    })
}

fn write_committed_terminal(w: &mut Writer, terminal: CommittedBindingTerminalRestore) {
    write_active_binding(w, terminal.binding);
    write_close_cause(w, terminal.cause);
    w.put_u64(terminal.transaction_order);
    w.put_u64(terminal.delivery_seq);
}

fn read_committed_terminal(
    r: &mut Reader<'_>,
) -> Result<CommittedBindingTerminalRestore, CapsuleDecodeError> {
    let binding = read_active_binding(r)?;
    let cause = read_close_cause(r)?;
    let transaction_order = r.take_u64()?;
    Ok(CommittedBindingTerminalRestore {
        binding,
        cause,
        transaction_order,
        delivery_seq: r.take_u64()?,
    })
}

fn write_pending_terminal(w: &mut Writer, pending: PendingFinalizationRestore) {
    write_active_binding(w, pending.binding);
    write_close_cause(w, pending.cause);
    w.put_u64(pending.transaction_order);
}

fn read_pending_terminal(
    r: &mut Reader<'_>,
) -> Result<PendingFinalizationRestore, CapsuleDecodeError> {
    let binding = read_active_binding(r)?;
    let cause = read_close_cause(r)?;
    Ok(PendingFinalizationRestore {
        binding,
        cause,
        transaction_order: r.take_u64()?,
    })
}

fn write_binding_state(w: &mut Writer, binding: BindingStateRestore) {
    match binding {
        BindingStateRestore::Detached => w.put_u8(1),
        BindingStateRestore::Bound(binding) => {
            w.put_u8(2);
            write_active_binding(w, binding);
        }
        BindingStateRestore::PendingFinalization(pending) => {
            w.put_u8(3);
            write_pending_terminal(w, pending);
        }
    }
}

fn read_binding_state(r: &mut Reader<'_>) -> Result<BindingStateRestore, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => Ok(BindingStateRestore::Detached),
        2 => read_active_binding(r).map(BindingStateRestore::Bound),
        3 => read_pending_terminal(r).map(BindingStateRestore::PendingFinalization),
        _ => Err(r.invalid()),
    }
}

/// An origin is its attached record's binding and position, then its kind.
/// The record's admission order is not written: it is always the
/// attach-lifecycle phase at the record's own participant, which the reader
/// rebuilds exactly as the producing commit did.
pub(super) fn encode_binding_origin_source(origin: BindingOrigin) -> Vec<u8> {
    let mut writer = Writer::default();
    write_binding_origin(&mut writer, origin);
    writer.bytes
}

pub(super) fn decode_binding_origin_source(
    bytes: &[u8],
) -> Result<BindingOrigin, CapsuleDecodeError> {
    let mut reader = Reader::new(bytes, PARTICIPANTS_SECTION_TAG);
    let origin = read_binding_origin(&mut reader)?;
    reader.finish()?;
    Ok(origin)
}

fn write_binding_origin(w: &mut Writer, origin: BindingOrigin) {
    let attached = origin.attached();
    w.put_u64(attached.participant_id());
    w.put_u64(attached.conversation_id());
    write_epoch(w, attached.binding_epoch());
    w.put_u64(attached.admission_order().transaction_order());
    w.put_u64(attached.delivery_seq());
    let recovered = origin.recovered_marker();
    w.put_u8(if recovered.is_none() { 1 } else { 2 });
    if let Some((marker_delivery_seq, prior_binding_epoch)) = recovered {
        w.put_u64(marker_delivery_seq);
        write_epoch(w, prior_binding_epoch);
    }
}

fn read_binding_origin(r: &mut Reader<'_>) -> Result<BindingOrigin, CapsuleDecodeError> {
    let binding = read_active_binding(r)?;
    let transaction_order = r.take_u64()?;
    let delivery_seq = r.take_u64()?;
    let attached = AttachedLifecycleRecord::from_binding(
        binding,
        AttachedRecordPosition::new(transaction_order, delivery_seq),
    );
    match r.take_u8()? {
        1 => Ok(BindingOrigin::unfenced(attached)),
        2 => {
            let marker_delivery_seq = r.take_u64()?;
            Ok(BindingOrigin::recovered(
                attached,
                marker_delivery_seq,
                read_epoch(r)?,
            ))
        }
        _ => Err(r.invalid()),
    }
}

fn write_detach_cell<const N: usize, D: CanonicalDigest<N>>(
    w: &mut Writer,
    cell: &DetachCellRestore<D>,
) {
    match cell {
        DetachCellRestore::Empty => w.put_u8(1),
        DetachCellRestore::Pending {
            token,
            participant_id,
            request_generation,
            request_verifier,
            committed_binding_epoch,
            admission_order,
            refused_epoch,
        } => {
            w.put_u8(2);
            write_detach_head(w, *token, *participant_id, *request_generation);
            write_digest(w, request_verifier);
            write_epoch(w, *committed_binding_epoch);
            write_admission_order(w, *admission_order);
            w.put_u64(*refused_epoch);
        }
        DetachCellRestore::Committed {
            token,
            participant_id,
            request_generation,
            request_verifier,
            committed_binding_epoch,
            detached_delivery_seq,
        } => {
            w.put_u8(3);
            write_detach_head(w, *token, *participant_id, *request_generation);
            write_digest(w, request_verifier);
            write_epoch(w, *committed_binding_epoch);
            w.put_u64(*detached_delivery_seq);
        }
        DetachCellRestore::Terminalized {
            token,
            participant_id,
            request_generation,
            request_verifier,
            committed_binding_epoch,
        } => {
            w.put_u8(4);
            write_detach_head(w, *token, *participant_id, *request_generation);
            write_digest(w, request_verifier);
            write_epoch(w, *committed_binding_epoch);
        }
    }
}

fn write_detach_head(
    w: &mut Writer,
    token: DetachAttemptToken,
    participant_id: ParticipantId,
    request_generation: Generation,
) {
    w.put_bytes(token.as_bytes());
    w.put_u64(participant_id);
    write_generation(w, request_generation);
}

fn read_detach_cell<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<DetachCellRestore<D>, CapsuleDecodeError> {
    let variant = r.take_u8()?;
    if variant == 1 {
        return Ok(DetachCellRestore::Empty);
    }
    if !(2..=4).contains(&variant) {
        return Err(r.invalid());
    }
    let token = DetachAttemptToken::new(r.take()?);
    let participant_id = r.take_u64()?;
    let request_generation = read_generation(r)?;
    let request_verifier = read_digest::<N, D>(r)?;
    let committed_binding_epoch = read_epoch(r)?;
    Ok(match variant {
        2 => {
            let admission_order = read_admission_order(r)?;
            DetachCellRestore::Pending {
                token,
                participant_id,
                request_generation,
                request_verifier,
                committed_binding_epoch,
                admission_order,
                refused_epoch: r.take_u64()?,
            }
        }
        3 => DetachCellRestore::Committed {
            token,
            participant_id,
            request_generation,
            request_verifier,
            committed_binding_epoch,
            detached_delivery_seq: r.take_u64()?,
        },
        _ => DetachCellRestore::Terminalized {
            token,
            participant_id,
            request_generation,
            request_verifier,
            committed_binding_epoch,
        },
    })
}

fn write_retired<const N: usize, D: CanonicalDigest<N>>(
    w: &mut Writer,
    retired: &RetiredIdentityRestore<D, D, D>,
) {
    w.put_u64(retired.participant_id);
    w.put_u64(retired.conversation_id);
    write_generation(w, retired.retired_generation);
    write_digest(w, retired.enrollment_fingerprint.value());
    w.put_bytes(retired.leave_attempt_token.as_bytes());
    write_digest(w, &retired.leave_request_verifier);
    write_digest(w, retired.leave_fingerprint.value());
    w.put_u64(retired.left_transaction_order);
    write_leave_committed(w, retired.committed_result);
}

impl<D> RetiredIdentityRestore<D, D, D> {
    /// Encodes one complete tombstone using the lifecycle's canonical fields.
    ///
    /// The enclosing history value supplies the format version and typed key.
    #[must_use]
    pub fn encode_canonical<const N: usize>(&self) -> Vec<u8>
    where
        D: CanonicalDigest<N>,
    {
        let mut writer = Writer::default();
        write_retired(&mut writer, self);
        writer.bytes
    }

    /// Decodes one exact tombstone belonging to the requested history key.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleDecodeError`] for an incomplete or noncanonical field,
    /// trailing bytes, a different conversation or participant, or a committed
    /// Leave result that does not belong to the tombstone.
    pub fn decode_canonical<const N: usize>(
        bytes: &[u8],
        conversation_id: u64,
        participant_id: ParticipantId,
    ) -> Result<Self, CapsuleDecodeError>
    where
        D: CanonicalDigest<N>,
    {
        let mut reader = Reader::new(bytes, PARTICIPANTS_SECTION_TAG);
        let retired = read_retired(&mut reader)?;
        reader.finish()?;
        if retired.conversation_id != conversation_id {
            return Err(CapsuleDecodeError::ForeignConversation {
                tag: PARTICIPANTS_SECTION_TAG,
            });
        }
        if retired.participant_id != participant_id {
            return Err(reader.invalid());
        }
        retired.restore_result().map_err(|_| reader.invalid())?;
        Ok(retired)
    }
}

/// A retired identity over canonical digests.
type RetiredRestore<D> = RetiredIdentityRestore<D, D, D>;

fn read_retired<const N: usize, D: CanonicalDigest<N>>(
    r: &mut Reader<'_>,
) -> Result<RetiredRestore<D>, CapsuleDecodeError> {
    let participant_id = r.take_u64()?;
    let conversation_id = r.take_u64()?;
    let retired_generation = read_generation(r)?;
    let enrollment_fingerprint = EnrollmentFingerprint::new(read_digest::<N, D>(r)?);
    let leave_attempt_token = LeaveAttemptToken::new(r.take()?);
    let leave_request_verifier = read_digest::<N, D>(r)?;
    let leave_fingerprint = LeaveFingerprint::new(read_digest::<N, D>(r)?);
    let left_transaction_order = r.take_u64()?;
    Ok(RetiredIdentityRestore {
        participant_id,
        conversation_id,
        retired_generation,
        enrollment_fingerprint,
        leave_attempt_token,
        leave_request_verifier,
        leave_fingerprint,
        left_transaction_order,
        committed_result: read_leave_committed(r)?,
    })
}

fn write_leave_committed(w: &mut Writer, result: LeaveCommittedRestore) {
    w.put_u64(result.conversation_id);
    w.put_bytes(result.leave_attempt_token.as_bytes());
    w.put_u64(result.participant_id);
    write_generation(w, result.retired_generation);
    write_optional_epoch(w, result.ended_binding_epoch);
    write_optional_u64(w, result.prior_terminal_delivery_seq);
    w.put_u64(result.left_delivery_seq);
}

fn read_leave_committed(r: &mut Reader<'_>) -> Result<LeaveCommittedRestore, CapsuleDecodeError> {
    let conversation_id = r.take_u64()?;
    let leave_attempt_token = LeaveAttemptToken::new(r.take()?);
    let participant_id = r.take_u64()?;
    let retired_generation = read_generation(r)?;
    let ended_binding_epoch = read_optional_epoch(r)?;
    let prior_terminal_delivery_seq = read_optional_u64(r)?;
    Ok(LeaveCommittedRestore {
        conversation_id,
        leave_attempt_token,
        participant_id,
        retired_generation,
        ended_binding_epoch,
        prior_terminal_delivery_seq,
        left_delivery_seq: r.take_u64()?,
    })
}

// ---------------------------------------------------------------------------
// Frontiers section
// ---------------------------------------------------------------------------

fn write_frontiers(w: &mut Writer, restore: &ClaimFrontiersRestore) {
    w.put_u64(restore.conversation_id);
    write_list(w, &restore.active_identities, |w, participant| {
        w.put_u64(participant.participant_index());
        w.put_u64(participant.cursor());
        write_frontier_binding(w, participant.binding());
    });
    w.put_u64(restore.identity_slot_limit);
    w.put_u128(restore.retained_floor);
    w.put_u64(restore.retained_record_limit);
    write_list(w, &restore.retained_records, |w, record| {
        write_retained_record(w, *record);
    });
    write_list(w, &restore.active_marker_anchors, |w, anchor| {
        w.put_u64(*anchor);
    });
    write_list(w, &restore.historical_marker_deliveries, |w, fact| {
        w.put_u64(fact.conversation_id);
        w.put_u64(fact.participant_index);
        w.put_u64(fact.marker_delivery_seq);
        write_epoch(w, fact.delivered_binding_epoch);
    });
    write_list(w, &restore.historical_causal_facts, |w, fact| {
        write_causal_fact(w, *fact);
    });
    write_sequence_frontier(w, &restore.sequence);
    write_order_frontier(w, &restore.order);
    write_optional_u64(w, restore.recovery_marker_delivery_seq);
}

fn read_frontiers(r: &mut Reader<'_>) -> Result<ClaimFrontiersRestore, CapsuleDecodeError> {
    let conversation_id = r.take_u64()?;
    let active_identities = read_list(r, |r| {
        let participant_index = r.take_u64()?;
        let cursor = r.take_u64()?;
        Ok(FrontierParticipant::new(
            participant_index,
            cursor,
            read_frontier_binding(r)?,
        ))
    })?;
    let identity_slot_limit = r.take_u64()?;
    let retained_floor = r.take_u128()?;
    let retained_record_limit = r.take_u64()?;
    let retained_records = read_list(r, read_retained_record)?;
    let active_marker_anchors = read_list(r, Reader::take_u64)?;
    let historical_marker_deliveries = read_list(r, |r| {
        let conversation_id = r.take_u64()?;
        let participant_index = r.take_u64()?;
        let marker_delivery_seq = r.take_u64()?;
        Ok(HistoricalMarkerDeliveryFactRestore {
            conversation_id,
            participant_index,
            marker_delivery_seq,
            delivered_binding_epoch: read_epoch(r)?,
        })
    })?;
    let historical_causal_facts = read_list(r, read_causal_fact)?;
    let sequence = read_sequence_frontier(r)?;
    let order = read_order_frontier(r)?;
    Ok(ClaimFrontiersRestore {
        conversation_id,
        active_identities,
        identity_slot_limit,
        retained_floor,
        retained_record_limit,
        retained_records,
        active_marker_anchors,
        historical_marker_deliveries,
        historical_causal_facts,
        sequence,
        order,
        recovery_marker_delivery_seq: read_optional_u64(r)?,
    })
}

fn write_retained_record(w: &mut Writer, record: RetainedCausalRecord) {
    w.put_u64(record.delivery_seq);
    write_admission_order(w, record.admission_order);
    match record.kind {
        RetainedCausalRecordKind::BindingTerminal(owner) => {
            w.put_u8(1);
            write_terminal_owner(w, owner);
        }
        RetainedCausalRecordKind::MembershipExit { participant_index } => {
            w.put_u8(2);
            w.put_u64(participant_index);
        }
        RetainedCausalRecordKind::AttachLifecycle {
            participant_index,
            binding_epoch,
        } => {
            w.put_u8(3);
            w.put_u64(participant_index);
            write_epoch(w, binding_epoch);
        }
        RetainedCausalRecordKind::OrdinaryRecord { participant_index } => {
            w.put_u8(4);
            w.put_u64(participant_index);
        }
        RetainedCausalRecordKind::CompactionMarker {
            participant_index,
            provenance,
        } => {
            w.put_u8(5);
            w.put_u64(participant_index);
            write_marker_provenance(w, provenance);
        }
    }
}

fn read_retained_record(r: &mut Reader<'_>) -> Result<RetainedCausalRecord, CapsuleDecodeError> {
    let delivery_seq = r.take_u64()?;
    let admission_order = read_admission_order(r)?;
    let kind = match r.take_u8()? {
        1 => RetainedCausalRecordKind::BindingTerminal(read_terminal_owner(r)?),
        2 => RetainedCausalRecordKind::MembershipExit {
            participant_index: r.take_u64()?,
        },
        3 => {
            let participant_index = r.take_u64()?;
            RetainedCausalRecordKind::AttachLifecycle {
                participant_index,
                binding_epoch: read_epoch(r)?,
            }
        }
        4 => RetainedCausalRecordKind::OrdinaryRecord {
            participant_index: r.take_u64()?,
        },
        5 => {
            let participant_index = r.take_u64()?;
            RetainedCausalRecordKind::CompactionMarker {
                participant_index,
                provenance: read_marker_provenance(r)?,
            }
        }
        _ => return Err(r.invalid()),
    };
    Ok(RetainedCausalRecord {
        delivery_seq,
        admission_order,
        kind,
    })
}

fn write_marker_provenance(w: &mut Writer, provenance: MarkerProvenance) {
    match provenance {
        MarkerProvenance::NonProductM => w.put_u8(1),
        MarkerProvenance::TerminalProduct {
            terminal,
            affected_participant,
        } => {
            w.put_u8(2);
            match terminal {
                TerminalProductSource::Binding(owner) => {
                    w.put_u8(1);
                    write_terminal_owner(w, owner);
                }
                TerminalProductSource::RecoveryReplacement {
                    participant_index,
                    binding_epoch,
                } => {
                    w.put_u8(2);
                    w.put_u64(participant_index);
                    write_epoch(w, binding_epoch);
                }
            }
            w.put_u64(affected_participant);
        }
        MarkerProvenance::ExitProduct {
            exit_participant,
            remaining_participant,
        } => {
            w.put_u8(3);
            w.put_u64(exit_participant);
            w.put_u64(remaining_participant);
        }
    }
}

fn read_marker_provenance(r: &mut Reader<'_>) -> Result<MarkerProvenance, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => Ok(MarkerProvenance::NonProductM),
        2 => {
            let terminal = match r.take_u8()? {
                1 => TerminalProductSource::Binding(read_terminal_owner(r)?),
                2 => {
                    let participant_index = r.take_u64()?;
                    TerminalProductSource::recovery_replacement(participant_index, read_epoch(r)?)
                }
                _ => return Err(r.invalid()),
            };
            Ok(MarkerProvenance::terminal_product(terminal, r.take_u64()?))
        }
        3 => {
            let exit_participant = r.take_u64()?;
            Ok(MarkerProvenance::exit_product(
                exit_participant,
                r.take_u64()?,
            ))
        }
        _ => Err(r.invalid()),
    }
}

fn write_causal_fact(w: &mut Writer, fact: HistoricalCausalFactRestore) {
    match fact {
        HistoricalCausalFactRestore::BindingTerminal {
            conversation_id,
            participant_index,
            binding_epoch,
            admission_order,
        } => {
            w.put_u8(1);
            w.put_u64(conversation_id);
            w.put_u64(participant_index);
            write_epoch(w, binding_epoch);
            write_admission_order(w, admission_order);
        }
        HistoricalCausalFactRestore::MembershipExit {
            conversation_id,
            participant_index,
            admission_order,
        } => {
            w.put_u8(2);
            w.put_u64(conversation_id);
            w.put_u64(participant_index);
            write_admission_order(w, admission_order);
        }
    }
}

fn read_causal_fact(r: &mut Reader<'_>) -> Result<HistoricalCausalFactRestore, CapsuleDecodeError> {
    let variant = r.take_u8()?;
    if !(1..=2).contains(&variant) {
        return Err(r.invalid());
    }
    let conversation_id = r.take_u64()?;
    let participant_index = r.take_u64()?;
    if variant == 1 {
        let binding_epoch = read_epoch(r)?;
        Ok(HistoricalCausalFactRestore::BindingTerminal {
            conversation_id,
            participant_index,
            binding_epoch,
            admission_order: read_admission_order(r)?,
        })
    } else {
        Ok(HistoricalCausalFactRestore::MembershipExit {
            conversation_id,
            participant_index,
            admission_order: read_admission_order(r)?,
        })
    }
}

fn write_sequence_frontier(w: &mut Writer, sequence: &SequenceClaimFrontierRestore) {
    write_list(w, &sequence.movable_claims, |w, claim| {
        w.put_u64(claim.delivery_seq);
        write_sequence_owner(w, claim.owner);
    });
    write_list(w, &sequence.immutable_candidates, |w, candidate| {
        write_sequence_candidate(w, *candidate);
    });
    let products = &sequence.products;
    write_list(w, &products.live_times_terminal, |w, range| {
        w.put_u64(range.start);
        w.put_u64(range.length);
        write_terminal_owner(w, range.terminal);
    });
    write_option(
        w,
        products.live_times_replacement_terminal.as_ref(),
        |w, range| {
            w.put_u64(range.start);
            w.put_u64(range.length);
        },
    );
    write_list(w, &products.other_live_times_exit, |w, range| {
        w.put_u64(range.start);
        w.put_u64(range.length);
        w.put_u64(range.exit_participant);
    });
    write_option(w, sequence.recovery.as_ref(), |w, block| {
        write_option(w, block.terminal.as_ref(), |w, terminal| {
            w.put_u64(terminal.delivery_seq);
            write_terminal_owner(w, terminal.owner);
        });
        w.put_u64(block.recovery_attach_seq);
        w.put_u64(block.replacement_terminal_seq);
    });
}

fn read_sequence_frontier(
    r: &mut Reader<'_>,
) -> Result<SequenceClaimFrontierRestore, CapsuleDecodeError> {
    let movable_claims = read_list(r, |r| {
        let delivery_seq = r.take_u64()?;
        Ok(MovableSequenceClaim {
            delivery_seq,
            owner: read_sequence_owner(r)?,
        })
    })?;
    let immutable_candidates = read_list(r, read_sequence_candidate)?;
    let live_times_terminal = read_list(r, |r| {
        let start = r.take_u64()?;
        let length = r.take_u64()?;
        Ok(TerminalProductRangeRestore {
            start,
            length,
            terminal: read_terminal_owner(r)?,
        })
    })?;
    let live_times_replacement_terminal = read_option(r, |r| {
        let start = r.take_u64()?;
        Ok(ReplacementTerminalProductRangeRestore {
            start,
            length: r.take_u64()?,
        })
    })?;
    let other_live_times_exit = read_list(r, |r| {
        let start = r.take_u64()?;
        let length = r.take_u64()?;
        Ok(ExitProductRangeRestore {
            start,
            length,
            exit_participant: r.take_u64()?,
        })
    })?;
    let recovery = read_option(r, |r| {
        let terminal = read_option(r, |r| {
            let delivery_seq = r.take_u64()?;
            Ok(RecoverySequenceTerminalRestore {
                delivery_seq,
                owner: read_terminal_owner(r)?,
            })
        })?;
        let recovery_attach_seq = r.take_u64()?;
        Ok(RecoverySequenceBlockRestore {
            terminal,
            recovery_attach_seq,
            replacement_terminal_seq: r.take_u64()?,
        })
    })?;
    Ok(SequenceClaimFrontierRestore {
        movable_claims,
        immutable_candidates,
        products: SequenceProductRangesRestore {
            live_times_terminal,
            live_times_replacement_terminal,
            other_live_times_exit,
        },
        recovery,
    })
}

fn write_sequence_owner(w: &mut Writer, owner: SequenceDirectOwner) {
    match owner {
        SequenceDirectOwner::MembershipExit { participant_index } => {
            w.put_u8(1);
            w.put_u64(participant_index);
        }
        SequenceDirectOwner::BindingTerminal(owner) => {
            w.put_u8(2);
            write_terminal_owner(w, owner);
        }
    }
}

fn read_sequence_owner(r: &mut Reader<'_>) -> Result<SequenceDirectOwner, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => Ok(SequenceDirectOwner::MembershipExit {
            participant_index: r.take_u64()?,
        }),
        2 => read_terminal_owner(r).map(SequenceDirectOwner::BindingTerminal),
        _ => Err(r.invalid()),
    }
}

fn write_sequence_candidate(w: &mut Writer, candidate: ImmutableSequenceCandidate) {
    match candidate {
        ImmutableSequenceCandidate::BindingTerminal {
            delivery_seq,
            admission_order,
            owner,
        } => {
            w.put_u8(1);
            w.put_u64(delivery_seq);
            write_admission_order(w, admission_order);
            write_terminal_owner(w, owner);
        }
        ImmutableSequenceCandidate::Marker(marker) => {
            w.put_u8(2);
            w.put_u64(marker.delivery_seq);
            write_admission_order(w, marker.admission_order);
            write_frontier_binding(w, marker.target_binding);
            write_marker_provenance(w, marker.provenance);
            w.put_u64(marker.abandoned_after);
            w.put_u64(marker.abandoned_through);
            w.put_u64(marker.physical_floor_at_decision);
            match marker.current_owner {
                MarkerSequenceOwner::Marker => w.put_u8(1),
                MarkerSequenceOwner::ConditionalProduct(class) => {
                    w.put_u8(2);
                    w.put_u8(match class {
                        SequenceProductClass::LiveTimesTerminal => 1,
                        SequenceProductClass::LiveTimesReplacementTerminal => 2,
                        SequenceProductClass::OtherLiveTimesExit => 3,
                    });
                }
            }
        }
    }
}

fn read_sequence_candidate(
    r: &mut Reader<'_>,
) -> Result<ImmutableSequenceCandidate, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => {
            let delivery_seq = r.take_u64()?;
            let admission_order = read_admission_order(r)?;
            Ok(ImmutableSequenceCandidate::BindingTerminal {
                delivery_seq,
                admission_order,
                owner: read_terminal_owner(r)?,
            })
        }
        2 => read_marker_candidate(r).map(ImmutableSequenceCandidate::Marker),
        _ => Err(r.invalid()),
    }
}

fn read_marker_candidate(
    r: &mut Reader<'_>,
) -> Result<MarkerCandidateAuthority, CapsuleDecodeError> {
    let delivery_seq = r.take_u64()?;
    let admission_order = read_admission_order(r)?;
    let target_binding = read_frontier_binding(r)?;
    let provenance = read_marker_provenance(r)?;
    let abandoned_after = r.take_u64()?;
    let abandoned_through = r.take_u64()?;
    let physical_floor_at_decision = r.take_u64()?;
    let current_owner = match r.take_u8()? {
        1 => MarkerSequenceOwner::Marker,
        2 => MarkerSequenceOwner::ConditionalProduct(match r.take_u8()? {
            1 => SequenceProductClass::LiveTimesTerminal,
            2 => SequenceProductClass::LiveTimesReplacementTerminal,
            3 => SequenceProductClass::OtherLiveTimesExit,
            _ => return Err(r.invalid()),
        }),
        _ => return Err(r.invalid()),
    };
    Ok(MarkerCandidateAuthority {
        delivery_seq,
        admission_order,
        target_binding,
        provenance,
        abandoned_after,
        abandoned_through,
        physical_floor_at_decision,
        current_owner,
    })
}

fn write_order_frontier(w: &mut Writer, order: &OrderClaimFrontierRestore) {
    write_list(w, &order.movable_claims, |w, claim| {
        w.put_u64(claim.transaction_order);
        match claim.owner {
            OrderDirectOwner::ActiveBindingTerminal(owner) => {
                w.put_u8(1);
                write_terminal_owner(w, owner);
            }
            OrderDirectOwner::MembershipExit { participant_index } => {
                w.put_u8(2);
                w.put_u64(participant_index);
            }
        }
    });
    write_list(w, &order.immutable_candidates, |w, major| {
        w.put_u64(major.transaction_order);
        write_list(w, &major.candidate_keys, |w, key| {
            write_admission_order(w, *key);
        });
    });
    write_option(w, order.recovery.as_ref(), |w, block| {
        write_option(w, block.active_binding.as_ref(), |w, active| {
            w.put_u64(active.transaction_order);
            write_terminal_owner(w, active.owner);
        });
        w.put_u64(block.recovery_operation_order);
        w.put_u64(block.replacement_terminal_order);
    });
}

fn read_order_frontier(
    r: &mut Reader<'_>,
) -> Result<OrderClaimFrontierRestore, CapsuleDecodeError> {
    let movable_claims = read_list(r, |r| {
        let transaction_order = r.take_u64()?;
        let owner = match r.take_u8()? {
            1 => OrderDirectOwner::ActiveBindingTerminal(read_terminal_owner(r)?),
            2 => OrderDirectOwner::MembershipExit {
                participant_index: r.take_u64()?,
            },
            _ => return Err(r.invalid()),
        };
        Ok(MovableOrderClaim {
            transaction_order,
            owner,
        })
    })?;
    let immutable_candidates = read_list(r, |r| {
        let transaction_order = r.take_u64()?;
        Ok(ImmutableOrderCandidateMajorRestore {
            transaction_order,
            candidate_keys: read_list(r, read_admission_order)?,
        })
    })?;
    let recovery = read_option(r, |r| {
        let active_binding = read_option(r, |r| {
            let transaction_order = r.take_u64()?;
            Ok(RecoveryOrderActiveBindingRestore {
                transaction_order,
                owner: read_terminal_owner(r)?,
            })
        })?;
        let recovery_operation_order = r.take_u64()?;
        Ok(RecoveryOrderBlockRestore {
            active_binding,
            recovery_operation_order,
            replacement_terminal_order: r.take_u64()?,
        })
    })?;
    Ok(OrderClaimFrontierRestore {
        movable_claims,
        immutable_candidates,
        recovery,
    })
}

fn write_sequence_ledger(w: &mut Writer, ledger: SequenceLedger) {
    w.put_u64(ledger.high_watermark());
    let claims = ledger.claims();
    w.put_u64(claims.live_members());
    w.put_u64(claims.binding_terminals());
    w.put_u64(claims.markers());
    w.put_u8(match claims.recovery() {
        RecoverySequenceReserve::None => 1,
        RecoverySequenceReserve::DetachedCredentialRecovery => 2,
    });
}

/// The ledger's required reserve is derived, never stored, and a ledger whose
/// claims exceed what remains is refused exactly as `try_new` refuses it.
fn read_sequence_ledger(r: &mut Reader<'_>) -> Result<SequenceLedger, CapsuleDecodeError> {
    let high_watermark = r.take_u64()?;
    let live_members = r.take_u64()?;
    let binding_terminals = r.take_u64()?;
    let markers = r.take_u64()?;
    let recovery = match r.take_u8()? {
        1 => RecoverySequenceReserve::None,
        2 => RecoverySequenceReserve::DetachedCredentialRecovery,
        _ => return Err(r.invalid()),
    };
    SequenceLedger::try_new(
        high_watermark,
        SequenceClaims::new(live_members, binding_terminals, markers, recovery),
    )
    .map_err(|_| r.invalid())
}

fn write_order_ledger(w: &mut Writer, ledger: OrderLedger) {
    match ledger.high() {
        OrderHigh::Empty => w.put_u8(1),
        OrderHigh::Allocated(order) => {
            w.put_u8(2);
            w.put_u64(order);
        }
    }
    let claims = ledger.claims();
    w.put_u64(claims.active_binding_terminals());
    w.put_u64(claims.membership_exits());
    w.put_flag(claims.recovery_operation());
    w.put_flag(claims.recovery_replacement_terminal());
}

fn read_order_ledger(r: &mut Reader<'_>) -> Result<OrderLedger, CapsuleDecodeError> {
    let high = match r.take_u8()? {
        1 => OrderHigh::Empty,
        2 => OrderHigh::Allocated(r.take_u64()?),
        _ => return Err(r.invalid()),
    };
    let active_binding_terminals = r.take_u64()?;
    let membership_exits = r.take_u64()?;
    let recovery_operation = r.take_flag()?;
    let recovery_replacement_terminal = r.take_flag()?;
    let claims = OrderClaims::new(
        active_binding_terminals,
        membership_exits,
        recovery_operation,
        recovery_replacement_terminal,
    )
    .map_err(|_| r.invalid())?;
    OrderLedger::try_new(high, claims).map_err(|_| r.invalid())
}

// ---------------------------------------------------------------------------
// Closure section
// ---------------------------------------------------------------------------

fn write_closure(w: &mut Writer, closure: &ClosureStateRestore) {
    match closure {
        ClosureStateRestore::Clear => w.put_u8(1),
        ClosureStateRestore::Owed { debt, edge } => {
            w.put_u8(2);
            write_wide(w, *debt);
            write_edge(w, edge);
        }
    }
}

fn read_closure(r: &mut Reader<'_>) -> Result<ClosureStateRestore, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => Ok(ClosureStateRestore::Clear),
        2 => {
            let debt = read_wide(r)?;
            Ok(ClosureStateRestore::Owed {
                debt,
                edge: read_edge(r)?,
            })
        }
        _ => Err(r.invalid()),
    }
}

fn write_edge(w: &mut Writer, edge: &StoredEdgeRestore) {
    match edge {
        StoredEdgeRestore::ObserverProjection { through_seq } => {
            w.put_u8(1);
            w.put_u64(*through_seq);
        }
        StoredEdgeRestore::PhysicalCompaction {
            from_floor,
            through_seq,
        } => {
            w.put_u8(2);
            w.put_u64(*from_floor);
            w.put_u64(*through_seq);
        }
        StoredEdgeRestore::MarkerDelivery(delivery) => {
            w.put_u8(3);
            write_marker_delivery(w, *delivery);
        }
        StoredEdgeRestore::ParticipantCursorProgressContinuous {
            participant_id,
            binding_epoch,
            through_seq,
            authority,
        } => {
            w.put_u8(4);
            w.put_u64(*participant_id);
            write_epoch(w, *binding_epoch);
            w.put_u64(*through_seq);
            write_ordinary_authority(w, *authority);
        }
        StoredEdgeRestore::ParticipantCursorProgressMarker(progress) => {
            w.put_u8(5);
            write_marker_progress(w, progress);
        }
        StoredEdgeRestore::DetachedCredentialRecovery(recovery) => {
            w.put_u8(6);
            write_credential_recovery(w, recovery);
        }
        StoredEdgeRestore::DetachedMarkerRelease(release) => {
            w.put_u8(7);
            write_marker_release(w, release);
        }
        StoredEdgeRestore::DetachedCursorRelease {
            participant_id,
            last_dead_binding_epoch,
            provenance,
        } => {
            w.put_u8(8);
            w.put_u64(*participant_id);
            write_epoch(w, *last_dead_binding_epoch);
            write_cursor_release_provenance(w, provenance);
        }
    }
}

fn read_edge(r: &mut Reader<'_>) -> Result<StoredEdgeRestore, CapsuleDecodeError> {
    Ok(match r.take_u8()? {
        1 => StoredEdgeRestore::ObserverProjection {
            through_seq: r.take_u64()?,
        },
        2 => {
            let from_floor = r.take_u64()?;
            StoredEdgeRestore::PhysicalCompaction {
                from_floor,
                through_seq: r.take_u64()?,
            }
        }
        3 => StoredEdgeRestore::MarkerDelivery(read_marker_delivery(r)?),
        4 => {
            let participant_id = r.take_u64()?;
            let binding_epoch = read_epoch(r)?;
            let through_seq = r.take_u64()?;
            StoredEdgeRestore::ParticipantCursorProgressContinuous {
                participant_id,
                binding_epoch,
                through_seq,
                authority: read_ordinary_authority(r)?,
            }
        }
        5 => StoredEdgeRestore::ParticipantCursorProgressMarker(read_marker_progress(r)?),
        6 => StoredEdgeRestore::DetachedCredentialRecovery(read_credential_recovery(r)?),
        7 => StoredEdgeRestore::DetachedMarkerRelease(read_marker_release(r)?),
        8 => {
            let participant_id = r.take_u64()?;
            let last_dead_binding_epoch = read_epoch(r)?;
            StoredEdgeRestore::DetachedCursorRelease {
                participant_id,
                last_dead_binding_epoch,
                provenance: read_cursor_release_provenance(r)?,
            }
        }
        _ => return Err(r.invalid()),
    })
}

fn write_marker_delivery(w: &mut Writer, delivery: MarkerDeliveryRestore) {
    w.put_u64(delivery.participant_id);
    write_epoch(w, delivery.binding_epoch);
    w.put_u64(delivery.marker_delivery_seq);
}

fn read_marker_delivery(r: &mut Reader<'_>) -> Result<MarkerDeliveryRestore, CapsuleDecodeError> {
    let participant_id = r.take_u64()?;
    let binding_epoch = read_epoch(r)?;
    Ok(MarkerDeliveryRestore {
        participant_id,
        binding_epoch,
        marker_delivery_seq: r.take_u64()?,
    })
}

fn write_ordinary_authority(w: &mut Writer, authority: OrdinaryBindingAuthorityRestore) {
    write_active_binding(w, authority.binding);
    w.put_u64(authority.through_seq);
}

fn read_ordinary_authority(
    r: &mut Reader<'_>,
) -> Result<OrdinaryBindingAuthorityRestore, CapsuleDecodeError> {
    let binding = read_active_binding(r)?;
    Ok(OrdinaryBindingAuthorityRestore {
        binding,
        through_seq: r.take_u64()?,
    })
}

fn write_marker_progress(w: &mut Writer, progress: &MarkerCursorProgressRestore) {
    w.put_u64(progress.conversation_id);
    w.put_u64(progress.participant_id);
    write_epoch(w, progress.binding_epoch);
    w.put_u64(progress.through_seq);
    w.put_u64(progress.marker_delivery_seq);
    write_marker_delivery(w, progress.delivery);
}

fn read_marker_progress(
    r: &mut Reader<'_>,
) -> Result<MarkerCursorProgressRestore, CapsuleDecodeError> {
    let conversation_id = r.take_u64()?;
    let participant_id = r.take_u64()?;
    let binding_epoch = read_epoch(r)?;
    let through_seq = r.take_u64()?;
    let marker_delivery_seq = r.take_u64()?;
    Ok(MarkerCursorProgressRestore {
        conversation_id,
        participant_id,
        binding_epoch,
        through_seq,
        marker_delivery_seq,
        delivery: read_marker_delivery(r)?,
    })
}

fn write_fate_terminal(w: &mut Writer, terminal: BindingFateTerminalRestore) {
    match terminal {
        BindingFateTerminalRestore::Committed(committed) => {
            w.put_u8(1);
            write_committed_terminal(w, committed);
        }
        BindingFateTerminalRestore::Pending(pending) => {
            w.put_u8(2);
            write_pending_terminal(w, pending);
        }
    }
}

fn read_fate_terminal(
    r: &mut Reader<'_>,
) -> Result<BindingFateTerminalRestore, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => read_committed_terminal(r).map(BindingFateTerminalRestore::Committed),
        2 => read_pending_terminal(r).map(BindingFateTerminalRestore::Pending),
        _ => Err(r.invalid()),
    }
}

fn write_credential_recovery(w: &mut Writer, recovery: &DetachedCredentialRecoveryRestore) {
    w.put_u64(recovery.participant_id);
    w.put_u64(recovery.marker_delivery_seq);
    write_epoch(w, recovery.prior_binding_epoch);
    w.put_u64(recovery.resulting_floor);
    write_fate_terminal(w, recovery.terminal);
    write_marker_progress(w, &recovery.progress);
}

fn read_credential_recovery(
    r: &mut Reader<'_>,
) -> Result<DetachedCredentialRecoveryRestore, CapsuleDecodeError> {
    let participant_id = r.take_u64()?;
    let marker_delivery_seq = r.take_u64()?;
    let prior_binding_epoch = read_epoch(r)?;
    let resulting_floor = r.take_u64()?;
    let terminal = read_fate_terminal(r)?;
    Ok(DetachedCredentialRecoveryRestore {
        participant_id,
        marker_delivery_seq,
        prior_binding_epoch,
        resulting_floor,
        terminal,
        progress: read_marker_progress(r)?,
    })
}

fn write_marker_release(w: &mut Writer, release: &DetachedMarkerReleaseRestore) {
    w.put_u64(release.conversation_id);
    w.put_u64(release.participant_id);
    w.put_u64(release.marker_delivery_seq);
    write_epoch(w, release.last_dead_binding_epoch);
    write_marker_delivery(w, release.delivery);
    match release.provenance {
        DetachedMarkerReleaseProvenanceRestore::Fate { terminal } => {
            w.put_u8(1);
            write_fate_terminal(w, terminal);
        }
        DetachedMarkerReleaseProvenanceRestore::Candidate => w.put_u8(2),
    }
}

fn read_marker_release(
    r: &mut Reader<'_>,
) -> Result<DetachedMarkerReleaseRestore, CapsuleDecodeError> {
    let conversation_id = r.take_u64()?;
    let participant_id = r.take_u64()?;
    let marker_delivery_seq = r.take_u64()?;
    let last_dead_binding_epoch = read_epoch(r)?;
    let delivery = read_marker_delivery(r)?;
    let provenance = match r.take_u8()? {
        1 => DetachedMarkerReleaseProvenanceRestore::Fate {
            terminal: read_fate_terminal(r)?,
        },
        2 => DetachedMarkerReleaseProvenanceRestore::Candidate,
        _ => return Err(r.invalid()),
    };
    Ok(DetachedMarkerReleaseRestore {
        conversation_id,
        participant_id,
        marker_delivery_seq,
        last_dead_binding_epoch,
        delivery,
        provenance,
    })
}

fn write_cursor_release_provenance(
    w: &mut Writer,
    provenance: &DetachedCursorReleaseProvenanceRestore,
) {
    match provenance {
        DetachedCursorReleaseProvenanceRestore::Ordinary(fate) => {
            w.put_u8(1);
            write_ordinary_authority(w, fate.authority);
            write_committed_terminal(w, fate.terminal);
            w.put_u64(fate.resulting_floor);
        }
        DetachedCursorReleaseProvenanceRestore::RecoveredDirect {
            fate,
            resulting_debt,
        } => {
            w.put_u8(2);
            write_recovered_fate(w, fate);
            write_wide(w, *resulting_debt);
        }
        DetachedCursorReleaseProvenanceRestore::RecoveredAfterStorage {
            pending,
            completion,
        } => {
            w.put_u8(3);
            write_recovered_fate(w, &pending.fate);
            write_wide(w, pending.resulting_debt);
            write_storage_completion(w, *completion);
        }
    }
}

fn read_cursor_release_provenance(
    r: &mut Reader<'_>,
) -> Result<DetachedCursorReleaseProvenanceRestore, CapsuleDecodeError> {
    Ok(match r.take_u8()? {
        1 => {
            let authority = read_ordinary_authority(r)?;
            let terminal = read_committed_terminal(r)?;
            DetachedCursorReleaseProvenanceRestore::Ordinary(OrdinaryBindingFateRestore {
                authority,
                terminal,
                resulting_floor: r.take_u64()?,
            })
        }
        2 => {
            let fate = read_recovered_fate(r)?;
            DetachedCursorReleaseProvenanceRestore::RecoveredDirect {
                fate,
                resulting_debt: read_wide(r)?,
            }
        }
        3 => {
            let fate = read_recovered_fate(r)?;
            let resulting_debt = read_wide(r)?;
            DetachedCursorReleaseProvenanceRestore::RecoveredAfterStorage {
                pending: PendingRecoveredCursorReleaseRestore {
                    fate,
                    resulting_debt,
                },
                completion: read_storage_completion(r)?,
            }
        }
        _ => return Err(r.invalid()),
    })
}

fn write_recovered_fate(w: &mut Writer, fate: &RecoveredBindingFateRestore) {
    write_fenced_attach(w, &fate.fenced_attach);
    w.put_u64(fate.participant_id);
    write_epoch(w, fate.binding_epoch);
    w.put_u64(fate.resulting_floor);
}

fn read_recovered_fate(
    r: &mut Reader<'_>,
) -> Result<RecoveredBindingFateRestore, CapsuleDecodeError> {
    let fenced_attach = read_fenced_attach(r)?;
    let participant_id = r.take_u64()?;
    let binding_epoch = read_epoch(r)?;
    Ok(RecoveredBindingFateRestore {
        fenced_attach,
        participant_id,
        binding_epoch,
        resulting_floor: r.take_u64()?,
    })
}

fn write_fenced_attach(w: &mut Writer, fenced: &FencedAttachCommitRestore) {
    write_credential_recovery(w, &fenced.predecessor);
    write_wide(w, fenced.predecessor_debt);
    w.put_u64(fenced.participant_id);
    w.put_u64(fenced.marker_delivery_seq);
    write_epoch(w, fenced.prior_binding_epoch);
    write_epoch(w, fenced.new_binding_epoch);
    w.put_u64(fenced.resulting_floor);
    match fenced.successor {
        DebtCompletionRestore::Clear => w.put_u8(1),
        DebtCompletionRestore::ObserverProjection { debt, through_seq } => {
            w.put_u8(2);
            write_wide(w, debt);
            w.put_u64(through_seq);
        }
        DebtCompletionRestore::PhysicalCompaction {
            debt,
            from_floor,
            through_seq,
        } => {
            w.put_u8(3);
            write_wide(w, debt);
            w.put_u64(from_floor);
            w.put_u64(through_seq);
        }
    }
}

fn read_fenced_attach(r: &mut Reader<'_>) -> Result<FencedAttachCommitRestore, CapsuleDecodeError> {
    let predecessor = read_credential_recovery(r)?;
    let predecessor_debt = read_wide(r)?;
    let participant_id = r.take_u64()?;
    let marker_delivery_seq = r.take_u64()?;
    let prior_binding_epoch = read_epoch(r)?;
    let new_binding_epoch = read_epoch(r)?;
    let resulting_floor = r.take_u64()?;
    let successor = match r.take_u8()? {
        1 => DebtCompletionRestore::Clear,
        2 => {
            let debt = read_wide(r)?;
            DebtCompletionRestore::ObserverProjection {
                debt,
                through_seq: r.take_u64()?,
            }
        }
        3 => {
            let debt = read_wide(r)?;
            let from_floor = r.take_u64()?;
            DebtCompletionRestore::PhysicalCompaction {
                debt,
                from_floor,
                through_seq: r.take_u64()?,
            }
        }
        _ => return Err(r.invalid()),
    };
    Ok(FencedAttachCommitRestore {
        predecessor,
        predecessor_debt,
        participant_id,
        marker_delivery_seq,
        prior_binding_epoch,
        new_binding_epoch,
        resulting_floor,
        successor,
    })
}

fn write_storage_completion(w: &mut Writer, completion: RecoveredStorageCompletionRestore) {
    match completion {
        RecoveredStorageCompletionRestore::ObserverProjection {
            through_seq,
            resulting_debt,
        } => {
            w.put_u8(1);
            w.put_u64(through_seq);
            write_optional_wide(w, resulting_debt);
        }
        RecoveredStorageCompletionRestore::PhysicalCompaction {
            from_floor,
            through_seq,
            resulting_floor,
            resulting_debt,
        } => {
            w.put_u8(2);
            w.put_u64(from_floor);
            w.put_u64(through_seq);
            w.put_u64(resulting_floor);
            write_optional_wide(w, resulting_debt);
        }
    }
}

fn read_storage_completion(
    r: &mut Reader<'_>,
) -> Result<RecoveredStorageCompletionRestore, CapsuleDecodeError> {
    match r.take_u8()? {
        1 => {
            let through_seq = r.take_u64()?;
            Ok(RecoveredStorageCompletionRestore::ObserverProjection {
                through_seq,
                resulting_debt: read_optional_wide(r)?,
            })
        }
        2 => {
            let from_floor = r.take_u64()?;
            let through_seq = r.take_u64()?;
            let resulting_floor = r.take_u64()?;
            Ok(RecoveredStorageCompletionRestore::PhysicalCompaction {
                from_floor,
                through_seq,
                resulting_floor,
                resulting_debt: read_optional_wide(r)?,
            })
        }
        _ => Err(r.invalid()),
    }
}

#[cfg(test)]
mod binding_source_tests {
    use super::*;

    #[test]
    fn binding_origin_source_reuses_the_capsule_codec_and_rejects_trailing_bytes() {
        let origin = BindingOrigin::unfenced(AttachedLifecycleRecord::from_binding(
            ActiveBinding {
                participant_id: 0,
                conversation_id: 7,
                binding_epoch: crate::lifecycle::storage_tests::epoch(1, 1),
            },
            AttachedRecordPosition::new(0, 0),
        ));
        let mut existing = Writer::default();
        write_binding_origin(&mut existing, origin);
        let mut source = encode_binding_origin_source(origin);
        assert_eq!(source, existing.bytes);
        assert_eq!(decode_binding_origin_source(&source), Ok(origin));
        source.push(0);
        assert!(decode_binding_origin_source(&source).is_err());
        assert!(decode_binding_origin_source(&[]).is_err());
    }
}

pub(super) use frontier_accounting::{
    decode_accounting, decode_charges, encode_accounting, encode_charges,
};

pub(super) fn encode_marker_source(
    record: &super::claim_frontier::ValidatedMarkerRecord,
) -> Vec<u8> {
    let mut w = Writer::default();
    w.put_u64(record.conversation_id());
    w.put_u64(record.delivery_seq());
    write_admission_order(&mut w, record.admission_order());
    w.put_u8(5);
    w.put_u64(record.participant_id());
    write_marker_provenance(&mut w, record.provenance());
    write_frontier_binding(&mut w, record.target_binding());
    w.put_u8(match record.occurrence() {
        super::claim_frontier::MarkerRecordOccurrence::Undelivered => 1,
        super::claim_frontier::MarkerRecordOccurrence::Delivered => 2,
    });
    w.bytes
}

#[cfg(test)]
mod marker_source_checkpoint_tests {
    use super::*;

    #[test]
    fn marker_source_matches_canonical_row_binding_and_occurrence_writers()
    -> Result<(), &'static str> {
        let generation = Generation::new(2).ok_or("nonzero generation missing")?;
        let epoch = BindingEpoch::new(ConnectionIncarnation::new(7, 9), generation);
        for binding in [
            FrontierBinding::Bound(epoch),
            FrontierBinding::Detached(epoch),
        ] {
            for delivered in [false, true] {
                let record =
                    super::super::edge::validated_marker_record_for_test(51, 0, binding, 5, 0);
                let record = if delivered {
                    record.delivered_for_test()
                } else {
                    record
                };
                let row = RetainedCausalRecord {
                    delivery_seq: record.delivery_seq(),
                    admission_order: record.admission_order(),
                    kind: RetainedCausalRecordKind::CompactionMarker {
                        participant_index: record.participant_id(),
                        provenance: record.provenance(),
                    },
                };
                let mut canonical = Writer::default();
                canonical.put_u64(record.conversation_id());
                write_retained_record(&mut canonical, row);
                write_frontier_binding(&mut canonical, record.target_binding());
                canonical.put_u8(if delivered { 2 } else { 1 });
                assert_eq!(encode_marker_source(&record), canonical.bytes);
            }
        }
        Ok(())
    }
}