1use alloc::vec::Vec;
9
10use crate::algebra::WideResourceVector;
11use crate::outcome::ParticipantStateCorruptReason;
12use crate::wire::{
13 AttachSecret, BindingEpoch, CloseCause, ConversationId, DeliverySeq, DetachAttemptToken,
14 Generation, LeaveAttemptToken, LeaveCommitted, ObserverEpoch, ParticipantId, TransactionOrder,
15};
16
17use super::{
18 ActiveBinding, AdmissionOrder, BindingOrigin, BindingState, BoundParticipantCursor,
19 ClaimFrontiers, ClaimFrontiersRestore, ClosureDebt, ClosureState, CommittedBindingTerminal,
20 DebtCompletion, DetachCell, DetachedCredentialRecovery, DetachedMarkerRelease,
21 DetachedMarkerReleaseProvenance, EmptyDetach, EnrollmentFingerprint, Event, FencedAttachCommit,
22 FrontierBinding, IdentityState, LeaveFingerprint, LiveMember, LiveMemberRestore,
23 MarkerDelivery, NonzeroDebtCursorEpisode, ObserverProjection, OrderLedger,
24 OrdinaryBindingAuthority, OrdinaryBindingFate, ParticipantCursorProgress, PendingFinalization,
25 PendingRecoveredCursorRelease, PhysicalCompaction, RecoveredBindingFate,
26 RecoveredBindingFateTransition, RetiredIdentity, SequenceLedger, StoredEdge,
27 binding::{restore_committed_terminal, restore_pending_finalization},
28 claim_frontier::{
29 ClaimFrontiersDescription, HistoricalCausalAuthority, MarkerRecordOccurrence,
30 MarkerRecordRequest, ValidatedConversationHistory, ValidatedMarkerRecord,
31 },
32 cursor_facts::{CursorProgressFact, CursorProgressKey},
33 detach::{
34 restore_committed_detach, restore_pending_detach, restore_terminalized_detach,
35 validate_pending_pair,
36 },
37};
38
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum StorageRestoreError {
42 CommittedBindingTerminal,
44 PendingFinalization,
46 BindingAuthority,
48 MembershipInvariant,
50 LeaveResult,
52 RetiredIdentity,
54 DetachCell,
56 DetachBindingPair,
58 CursorEpisode,
60 ClosureDebt,
62 StoredEdgeProvenance,
64 FencedMarkerLatch,
70}
71
72#[derive(Clone, Debug, PartialEq, Eq)]
74pub enum ConversationStateRestoreError {
75 ClaimFrontier(ParticipantStateCorruptReason),
77 Storage(StorageRestoreError),
79}
80
81#[derive(Debug, PartialEq, Eq)]
88pub(super) struct RestoredConversationState {
89 frontiers: ClaimFrontiers,
90 closure: ClosureState,
91}
92
93#[derive(Clone, Debug, PartialEq, Eq)]
107pub struct ParticipantConversationRestore<EF, V, LF, D> {
108 pub participants: Vec<ParticipantLifecycleRestore<EF, V, LF, D>>,
110 pub frontiers: ClaimFrontiersRestore,
112 pub sequence_ledger: SequenceLedger,
114 pub order_ledger: OrderLedger,
116 pub closure: ClosureStateRestore,
118}
119
120#[derive(Debug, PartialEq, Eq)]
139pub struct ParticipantConversationState<EF, V, LF, D> {
140 participants: Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
141 frontiers: ClaimFrontiers,
142 closure: ClosureState,
143}
144
145impl<EF, V, LF, D> ParticipantConversationState<EF, V, LF, D> {
146 #[must_use]
148 pub fn participants(&self) -> &[RestoredParticipantLifecycle<EF, V, LF, D>] {
149 &self.participants
150 }
151
152 #[must_use]
154 pub const fn frontiers(&self) -> &ClaimFrontiers {
155 &self.frontiers
156 }
157
158 #[must_use]
160 pub const fn closure(&self) -> ClosureState {
161 self.closure
162 }
163
164 #[must_use]
171 #[allow(clippy::type_complexity)]
172 pub fn into_parts(
173 self,
174 ) -> (
175 Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
176 ClaimFrontiers,
177 ClosureState,
178 ) {
179 (self.participants, self.frontiers, self.closure)
180 }
181}
182
183impl<EF: Clone, V: Clone, LF: Clone, D: Clone> ParticipantConversationState<EF, V, LF, D> {
184 #[cfg(test)]
192 pub(super) fn describe(
193 &self,
194 retained_record_limit: u64,
195 ) -> Result<ParticipantConversationDescription<EF, V, LF, D>, ConversationDescribeError> {
196 let views: Vec<_> = self
197 .participants
198 .iter()
199 .map(RestoredParticipantLifecycle::view)
200 .collect();
201 describe_conversation(&views, &self.frontiers, retained_record_limit, self.closure)
202 }
203}
204
205impl RestoredConversationState {
206 #[must_use]
208 #[cfg(test)]
209 pub(crate) const fn frontiers(&self) -> &ClaimFrontiers {
210 &self.frontiers
211 }
212
213 #[must_use]
215 #[cfg(test)]
216 pub(crate) const fn closure(&self) -> ClosureState {
217 self.closure
218 }
219
220 #[must_use]
222 pub(crate) fn into_parts(self) -> (ClaimFrontiers, ClosureState) {
223 (self.frontiers, self.closure)
224 }
225}
226
227#[cfg(test)]
245pub(super) fn restore_conversation_state(
246 frontier_restore: ClaimFrontiersRestore,
247 sequence_ledger: SequenceLedger,
248 order_ledger: OrderLedger,
249 closure_restore: &ClosureStateRestore,
250) -> Result<RestoredConversationState, ConversationStateRestoreError> {
251 let history = ValidatedConversationHistory::empty();
252 restore_conversation_with_history(
253 ClaimFrontiersDescription {
254 restore: frontier_restore,
255 sequence_ledger,
256 order_ledger,
257 fenced_marker_issued: None,
258 },
259 closure_restore,
260 &history,
261 )
262}
263
264fn restore_conversation_with_history(
272 frontier_description: ClaimFrontiersDescription,
273 closure_restore: &ClosureStateRestore,
274 history: &ValidatedConversationHistory,
275) -> Result<RestoredConversationState, ConversationStateRestoreError> {
276 let mut prevalidated = ClaimFrontiers::prevalidate_description(frontier_description, history)
277 .map_err(ConversationStateRestoreError::ClaimFrontier)?;
278 let marker_request = closure_restore.marker_record_request();
279 let ordinary_request = closure_restore.ordinary_binding_request();
280 let closure = match (marker_request, ordinary_request) {
281 (Some(marker_request), None) => {
282 let record = prevalidated.take_marker_record(marker_request).ok_or(
283 ConversationStateRestoreError::Storage(StorageRestoreError::StoredEdgeProvenance),
284 )?;
285 (*closure_restore)
286 .restore_with_marker_record(prevalidated.conversation_id(), record)
287 .map_err(ConversationStateRestoreError::Storage)?
288 }
289 (None, Some(binding)) => {
290 let origin = history
291 .ordinary_origin(
292 binding.conversation_id,
293 binding.participant_id,
294 binding.binding_epoch,
295 )
296 .ok_or(ConversationStateRestoreError::Storage(
297 StorageRestoreError::StoredEdgeProvenance,
298 ))?;
299 (*closure_restore)
300 .restore_with_binding_origin(origin)
301 .map_err(ConversationStateRestoreError::Storage)?
302 }
303 (None, None) => (*closure_restore)
304 .restore()
305 .map_err(ConversationStateRestoreError::Storage)?,
306 (Some(_), Some(_)) => {
307 return Err(ConversationStateRestoreError::Storage(
308 StorageRestoreError::StoredEdgeProvenance,
309 ));
310 }
311 };
312 let current_edge = match closure {
313 ClosureState::Clear => None,
314 ClosureState::Owed { edge, .. } => Some(edge),
315 };
316 let mut frontiers = prevalidated
317 .finish(current_edge)
318 .map_err(ConversationStateRestoreError::ClaimFrontier)?;
319 if !frontiers.fenced_latch_matches(current_edge) {
320 return Err(ConversationStateRestoreError::Storage(
321 StorageRestoreError::FencedMarkerLatch,
322 ));
323 }
324 if let ClosureStateRestore::Owed {
325 edge:
326 StoredEdgeRestore::DetachedCursorRelease {
327 provenance: DetachedCursorReleaseProvenanceRestore::Ordinary(source),
328 ..
329 },
330 ..
331 } = closure_restore
332 {
333 frontiers.retain_ordinary_cursor_release(alloc::boxed::Box::new(*source));
334 }
335 Ok(RestoredConversationState { frontiers, closure })
336}
337
338#[derive(Clone, Copy, Debug, PartialEq, Eq)]
340pub struct CommittedBindingTerminalRestore {
341 pub binding: ActiveBinding,
343 pub cause: CloseCause,
345 pub transaction_order: TransactionOrder,
347 pub delivery_seq: DeliverySeq,
349}
350
351impl CommittedBindingTerminalRestore {
352 pub fn restore(self) -> Result<CommittedBindingTerminal, StorageRestoreError> {
359 restore_committed_terminal(
360 self.binding,
361 self.cause,
362 self.transaction_order,
363 self.delivery_seq,
364 )
365 .ok_or(StorageRestoreError::CommittedBindingTerminal)
366 }
367}
368
369#[derive(Clone, Copy, Debug, PartialEq, Eq)]
371pub struct PendingFinalizationRestore {
372 pub binding: ActiveBinding,
374 pub cause: CloseCause,
376 pub transaction_order: TransactionOrder,
378}
379
380impl PendingFinalizationRestore {
381 pub fn restore(self) -> Result<PendingFinalization, StorageRestoreError> {
388 restore_pending_finalization(self.binding, self.cause, self.transaction_order)
389 .ok_or(StorageRestoreError::PendingFinalization)
390 }
391}
392
393#[derive(Clone, Copy, Debug, PartialEq, Eq)]
395pub enum BindingFateTerminalRestore {
396 Committed(CommittedBindingTerminalRestore),
398 Pending(PendingFinalizationRestore),
400}
401
402#[derive(Clone, Copy, Debug, PartialEq, Eq)]
404pub enum RestoredBindingFateTerminal {
405 Committed(CommittedBindingTerminal),
407 Pending(PendingFinalization),
409}
410
411impl BindingFateTerminalRestore {
412 pub fn restore(self) -> Result<RestoredBindingFateTerminal, StorageRestoreError> {
418 match self {
419 Self::Committed(value) => value.restore().map(RestoredBindingFateTerminal::Committed),
420 Self::Pending(value) => value.restore().map(RestoredBindingFateTerminal::Pending),
421 }
422 }
423}
424
425impl RestoredBindingFateTerminal {
426 #[cfg(any(test, feature = "test-support"))]
449 #[allow(clippy::panic, clippy::expect_used)]
450 #[must_use]
451 pub fn committed_for_marker_progress(
452 progress: super::edge::ParticipantCursorProgress,
453 transaction_order: TransactionOrder,
454 delivery_seq: DeliverySeq,
455 ) -> Self {
456 let super::edge::ParticipantCursorProgress::Marker(marker) = progress else {
457 panic!("a continuous cursor witness names no conversation to build a terminal for")
458 };
459 Self::committed_for_test(
460 marker.conversation_id(),
461 marker.participant_id(),
462 marker.binding_epoch(),
463 transaction_order,
464 delivery_seq,
465 )
466 }
467
468 #[cfg(any(test, feature = "test-support"))]
474 #[must_use]
475 pub fn committed_for_marker_delivery(
476 delivery: super::edge::MarkerDelivery,
477 transaction_order: TransactionOrder,
478 delivery_seq: DeliverySeq,
479 ) -> Self {
480 Self::committed_for_test(
481 delivery.conversation_id(),
482 delivery.participant_id(),
483 delivery.binding_epoch(),
484 transaction_order,
485 delivery_seq,
486 )
487 }
488
489 #[cfg(any(test, feature = "test-support"))]
499 #[allow(clippy::expect_used)]
500 #[must_use]
501 pub fn committed_for_test(
502 conversation_id: ConversationId,
503 participant_id: ParticipantId,
504 binding_epoch: BindingEpoch,
505 transaction_order: TransactionOrder,
506 delivery_seq: DeliverySeq,
507 ) -> Self {
508 BindingFateTerminalRestore::Committed(CommittedBindingTerminalRestore {
509 binding: ActiveBinding {
510 participant_id,
511 conversation_id,
512 binding_epoch,
513 },
514 cause: CloseCause::ConnectionLost,
515 transaction_order,
516 delivery_seq,
517 })
518 .restore()
519 .expect("test fate terminal fields form a valid committed terminal")
520 }
521
522 #[must_use]
524 pub const fn storage_description(self) -> BindingFateTerminalRestore {
525 match self {
526 Self::Committed(value) => {
527 BindingFateTerminalRestore::Committed(value.storage_description())
528 }
529 Self::Pending(value) => {
530 BindingFateTerminalRestore::Pending(value.storage_description())
531 }
532 }
533 }
534
535 pub(super) const fn participant_id(self) -> ParticipantId {
536 match self {
537 Self::Committed(value) => value.participant_id(),
538 Self::Pending(value) => value.participant_id(),
539 }
540 }
541
542 pub(super) const fn conversation_id(self) -> ConversationId {
543 match self {
544 Self::Committed(value) => value.conversation_id(),
545 Self::Pending(value) => value.conversation_id(),
546 }
547 }
548
549 pub(super) const fn binding_epoch(self) -> BindingEpoch {
550 match self {
551 Self::Committed(value) => value.binding_epoch(),
552 Self::Pending(value) => value.binding_epoch(),
553 }
554 }
555}
556
557#[derive(Clone, Copy, Debug, PartialEq, Eq)]
559pub enum BindingStateRestore {
560 Detached,
562 Bound(ActiveBinding),
564 PendingFinalization(PendingFinalizationRestore),
566}
567
568impl BindingStateRestore {
569 fn restore_for<EF>(self, member: &LiveMember<EF>) -> Result<BindingState, StorageRestoreError> {
570 let state = match self {
571 Self::Detached => BindingState::Detached,
572 Self::Bound(binding) => BindingState::Bound(binding),
573 Self::PendingFinalization(raw) => BindingState::PendingFinalization(raw.restore()?),
574 };
575 let authority_matches = match state {
576 BindingState::Detached => true,
577 BindingState::Bound(binding) => {
578 binding.participant_id == member.participant_id()
579 && binding.conversation_id == member.conversation_id()
580 && binding.binding_epoch.capability_generation == member.generation()
581 }
582 BindingState::PendingFinalization(pending) => {
583 pending.participant_id() == member.participant_id()
584 && pending.conversation_id() == member.conversation_id()
585 && pending.binding_epoch().capability_generation == member.generation()
586 }
587 };
588 if authority_matches {
589 Ok(state)
590 } else {
591 Err(StorageRestoreError::BindingAuthority)
592 }
593 }
594}
595
596#[derive(Clone, Debug, PartialEq, Eq)]
598pub struct LiveIdentityRestore<EF> {
599 pub participant_id: ParticipantId,
601 pub conversation_id: ConversationId,
603 pub generation: Generation,
605 pub attach_secret: AttachSecret,
607 pub cursor: DeliverySeq,
609 pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
611 pub latest_terminal: Option<CommittedBindingTerminalRestore>,
613}
614
615impl<EF> LiveIdentityRestore<EF> {
616 fn restore(self) -> Result<LiveMember<EF>, StorageRestoreError> {
617 let latest_terminal = self
618 .latest_terminal
619 .map(CommittedBindingTerminalRestore::restore)
620 .transpose()?;
621 LiveMember::restore(LiveMemberRestore {
622 participant_id: self.participant_id,
623 conversation_id: self.conversation_id,
624 generation: self.generation,
625 attach_secret: self.attach_secret,
626 cursor: self.cursor,
627 enrollment_fingerprint: self.enrollment_fingerprint,
628 latest_terminal,
629 })
630 .map_err(|_| StorageRestoreError::MembershipInvariant)
631 }
632}
633
634#[derive(Clone, Copy, Debug, PartialEq, Eq)]
636pub struct LeaveCommittedRestore {
637 pub conversation_id: ConversationId,
639 pub leave_attempt_token: LeaveAttemptToken,
641 pub participant_id: ParticipantId,
643 pub retired_generation: Generation,
645 pub ended_binding_epoch: Option<BindingEpoch>,
647 pub prior_terminal_delivery_seq: Option<DeliverySeq>,
649 pub left_delivery_seq: DeliverySeq,
651}
652
653impl LeaveCommittedRestore {
654 pub fn restore(self) -> Result<LeaveCommitted, StorageRestoreError> {
661 LeaveCommitted::new(
662 self.conversation_id,
663 self.leave_attempt_token,
664 self.participant_id,
665 self.retired_generation,
666 self.ended_binding_epoch,
667 self.prior_terminal_delivery_seq,
668 self.left_delivery_seq,
669 )
670 .ok_or(StorageRestoreError::LeaveResult)
671 }
672}
673
674#[derive(Clone, Debug, PartialEq, Eq)]
676pub struct RetiredIdentityRestore<EF, V, LF> {
677 pub participant_id: ParticipantId,
679 pub conversation_id: ConversationId,
681 pub retired_generation: Generation,
683 pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
685 pub leave_attempt_token: LeaveAttemptToken,
687 pub leave_request_verifier: V,
689 pub leave_fingerprint: LeaveFingerprint<LF>,
691 pub left_transaction_order: TransactionOrder,
693 pub committed_result: LeaveCommittedRestore,
695}
696
697impl<EF, V, LF> RetiredIdentityRestore<EF, V, LF> {
698 pub(super) fn restore_result(&self) -> Result<LeaveCommitted, StorageRestoreError> {
699 let result = self.committed_result.restore()?;
700 if result.conversation_id() != self.conversation_id
701 || result.participant_id() != self.participant_id
702 || result.presented_generation() != self.retired_generation
703 || result.retired_generation() != self.retired_generation
704 || result.leave_attempt_token() != self.leave_attempt_token
705 {
706 return Err(StorageRestoreError::RetiredIdentity);
707 }
708 Ok(result)
709 }
710
711 pub fn restore(self) -> Result<RetiredIdentity<EF, V, LF>, StorageRestoreError> {
719 let result = self.restore_result()?;
720 RetiredIdentity::restore(
721 self.participant_id,
722 self.conversation_id,
723 self.retired_generation,
724 self.enrollment_fingerprint,
725 self.leave_attempt_token,
726 self.leave_request_verifier,
727 self.leave_fingerprint,
728 self.left_transaction_order,
729 result,
730 )
731 .map_err(|_| StorageRestoreError::RetiredIdentity)
732 }
733}
734
735#[derive(Clone, Debug, PartialEq, Eq)]
737pub enum DetachCellRestore<V> {
738 Empty,
740 Pending {
742 token: DetachAttemptToken,
744 participant_id: ParticipantId,
746 request_generation: Generation,
748 request_verifier: V,
750 committed_binding_epoch: BindingEpoch,
752 admission_order: AdmissionOrder,
754 refused_epoch: ObserverEpoch,
756 },
757 Committed {
759 token: DetachAttemptToken,
761 participant_id: ParticipantId,
763 request_generation: Generation,
765 request_verifier: V,
767 committed_binding_epoch: BindingEpoch,
769 detached_delivery_seq: DeliverySeq,
771 },
772 Terminalized {
774 token: DetachAttemptToken,
776 participant_id: ParticipantId,
778 request_generation: Generation,
780 request_verifier: V,
782 committed_binding_epoch: BindingEpoch,
784 },
785}
786
787impl<V> DetachCellRestore<V> {
788 fn restore(self) -> Result<DetachCell<V>, StorageRestoreError> {
789 match self {
790 Self::Empty => Ok(DetachCell::Empty(EmptyDetach)),
791 Self::Pending {
792 token,
793 participant_id,
794 request_generation,
795 request_verifier,
796 committed_binding_epoch,
797 admission_order,
798 refused_epoch,
799 } => restore_pending_detach(
800 token,
801 participant_id,
802 request_generation,
803 request_verifier,
804 committed_binding_epoch,
805 admission_order,
806 refused_epoch,
807 )
808 .map(DetachCell::Pending)
809 .ok_or(StorageRestoreError::DetachCell),
810 Self::Committed {
811 token,
812 participant_id,
813 request_generation,
814 request_verifier,
815 committed_binding_epoch,
816 detached_delivery_seq,
817 } => restore_committed_detach(
818 token,
819 participant_id,
820 request_generation,
821 request_verifier,
822 committed_binding_epoch,
823 detached_delivery_seq,
824 )
825 .map(DetachCell::Committed)
826 .ok_or(StorageRestoreError::DetachCell),
827 Self::Terminalized {
828 token,
829 participant_id,
830 request_generation,
831 request_verifier,
832 committed_binding_epoch,
833 } => restore_terminalized_detach(
834 token,
835 participant_id,
836 request_generation,
837 request_verifier,
838 committed_binding_epoch,
839 )
840 .map(DetachCell::Terminalized)
841 .ok_or(StorageRestoreError::DetachCell),
842 }
843 }
844}
845
846#[allow(
848 clippy::large_enum_variant,
849 reason = "the live storage capsule remains inline so its atomic slots cannot be restored separately"
850)]
851#[derive(Clone, Debug, PartialEq, Eq)]
852pub enum ParticipantLifecycleRestore<EF, V, LF, D> {
853 Live {
855 identity: LiveIdentityRestore<EF>,
857 binding: BindingStateRestore,
859 binding_origin: Option<BindingOrigin>,
861 detach_cell: DetachCellRestore<D>,
863 },
864 Retired(RetiredIdentityRestore<EF, V, LF>),
866}
867
868#[allow(
870 clippy::large_enum_variant,
871 reason = "the validated live capsule remains inline as one atomic lifecycle result"
872)]
873#[derive(Clone, Debug, PartialEq, Eq)]
874pub enum RestoredParticipantLifecycle<EF, V, LF, D> {
875 Live {
877 member: LiveMember<EF>,
879 binding: BindingState,
881 binding_origin: Option<BindingOrigin>,
883 detach_cell: DetachCell<D>,
885 },
886 Retired(RetiredIdentity<EF, V, LF>),
888}
889
890impl<EF, V, LF, D> ParticipantLifecycleRestore<EF, V, LF, D> {
891 pub fn restore(
898 self,
899 ) -> Result<RestoredParticipantLifecycle<EF, V, LF, D>, StorageRestoreError> {
900 match self {
901 Self::Retired(identity) => identity
902 .restore()
903 .map(RestoredParticipantLifecycle::Retired),
904 Self::Live {
905 identity,
906 binding,
907 binding_origin,
908 detach_cell,
909 } => {
910 let member = identity.restore()?;
911 let binding = binding.restore_for(&member)?;
912 let binding_origin = binding_origin
913 .map(|origin| validate_binding_origin(origin, &member, binding))
914 .transpose()?;
915 let origin_required = !matches!(binding, BindingState::Detached)
916 || member.latest_terminal().is_some();
917 if origin_required != binding_origin.is_some() {
918 return Err(StorageRestoreError::BindingAuthority);
919 }
920 let detach_cell = detach_cell.restore()?;
921 validate_live_pair(&member, binding, &detach_cell)?;
922 Ok(RestoredParticipantLifecycle::Live {
923 member,
924 binding,
925 binding_origin,
926 detach_cell,
927 })
928 }
929 }
930 }
931}
932
933pub(super) fn validate_binding_origin<EF>(
934 origin: BindingOrigin,
935 member: &LiveMember<EF>,
936 binding_state: BindingState,
937) -> Result<BindingOrigin, StorageRestoreError> {
938 let expected_epoch = match binding_state {
939 BindingState::Bound(current) => Some(current.binding_epoch),
940 BindingState::PendingFinalization(pending) => Some(pending.binding_epoch()),
941 BindingState::Detached => member
942 .latest_terminal()
943 .map(CommittedBindingTerminal::binding_epoch),
944 };
945 let attached = origin.attached();
946 if origin.participant_id() != member.participant_id()
947 || origin.conversation_id() != member.conversation_id()
948 || expected_epoch != Some(origin.binding_epoch())
949 || attached.participant_id() != member.participant_id()
950 || attached.conversation_id() != member.conversation_id()
951 {
952 Err(StorageRestoreError::BindingAuthority)
953 } else {
954 Ok(origin)
955 }
956}
957
958impl<EF, V, LF, D> RestoredParticipantLifecycle<EF, V, LF, D> {
959 #[must_use]
961 #[allow(clippy::type_complexity)]
962 pub fn into_parts(
963 self,
964 ) -> (
965 IdentityState<EF, V, LF>,
966 Option<BindingState>,
967 Option<DetachCell<D>>,
968 ) {
969 match self {
970 Self::Live {
971 member,
972 binding,
973 binding_origin: _,
974 detach_cell,
975 } => (
976 IdentityState::Live(member),
977 Some(binding),
978 Some(detach_cell),
979 ),
980 Self::Retired(identity) => (IdentityState::Retired(identity), None, None),
981 }
982 }
983}
984
985impl<EF, V, LF, D> ParticipantConversationRestore<EF, V, LF, D> {
986 pub fn restore(
1001 self,
1002 ) -> Result<ParticipantConversationState<EF, V, LF, D>, ConversationStateRestoreError> {
1003 self.restore_with_fenced_latch(None)
1004 }
1005
1006 pub(in crate::lifecycle) fn restore_with_fenced_latch(
1025 self,
1026 fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1027 ) -> Result<ParticipantConversationState<EF, V, LF, D>, ConversationStateRestoreError> {
1028 let participants = self
1029 .participants
1030 .into_iter()
1031 .map(ParticipantLifecycleRestore::restore)
1032 .collect::<Result<Vec<_>, _>>()
1033 .map_err(ConversationStateRestoreError::Storage)?;
1034 validate_participant_frontier_projection(
1035 &participants,
1036 self.frontiers.conversation_id,
1037 &self.frontiers.active_identities,
1038 )?;
1039 let history = validated_conversation_history(&participants)?;
1040 let restored = restore_conversation_with_history(
1041 ClaimFrontiersDescription {
1042 restore: self.frontiers,
1043 sequence_ledger: self.sequence_ledger,
1044 order_ledger: self.order_ledger,
1045 fenced_marker_issued,
1046 },
1047 &self.closure,
1048 &history,
1049 )?;
1050 let (frontiers, closure) = restored.into_parts();
1051 Ok(ParticipantConversationState {
1052 participants,
1053 frontiers,
1054 closure,
1055 })
1056 }
1057}
1058
1059#[derive(Debug)]
1065pub enum ParticipantLifecycleView<'a, EF, V, LF, D> {
1066 Live {
1068 member: &'a LiveMember<EF>,
1070 binding: &'a BindingState,
1072 binding_origin: Option<&'a BindingOrigin>,
1075 detach_cell: &'a DetachCell<D>,
1077 },
1078 Retired(&'a RetiredIdentity<EF, V, LF>),
1080}
1081
1082impl<EF, V, LF, D> ParticipantLifecycleView<'_, EF, V, LF, D> {
1083 #[must_use]
1085 pub const fn participant_id(&self) -> ParticipantId {
1086 match self {
1087 Self::Live { member, .. } => member.participant_id(),
1088 Self::Retired(retired) => retired.participant_id(),
1089 }
1090 }
1091}
1092
1093impl<EF: Clone, V: Clone, LF: Clone, D: Clone> ParticipantLifecycleView<'_, EF, V, LF, D> {
1094 pub(super) fn storage_description(&self) -> ParticipantLifecycleRestore<EF, V, LF, D> {
1097 match *self {
1098 Self::Live {
1099 member,
1100 binding,
1101 binding_origin,
1102 detach_cell,
1103 } => ParticipantLifecycleRestore::Live {
1104 identity: LiveIdentityRestore {
1105 participant_id: member.participant_id(),
1106 conversation_id: member.conversation_id(),
1107 generation: member.generation(),
1108 attach_secret: member.attach_secret(),
1109 cursor: member.cursor(),
1110 enrollment_fingerprint: member.enrollment_fingerprint().clone(),
1111 latest_terminal: member
1112 .latest_terminal()
1113 .map(CommittedBindingTerminal::storage_description),
1114 },
1115 binding: describe_binding_state(*binding),
1116 binding_origin: binding_origin.copied(),
1117 detach_cell: detach_cell.storage_description(),
1118 },
1119 Self::Retired(retired) => {
1120 ParticipantLifecycleRestore::Retired(describe_retired(retired))
1121 }
1122 }
1123 }
1124}
1125
1126impl<EF, V, LF, D> RestoredParticipantLifecycle<EF, V, LF, D> {
1127 #[must_use]
1129 pub const fn view(&self) -> ParticipantLifecycleView<'_, EF, V, LF, D> {
1130 match self {
1131 Self::Live {
1132 member,
1133 binding,
1134 binding_origin,
1135 detach_cell,
1136 } => ParticipantLifecycleView::Live {
1137 member,
1138 binding,
1139 binding_origin: binding_origin.as_ref(),
1140 detach_cell,
1141 },
1142 Self::Retired(retired) => ParticipantLifecycleView::Retired(retired),
1143 }
1144 }
1145}
1146
1147const fn describe_binding_state(binding: BindingState) -> BindingStateRestore {
1148 match binding {
1149 BindingState::Detached => BindingStateRestore::Detached,
1150 BindingState::Bound(binding) => BindingStateRestore::Bound(binding),
1151 BindingState::PendingFinalization(pending) => {
1152 BindingStateRestore::PendingFinalization(pending.storage_description())
1153 }
1154 }
1155}
1156
1157impl<D: Copy> RetiredIdentity<D, D, D> {
1158 #[must_use]
1160 pub const fn storage_description(&self) -> RetiredIdentityRestore<D, D, D> {
1161 let result = self.committed_result();
1162 RetiredIdentityRestore {
1163 participant_id: self.participant_id(),
1164 conversation_id: self.conversation_id(),
1165 retired_generation: self.retired_generation(),
1166 enrollment_fingerprint: EnrollmentFingerprint::new(
1167 *self.enrollment_fingerprint().value(),
1168 ),
1169 leave_attempt_token: self.leave_attempt_token(),
1170 leave_request_verifier: *self.leave_request_verifier(),
1171 leave_fingerprint: LeaveFingerprint::new(*self.leave_fingerprint().value()),
1172 left_transaction_order: self.left_admission_order().transaction_order(),
1173 committed_result: LeaveCommittedRestore {
1174 conversation_id: result.conversation_id(),
1175 leave_attempt_token: result.leave_attempt_token(),
1176 participant_id: result.participant_id(),
1177 retired_generation: result.retired_generation(),
1178 ended_binding_epoch: result.ended_binding_epoch(),
1179 prior_terminal_delivery_seq: result.prior_terminal_delivery_seq(),
1180 left_delivery_seq: result.left_delivery_seq(),
1181 },
1182 }
1183 }
1184}
1185
1186fn describe_retired<EF: Clone, V: Clone, LF: Clone>(
1187 retired: &RetiredIdentity<EF, V, LF>,
1188) -> RetiredIdentityRestore<EF, V, LF> {
1189 let result = retired.committed_result();
1190 RetiredIdentityRestore {
1191 participant_id: retired.participant_id(),
1192 conversation_id: retired.conversation_id(),
1193 retired_generation: retired.retired_generation(),
1194 enrollment_fingerprint: retired.enrollment_fingerprint().clone(),
1195 leave_attempt_token: retired.leave_attempt_token(),
1196 leave_request_verifier: retired.leave_request_verifier().clone(),
1197 leave_fingerprint: retired.leave_fingerprint().clone(),
1198 left_transaction_order: retired.left_admission_order().transaction_order(),
1199 committed_result: LeaveCommittedRestore {
1200 conversation_id: result.conversation_id(),
1201 leave_attempt_token: result.leave_attempt_token(),
1202 participant_id: result.participant_id(),
1203 retired_generation: result.retired_generation(),
1204 ended_binding_epoch: result.ended_binding_epoch(),
1205 prior_terminal_delivery_seq: result.prior_terminal_delivery_seq(),
1206 left_delivery_seq: result.left_delivery_seq(),
1207 },
1208 }
1209}
1210
1211#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1214#[non_exhaustive]
1215pub enum ConversationDescribeError {
1216 DetachedCursorReleaseProvenance,
1219 EdgeTerminal,
1223}
1224
1225#[derive(Clone, Debug, PartialEq, Eq)]
1229pub(super) struct ParticipantConversationDescription<EF, V, LF, D> {
1230 pub(super) restore: ParticipantConversationRestore<EF, V, LF, D>,
1233 pub(super) fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1235}
1236
1237impl<EF, V, LF, D> ParticipantConversationDescription<EF, V, LF, D> {
1238 pub(super) fn into_state(
1247 self,
1248 ) -> Result<ParticipantConversationState<EF, V, LF, D>, ConversationStateRestoreError> {
1249 self.restore
1250 .restore_with_fenced_latch(self.fenced_marker_issued)
1251 }
1252}
1253
1254pub(super) fn describe_conversation<EF: Clone, V: Clone, LF: Clone, D: Clone>(
1269 participants: &[ParticipantLifecycleView<'_, EF, V, LF, D>],
1270 frontiers: &ClaimFrontiers,
1271 retained_record_limit: u64,
1272 closure: ClosureState,
1273) -> Result<ParticipantConversationDescription<EF, V, LF, D>, ConversationDescribeError> {
1274 let closure = describe_closure(participants, frontiers, closure)?;
1275 let ClaimFrontiersDescription {
1276 restore,
1277 sequence_ledger,
1278 order_ledger,
1279 fenced_marker_issued,
1280 } = frontiers.describe(retained_record_limit);
1281 Ok(ParticipantConversationDescription {
1282 restore: ParticipantConversationRestore {
1283 participants: participants
1284 .iter()
1285 .map(ParticipantLifecycleView::storage_description)
1286 .collect(),
1287 frontiers: restore,
1288 sequence_ledger,
1289 order_ledger,
1290 closure,
1291 },
1292 fenced_marker_issued,
1293 })
1294}
1295
1296fn describe_closure<EF, V, LF, D>(
1297 participants: &[ParticipantLifecycleView<'_, EF, V, LF, D>],
1298 frontiers: &ClaimFrontiers,
1299 closure: ClosureState,
1300) -> Result<ClosureStateRestore, ConversationDescribeError> {
1301 let conversation_id = frontiers.conversation_id();
1302 let ClosureState::Owed { debt, edge } = closure else {
1303 return Ok(ClosureStateRestore::Clear);
1304 };
1305 let edge = match edge {
1306 StoredEdge::ObserverProjection(edge) => edge.storage_description(),
1307 StoredEdge::PhysicalCompaction(edge) => edge.storage_description(),
1308 StoredEdge::MarkerDelivery(edge) => edge.storage_description(),
1309 StoredEdge::ParticipantCursorProgress(ParticipantCursorProgress::Continuous(edge)) => {
1310 edge.storage_description(conversation_id)
1311 }
1312 StoredEdge::ParticipantCursorProgress(ParticipantCursorProgress::Marker(edge)) => {
1313 edge.storage_description()
1314 }
1315 StoredEdge::DetachedCredentialRecovery(edge) => {
1316 let terminal = selecting_fate_terminal(
1317 participants,
1318 edge.participant_id(),
1319 edge.prior_binding_epoch(),
1320 )
1321 .ok_or(ConversationDescribeError::EdgeTerminal)?;
1322 edge.storage_description(terminal)
1323 .map_err(|_| ConversationDescribeError::EdgeTerminal)?
1324 }
1325 StoredEdge::DetachedMarkerRelease(edge) => {
1326 let terminal = match edge.provenance() {
1327 DetachedMarkerReleaseProvenance::Fate => Some(
1328 selecting_fate_terminal(
1329 participants,
1330 edge.participant_id(),
1331 edge.last_dead_binding_epoch(),
1332 )
1333 .ok_or(ConversationDescribeError::EdgeTerminal)?,
1334 ),
1335 DetachedMarkerReleaseProvenance::Candidate => None,
1336 };
1337 edge.storage_description(terminal)
1338 .map_err(|_| ConversationDescribeError::EdgeTerminal)?
1339 }
1340 StoredEdge::DetachedCursorRelease(edge) => {
1341 let source = frontiers
1342 .ordinary_cursor_release()
1343 .ok_or(ConversationDescribeError::DetachedCursorReleaseProvenance)?;
1344 let binding = source.authority.binding;
1345 if binding.conversation_id != conversation_id
1346 || binding.participant_id != edge.participant_id()
1347 || binding.binding_epoch != edge.last_dead_binding_epoch()
1348 {
1349 return Err(ConversationDescribeError::DetachedCursorReleaseProvenance);
1350 }
1351 StoredEdgeRestore::DetachedCursorRelease {
1352 participant_id: edge.participant_id(),
1353 last_dead_binding_epoch: edge.last_dead_binding_epoch(),
1354 provenance: DetachedCursorReleaseProvenanceRestore::Ordinary(*source),
1355 }
1356 }
1357 };
1358 Ok(ClosureStateRestore::Owed {
1359 debt: debt.value(),
1360 edge,
1361 })
1362}
1363
1364fn selecting_fate_terminal<EF, V, LF, D>(
1368 participants: &[ParticipantLifecycleView<'_, EF, V, LF, D>],
1369 participant_id: ParticipantId,
1370 binding_epoch: BindingEpoch,
1371) -> Option<RestoredBindingFateTerminal> {
1372 participants.iter().find_map(|participant| {
1373 let ParticipantLifecycleView::Live {
1374 member, binding, ..
1375 } = participant
1376 else {
1377 return None;
1378 };
1379 if member.participant_id() != participant_id {
1380 return None;
1381 }
1382 if let BindingState::PendingFinalization(pending) = **binding
1383 && pending.binding_epoch() == binding_epoch
1384 {
1385 return Some(RestoredBindingFateTerminal::Pending(pending));
1386 }
1387 member
1388 .latest_terminal()
1389 .filter(|terminal| terminal.binding_epoch() == binding_epoch)
1390 .map(RestoredBindingFateTerminal::Committed)
1391 })
1392}
1393
1394fn validated_conversation_history<EF, V, LF, D>(
1395 participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
1396) -> Result<ValidatedConversationHistory, ConversationStateRestoreError> {
1397 let mut causal_authorities = Vec::new();
1398 let mut binding_origins = Vec::new();
1399 let mut seen = Vec::with_capacity(participants.len());
1400 for participant in participants {
1401 let participant_id = match participant {
1402 RestoredParticipantLifecycle::Live {
1403 member,
1404 binding_origin,
1405 ..
1406 } => {
1407 if let Some(terminal) = member.latest_terminal() {
1408 causal_authorities
1409 .push(HistoricalCausalAuthority::from_committed_terminal(terminal));
1410 }
1411 if let Some(origin) = binding_origin {
1412 binding_origins.push(*origin);
1413 }
1414 member.participant_id()
1415 }
1416 RestoredParticipantLifecycle::Retired(retired) => {
1417 causal_authorities.push(HistoricalCausalAuthority::from_retired(retired));
1418 retired.participant_id()
1419 }
1420 };
1421 seen.push(participant_id);
1422 }
1423 seen.sort_unstable();
1427 let has_duplicate = seen
1428 .iter()
1429 .zip(seen.iter().skip(1))
1430 .any(|(current, next)| current == next);
1431 if has_duplicate {
1432 return Err(ConversationStateRestoreError::Storage(
1433 StorageRestoreError::MembershipInvariant,
1434 ));
1435 }
1436 Ok(ValidatedConversationHistory::new(
1437 causal_authorities,
1438 binding_origins,
1439 ))
1440}
1441
1442fn validate_participant_frontier_projection<EF, V, LF, D>(
1443 participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
1444 conversation_id: ConversationId,
1445 frontier: &[super::FrontierParticipant],
1446) -> Result<(), ConversationStateRestoreError> {
1447 let mut projected = Vec::new();
1448 for participant in participants {
1449 match participant {
1450 RestoredParticipantLifecycle::Live {
1451 member, binding, ..
1452 } => {
1453 if member.conversation_id() != conversation_id {
1454 return Err(ConversationStateRestoreError::Storage(
1455 StorageRestoreError::MembershipInvariant,
1456 ));
1457 }
1458 let binding = match binding {
1459 BindingState::Bound(binding) => FrontierBinding::Bound(binding.binding_epoch),
1460 BindingState::PendingFinalization(pending) => {
1461 FrontierBinding::Detached(pending.binding_epoch())
1462 }
1463 BindingState::Detached => {
1464 let Some(terminal) = member.latest_terminal() else {
1465 return Err(ConversationStateRestoreError::Storage(
1466 StorageRestoreError::BindingAuthority,
1467 ));
1468 };
1469 FrontierBinding::Detached(terminal.binding_epoch())
1470 }
1471 };
1472 projected.push(super::FrontierParticipant::new(
1473 member.participant_id(),
1474 member.cursor(),
1475 binding,
1476 ));
1477 }
1478 RestoredParticipantLifecycle::Retired(retired) => {
1479 if retired.conversation_id() != conversation_id {
1480 return Err(ConversationStateRestoreError::Storage(
1481 StorageRestoreError::MembershipInvariant,
1482 ));
1483 }
1484 }
1485 }
1486 }
1487 projected.sort_by_key(|participant| participant.participant_index());
1488 if projected == frontier {
1489 Ok(())
1490 } else {
1491 Err(ConversationStateRestoreError::Storage(
1492 StorageRestoreError::MembershipInvariant,
1493 ))
1494 }
1495}
1496
1497#[allow(clippy::suspicious_operation_groupings)]
1498fn validate_live_pair<EF, D>(
1499 member: &LiveMember<EF>,
1500 binding: BindingState,
1501 detach_cell: &DetachCell<D>,
1502) -> Result<(), StorageRestoreError> {
1503 match detach_cell {
1504 DetachCell::Empty(_) => Ok(()),
1505 DetachCell::Pending(cell) => {
1506 if cell.participant_id() != member.participant_id()
1507 || cell.request_generation() != member.generation()
1508 || validate_pending_pair(binding, cell, Some(member.conversation_id())).is_err()
1509 {
1510 Err(StorageRestoreError::DetachBindingPair)
1511 } else {
1512 Ok(())
1513 }
1514 }
1515 DetachCell::Committed(cell) => {
1516 let terminal_matches = member.latest_terminal().is_some_and(|terminal| {
1517 terminal.participant_id() == cell.participant_id()
1518 && terminal.conversation_id() == member.conversation_id()
1519 && terminal.binding_epoch() == cell.committed_binding_epoch()
1520 && terminal.delivery_seq() == cell.detached_delivery_seq()
1521 && terminal.detached_cause()
1522 == Some(crate::wire::DetachedCause::CleanDeregister)
1523 });
1524 if cell.participant_id() != member.participant_id()
1525 || cell.request_generation() != member.generation()
1526 || binding != BindingState::Detached
1527 || !terminal_matches
1528 {
1529 Err(StorageRestoreError::DetachBindingPair)
1530 } else {
1531 Ok(())
1532 }
1533 }
1534 DetachCell::Terminalized(cell) => {
1535 if cell.participant_id() != member.participant_id()
1536 || cell.request_generation().get() >= member.generation().get()
1537 {
1538 Err(StorageRestoreError::DetachBindingPair)
1539 } else {
1540 Ok(())
1541 }
1542 }
1543 }
1544}
1545
1546#[derive(Clone, Debug, PartialEq, Eq)]
1548pub struct CursorEpisodeRestore {
1549 pub conversation_id: ConversationId,
1551 pub debt: WideResourceVector,
1553 pub observer_progress: DeliverySeq,
1555 pub candidate_high_watermark: DeliverySeq,
1557 pub current_floor: u128,
1559 pub cap_floor: u128,
1561 pub participants: Vec<BoundParticipantCursor>,
1563 pub facts: Vec<(CursorProgressKey, CursorProgressFact)>,
1565}
1566
1567impl CursorEpisodeRestore {
1568 pub fn restore(self) -> Result<NonzeroDebtCursorEpisode, StorageRestoreError> {
1576 let debt = ClosureDebt::new(self.debt).ok_or(StorageRestoreError::ClosureDebt)?;
1577 NonzeroDebtCursorEpisode::restore(
1578 self.conversation_id,
1579 debt,
1580 self.observer_progress,
1581 self.candidate_high_watermark,
1582 self.current_floor,
1583 self.cap_floor,
1584 self.participants,
1585 self.facts,
1586 )
1587 .ok_or(StorageRestoreError::CursorEpisode)
1588 }
1589}
1590
1591#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1605pub struct OrdinaryBindingAuthorityRestore {
1606 pub binding: ActiveBinding,
1608 pub through_seq: DeliverySeq,
1610}
1611
1612impl OrdinaryBindingAuthorityRestore {
1613 pub(in crate::lifecycle) fn restore_with_origin(
1614 self,
1615 origin: &BindingOrigin,
1616 ) -> Result<OrdinaryBindingAuthority, StorageRestoreError> {
1617 if !origin.is_unfenced()
1618 || origin.conversation_id() != self.binding.conversation_id
1619 || origin.participant_id() != self.binding.participant_id
1620 || origin.binding_epoch() != self.binding.binding_epoch
1621 {
1622 return Err(StorageRestoreError::StoredEdgeProvenance);
1623 }
1624 Ok(OrdinaryBindingAuthority::new(
1625 self.binding,
1626 self.through_seq,
1627 ))
1628 }
1629}
1630
1631#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1652pub struct MarkerDeliveryRestore {
1653 pub participant_id: ParticipantId,
1655 pub binding_epoch: BindingEpoch,
1657 pub marker_delivery_seq: DeliverySeq,
1659}
1660
1661impl MarkerDeliveryRestore {
1662 #[cfg(any(test, feature = "test-support"))]
1670 pub(super) fn restore_bound(
1671 self,
1672 conversation_id: ConversationId,
1673 record_authority: ValidatedMarkerRecord,
1674 ) -> Result<MarkerDelivery, StorageRestoreError> {
1675 let restored = self.restore_with_target(
1676 conversation_id,
1677 &record_authority,
1678 MarkerRecordTarget::Bound,
1679 MarkerRecordOccurrence::Undelivered,
1680 );
1681 record_authority.consume();
1682 restored
1683 }
1684
1685 #[cfg(test)]
1687 pub(super) fn restore_detached_delivered_for_test(
1688 self,
1689 conversation_id: ConversationId,
1690 record_authority: ValidatedMarkerRecord,
1691 ) -> Result<MarkerDelivery, StorageRestoreError> {
1692 let restored = self.restore_with_target(
1693 conversation_id,
1694 &record_authority,
1695 MarkerRecordTarget::Detached,
1696 MarkerRecordOccurrence::Delivered,
1697 );
1698 record_authority.consume();
1699 restored
1700 }
1701
1702 fn restore_detached(
1703 self,
1704 conversation_id: ConversationId,
1705 record_authority: &ValidatedMarkerRecord,
1706 ) -> Result<MarkerDelivery, StorageRestoreError> {
1707 self.restore_with_target(
1708 conversation_id,
1709 record_authority,
1710 MarkerRecordTarget::Detached,
1711 MarkerRecordOccurrence::Undelivered,
1712 )
1713 }
1714
1715 fn restore_with_target(
1716 self,
1717 conversation_id: ConversationId,
1718 record_authority: &ValidatedMarkerRecord,
1719 target: MarkerRecordTarget,
1720 occurrence: MarkerRecordOccurrence,
1721 ) -> Result<MarkerDelivery, StorageRestoreError> {
1722 if record_authority.conversation_id() != conversation_id
1723 || !target.matches(record_authority.target_binding(), self.binding_epoch)
1724 || record_authority.occurrence() != occurrence
1725 {
1726 return Err(StorageRestoreError::StoredEdgeProvenance);
1727 }
1728 let delivery = MarkerDelivery::from_validated_record(record_authority);
1729 if delivery.participant_id() != self.participant_id
1730 || delivery.binding_epoch() != self.binding_epoch
1731 || delivery.marker_delivery_seq() != self.marker_delivery_seq
1732 {
1733 return Err(StorageRestoreError::StoredEdgeProvenance);
1734 }
1735 Ok(delivery)
1736 }
1737}
1738
1739#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1740enum MarkerRecordTarget {
1741 Bound,
1742 Detached,
1743}
1744
1745impl MarkerRecordTarget {
1746 fn matches(self, target: super::FrontierBinding, epoch: BindingEpoch) -> bool {
1747 matches!(
1748 (self, target),
1749 (Self::Bound, super::FrontierBinding::Bound(actual))
1750 | (Self::Detached, super::FrontierBinding::Detached(actual))
1751 if actual == epoch
1752 )
1753 }
1754}
1755
1756#[derive(Debug)]
1757enum MarkerRestoreAuthority<'a> {
1758 Absent,
1759 Record {
1760 conversation_id: ConversationId,
1761 record: &'a ValidatedMarkerRecord,
1762 },
1763}
1764
1765impl MarkerRestoreAuthority<'_> {
1766 const fn require_absent(&self) -> Result<(), StorageRestoreError> {
1767 match self {
1768 Self::Absent => Ok(()),
1769 Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
1770 }
1771 }
1772
1773 const fn require_record(
1774 &self,
1775 ) -> Result<(ConversationId, &ValidatedMarkerRecord), StorageRestoreError> {
1776 match self {
1777 Self::Record {
1778 conversation_id,
1779 record,
1780 } if record.conversation_id() == *conversation_id => Ok((*conversation_id, record)),
1781 Self::Absent | Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
1782 }
1783 }
1784
1785 fn record_for(
1786 &self,
1787 expected_conversation_id: ConversationId,
1788 ) -> Result<&ValidatedMarkerRecord, StorageRestoreError> {
1789 let (conversation_id, record) = self.require_record()?;
1790 if conversation_id == expected_conversation_id {
1791 Ok(record)
1792 } else {
1793 Err(StorageRestoreError::StoredEdgeProvenance)
1794 }
1795 }
1796}
1797
1798#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1800pub struct MarkerCursorProgressRestore {
1801 pub conversation_id: ConversationId,
1803 pub participant_id: ParticipantId,
1805 pub binding_epoch: BindingEpoch,
1807 pub through_seq: DeliverySeq,
1809 pub marker_delivery_seq: DeliverySeq,
1811 pub delivery: MarkerDeliveryRestore,
1813}
1814
1815impl MarkerCursorProgressRestore {
1816 fn restore_with_debt(
1817 self,
1818 debt: ClosureDebt,
1819 record_authority: &ValidatedMarkerRecord,
1820 target: MarkerRecordTarget,
1821 ) -> Result<ParticipantCursorProgress, StorageRestoreError> {
1822 if self.participant_id != self.delivery.participant_id
1823 || self.binding_epoch != self.delivery.binding_epoch
1824 || self.marker_delivery_seq != self.delivery.marker_delivery_seq
1825 || self.through_seq != self.marker_delivery_seq
1826 {
1827 return Err(StorageRestoreError::StoredEdgeProvenance);
1828 }
1829 let marker = self.delivery.restore_with_target(
1830 self.conversation_id,
1831 record_authority,
1832 target,
1833 MarkerRecordOccurrence::Delivered,
1834 )?;
1835 let state = marker
1836 .delivered(
1837 debt,
1838 Event::marker_delivered(
1839 self.participant_id,
1840 self.binding_epoch,
1841 self.marker_delivery_seq,
1842 ),
1843 )
1844 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1845 match state {
1846 ClosureState::Owed {
1847 edge: StoredEdge::ParticipantCursorProgress(progress),
1848 ..
1849 } => Ok(progress),
1850 ClosureState::Clear | ClosureState::Owed { .. } => {
1851 Err(StorageRestoreError::StoredEdgeProvenance)
1852 }
1853 }
1854 }
1855}
1856
1857#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1859pub struct DetachedCredentialRecoveryRestore {
1860 pub participant_id: ParticipantId,
1862 pub marker_delivery_seq: DeliverySeq,
1864 pub prior_binding_epoch: BindingEpoch,
1866 pub resulting_floor: DeliverySeq,
1868 pub terminal: BindingFateTerminalRestore,
1870 pub progress: MarkerCursorProgressRestore,
1872}
1873
1874impl DetachedCredentialRecoveryRestore {
1875 pub fn restore_description(self) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
1887 if self.participant_id != self.progress.participant_id
1888 || self.marker_delivery_seq != self.progress.marker_delivery_seq
1889 || self.prior_binding_epoch != self.progress.binding_epoch
1890 || self.progress.through_seq != self.marker_delivery_seq
1891 || self.progress.participant_id != self.progress.delivery.participant_id
1892 || self.progress.binding_epoch != self.progress.delivery.binding_epoch
1893 || self.progress.marker_delivery_seq != self.progress.delivery.marker_delivery_seq
1894 {
1895 return Err(StorageRestoreError::StoredEdgeProvenance);
1896 }
1897 let terminal = self.terminal.restore()?;
1898 if terminal.participant_id() != self.participant_id
1899 || terminal.binding_epoch() != self.prior_binding_epoch
1900 || terminal.conversation_id() != self.progress.conversation_id
1901 {
1902 return Err(StorageRestoreError::StoredEdgeProvenance);
1903 }
1904 Ok(DetachedCredentialRecovery::from_storage_description(
1905 self.progress.conversation_id,
1906 self.participant_id,
1907 self.marker_delivery_seq,
1908 self.prior_binding_epoch,
1909 self.resulting_floor,
1910 ))
1911 }
1912
1913 fn restore_with_debt(
1914 self,
1915 debt: ClosureDebt,
1916 record_authority: &ValidatedMarkerRecord,
1917 ) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
1918 if self.participant_id != self.progress.participant_id
1919 || self.marker_delivery_seq != self.progress.marker_delivery_seq
1920 || self.prior_binding_epoch != self.progress.binding_epoch
1921 {
1922 return Err(StorageRestoreError::StoredEdgeProvenance);
1923 }
1924 let terminal = self.terminal.restore()?;
1925 if terminal.participant_id() != self.participant_id
1926 || terminal.binding_epoch() != self.prior_binding_epoch
1927 || terminal.conversation_id() != self.progress.conversation_id
1928 {
1929 return Err(StorageRestoreError::StoredEdgeProvenance);
1930 }
1931 let progress = self.progress.restore_with_debt(
1932 debt,
1933 record_authority,
1934 MarkerRecordTarget::Detached,
1935 )?;
1936 let successor = progress
1937 .binding_fate(
1938 debt,
1939 terminal,
1940 Event::binding_fate_observed(
1941 self.participant_id,
1942 self.prior_binding_epoch,
1943 self.resulting_floor,
1944 ),
1945 )
1946 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1947 match successor.into_stored_edge() {
1948 StoredEdge::DetachedCredentialRecovery(edge) => Ok(edge),
1949 _ => Err(StorageRestoreError::StoredEdgeProvenance),
1950 }
1951 }
1952}
1953
1954#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1956pub struct DetachedMarkerReleaseRestore {
1957 pub conversation_id: ConversationId,
1959 pub participant_id: ParticipantId,
1961 pub marker_delivery_seq: DeliverySeq,
1963 pub last_dead_binding_epoch: BindingEpoch,
1965 pub delivery: MarkerDeliveryRestore,
1967 pub provenance: DetachedMarkerReleaseProvenanceRestore,
1969}
1970
1971#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1976pub enum DetachedMarkerReleaseProvenanceRestore {
1977 Fate {
1985 terminal: BindingFateTerminalRestore,
1987 },
1988 Candidate,
1991}
1992
1993impl DetachedMarkerReleaseRestore {
1994 fn restore_with_marker_record(
2000 self,
2001 record_authority: &ValidatedMarkerRecord,
2002 ) -> Result<DetachedMarkerRelease, StorageRestoreError> {
2003 if self.participant_id != self.delivery.participant_id
2004 || self.marker_delivery_seq != self.delivery.marker_delivery_seq
2005 || self.last_dead_binding_epoch != self.delivery.binding_epoch
2006 {
2007 return Err(StorageRestoreError::StoredEdgeProvenance);
2008 }
2009 let provenance = match self.provenance {
2010 DetachedMarkerReleaseProvenanceRestore::Fate { terminal } => {
2011 let terminal = terminal.restore()?;
2012 if terminal.participant_id() != self.participant_id
2013 || terminal.binding_epoch() != self.last_dead_binding_epoch
2014 || terminal.conversation_id() != self.conversation_id
2015 {
2016 return Err(StorageRestoreError::StoredEdgeProvenance);
2017 }
2018 DetachedMarkerReleaseProvenance::Fate
2019 }
2020 DetachedMarkerReleaseProvenanceRestore::Candidate => {
2021 DetachedMarkerReleaseProvenance::Candidate
2022 }
2023 };
2024 self.delivery
2025 .restore_detached(self.conversation_id, record_authority)?;
2026 Ok(DetachedMarkerRelease::from_storage_description(
2027 self.conversation_id,
2028 self.participant_id,
2029 self.marker_delivery_seq,
2030 self.last_dead_binding_epoch,
2031 provenance,
2032 ))
2033 }
2034}
2035
2036#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2049pub struct OrdinaryBindingFateRestore {
2050 pub authority: OrdinaryBindingAuthorityRestore,
2052 pub terminal: CommittedBindingTerminalRestore,
2054 pub resulting_floor: DeliverySeq,
2056}
2057
2058impl OrdinaryBindingFateRestore {
2059 pub fn restore_with_origin(
2066 self,
2067 origin: &BindingOrigin,
2068 ) -> Result<OrdinaryBindingFate, StorageRestoreError> {
2069 let authority = self.authority.restore_with_origin(origin)?;
2070 let terminal = self.terminal.restore()?;
2071 let CommittedBindingTerminal::Died(terminal) = terminal else {
2072 return Err(StorageRestoreError::StoredEdgeProvenance);
2073 };
2074 authority
2075 .binding_fate(terminal, self.resulting_floor)
2076 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
2077 }
2078}
2079
2080#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2082pub enum DebtCompletionRestore {
2083 Clear,
2085 ObserverProjection {
2087 debt: WideResourceVector,
2089 through_seq: DeliverySeq,
2091 },
2092 PhysicalCompaction {
2094 debt: WideResourceVector,
2096 from_floor: DeliverySeq,
2098 through_seq: DeliverySeq,
2100 },
2101}
2102
2103impl DebtCompletionRestore {
2104 pub fn restore(self) -> Result<DebtCompletion, StorageRestoreError> {
2110 match self {
2111 Self::Clear => Ok(DebtCompletion::clear()),
2112 Self::ObserverProjection { debt, through_seq } => {
2113 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
2114 Ok(DebtCompletion::observer_projection(
2115 debt,
2116 ObserverProjection::new(through_seq),
2117 ))
2118 }
2119 Self::PhysicalCompaction {
2120 debt,
2121 from_floor,
2122 through_seq,
2123 } => {
2124 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
2125 let edge = PhysicalCompaction::new(from_floor, through_seq)
2126 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2127 Ok(DebtCompletion::physical_compaction(debt, edge))
2128 }
2129 }
2130 }
2131}
2132
2133#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2147pub struct FencedAttachCommitRestore {
2148 pub predecessor: DetachedCredentialRecoveryRestore,
2150 pub predecessor_debt: WideResourceVector,
2152 pub participant_id: ParticipantId,
2154 pub marker_delivery_seq: DeliverySeq,
2156 pub prior_binding_epoch: BindingEpoch,
2158 pub new_binding_epoch: BindingEpoch,
2160 pub resulting_floor: DeliverySeq,
2162 pub successor: DebtCompletionRestore,
2164}
2165
2166impl FencedAttachCommitRestore {
2167 #[cfg(test)]
2174 pub(super) fn restore(
2175 self,
2176 record_authority: ValidatedMarkerRecord,
2177 ) -> Result<FencedAttachCommit, StorageRestoreError> {
2178 let restored = self.restore_with_record(&record_authority);
2179 record_authority.consume();
2180 restored
2181 }
2182
2183 pub(super) fn restore_with_record(
2184 self,
2185 record_authority: &ValidatedMarkerRecord,
2186 ) -> Result<FencedAttachCommit, StorageRestoreError> {
2187 let debt =
2188 ClosureDebt::new(self.predecessor_debt).ok_or(StorageRestoreError::ClosureDebt)?;
2189 let predecessor = self.predecessor.restore_with_debt(debt, record_authority)?;
2190 FencedAttachCommit::restore_validated(
2191 predecessor,
2192 record_authority,
2193 debt,
2194 Event::fenced_recovery_committed(
2195 self.participant_id,
2196 self.marker_delivery_seq,
2197 self.prior_binding_epoch,
2198 self.new_binding_epoch,
2199 self.resulting_floor,
2200 ),
2201 self.successor.restore()?,
2202 )
2203 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2204 }
2205}
2206
2207#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2217pub struct RecoveredBindingFateRestore {
2218 pub fenced_attach: FencedAttachCommitRestore,
2220 pub participant_id: ParticipantId,
2222 pub binding_epoch: BindingEpoch,
2224 pub resulting_floor: DeliverySeq,
2226}
2227
2228impl RecoveredBindingFateRestore {
2229 #[cfg(test)]
2236 pub(super) fn restore(
2237 self,
2238 record_authority: ValidatedMarkerRecord,
2239 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
2240 let restored = self.restore_with_record(&record_authority);
2241 record_authority.consume();
2242 restored
2243 }
2244
2245 fn restore_with_record(
2246 self,
2247 record_authority: &ValidatedMarkerRecord,
2248 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
2249 let commit = self.fenced_attach.restore_with_record(record_authority)?;
2250 commit
2251 .recovered_binding_fate(Event::binding_fate_observed(
2252 self.participant_id,
2253 self.binding_epoch,
2254 self.resulting_floor,
2255 ))
2256 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
2257 }
2258}
2259
2260#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2262pub struct PendingRecoveredCursorReleaseRestore {
2263 pub fate: RecoveredBindingFateRestore,
2265 pub resulting_debt: WideResourceVector,
2267}
2268
2269impl PendingRecoveredCursorReleaseRestore {
2270 #[cfg(test)]
2277 pub(super) fn restore(
2278 self,
2279 record_authority: ValidatedMarkerRecord,
2280 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
2281 let restored = self.restore_with_record(&record_authority);
2282 record_authority.consume();
2283 restored
2284 }
2285
2286 fn restore_with_record(
2287 self,
2288 record_authority: &ValidatedMarkerRecord,
2289 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
2290 let authority = self.fate.restore_with_record(record_authority)?;
2291 let predecessor_state = authority.predecessor_state();
2292 let resulting_debt =
2293 ClosureDebt::new(self.resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
2294 let transition = match predecessor_state {
2295 ClosureState::Owed {
2296 debt,
2297 edge: StoredEdge::ObserverProjection(edge),
2298 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
2299 ClosureState::Owed {
2300 debt,
2301 edge: StoredEdge::PhysicalCompaction(edge),
2302 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
2303 ClosureState::Clear | ClosureState::Owed { .. } => {
2304 return Err(StorageRestoreError::StoredEdgeProvenance);
2305 }
2306 }
2307 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
2308 match transition {
2309 RecoveredBindingFateTransition::PendingStorage(pending) => Ok(pending),
2310 RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
2311 Err(StorageRestoreError::StoredEdgeProvenance)
2312 }
2313 }
2314 }
2315}
2316
2317#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2319pub enum RecoveredStorageCompletionRestore {
2320 ObserverProjection {
2322 through_seq: DeliverySeq,
2324 resulting_debt: Option<WideResourceVector>,
2326 },
2327 PhysicalCompaction {
2329 from_floor: DeliverySeq,
2331 through_seq: DeliverySeq,
2333 resulting_floor: DeliverySeq,
2335 resulting_debt: Option<WideResourceVector>,
2337 },
2338}
2339
2340impl RecoveredStorageCompletionRestore {
2341 fn restore(
2342 self,
2343 pending: PendingRecoveredCursorRelease,
2344 ) -> Result<ClosureState, StorageRestoreError> {
2345 let current = pending.current_state();
2346 match (current, self) {
2347 (
2348 ClosureState::Owed {
2349 edge: StoredEdge::ObserverProjection(edge),
2350 ..
2351 },
2352 Self::ObserverProjection {
2353 through_seq,
2354 resulting_debt,
2355 },
2356 ) => edge
2357 .complete_after_recovered_binding_fate(
2358 Event::projection_completed(through_seq),
2359 optional_debt(resulting_debt)?,
2360 pending,
2361 )
2362 .map_err(|_| StorageRestoreError::StoredEdgeProvenance),
2363 (
2364 ClosureState::Owed {
2365 edge: StoredEdge::PhysicalCompaction(edge),
2366 ..
2367 },
2368 Self::PhysicalCompaction {
2369 from_floor,
2370 through_seq,
2371 resulting_floor,
2372 resulting_debt,
2373 },
2374 ) => {
2375 let event = Event::compaction_completed(from_floor, through_seq, resulting_floor)
2376 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2377 edge.complete_after_recovered_binding_fate(
2378 event,
2379 optional_debt(resulting_debt)?,
2380 pending,
2381 )
2382 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
2383 }
2384 _ => Err(StorageRestoreError::StoredEdgeProvenance),
2385 }
2386 }
2387}
2388
2389#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2391pub enum DetachedCursorReleaseProvenanceRestore {
2392 Ordinary(OrdinaryBindingFateRestore),
2394 RecoveredDirect {
2396 fate: RecoveredBindingFateRestore,
2398 resulting_debt: WideResourceVector,
2400 },
2401 RecoveredAfterStorage {
2403 pending: PendingRecoveredCursorReleaseRestore,
2405 completion: RecoveredStorageCompletionRestore,
2407 },
2408}
2409
2410impl DetachedCursorReleaseProvenanceRestore {
2411 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
2412 match self {
2413 Self::Ordinary(fate) => Some(fate.authority.binding),
2414 Self::RecoveredDirect { .. } | Self::RecoveredAfterStorage { .. } => None,
2415 }
2416 }
2417
2418 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2419 match self {
2420 Self::Ordinary(_) => None,
2421 Self::RecoveredDirect { fate, .. } => Some(MarkerRecordRequest::recovered(
2422 fate.participant_id,
2423 fate.fenced_attach.marker_delivery_seq,
2424 fate.fenced_attach.prior_binding_epoch,
2425 fate.fenced_attach.new_binding_epoch,
2426 )),
2427 Self::RecoveredAfterStorage { pending, .. } => Some(MarkerRecordRequest::recovered(
2428 pending.fate.participant_id,
2429 pending.fate.fenced_attach.marker_delivery_seq,
2430 pending.fate.fenced_attach.prior_binding_epoch,
2431 pending.fate.fenced_attach.new_binding_epoch,
2432 )),
2433 }
2434 }
2435
2436 fn restore_state(
2437 self,
2438 debt: ClosureDebt,
2439 marker_authority: &MarkerRestoreAuthority<'_>,
2440 ordinary_origin: Option<&BindingOrigin>,
2441 ) -> Result<ClosureState, StorageRestoreError> {
2442 match self {
2443 Self::Ordinary(provenance) => {
2444 marker_authority.require_absent()?;
2445 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2446 Ok(provenance
2447 .restore_with_origin(origin)?
2448 .into_direct_state(debt))
2449 }
2450 Self::RecoveredDirect {
2451 fate,
2452 resulting_debt,
2453 } => {
2454 let (_, record_authority) = marker_authority.require_record()?;
2455 let authority = fate.restore_with_record(record_authority)?;
2456 let predecessor = authority.predecessor_state();
2457 let resulting_debt =
2458 ClosureDebt::new(resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
2459 let transition = match predecessor {
2460 ClosureState::Owed {
2461 debt: predecessor_debt,
2462 edge: StoredEdge::PhysicalCompaction(edge),
2463 } => edge.apply_recovered_binding_fate(
2464 predecessor_debt,
2465 resulting_debt,
2466 authority,
2467 ),
2468 ClosureState::Clear | ClosureState::Owed { .. } => {
2469 return Err(StorageRestoreError::StoredEdgeProvenance);
2470 }
2471 }
2472 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
2473 match transition {
2474 RecoveredBindingFateTransition::DetachedCursorRelease(release)
2475 if release.debt() == debt =>
2476 {
2477 Ok(release.into_state())
2478 }
2479 RecoveredBindingFateTransition::PendingStorage(_)
2480 | RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
2481 Err(StorageRestoreError::StoredEdgeProvenance)
2482 }
2483 }
2484 }
2485 Self::RecoveredAfterStorage {
2486 pending,
2487 completion,
2488 } => {
2489 let (_, record_authority) = marker_authority.require_record()?;
2490 let state = completion.restore(pending.restore_with_record(record_authority)?)?;
2491 match state {
2492 ClosureState::Owed {
2493 debt: restored_debt,
2494 edge: StoredEdge::DetachedCursorRelease(_),
2495 } if restored_debt == debt => Ok(state),
2496 ClosureState::Clear | ClosureState::Owed { .. } => {
2497 Err(StorageRestoreError::StoredEdgeProvenance)
2498 }
2499 }
2500 }
2501 }
2502 }
2503}
2504
2505#[allow(clippy::large_enum_variant)]
2507#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2508pub enum StoredEdgeRestore {
2509 ObserverProjection {
2511 through_seq: DeliverySeq,
2513 },
2514 PhysicalCompaction {
2516 from_floor: DeliverySeq,
2518 through_seq: DeliverySeq,
2520 },
2521 MarkerDelivery(MarkerDeliveryRestore),
2523 ParticipantCursorProgressContinuous {
2525 participant_id: ParticipantId,
2527 binding_epoch: BindingEpoch,
2529 through_seq: DeliverySeq,
2531 authority: OrdinaryBindingAuthorityRestore,
2533 },
2534 ParticipantCursorProgressMarker(MarkerCursorProgressRestore),
2536 DetachedCredentialRecovery(DetachedCredentialRecoveryRestore),
2538 DetachedMarkerRelease(DetachedMarkerReleaseRestore),
2540 DetachedCursorRelease {
2542 participant_id: ParticipantId,
2544 last_dead_binding_epoch: BindingEpoch,
2546 provenance: DetachedCursorReleaseProvenanceRestore,
2548 },
2549}
2550
2551impl StoredEdgeRestore {
2552 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
2553 match self {
2554 Self::ParticipantCursorProgressContinuous { authority, .. } => Some(authority.binding),
2555 Self::DetachedCursorRelease { provenance, .. } => provenance.ordinary_binding_request(),
2556 Self::ObserverProjection { .. }
2557 | Self::PhysicalCompaction { .. }
2558 | Self::MarkerDelivery(_)
2559 | Self::ParticipantCursorProgressMarker(_)
2560 | Self::DetachedCredentialRecovery(_)
2561 | Self::DetachedMarkerRelease(_) => None,
2562 }
2563 }
2564
2565 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2566 match self {
2567 Self::MarkerDelivery(value) => Some(MarkerRecordRequest::planned(
2568 value.participant_id,
2569 value.marker_delivery_seq,
2570 super::FrontierBinding::Bound(value.binding_epoch),
2571 )),
2572 Self::ParticipantCursorProgressMarker(value) => Some(MarkerRecordRequest::delivered(
2573 value.participant_id,
2574 value.marker_delivery_seq,
2575 super::FrontierBinding::Bound(value.binding_epoch),
2576 )),
2577 Self::DetachedCredentialRecovery(value) => Some(MarkerRecordRequest::delivered(
2578 value.participant_id,
2579 value.marker_delivery_seq,
2580 super::FrontierBinding::Detached(value.prior_binding_epoch),
2581 )),
2582 Self::DetachedMarkerRelease(value) => Some(MarkerRecordRequest::planned(
2583 value.participant_id,
2584 value.marker_delivery_seq,
2585 super::FrontierBinding::Detached(value.last_dead_binding_epoch),
2586 )),
2587 Self::DetachedCursorRelease { provenance, .. } => provenance.marker_record_request(),
2588 Self::ObserverProjection { .. }
2589 | Self::PhysicalCompaction { .. }
2590 | Self::ParticipantCursorProgressContinuous { .. } => None,
2591 }
2592 }
2593
2594 fn restore_with_debt(
2595 self,
2596 debt: ClosureDebt,
2597 marker_authority: &MarkerRestoreAuthority<'_>,
2598 ordinary_origin: Option<&BindingOrigin>,
2599 ) -> Result<StoredEdge, StorageRestoreError> {
2600 match self {
2601 Self::ObserverProjection { through_seq } => {
2602 marker_authority.require_absent()?;
2603 Ok(StoredEdge::ObserverProjection(ObserverProjection::new(
2604 through_seq,
2605 )))
2606 }
2607 Self::PhysicalCompaction {
2608 from_floor,
2609 through_seq,
2610 } => {
2611 marker_authority.require_absent()?;
2612 PhysicalCompaction::new(from_floor, through_seq)
2613 .map(StoredEdge::PhysicalCompaction)
2614 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2615 }
2616 Self::MarkerDelivery(value) => {
2617 let (conversation_id, record_authority) = marker_authority.require_record()?;
2618 value
2619 .restore_with_target(
2620 conversation_id,
2621 record_authority,
2622 MarkerRecordTarget::Bound,
2623 MarkerRecordOccurrence::Undelivered,
2624 )
2625 .map(StoredEdge::MarkerDelivery)
2626 }
2627 Self::ParticipantCursorProgressContinuous {
2628 participant_id,
2629 binding_epoch,
2630 through_seq,
2631 authority,
2632 } => {
2633 marker_authority.require_absent()?;
2634 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2635 ParticipantCursorProgress::restore_continuous(
2636 authority.restore_with_origin(origin)?,
2637 participant_id,
2638 binding_epoch,
2639 through_seq,
2640 )
2641 .map(StoredEdge::ParticipantCursorProgress)
2642 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2643 }
2644 Self::ParticipantCursorProgressMarker(value) => {
2645 let record_authority = marker_authority.record_for(value.conversation_id)?;
2646 value
2647 .restore_with_debt(debt, record_authority, MarkerRecordTarget::Bound)
2648 .map(StoredEdge::ParticipantCursorProgress)
2649 }
2650 Self::DetachedCredentialRecovery(value) => {
2651 let record_authority =
2652 marker_authority.record_for(value.progress.conversation_id)?;
2653 value
2654 .restore_with_debt(debt, record_authority)
2655 .map(StoredEdge::DetachedCredentialRecovery)
2656 }
2657 Self::DetachedMarkerRelease(value) => {
2658 let record_authority = marker_authority.record_for(value.conversation_id)?;
2659 value
2660 .restore_with_marker_record(record_authority)
2661 .map(StoredEdge::DetachedMarkerRelease)
2662 }
2663 Self::DetachedCursorRelease {
2664 participant_id,
2665 last_dead_binding_epoch,
2666 provenance,
2667 } => {
2668 let state = provenance.restore_state(debt, marker_authority, ordinary_origin)?;
2669 match state {
2670 ClosureState::Owed {
2671 edge: StoredEdge::DetachedCursorRelease(edge),
2672 ..
2673 } if edge.participant_id() == participant_id
2674 && edge.last_dead_binding_epoch() == last_dead_binding_epoch =>
2675 {
2676 Ok(StoredEdge::DetachedCursorRelease(edge))
2677 }
2678 ClosureState::Clear | ClosureState::Owed { .. } => {
2679 Err(StorageRestoreError::StoredEdgeProvenance)
2680 }
2681 }
2682 }
2683 }
2684 }
2685}
2686
2687#[allow(clippy::large_enum_variant)]
2689#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2690pub enum ClosureStateRestore {
2691 Clear,
2693 Owed {
2695 debt: WideResourceVector,
2697 edge: StoredEdgeRestore,
2699 },
2700}
2701
2702impl ClosureStateRestore {
2703 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
2704 match self {
2705 Self::Clear => None,
2706 Self::Owed { edge, .. } => edge.ordinary_binding_request(),
2707 }
2708 }
2709
2710 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2711 match self {
2712 Self::Clear => None,
2713 Self::Owed { edge, .. } => edge.marker_record_request(),
2714 }
2715 }
2716
2717 pub fn restore(self) -> Result<ClosureState, StorageRestoreError> {
2724 self.restore_with_authority(&MarkerRestoreAuthority::Absent, None)
2725 }
2726
2727 pub(super) fn restore_with_marker_record(
2740 self,
2741 conversation_id: ConversationId,
2742 record: ValidatedMarkerRecord,
2743 ) -> Result<ClosureState, StorageRestoreError> {
2744 let restored = self.restore_with_authority(
2745 &MarkerRestoreAuthority::Record {
2746 conversation_id,
2747 record: &record,
2748 },
2749 None,
2750 );
2751 record.consume();
2752 restored
2753 }
2754
2755 fn restore_with_binding_origin(
2756 self,
2757 origin: &BindingOrigin,
2758 ) -> Result<ClosureState, StorageRestoreError> {
2759 self.restore_with_authority(&MarkerRestoreAuthority::Absent, Some(origin))
2760 }
2761
2762 fn restore_with_authority(
2763 self,
2764 marker_authority: &MarkerRestoreAuthority<'_>,
2765 ordinary_origin: Option<&BindingOrigin>,
2766 ) -> Result<ClosureState, StorageRestoreError> {
2767 match self {
2768 Self::Clear => {
2769 marker_authority.require_absent()?;
2770 Ok(ClosureState::Clear)
2771 }
2772 Self::Owed { debt, edge } => {
2773 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
2774 Ok(ClosureState::Owed {
2775 debt,
2776 edge: edge.restore_with_debt(debt, marker_authority, ordinary_origin)?,
2777 })
2778 }
2779 }
2780 }
2781}
2782
2783fn optional_debt(
2784 raw: Option<WideResourceVector>,
2785) -> Result<Option<ClosureDebt>, StorageRestoreError> {
2786 raw.map(|value| ClosureDebt::new(value).ok_or(StorageRestoreError::ClosureDebt))
2787 .transpose()
2788}
2789
2790impl OrdinaryBindingFateRestore {
2791 #[must_use]
2793 pub fn encode_checkpoint(&self) -> alloc::vec::Vec<u8> {
2794 super::checkpoint_codec::encode_ordinary_fate(self)
2795 }
2796
2797 pub fn restore_checkpoint(
2803 body: &[u8],
2804 origin: &BindingOrigin,
2805 ) -> Result<OrdinaryBindingFate, super::OwnerCheckpointRestoreError> {
2806 super::checkpoint_codec::decode_ordinary_fate(body)?
2807 .restore_with_origin(origin)
2808 .map_err(super::OwnerCheckpointRestoreError::Provenance)
2809 }
2810}