1use alloc::collections::BTreeSet;
21use alloc::format;
22use alloc::string::String;
23use alloc::vec::Vec;
24
25use crate::wire::{ConversationId, CredentialDigest, ParticipantId};
26
27use super::checkpoint_codec::{
28 CLOSURE_SECTION_TAG, DigestDescription, FRONTIERS_SECTION_TAG, PARTICIPANTS_SECTION_TAG,
29 SEALED_SECTION_TAG, decode_description, decode_sealed_description, encode_description,
30 encode_sealed_description,
31};
32use super::claim_frontier::ClaimFrontiers;
33use super::conversation::{ConversationGenesis, ParticipantConversation};
34use super::edge::ClosureState;
35use super::membership::RetiredIdentity;
36use super::storage::{
37 ConversationDescribeError, ConversationStateRestoreError, ParticipantConversationState,
38 ParticipantLifecycleRestore, ParticipantLifecycleView, StorageRestoreError,
39 describe_conversation,
40};
41
42pub type DigestParticipantView<'a> = ParticipantLifecycleView<
44 'a,
45 CredentialDigest,
46 CredentialDigest,
47 CredentialDigest,
48 CredentialDigest,
49>;
50
51pub type DigestConversationState = ParticipantConversationState<
53 CredentialDigest,
54 CredentialDigest,
55 CredentialDigest,
56 CredentialDigest,
57>;
58
59pub type DigestRetiredIdentity =
61 RetiredIdentity<CredentialDigest, CredentialDigest, CredentialDigest>;
62
63type DigestParticipantRestore = ParticipantLifecycleRestore<
64 CredentialDigest,
65 CredentialDigest,
66 CredentialDigest,
67 CredentialDigest,
68>;
69
70const CAPSULE_MAGIC: [u8; 4] = *b"LPCK";
71const CAPSULE_CODEC_MAJOR: u16 = 1;
72const CAPSULE_CODEC_MINOR: u16 = 0;
75const CAPSULE_CODEC_MINOR_SEALED: u16 = 1;
78const CAPSULE_HEADER_LEN: usize = 28;
80const SECTION_HEADER_LEN: u32 = 6;
82const CONVERSATION_SECTION_TAG: u16 = 1;
83const CONVERSATION_SECTION_LEN: u32 = 9;
85const GENESIS_VALIDATED_FLAG: u8 = 0b1;
86const CONVERSATION_FRAME_LEN: u32 = SECTION_HEADER_LEN + CONVERSATION_SECTION_LEN;
88const SECTION_TAGS: [u16; 5] = [
93 CONVERSATION_SECTION_TAG,
94 PARTICIPANTS_SECTION_TAG,
95 FRONTIERS_SECTION_TAG,
96 CLOSURE_SECTION_TAG,
97 SEALED_SECTION_TAG,
98];
99
100#[derive(Debug)]
103pub struct LifecycleView<'a> {
104 pub conversation: &'a ParticipantConversation,
106 pub lifecycle: Option<ConversationLifecycleView<'a>>,
109 pub sealed: Option<SealedLifecycleView<'a>>,
112 pub covered_log_sequence: u64,
114}
115
116#[derive(Debug)]
120pub struct SealedLifecycleView<'a> {
121 pub participants: Vec<DigestParticipantView<'a>>,
124 pub ledger_high_watermark: u64,
126}
127
128#[derive(Debug)]
130pub struct ConversationLifecycleView<'a> {
131 pub participants: Vec<DigestParticipantView<'a>>,
134 pub frontiers: &'a ClaimFrontiers,
136 pub retained_record_limit: u64,
138 pub closure: &'a ClosureState,
140}
141
142#[derive(Debug, PartialEq, Eq)]
146pub struct LifecycleCapsule {
147 conversation_id: ConversationId,
148 covered_log_sequence: u64,
149 conversation: ConversationSection,
150 lifecycle: Option<LifecycleSections>,
151 sealed: Option<SealedSections>,
152 body_len: u32,
155}
156
157#[derive(Debug, PartialEq, Eq)]
160struct SealedSections {
161 participants: Vec<DigestParticipantRestore>,
162 ledger_high_watermark: u64,
163 encoded: Vec<u8>,
164}
165
166#[derive(Clone, Copy, Debug, PartialEq, Eq)]
168struct ConversationSection {
169 next_event_ordinal: u64,
170 genesis_validated: bool,
171}
172
173impl ConversationSection {
174 const fn is_consistent(self) -> bool {
177 self.genesis_validated == (self.next_event_ordinal > 0)
178 }
179}
180
181#[derive(Debug, PartialEq, Eq)]
184struct LifecycleSections {
185 description: DigestDescription<CredentialDigest>,
186 encoded: Vec<u8>,
187}
188
189impl LifecycleCapsule {
190 pub fn capture(view: &LifecycleView<'_>) -> Result<Self, CapsuleCaptureError> {
199 let conversation_id = view.conversation.conversation_id();
200 if view.lifecycle.is_some() && view.sealed.is_some() {
201 return Err(CapsuleCaptureError::SealedBesideLifecycle);
202 }
203 let lifecycle = view
204 .lifecycle
205 .as_ref()
206 .map(|lifecycle| capture_lifecycle(conversation_id, lifecycle))
207 .transpose()?;
208 let sealed = view.sealed.as_ref().map(capture_sealed).transpose()?;
209 let lifecycle_len = lifecycle
210 .as_ref()
211 .map_or(0, |sections| sections.encoded.len())
212 .checked_add(sealed.as_ref().map_or(0, |sections| sections.encoded.len()))
213 .ok_or(CapsuleCaptureError::LengthOverflow)?;
214 let body_len = u32::try_from(lifecycle_len)
215 .map_err(|_| CapsuleCaptureError::LengthOverflow)?
216 .checked_add(CONVERSATION_FRAME_LEN)
217 .ok_or(CapsuleCaptureError::LengthOverflow)?;
218 Ok(Self {
219 conversation_id,
220 covered_log_sequence: view.covered_log_sequence,
221 conversation: ConversationSection {
222 next_event_ordinal: view.conversation.next_event_ordinal(),
223 genesis_validated: view.conversation.genesis_validated(),
224 },
225 lifecycle,
226 sealed,
227 body_len,
228 })
229 }
230
231 #[must_use]
233 pub const fn is_sealed(&self) -> bool {
234 self.sealed.is_some()
235 }
236
237 #[must_use]
239 pub const fn conversation_id(&self) -> ConversationId {
240 self.conversation_id
241 }
242
243 #[must_use]
245 pub const fn covered_log_sequence(&self) -> u64 {
246 self.covered_log_sequence
247 }
248
249 #[must_use]
251 pub fn encode_canonical(&self) -> Vec<u8> {
252 let mut encoded = Vec::with_capacity(CAPSULE_HEADER_LEN + self.body_len as usize);
253 let minor = if self.sealed.is_some() {
254 CAPSULE_CODEC_MINOR_SEALED
255 } else {
256 CAPSULE_CODEC_MINOR
257 };
258 encoded.extend_from_slice(&CAPSULE_MAGIC);
259 encoded.extend_from_slice(&CAPSULE_CODEC_MAJOR.to_be_bytes());
260 encoded.extend_from_slice(&minor.to_be_bytes());
261 encoded.extend_from_slice(&self.conversation_id.to_be_bytes());
262 encoded.extend_from_slice(&self.covered_log_sequence.to_be_bytes());
263 encoded.extend_from_slice(&self.body_len.to_be_bytes());
264 encoded.extend_from_slice(&CONVERSATION_SECTION_TAG.to_be_bytes());
265 encoded.extend_from_slice(&CONVERSATION_SECTION_LEN.to_be_bytes());
266 encoded.extend_from_slice(&self.conversation.next_event_ordinal.to_be_bytes());
267 encoded.push(if self.conversation.genesis_validated {
268 GENESIS_VALIDATED_FLAG
269 } else {
270 0
271 });
272 if let Some(sections) = &self.lifecycle {
273 encoded.extend_from_slice(§ions.encoded);
274 }
275 if let Some(sections) = &self.sealed {
276 encoded.extend_from_slice(§ions.encoded);
277 }
278 encoded
279 }
280
281 pub fn decode_canonical(input: &[u8]) -> Result<Self, CapsuleDecodeError> {
291 if input.len() < CAPSULE_HEADER_LEN {
292 return Err(CapsuleDecodeError::Truncated {
293 required: CAPSULE_HEADER_LEN,
294 available: input.len(),
295 });
296 }
297 if input.get(0..4) != Some(CAPSULE_MAGIC.as_slice()) {
298 return Err(CapsuleDecodeError::InvalidMagic);
299 }
300 let major = take_u16(input, 4)?;
301 let minor = take_u16(input, 6)?;
302 if major != CAPSULE_CODEC_MAJOR || minor > CAPSULE_CODEC_MINOR_SEALED {
303 return Err(CapsuleDecodeError::UnsupportedCodec { major, minor });
304 }
305 let conversation_id = take_u64(input, 8)?;
306 let covered_log_sequence = take_u64(input, 16)?;
307 let declared_body_len = take_u32(input, 24)?;
308 let actual_body_len = input.len() - CAPSULE_HEADER_LEN;
309 if usize::try_from(declared_body_len).map_err(|_| CapsuleDecodeError::LengthOverflow)?
310 != actual_body_len
311 {
312 return Err(CapsuleDecodeError::NonCanonicalLength {
313 declared_body_len,
314 actual_body_len,
315 });
316 }
317 let mut at = CAPSULE_HEADER_LEN;
318 let conversation =
319 decode_conversation(take_section(input, &mut at, CONVERSATION_SECTION_TAG)?)?;
320 let (lifecycle, sealed) = if input.len() - at < SECTION_HEADER_LEN as usize {
321 (None, None)
322 } else {
323 let start = at;
324 let participants = take_section(input, &mut at, PARTICIPANTS_SECTION_TAG)?;
325 let next_tag = if input.len() - at < SECTION_HEADER_LEN as usize {
326 None
327 } else {
328 Some(take_u16(input, at)?)
329 };
330 if next_tag == Some(SEALED_SECTION_TAG) {
331 if minor != CAPSULE_CODEC_MINOR_SEALED {
332 return Err(CapsuleDecodeError::UnknownSection {
333 tag: SEALED_SECTION_TAG,
334 });
335 }
336 let sealed = take_section(input, &mut at, SEALED_SECTION_TAG)?;
337 let (participants, ledger_high_watermark) =
338 decode_sealed_description::<32, CredentialDigest>(participants, sealed)?;
339 let encoded = input
340 .get(start..at)
341 .ok_or(CapsuleDecodeError::LengthOverflow)?
342 .to_vec();
343 (
344 None,
345 Some(SealedSections {
346 participants,
347 ledger_high_watermark,
348 encoded,
349 }),
350 )
351 } else {
352 (
353 Some(decode_lifecycle(
354 input,
355 &mut at,
356 start,
357 participants,
358 conversation_id,
359 )?),
360 None,
361 )
362 }
363 };
364 if minor == CAPSULE_CODEC_MINOR_SEALED && sealed.is_none() {
365 return Err(CapsuleDecodeError::MissingSection {
366 tag: SEALED_SECTION_TAG,
367 });
368 }
369 if at != input.len() {
370 return Err(CapsuleDecodeError::TrailingBytes {
371 trailing: input.len() - at,
372 });
373 }
374 Ok(Self {
375 conversation_id,
376 covered_log_sequence,
377 conversation,
378 lifecycle,
379 sealed,
380 body_len: declared_body_len,
381 })
382 }
383
384 #[doc(hidden)]
391 pub fn restore_with_binding_sources(
392 self,
393 sources: &[(ParticipantId, &[u8])],
394 ) -> Result<RestoredLifecycle, BindingOriginSourceError> {
395 let mut sources = sources.iter();
396 if let Some(lifecycle) = &self.lifecycle {
397 for participant in &lifecycle.description.restore.participants {
398 if let super::storage::ParticipantLifecycleRestore::Live {
399 identity,
400 binding_origin: Some(expected),
401 ..
402 } = participant
403 {
404 let Some(&(participant_id, bytes)) = sources.next() else {
405 return Err(BindingOriginSourceError::Association {
406 participant_id: identity.participant_id,
407 });
408 };
409 let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
410 .map_err(|error| BindingOriginSourceError::Decode {
411 participant_id,
412 error,
413 })?;
414 if participant_id != identity.participant_id || origin != *expected {
415 return Err(BindingOriginSourceError::Association { participant_id });
416 }
417 }
418 }
419 }
420 if let Some(&(participant_id, _)) = sources.next() {
421 return Err(BindingOriginSourceError::Association { participant_id });
422 }
423 self.restore().map_err(BindingOriginSourceError::Restore)
424 }
425
426 pub fn restore(self) -> Result<RestoredLifecycle, CapsuleRestoreError> {
438 if !self.conversation.is_consistent() {
439 return Err(CapsuleRestoreError::InconsistentConversation);
440 }
441 let lifecycle = self
442 .lifecycle
443 .map(|sections| sections.description.into_state())
444 .transpose()
445 .map_err(CapsuleRestoreError::Lifecycle)?;
446 let sealed = self.sealed.map(restore_sealed).transpose()?;
447 Ok(RestoredLifecycle {
448 covered_log_sequence: self.covered_log_sequence,
449 restored_fenced_sources: BTreeSet::new(),
450 conversation: ParticipantConversation::from_checkpoint(
451 ConversationGenesis::new(self.conversation_id),
452 self.conversation.next_event_ordinal,
453 self.conversation.genesis_validated,
454 ),
455 lifecycle,
456 sealed,
457 })
458 }
459}
460
461fn capture_lifecycle(
464 conversation_id: ConversationId,
465 view: &ConversationLifecycleView<'_>,
466) -> Result<LifecycleSections, CapsuleCaptureError> {
467 if view.frontiers.conversation_id() != conversation_id {
468 return Err(CapsuleCaptureError::ForeignConversation);
469 }
470 for pair in view.participants.windows(2) {
471 if let [earlier, later] = pair
472 && earlier.participant_id() >= later.participant_id()
473 {
474 return Err(CapsuleCaptureError::ParticipantOutOfOrder {
475 participant_id: later.participant_id(),
476 });
477 }
478 }
479 let description = describe_conversation(
480 &view.participants,
481 view.frontiers,
482 view.retained_record_limit,
483 *view.closure,
484 )
485 .map_err(CapsuleCaptureError::Describe)?;
486 let encoded = encode_description(&description).ok_or(CapsuleCaptureError::LengthOverflow)?;
487 Ok(LifecycleSections {
488 description,
489 encoded,
490 })
491}
492
493fn capture_sealed(view: &SealedLifecycleView<'_>) -> Result<SealedSections, CapsuleCaptureError> {
496 for pair in view.participants.windows(2) {
497 if let [earlier, later] = pair
498 && earlier.participant_id() >= later.participant_id()
499 {
500 return Err(CapsuleCaptureError::ParticipantOutOfOrder {
501 participant_id: later.participant_id(),
502 });
503 }
504 }
505 if let Some(live) = view
506 .participants
507 .iter()
508 .find(|participant| matches!(participant, DigestParticipantView::Live { .. }))
509 {
510 return Err(CapsuleCaptureError::LiveParticipantSealed {
511 participant_id: live.participant_id(),
512 });
513 }
514 let participants: Vec<_> = view
515 .participants
516 .iter()
517 .map(ParticipantLifecycleView::storage_description)
518 .collect();
519 let encoded = encode_sealed_description(&participants, view.ledger_high_watermark)
520 .ok_or(CapsuleCaptureError::LengthOverflow)?;
521 Ok(SealedSections {
522 participants,
523 ledger_high_watermark: view.ledger_high_watermark,
524 encoded,
525 })
526}
527
528fn restore_sealed(sections: SealedSections) -> Result<SealedLifecycle, CapsuleRestoreError> {
530 let mut retired = Vec::with_capacity(sections.participants.len());
531 for participant in sections.participants {
532 match participant {
533 ParticipantLifecycleRestore::Retired(identity) => {
534 retired.push(identity.restore().map_err(CapsuleRestoreError::Sealed)?);
535 }
536 ParticipantLifecycleRestore::Live { identity, .. } => {
537 return Err(CapsuleRestoreError::SealedLiveParticipant {
538 participant_id: identity.participant_id,
539 });
540 }
541 }
542 }
543 Ok(SealedLifecycle {
544 retired,
545 ledger_high_watermark: sections.ledger_high_watermark,
546 })
547}
548
549fn decode_lifecycle(
552 input: &[u8],
553 at: &mut usize,
554 start: usize,
555 participants: &[u8],
556 conversation_id: ConversationId,
557) -> Result<LifecycleSections, CapsuleDecodeError> {
558 let frontiers = take_section(input, at, FRONTIERS_SECTION_TAG)?;
559 let closure = take_section(input, at, CLOSURE_SECTION_TAG)?;
560 let description = decode_description::<32, CredentialDigest>(participants, frontiers, closure)?;
561 if description.restore.frontiers.conversation_id != conversation_id {
562 return Err(CapsuleDecodeError::ForeignConversation {
563 tag: FRONTIERS_SECTION_TAG,
564 });
565 }
566 let encoded = input
567 .get(start..*at)
568 .ok_or(CapsuleDecodeError::LengthOverflow)?
569 .to_vec();
570 Ok(LifecycleSections {
571 description,
572 encoded,
573 })
574}
575
576fn take_section<'a>(
578 input: &'a [u8],
579 at: &mut usize,
580 expected: u16,
581) -> Result<&'a [u8], CapsuleDecodeError> {
582 if *at == input.len() {
583 return Err(CapsuleDecodeError::MissingSection { tag: expected });
584 }
585 let tag = take_u16(input, *at)?;
586 let len = take_u32(input, *at + 2)?;
587 let start = *at + SECTION_HEADER_LEN as usize;
588 let end = start
589 .checked_add(usize::try_from(len).map_err(|_| CapsuleDecodeError::LengthOverflow)?)
590 .ok_or(CapsuleDecodeError::LengthOverflow)?;
591 let body = input.get(start..end).ok_or(CapsuleDecodeError::Truncated {
592 required: end,
593 available: input.len(),
594 })?;
595 if tag != expected {
596 return Err(if SECTION_TAGS.contains(&tag) {
597 CapsuleDecodeError::SectionOutOfOrder { tag }
598 } else {
599 CapsuleDecodeError::UnknownSection { tag }
600 });
601 }
602 *at = end;
603 Ok(body)
604}
605
606#[derive(Debug, PartialEq, Eq)]
608pub struct RestoredLifecycle {
609 covered_log_sequence: u64,
610 restored_fenced_sources: BTreeSet<ParticipantId>,
611 conversation: ParticipantConversation,
612 lifecycle: Option<DigestConversationState>,
613 sealed: Option<SealedLifecycle>,
614}
615
616#[derive(Debug, PartialEq, Eq)]
619pub struct SealedLifecycle {
620 retired: Vec<DigestRetiredIdentity>,
621 ledger_high_watermark: u64,
622}
623
624impl SealedLifecycle {
625 #[must_use]
627 pub fn retired(&self) -> &[DigestRetiredIdentity] {
628 &self.retired
629 }
630
631 #[must_use]
633 pub const fn ledger_high_watermark(&self) -> u64 {
634 self.ledger_high_watermark
635 }
636
637 #[must_use]
639 pub fn into_parts(self) -> (Vec<DigestRetiredIdentity>, u64) {
640 (self.retired, self.ledger_high_watermark)
641 }
642}
643
644impl RestoredLifecycle {
645 #[doc(hidden)]
653 pub fn restore_checkpoint_observer_projection(
654 &self,
655 bytes: &[u8],
656 participant_id: ParticipantId,
657 ) -> Result<super::ObserverProgressProjection, super::OwnerCheckpointRestoreError> {
658 let (conversation_id, progress, source) =
659 super::checkpoint_codec::decode_observer_source(bytes)?;
660 if conversation_id != self.conversation.conversation_id()
661 || source.participant_id() != participant_id
662 {
663 return Err(super::OwnerCheckpointRestoreError::Field(
664 "observer_projection.capsule_association",
665 ));
666 }
667 let high_watermark = match (&self.lifecycle, &self.sealed) {
668 (Some(lifecycle), _) => lifecycle.frontiers().sequence().ledger().high_watermark(),
669 (None, Some(sealed)) => sealed.ledger_high_watermark,
670 (None, None) => {
671 return Err(super::OwnerCheckpointRestoreError::Field(
672 "observer_projection.lifecycle",
673 ));
674 }
675 };
676 if u128::from(progress) > u128::from(high_watermark) + 1 {
677 return Err(super::OwnerCheckpointRestoreError::Field(
678 "observer_projection.delivery_coverage",
679 ));
680 }
681 let generation = match (&self.lifecycle, &self.sealed) {
692 (Some(lifecycle), _) => match lifecycle
693 .participants()
694 .iter()
695 .find(|participant| participant.view().participant_id() == participant_id)
696 {
697 Some(super::RestoredParticipantLifecycle::Live { member, .. }) => {
698 Some(member.generation())
699 }
700 Some(super::RestoredParticipantLifecycle::Retired(retired)) => {
701 Some(retired.retired_generation())
702 }
703 None if lifecycle
704 .frontiers()
705 .active_identities()
706 .participants()
707 .iter()
708 .all(|active| active.participant_index() != participant_id) =>
709 {
710 None
711 }
712 None => {
713 return Err(super::OwnerCheckpointRestoreError::Field(
714 "observer_projection.participant",
715 ));
716 }
717 },
718 (None, Some(sealed)) => sealed
719 .retired
720 .iter()
721 .find(|retired| retired.participant_id() == participant_id)
722 .map(super::RetiredIdentity::retired_generation),
723 (None, None) => {
724 return Err(super::OwnerCheckpointRestoreError::Field(
725 "observer_projection.lifecycle",
726 ));
727 }
728 };
729 if let Some(generation) = generation
730 && source
731 .binding_epoch()
732 .is_some_and(|epoch| epoch.capability_generation.get() > generation.get())
733 {
734 return Err(super::OwnerCheckpointRestoreError::Field(
735 "observer_projection.binding_generation",
736 ));
737 }
738 Ok(super::ObserverProgressProjection::sourced(
739 conversation_id,
740 progress,
741 source,
742 ))
743 }
744
745 #[doc(hidden)]
756 pub fn restore_checkpoint_fenced_attach(
757 &mut self,
758 source: super::FencedAttachCommitRestore,
759 marker_record: &[u8],
760 ) -> Result<super::FencedAttachCommit, super::OwnerCheckpointRestoreError> {
761 use super::OwnerCheckpointRestoreError as Error;
762 if self
763 .restored_fenced_sources
764 .contains(&source.participant_id)
765 {
766 return Err(Error::Field("fenced_source.duplicate"));
767 }
768 let lifecycle = self
769 .lifecycle
770 .as_ref()
771 .ok_or(Error::Field("fenced_source.lifecycle"))?;
772 let origin = lifecycle
773 .participants()
774 .iter()
775 .find_map(|participant| match participant {
776 super::RestoredParticipantLifecycle::Live {
777 member,
778 binding_origin: Some(origin),
779 ..
780 } if member.participant_id() == source.participant_id => Some(*origin),
781 _ => None,
782 })
783 .ok_or(Error::Field("fenced_source.binding_origin"))?;
784 if origin.binding_epoch() != source.new_binding_epoch
785 || origin.recovered_marker()
786 != Some((source.marker_delivery_seq, source.prior_binding_epoch))
787 {
788 return Err(Error::Field("fenced_source.origin_association"));
789 }
790 let record = lifecycle
791 .frontiers()
792 .checkpoint_recovered_marker_record(origin)
793 .ok_or(Error::Field("fenced_source.historical_marker"))?;
794 if super::checkpoint_codec::encode_marker_source(&record) != marker_record {
795 return Err(Error::Field("fenced_source.marker_bytes"));
796 }
797 let restored = source
798 .restore_with_record(&record)
799 .map_err(Error::Provenance);
800 record.consume();
801 let restored = restored?;
802 self.restored_fenced_sources.insert(source.participant_id);
803 Ok(restored)
804 }
805
806 #[must_use]
809 pub const fn covered_log_sequence(&self) -> u64 {
810 self.covered_log_sequence
811 }
812
813 #[must_use]
815 pub fn into_parts(self) -> RestoredLifecycleParts {
816 RestoredLifecycleParts {
817 conversation: self.conversation,
818 lifecycle: self.lifecycle,
819 sealed: self.sealed,
820 }
821 }
822
823 #[must_use]
826 pub const fn sealed(&self) -> Option<&SealedLifecycle> {
827 self.sealed.as_ref()
828 }
829}
830
831#[derive(Debug)]
835#[non_exhaustive]
836pub struct RestoredLifecycleParts {
837 pub conversation: ParticipantConversation,
839 pub lifecycle: Option<DigestConversationState>,
843 pub sealed: Option<SealedLifecycle>,
846}
847
848#[derive(Clone, Copy, Debug, PartialEq, Eq)]
850#[non_exhaustive]
851pub enum CapsuleCaptureError {
852 ForeignConversation,
854 ParticipantOutOfOrder {
856 participant_id: ParticipantId,
858 },
859 Describe(ConversationDescribeError),
861 LengthOverflow,
863 SealedBesideLifecycle,
866 LiveParticipantSealed {
868 participant_id: ParticipantId,
870 },
871}
872
873#[derive(Clone, Copy, Debug, PartialEq, Eq)]
875#[non_exhaustive]
876pub enum CapsuleDecodeError {
877 Truncated {
879 required: usize,
881 available: usize,
883 },
884 InvalidMagic,
886 UnsupportedCodec {
888 major: u16,
890 minor: u16,
892 },
893 NonCanonicalLength {
896 declared_body_len: u32,
898 actual_body_len: usize,
900 },
901 MissingSection {
903 tag: u16,
905 },
906 SectionOutOfOrder {
908 tag: u16,
910 },
911 UnknownSection {
913 tag: u16,
915 },
916 NonCanonicalSectionLength {
918 tag: u16,
920 declared: u32,
922 },
923 NonCanonicalSection {
925 tag: u16,
927 },
928 ParticipantOutOfOrder {
930 participant_id: ParticipantId,
932 },
933 ForeignConversation {
935 tag: u16,
937 },
938 TrailingBytes {
940 trailing: usize,
942 },
943 LengthOverflow,
945}
946
947#[derive(Clone, Debug, PartialEq, Eq)]
949#[non_exhaustive]
950pub enum CapsuleRestoreError {
951 InconsistentConversation,
953 Lifecycle(ConversationStateRestoreError),
956 Sealed(StorageRestoreError),
958 SealedLiveParticipant {
960 participant_id: ParticipantId,
962 },
963}
964
965fn decode_conversation(body: &[u8]) -> Result<ConversationSection, CapsuleDecodeError> {
966 if body.len() != CONVERSATION_SECTION_LEN as usize {
967 return Err(CapsuleDecodeError::NonCanonicalSectionLength {
968 tag: CONVERSATION_SECTION_TAG,
969 declared: u32::try_from(body.len()).map_err(|_| CapsuleDecodeError::LengthOverflow)?,
970 });
971 }
972 let next_event_ordinal = take_u64(body, 0)?;
973 let flags = body.get(8).copied().ok_or(CapsuleDecodeError::Truncated {
974 required: 9,
975 available: body.len(),
976 })?;
977 if flags & !GENESIS_VALIDATED_FLAG != 0 {
978 return Err(CapsuleDecodeError::NonCanonicalSection {
979 tag: CONVERSATION_SECTION_TAG,
980 });
981 }
982 let section = ConversationSection {
983 next_event_ordinal,
984 genesis_validated: flags & GENESIS_VALIDATED_FLAG != 0,
985 };
986 if !section.is_consistent() {
987 return Err(CapsuleDecodeError::NonCanonicalSection {
988 tag: CONVERSATION_SECTION_TAG,
989 });
990 }
991 Ok(section)
992}
993
994fn take<const N: usize>(input: &[u8], start: usize) -> Result<[u8; N], CapsuleDecodeError> {
995 let end = start
996 .checked_add(N)
997 .ok_or(CapsuleDecodeError::LengthOverflow)?;
998 input
999 .get(start..end)
1000 .ok_or(CapsuleDecodeError::Truncated {
1001 required: end,
1002 available: input.len(),
1003 })?
1004 .try_into()
1005 .map_err(|_| CapsuleDecodeError::LengthOverflow)
1006}
1007
1008fn take_u16(input: &[u8], start: usize) -> Result<u16, CapsuleDecodeError> {
1009 take(input, start).map(u16::from_be_bytes)
1010}
1011
1012fn take_u32(input: &[u8], start: usize) -> Result<u32, CapsuleDecodeError> {
1013 take(input, start).map(u32::from_be_bytes)
1014}
1015
1016fn take_u64(input: &[u8], start: usize) -> Result<u64, CapsuleDecodeError> {
1017 take(input, start).map(u64::from_be_bytes)
1018}
1019
1020#[doc(hidden)]
1022#[derive(Clone, Debug, PartialEq, Eq)]
1023pub enum BindingOriginSourceError {
1024 Decode {
1026 participant_id: ParticipantId,
1028 error: CapsuleDecodeError,
1030 },
1031 Association {
1033 participant_id: ParticipantId,
1035 },
1036 Restore(CapsuleRestoreError),
1038}
1039
1040impl RestoredLifecycle {
1041 pub fn restore_cursor_episode(
1047 self,
1048 bytes: &[u8],
1049 observer_progress: u64,
1050 ) -> Result<RestoredLifecycleWithCursorEpisode, super::StorageRestoreError> {
1051 let lifecycle = self
1052 .lifecycle
1053 .as_ref()
1054 .ok_or(super::StorageRestoreError::CursorEpisode)?;
1055 let ClosureState::Owed { debt, .. } = lifecycle.closure() else {
1056 return Err(super::StorageRestoreError::CursorEpisode);
1057 };
1058 let episode = super::NonzeroDebtCursorEpisode::decode_checkpoint(
1059 bytes,
1060 lifecycle.frontiers(),
1061 debt,
1062 observer_progress,
1063 )?;
1064 Ok(RestoredLifecycleWithCursorEpisode {
1065 lifecycle: self,
1066 episode,
1067 })
1068 }
1069}
1070
1071#[derive(Debug)]
1073pub struct RestoredLifecycleWithCursorEpisode {
1074 lifecycle: RestoredLifecycle,
1075 episode: super::NonzeroDebtCursorEpisode,
1076}
1077impl RestoredLifecycleWithCursorEpisode {
1078 #[must_use]
1080 pub fn into_parts(self) -> (RestoredLifecycleParts, super::NonzeroDebtCursorEpisode) {
1081 (self.lifecycle.into_parts(), self.episode)
1082 }
1083}
1084
1085impl LifecycleCapsule {
1086 pub fn capture_with_binding_sources(
1094 view: &LifecycleView<'_>,
1095 sources: &[(ParticipantId, &[u8])],
1096 ) -> Result<Self, BindingOriginCaptureError> {
1097 let Some(lifecycle) = view.lifecycle.as_ref() else {
1098 if let Some(&(participant_id, _)) = sources.first() {
1099 return Err(BindingOriginCaptureError::Association { participant_id });
1100 }
1101 return Self::capture(view).map_err(BindingOriginCaptureError::Capture);
1102 };
1103 let mut remaining = sources.iter().peekable();
1104 let mut origins = Vec::new();
1105 for participant in &lifecycle.participants {
1106 match participant {
1107 DigestParticipantView::Retired(_) => origins.push(None),
1108 DigestParticipantView::Live {
1109 member,
1110 binding,
1111 binding_origin,
1112 ..
1113 } => {
1114 let participant_id = member.participant_id();
1115 let origin_required = !matches!(binding, super::BindingState::Detached)
1116 || member.latest_terminal().is_some()
1117 || binding_origin.is_some();
1118 let Some(&&(source_id, bytes)) = remaining.peek() else {
1119 if origin_required {
1120 return Err(BindingOriginCaptureError::Association { participant_id });
1121 }
1122 origins.push(None);
1123 continue;
1124 };
1125 if source_id != participant_id {
1126 if origin_required || source_id < participant_id {
1127 return Err(BindingOriginCaptureError::Association { participant_id });
1128 }
1129 origins.push(None);
1130 continue;
1131 }
1132 remaining.next();
1133 let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
1134 .map_err(|error| BindingOriginCaptureError::Decode {
1135 participant_id,
1136 error,
1137 })?;
1138 super::storage::validate_binding_origin(origin, member, **binding).map_err(
1139 |error| BindingOriginCaptureError::Validation {
1140 participant_id,
1141 error,
1142 detail: origin_disagreement(&**binding, &origin, member),
1145 },
1146 )?;
1147 if binding_origin.is_some_and(|expected| *expected != origin) {
1148 return Err(BindingOriginCaptureError::Association { participant_id });
1149 }
1150 origins.push(Some(origin));
1151 }
1152 }
1153 }
1154 if let Some(&(participant_id, _)) = remaining.next() {
1155 return Err(BindingOriginCaptureError::Association { participant_id });
1156 }
1157 let participants = lifecycle
1158 .participants
1159 .iter()
1160 .zip(&origins)
1161 .map(|(participant, origin)| match participant {
1162 DigestParticipantView::Live {
1163 member,
1164 binding,
1165 detach_cell,
1166 ..
1167 } => DigestParticipantView::Live {
1168 member,
1169 binding,
1170 detach_cell,
1171 binding_origin: origin.as_ref(),
1172 },
1173 DigestParticipantView::Retired(retired) => DigestParticipantView::Retired(retired),
1174 })
1175 .collect();
1176 let capsule = Self::capture(&LifecycleView {
1177 sealed: None,
1178 conversation: view.conversation,
1179 covered_log_sequence: view.covered_log_sequence,
1180 lifecycle: Some(ConversationLifecycleView {
1181 participants,
1182 frontiers: lifecycle.frontiers,
1183 retained_record_limit: lifecycle.retained_record_limit,
1184 closure: lifecycle.closure,
1185 }),
1186 })
1187 .map_err(BindingOriginCaptureError::Capture)?;
1188 let canonical = capsule.encode_canonical();
1189 capsule
1190 .restore()
1191 .map_err(BindingOriginCaptureError::Restore)?;
1192 Self::decode_canonical(&canonical).map_err(BindingOriginCaptureError::Canonical)
1193 }
1194}
1195
1196#[derive(Clone, Debug, PartialEq, Eq)]
1198pub enum BindingOriginCaptureError {
1199 Restore(CapsuleRestoreError),
1201 Canonical(CapsuleDecodeError),
1203 Decode {
1205 participant_id: ParticipantId,
1207 error: CapsuleDecodeError,
1209 },
1210 Association {
1212 participant_id: ParticipantId,
1214 },
1215 Validation {
1217 participant_id: ParticipantId,
1219 error: super::StorageRestoreError,
1221 detail: String,
1224 },
1225 Capture(CapsuleCaptureError),
1227}
1228
1229fn origin_disagreement(
1232 binding: &impl core::fmt::Debug,
1233 origin: &super::BindingOrigin,
1234 member: &super::LiveMember<CredentialDigest>,
1235) -> String {
1236 format!(
1237 "binding {binding:?}; origin for participant {} conversation {} epoch {:?}; \
1238 member conversation {} generation {:?}; latest terminal epoch {:?}",
1239 origin.participant_id(),
1240 origin.conversation_id(),
1241 origin.binding_epoch(),
1242 member.conversation_id(),
1243 member.generation(),
1244 member
1245 .latest_terminal()
1246 .map(super::CommittedBindingTerminal::binding_epoch),
1247 )
1248}