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(test)]
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 fn restore_with_debt(
1337 self,
1338 debt: ClosureDebt,
1339 record_authority: &ValidatedMarkerRecord,
1340 ) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
1341 if self.participant_id != self.progress.participant_id
1342 || self.marker_delivery_seq != self.progress.marker_delivery_seq
1343 || self.prior_binding_epoch != self.progress.binding_epoch
1344 {
1345 return Err(StorageRestoreError::StoredEdgeProvenance);
1346 }
1347 let terminal = self.terminal.restore()?;
1348 if terminal.participant_id() != self.participant_id
1349 || terminal.binding_epoch() != self.prior_binding_epoch
1350 || terminal.conversation_id() != self.progress.conversation_id
1351 {
1352 return Err(StorageRestoreError::StoredEdgeProvenance);
1353 }
1354 let progress = self.progress.restore_with_debt(
1355 debt,
1356 record_authority,
1357 MarkerRecordTarget::Detached,
1358 )?;
1359 let successor = progress
1360 .binding_fate(
1361 debt,
1362 Event::binding_fate_observed(
1363 self.participant_id,
1364 self.prior_binding_epoch,
1365 self.resulting_floor,
1366 ),
1367 )
1368 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1369 match successor.into_stored_edge() {
1370 StoredEdge::DetachedCredentialRecovery(edge) => Ok(edge),
1371 _ => Err(StorageRestoreError::StoredEdgeProvenance),
1372 }
1373 }
1374}
1375
1376#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1378pub struct DetachedMarkerReleaseRestore {
1379 pub conversation_id: ConversationId,
1381 pub participant_id: ParticipantId,
1383 pub marker_delivery_seq: DeliverySeq,
1385 pub last_dead_binding_epoch: BindingEpoch,
1387 pub resulting_floor: DeliverySeq,
1389 pub terminal: BindingFateTerminalRestore,
1391 pub delivery: MarkerDeliveryRestore,
1393}
1394
1395impl DetachedMarkerReleaseRestore {
1396 fn restore_with_debt(
1397 self,
1398 debt: ClosureDebt,
1399 record_authority: &ValidatedMarkerRecord,
1400 ) -> Result<DetachedMarkerRelease, StorageRestoreError> {
1401 if self.participant_id != self.delivery.participant_id
1402 || self.marker_delivery_seq != self.delivery.marker_delivery_seq
1403 || self.last_dead_binding_epoch != self.delivery.binding_epoch
1404 {
1405 return Err(StorageRestoreError::StoredEdgeProvenance);
1406 }
1407 let terminal = self.terminal.restore()?;
1408 if terminal.participant_id() != self.participant_id
1409 || terminal.binding_epoch() != self.last_dead_binding_epoch
1410 || terminal.conversation_id() != self.conversation_id
1411 {
1412 return Err(StorageRestoreError::StoredEdgeProvenance);
1413 }
1414 let marker = self
1415 .delivery
1416 .restore_detached(self.conversation_id, record_authority)?;
1417 let state = marker
1418 .binding_fate(
1419 debt,
1420 Event::binding_fate_observed(
1421 self.participant_id,
1422 self.last_dead_binding_epoch,
1423 self.resulting_floor,
1424 ),
1425 )
1426 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1427 match state {
1428 ClosureState::Owed {
1429 edge: StoredEdge::DetachedMarkerRelease(edge),
1430 ..
1431 } => Ok(edge),
1432 ClosureState::Clear | ClosureState::Owed { .. } => {
1433 Err(StorageRestoreError::StoredEdgeProvenance)
1434 }
1435 }
1436 }
1437}
1438
1439#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1452pub struct OrdinaryBindingFateRestore {
1453 pub authority: OrdinaryBindingAuthorityRestore,
1455 pub terminal: CommittedBindingTerminalRestore,
1457 pub resulting_floor: DeliverySeq,
1459}
1460
1461impl OrdinaryBindingFateRestore {
1462 fn restore_with_origin(
1469 self,
1470 origin: &BindingOrigin,
1471 ) -> Result<OrdinaryBindingFate, StorageRestoreError> {
1472 let authority = self.authority.restore_with_origin(origin)?;
1473 let terminal = self.terminal.restore()?;
1474 let CommittedBindingTerminal::Died(terminal) = terminal else {
1475 return Err(StorageRestoreError::StoredEdgeProvenance);
1476 };
1477 authority
1478 .binding_fate(terminal, self.resulting_floor)
1479 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1480 }
1481}
1482
1483#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1485pub enum DebtCompletionRestore {
1486 Clear,
1488 ObserverProjection {
1490 debt: WideResourceVector,
1492 through_seq: DeliverySeq,
1494 },
1495 PhysicalCompaction {
1497 debt: WideResourceVector,
1499 from_floor: DeliverySeq,
1501 through_seq: DeliverySeq,
1503 },
1504}
1505
1506impl DebtCompletionRestore {
1507 pub fn restore(self) -> Result<DebtCompletion, StorageRestoreError> {
1513 match self {
1514 Self::Clear => Ok(DebtCompletion::clear()),
1515 Self::ObserverProjection { debt, through_seq } => {
1516 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
1517 Ok(DebtCompletion::observer_projection(
1518 debt,
1519 ObserverProjection::new(through_seq),
1520 ))
1521 }
1522 Self::PhysicalCompaction {
1523 debt,
1524 from_floor,
1525 through_seq,
1526 } => {
1527 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
1528 let edge = PhysicalCompaction::new(from_floor, through_seq)
1529 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1530 Ok(DebtCompletion::physical_compaction(debt, edge))
1531 }
1532 }
1533 }
1534}
1535
1536#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1550pub struct FencedAttachCommitRestore {
1551 pub predecessor: DetachedCredentialRecoveryRestore,
1553 pub predecessor_debt: WideResourceVector,
1555 pub participant_id: ParticipantId,
1557 pub marker_delivery_seq: DeliverySeq,
1559 pub prior_binding_epoch: BindingEpoch,
1561 pub new_binding_epoch: BindingEpoch,
1563 pub resulting_floor: DeliverySeq,
1565 pub successor: DebtCompletionRestore,
1567}
1568
1569impl FencedAttachCommitRestore {
1570 #[cfg(test)]
1577 pub(super) fn restore(
1578 self,
1579 record_authority: ValidatedMarkerRecord,
1580 ) -> Result<FencedAttachCommit, StorageRestoreError> {
1581 let restored = self.restore_with_record(&record_authority);
1582 record_authority.consume();
1583 restored
1584 }
1585
1586 fn restore_with_record(
1587 self,
1588 record_authority: &ValidatedMarkerRecord,
1589 ) -> Result<FencedAttachCommit, StorageRestoreError> {
1590 let debt =
1591 ClosureDebt::new(self.predecessor_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1592 let predecessor = self.predecessor.restore_with_debt(debt, record_authority)?;
1593 predecessor
1594 .fenced_attach(
1595 debt,
1596 Event::fenced_recovery_committed(
1597 self.participant_id,
1598 self.marker_delivery_seq,
1599 self.prior_binding_epoch,
1600 self.new_binding_epoch,
1601 self.resulting_floor,
1602 ),
1603 self.successor.restore()?,
1604 )
1605 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1606 }
1607}
1608
1609#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1619pub struct RecoveredBindingFateRestore {
1620 pub fenced_attach: FencedAttachCommitRestore,
1622 pub participant_id: ParticipantId,
1624 pub binding_epoch: BindingEpoch,
1626 pub resulting_floor: DeliverySeq,
1628}
1629
1630impl RecoveredBindingFateRestore {
1631 #[cfg(test)]
1638 pub(super) fn restore(
1639 self,
1640 record_authority: ValidatedMarkerRecord,
1641 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
1642 let restored = self.restore_with_record(&record_authority);
1643 record_authority.consume();
1644 restored
1645 }
1646
1647 fn restore_with_record(
1648 self,
1649 record_authority: &ValidatedMarkerRecord,
1650 ) -> Result<RecoveredBindingFate, StorageRestoreError> {
1651 let commit = self.fenced_attach.restore_with_record(record_authority)?;
1652 commit
1653 .recovered_binding_fate(Event::binding_fate_observed(
1654 self.participant_id,
1655 self.binding_epoch,
1656 self.resulting_floor,
1657 ))
1658 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1659 }
1660}
1661
1662#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1664pub struct PendingRecoveredCursorReleaseRestore {
1665 pub fate: RecoveredBindingFateRestore,
1667 pub resulting_debt: WideResourceVector,
1669}
1670
1671impl PendingRecoveredCursorReleaseRestore {
1672 #[cfg(test)]
1679 pub(super) fn restore(
1680 self,
1681 record_authority: ValidatedMarkerRecord,
1682 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
1683 let restored = self.restore_with_record(&record_authority);
1684 record_authority.consume();
1685 restored
1686 }
1687
1688 fn restore_with_record(
1689 self,
1690 record_authority: &ValidatedMarkerRecord,
1691 ) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
1692 let authority = self.fate.restore_with_record(record_authority)?;
1693 let predecessor_state = authority.predecessor_state();
1694 let resulting_debt =
1695 ClosureDebt::new(self.resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1696 let transition = match predecessor_state {
1697 ClosureState::Owed {
1698 debt,
1699 edge: StoredEdge::ObserverProjection(edge),
1700 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
1701 ClosureState::Owed {
1702 debt,
1703 edge: StoredEdge::PhysicalCompaction(edge),
1704 } => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
1705 ClosureState::Clear | ClosureState::Owed { .. } => {
1706 return Err(StorageRestoreError::StoredEdgeProvenance);
1707 }
1708 }
1709 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1710 match transition {
1711 RecoveredBindingFateTransition::PendingStorage(pending) => Ok(pending),
1712 RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
1713 Err(StorageRestoreError::StoredEdgeProvenance)
1714 }
1715 }
1716 }
1717}
1718
1719#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1721pub enum RecoveredStorageCompletionRestore {
1722 ObserverProjection {
1724 through_seq: DeliverySeq,
1726 resulting_debt: Option<WideResourceVector>,
1728 },
1729 PhysicalCompaction {
1731 from_floor: DeliverySeq,
1733 through_seq: DeliverySeq,
1735 resulting_floor: DeliverySeq,
1737 resulting_debt: Option<WideResourceVector>,
1739 },
1740}
1741
1742impl RecoveredStorageCompletionRestore {
1743 fn restore(
1744 self,
1745 pending: PendingRecoveredCursorRelease,
1746 ) -> Result<ClosureState, StorageRestoreError> {
1747 let current = pending.current_state();
1748 match (current, self) {
1749 (
1750 ClosureState::Owed {
1751 edge: StoredEdge::ObserverProjection(edge),
1752 ..
1753 },
1754 Self::ObserverProjection {
1755 through_seq,
1756 resulting_debt,
1757 },
1758 ) => edge
1759 .complete_after_recovered_binding_fate(
1760 Event::projection_completed(through_seq),
1761 optional_debt(resulting_debt)?,
1762 pending,
1763 )
1764 .map_err(|_| StorageRestoreError::StoredEdgeProvenance),
1765 (
1766 ClosureState::Owed {
1767 edge: StoredEdge::PhysicalCompaction(edge),
1768 ..
1769 },
1770 Self::PhysicalCompaction {
1771 from_floor,
1772 through_seq,
1773 resulting_floor,
1774 resulting_debt,
1775 },
1776 ) => {
1777 let event = Event::compaction_completed(from_floor, through_seq, resulting_floor)
1778 .ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1779 edge.complete_after_recovered_binding_fate(
1780 event,
1781 optional_debt(resulting_debt)?,
1782 pending,
1783 )
1784 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)
1785 }
1786 _ => Err(StorageRestoreError::StoredEdgeProvenance),
1787 }
1788 }
1789}
1790
1791#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1793pub enum DetachedCursorReleaseProvenanceRestore {
1794 Ordinary(OrdinaryBindingFateRestore),
1796 RecoveredDirect {
1798 fate: RecoveredBindingFateRestore,
1800 resulting_debt: WideResourceVector,
1802 },
1803 RecoveredAfterStorage {
1805 pending: PendingRecoveredCursorReleaseRestore,
1807 completion: RecoveredStorageCompletionRestore,
1809 },
1810}
1811
1812impl DetachedCursorReleaseProvenanceRestore {
1813 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
1814 match self {
1815 Self::Ordinary(fate) => Some(fate.authority.binding),
1816 Self::RecoveredDirect { .. } | Self::RecoveredAfterStorage { .. } => None,
1817 }
1818 }
1819
1820 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
1821 match self {
1822 Self::Ordinary(_) => None,
1823 Self::RecoveredDirect { fate, .. } => Some(MarkerRecordRequest::recovered(
1824 fate.participant_id,
1825 fate.fenced_attach.marker_delivery_seq,
1826 fate.fenced_attach.prior_binding_epoch,
1827 fate.fenced_attach.new_binding_epoch,
1828 )),
1829 Self::RecoveredAfterStorage { pending, .. } => Some(MarkerRecordRequest::recovered(
1830 pending.fate.participant_id,
1831 pending.fate.fenced_attach.marker_delivery_seq,
1832 pending.fate.fenced_attach.prior_binding_epoch,
1833 pending.fate.fenced_attach.new_binding_epoch,
1834 )),
1835 }
1836 }
1837
1838 fn restore_state(
1839 self,
1840 debt: ClosureDebt,
1841 marker_authority: &MarkerRestoreAuthority<'_>,
1842 ordinary_origin: Option<&BindingOrigin>,
1843 ) -> Result<ClosureState, StorageRestoreError> {
1844 match self {
1845 Self::Ordinary(provenance) => {
1846 marker_authority.require_absent()?;
1847 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
1848 Ok(provenance
1849 .restore_with_origin(origin)?
1850 .into_direct_state(debt))
1851 }
1852 Self::RecoveredDirect {
1853 fate,
1854 resulting_debt,
1855 } => {
1856 let (_, record_authority) = marker_authority.require_record()?;
1857 let authority = fate.restore_with_record(record_authority)?;
1858 let predecessor = authority.predecessor_state();
1859 let resulting_debt =
1860 ClosureDebt::new(resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
1861 let transition = match predecessor {
1862 ClosureState::Owed {
1863 debt: predecessor_debt,
1864 edge: StoredEdge::PhysicalCompaction(edge),
1865 } => edge.apply_recovered_binding_fate(
1866 predecessor_debt,
1867 resulting_debt,
1868 authority,
1869 ),
1870 ClosureState::Clear | ClosureState::Owed { .. } => {
1871 return Err(StorageRestoreError::StoredEdgeProvenance);
1872 }
1873 }
1874 .map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
1875 match transition {
1876 RecoveredBindingFateTransition::DetachedCursorRelease(release)
1877 if release.debt() == debt =>
1878 {
1879 Ok(release.into_state())
1880 }
1881 RecoveredBindingFateTransition::PendingStorage(_)
1882 | RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
1883 Err(StorageRestoreError::StoredEdgeProvenance)
1884 }
1885 }
1886 }
1887 Self::RecoveredAfterStorage {
1888 pending,
1889 completion,
1890 } => {
1891 let (_, record_authority) = marker_authority.require_record()?;
1892 let state = completion.restore(pending.restore_with_record(record_authority)?)?;
1893 match state {
1894 ClosureState::Owed {
1895 debt: restored_debt,
1896 edge: StoredEdge::DetachedCursorRelease(_),
1897 } if restored_debt == debt => Ok(state),
1898 ClosureState::Clear | ClosureState::Owed { .. } => {
1899 Err(StorageRestoreError::StoredEdgeProvenance)
1900 }
1901 }
1902 }
1903 }
1904 }
1905}
1906
1907#[allow(clippy::large_enum_variant)]
1909#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1910pub enum StoredEdgeRestore {
1911 ObserverProjection {
1913 through_seq: DeliverySeq,
1915 },
1916 PhysicalCompaction {
1918 from_floor: DeliverySeq,
1920 through_seq: DeliverySeq,
1922 },
1923 MarkerDelivery(MarkerDeliveryRestore),
1925 ParticipantCursorProgressContinuous {
1927 participant_id: ParticipantId,
1929 binding_epoch: BindingEpoch,
1931 through_seq: DeliverySeq,
1933 authority: OrdinaryBindingAuthorityRestore,
1935 },
1936 ParticipantCursorProgressMarker(MarkerCursorProgressRestore),
1938 DetachedCredentialRecovery(DetachedCredentialRecoveryRestore),
1940 DetachedMarkerRelease(DetachedMarkerReleaseRestore),
1942 DetachedCursorRelease {
1944 participant_id: ParticipantId,
1946 last_dead_binding_epoch: BindingEpoch,
1948 provenance: DetachedCursorReleaseProvenanceRestore,
1950 },
1951}
1952
1953impl StoredEdgeRestore {
1954 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
1955 match self {
1956 Self::ParticipantCursorProgressContinuous { authority, .. } => Some(authority.binding),
1957 Self::DetachedCursorRelease { provenance, .. } => provenance.ordinary_binding_request(),
1958 Self::ObserverProjection { .. }
1959 | Self::PhysicalCompaction { .. }
1960 | Self::MarkerDelivery(_)
1961 | Self::ParticipantCursorProgressMarker(_)
1962 | Self::DetachedCredentialRecovery(_)
1963 | Self::DetachedMarkerRelease(_) => None,
1964 }
1965 }
1966
1967 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
1968 match self {
1969 Self::MarkerDelivery(value) => Some(MarkerRecordRequest::planned(
1970 value.participant_id,
1971 value.marker_delivery_seq,
1972 super::FrontierBinding::Bound(value.binding_epoch),
1973 )),
1974 Self::ParticipantCursorProgressMarker(value) => Some(MarkerRecordRequest::delivered(
1975 value.participant_id,
1976 value.marker_delivery_seq,
1977 super::FrontierBinding::Bound(value.binding_epoch),
1978 )),
1979 Self::DetachedCredentialRecovery(value) => Some(MarkerRecordRequest::delivered(
1980 value.participant_id,
1981 value.marker_delivery_seq,
1982 super::FrontierBinding::Detached(value.prior_binding_epoch),
1983 )),
1984 Self::DetachedMarkerRelease(value) => Some(MarkerRecordRequest::planned(
1985 value.participant_id,
1986 value.marker_delivery_seq,
1987 super::FrontierBinding::Detached(value.last_dead_binding_epoch),
1988 )),
1989 Self::DetachedCursorRelease { provenance, .. } => provenance.marker_record_request(),
1990 Self::ObserverProjection { .. }
1991 | Self::PhysicalCompaction { .. }
1992 | Self::ParticipantCursorProgressContinuous { .. } => None,
1993 }
1994 }
1995
1996 fn restore_with_debt(
1997 self,
1998 debt: ClosureDebt,
1999 marker_authority: &MarkerRestoreAuthority<'_>,
2000 ordinary_origin: Option<&BindingOrigin>,
2001 ) -> Result<StoredEdge, StorageRestoreError> {
2002 match self {
2003 Self::ObserverProjection { through_seq } => {
2004 marker_authority.require_absent()?;
2005 Ok(StoredEdge::ObserverProjection(ObserverProjection::new(
2006 through_seq,
2007 )))
2008 }
2009 Self::PhysicalCompaction {
2010 from_floor,
2011 through_seq,
2012 } => {
2013 marker_authority.require_absent()?;
2014 PhysicalCompaction::new(from_floor, through_seq)
2015 .map(StoredEdge::PhysicalCompaction)
2016 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2017 }
2018 Self::MarkerDelivery(value) => {
2019 let (conversation_id, record_authority) = marker_authority.require_record()?;
2020 value
2021 .restore_with_target(
2022 conversation_id,
2023 record_authority,
2024 MarkerRecordTarget::Bound,
2025 MarkerRecordOccurrence::Undelivered,
2026 )
2027 .map(StoredEdge::MarkerDelivery)
2028 }
2029 Self::ParticipantCursorProgressContinuous {
2030 participant_id,
2031 binding_epoch,
2032 through_seq,
2033 authority,
2034 } => {
2035 marker_authority.require_absent()?;
2036 let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
2037 ParticipantCursorProgress::restore_continuous(
2038 authority.restore_with_origin(origin)?,
2039 participant_id,
2040 binding_epoch,
2041 through_seq,
2042 )
2043 .map(StoredEdge::ParticipantCursorProgress)
2044 .ok_or(StorageRestoreError::StoredEdgeProvenance)
2045 }
2046 Self::ParticipantCursorProgressMarker(value) => {
2047 let record_authority = marker_authority.record_for(value.conversation_id)?;
2048 value
2049 .restore_with_debt(debt, record_authority, MarkerRecordTarget::Bound)
2050 .map(StoredEdge::ParticipantCursorProgress)
2051 }
2052 Self::DetachedCredentialRecovery(value) => {
2053 let record_authority =
2054 marker_authority.record_for(value.progress.conversation_id)?;
2055 value
2056 .restore_with_debt(debt, record_authority)
2057 .map(StoredEdge::DetachedCredentialRecovery)
2058 }
2059 Self::DetachedMarkerRelease(value) => {
2060 let record_authority = marker_authority.record_for(value.conversation_id)?;
2061 value
2062 .restore_with_debt(debt, record_authority)
2063 .map(StoredEdge::DetachedMarkerRelease)
2064 }
2065 Self::DetachedCursorRelease {
2066 participant_id,
2067 last_dead_binding_epoch,
2068 provenance,
2069 } => {
2070 let state = provenance.restore_state(debt, marker_authority, ordinary_origin)?;
2071 match state {
2072 ClosureState::Owed {
2073 edge: StoredEdge::DetachedCursorRelease(edge),
2074 ..
2075 } if edge.participant_id() == participant_id
2076 && edge.last_dead_binding_epoch() == last_dead_binding_epoch =>
2077 {
2078 Ok(StoredEdge::DetachedCursorRelease(edge))
2079 }
2080 ClosureState::Clear | ClosureState::Owed { .. } => {
2081 Err(StorageRestoreError::StoredEdgeProvenance)
2082 }
2083 }
2084 }
2085 }
2086 }
2087}
2088
2089#[allow(clippy::large_enum_variant)]
2091#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2092pub enum ClosureStateRestore {
2093 Clear,
2095 Owed {
2097 debt: WideResourceVector,
2099 edge: StoredEdgeRestore,
2101 },
2102}
2103
2104impl ClosureStateRestore {
2105 const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
2106 match self {
2107 Self::Clear => None,
2108 Self::Owed { edge, .. } => edge.ordinary_binding_request(),
2109 }
2110 }
2111
2112 const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
2113 match self {
2114 Self::Clear => None,
2115 Self::Owed { edge, .. } => edge.marker_record_request(),
2116 }
2117 }
2118
2119 pub fn restore(self) -> Result<ClosureState, StorageRestoreError> {
2126 self.restore_with_authority(&MarkerRestoreAuthority::Absent, None)
2127 }
2128
2129 pub(super) fn restore_with_marker_record(
2142 self,
2143 conversation_id: ConversationId,
2144 record: ValidatedMarkerRecord,
2145 ) -> Result<ClosureState, StorageRestoreError> {
2146 let restored = self.restore_with_authority(
2147 &MarkerRestoreAuthority::Record {
2148 conversation_id,
2149 record: &record,
2150 },
2151 None,
2152 );
2153 record.consume();
2154 restored
2155 }
2156
2157 fn restore_with_binding_origin(
2158 self,
2159 origin: &BindingOrigin,
2160 ) -> Result<ClosureState, StorageRestoreError> {
2161 self.restore_with_authority(&MarkerRestoreAuthority::Absent, Some(origin))
2162 }
2163
2164 fn restore_with_authority(
2165 self,
2166 marker_authority: &MarkerRestoreAuthority<'_>,
2167 ordinary_origin: Option<&BindingOrigin>,
2168 ) -> Result<ClosureState, StorageRestoreError> {
2169 match self {
2170 Self::Clear => {
2171 marker_authority.require_absent()?;
2172 Ok(ClosureState::Clear)
2173 }
2174 Self::Owed { debt, edge } => {
2175 let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
2176 Ok(ClosureState::Owed {
2177 debt,
2178 edge: edge.restore_with_debt(debt, marker_authority, ordinary_origin)?,
2179 })
2180 }
2181 }
2182 }
2183}
2184
2185fn optional_debt(
2186 raw: Option<WideResourceVector>,
2187) -> Result<Option<ClosureDebt>, StorageRestoreError> {
2188 raw.map(|value| ClosureDebt::new(value).ok_or(StorageRestoreError::ClosureDebt))
2189 .transpose()
2190}