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,
};
pub(super) const PARTICIPANTS_SECTION_TAG: u16 = 2;
pub(super) const FRONTIERS_SECTION_TAG: u16 = 3;
pub(super) const CLOSURE_SECTION_TAG: u16 = 4;
pub(super) const SEALED_SECTION_TAG: u16 = 5;
pub(super) type DigestDescription<const N: usize> =
ParticipantConversationDescription<[u8; N], [u8; N], [u8; N], [u8; N]>;
type DigestParticipant<const N: usize> =
ParticipantLifecycleRestore<[u8; N], [u8; N], [u8; N], [u8; N]>;
pub(super) fn encode_description<const N: usize>(
description: &DigestDescription<N>,
) -> Option<Vec<u8>> {
let restore = &description.restore;
let mut participants = Writer::default();
write_list(
&mut participants,
&restore.participants,
write_participant::<N>,
);
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()
}
pub(super) fn encode_sealed_description<const N: usize>(
participants: &[DigestParticipant<N>],
ledger_high_watermark: u64,
) -> Option<Vec<u8>> {
let mut retired = Writer::default();
write_list(&mut retired, participants, write_participant::<N>);
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()
}
pub(super) fn decode_sealed_description<const N: usize>(
participants: &[u8],
sealed: &[u8],
) -> Result<(Vec<DigestParticipant<N>>, u64), CapsuleDecodeError> {
let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
let participants = read_participants::<N>(&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))
}
pub(super) fn decode_description<const N: usize>(
participants: &[u8],
frontiers: &[u8],
closure: &[u8],
) -> Result<DigestDescription<N>, CapsuleDecodeError> {
let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
let participants = read_participants::<N>(&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,
})
}
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,
}
}
#[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));
}
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)
}
}
}
#[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 }
}
fn wrong_length(&self) -> CapsuleDecodeError {
CapsuleDecodeError::NonCanonicalSectionLength {
tag: self.tag,
declared: u32::try_from(self.body.len()).unwrap_or(u32::MAX),
}
}
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)
}
}
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>(w: &mut Writer, digest: &[u8; N]) {
w.put_bytes(digest);
}
fn read_digest<const N: usize>(r: &mut Reader<'_>) -> Result<[u8; N], CapsuleDecodeError> {
r.take::<N>()
}
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)?,
})
}
fn read_participants<const N: usize>(
r: &mut Reader<'_>,
) -> Result<Vec<DigestParticipant<N>>, CapsuleDecodeError> {
let participants = read_list(r, read_participant::<N>)?;
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>(w: &mut Writer, participant: &DigestParticipant<N>) {
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>(
r: &mut Reader<'_>,
) -> Result<DigestParticipant<N>, CapsuleDecodeError> {
match r.take_u8()? {
1 => {
let identity = read_live_identity::<N>(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>(r)?,
})
}
2 => read_retired::<N>(r).map(ParticipantLifecycleRestore::Retired),
_ => Err(r.invalid()),
}
}
fn write_live_identity<const N: usize>(w: &mut Writer, identity: &LiveIdentityRestore<[u8; N]>) {
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>(
r: &mut Reader<'_>,
) -> Result<LiveIdentityRestore<[u8; N]>, 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>(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()),
}
}
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>(w: &mut Writer, cell: &DetachCellRestore<[u8; N]>) {
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>(
r: &mut Reader<'_>,
) -> Result<DetachCellRestore<[u8; N]>, 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>(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>(
w: &mut Writer,
retired: &RetiredIdentityRestore<[u8; N], [u8; N], [u8; N]>,
) {
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<const N: usize> RetiredIdentityRestore<[u8; N], [u8; N], [u8; N]> {
#[must_use]
pub fn encode_canonical(&self) -> Vec<u8> {
let mut writer = Writer::default();
write_retired(&mut writer, self);
writer.bytes
}
pub fn decode_canonical(
bytes: &[u8],
conversation_id: u64,
participant_id: ParticipantId,
) -> Result<Self, CapsuleDecodeError> {
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)
}
}
type RetiredRestore<const N: usize> = RetiredIdentityRestore<[u8; N], [u8; N], [u8; N]>;
fn read_retired<const N: usize>(
r: &mut Reader<'_>,
) -> Result<RetiredRestore<N>, 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>(r)?);
let leave_attempt_token = LeaveAttemptToken::new(r.take()?);
let leave_request_verifier = read_digest::<N>(r)?;
let leave_fingerprint = LeaveFingerprint::new(read_digest::<N>(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()?,
})
}
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,
});
}
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())
}
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(())
}
}