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, EmptyDetach,
21 EnrollmentFingerprint, Event, FencedAttachCommit, FrontierBinding, IdentityState,
22 LeaveFingerprint, LiveMember, LiveMemberRestore, MarkerDelivery, NonzeroDebtCursorEpisode,
23 ObserverProjection, OrderLedger, OrdinaryBindingAuthority, OrdinaryBindingFate,
24 ParticipantCursorProgress, PendingFinalization, PendingRecoveredCursorRelease,
25 PhysicalCompaction, RecoveredBindingFate, RecoveredBindingFateTransition, RetiredIdentity,
26 SequenceLedger, StoredEdge,
27 binding::{restore_committed_terminal, restore_pending_finalization},
28 claim_frontier::{
29 HistoricalCausalAuthority, MarkerRecordOccurrence, MarkerRecordRequest,
30 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}
65
66#[derive(Clone, Debug, PartialEq, Eq)]
68pub enum ConversationStateRestoreError {
69 ClaimFrontier(ParticipantStateCorruptReason),
71 Storage(StorageRestoreError),
73}
74
75#[derive(Debug, PartialEq, Eq)]
82pub(super) struct RestoredConversationState {
83 frontiers: ClaimFrontiers,
84 closure: ClosureState,
85}
86
87#[derive(Clone, Debug, PartialEq, Eq)]
101pub struct ParticipantConversationRestore<EF, V, LF, D> {
102 pub participants: Vec<ParticipantLifecycleRestore<EF, V, LF, D>>,
104 pub frontiers: ClaimFrontiersRestore,
106 pub sequence_ledger: SequenceLedger,
108 pub order_ledger: OrderLedger,
110 pub closure: ClosureStateRestore,
112}
113
114#[derive(Debug, PartialEq, Eq)]
133pub struct ParticipantConversationState<EF, V, LF, D> {
134 participants: Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
135 frontiers: ClaimFrontiers,
136 closure: ClosureState,
137}
138
139impl<EF, V, LF, D> ParticipantConversationState<EF, V, LF, D> {
140 #[must_use]
142 pub fn participants(&self) -> &[RestoredParticipantLifecycle<EF, V, LF, D>] {
143 &self.participants
144 }
145
146 #[must_use]
148 pub const fn frontiers(&self) -> &ClaimFrontiers {
149 &self.frontiers
150 }
151
152 #[must_use]
154 pub const fn closure(&self) -> ClosureState {
155 self.closure
156 }
157
158 #[must_use]
165 #[allow(clippy::type_complexity)]
166 pub fn into_parts(
167 self,
168 ) -> (
169 Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
170 ClaimFrontiers,
171 ClosureState,
172 ) {
173 (self.participants, self.frontiers, self.closure)
174 }
175}
176
177impl RestoredConversationState {
178 #[must_use]
180 #[cfg(test)]
181 pub(crate) const fn frontiers(&self) -> &ClaimFrontiers {
182 &self.frontiers
183 }
184
185 #[must_use]
187 #[cfg(test)]
188 pub(crate) const fn closure(&self) -> ClosureState {
189 self.closure
190 }
191
192 #[must_use]
194 pub(crate) fn into_parts(self) -> (ClaimFrontiers, ClosureState) {
195 (self.frontiers, self.closure)
196 }
197}
198
199#[cfg(test)]
217pub(super) fn restore_conversation_state(
218 frontier_restore: ClaimFrontiersRestore,
219 sequence_ledger: SequenceLedger,
220 order_ledger: OrderLedger,
221 closure_restore: &ClosureStateRestore,
222) -> Result<RestoredConversationState, ConversationStateRestoreError> {
223 let history = ValidatedConversationHistory::empty();
224 restore_conversation_with_history(
225 frontier_restore,
226 sequence_ledger,
227 order_ledger,
228 closure_restore,
229 &history,
230 )
231}
232
233fn restore_conversation_with_history(
234 frontier_restore: ClaimFrontiersRestore,
235 sequence_ledger: SequenceLedger,
236 order_ledger: OrderLedger,
237 closure_restore: &ClosureStateRestore,
238 history: &ValidatedConversationHistory,
239) -> Result<RestoredConversationState, ConversationStateRestoreError> {
240 let mut prevalidated = ClaimFrontiers::prevalidate_with_history(
241 frontier_restore,
242 sequence_ledger,
243 order_ledger,
244 history,
245 )
246 .map_err(ConversationStateRestoreError::ClaimFrontier)?;
247 let marker_request = closure_restore.marker_record_request();
248 let ordinary_request = closure_restore.ordinary_binding_request();
249 let closure = match (marker_request, ordinary_request) {
250 (Some(marker_request), None) => {
251 let record = prevalidated.take_marker_record(marker_request).ok_or(
252 ConversationStateRestoreError::Storage(StorageRestoreError::StoredEdgeProvenance),
253 )?;
254 (*closure_restore)
255 .restore_with_marker_record(prevalidated.conversation_id(), record)
256 .map_err(ConversationStateRestoreError::Storage)?
257 }
258 (None, Some(binding)) => {
259 let origin = history
260 .ordinary_origin(
261 binding.conversation_id,
262 binding.participant_id,
263 binding.binding_epoch,
264 )
265 .ok_or(ConversationStateRestoreError::Storage(
266 StorageRestoreError::StoredEdgeProvenance,
267 ))?;
268 (*closure_restore)
269 .restore_with_binding_origin(origin)
270 .map_err(ConversationStateRestoreError::Storage)?
271 }
272 (None, None) => (*closure_restore)
273 .restore()
274 .map_err(ConversationStateRestoreError::Storage)?,
275 (Some(_), Some(_)) => {
276 return Err(ConversationStateRestoreError::Storage(
277 StorageRestoreError::StoredEdgeProvenance,
278 ));
279 }
280 };
281 let current_edge = match closure {
282 ClosureState::Clear => None,
283 ClosureState::Owed { edge, .. } => Some(edge),
284 };
285 let frontiers = prevalidated
286 .finish(current_edge)
287 .map_err(ConversationStateRestoreError::ClaimFrontier)?;
288 Ok(RestoredConversationState { frontiers, closure })
289}
290
291#[derive(Clone, Copy, Debug, PartialEq, Eq)]
293pub struct CommittedBindingTerminalRestore {
294 pub binding: ActiveBinding,
296 pub cause: CloseCause,
298 pub transaction_order: TransactionOrder,
300 pub delivery_seq: DeliverySeq,
302}
303
304impl CommittedBindingTerminalRestore {
305 pub fn restore(self) -> Result<CommittedBindingTerminal, StorageRestoreError> {
312 restore_committed_terminal(
313 self.binding,
314 self.cause,
315 self.transaction_order,
316 self.delivery_seq,
317 )
318 .ok_or(StorageRestoreError::CommittedBindingTerminal)
319 }
320}
321
322#[derive(Clone, Copy, Debug, PartialEq, Eq)]
324pub struct PendingFinalizationRestore {
325 pub binding: ActiveBinding,
327 pub cause: CloseCause,
329 pub transaction_order: TransactionOrder,
331}
332
333impl PendingFinalizationRestore {
334 pub fn restore(self) -> Result<PendingFinalization, StorageRestoreError> {
341 restore_pending_finalization(self.binding, self.cause, self.transaction_order)
342 .ok_or(StorageRestoreError::PendingFinalization)
343 }
344}
345
346#[derive(Clone, Copy, Debug, PartialEq, Eq)]
348pub enum BindingFateTerminalRestore {
349 Committed(CommittedBindingTerminalRestore),
351 Pending(PendingFinalizationRestore),
353}
354
355#[derive(Clone, Copy, Debug, PartialEq, Eq)]
357pub enum RestoredBindingFateTerminal {
358 Committed(CommittedBindingTerminal),
360 Pending(PendingFinalization),
362}
363
364impl BindingFateTerminalRestore {
365 pub fn restore(self) -> Result<RestoredBindingFateTerminal, StorageRestoreError> {
371 match self {
372 Self::Committed(value) => value.restore().map(RestoredBindingFateTerminal::Committed),
373 Self::Pending(value) => value.restore().map(RestoredBindingFateTerminal::Pending),
374 }
375 }
376}
377
378impl RestoredBindingFateTerminal {
379 const fn participant_id(self) -> ParticipantId {
380 match self {
381 Self::Committed(value) => value.participant_id(),
382 Self::Pending(value) => value.participant_id(),
383 }
384 }
385
386 const fn conversation_id(self) -> ConversationId {
387 match self {
388 Self::Committed(value) => value.conversation_id(),
389 Self::Pending(value) => value.conversation_id(),
390 }
391 }
392
393 const fn binding_epoch(self) -> BindingEpoch {
394 match self {
395 Self::Committed(value) => value.binding_epoch(),
396 Self::Pending(value) => value.binding_epoch(),
397 }
398 }
399}
400
401#[derive(Clone, Copy, Debug, PartialEq, Eq)]
403pub enum BindingStateRestore {
404 Detached,
406 Bound(ActiveBinding),
408 PendingFinalization(PendingFinalizationRestore),
410}
411
412impl BindingStateRestore {
413 fn restore_for<EF>(self, member: &LiveMember<EF>) -> Result<BindingState, StorageRestoreError> {
414 let state = match self {
415 Self::Detached => BindingState::Detached,
416 Self::Bound(binding) => BindingState::Bound(binding),
417 Self::PendingFinalization(raw) => BindingState::PendingFinalization(raw.restore()?),
418 };
419 let authority_matches = match state {
420 BindingState::Detached => true,
421 BindingState::Bound(binding) => {
422 binding.participant_id == member.participant_id()
423 && binding.conversation_id == member.conversation_id()
424 && binding.binding_epoch.capability_generation == member.generation()
425 }
426 BindingState::PendingFinalization(pending) => {
427 pending.participant_id() == member.participant_id()
428 && pending.conversation_id() == member.conversation_id()
429 && pending.binding_epoch().capability_generation == member.generation()
430 }
431 };
432 if authority_matches {
433 Ok(state)
434 } else {
435 Err(StorageRestoreError::BindingAuthority)
436 }
437 }
438}
439
440#[derive(Clone, Debug, PartialEq, Eq)]
442pub struct LiveIdentityRestore<EF> {
443 pub participant_id: ParticipantId,
445 pub conversation_id: ConversationId,
447 pub generation: Generation,
449 pub attach_secret: AttachSecret,
451 pub cursor: DeliverySeq,
453 pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
455 pub latest_terminal: Option<CommittedBindingTerminalRestore>,
457}
458
459impl<EF> LiveIdentityRestore<EF> {
460 fn restore(self) -> Result<LiveMember<EF>, StorageRestoreError> {
461 let latest_terminal = self
462 .latest_terminal
463 .map(CommittedBindingTerminalRestore::restore)
464 .transpose()?;
465 LiveMember::restore(LiveMemberRestore {
466 participant_id: self.participant_id,
467 conversation_id: self.conversation_id,
468 generation: self.generation,
469 attach_secret: self.attach_secret,
470 cursor: self.cursor,
471 enrollment_fingerprint: self.enrollment_fingerprint,
472 latest_terminal,
473 })
474 .map_err(|_| StorageRestoreError::MembershipInvariant)
475 }
476}
477
478#[derive(Clone, Copy, Debug, PartialEq, Eq)]
480pub struct LeaveCommittedRestore {
481 pub conversation_id: ConversationId,
483 pub leave_attempt_token: LeaveAttemptToken,
485 pub participant_id: ParticipantId,
487 pub retired_generation: Generation,
489 pub ended_binding_epoch: Option<BindingEpoch>,
491 pub prior_terminal_delivery_seq: Option<DeliverySeq>,
493 pub left_delivery_seq: DeliverySeq,
495}
496
497impl LeaveCommittedRestore {
498 pub fn restore(self) -> Result<LeaveCommitted, StorageRestoreError> {
505 LeaveCommitted::new(
506 self.conversation_id,
507 self.leave_attempt_token,
508 self.participant_id,
509 self.retired_generation,
510 self.ended_binding_epoch,
511 self.prior_terminal_delivery_seq,
512 self.left_delivery_seq,
513 )
514 .ok_or(StorageRestoreError::LeaveResult)
515 }
516}
517
518#[derive(Clone, Debug, PartialEq, Eq)]
520pub struct RetiredIdentityRestore<EF, V, LF> {
521 pub participant_id: ParticipantId,
523 pub conversation_id: ConversationId,
525 pub retired_generation: Generation,
527 pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
529 pub leave_attempt_token: LeaveAttemptToken,
531 pub leave_request_verifier: V,
533 pub leave_fingerprint: LeaveFingerprint<LF>,
535 pub left_transaction_order: TransactionOrder,
537 pub committed_result: LeaveCommittedRestore,
539}
540
541impl<EF, V, LF> RetiredIdentityRestore<EF, V, LF> {
542 fn restore(self) -> Result<RetiredIdentity<EF, V, LF>, StorageRestoreError> {
543 let result = self.committed_result.restore()?;
544 RetiredIdentity::restore(
545 self.participant_id,
546 self.conversation_id,
547 self.retired_generation,
548 self.enrollment_fingerprint,
549 self.leave_attempt_token,
550 self.leave_request_verifier,
551 self.leave_fingerprint,
552 self.left_transaction_order,
553 result,
554 )
555 .map_err(|_| StorageRestoreError::RetiredIdentity)
556 }
557}
558
559#[derive(Clone, Debug, PartialEq, Eq)]
561pub enum DetachCellRestore<V> {
562 Empty,
564 Pending {
566 token: DetachAttemptToken,
568 participant_id: ParticipantId,
570 request_generation: Generation,
572 request_verifier: V,
574 committed_binding_epoch: BindingEpoch,
576 admission_order: AdmissionOrder,
578 refused_epoch: ObserverEpoch,
580 },
581 Committed {
583 token: DetachAttemptToken,
585 participant_id: ParticipantId,
587 request_generation: Generation,
589 request_verifier: V,
591 committed_binding_epoch: BindingEpoch,
593 detached_delivery_seq: DeliverySeq,
595 },
596 Terminalized {
598 token: DetachAttemptToken,
600 participant_id: ParticipantId,
602 request_generation: Generation,
604 request_verifier: V,
606 committed_binding_epoch: BindingEpoch,
608 },
609}
610
611impl<V> DetachCellRestore<V> {
612 fn restore(self) -> Result<DetachCell<V>, StorageRestoreError> {
613 match self {
614 Self::Empty => Ok(DetachCell::Empty(EmptyDetach)),
615 Self::Pending {
616 token,
617 participant_id,
618 request_generation,
619 request_verifier,
620 committed_binding_epoch,
621 admission_order,
622 refused_epoch,
623 } => restore_pending_detach(
624 token,
625 participant_id,
626 request_generation,
627 request_verifier,
628 committed_binding_epoch,
629 admission_order,
630 refused_epoch,
631 )
632 .map(DetachCell::Pending)
633 .ok_or(StorageRestoreError::DetachCell),
634 Self::Committed {
635 token,
636 participant_id,
637 request_generation,
638 request_verifier,
639 committed_binding_epoch,
640 detached_delivery_seq,
641 } => restore_committed_detach(
642 token,
643 participant_id,
644 request_generation,
645 request_verifier,
646 committed_binding_epoch,
647 detached_delivery_seq,
648 )
649 .map(DetachCell::Committed)
650 .ok_or(StorageRestoreError::DetachCell),
651 Self::Terminalized {
652 token,
653 participant_id,
654 request_generation,
655 request_verifier,
656 committed_binding_epoch,
657 } => restore_terminalized_detach(
658 token,
659 participant_id,
660 request_generation,
661 request_verifier,
662 committed_binding_epoch,
663 )
664 .map(DetachCell::Terminalized)
665 .ok_or(StorageRestoreError::DetachCell),
666 }
667 }
668}
669
670#[allow(
672 clippy::large_enum_variant,
673 reason = "the live storage capsule remains inline so its atomic slots cannot be restored separately"
674)]
675#[derive(Clone, Debug, PartialEq, Eq)]
676pub enum ParticipantLifecycleRestore<EF, V, LF, D> {
677 Live {
679 identity: LiveIdentityRestore<EF>,
681 binding: BindingStateRestore,
683 binding_origin: Option<BindingOrigin>,
685 detach_cell: DetachCellRestore<D>,
687 },
688 Retired(RetiredIdentityRestore<EF, V, LF>),
690}
691
692#[allow(
694 clippy::large_enum_variant,
695 reason = "the validated live capsule remains inline as one atomic lifecycle result"
696)]
697#[derive(Clone, Debug, PartialEq, Eq)]
698pub enum RestoredParticipantLifecycle<EF, V, LF, D> {
699 Live {
701 member: LiveMember<EF>,
703 binding: BindingState,
705 binding_origin: Option<BindingOrigin>,
707 detach_cell: DetachCell<D>,
709 },
710 Retired(RetiredIdentity<EF, V, LF>),
712}
713
714impl<EF, V, LF, D> ParticipantLifecycleRestore<EF, V, LF, D> {
715 pub fn restore(
722 self,
723 ) -> Result<RestoredParticipantLifecycle<EF, V, LF, D>, StorageRestoreError> {
724 match self {
725 Self::Retired(identity) => identity
726 .restore()
727 .map(RestoredParticipantLifecycle::Retired),
728 Self::Live {
729 identity,
730 binding,
731 binding_origin,
732 detach_cell,
733 } => {
734 let member = identity.restore()?;
735 let binding = binding.restore_for(&member)?;
736 let binding_origin = binding_origin
737 .map(|origin| validate_binding_origin(origin, &member, binding))
738 .transpose()?;
739 let origin_required = !matches!(binding, BindingState::Detached)
740 || member.latest_terminal().is_some();
741 if origin_required != binding_origin.is_some() {
742 return Err(StorageRestoreError::BindingAuthority);
743 }
744 let detach_cell = detach_cell.restore()?;
745 validate_live_pair(&member, binding, &detach_cell)?;
746 Ok(RestoredParticipantLifecycle::Live {
747 member,
748 binding,
749 binding_origin,
750 detach_cell,
751 })
752 }
753 }
754 }
755}
756
757fn validate_binding_origin<EF>(
758 origin: BindingOrigin,
759 member: &LiveMember<EF>,
760 binding_state: BindingState,
761) -> Result<BindingOrigin, StorageRestoreError> {
762 let expected_epoch = match binding_state {
763 BindingState::Bound(current) => Some(current.binding_epoch),
764 BindingState::PendingFinalization(pending) => Some(pending.binding_epoch()),
765 BindingState::Detached => member
766 .latest_terminal()
767 .map(CommittedBindingTerminal::binding_epoch),
768 };
769 let attached = origin.attached();
770 if origin.participant_id() != member.participant_id()
771 || origin.conversation_id() != member.conversation_id()
772 || expected_epoch != Some(origin.binding_epoch())
773 || attached.participant_id() != member.participant_id()
774 || attached.conversation_id() != member.conversation_id()
775 {
776 Err(StorageRestoreError::BindingAuthority)
777 } else {
778 Ok(origin)
779 }
780}
781
782impl<EF, V, LF, D> RestoredParticipantLifecycle<EF, V, LF, D> {
783 #[must_use]
785 #[allow(clippy::type_complexity)]
786 pub fn into_parts(
787 self,
788 ) -> (
789 IdentityState<EF, V, LF>,
790 Option<BindingState>,
791 Option<DetachCell<D>>,
792 ) {
793 match self {
794 Self::Live {
795 member,
796 binding,
797 binding_origin: _,
798 detach_cell,
799 } => (
800 IdentityState::Live(member),
801 Some(binding),
802 Some(detach_cell),
803 ),
804 Self::Retired(identity) => (IdentityState::Retired(identity), None, None),
805 }
806 }
807}
808
809impl<EF, V, LF, D> ParticipantConversationRestore<EF, V, LF, D> {
810 pub fn restore(
825 self,
826 ) -> Result<ParticipantConversationState<EF, V, LF, D>, ConversationStateRestoreError> {
827 let participants = self
828 .participants
829 .into_iter()
830 .map(ParticipantLifecycleRestore::restore)
831 .collect::<Result<Vec<_>, _>>()
832 .map_err(ConversationStateRestoreError::Storage)?;
833 validate_participant_frontier_projection(
834 &participants,
835 self.frontiers.conversation_id,
836 &self.frontiers.active_identities,
837 )?;
838 let history = validated_conversation_history(&participants)?;
839 let restored = restore_conversation_with_history(
840 self.frontiers,
841 self.sequence_ledger,
842 self.order_ledger,
843 &self.closure,
844 &history,
845 )?;
846 let (frontiers, closure) = restored.into_parts();
847 Ok(ParticipantConversationState {
848 participants,
849 frontiers,
850 closure,
851 })
852 }
853}
854
855fn validated_conversation_history<EF, V, LF, D>(
856 participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
857) -> Result<ValidatedConversationHistory, ConversationStateRestoreError> {
858 let mut causal_authorities = Vec::new();
859 let mut binding_origins = Vec::new();
860 let mut seen = Vec::with_capacity(participants.len());
861 for participant in participants {
862 let participant_id = match participant {
863 RestoredParticipantLifecycle::Live {
864 member,
865 binding_origin,
866 ..
867 } => {
868 if let Some(terminal) = member.latest_terminal() {
869 causal_authorities
870 .push(HistoricalCausalAuthority::from_committed_terminal(terminal));
871 }
872 if let Some(origin) = binding_origin {
873 binding_origins.push(*origin);
874 }
875 member.participant_id()
876 }
877 RestoredParticipantLifecycle::Retired(retired) => {
878 causal_authorities.push(HistoricalCausalAuthority::from_retired(retired));
879 retired.participant_id()
880 }
881 };
882 seen.push(participant_id);
883 }
884 seen.sort_unstable();
888 let has_duplicate = seen
889 .iter()
890 .zip(seen.iter().skip(1))
891 .any(|(current, next)| current == next);
892 if has_duplicate {
893 return Err(ConversationStateRestoreError::Storage(
894 StorageRestoreError::MembershipInvariant,
895 ));
896 }
897 Ok(ValidatedConversationHistory::new(
898 causal_authorities,
899 binding_origins,
900 ))
901}
902
903fn validate_participant_frontier_projection<EF, V, LF, D>(
904 participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
905 conversation_id: ConversationId,
906 frontier: &[super::FrontierParticipant],
907) -> Result<(), ConversationStateRestoreError> {
908 let mut projected = Vec::new();
909 for participant in participants {
910 match participant {
911 RestoredParticipantLifecycle::Live {
912 member, binding, ..
913 } => {
914 if member.conversation_id() != conversation_id {
915 return Err(ConversationStateRestoreError::Storage(
916 StorageRestoreError::MembershipInvariant,
917 ));
918 }
919 let binding = match binding {
920 BindingState::Bound(binding) => FrontierBinding::Bound(binding.binding_epoch),
921 BindingState::PendingFinalization(pending) => {
922 FrontierBinding::Detached(pending.binding_epoch())
923 }
924 BindingState::Detached => {
925 let Some(terminal) = member.latest_terminal() else {
926 return Err(ConversationStateRestoreError::Storage(
927 StorageRestoreError::BindingAuthority,
928 ));
929 };
930 FrontierBinding::Detached(terminal.binding_epoch())
931 }
932 };
933 projected.push(super::FrontierParticipant::new(
934 member.participant_id(),
935 member.cursor(),
936 binding,
937 ));
938 }
939 RestoredParticipantLifecycle::Retired(retired) => {
940 if retired.conversation_id() != conversation_id {
941 return Err(ConversationStateRestoreError::Storage(
942 StorageRestoreError::MembershipInvariant,
943 ));
944 }
945 }
946 }
947 }
948 projected.sort_by_key(|participant| participant.participant_index());
949 if projected == frontier {
950 Ok(())
951 } else {
952 Err(ConversationStateRestoreError::Storage(
953 StorageRestoreError::MembershipInvariant,
954 ))
955 }
956}
957
958#[allow(clippy::suspicious_operation_groupings)]
959fn validate_live_pair<EF, D>(
960 member: &LiveMember<EF>,
961 binding: BindingState,
962 detach_cell: &DetachCell<D>,
963) -> Result<(), StorageRestoreError> {
964 match detach_cell {
965 DetachCell::Empty(_) => Ok(()),
966 DetachCell::Pending(cell) => {
967 if cell.participant_id() != member.participant_id()
968 || cell.request_generation() != member.generation()
969 || validate_pending_pair(binding, cell, Some(member.conversation_id())).is_err()
970 {
971 Err(StorageRestoreError::DetachBindingPair)
972 } else {
973 Ok(())
974 }
975 }
976 DetachCell::Committed(cell) => {
977 let terminal_matches = member.latest_terminal().is_some_and(|terminal| {
978 terminal.participant_id() == cell.participant_id()
979 && terminal.conversation_id() == member.conversation_id()
980 && terminal.binding_epoch() == cell.committed_binding_epoch()
981 && terminal.delivery_seq() == cell.detached_delivery_seq()
982 && terminal.detached_cause()
983 == Some(crate::wire::DetachedCause::CleanDeregister)
984 });
985 if cell.participant_id() != member.participant_id()
986 || cell.request_generation() != member.generation()
987 || binding != BindingState::Detached
988 || !terminal_matches
989 {
990 Err(StorageRestoreError::DetachBindingPair)
991 } else {
992 Ok(())
993 }
994 }
995 DetachCell::Terminalized(cell) => {
996 if cell.participant_id() != member.participant_id()
997 || cell.request_generation().get() >= member.generation().get()
998 {
999 Err(StorageRestoreError::DetachBindingPair)
1000 } else {
1001 Ok(())
1002 }
1003 }
1004 }
1005}
1006
1007#[derive(Clone, Debug, PartialEq, Eq)]
1009pub struct CursorEpisodeRestore {
1010 pub conversation_id: ConversationId,
1012 pub debt: WideResourceVector,
1014 pub observer_progress: DeliverySeq,
1016 pub candidate_high_watermark: DeliverySeq,
1018 pub current_floor: u128,
1020 pub cap_floor: u128,
1022 pub participants: Vec<BoundParticipantCursor>,
1024 pub facts: Vec<(CursorProgressKey, CursorProgressFact)>,
1026}
1027
1028impl CursorEpisodeRestore {
1029 pub fn restore(self) -> Result<NonzeroDebtCursorEpisode, StorageRestoreError> {
1037 let debt = ClosureDebt::new(self.debt).ok_or(StorageRestoreError::ClosureDebt)?;
1038 NonzeroDebtCursorEpisode::restore(
1039 self.conversation_id,
1040 debt,
1041 self.observer_progress,
1042 self.candidate_high_watermark,
1043 self.current_floor,
1044 self.cap_floor,
1045 self.participants,
1046 self.facts,
1047 )
1048 .ok_or(StorageRestoreError::CursorEpisode)
1049 }
1050}
1051
1052#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1066pub struct OrdinaryBindingAuthorityRestore {
1067 pub binding: ActiveBinding,
1069 pub through_seq: DeliverySeq,
1071}
1072
1073impl OrdinaryBindingAuthorityRestore {
1074 fn restore_with_origin(
1075 self,
1076 origin: &BindingOrigin,
1077 ) -> Result<OrdinaryBindingAuthority, StorageRestoreError> {
1078 if !origin.is_unfenced()
1079 || origin.conversation_id() != self.binding.conversation_id
1080 || origin.participant_id() != self.binding.participant_id
1081 || origin.binding_epoch() != self.binding.binding_epoch
1082 {
1083 return Err(StorageRestoreError::StoredEdgeProvenance);
1084 }
1085 Ok(OrdinaryBindingAuthority::new(
1086 self.binding,
1087 self.through_seq,
1088 ))
1089 }
1090}
1091
1092#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1113pub struct MarkerDeliveryRestore {
1114 pub participant_id: ParticipantId,
1116 pub binding_epoch: BindingEpoch,
1118 pub marker_delivery_seq: DeliverySeq,
1120}
1121
1122impl MarkerDeliveryRestore {
1123 #[cfg(any(test, feature = "test-support"))]
1131 pub(super) fn restore_bound(
1132 self,
1133 conversation_id: ConversationId,
1134 record_authority: ValidatedMarkerRecord,
1135 ) -> Result<MarkerDelivery, StorageRestoreError> {
1136 let restored = self.restore_with_target(
1137 conversation_id,
1138 &record_authority,
1139 MarkerRecordTarget::Bound,
1140 MarkerRecordOccurrence::Undelivered,
1141 );
1142 record_authority.consume();
1143 restored
1144 }
1145
1146 #[cfg(test)]
1148 pub(super) fn restore_detached_delivered_for_test(
1149 self,
1150 conversation_id: ConversationId,
1151 record_authority: ValidatedMarkerRecord,
1152 ) -> Result<MarkerDelivery, StorageRestoreError> {
1153 let restored = self.restore_with_target(
1154 conversation_id,
1155 &record_authority,
1156 MarkerRecordTarget::Detached,
1157 MarkerRecordOccurrence::Delivered,
1158 );
1159 record_authority.consume();
1160 restored
1161 }
1162
1163 fn restore_detached(
1164 self,
1165 conversation_id: ConversationId,
1166 record_authority: &ValidatedMarkerRecord,
1167 ) -> Result<MarkerDelivery, StorageRestoreError> {
1168 self.restore_with_target(
1169 conversation_id,
1170 record_authority,
1171 MarkerRecordTarget::Detached,
1172 MarkerRecordOccurrence::Undelivered,
1173 )
1174 }
1175
1176 fn restore_with_target(
1177 self,
1178 conversation_id: ConversationId,
1179 record_authority: &ValidatedMarkerRecord,
1180 target: MarkerRecordTarget,
1181 occurrence: MarkerRecordOccurrence,
1182 ) -> Result<MarkerDelivery, StorageRestoreError> {
1183 if record_authority.conversation_id() != conversation_id
1184 || !target.matches(record_authority.target_binding(), self.binding_epoch)
1185 || record_authority.occurrence() != occurrence
1186 {
1187 return Err(StorageRestoreError::StoredEdgeProvenance);
1188 }
1189 let delivery = MarkerDelivery::from_validated_record(record_authority);
1190 if delivery.participant_id() != self.participant_id
1191 || delivery.binding_epoch() != self.binding_epoch
1192 || delivery.marker_delivery_seq() != self.marker_delivery_seq
1193 {
1194 return Err(StorageRestoreError::StoredEdgeProvenance);
1195 }
1196 Ok(delivery)
1197 }
1198}
1199
1200#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1201enum MarkerRecordTarget {
1202 Bound,
1203 Detached,
1204}
1205
1206impl MarkerRecordTarget {
1207 fn matches(self, target: super::FrontierBinding, epoch: BindingEpoch) -> bool {
1208 matches!(
1209 (self, target),
1210 (Self::Bound, super::FrontierBinding::Bound(actual))
1211 | (Self::Detached, super::FrontierBinding::Detached(actual))
1212 if actual == epoch
1213 )
1214 }
1215}
1216
1217#[derive(Debug)]
1218enum MarkerRestoreAuthority<'a> {
1219 Absent,
1220 Record {
1221 conversation_id: ConversationId,
1222 record: &'a ValidatedMarkerRecord,
1223 },
1224}
1225
1226impl MarkerRestoreAuthority<'_> {
1227 const fn require_absent(&self) -> Result<(), StorageRestoreError> {
1228 match self {
1229 Self::Absent => Ok(()),
1230 Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
1231 }
1232 }
1233
1234 const fn require_record(
1235 &self,
1236 ) -> Result<(ConversationId, &ValidatedMarkerRecord), StorageRestoreError> {
1237 match self {
1238 Self::Record {
1239 conversation_id,
1240 record,
1241 } if record.conversation_id() == *conversation_id => Ok((*conversation_id, record)),
1242 Self::Absent | Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
1243 }
1244 }
1245
1246 fn record_for(
1247 &self,
1248 expected_conversation_id: ConversationId,
1249 ) -> Result<&ValidatedMarkerRecord, StorageRestoreError> {
1250 let (conversation_id, record) = self.require_record()?;
1251 if conversation_id == expected_conversation_id {
1252 Ok(record)
1253 } else {
1254 Err(StorageRestoreError::StoredEdgeProvenance)
1255 }
1256 }
1257}
1258
1259#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1261pub struct MarkerCursorProgressRestore {
1262 pub conversation_id: ConversationId,
1264 pub participant_id: ParticipantId,
1266 pub binding_epoch: BindingEpoch,
1268 pub through_seq: DeliverySeq,
1270 pub marker_delivery_seq: DeliverySeq,
1272 pub delivery: MarkerDeliveryRestore,
1274}
1275
1276impl MarkerCursorProgressRestore {
1277 fn restore_with_debt(
1278 self,
1279 debt: ClosureDebt,
1280 record_authority: &ValidatedMarkerRecord,
1281 target: MarkerRecordTarget,
1282 ) -> Result<ParticipantCursorProgress, StorageRestoreError> {
1283 if self.participant_id != self.delivery.participant_id
1284 || self.binding_epoch != self.delivery.binding_epoch
1285 || self.marker_delivery_seq != self.delivery.marker_delivery_seq
1286 || self.through_seq != self.marker_delivery_seq
1287 {
1288 return Err(StorageRestoreError::StoredEdgeProvenance);
1289 }
1290 let marker = self.delivery.restore_with_target(
1291 self.conversation_id,
1292 record_authority,
1293 target,
1294 MarkerRecordOccurrence::Delivered,
1295 )?;
1296 let state = marker
1297 .delivered(
1298 debt,
1299 Event::marker_delivered(
1300 self.participant_id,
1301 self.binding_epoch,
1302 self.marker_delivery_seq,
1303 ),
1304 )
1305 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1306 match state {
1307 ClosureState::Owed {
1308 edge: StoredEdge::ParticipantCursorProgress(progress),
1309 ..
1310 } => Ok(progress),
1311 ClosureState::Clear | ClosureState::Owed { .. } => {
1312 Err(StorageRestoreError::StoredEdgeProvenance)
1313 }
1314 }
1315 }
1316}
1317
1318#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1320pub struct DetachedCredentialRecoveryRestore {
1321 pub participant_id: ParticipantId,
1323 pub marker_delivery_seq: DeliverySeq,
1325 pub prior_binding_epoch: BindingEpoch,
1327 pub resulting_floor: DeliverySeq,
1329 pub terminal: BindingFateTerminalRestore,
1331 pub progress: MarkerCursorProgressRestore,
1333}
1334
1335impl DetachedCredentialRecoveryRestore {
1336 pub fn restore_description(self) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
1348 if self.participant_id != self.progress.participant_id
1349 || self.marker_delivery_seq != self.progress.marker_delivery_seq
1350 || self.prior_binding_epoch != self.progress.binding_epoch
1351 || self.progress.through_seq != self.marker_delivery_seq
1352 || self.progress.participant_id != self.progress.delivery.participant_id
1353 || self.progress.binding_epoch != self.progress.delivery.binding_epoch
1354 || self.progress.marker_delivery_seq != self.progress.delivery.marker_delivery_seq
1355 {
1356 return Err(StorageRestoreError::StoredEdgeProvenance);
1357 }
1358 let terminal = self.terminal.restore()?;
1359 if terminal.participant_id() != self.participant_id
1360 || terminal.binding_epoch() != self.prior_binding_epoch
1361 || terminal.conversation_id() != self.progress.conversation_id
1362 {
1363 return Err(StorageRestoreError::StoredEdgeProvenance);
1364 }
1365 Ok(DetachedCredentialRecovery::from_storage_description(
1366 self.progress.conversation_id,
1367 self.participant_id,
1368 self.marker_delivery_seq,
1369 self.prior_binding_epoch,
1370 ))
1371 }
1372
1373 fn restore_with_debt(
1374 self,
1375 debt: ClosureDebt,
1376 record_authority: &ValidatedMarkerRecord,
1377 ) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
1378 if self.participant_id != self.progress.participant_id
1379 || self.marker_delivery_seq != self.progress.marker_delivery_seq
1380 || self.prior_binding_epoch != self.progress.binding_epoch
1381 {
1382 return Err(StorageRestoreError::StoredEdgeProvenance);
1383 }
1384 let terminal = self.terminal.restore()?;
1385 if terminal.participant_id() != self.participant_id
1386 || terminal.binding_epoch() != self.prior_binding_epoch
1387 || terminal.conversation_id() != self.progress.conversation_id
1388 {
1389 return Err(StorageRestoreError::StoredEdgeProvenance);
1390 }
1391 let progress = self.progress.restore_with_debt(
1392 debt,
1393 record_authority,
1394 MarkerRecordTarget::Detached,
1395 )?;
1396 let successor = progress
1397 .binding_fate(
1398 debt,
1399 Event::binding_fate_observed(
1400 self.participant_id,
1401 self.prior_binding_epoch,
1402 self.resulting_floor,
1403 ),
1404 )
1405 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1406 match successor.into_stored_edge() {
1407 StoredEdge::DetachedCredentialRecovery(edge) => Ok(edge),
1408 _ => Err(StorageRestoreError::StoredEdgeProvenance),
1409 }
1410 }
1411}
1412
1413#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1415pub struct DetachedMarkerReleaseRestore {
1416 pub conversation_id: ConversationId,
1418 pub participant_id: ParticipantId,
1420 pub marker_delivery_seq: DeliverySeq,
1422 pub last_dead_binding_epoch: BindingEpoch,
1424 pub resulting_floor: DeliverySeq,
1426 pub terminal: BindingFateTerminalRestore,
1428 pub delivery: MarkerDeliveryRestore,
1430}
1431
1432impl DetachedMarkerReleaseRestore {
1433 fn restore_with_debt(
1434 self,
1435 debt: ClosureDebt,
1436 record_authority: &ValidatedMarkerRecord,
1437 ) -> Result<DetachedMarkerRelease, StorageRestoreError> {
1438 if self.participant_id != self.delivery.participant_id
1439 || self.marker_delivery_seq != self.delivery.marker_delivery_seq
1440 || self.last_dead_binding_epoch != self.delivery.binding_epoch
1441 {
1442 return Err(StorageRestoreError::StoredEdgeProvenance);
1443 }
1444 let terminal = self.terminal.restore()?;
1445 if terminal.participant_id() != self.participant_id
1446 || terminal.binding_epoch() != self.last_dead_binding_epoch
1447 || terminal.conversation_id() != self.conversation_id
1448 {
1449 return Err(StorageRestoreError::StoredEdgeProvenance);
1450 }
1451 let marker = self
1452 .delivery
1453 .restore_detached(self.conversation_id, record_authority)?;
1454 let state = marker
1455 .binding_fate(
1456 debt,
1457 Event::binding_fate_observed(
1458 self.participant_id,
1459 self.last_dead_binding_epoch,
1460 self.resulting_floor,
1461 ),
1462 )
1463 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1464 match state {
1465 ClosureState::Owed {
1466 edge: StoredEdge::DetachedMarkerRelease(edge),
1467 ..
1468 } => Ok(edge),
1469 ClosureState::Clear | ClosureState::Owed { .. } => {
1470 Err(StorageRestoreError::StoredEdgeProvenance)
1471 }
1472 }
1473 }
1474}
1475
1476#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1489pub struct OrdinaryBindingFateRestore {
1490 pub authority: OrdinaryBindingAuthorityRestore,
1492 pub terminal: CommittedBindingTerminalRestore,
1494 pub resulting_floor: DeliverySeq,
1496}
1497
1498impl OrdinaryBindingFateRestore {
1499 fn restore_with_origin(
1506 self,
1507 origin: &BindingOrigin,
1508 ) -> Result<OrdinaryBindingFate, StorageRestoreError> {
1509 let authority = self.authority.restore_with_origin(origin)?;
1510 let terminal = self.terminal.restore()?;
1511 let CommittedBindingTerminal::Died(terminal) = terminal else {
1512 return Err(StorageRestoreError::StoredEdgeProvenance);
1513 };
1514 authority
1515 .binding_fate(terminal, self.resulting_floor)
1516 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1517 }
1518}
1519
1520#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1522pub enum DebtCompletionRestore {
1523 Clear,
1525 ObserverProjection {
1527 debt: WideResourceVector,
1529 through_seq: DeliverySeq,
1531 },
1532 PhysicalCompaction {
1534 debt: WideResourceVector,
1536 from_floor: DeliverySeq,
1538 through_seq: DeliverySeq,
1540 },
1541}
1542
1543impl DebtCompletionRestore {
1544 pub fn restore(self) -> Result<DebtCompletion, StorageRestoreError> {
1550 match self {
1551 Self::Clear => Ok(DebtCompletion::clear()),
1552 Self::ObserverProjection { debt, through_seq } => {
1553 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
1554 Ok(DebtCompletion::observer_projection(
1555 debt,
1556 ObserverProjection::new(through_seq),
1557 ))
1558 }
1559 Self::PhysicalCompaction {
1560 debt,
1561 from_floor,
1562 through_seq,
1563 } => {
1564 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
1565 let edge = PhysicalCompaction::new(from_floor, through_seq)
1566 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1567 Ok(DebtCompletion::physical_compaction(debt, edge))
1568 }
1569 }
1570 }
1571}
1572
1573#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1587pub struct FencedAttachCommitRestore {
1588 pub predecessor: DetachedCredentialRecoveryRestore,
1590 pub predecessor_debt: WideResourceVector,
1592 pub participant_id: ParticipantId,
1594 pub marker_delivery_seq: DeliverySeq,
1596 pub prior_binding_epoch: BindingEpoch,
1598 pub new_binding_epoch: BindingEpoch,
1600 pub resulting_floor: DeliverySeq,
1602 pub successor: DebtCompletionRestore,
1604}
1605
1606impl FencedAttachCommitRestore {
1607 #[cfg(test)]
1614 pub(super) fn restore(
1615 self,
1616 record_authority: ValidatedMarkerRecord,
1617 ) -> Result<FencedAttachCommit, StorageRestoreError> {
1618 let restored = self.restore_with_record(&record_authority);
1619 record_authority.consume();
1620 restored
1621 }
1622
1623 fn restore_with_record(
1624 self,
1625 record_authority: &ValidatedMarkerRecord,
1626 ) -> Result<FencedAttachCommit, StorageRestoreError> {
1627 let debt =
1628 ClosureDebt::new(self.predecessor_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1629 let predecessor = self.predecessor.restore_with_debt(debt, record_authority)?;
1630 FencedAttachCommit::restore_validated(
1631 predecessor,
1632 record_authority,
1633 debt,
1634 Event::fenced_recovery_committed(
1635 self.participant_id,
1636 self.marker_delivery_seq,
1637 self.prior_binding_epoch,
1638 self.new_binding_epoch,
1639 self.resulting_floor,
1640 ),
1641 self.successor.restore()?,
1642 )
1643 .ok_or(StorageRestoreError::StoredEdgeProvenance)
1644 }
1645}
1646
1647#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1657pub struct RecoveredBindingFateRestore {
1658 pub fenced_attach: FencedAttachCommitRestore,
1660 pub participant_id: ParticipantId,
1662 pub binding_epoch: BindingEpoch,
1664 pub resulting_floor: DeliverySeq,
1666}
1667
1668impl RecoveredBindingFateRestore {
1669 #[cfg(test)]
1676 pub(super) fn restore(
1677 self,
1678 record_authority: ValidatedMarkerRecord,
1679 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
1680 let restored = self.restore_with_record(&record_authority);
1681 record_authority.consume();
1682 restored
1683 }
1684
1685 fn restore_with_record(
1686 self,
1687 record_authority: &ValidatedMarkerRecord,
1688 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
1689 let commit = self.fenced_attach.restore_with_record(record_authority)?;
1690 commit
1691 .recovered_binding_fate(Event::binding_fate_observed(
1692 self.participant_id,
1693 self.binding_epoch,
1694 self.resulting_floor,
1695 ))
1696 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1697 }
1698}
1699
1700#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1702pub struct PendingRecoveredCursorReleaseRestore {
1703 pub fate: RecoveredBindingFateRestore,
1705 pub resulting_debt: WideResourceVector,
1707}
1708
1709impl PendingRecoveredCursorReleaseRestore {
1710 #[cfg(test)]
1717 pub(super) fn restore(
1718 self,
1719 record_authority: ValidatedMarkerRecord,
1720 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
1721 let restored = self.restore_with_record(&record_authority);
1722 record_authority.consume();
1723 restored
1724 }
1725
1726 fn restore_with_record(
1727 self,
1728 record_authority: &ValidatedMarkerRecord,
1729 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
1730 let authority = self.fate.restore_with_record(record_authority)?;
1731 let predecessor_state = authority.predecessor_state();
1732 let resulting_debt =
1733 ClosureDebt::new(self.resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1734 let transition = match predecessor_state {
1735 ClosureState::Owed {
1736 debt,
1737 edge: StoredEdge::ObserverProjection(edge),
1738 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
1739 ClosureState::Owed {
1740 debt,
1741 edge: StoredEdge::PhysicalCompaction(edge),
1742 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
1743 ClosureState::Clear | ClosureState::Owed { .. } => {
1744 return Err(StorageRestoreError::StoredEdgeProvenance);
1745 }
1746 }
1747 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1748 match transition {
1749 RecoveredBindingFateTransition::PendingStorage(pending) => Ok(pending),
1750 RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
1751 Err(StorageRestoreError::StoredEdgeProvenance)
1752 }
1753 }
1754 }
1755}
1756
1757#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1759pub enum RecoveredStorageCompletionRestore {
1760 ObserverProjection {
1762 through_seq: DeliverySeq,
1764 resulting_debt: Option<WideResourceVector>,
1766 },
1767 PhysicalCompaction {
1769 from_floor: DeliverySeq,
1771 through_seq: DeliverySeq,
1773 resulting_floor: DeliverySeq,
1775 resulting_debt: Option<WideResourceVector>,
1777 },
1778}
1779
1780impl RecoveredStorageCompletionRestore {
1781 fn restore(
1782 self,
1783 pending: PendingRecoveredCursorRelease,
1784 ) -> Result<ClosureState, StorageRestoreError> {
1785 let current = pending.current_state();
1786 match (current, self) {
1787 (
1788 ClosureState::Owed {
1789 edge: StoredEdge::ObserverProjection(edge),
1790 ..
1791 },
1792 Self::ObserverProjection {
1793 through_seq,
1794 resulting_debt,
1795 },
1796 ) => edge
1797 .complete_after_recovered_binding_fate(
1798 Event::projection_completed(through_seq),
1799 optional_debt(resulting_debt)?,
1800 pending,
1801 )
1802 .map_err(|_| StorageRestoreError::StoredEdgeProvenance),
1803 (
1804 ClosureState::Owed {
1805 edge: StoredEdge::PhysicalCompaction(edge),
1806 ..
1807 },
1808 Self::PhysicalCompaction {
1809 from_floor,
1810 through_seq,
1811 resulting_floor,
1812 resulting_debt,
1813 },
1814 ) => {
1815 let event = Event::compaction_completed(from_floor, through_seq, resulting_floor)
1816 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1817 edge.complete_after_recovered_binding_fate(
1818 event,
1819 optional_debt(resulting_debt)?,
1820 pending,
1821 )
1822 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1823 }
1824 _ => Err(StorageRestoreError::StoredEdgeProvenance),
1825 }
1826 }
1827}
1828
1829#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1831pub enum DetachedCursorReleaseProvenanceRestore {
1832 Ordinary(OrdinaryBindingFateRestore),
1834 RecoveredDirect {
1836 fate: RecoveredBindingFateRestore,
1838 resulting_debt: WideResourceVector,
1840 },
1841 RecoveredAfterStorage {
1843 pending: PendingRecoveredCursorReleaseRestore,
1845 completion: RecoveredStorageCompletionRestore,
1847 },
1848}
1849
1850impl DetachedCursorReleaseProvenanceRestore {
1851 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
1852 match self {
1853 Self::Ordinary(fate) => Some(fate.authority.binding),
1854 Self::RecoveredDirect { .. } | Self::RecoveredAfterStorage { .. } => None,
1855 }
1856 }
1857
1858 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
1859 match self {
1860 Self::Ordinary(_) => None,
1861 Self::RecoveredDirect { fate, .. } => Some(MarkerRecordRequest::recovered(
1862 fate.participant_id,
1863 fate.fenced_attach.marker_delivery_seq,
1864 fate.fenced_attach.prior_binding_epoch,
1865 fate.fenced_attach.new_binding_epoch,
1866 )),
1867 Self::RecoveredAfterStorage { pending, .. } => Some(MarkerRecordRequest::recovered(
1868 pending.fate.participant_id,
1869 pending.fate.fenced_attach.marker_delivery_seq,
1870 pending.fate.fenced_attach.prior_binding_epoch,
1871 pending.fate.fenced_attach.new_binding_epoch,
1872 )),
1873 }
1874 }
1875
1876 fn restore_state(
1877 self,
1878 debt: ClosureDebt,
1879 marker_authority: &MarkerRestoreAuthority<'_>,
1880 ordinary_origin: Option<&BindingOrigin>,
1881 ) -> Result<ClosureState, StorageRestoreError> {
1882 match self {
1883 Self::Ordinary(provenance) => {
1884 marker_authority.require_absent()?;
1885 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1886 Ok(provenance
1887 .restore_with_origin(origin)?
1888 .into_direct_state(debt))
1889 }
1890 Self::RecoveredDirect {
1891 fate,
1892 resulting_debt,
1893 } => {
1894 let (_, record_authority) = marker_authority.require_record()?;
1895 let authority = fate.restore_with_record(record_authority)?;
1896 let predecessor = authority.predecessor_state();
1897 let resulting_debt =
1898 ClosureDebt::new(resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1899 let transition = match predecessor {
1900 ClosureState::Owed {
1901 debt: predecessor_debt,
1902 edge: StoredEdge::PhysicalCompaction(edge),
1903 } => edge.apply_recovered_binding_fate(
1904 predecessor_debt,
1905 resulting_debt,
1906 authority,
1907 ),
1908 ClosureState::Clear | ClosureState::Owed { .. } => {
1909 return Err(StorageRestoreError::StoredEdgeProvenance);
1910 }
1911 }
1912 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1913 match transition {
1914 RecoveredBindingFateTransition::DetachedCursorRelease(release)
1915 if release.debt() == debt =>
1916 {
1917 Ok(release.into_state())
1918 }
1919 RecoveredBindingFateTransition::PendingStorage(_)
1920 | RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
1921 Err(StorageRestoreError::StoredEdgeProvenance)
1922 }
1923 }
1924 }
1925 Self::RecoveredAfterStorage {
1926 pending,
1927 completion,
1928 } => {
1929 let (_, record_authority) = marker_authority.require_record()?;
1930 let state = completion.restore(pending.restore_with_record(record_authority)?)?;
1931 match state {
1932 ClosureState::Owed {
1933 debt: restored_debt,
1934 edge: StoredEdge::DetachedCursorRelease(_),
1935 } if restored_debt == debt => Ok(state),
1936 ClosureState::Clear | ClosureState::Owed { .. } => {
1937 Err(StorageRestoreError::StoredEdgeProvenance)
1938 }
1939 }
1940 }
1941 }
1942 }
1943}
1944
1945#[allow(clippy::large_enum_variant)]
1947#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1948pub enum StoredEdgeRestore {
1949 ObserverProjection {
1951 through_seq: DeliverySeq,
1953 },
1954 PhysicalCompaction {
1956 from_floor: DeliverySeq,
1958 through_seq: DeliverySeq,
1960 },
1961 MarkerDelivery(MarkerDeliveryRestore),
1963 ParticipantCursorProgressContinuous {
1965 participant_id: ParticipantId,
1967 binding_epoch: BindingEpoch,
1969 through_seq: DeliverySeq,
1971 authority: OrdinaryBindingAuthorityRestore,
1973 },
1974 ParticipantCursorProgressMarker(MarkerCursorProgressRestore),
1976 DetachedCredentialRecovery(DetachedCredentialRecoveryRestore),
1978 DetachedMarkerRelease(DetachedMarkerReleaseRestore),
1980 DetachedCursorRelease {
1982 participant_id: ParticipantId,
1984 last_dead_binding_epoch: BindingEpoch,
1986 provenance: DetachedCursorReleaseProvenanceRestore,
1988 },
1989}
1990
1991impl StoredEdgeRestore {
1992 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
1993 match self {
1994 Self::ParticipantCursorProgressContinuous { authority, .. } => Some(authority.binding),
1995 Self::DetachedCursorRelease { provenance, .. } => provenance.ordinary_binding_request(),
1996 Self::ObserverProjection { .. }
1997 | Self::PhysicalCompaction { .. }
1998 | Self::MarkerDelivery(_)
1999 | Self::ParticipantCursorProgressMarker(_)
2000 | Self::DetachedCredentialRecovery(_)
2001 | Self::DetachedMarkerRelease(_) => None,
2002 }
2003 }
2004
2005 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2006 match self {
2007 Self::MarkerDelivery(value) => Some(MarkerRecordRequest::planned(
2008 value.participant_id,
2009 value.marker_delivery_seq,
2010 super::FrontierBinding::Bound(value.binding_epoch),
2011 )),
2012 Self::ParticipantCursorProgressMarker(value) => Some(MarkerRecordRequest::delivered(
2013 value.participant_id,
2014 value.marker_delivery_seq,
2015 super::FrontierBinding::Bound(value.binding_epoch),
2016 )),
2017 Self::DetachedCredentialRecovery(value) => Some(MarkerRecordRequest::delivered(
2018 value.participant_id,
2019 value.marker_delivery_seq,
2020 super::FrontierBinding::Detached(value.prior_binding_epoch),
2021 )),
2022 Self::DetachedMarkerRelease(value) => Some(MarkerRecordRequest::planned(
2023 value.participant_id,
2024 value.marker_delivery_seq,
2025 super::FrontierBinding::Detached(value.last_dead_binding_epoch),
2026 )),
2027 Self::DetachedCursorRelease { provenance, .. } => provenance.marker_record_request(),
2028 Self::ObserverProjection { .. }
2029 | Self::PhysicalCompaction { .. }
2030 | Self::ParticipantCursorProgressContinuous { .. } => None,
2031 }
2032 }
2033
2034 fn restore_with_debt(
2035 self,
2036 debt: ClosureDebt,
2037 marker_authority: &MarkerRestoreAuthority<'_>,
2038 ordinary_origin: Option<&BindingOrigin>,
2039 ) -> Result<StoredEdge, StorageRestoreError> {
2040 match self {
2041 Self::ObserverProjection { through_seq } => {
2042 marker_authority.require_absent()?;
2043 Ok(StoredEdge::ObserverProjection(ObserverProjection::new(
2044 through_seq,
2045 )))
2046 }
2047 Self::PhysicalCompaction {
2048 from_floor,
2049 through_seq,
2050 } => {
2051 marker_authority.require_absent()?;
2052 PhysicalCompaction::new(from_floor, through_seq)
2053 .map(StoredEdge::PhysicalCompaction)
2054 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2055 }
2056 Self::MarkerDelivery(value) => {
2057 let (conversation_id, record_authority) = marker_authority.require_record()?;
2058 value
2059 .restore_with_target(
2060 conversation_id,
2061 record_authority,
2062 MarkerRecordTarget::Bound,
2063 MarkerRecordOccurrence::Undelivered,
2064 )
2065 .map(StoredEdge::MarkerDelivery)
2066 }
2067 Self::ParticipantCursorProgressContinuous {
2068 participant_id,
2069 binding_epoch,
2070 through_seq,
2071 authority,
2072 } => {
2073 marker_authority.require_absent()?;
2074 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2075 ParticipantCursorProgress::restore_continuous(
2076 authority.restore_with_origin(origin)?,
2077 participant_id,
2078 binding_epoch,
2079 through_seq,
2080 )
2081 .map(StoredEdge::ParticipantCursorProgress)
2082 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2083 }
2084 Self::ParticipantCursorProgressMarker(value) => {
2085 let record_authority = marker_authority.record_for(value.conversation_id)?;
2086 value
2087 .restore_with_debt(debt, record_authority, MarkerRecordTarget::Bound)
2088 .map(StoredEdge::ParticipantCursorProgress)
2089 }
2090 Self::DetachedCredentialRecovery(value) => {
2091 let record_authority =
2092 marker_authority.record_for(value.progress.conversation_id)?;
2093 value
2094 .restore_with_debt(debt, record_authority)
2095 .map(StoredEdge::DetachedCredentialRecovery)
2096 }
2097 Self::DetachedMarkerRelease(value) => {
2098 let record_authority = marker_authority.record_for(value.conversation_id)?;
2099 value
2100 .restore_with_debt(debt, record_authority)
2101 .map(StoredEdge::DetachedMarkerRelease)
2102 }
2103 Self::DetachedCursorRelease {
2104 participant_id,
2105 last_dead_binding_epoch,
2106 provenance,
2107 } => {
2108 let state = provenance.restore_state(debt, marker_authority, ordinary_origin)?;
2109 match state {
2110 ClosureState::Owed {
2111 edge: StoredEdge::DetachedCursorRelease(edge),
2112 ..
2113 } if edge.participant_id() == participant_id
2114 && edge.last_dead_binding_epoch() == last_dead_binding_epoch =>
2115 {
2116 Ok(StoredEdge::DetachedCursorRelease(edge))
2117 }
2118 ClosureState::Clear | ClosureState::Owed { .. } => {
2119 Err(StorageRestoreError::StoredEdgeProvenance)
2120 }
2121 }
2122 }
2123 }
2124 }
2125}
2126
2127#[allow(clippy::large_enum_variant)]
2129#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2130pub enum ClosureStateRestore {
2131 Clear,
2133 Owed {
2135 debt: WideResourceVector,
2137 edge: StoredEdgeRestore,
2139 },
2140}
2141
2142impl ClosureStateRestore {
2143 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
2144 match self {
2145 Self::Clear => None,
2146 Self::Owed { edge, .. } => edge.ordinary_binding_request(),
2147 }
2148 }
2149
2150 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2151 match self {
2152 Self::Clear => None,
2153 Self::Owed { edge, .. } => edge.marker_record_request(),
2154 }
2155 }
2156
2157 pub fn restore(self) -> Result<ClosureState, StorageRestoreError> {
2164 self.restore_with_authority(&MarkerRestoreAuthority::Absent, None)
2165 }
2166
2167 pub(super) fn restore_with_marker_record(
2180 self,
2181 conversation_id: ConversationId,
2182 record: ValidatedMarkerRecord,
2183 ) -> Result<ClosureState, StorageRestoreError> {
2184 let restored = self.restore_with_authority(
2185 &MarkerRestoreAuthority::Record {
2186 conversation_id,
2187 record: &record,
2188 },
2189 None,
2190 );
2191 record.consume();
2192 restored
2193 }
2194
2195 fn restore_with_binding_origin(
2196 self,
2197 origin: &BindingOrigin,
2198 ) -> Result<ClosureState, StorageRestoreError> {
2199 self.restore_with_authority(&MarkerRestoreAuthority::Absent, Some(origin))
2200 }
2201
2202 fn restore_with_authority(
2203 self,
2204 marker_authority: &MarkerRestoreAuthority<'_>,
2205 ordinary_origin: Option<&BindingOrigin>,
2206 ) -> Result<ClosureState, StorageRestoreError> {
2207 match self {
2208 Self::Clear => {
2209 marker_authority.require_absent()?;
2210 Ok(ClosureState::Clear)
2211 }
2212 Self::Owed { debt, edge } => {
2213 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
2214 Ok(ClosureState::Owed {
2215 debt,
2216 edge: edge.restore_with_debt(debt, marker_authority, ordinary_origin)?,
2217 })
2218 }
2219 }
2220 }
2221}
2222
2223fn optional_debt(
2224 raw: Option<WideResourceVector>,
2225) -> Result<Option<ClosureDebt>, StorageRestoreError> {
2226 raw.map(|value| ClosureDebt::new(value).ok_or(StorageRestoreError::ClosureDebt))
2227 .transpose()
2228}