1mod frontier_accounting;
12mod observer_sources;
13mod owner_tokens;
14pub use frontier_accounting::FrontierAccountingDecodeError;
15pub use observer_sources::ObserverProjectionCheckpointKind;
16pub(in crate::lifecycle) use observer_sources::{
17 ObserverProjectionSource, decode_observer_source, encode_observer_source,
18};
19pub use owner_tokens::OwnerCheckpointRestoreError;
20pub(in crate::lifecycle) use owner_tokens::{
21 decode_ordinary_fate, encode_fate_token, encode_ordinary_fate, restore_fate_token,
22};
23
24use alloc::vec::Vec;
25
26use crate::algebra::WideResourceVector;
27use crate::outcome::CandidatePhase;
28use crate::wire::{
29 AttachSecret, BindingEpoch, CloseCause, ConnectionIncarnation, DetachAttemptToken, Generation,
30 LeaveAttemptToken, ParticipantId,
31};
32
33use super::checkpoint::CapsuleDecodeError;
34use super::storage::{
35 BindingFateTerminalRestore, BindingStateRestore, ClosureStateRestore,
36 CommittedBindingTerminalRestore, DebtCompletionRestore, DetachCellRestore,
37 DetachedCredentialRecoveryRestore, DetachedCursorReleaseProvenanceRestore,
38 DetachedMarkerReleaseProvenanceRestore, DetachedMarkerReleaseRestore,
39 FencedAttachCommitRestore, LeaveCommittedRestore, LiveIdentityRestore,
40 MarkerCursorProgressRestore, MarkerDeliveryRestore, OrdinaryBindingAuthorityRestore,
41 OrdinaryBindingFateRestore, ParticipantConversationDescription, ParticipantConversationRestore,
42 ParticipantLifecycleRestore, PendingFinalizationRestore, PendingRecoveredCursorReleaseRestore,
43 RecoveredBindingFateRestore, RecoveredStorageCompletionRestore, RetiredIdentityRestore,
44 StoredEdgeRestore,
45};
46use super::{
47 ActiveBinding, AdmissionOrder, AttachedLifecycleRecord, AttachedRecordPosition, BindingOrigin,
48 BindingTerminalOwner, ClaimFrontiersRestore, EnrollmentFingerprint, ExitProductRangeRestore,
49 FrontierBinding, FrontierParticipant, HistoricalCausalFactRestore,
50 HistoricalMarkerDeliveryFactRestore, ImmutableOrderCandidateMajorRestore,
51 ImmutableSequenceCandidate, LeaveFingerprint, MarkerCandidateAuthority, MarkerProvenance,
52 MarkerSequenceOwner, MovableOrderClaim, MovableSequenceClaim, OrderClaimFrontierRestore,
53 OrderClaims, OrderDirectOwner, OrderHigh, OrderLedger, RecoveryOrderActiveBindingRestore,
54 RecoveryOrderBlockRestore, RecoverySequenceBlockRestore, RecoverySequenceReserve,
55 RecoverySequenceTerminalRestore, ReplacementTerminalProductRangeRestore, RetainedCausalRecord,
56 RetainedCausalRecordKind, SequenceClaimFrontierRestore, SequenceClaims, SequenceDirectOwner,
57 SequenceLedger, SequenceProductClass, SequenceProductRangesRestore,
58 TerminalProductRangeRestore, TerminalProductSource,
59};
60
61pub(super) const PARTICIPANTS_SECTION_TAG: u16 = 2;
63pub(super) const FRONTIERS_SECTION_TAG: u16 = 3;
65pub(super) const CLOSURE_SECTION_TAG: u16 = 4;
67pub(super) const SEALED_SECTION_TAG: u16 = 5;
72
73pub const CANONICAL_DIGEST_LEN: usize = 32;
75
76pub trait CanonicalDigest<const N: usize>: Copy {
82 fn to_canonical(self) -> [u8; N];
84 fn from_canonical(bytes: [u8; N]) -> Self;
86}
87
88impl<const N: usize> CanonicalDigest<N> for [u8; N] {
89 fn to_canonical(self) -> [u8; N] {
90 self
91 }
92 fn from_canonical(bytes: [u8; N]) -> Self {
93 bytes
94 }
95}
96
97impl CanonicalDigest<CANONICAL_DIGEST_LEN> for crate::wire::CredentialDigest {
98 fn to_canonical(self) -> [u8; CANONICAL_DIGEST_LEN] {
99 self.into_bytes()
100 }
101 fn from_canonical(bytes: [u8; CANONICAL_DIGEST_LEN]) -> Self {
102 Self::new(bytes)
103 }
104}
105
106pub(super) type DigestDescription<D> = ParticipantConversationDescription<D, D, D, D>;
108
109type DigestParticipant<D> = ParticipantLifecycleRestore<D, D, D, D>;
110
111pub(super) fn encode_description<const N: usize, D: CanonicalDigest<N>>(
116 description: &DigestDescription<D>,
117) -> Option<Vec<u8>> {
118 let restore = &description.restore;
119 let mut participants = Writer::default();
120 write_list(
121 &mut participants,
122 &restore.participants,
123 write_participant::<N, D>,
124 );
125 let mut frontiers = Writer::default();
126 write_frontiers(&mut frontiers, &restore.frontiers);
127 write_sequence_ledger(&mut frontiers, restore.sequence_ledger);
128 write_order_ledger(&mut frontiers, restore.order_ledger);
129 write_option(
130 &mut frontiers,
131 description.fenced_marker_issued.as_ref(),
132 |w, &(participant_id, marker_delivery_seq, binding_epoch)| {
133 w.put_u64(participant_id);
134 w.put_u64(marker_delivery_seq);
135 write_epoch(w, binding_epoch);
136 },
137 );
138 let mut closure = Writer::default();
139 write_closure(&mut closure, &restore.closure);
140 let mut sections = Writer::default();
141 sections.section(PARTICIPANTS_SECTION_TAG, participants);
142 sections.section(FRONTIERS_SECTION_TAG, frontiers);
143 sections.section(CLOSURE_SECTION_TAG, closure);
144 sections.finish()
145}
146
147pub(super) fn encode_sealed_description<const N: usize, D: CanonicalDigest<N>>(
153 participants: &[DigestParticipant<D>],
154 ledger_high_watermark: u64,
155) -> Option<Vec<u8>> {
156 let mut retired = Writer::default();
157 write_list(&mut retired, participants, write_participant::<N, D>);
158 let mut sealed = Writer::default();
159 sealed.put_u64(ledger_high_watermark);
160 let mut sections = Writer::default();
161 sections.section(PARTICIPANTS_SECTION_TAG, retired);
162 sections.section(SEALED_SECTION_TAG, sealed);
163 sections.finish()
164}
165
166pub(super) fn decode_sealed_description<const N: usize, D: CanonicalDigest<N>>(
174 participants: &[u8],
175 sealed: &[u8],
176) -> Result<(Vec<DigestParticipant<D>>, u64), CapsuleDecodeError> {
177 let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
178 let participants = read_participants::<N, D>(&mut reader)?;
179 reader.finish()?;
180 if participants
181 .iter()
182 .any(|participant| matches!(participant, ParticipantLifecycleRestore::Live { .. }))
183 {
184 return Err(reader.invalid());
185 }
186 let mut reader = Reader::new(sealed, SEALED_SECTION_TAG);
187 let ledger_high_watermark = reader.take_u64()?;
188 reader.finish()?;
189 Ok((participants, ledger_high_watermark))
190}
191
192pub(super) fn decode_description<const N: usize, D: CanonicalDigest<N>>(
202 participants: &[u8],
203 frontiers: &[u8],
204 closure: &[u8],
205) -> Result<DigestDescription<D>, CapsuleDecodeError> {
206 let mut reader = Reader::new(participants, PARTICIPANTS_SECTION_TAG);
207 let participants = read_participants::<N, D>(&mut reader)?;
208 reader.finish()?;
209
210 let mut reader = Reader::new(frontiers, FRONTIERS_SECTION_TAG);
211 let frontier_restore = read_frontiers(&mut reader)?;
212 let sequence_ledger = read_sequence_ledger(&mut reader)?;
213 let order_ledger = read_order_ledger(&mut reader)?;
214 let fenced_marker_issued = read_option(&mut reader, |r| {
215 Ok((r.take_u64()?, r.take_u64()?, read_epoch(r)?))
216 })?;
217 reader.finish()?;
218
219 let mut reader = Reader::new(closure, CLOSURE_SECTION_TAG);
220 let closure = read_closure(&mut reader)?;
221 reader.finish()?;
222
223 Ok(ParticipantConversationDescription {
224 restore: ParticipantConversationRestore {
225 participants,
226 frontiers: frontier_restore,
227 sequence_ledger,
228 order_ledger,
229 closure,
230 },
231 fenced_marker_issued,
232 })
233}
234
235pub(super) const fn participant_id_of<EF, V, LF, D>(
237 participant: &ParticipantLifecycleRestore<EF, V, LF, D>,
238) -> ParticipantId {
239 match participant {
240 ParticipantLifecycleRestore::Live { identity, .. } => identity.participant_id,
241 ParticipantLifecycleRestore::Retired(retired) => retired.participant_id,
242 }
243}
244
245#[derive(Debug, Default)]
252struct Writer {
253 bytes: Vec<u8>,
254 overflow: bool,
255}
256
257impl Writer {
258 fn put_u8(&mut self, value: u8) {
259 self.bytes.push(value);
260 }
261
262 fn put_u16(&mut self, value: u16) {
263 self.bytes.extend_from_slice(&value.to_be_bytes());
264 }
265
266 fn put_u32(&mut self, value: u32) {
267 self.bytes.extend_from_slice(&value.to_be_bytes());
268 }
269
270 fn put_u64(&mut self, value: u64) {
271 self.bytes.extend_from_slice(&value.to_be_bytes());
272 }
273
274 fn put_u128(&mut self, value: u128) {
275 self.bytes.extend_from_slice(&value.to_be_bytes());
276 }
277
278 fn put_bytes(&mut self, value: &[u8]) {
279 self.bytes.extend_from_slice(value);
280 }
281
282 fn put_flag(&mut self, value: bool) {
283 self.put_u8(u8::from(value));
284 }
285
286 fn put_count(&mut self, len: usize) {
287 let count = u32::try_from(len);
288 self.overflow |= count.is_err();
289 self.put_u32(count.unwrap_or(u32::MAX));
290 }
291
292 fn section(&mut self, tag: u16, body: Self) {
294 self.overflow |= body.overflow;
295 self.put_u16(tag);
296 self.put_count(body.bytes.len());
297 self.bytes.extend(body.bytes);
298 }
299
300 fn finish(self) -> Option<Vec<u8>> {
301 if self.overflow {
302 None
303 } else {
304 Some(self.bytes)
305 }
306 }
307}
308
309#[derive(Debug)]
311struct Reader<'a> {
312 body: &'a [u8],
313 at: usize,
314 tag: u16,
315}
316
317impl<'a> Reader<'a> {
318 const fn new(body: &'a [u8], tag: u16) -> Self {
319 Self { body, at: 0, tag }
320 }
321
322 fn wrong_length(&self) -> CapsuleDecodeError {
324 CapsuleDecodeError::NonCanonicalSectionLength {
325 tag: self.tag,
326 declared: u32::try_from(self.body.len()).unwrap_or(u32::MAX),
327 }
328 }
329
330 const fn invalid(&self) -> CapsuleDecodeError {
332 CapsuleDecodeError::NonCanonicalSection { tag: self.tag }
333 }
334
335 fn take<const W: usize>(&mut self) -> Result<[u8; W], CapsuleDecodeError> {
336 let body = self.body;
337 let end = self
338 .at
339 .checked_add(W)
340 .ok_or(CapsuleDecodeError::LengthOverflow)?;
341 let taken = body.get(self.at..end).ok_or_else(|| self.wrong_length())?;
342 let array = <[u8; W]>::try_from(taken).map_err(|_| CapsuleDecodeError::LengthOverflow)?;
343 self.at = end;
344 Ok(array)
345 }
346
347 fn take_u8(&mut self) -> Result<u8, CapsuleDecodeError> {
348 self.take::<1>().map(|[value]| value)
349 }
350
351 fn take_u32(&mut self) -> Result<u32, CapsuleDecodeError> {
352 self.take().map(u32::from_be_bytes)
353 }
354
355 fn take_u64(&mut self) -> Result<u64, CapsuleDecodeError> {
356 self.take().map(u64::from_be_bytes)
357 }
358
359 fn take_u128(&mut self) -> Result<u128, CapsuleDecodeError> {
360 self.take().map(u128::from_be_bytes)
361 }
362
363 fn take_flag(&mut self) -> Result<bool, CapsuleDecodeError> {
364 match self.take_u8()? {
365 0 => Ok(false),
366 1 => Ok(true),
367 _ => Err(self.invalid()),
368 }
369 }
370
371 fn take_count(&mut self) -> Result<usize, CapsuleDecodeError> {
372 usize::try_from(self.take_u32()?).map_err(|_| CapsuleDecodeError::LengthOverflow)
373 }
374
375 fn finish(&self) -> Result<(), CapsuleDecodeError> {
376 if self.at == self.body.len() {
377 Ok(())
378 } else {
379 Err(self.wrong_length())
380 }
381 }
382}
383
384fn write_list<T>(w: &mut Writer, items: &[T], mut write: impl FnMut(&mut Writer, &T)) {
385 w.put_count(items.len());
386 for item in items {
387 write(w, item);
388 }
389}
390
391fn read_list<'a, T>(
392 r: &mut Reader<'a>,
393 mut read: impl FnMut(&mut Reader<'a>) -> Result<T, CapsuleDecodeError>,
394) -> Result<Vec<T>, CapsuleDecodeError> {
395 let count = r.take_count()?;
396 let mut items = Vec::new();
397 for _ in 0..count {
398 items.push(read(r)?);
399 }
400 Ok(items)
401}
402
403fn write_option<T>(w: &mut Writer, item: Option<&T>, write: impl FnOnce(&mut Writer, &T)) {
404 w.put_flag(item.is_some());
405 if let Some(item) = item {
406 write(w, item);
407 }
408}
409
410fn read_option<'a, T>(
411 r: &mut Reader<'a>,
412 read: impl FnOnce(&mut Reader<'a>) -> Result<T, CapsuleDecodeError>,
413) -> Result<Option<T>, CapsuleDecodeError> {
414 if r.take_flag()? {
415 read(r).map(Some)
416 } else {
417 Ok(None)
418 }
419}
420
421fn write_generation(w: &mut Writer, generation: Generation) {
426 w.put_u64(generation.get());
427}
428
429fn read_generation(r: &mut Reader<'_>) -> Result<Generation, CapsuleDecodeError> {
430 let value = r.take_u64()?;
431 Generation::new(value).ok_or_else(|| r.invalid())
432}
433
434fn write_epoch(w: &mut Writer, epoch: BindingEpoch) {
435 w.put_u64(epoch.connection_incarnation.server_incarnation);
436 w.put_u64(epoch.connection_incarnation.connection_ordinal);
437 write_generation(w, epoch.capability_generation);
438}
439
440fn read_epoch(r: &mut Reader<'_>) -> Result<BindingEpoch, CapsuleDecodeError> {
441 let server_incarnation = r.take_u64()?;
442 let connection_ordinal = r.take_u64()?;
443 Ok(BindingEpoch::new(
444 ConnectionIncarnation::new(server_incarnation, connection_ordinal),
445 read_generation(r)?,
446 ))
447}
448
449fn write_optional_epoch(w: &mut Writer, epoch: Option<BindingEpoch>) {
450 write_option(w, epoch.as_ref(), |w, epoch| write_epoch(w, *epoch));
451}
452
453fn read_optional_epoch(r: &mut Reader<'_>) -> Result<Option<BindingEpoch>, CapsuleDecodeError> {
454 read_option(r, read_epoch)
455}
456
457fn write_optional_u64(w: &mut Writer, value: Option<u64>) {
458 write_option(w, value.as_ref(), |w, value| w.put_u64(*value));
459}
460
461fn read_optional_u64(r: &mut Reader<'_>) -> Result<Option<u64>, CapsuleDecodeError> {
462 read_option(r, Reader::take_u64)
463}
464
465fn write_wide(w: &mut Writer, value: WideResourceVector) {
466 w.put_u128(value.entries);
467 w.put_u128(value.bytes);
468}
469
470fn read_wide(r: &mut Reader<'_>) -> Result<WideResourceVector, CapsuleDecodeError> {
471 let entries = r.take_u128()?;
472 Ok(WideResourceVector::new(entries, r.take_u128()?))
473}
474
475fn write_optional_wide(w: &mut Writer, value: Option<WideResourceVector>) {
476 write_option(w, value.as_ref(), |w, value| write_wide(w, *value));
477}
478
479fn read_optional_wide(
480 r: &mut Reader<'_>,
481) -> Result<Option<WideResourceVector>, CapsuleDecodeError> {
482 read_option(r, read_wide)
483}
484
485fn write_digest<const N: usize, D: CanonicalDigest<N>>(w: &mut Writer, digest: &D) {
486 w.put_bytes(&digest.to_canonical());
487}
488
489fn read_digest<const N: usize, D: CanonicalDigest<N>>(
490 r: &mut Reader<'_>,
491) -> Result<D, CapsuleDecodeError> {
492 r.take::<N>().map(D::from_canonical)
493}
494
495fn write_active_binding(w: &mut Writer, binding: ActiveBinding) {
496 w.put_u64(binding.participant_id);
497 w.put_u64(binding.conversation_id);
498 write_epoch(w, binding.binding_epoch);
499}
500
501fn read_active_binding(r: &mut Reader<'_>) -> Result<ActiveBinding, CapsuleDecodeError> {
502 let participant_id = r.take_u64()?;
503 let conversation_id = r.take_u64()?;
504 Ok(ActiveBinding {
505 participant_id,
506 conversation_id,
507 binding_epoch: read_epoch(r)?,
508 })
509}
510
511fn write_close_cause(w: &mut Writer, cause: CloseCause) {
512 match cause {
513 CloseCause::CleanDeregister => w.put_u8(1),
514 CloseCause::ConnectionLost => w.put_u8(2),
515 CloseCause::ProcessKilled => w.put_u8(3),
516 CloseCause::ProtocolError => w.put_u8(4),
517 CloseCause::Superseded => w.put_u8(5),
518 CloseCause::ServerShutdown => w.put_u8(6),
519 CloseCause::UncleanServerRestart {
520 prior_server_incarnation,
521 } => {
522 w.put_u8(7);
523 w.put_u64(prior_server_incarnation);
524 }
525 }
526}
527
528fn read_close_cause(r: &mut Reader<'_>) -> Result<CloseCause, CapsuleDecodeError> {
529 Ok(match r.take_u8()? {
530 1 => CloseCause::CleanDeregister,
531 2 => CloseCause::ConnectionLost,
532 3 => CloseCause::ProcessKilled,
533 4 => CloseCause::ProtocolError,
534 5 => CloseCause::Superseded,
535 6 => CloseCause::ServerShutdown,
536 7 => CloseCause::UncleanServerRestart {
537 prior_server_incarnation: r.take_u64()?,
538 },
539 _ => return Err(r.invalid()),
540 })
541}
542
543fn write_admission_order(w: &mut Writer, order: AdmissionOrder) {
544 w.put_u64(order.transaction_order());
545 w.put_u8(match order.candidate_phase() {
546 CandidatePhase::BindingTerminal => 1,
547 CandidatePhase::MembershipExit => 2,
548 CandidatePhase::AttachLifecycle => 3,
549 CandidatePhase::OrdinaryRecord => 4,
550 CandidatePhase::CompactionMarker => 5,
551 });
552 w.put_u64(order.participant_index());
553}
554
555fn read_admission_order(r: &mut Reader<'_>) -> Result<AdmissionOrder, CapsuleDecodeError> {
556 let transaction_order = r.take_u64()?;
557 let phase = match r.take_u8()? {
558 1 => CandidatePhase::BindingTerminal,
559 2 => CandidatePhase::MembershipExit,
560 3 => CandidatePhase::AttachLifecycle,
561 4 => CandidatePhase::OrdinaryRecord,
562 5 => CandidatePhase::CompactionMarker,
563 _ => return Err(r.invalid()),
564 };
565 Ok(AdmissionOrder::new(transaction_order, phase, r.take_u64()?))
566}
567
568fn write_frontier_binding(w: &mut Writer, binding: FrontierBinding) {
569 match binding {
570 FrontierBinding::Bound(epoch) => {
571 w.put_u8(1);
572 write_epoch(w, epoch);
573 }
574 FrontierBinding::Detached(epoch) => {
575 w.put_u8(2);
576 write_epoch(w, epoch);
577 }
578 }
579}
580
581fn read_frontier_binding(r: &mut Reader<'_>) -> Result<FrontierBinding, CapsuleDecodeError> {
582 match r.take_u8()? {
583 1 => read_epoch(r).map(FrontierBinding::Bound),
584 2 => read_epoch(r).map(FrontierBinding::Detached),
585 _ => Err(r.invalid()),
586 }
587}
588
589fn write_terminal_owner(w: &mut Writer, owner: BindingTerminalOwner) {
590 w.put_u64(owner.participant_index);
591 write_epoch(w, owner.binding_epoch);
592}
593
594fn read_terminal_owner(r: &mut Reader<'_>) -> Result<BindingTerminalOwner, CapsuleDecodeError> {
595 let participant_index = r.take_u64()?;
596 Ok(BindingTerminalOwner {
597 participant_index,
598 binding_epoch: read_epoch(r)?,
599 })
600}
601
602fn read_participants<const N: usize, D: CanonicalDigest<N>>(
607 r: &mut Reader<'_>,
608) -> Result<Vec<DigestParticipant<D>>, CapsuleDecodeError> {
609 let participants = read_list(r, read_participant::<N, D>)?;
610 for pair in participants.windows(2) {
611 if let [earlier, later] = pair
612 && participant_id_of(earlier) >= participant_id_of(later)
613 {
614 return Err(CapsuleDecodeError::ParticipantOutOfOrder {
615 participant_id: participant_id_of(later),
616 });
617 }
618 }
619 Ok(participants)
620}
621
622fn write_participant<const N: usize, D: CanonicalDigest<N>>(
623 w: &mut Writer,
624 participant: &DigestParticipant<D>,
625) {
626 match participant {
627 ParticipantLifecycleRestore::Live {
628 identity,
629 binding,
630 binding_origin,
631 detach_cell,
632 } => {
633 w.put_u8(1);
634 write_live_identity(w, identity);
635 write_binding_state(w, *binding);
636 write_option(w, binding_origin.as_ref(), |w, origin| {
637 write_binding_origin(w, *origin);
638 });
639 write_detach_cell(w, detach_cell);
640 }
641 ParticipantLifecycleRestore::Retired(retired) => {
642 w.put_u8(2);
643 write_retired(w, retired);
644 }
645 }
646}
647
648fn read_participant<const N: usize, D: CanonicalDigest<N>>(
649 r: &mut Reader<'_>,
650) -> Result<DigestParticipant<D>, CapsuleDecodeError> {
651 match r.take_u8()? {
652 1 => {
653 let identity = read_live_identity::<N, D>(r)?;
654 let binding = read_binding_state(r)?;
655 let binding_origin = read_option(r, read_binding_origin)?;
656 Ok(ParticipantLifecycleRestore::Live {
657 identity,
658 binding,
659 binding_origin,
660 detach_cell: read_detach_cell::<N, D>(r)?,
661 })
662 }
663 2 => read_retired::<N, D>(r).map(ParticipantLifecycleRestore::Retired),
664 _ => Err(r.invalid()),
665 }
666}
667
668fn write_live_identity<const N: usize, D: CanonicalDigest<N>>(
669 w: &mut Writer,
670 identity: &LiveIdentityRestore<D>,
671) {
672 w.put_u64(identity.participant_id);
673 w.put_u64(identity.conversation_id);
674 write_generation(w, identity.generation);
675 w.put_bytes(identity.attach_secret.as_bytes());
676 w.put_u64(identity.cursor);
677 write_digest(w, identity.enrollment_fingerprint.value());
678 write_option(w, identity.latest_terminal.as_ref(), |w, terminal| {
679 write_committed_terminal(w, *terminal);
680 });
681}
682
683fn read_live_identity<const N: usize, D: CanonicalDigest<N>>(
684 r: &mut Reader<'_>,
685) -> Result<LiveIdentityRestore<D>, CapsuleDecodeError> {
686 let participant_id = r.take_u64()?;
687 let conversation_id = r.take_u64()?;
688 let generation = read_generation(r)?;
689 let attach_secret = AttachSecret::new(r.take()?);
690 let cursor = r.take_u64()?;
691 let enrollment_fingerprint = EnrollmentFingerprint::new(read_digest::<N, D>(r)?);
692 Ok(LiveIdentityRestore {
693 participant_id,
694 conversation_id,
695 generation,
696 attach_secret,
697 cursor,
698 enrollment_fingerprint,
699 latest_terminal: read_option(r, read_committed_terminal)?,
700 })
701}
702
703fn write_committed_terminal(w: &mut Writer, terminal: CommittedBindingTerminalRestore) {
704 write_active_binding(w, terminal.binding);
705 write_close_cause(w, terminal.cause);
706 w.put_u64(terminal.transaction_order);
707 w.put_u64(terminal.delivery_seq);
708}
709
710fn read_committed_terminal(
711 r: &mut Reader<'_>,
712) -> Result<CommittedBindingTerminalRestore, CapsuleDecodeError> {
713 let binding = read_active_binding(r)?;
714 let cause = read_close_cause(r)?;
715 let transaction_order = r.take_u64()?;
716 Ok(CommittedBindingTerminalRestore {
717 binding,
718 cause,
719 transaction_order,
720 delivery_seq: r.take_u64()?,
721 })
722}
723
724fn write_pending_terminal(w: &mut Writer, pending: PendingFinalizationRestore) {
725 write_active_binding(w, pending.binding);
726 write_close_cause(w, pending.cause);
727 w.put_u64(pending.transaction_order);
728}
729
730fn read_pending_terminal(
731 r: &mut Reader<'_>,
732) -> Result<PendingFinalizationRestore, CapsuleDecodeError> {
733 let binding = read_active_binding(r)?;
734 let cause = read_close_cause(r)?;
735 Ok(PendingFinalizationRestore {
736 binding,
737 cause,
738 transaction_order: r.take_u64()?,
739 })
740}
741
742fn write_binding_state(w: &mut Writer, binding: BindingStateRestore) {
743 match binding {
744 BindingStateRestore::Detached => w.put_u8(1),
745 BindingStateRestore::Bound(binding) => {
746 w.put_u8(2);
747 write_active_binding(w, binding);
748 }
749 BindingStateRestore::PendingFinalization(pending) => {
750 w.put_u8(3);
751 write_pending_terminal(w, pending);
752 }
753 }
754}
755
756fn read_binding_state(r: &mut Reader<'_>) -> Result<BindingStateRestore, CapsuleDecodeError> {
757 match r.take_u8()? {
758 1 => Ok(BindingStateRestore::Detached),
759 2 => read_active_binding(r).map(BindingStateRestore::Bound),
760 3 => read_pending_terminal(r).map(BindingStateRestore::PendingFinalization),
761 _ => Err(r.invalid()),
762 }
763}
764
765pub(super) fn encode_binding_origin_source(origin: BindingOrigin) -> Vec<u8> {
770 let mut writer = Writer::default();
771 write_binding_origin(&mut writer, origin);
772 writer.bytes
773}
774
775pub(super) fn decode_binding_origin_source(
776 bytes: &[u8],
777) -> Result<BindingOrigin, CapsuleDecodeError> {
778 let mut reader = Reader::new(bytes, PARTICIPANTS_SECTION_TAG);
779 let origin = read_binding_origin(&mut reader)?;
780 reader.finish()?;
781 Ok(origin)
782}
783
784fn write_binding_origin(w: &mut Writer, origin: BindingOrigin) {
785 let attached = origin.attached();
786 w.put_u64(attached.participant_id());
787 w.put_u64(attached.conversation_id());
788 write_epoch(w, attached.binding_epoch());
789 w.put_u64(attached.admission_order().transaction_order());
790 w.put_u64(attached.delivery_seq());
791 let recovered = origin.recovered_marker();
792 w.put_u8(if recovered.is_none() { 1 } else { 2 });
793 if let Some((marker_delivery_seq, prior_binding_epoch)) = recovered {
794 w.put_u64(marker_delivery_seq);
795 write_epoch(w, prior_binding_epoch);
796 }
797}
798
799fn read_binding_origin(r: &mut Reader<'_>) -> Result<BindingOrigin, CapsuleDecodeError> {
800 let binding = read_active_binding(r)?;
801 let transaction_order = r.take_u64()?;
802 let delivery_seq = r.take_u64()?;
803 let attached = AttachedLifecycleRecord::from_binding(
804 binding,
805 AttachedRecordPosition::new(transaction_order, delivery_seq),
806 );
807 match r.take_u8()? {
808 1 => Ok(BindingOrigin::unfenced(attached)),
809 2 => {
810 let marker_delivery_seq = r.take_u64()?;
811 Ok(BindingOrigin::recovered(
812 attached,
813 marker_delivery_seq,
814 read_epoch(r)?,
815 ))
816 }
817 _ => Err(r.invalid()),
818 }
819}
820
821fn write_detach_cell<const N: usize, D: CanonicalDigest<N>>(
822 w: &mut Writer,
823 cell: &DetachCellRestore<D>,
824) {
825 match cell {
826 DetachCellRestore::Empty => w.put_u8(1),
827 DetachCellRestore::Pending {
828 token,
829 participant_id,
830 request_generation,
831 request_verifier,
832 committed_binding_epoch,
833 admission_order,
834 refused_epoch,
835 } => {
836 w.put_u8(2);
837 write_detach_head(w, *token, *participant_id, *request_generation);
838 write_digest(w, request_verifier);
839 write_epoch(w, *committed_binding_epoch);
840 write_admission_order(w, *admission_order);
841 w.put_u64(*refused_epoch);
842 }
843 DetachCellRestore::Committed {
844 token,
845 participant_id,
846 request_generation,
847 request_verifier,
848 committed_binding_epoch,
849 detached_delivery_seq,
850 } => {
851 w.put_u8(3);
852 write_detach_head(w, *token, *participant_id, *request_generation);
853 write_digest(w, request_verifier);
854 write_epoch(w, *committed_binding_epoch);
855 w.put_u64(*detached_delivery_seq);
856 }
857 DetachCellRestore::Terminalized {
858 token,
859 participant_id,
860 request_generation,
861 request_verifier,
862 committed_binding_epoch,
863 } => {
864 w.put_u8(4);
865 write_detach_head(w, *token, *participant_id, *request_generation);
866 write_digest(w, request_verifier);
867 write_epoch(w, *committed_binding_epoch);
868 }
869 }
870}
871
872fn write_detach_head(
873 w: &mut Writer,
874 token: DetachAttemptToken,
875 participant_id: ParticipantId,
876 request_generation: Generation,
877) {
878 w.put_bytes(token.as_bytes());
879 w.put_u64(participant_id);
880 write_generation(w, request_generation);
881}
882
883fn read_detach_cell<const N: usize, D: CanonicalDigest<N>>(
884 r: &mut Reader<'_>,
885) -> Result<DetachCellRestore<D>, CapsuleDecodeError> {
886 let variant = r.take_u8()?;
887 if variant == 1 {
888 return Ok(DetachCellRestore::Empty);
889 }
890 if !(2..=4).contains(&variant) {
891 return Err(r.invalid());
892 }
893 let token = DetachAttemptToken::new(r.take()?);
894 let participant_id = r.take_u64()?;
895 let request_generation = read_generation(r)?;
896 let request_verifier = read_digest::<N, D>(r)?;
897 let committed_binding_epoch = read_epoch(r)?;
898 Ok(match variant {
899 2 => {
900 let admission_order = read_admission_order(r)?;
901 DetachCellRestore::Pending {
902 token,
903 participant_id,
904 request_generation,
905 request_verifier,
906 committed_binding_epoch,
907 admission_order,
908 refused_epoch: r.take_u64()?,
909 }
910 }
911 3 => DetachCellRestore::Committed {
912 token,
913 participant_id,
914 request_generation,
915 request_verifier,
916 committed_binding_epoch,
917 detached_delivery_seq: r.take_u64()?,
918 },
919 _ => DetachCellRestore::Terminalized {
920 token,
921 participant_id,
922 request_generation,
923 request_verifier,
924 committed_binding_epoch,
925 },
926 })
927}
928
929fn write_retired<const N: usize, D: CanonicalDigest<N>>(
930 w: &mut Writer,
931 retired: &RetiredIdentityRestore<D, D, D>,
932) {
933 w.put_u64(retired.participant_id);
934 w.put_u64(retired.conversation_id);
935 write_generation(w, retired.retired_generation);
936 write_digest(w, retired.enrollment_fingerprint.value());
937 w.put_bytes(retired.leave_attempt_token.as_bytes());
938 write_digest(w, &retired.leave_request_verifier);
939 write_digest(w, retired.leave_fingerprint.value());
940 w.put_u64(retired.left_transaction_order);
941 write_leave_committed(w, retired.committed_result);
942}
943
944impl<D> RetiredIdentityRestore<D, D, D> {
945 #[must_use]
949 pub fn encode_canonical<const N: usize>(&self) -> Vec<u8>
950 where
951 D: CanonicalDigest<N>,
952 {
953 let mut writer = Writer::default();
954 write_retired(&mut writer, self);
955 writer.bytes
956 }
957
958 pub fn decode_canonical<const N: usize>(
966 bytes: &[u8],
967 conversation_id: u64,
968 participant_id: ParticipantId,
969 ) -> Result<Self, CapsuleDecodeError>
970 where
971 D: CanonicalDigest<N>,
972 {
973 let mut reader = Reader::new(bytes, PARTICIPANTS_SECTION_TAG);
974 let retired = read_retired(&mut reader)?;
975 reader.finish()?;
976 if retired.conversation_id != conversation_id {
977 return Err(CapsuleDecodeError::ForeignConversation {
978 tag: PARTICIPANTS_SECTION_TAG,
979 });
980 }
981 if retired.participant_id != participant_id {
982 return Err(reader.invalid());
983 }
984 retired.restore_result().map_err(|_| reader.invalid())?;
985 Ok(retired)
986 }
987}
988
989type RetiredRestore<D> = RetiredIdentityRestore<D, D, D>;
991
992fn read_retired<const N: usize, D: CanonicalDigest<N>>(
993 r: &mut Reader<'_>,
994) -> Result<RetiredRestore<D>, CapsuleDecodeError> {
995 let participant_id = r.take_u64()?;
996 let conversation_id = r.take_u64()?;
997 let retired_generation = read_generation(r)?;
998 let enrollment_fingerprint = EnrollmentFingerprint::new(read_digest::<N, D>(r)?);
999 let leave_attempt_token = LeaveAttemptToken::new(r.take()?);
1000 let leave_request_verifier = read_digest::<N, D>(r)?;
1001 let leave_fingerprint = LeaveFingerprint::new(read_digest::<N, D>(r)?);
1002 let left_transaction_order = r.take_u64()?;
1003 Ok(RetiredIdentityRestore {
1004 participant_id,
1005 conversation_id,
1006 retired_generation,
1007 enrollment_fingerprint,
1008 leave_attempt_token,
1009 leave_request_verifier,
1010 leave_fingerprint,
1011 left_transaction_order,
1012 committed_result: read_leave_committed(r)?,
1013 })
1014}
1015
1016fn write_leave_committed(w: &mut Writer, result: LeaveCommittedRestore) {
1017 w.put_u64(result.conversation_id);
1018 w.put_bytes(result.leave_attempt_token.as_bytes());
1019 w.put_u64(result.participant_id);
1020 write_generation(w, result.retired_generation);
1021 write_optional_epoch(w, result.ended_binding_epoch);
1022 write_optional_u64(w, result.prior_terminal_delivery_seq);
1023 w.put_u64(result.left_delivery_seq);
1024}
1025
1026fn read_leave_committed(r: &mut Reader<'_>) -> Result<LeaveCommittedRestore, CapsuleDecodeError> {
1027 let conversation_id = r.take_u64()?;
1028 let leave_attempt_token = LeaveAttemptToken::new(r.take()?);
1029 let participant_id = r.take_u64()?;
1030 let retired_generation = read_generation(r)?;
1031 let ended_binding_epoch = read_optional_epoch(r)?;
1032 let prior_terminal_delivery_seq = read_optional_u64(r)?;
1033 Ok(LeaveCommittedRestore {
1034 conversation_id,
1035 leave_attempt_token,
1036 participant_id,
1037 retired_generation,
1038 ended_binding_epoch,
1039 prior_terminal_delivery_seq,
1040 left_delivery_seq: r.take_u64()?,
1041 })
1042}
1043
1044fn write_frontiers(w: &mut Writer, restore: &ClaimFrontiersRestore) {
1049 w.put_u64(restore.conversation_id);
1050 write_list(w, &restore.active_identities, |w, participant| {
1051 w.put_u64(participant.participant_index());
1052 w.put_u64(participant.cursor());
1053 write_frontier_binding(w, participant.binding());
1054 });
1055 w.put_u64(restore.identity_slot_limit);
1056 w.put_u128(restore.retained_floor);
1057 w.put_u64(restore.retained_record_limit);
1058 write_list(w, &restore.retained_records, |w, record| {
1059 write_retained_record(w, *record);
1060 });
1061 write_list(w, &restore.active_marker_anchors, |w, anchor| {
1062 w.put_u64(*anchor);
1063 });
1064 write_list(w, &restore.historical_marker_deliveries, |w, fact| {
1065 w.put_u64(fact.conversation_id);
1066 w.put_u64(fact.participant_index);
1067 w.put_u64(fact.marker_delivery_seq);
1068 write_epoch(w, fact.delivered_binding_epoch);
1069 });
1070 write_list(w, &restore.historical_causal_facts, |w, fact| {
1071 write_causal_fact(w, *fact);
1072 });
1073 write_sequence_frontier(w, &restore.sequence);
1074 write_order_frontier(w, &restore.order);
1075 write_optional_u64(w, restore.recovery_marker_delivery_seq);
1076}
1077
1078fn read_frontiers(r: &mut Reader<'_>) -> Result<ClaimFrontiersRestore, CapsuleDecodeError> {
1079 let conversation_id = r.take_u64()?;
1080 let active_identities = read_list(r, |r| {
1081 let participant_index = r.take_u64()?;
1082 let cursor = r.take_u64()?;
1083 Ok(FrontierParticipant::new(
1084 participant_index,
1085 cursor,
1086 read_frontier_binding(r)?,
1087 ))
1088 })?;
1089 let identity_slot_limit = r.take_u64()?;
1090 let retained_floor = r.take_u128()?;
1091 let retained_record_limit = r.take_u64()?;
1092 let retained_records = read_list(r, read_retained_record)?;
1093 let active_marker_anchors = read_list(r, Reader::take_u64)?;
1094 let historical_marker_deliveries = read_list(r, |r| {
1095 let conversation_id = r.take_u64()?;
1096 let participant_index = r.take_u64()?;
1097 let marker_delivery_seq = r.take_u64()?;
1098 Ok(HistoricalMarkerDeliveryFactRestore {
1099 conversation_id,
1100 participant_index,
1101 marker_delivery_seq,
1102 delivered_binding_epoch: read_epoch(r)?,
1103 })
1104 })?;
1105 let historical_causal_facts = read_list(r, read_causal_fact)?;
1106 let sequence = read_sequence_frontier(r)?;
1107 let order = read_order_frontier(r)?;
1108 Ok(ClaimFrontiersRestore {
1109 conversation_id,
1110 active_identities,
1111 identity_slot_limit,
1112 retained_floor,
1113 retained_record_limit,
1114 retained_records,
1115 active_marker_anchors,
1116 historical_marker_deliveries,
1117 historical_causal_facts,
1118 sequence,
1119 order,
1120 recovery_marker_delivery_seq: read_optional_u64(r)?,
1121 })
1122}
1123
1124fn write_retained_record(w: &mut Writer, record: RetainedCausalRecord) {
1125 w.put_u64(record.delivery_seq);
1126 write_admission_order(w, record.admission_order);
1127 match record.kind {
1128 RetainedCausalRecordKind::BindingTerminal(owner) => {
1129 w.put_u8(1);
1130 write_terminal_owner(w, owner);
1131 }
1132 RetainedCausalRecordKind::MembershipExit { participant_index } => {
1133 w.put_u8(2);
1134 w.put_u64(participant_index);
1135 }
1136 RetainedCausalRecordKind::AttachLifecycle {
1137 participant_index,
1138 binding_epoch,
1139 } => {
1140 w.put_u8(3);
1141 w.put_u64(participant_index);
1142 write_epoch(w, binding_epoch);
1143 }
1144 RetainedCausalRecordKind::OrdinaryRecord { participant_index } => {
1145 w.put_u8(4);
1146 w.put_u64(participant_index);
1147 }
1148 RetainedCausalRecordKind::CompactionMarker {
1149 participant_index,
1150 provenance,
1151 } => {
1152 w.put_u8(5);
1153 w.put_u64(participant_index);
1154 write_marker_provenance(w, provenance);
1155 }
1156 }
1157}
1158
1159fn read_retained_record(r: &mut Reader<'_>) -> Result<RetainedCausalRecord, CapsuleDecodeError> {
1160 let delivery_seq = r.take_u64()?;
1161 let admission_order = read_admission_order(r)?;
1162 let kind = match r.take_u8()? {
1163 1 => RetainedCausalRecordKind::BindingTerminal(read_terminal_owner(r)?),
1164 2 => RetainedCausalRecordKind::MembershipExit {
1165 participant_index: r.take_u64()?,
1166 },
1167 3 => {
1168 let participant_index = r.take_u64()?;
1169 RetainedCausalRecordKind::AttachLifecycle {
1170 participant_index,
1171 binding_epoch: read_epoch(r)?,
1172 }
1173 }
1174 4 => RetainedCausalRecordKind::OrdinaryRecord {
1175 participant_index: r.take_u64()?,
1176 },
1177 5 => {
1178 let participant_index = r.take_u64()?;
1179 RetainedCausalRecordKind::CompactionMarker {
1180 participant_index,
1181 provenance: read_marker_provenance(r)?,
1182 }
1183 }
1184 _ => return Err(r.invalid()),
1185 };
1186 Ok(RetainedCausalRecord {
1187 delivery_seq,
1188 admission_order,
1189 kind,
1190 })
1191}
1192
1193fn write_marker_provenance(w: &mut Writer, provenance: MarkerProvenance) {
1194 match provenance {
1195 MarkerProvenance::NonProductM => w.put_u8(1),
1196 MarkerProvenance::TerminalProduct {
1197 terminal,
1198 affected_participant,
1199 } => {
1200 w.put_u8(2);
1201 match terminal {
1202 TerminalProductSource::Binding(owner) => {
1203 w.put_u8(1);
1204 write_terminal_owner(w, owner);
1205 }
1206 TerminalProductSource::RecoveryReplacement {
1207 participant_index,
1208 binding_epoch,
1209 } => {
1210 w.put_u8(2);
1211 w.put_u64(participant_index);
1212 write_epoch(w, binding_epoch);
1213 }
1214 }
1215 w.put_u64(affected_participant);
1216 }
1217 MarkerProvenance::ExitProduct {
1218 exit_participant,
1219 remaining_participant,
1220 } => {
1221 w.put_u8(3);
1222 w.put_u64(exit_participant);
1223 w.put_u64(remaining_participant);
1224 }
1225 }
1226}
1227
1228fn read_marker_provenance(r: &mut Reader<'_>) -> Result<MarkerProvenance, CapsuleDecodeError> {
1229 match r.take_u8()? {
1230 1 => Ok(MarkerProvenance::NonProductM),
1231 2 => {
1232 let terminal = match r.take_u8()? {
1233 1 => TerminalProductSource::Binding(read_terminal_owner(r)?),
1234 2 => {
1235 let participant_index = r.take_u64()?;
1236 TerminalProductSource::recovery_replacement(participant_index, read_epoch(r)?)
1237 }
1238 _ => return Err(r.invalid()),
1239 };
1240 Ok(MarkerProvenance::terminal_product(terminal, r.take_u64()?))
1241 }
1242 3 => {
1243 let exit_participant = r.take_u64()?;
1244 Ok(MarkerProvenance::exit_product(
1245 exit_participant,
1246 r.take_u64()?,
1247 ))
1248 }
1249 _ => Err(r.invalid()),
1250 }
1251}
1252
1253fn write_causal_fact(w: &mut Writer, fact: HistoricalCausalFactRestore) {
1254 match fact {
1255 HistoricalCausalFactRestore::BindingTerminal {
1256 conversation_id,
1257 participant_index,
1258 binding_epoch,
1259 admission_order,
1260 } => {
1261 w.put_u8(1);
1262 w.put_u64(conversation_id);
1263 w.put_u64(participant_index);
1264 write_epoch(w, binding_epoch);
1265 write_admission_order(w, admission_order);
1266 }
1267 HistoricalCausalFactRestore::MembershipExit {
1268 conversation_id,
1269 participant_index,
1270 admission_order,
1271 } => {
1272 w.put_u8(2);
1273 w.put_u64(conversation_id);
1274 w.put_u64(participant_index);
1275 write_admission_order(w, admission_order);
1276 }
1277 }
1278}
1279
1280fn read_causal_fact(r: &mut Reader<'_>) -> Result<HistoricalCausalFactRestore, CapsuleDecodeError> {
1281 let variant = r.take_u8()?;
1282 if !(1..=2).contains(&variant) {
1283 return Err(r.invalid());
1284 }
1285 let conversation_id = r.take_u64()?;
1286 let participant_index = r.take_u64()?;
1287 if variant == 1 {
1288 let binding_epoch = read_epoch(r)?;
1289 Ok(HistoricalCausalFactRestore::BindingTerminal {
1290 conversation_id,
1291 participant_index,
1292 binding_epoch,
1293 admission_order: read_admission_order(r)?,
1294 })
1295 } else {
1296 Ok(HistoricalCausalFactRestore::MembershipExit {
1297 conversation_id,
1298 participant_index,
1299 admission_order: read_admission_order(r)?,
1300 })
1301 }
1302}
1303
1304fn write_sequence_frontier(w: &mut Writer, sequence: &SequenceClaimFrontierRestore) {
1305 write_list(w, &sequence.movable_claims, |w, claim| {
1306 w.put_u64(claim.delivery_seq);
1307 write_sequence_owner(w, claim.owner);
1308 });
1309 write_list(w, &sequence.immutable_candidates, |w, candidate| {
1310 write_sequence_candidate(w, *candidate);
1311 });
1312 let products = &sequence.products;
1313 write_list(w, &products.live_times_terminal, |w, range| {
1314 w.put_u64(range.start);
1315 w.put_u64(range.length);
1316 write_terminal_owner(w, range.terminal);
1317 });
1318 write_option(
1319 w,
1320 products.live_times_replacement_terminal.as_ref(),
1321 |w, range| {
1322 w.put_u64(range.start);
1323 w.put_u64(range.length);
1324 },
1325 );
1326 write_list(w, &products.other_live_times_exit, |w, range| {
1327 w.put_u64(range.start);
1328 w.put_u64(range.length);
1329 w.put_u64(range.exit_participant);
1330 });
1331 write_option(w, sequence.recovery.as_ref(), |w, block| {
1332 write_option(w, block.terminal.as_ref(), |w, terminal| {
1333 w.put_u64(terminal.delivery_seq);
1334 write_terminal_owner(w, terminal.owner);
1335 });
1336 w.put_u64(block.recovery_attach_seq);
1337 w.put_u64(block.replacement_terminal_seq);
1338 });
1339}
1340
1341fn read_sequence_frontier(
1342 r: &mut Reader<'_>,
1343) -> Result<SequenceClaimFrontierRestore, CapsuleDecodeError> {
1344 let movable_claims = read_list(r, |r| {
1345 let delivery_seq = r.take_u64()?;
1346 Ok(MovableSequenceClaim {
1347 delivery_seq,
1348 owner: read_sequence_owner(r)?,
1349 })
1350 })?;
1351 let immutable_candidates = read_list(r, read_sequence_candidate)?;
1352 let live_times_terminal = read_list(r, |r| {
1353 let start = r.take_u64()?;
1354 let length = r.take_u64()?;
1355 Ok(TerminalProductRangeRestore {
1356 start,
1357 length,
1358 terminal: read_terminal_owner(r)?,
1359 })
1360 })?;
1361 let live_times_replacement_terminal = read_option(r, |r| {
1362 let start = r.take_u64()?;
1363 Ok(ReplacementTerminalProductRangeRestore {
1364 start,
1365 length: r.take_u64()?,
1366 })
1367 })?;
1368 let other_live_times_exit = read_list(r, |r| {
1369 let start = r.take_u64()?;
1370 let length = r.take_u64()?;
1371 Ok(ExitProductRangeRestore {
1372 start,
1373 length,
1374 exit_participant: r.take_u64()?,
1375 })
1376 })?;
1377 let recovery = read_option(r, |r| {
1378 let terminal = read_option(r, |r| {
1379 let delivery_seq = r.take_u64()?;
1380 Ok(RecoverySequenceTerminalRestore {
1381 delivery_seq,
1382 owner: read_terminal_owner(r)?,
1383 })
1384 })?;
1385 let recovery_attach_seq = r.take_u64()?;
1386 Ok(RecoverySequenceBlockRestore {
1387 terminal,
1388 recovery_attach_seq,
1389 replacement_terminal_seq: r.take_u64()?,
1390 })
1391 })?;
1392 Ok(SequenceClaimFrontierRestore {
1393 movable_claims,
1394 immutable_candidates,
1395 products: SequenceProductRangesRestore {
1396 live_times_terminal,
1397 live_times_replacement_terminal,
1398 other_live_times_exit,
1399 },
1400 recovery,
1401 })
1402}
1403
1404fn write_sequence_owner(w: &mut Writer, owner: SequenceDirectOwner) {
1405 match owner {
1406 SequenceDirectOwner::MembershipExit { participant_index } => {
1407 w.put_u8(1);
1408 w.put_u64(participant_index);
1409 }
1410 SequenceDirectOwner::BindingTerminal(owner) => {
1411 w.put_u8(2);
1412 write_terminal_owner(w, owner);
1413 }
1414 }
1415}
1416
1417fn read_sequence_owner(r: &mut Reader<'_>) -> Result<SequenceDirectOwner, CapsuleDecodeError> {
1418 match r.take_u8()? {
1419 1 => Ok(SequenceDirectOwner::MembershipExit {
1420 participant_index: r.take_u64()?,
1421 }),
1422 2 => read_terminal_owner(r).map(SequenceDirectOwner::BindingTerminal),
1423 _ => Err(r.invalid()),
1424 }
1425}
1426
1427fn write_sequence_candidate(w: &mut Writer, candidate: ImmutableSequenceCandidate) {
1428 match candidate {
1429 ImmutableSequenceCandidate::BindingTerminal {
1430 delivery_seq,
1431 admission_order,
1432 owner,
1433 } => {
1434 w.put_u8(1);
1435 w.put_u64(delivery_seq);
1436 write_admission_order(w, admission_order);
1437 write_terminal_owner(w, owner);
1438 }
1439 ImmutableSequenceCandidate::Marker(marker) => {
1440 w.put_u8(2);
1441 w.put_u64(marker.delivery_seq);
1442 write_admission_order(w, marker.admission_order);
1443 write_frontier_binding(w, marker.target_binding);
1444 write_marker_provenance(w, marker.provenance);
1445 w.put_u64(marker.abandoned_after);
1446 w.put_u64(marker.abandoned_through);
1447 w.put_u64(marker.physical_floor_at_decision);
1448 match marker.current_owner {
1449 MarkerSequenceOwner::Marker => w.put_u8(1),
1450 MarkerSequenceOwner::ConditionalProduct(class) => {
1451 w.put_u8(2);
1452 w.put_u8(match class {
1453 SequenceProductClass::LiveTimesTerminal => 1,
1454 SequenceProductClass::LiveTimesReplacementTerminal => 2,
1455 SequenceProductClass::OtherLiveTimesExit => 3,
1456 });
1457 }
1458 }
1459 }
1460 }
1461}
1462
1463fn read_sequence_candidate(
1464 r: &mut Reader<'_>,
1465) -> Result<ImmutableSequenceCandidate, CapsuleDecodeError> {
1466 match r.take_u8()? {
1467 1 => {
1468 let delivery_seq = r.take_u64()?;
1469 let admission_order = read_admission_order(r)?;
1470 Ok(ImmutableSequenceCandidate::BindingTerminal {
1471 delivery_seq,
1472 admission_order,
1473 owner: read_terminal_owner(r)?,
1474 })
1475 }
1476 2 => read_marker_candidate(r).map(ImmutableSequenceCandidate::Marker),
1477 _ => Err(r.invalid()),
1478 }
1479}
1480
1481fn read_marker_candidate(
1482 r: &mut Reader<'_>,
1483) -> Result<MarkerCandidateAuthority, CapsuleDecodeError> {
1484 let delivery_seq = r.take_u64()?;
1485 let admission_order = read_admission_order(r)?;
1486 let target_binding = read_frontier_binding(r)?;
1487 let provenance = read_marker_provenance(r)?;
1488 let abandoned_after = r.take_u64()?;
1489 let abandoned_through = r.take_u64()?;
1490 let physical_floor_at_decision = r.take_u64()?;
1491 let current_owner = match r.take_u8()? {
1492 1 => MarkerSequenceOwner::Marker,
1493 2 => MarkerSequenceOwner::ConditionalProduct(match r.take_u8()? {
1494 1 => SequenceProductClass::LiveTimesTerminal,
1495 2 => SequenceProductClass::LiveTimesReplacementTerminal,
1496 3 => SequenceProductClass::OtherLiveTimesExit,
1497 _ => return Err(r.invalid()),
1498 }),
1499 _ => return Err(r.invalid()),
1500 };
1501 Ok(MarkerCandidateAuthority {
1502 delivery_seq,
1503 admission_order,
1504 target_binding,
1505 provenance,
1506 abandoned_after,
1507 abandoned_through,
1508 physical_floor_at_decision,
1509 current_owner,
1510 })
1511}
1512
1513fn write_order_frontier(w: &mut Writer, order: &OrderClaimFrontierRestore) {
1514 write_list(w, &order.movable_claims, |w, claim| {
1515 w.put_u64(claim.transaction_order);
1516 match claim.owner {
1517 OrderDirectOwner::ActiveBindingTerminal(owner) => {
1518 w.put_u8(1);
1519 write_terminal_owner(w, owner);
1520 }
1521 OrderDirectOwner::MembershipExit { participant_index } => {
1522 w.put_u8(2);
1523 w.put_u64(participant_index);
1524 }
1525 }
1526 });
1527 write_list(w, &order.immutable_candidates, |w, major| {
1528 w.put_u64(major.transaction_order);
1529 write_list(w, &major.candidate_keys, |w, key| {
1530 write_admission_order(w, *key);
1531 });
1532 });
1533 write_option(w, order.recovery.as_ref(), |w, block| {
1534 write_option(w, block.active_binding.as_ref(), |w, active| {
1535 w.put_u64(active.transaction_order);
1536 write_terminal_owner(w, active.owner);
1537 });
1538 w.put_u64(block.recovery_operation_order);
1539 w.put_u64(block.replacement_terminal_order);
1540 });
1541}
1542
1543fn read_order_frontier(
1544 r: &mut Reader<'_>,
1545) -> Result<OrderClaimFrontierRestore, CapsuleDecodeError> {
1546 let movable_claims = read_list(r, |r| {
1547 let transaction_order = r.take_u64()?;
1548 let owner = match r.take_u8()? {
1549 1 => OrderDirectOwner::ActiveBindingTerminal(read_terminal_owner(r)?),
1550 2 => OrderDirectOwner::MembershipExit {
1551 participant_index: r.take_u64()?,
1552 },
1553 _ => return Err(r.invalid()),
1554 };
1555 Ok(MovableOrderClaim {
1556 transaction_order,
1557 owner,
1558 })
1559 })?;
1560 let immutable_candidates = read_list(r, |r| {
1561 let transaction_order = r.take_u64()?;
1562 Ok(ImmutableOrderCandidateMajorRestore {
1563 transaction_order,
1564 candidate_keys: read_list(r, read_admission_order)?,
1565 })
1566 })?;
1567 let recovery = read_option(r, |r| {
1568 let active_binding = read_option(r, |r| {
1569 let transaction_order = r.take_u64()?;
1570 Ok(RecoveryOrderActiveBindingRestore {
1571 transaction_order,
1572 owner: read_terminal_owner(r)?,
1573 })
1574 })?;
1575 let recovery_operation_order = r.take_u64()?;
1576 Ok(RecoveryOrderBlockRestore {
1577 active_binding,
1578 recovery_operation_order,
1579 replacement_terminal_order: r.take_u64()?,
1580 })
1581 })?;
1582 Ok(OrderClaimFrontierRestore {
1583 movable_claims,
1584 immutable_candidates,
1585 recovery,
1586 })
1587}
1588
1589fn write_sequence_ledger(w: &mut Writer, ledger: SequenceLedger) {
1590 w.put_u64(ledger.high_watermark());
1591 let claims = ledger.claims();
1592 w.put_u64(claims.live_members());
1593 w.put_u64(claims.binding_terminals());
1594 w.put_u64(claims.markers());
1595 w.put_u8(match claims.recovery() {
1596 RecoverySequenceReserve::None => 1,
1597 RecoverySequenceReserve::DetachedCredentialRecovery => 2,
1598 });
1599}
1600
1601fn read_sequence_ledger(r: &mut Reader<'_>) -> Result<SequenceLedger, CapsuleDecodeError> {
1604 let high_watermark = r.take_u64()?;
1605 let live_members = r.take_u64()?;
1606 let binding_terminals = r.take_u64()?;
1607 let markers = r.take_u64()?;
1608 let recovery = match r.take_u8()? {
1609 1 => RecoverySequenceReserve::None,
1610 2 => RecoverySequenceReserve::DetachedCredentialRecovery,
1611 _ => return Err(r.invalid()),
1612 };
1613 SequenceLedger::try_new(
1614 high_watermark,
1615 SequenceClaims::new(live_members, binding_terminals, markers, recovery),
1616 )
1617 .map_err(|_| r.invalid())
1618}
1619
1620fn write_order_ledger(w: &mut Writer, ledger: OrderLedger) {
1621 match ledger.high() {
1622 OrderHigh::Empty => w.put_u8(1),
1623 OrderHigh::Allocated(order) => {
1624 w.put_u8(2);
1625 w.put_u64(order);
1626 }
1627 }
1628 let claims = ledger.claims();
1629 w.put_u64(claims.active_binding_terminals());
1630 w.put_u64(claims.membership_exits());
1631 w.put_flag(claims.recovery_operation());
1632 w.put_flag(claims.recovery_replacement_terminal());
1633}
1634
1635fn read_order_ledger(r: &mut Reader<'_>) -> Result<OrderLedger, CapsuleDecodeError> {
1636 let high = match r.take_u8()? {
1637 1 => OrderHigh::Empty,
1638 2 => OrderHigh::Allocated(r.take_u64()?),
1639 _ => return Err(r.invalid()),
1640 };
1641 let active_binding_terminals = r.take_u64()?;
1642 let membership_exits = r.take_u64()?;
1643 let recovery_operation = r.take_flag()?;
1644 let recovery_replacement_terminal = r.take_flag()?;
1645 let claims = OrderClaims::new(
1646 active_binding_terminals,
1647 membership_exits,
1648 recovery_operation,
1649 recovery_replacement_terminal,
1650 )
1651 .map_err(|_| r.invalid())?;
1652 OrderLedger::try_new(high, claims).map_err(|_| r.invalid())
1653}
1654
1655fn write_closure(w: &mut Writer, closure: &ClosureStateRestore) {
1660 match closure {
1661 ClosureStateRestore::Clear => w.put_u8(1),
1662 ClosureStateRestore::Owed { debt, edge } => {
1663 w.put_u8(2);
1664 write_wide(w, *debt);
1665 write_edge(w, edge);
1666 }
1667 }
1668}
1669
1670fn read_closure(r: &mut Reader<'_>) -> Result<ClosureStateRestore, CapsuleDecodeError> {
1671 match r.take_u8()? {
1672 1 => Ok(ClosureStateRestore::Clear),
1673 2 => {
1674 let debt = read_wide(r)?;
1675 Ok(ClosureStateRestore::Owed {
1676 debt,
1677 edge: read_edge(r)?,
1678 })
1679 }
1680 _ => Err(r.invalid()),
1681 }
1682}
1683
1684fn write_edge(w: &mut Writer, edge: &StoredEdgeRestore) {
1685 match edge {
1686 StoredEdgeRestore::ObserverProjection { through_seq } => {
1687 w.put_u8(1);
1688 w.put_u64(*through_seq);
1689 }
1690 StoredEdgeRestore::PhysicalCompaction {
1691 from_floor,
1692 through_seq,
1693 } => {
1694 w.put_u8(2);
1695 w.put_u64(*from_floor);
1696 w.put_u64(*through_seq);
1697 }
1698 StoredEdgeRestore::MarkerDelivery(delivery) => {
1699 w.put_u8(3);
1700 write_marker_delivery(w, *delivery);
1701 }
1702 StoredEdgeRestore::ParticipantCursorProgressContinuous {
1703 participant_id,
1704 binding_epoch,
1705 through_seq,
1706 authority,
1707 } => {
1708 w.put_u8(4);
1709 w.put_u64(*participant_id);
1710 write_epoch(w, *binding_epoch);
1711 w.put_u64(*through_seq);
1712 write_ordinary_authority(w, *authority);
1713 }
1714 StoredEdgeRestore::ParticipantCursorProgressMarker(progress) => {
1715 w.put_u8(5);
1716 write_marker_progress(w, progress);
1717 }
1718 StoredEdgeRestore::DetachedCredentialRecovery(recovery) => {
1719 w.put_u8(6);
1720 write_credential_recovery(w, recovery);
1721 }
1722 StoredEdgeRestore::DetachedMarkerRelease(release) => {
1723 w.put_u8(7);
1724 write_marker_release(w, release);
1725 }
1726 StoredEdgeRestore::DetachedCursorRelease {
1727 participant_id,
1728 last_dead_binding_epoch,
1729 provenance,
1730 } => {
1731 w.put_u8(8);
1732 w.put_u64(*participant_id);
1733 write_epoch(w, *last_dead_binding_epoch);
1734 write_cursor_release_provenance(w, provenance);
1735 }
1736 }
1737}
1738
1739fn read_edge(r: &mut Reader<'_>) -> Result<StoredEdgeRestore, CapsuleDecodeError> {
1740 Ok(match r.take_u8()? {
1741 1 => StoredEdgeRestore::ObserverProjection {
1742 through_seq: r.take_u64()?,
1743 },
1744 2 => {
1745 let from_floor = r.take_u64()?;
1746 StoredEdgeRestore::PhysicalCompaction {
1747 from_floor,
1748 through_seq: r.take_u64()?,
1749 }
1750 }
1751 3 => StoredEdgeRestore::MarkerDelivery(read_marker_delivery(r)?),
1752 4 => {
1753 let participant_id = r.take_u64()?;
1754 let binding_epoch = read_epoch(r)?;
1755 let through_seq = r.take_u64()?;
1756 StoredEdgeRestore::ParticipantCursorProgressContinuous {
1757 participant_id,
1758 binding_epoch,
1759 through_seq,
1760 authority: read_ordinary_authority(r)?,
1761 }
1762 }
1763 5 => StoredEdgeRestore::ParticipantCursorProgressMarker(read_marker_progress(r)?),
1764 6 => StoredEdgeRestore::DetachedCredentialRecovery(read_credential_recovery(r)?),
1765 7 => StoredEdgeRestore::DetachedMarkerRelease(read_marker_release(r)?),
1766 8 => {
1767 let participant_id = r.take_u64()?;
1768 let last_dead_binding_epoch = read_epoch(r)?;
1769 StoredEdgeRestore::DetachedCursorRelease {
1770 participant_id,
1771 last_dead_binding_epoch,
1772 provenance: read_cursor_release_provenance(r)?,
1773 }
1774 }
1775 _ => return Err(r.invalid()),
1776 })
1777}
1778
1779fn write_marker_delivery(w: &mut Writer, delivery: MarkerDeliveryRestore) {
1780 w.put_u64(delivery.participant_id);
1781 write_epoch(w, delivery.binding_epoch);
1782 w.put_u64(delivery.marker_delivery_seq);
1783}
1784
1785fn read_marker_delivery(r: &mut Reader<'_>) -> Result<MarkerDeliveryRestore, CapsuleDecodeError> {
1786 let participant_id = r.take_u64()?;
1787 let binding_epoch = read_epoch(r)?;
1788 Ok(MarkerDeliveryRestore {
1789 participant_id,
1790 binding_epoch,
1791 marker_delivery_seq: r.take_u64()?,
1792 })
1793}
1794
1795fn write_ordinary_authority(w: &mut Writer, authority: OrdinaryBindingAuthorityRestore) {
1796 write_active_binding(w, authority.binding);
1797 w.put_u64(authority.through_seq);
1798}
1799
1800fn read_ordinary_authority(
1801 r: &mut Reader<'_>,
1802) -> Result<OrdinaryBindingAuthorityRestore, CapsuleDecodeError> {
1803 let binding = read_active_binding(r)?;
1804 Ok(OrdinaryBindingAuthorityRestore {
1805 binding,
1806 through_seq: r.take_u64()?,
1807 })
1808}
1809
1810fn write_marker_progress(w: &mut Writer, progress: &MarkerCursorProgressRestore) {
1811 w.put_u64(progress.conversation_id);
1812 w.put_u64(progress.participant_id);
1813 write_epoch(w, progress.binding_epoch);
1814 w.put_u64(progress.through_seq);
1815 w.put_u64(progress.marker_delivery_seq);
1816 write_marker_delivery(w, progress.delivery);
1817}
1818
1819fn read_marker_progress(
1820 r: &mut Reader<'_>,
1821) -> Result<MarkerCursorProgressRestore, CapsuleDecodeError> {
1822 let conversation_id = r.take_u64()?;
1823 let participant_id = r.take_u64()?;
1824 let binding_epoch = read_epoch(r)?;
1825 let through_seq = r.take_u64()?;
1826 let marker_delivery_seq = r.take_u64()?;
1827 Ok(MarkerCursorProgressRestore {
1828 conversation_id,
1829 participant_id,
1830 binding_epoch,
1831 through_seq,
1832 marker_delivery_seq,
1833 delivery: read_marker_delivery(r)?,
1834 })
1835}
1836
1837fn write_fate_terminal(w: &mut Writer, terminal: BindingFateTerminalRestore) {
1838 match terminal {
1839 BindingFateTerminalRestore::Committed(committed) => {
1840 w.put_u8(1);
1841 write_committed_terminal(w, committed);
1842 }
1843 BindingFateTerminalRestore::Pending(pending) => {
1844 w.put_u8(2);
1845 write_pending_terminal(w, pending);
1846 }
1847 }
1848}
1849
1850fn read_fate_terminal(
1851 r: &mut Reader<'_>,
1852) -> Result<BindingFateTerminalRestore, CapsuleDecodeError> {
1853 match r.take_u8()? {
1854 1 => read_committed_terminal(r).map(BindingFateTerminalRestore::Committed),
1855 2 => read_pending_terminal(r).map(BindingFateTerminalRestore::Pending),
1856 _ => Err(r.invalid()),
1857 }
1858}
1859
1860fn write_credential_recovery(w: &mut Writer, recovery: &DetachedCredentialRecoveryRestore) {
1861 w.put_u64(recovery.participant_id);
1862 w.put_u64(recovery.marker_delivery_seq);
1863 write_epoch(w, recovery.prior_binding_epoch);
1864 w.put_u64(recovery.resulting_floor);
1865 write_fate_terminal(w, recovery.terminal);
1866 write_marker_progress(w, &recovery.progress);
1867}
1868
1869fn read_credential_recovery(
1870 r: &mut Reader<'_>,
1871) -> Result<DetachedCredentialRecoveryRestore, CapsuleDecodeError> {
1872 let participant_id = r.take_u64()?;
1873 let marker_delivery_seq = r.take_u64()?;
1874 let prior_binding_epoch = read_epoch(r)?;
1875 let resulting_floor = r.take_u64()?;
1876 let terminal = read_fate_terminal(r)?;
1877 Ok(DetachedCredentialRecoveryRestore {
1878 participant_id,
1879 marker_delivery_seq,
1880 prior_binding_epoch,
1881 resulting_floor,
1882 terminal,
1883 progress: read_marker_progress(r)?,
1884 })
1885}
1886
1887fn write_marker_release(w: &mut Writer, release: &DetachedMarkerReleaseRestore) {
1888 w.put_u64(release.conversation_id);
1889 w.put_u64(release.participant_id);
1890 w.put_u64(release.marker_delivery_seq);
1891 write_epoch(w, release.last_dead_binding_epoch);
1892 write_marker_delivery(w, release.delivery);
1893 match release.provenance {
1894 DetachedMarkerReleaseProvenanceRestore::Fate { terminal } => {
1895 w.put_u8(1);
1896 write_fate_terminal(w, terminal);
1897 }
1898 DetachedMarkerReleaseProvenanceRestore::Candidate => w.put_u8(2),
1899 }
1900}
1901
1902fn read_marker_release(
1903 r: &mut Reader<'_>,
1904) -> Result<DetachedMarkerReleaseRestore, CapsuleDecodeError> {
1905 let conversation_id = r.take_u64()?;
1906 let participant_id = r.take_u64()?;
1907 let marker_delivery_seq = r.take_u64()?;
1908 let last_dead_binding_epoch = read_epoch(r)?;
1909 let delivery = read_marker_delivery(r)?;
1910 let provenance = match r.take_u8()? {
1911 1 => DetachedMarkerReleaseProvenanceRestore::Fate {
1912 terminal: read_fate_terminal(r)?,
1913 },
1914 2 => DetachedMarkerReleaseProvenanceRestore::Candidate,
1915 _ => return Err(r.invalid()),
1916 };
1917 Ok(DetachedMarkerReleaseRestore {
1918 conversation_id,
1919 participant_id,
1920 marker_delivery_seq,
1921 last_dead_binding_epoch,
1922 delivery,
1923 provenance,
1924 })
1925}
1926
1927fn write_cursor_release_provenance(
1928 w: &mut Writer,
1929 provenance: &DetachedCursorReleaseProvenanceRestore,
1930) {
1931 match provenance {
1932 DetachedCursorReleaseProvenanceRestore::Ordinary(fate) => {
1933 w.put_u8(1);
1934 write_ordinary_authority(w, fate.authority);
1935 write_committed_terminal(w, fate.terminal);
1936 w.put_u64(fate.resulting_floor);
1937 }
1938 DetachedCursorReleaseProvenanceRestore::RecoveredDirect {
1939 fate,
1940 resulting_debt,
1941 } => {
1942 w.put_u8(2);
1943 write_recovered_fate(w, fate);
1944 write_wide(w, *resulting_debt);
1945 }
1946 DetachedCursorReleaseProvenanceRestore::RecoveredAfterStorage {
1947 pending,
1948 completion,
1949 } => {
1950 w.put_u8(3);
1951 write_recovered_fate(w, &pending.fate);
1952 write_wide(w, pending.resulting_debt);
1953 write_storage_completion(w, *completion);
1954 }
1955 }
1956}
1957
1958fn read_cursor_release_provenance(
1959 r: &mut Reader<'_>,
1960) -> Result<DetachedCursorReleaseProvenanceRestore, CapsuleDecodeError> {
1961 Ok(match r.take_u8()? {
1962 1 => {
1963 let authority = read_ordinary_authority(r)?;
1964 let terminal = read_committed_terminal(r)?;
1965 DetachedCursorReleaseProvenanceRestore::Ordinary(OrdinaryBindingFateRestore {
1966 authority,
1967 terminal,
1968 resulting_floor: r.take_u64()?,
1969 })
1970 }
1971 2 => {
1972 let fate = read_recovered_fate(r)?;
1973 DetachedCursorReleaseProvenanceRestore::RecoveredDirect {
1974 fate,
1975 resulting_debt: read_wide(r)?,
1976 }
1977 }
1978 3 => {
1979 let fate = read_recovered_fate(r)?;
1980 let resulting_debt = read_wide(r)?;
1981 DetachedCursorReleaseProvenanceRestore::RecoveredAfterStorage {
1982 pending: PendingRecoveredCursorReleaseRestore {
1983 fate,
1984 resulting_debt,
1985 },
1986 completion: read_storage_completion(r)?,
1987 }
1988 }
1989 _ => return Err(r.invalid()),
1990 })
1991}
1992
1993fn write_recovered_fate(w: &mut Writer, fate: &RecoveredBindingFateRestore) {
1994 write_fenced_attach(w, &fate.fenced_attach);
1995 w.put_u64(fate.participant_id);
1996 write_epoch(w, fate.binding_epoch);
1997 w.put_u64(fate.resulting_floor);
1998}
1999
2000fn read_recovered_fate(
2001 r: &mut Reader<'_>,
2002) -> Result<RecoveredBindingFateRestore, CapsuleDecodeError> {
2003 let fenced_attach = read_fenced_attach(r)?;
2004 let participant_id = r.take_u64()?;
2005 let binding_epoch = read_epoch(r)?;
2006 Ok(RecoveredBindingFateRestore {
2007 fenced_attach,
2008 participant_id,
2009 binding_epoch,
2010 resulting_floor: r.take_u64()?,
2011 })
2012}
2013
2014fn write_fenced_attach(w: &mut Writer, fenced: &FencedAttachCommitRestore) {
2015 write_credential_recovery(w, &fenced.predecessor);
2016 write_wide(w, fenced.predecessor_debt);
2017 w.put_u64(fenced.participant_id);
2018 w.put_u64(fenced.marker_delivery_seq);
2019 write_epoch(w, fenced.prior_binding_epoch);
2020 write_epoch(w, fenced.new_binding_epoch);
2021 w.put_u64(fenced.resulting_floor);
2022 match fenced.successor {
2023 DebtCompletionRestore::Clear => w.put_u8(1),
2024 DebtCompletionRestore::ObserverProjection { debt, through_seq } => {
2025 w.put_u8(2);
2026 write_wide(w, debt);
2027 w.put_u64(through_seq);
2028 }
2029 DebtCompletionRestore::PhysicalCompaction {
2030 debt,
2031 from_floor,
2032 through_seq,
2033 } => {
2034 w.put_u8(3);
2035 write_wide(w, debt);
2036 w.put_u64(from_floor);
2037 w.put_u64(through_seq);
2038 }
2039 }
2040}
2041
2042fn read_fenced_attach(r: &mut Reader<'_>) -> Result<FencedAttachCommitRestore, CapsuleDecodeError> {
2043 let predecessor = read_credential_recovery(r)?;
2044 let predecessor_debt = read_wide(r)?;
2045 let participant_id = r.take_u64()?;
2046 let marker_delivery_seq = r.take_u64()?;
2047 let prior_binding_epoch = read_epoch(r)?;
2048 let new_binding_epoch = read_epoch(r)?;
2049 let resulting_floor = r.take_u64()?;
2050 let successor = match r.take_u8()? {
2051 1 => DebtCompletionRestore::Clear,
2052 2 => {
2053 let debt = read_wide(r)?;
2054 DebtCompletionRestore::ObserverProjection {
2055 debt,
2056 through_seq: r.take_u64()?,
2057 }
2058 }
2059 3 => {
2060 let debt = read_wide(r)?;
2061 let from_floor = r.take_u64()?;
2062 DebtCompletionRestore::PhysicalCompaction {
2063 debt,
2064 from_floor,
2065 through_seq: r.take_u64()?,
2066 }
2067 }
2068 _ => return Err(r.invalid()),
2069 };
2070 Ok(FencedAttachCommitRestore {
2071 predecessor,
2072 predecessor_debt,
2073 participant_id,
2074 marker_delivery_seq,
2075 prior_binding_epoch,
2076 new_binding_epoch,
2077 resulting_floor,
2078 successor,
2079 })
2080}
2081
2082fn write_storage_completion(w: &mut Writer, completion: RecoveredStorageCompletionRestore) {
2083 match completion {
2084 RecoveredStorageCompletionRestore::ObserverProjection {
2085 through_seq,
2086 resulting_debt,
2087 } => {
2088 w.put_u8(1);
2089 w.put_u64(through_seq);
2090 write_optional_wide(w, resulting_debt);
2091 }
2092 RecoveredStorageCompletionRestore::PhysicalCompaction {
2093 from_floor,
2094 through_seq,
2095 resulting_floor,
2096 resulting_debt,
2097 } => {
2098 w.put_u8(2);
2099 w.put_u64(from_floor);
2100 w.put_u64(through_seq);
2101 w.put_u64(resulting_floor);
2102 write_optional_wide(w, resulting_debt);
2103 }
2104 }
2105}
2106
2107fn read_storage_completion(
2108 r: &mut Reader<'_>,
2109) -> Result<RecoveredStorageCompletionRestore, CapsuleDecodeError> {
2110 match r.take_u8()? {
2111 1 => {
2112 let through_seq = r.take_u64()?;
2113 Ok(RecoveredStorageCompletionRestore::ObserverProjection {
2114 through_seq,
2115 resulting_debt: read_optional_wide(r)?,
2116 })
2117 }
2118 2 => {
2119 let from_floor = r.take_u64()?;
2120 let through_seq = r.take_u64()?;
2121 let resulting_floor = r.take_u64()?;
2122 Ok(RecoveredStorageCompletionRestore::PhysicalCompaction {
2123 from_floor,
2124 through_seq,
2125 resulting_floor,
2126 resulting_debt: read_optional_wide(r)?,
2127 })
2128 }
2129 _ => Err(r.invalid()),
2130 }
2131}
2132
2133#[cfg(test)]
2134mod binding_source_tests {
2135 use super::*;
2136
2137 #[test]
2138 fn binding_origin_source_reuses_the_capsule_codec_and_rejects_trailing_bytes() {
2139 let origin = BindingOrigin::unfenced(AttachedLifecycleRecord::from_binding(
2140 ActiveBinding {
2141 participant_id: 0,
2142 conversation_id: 7,
2143 binding_epoch: crate::lifecycle::storage_tests::epoch(1, 1),
2144 },
2145 AttachedRecordPosition::new(0, 0),
2146 ));
2147 let mut existing = Writer::default();
2148 write_binding_origin(&mut existing, origin);
2149 let mut source = encode_binding_origin_source(origin);
2150 assert_eq!(source, existing.bytes);
2151 assert_eq!(decode_binding_origin_source(&source), Ok(origin));
2152 source.push(0);
2153 assert!(decode_binding_origin_source(&source).is_err());
2154 assert!(decode_binding_origin_source(&[]).is_err());
2155 }
2156}
2157
2158pub(super) use frontier_accounting::{
2159 decode_accounting, decode_charges, encode_accounting, encode_charges,
2160};
2161
2162pub(super) fn encode_marker_source(
2163 record: &super::claim_frontier::ValidatedMarkerRecord,
2164) -> Vec<u8> {
2165 let mut w = Writer::default();
2166 w.put_u64(record.conversation_id());
2167 w.put_u64(record.delivery_seq());
2168 write_admission_order(&mut w, record.admission_order());
2169 w.put_u8(5);
2170 w.put_u64(record.participant_id());
2171 write_marker_provenance(&mut w, record.provenance());
2172 write_frontier_binding(&mut w, record.target_binding());
2173 w.put_u8(match record.occurrence() {
2174 super::claim_frontier::MarkerRecordOccurrence::Undelivered => 1,
2175 super::claim_frontier::MarkerRecordOccurrence::Delivered => 2,
2176 });
2177 w.bytes
2178}
2179
2180#[cfg(test)]
2181mod marker_source_checkpoint_tests {
2182 use super::*;
2183
2184 #[test]
2185 fn marker_source_matches_canonical_row_binding_and_occurrence_writers()
2186 -> Result<(), &'static str> {
2187 let generation = Generation::new(2).ok_or("nonzero generation missing")?;
2188 let epoch = BindingEpoch::new(ConnectionIncarnation::new(7, 9), generation);
2189 for binding in [
2190 FrontierBinding::Bound(epoch),
2191 FrontierBinding::Detached(epoch),
2192 ] {
2193 for delivered in [false, true] {
2194 let record =
2195 super::super::edge::validated_marker_record_for_test(51, 0, binding, 5, 0);
2196 let record = if delivered {
2197 record.delivered_for_test()
2198 } else {
2199 record
2200 };
2201 let row = RetainedCausalRecord {
2202 delivery_seq: record.delivery_seq(),
2203 admission_order: record.admission_order(),
2204 kind: RetainedCausalRecordKind::CompactionMarker {
2205 participant_index: record.participant_id(),
2206 provenance: record.provenance(),
2207 },
2208 };
2209 let mut canonical = Writer::default();
2210 canonical.put_u64(record.conversation_id());
2211 write_retained_record(&mut canonical, row);
2212 write_frontier_binding(&mut canonical, record.target_binding());
2213 canonical.put_u8(if delivered { 2 } else { 1 });
2214 assert_eq!(encode_marker_source(&record), canonical.bytes);
2215 }
2216 }
2217 Ok(())
2218 }
2219}