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