1use alloc::{boxed::Box, vec::Vec};
10
11mod binding_fate_transition;
12pub(in crate::lifecycle) use binding_fate_transition::BindingFateFrontierPlan;
13
14use crate::{
15 algebra::ResourceVector,
16 outcome::{CandidatePhase, ClaimCounter, ParticipantStateCorruptReason},
17 wire::{
18 BindingEpoch, ClosureCheckedEnvelope, ConversationId, DeliverySeq, ParticipantId,
19 TransactionOrder,
20 },
21};
22
23use super::{
24 AttachedLifecycleRecord, BindingOrigin, BindingState, ClosureAccounting, ClosureState,
25 CommittedBindingTerminal, Event, InitialEnrollmentClosureProjection,
26 InitialEnrollmentOperationCommit, LeaveCommitError, LiveMember, MarkerDelivery,
27 ObserverCheckedOperation, ObserverFloorDecision, ObserverProjection, OrderClaims, OrderHigh,
28 OrderLedger, ParticipantCursorProgress, PendingFinalization, PreparedLeaveAuthority,
29 RecoveryQuartetStatus, RecoverySequenceReserve, RemainingClosureDecision, SequenceClaims,
30 SequenceLedger, StoredEdge, check_observer_floor, check_remaining_closure,
31 operations::ordinary_record_projection::{
32 OrdinaryFixedPointPlan, OrdinaryProjectionError, OrdinaryProjectionFacts,
33 OrdinaryProjectionKernelDecision, OrdinaryRecordDrainFirst,
34 OrdinaryRecordProjectionDecision, OrdinaryRecordProjectionFailure,
35 OrdinaryRecordProjectionInput, ProjectedOrdinaryRecord, project_ordinary_fixed_point,
36 },
37};
38
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum ClaimFrontierCounter {
42 DeliverySequence,
44 TransactionOrder,
46}
47
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
50pub enum ClaimFrontierInvalidReason {
51 NumericPosition,
53 CandidateKey,
55 LogicalOwner,
57 ProductRange,
59 RecoveryBlock,
61 AggregateLedger,
63}
64
65#[derive(Clone, Copy, Debug, PartialEq, Eq)]
67pub struct ClaimFrontierError {
68 pub counter: ClaimFrontierCounter,
70 pub first_bad_position: u128,
72 pub reason: ClaimFrontierInvalidReason,
74}
75
76#[derive(Clone, Copy, Debug, PartialEq, Eq)]
78pub enum FrontierBinding {
79 Bound(BindingEpoch),
81 Detached(BindingEpoch),
83}
84
85#[derive(Clone, Copy, Debug, PartialEq, Eq)]
87pub struct FrontierParticipant {
88 participant_index: ParticipantId,
89 cursor: DeliverySeq,
90 binding: FrontierBinding,
91}
92
93impl FrontierParticipant {
94 #[must_use]
96 pub const fn new(
97 participant_index: ParticipantId,
98 cursor: DeliverySeq,
99 binding: FrontierBinding,
100 ) -> Self {
101 Self {
102 participant_index,
103 cursor,
104 binding,
105 }
106 }
107
108 #[must_use]
110 pub const fn participant_index(self) -> ParticipantId {
111 self.participant_index
112 }
113
114 #[must_use]
116 pub const fn cursor(self) -> DeliverySeq {
117 self.cursor
118 }
119
120 #[must_use]
122 pub const fn binding(self) -> FrontierBinding {
123 self.binding
124 }
125}
126
127#[derive(Clone, Debug, PartialEq, Eq)]
129pub struct ActiveIdentityRanks {
130 participants: Vec<FrontierParticipant>,
131}
132
133impl ActiveIdentityRanks {
134 pub fn try_new(
142 participants: Vec<FrontierParticipant>,
143 high_watermark: DeliverySeq,
144 identity_slot_limit: u64,
145 ) -> Result<Self, ClaimFrontierError> {
146 if usize_to_u128(participants.len()) > u128::from(identity_slot_limit) {
147 return Err(sequence_error(
148 u128::from(identity_slot_limit),
149 ClaimFrontierInvalidReason::LogicalOwner,
150 ));
151 }
152 let mut previous = None;
153 for (rank, participant) in participants.iter().enumerate() {
154 if previous.is_some_and(|value| value >= participant.participant_index)
155 || participant.participant_index >= identity_slot_limit
156 || participant.cursor > high_watermark
157 {
158 return Err(sequence_error(
159 rank_index(rank),
160 ClaimFrontierInvalidReason::LogicalOwner,
161 ));
162 }
163 previous = Some(participant.participant_index);
164 }
165 Ok(Self { participants })
166 }
167
168 #[must_use]
170 pub fn participants(&self) -> &[FrontierParticipant] {
171 &self.participants
172 }
173
174 #[must_use]
176 pub fn len(&self) -> u64 {
177 usize_to_u64(self.participants.len())
178 }
179
180 #[must_use]
182 pub fn is_empty(&self) -> bool {
183 self.participants.is_empty()
184 }
185
186 fn contains(&self, participant_index: ParticipantId) -> bool {
187 self.participants
188 .binary_search_by_key(&participant_index, |participant| {
189 participant.participant_index
190 })
191 .is_ok()
192 }
193}
194
195#[derive(Clone, Copy, Debug, PartialEq, Eq)]
197pub struct BindingTerminalOwner {
198 pub participant_index: ParticipantId,
200 pub binding_epoch: BindingEpoch,
202}
203
204#[derive(Clone, Copy, Debug, PartialEq, Eq)]
206pub enum SequenceDirectOwner {
207 MembershipExit {
209 participant_index: ParticipantId,
211 },
212 BindingTerminal(BindingTerminalOwner),
214}
215
216#[derive(Clone, Copy, Debug, PartialEq, Eq)]
218pub struct MovableSequenceClaim {
219 pub delivery_seq: DeliverySeq,
221 pub owner: SequenceDirectOwner,
223}
224
225#[derive(Clone, Copy, Debug, PartialEq, Eq)]
227pub enum TerminalProductSource {
228 Binding(BindingTerminalOwner),
230 RecoveryReplacement {
232 participant_index: ParticipantId,
234 binding_epoch: BindingEpoch,
236 },
237}
238
239impl TerminalProductSource {
240 #[must_use]
242 pub const fn recovery_replacement(
243 participant_index: ParticipantId,
244 binding_epoch: BindingEpoch,
245 ) -> Self {
246 Self::RecoveryReplacement {
247 participant_index,
248 binding_epoch,
249 }
250 }
251}
252
253#[derive(Clone, Copy, Debug, PartialEq, Eq)]
255pub enum MarkerProvenance {
256 NonProductM,
258 TerminalProduct {
260 terminal: TerminalProductSource,
262 affected_participant: ParticipantId,
264 },
265 ExitProduct {
267 exit_participant: ParticipantId,
269 remaining_participant: ParticipantId,
271 },
272}
273
274impl MarkerProvenance {
275 #[must_use]
277 pub const fn terminal_product(
278 terminal: TerminalProductSource,
279 affected_participant: ParticipantId,
280 ) -> Self {
281 Self::TerminalProduct {
282 terminal,
283 affected_participant,
284 }
285 }
286
287 #[must_use]
289 pub const fn exit_product(
290 exit_participant: ParticipantId,
291 remaining_participant: ParticipantId,
292 ) -> Self {
293 Self::ExitProduct {
294 exit_participant,
295 remaining_participant,
296 }
297 }
298}
299
300#[derive(Clone, Copy, Debug, PartialEq, Eq)]
302pub enum RetainedCausalRecordKind {
303 BindingTerminal(BindingTerminalOwner),
305 MembershipExit {
307 participant_index: ParticipantId,
309 },
310 AttachLifecycle {
312 participant_index: ParticipantId,
314 binding_epoch: BindingEpoch,
316 },
317 OrdinaryRecord {
319 participant_index: ParticipantId,
321 },
322 CompactionMarker {
324 participant_index: ParticipantId,
326 provenance: MarkerProvenance,
328 },
329}
330
331#[derive(Clone, Copy, Debug, PartialEq, Eq)]
333pub struct RetainedCausalRecord {
334 pub delivery_seq: DeliverySeq,
336 pub admission_order: super::AdmissionOrder,
338 pub kind: RetainedCausalRecordKind,
340}
341
342#[derive(Clone, Copy, Debug, PartialEq, Eq)]
343enum HistoricalCausalKind {
344 BindingTerminal(BindingTerminalOwner),
345 MembershipExit(ParticipantId),
346}
347
348#[derive(Clone, Copy, Debug, PartialEq, Eq)]
360pub enum HistoricalCausalFactRestore {
361 BindingTerminal {
363 conversation_id: ConversationId,
365 participant_index: ParticipantId,
367 binding_epoch: BindingEpoch,
369 admission_order: super::AdmissionOrder,
371 },
372 MembershipExit {
374 conversation_id: ConversationId,
376 participant_index: ParticipantId,
378 admission_order: super::AdmissionOrder,
380 },
381}
382
383#[derive(Clone, Copy, Debug, PartialEq, Eq)]
392pub struct HistoricalMarkerDeliveryFactRestore {
393 pub conversation_id: ConversationId,
395 pub participant_index: ParticipantId,
397 pub marker_delivery_seq: DeliverySeq,
399 pub delivered_binding_epoch: BindingEpoch,
401}
402
403#[derive(Clone, Copy, Debug, PartialEq, Eq)]
405pub(super) struct HistoricalCausalAuthority {
406 conversation_id: ConversationId,
407 admission_order: super::AdmissionOrder,
408 kind: HistoricalCausalKind,
409}
410
411#[derive(Clone, Copy, Debug, PartialEq, Eq)]
412struct HistoricalMarkerDeliveryAuthority {
413 participant_index: ParticipantId,
414 marker_delivery_seq: DeliverySeq,
415 delivered_binding_epoch: BindingEpoch,
416}
417
418impl HistoricalCausalAuthority {
419 pub(super) const fn from_committed_terminal(terminal: super::CommittedBindingTerminal) -> Self {
420 Self {
421 conversation_id: terminal.conversation_id(),
422 admission_order: terminal.admission_order(),
423 kind: HistoricalCausalKind::BindingTerminal(BindingTerminalOwner {
424 participant_index: terminal.participant_id(),
425 binding_epoch: terminal.binding_epoch(),
426 }),
427 }
428 }
429
430 pub(super) const fn from_retired<EF, V, LF>(
431 retired: &super::RetiredIdentity<EF, V, LF>,
432 ) -> Self {
433 Self {
434 conversation_id: retired.conversation_id(),
435 admission_order: retired.left_admission_order(),
436 kind: HistoricalCausalKind::MembershipExit(retired.participant_id()),
437 }
438 }
439
440 const fn from_restore(fact: HistoricalCausalFactRestore) -> Self {
441 match fact {
442 HistoricalCausalFactRestore::BindingTerminal {
443 conversation_id,
444 participant_index,
445 binding_epoch,
446 admission_order,
447 } => Self {
448 conversation_id,
449 admission_order,
450 kind: HistoricalCausalKind::BindingTerminal(BindingTerminalOwner {
451 participant_index,
452 binding_epoch,
453 }),
454 },
455 HistoricalCausalFactRestore::MembershipExit {
456 conversation_id,
457 participant_index,
458 admission_order,
459 } => Self {
460 conversation_id,
461 admission_order,
462 kind: HistoricalCausalKind::MembershipExit(participant_index),
463 },
464 }
465 }
466}
467
468#[derive(Debug)]
469pub(super) struct ValidatedConversationHistory {
470 causal_authorities: Vec<HistoricalCausalAuthority>,
471 binding_origins: Vec<BindingOrigin>,
472 total: bool,
473}
474
475impl ValidatedConversationHistory {
476 pub(super) const fn empty() -> Self {
477 Self {
478 causal_authorities: Vec::new(),
479 binding_origins: Vec::new(),
480 total: false,
481 }
482 }
483
484 pub(super) const fn new(
485 causal_authorities: Vec<HistoricalCausalAuthority>,
486 binding_origins: Vec<BindingOrigin>,
487 ) -> Self {
488 Self {
489 causal_authorities,
490 binding_origins,
491 total: true,
492 }
493 }
494
495 pub(super) fn ordinary_origin(
496 &self,
497 conversation_id: ConversationId,
498 participant_index: ParticipantId,
499 binding_epoch: BindingEpoch,
500 ) -> Option<&BindingOrigin> {
501 self.binding_origins.iter().find(|origin| {
502 origin.conversation_id() == conversation_id
503 && origin.participant_id() == participant_index
504 && origin.binding_epoch() == binding_epoch
505 && origin.is_unfenced()
506 })
507 }
508}
509
510#[derive(Clone, Copy, Debug, PartialEq, Eq)]
512pub enum SequenceProductClass {
513 LiveTimesTerminal,
515 LiveTimesReplacementTerminal,
517 OtherLiveTimesExit,
519}
520
521#[derive(Clone, Copy, Debug, PartialEq, Eq)]
523pub enum MarkerSequenceOwner {
524 Marker,
526 ConditionalProduct(SequenceProductClass),
528}
529
530#[derive(Clone, Copy, Debug, PartialEq, Eq)]
532pub struct MarkerCandidateAuthority {
533 pub delivery_seq: DeliverySeq,
535 pub admission_order: super::AdmissionOrder,
537 pub target_binding: FrontierBinding,
539 pub provenance: MarkerProvenance,
541 pub abandoned_after: DeliverySeq,
543 pub abandoned_through: DeliverySeq,
545 pub physical_floor_at_decision: DeliverySeq,
547 pub current_owner: MarkerSequenceOwner,
549}
550
551#[derive(Clone, Copy, Debug, PartialEq, Eq)]
553pub enum ImmutableSequenceCandidate {
554 BindingTerminal {
556 delivery_seq: DeliverySeq,
558 admission_order: super::AdmissionOrder,
560 owner: BindingTerminalOwner,
562 },
563 Marker(MarkerCandidateAuthority),
565}
566
567impl ImmutableSequenceCandidate {
568 #[must_use]
570 pub const fn delivery_seq(self) -> DeliverySeq {
571 match self {
572 Self::BindingTerminal { delivery_seq, .. } => delivery_seq,
573 Self::Marker(candidate) => candidate.delivery_seq,
574 }
575 }
576
577 #[must_use]
579 pub const fn admission_order(self) -> super::AdmissionOrder {
580 match self {
581 Self::BindingTerminal {
582 admission_order, ..
583 } => admission_order,
584 Self::Marker(candidate) => candidate.admission_order,
585 }
586 }
587}
588
589#[derive(Clone, Copy, Debug, PartialEq, Eq)]
591pub struct TerminalProductRangeRestore {
592 pub start: DeliverySeq,
594 pub length: u64,
596 pub terminal: BindingTerminalOwner,
598}
599
600#[derive(Clone, Copy, Debug, PartialEq, Eq)]
602pub struct ReplacementTerminalProductRangeRestore {
603 pub start: DeliverySeq,
605 pub length: u64,
607}
608
609#[derive(Clone, Copy, Debug, PartialEq, Eq)]
611pub struct ExitProductRangeRestore {
612 pub start: DeliverySeq,
614 pub length: u64,
616 pub exit_participant: ParticipantId,
618}
619
620#[derive(Clone, Debug, Default, PartialEq, Eq)]
622pub struct SequenceProductRangesRestore {
623 pub live_times_terminal: Vec<TerminalProductRangeRestore>,
625 pub live_times_replacement_terminal: Option<ReplacementTerminalProductRangeRestore>,
627 pub other_live_times_exit: Vec<ExitProductRangeRestore>,
629}
630
631#[derive(Clone, Copy, Debug, PartialEq, Eq)]
633pub struct TerminalProductRange {
634 start: DeliverySeq,
635 length: u64,
636 terminal: BindingTerminalOwner,
637}
638
639impl TerminalProductRange {
640 #[must_use]
642 pub const fn start(self) -> DeliverySeq {
643 self.start
644 }
645
646 #[must_use]
648 pub const fn terminal(self) -> BindingTerminalOwner {
649 self.terminal
650 }
651
652 #[must_use]
654 pub const fn length(self) -> u64 {
655 self.length
656 }
657
658 #[must_use]
660 pub fn value_at_rank(self, active_rank: usize) -> Option<DeliverySeq> {
661 if usize_to_u128(active_rank) >= u128::from(self.length) {
662 return None;
663 }
664 checked_rank_value(self.start, active_rank)
665 }
666}
667
668#[derive(Clone, Copy, Debug, PartialEq, Eq)]
670pub struct ReplacementTerminalProductRange {
671 start: DeliverySeq,
672 length: u64,
673 participant_index: ParticipantId,
674 marker_delivery_seq: DeliverySeq,
675 prior_binding_epoch: BindingEpoch,
676}
677
678impl ReplacementTerminalProductRange {
679 #[must_use]
681 pub const fn start(self) -> DeliverySeq {
682 self.start
683 }
684
685 #[must_use]
687 pub const fn length(self) -> u64 {
688 self.length
689 }
690
691 #[must_use]
693 pub const fn participant_index(self) -> ParticipantId {
694 self.participant_index
695 }
696
697 #[must_use]
699 pub const fn marker_delivery_seq(self) -> DeliverySeq {
700 self.marker_delivery_seq
701 }
702
703 #[must_use]
705 pub const fn prior_binding_epoch(self) -> BindingEpoch {
706 self.prior_binding_epoch
707 }
708
709 #[must_use]
711 pub fn value_at_rank(self, active_rank: usize) -> Option<DeliverySeq> {
712 if usize_to_u128(active_rank) >= u128::from(self.length) {
713 return None;
714 }
715 checked_rank_value(self.start, active_rank)
716 }
717}
718
719#[derive(Clone, Copy, Debug, PartialEq, Eq)]
721pub struct ExitProductRange {
722 start: DeliverySeq,
723 length: u64,
724 exit_participant: ParticipantId,
725}
726
727impl ExitProductRange {
728 #[must_use]
730 pub const fn start(self) -> DeliverySeq {
731 self.start
732 }
733
734 #[must_use]
736 pub const fn exit_participant(self) -> ParticipantId {
737 self.exit_participant
738 }
739
740 #[must_use]
742 pub const fn length(self) -> u64 {
743 self.length
744 }
745
746 #[must_use]
750 pub fn value_for_affected_rank(
751 self,
752 active_identities: &ActiveIdentityRanks,
753 affected_rank: usize,
754 ) -> Option<DeliverySeq> {
755 let affected = active_identities.participants().get(affected_rank)?;
756 if usize_to_u128(affected_rank) >= usize_to_u128(active_identities.participants.len()) {
757 return None;
758 }
759 if affected.participant_index == self.exit_participant {
760 return None;
761 }
762 let exit_rank = active_identities
763 .participants()
764 .binary_search_by_key(&self.exit_participant, |participant| {
765 participant.participant_index
766 })
767 .ok()?;
768 let compact_rank = if affected_rank < exit_rank {
769 affected_rank
770 } else {
771 affected_rank.checked_sub(1)?
772 };
773 if usize_to_u128(compact_rank) >= u128::from(self.length) {
774 return None;
775 }
776 checked_rank_value(self.start, compact_rank)
777 }
778}
779
780#[derive(Clone, Debug, Default, PartialEq, Eq)]
782pub struct SequenceProductRanges {
783 live_times_terminal: Vec<TerminalProductRange>,
784 live_times_replacement_terminal: Option<ReplacementTerminalProductRange>,
785 other_live_times_exit: Vec<ExitProductRange>,
786}
787
788impl SequenceProductRanges {
789 #[must_use]
791 pub fn live_times_terminal(&self) -> &[TerminalProductRange] {
792 &self.live_times_terminal
793 }
794
795 #[must_use]
797 pub const fn live_times_replacement_terminal(&self) -> Option<ReplacementTerminalProductRange> {
798 self.live_times_replacement_terminal
799 }
800
801 #[must_use]
803 pub fn other_live_times_exit(&self) -> &[ExitProductRange] {
804 &self.other_live_times_exit
805 }
806}
807
808#[derive(Clone, Copy, Debug, PartialEq, Eq)]
810pub struct RecoverySequenceTerminalRestore {
811 pub delivery_seq: DeliverySeq,
813 pub owner: BindingTerminalOwner,
815}
816
817#[derive(Clone, Copy, Debug, PartialEq, Eq)]
823pub struct RecoverySequenceBlockRestore {
824 pub terminal: Option<RecoverySequenceTerminalRestore>,
826 pub recovery_attach_seq: DeliverySeq,
828 pub replacement_terminal_seq: DeliverySeq,
830}
831
832#[derive(Clone, Copy, Debug, PartialEq, Eq)]
834pub struct RecoverySequenceBlock {
835 terminal: Option<RecoverySequenceTerminalRestore>,
836 recovery_attach_seq: DeliverySeq,
837 replacement_terminal_seq: DeliverySeq,
838 participant_index: ParticipantId,
839 marker_delivery_seq: DeliverySeq,
840 recovered_binding_epoch: BindingEpoch,
841}
842
843impl RecoverySequenceBlock {
844 #[must_use]
846 pub const fn terminal(self) -> Option<RecoverySequenceTerminalRestore> {
847 self.terminal
848 }
849
850 #[must_use]
852 pub const fn recovery_attach_seq(self) -> DeliverySeq {
853 self.recovery_attach_seq
854 }
855
856 #[must_use]
858 pub const fn replacement_terminal_seq(self) -> DeliverySeq {
859 self.replacement_terminal_seq
860 }
861
862 #[must_use]
864 pub const fn participant_index(self) -> ParticipantId {
865 self.participant_index
866 }
867
868 #[must_use]
870 pub const fn marker_delivery_seq(self) -> DeliverySeq {
871 self.marker_delivery_seq
872 }
873
874 #[must_use]
876 pub const fn recovered_binding_epoch(self) -> BindingEpoch {
877 self.recovered_binding_epoch
878 }
879}
880
881#[derive(Clone, Debug, Default, PartialEq, Eq)]
883pub struct SequenceClaimFrontierRestore {
884 pub movable_claims: Vec<MovableSequenceClaim>,
886 pub immutable_candidates: Vec<ImmutableSequenceCandidate>,
888 pub products: SequenceProductRangesRestore,
890 pub recovery: Option<RecoverySequenceBlockRestore>,
892}
893
894#[derive(Clone, Debug, PartialEq, Eq)]
896pub struct SequenceClaimFrontier {
897 ledger: SequenceLedger,
898 movable_claims: Vec<MovableSequenceClaim>,
899 immutable_candidates: Vec<ImmutableSequenceCandidate>,
900 products: SequenceProductRanges,
901 recovery: Option<RecoverySequenceBlock>,
902}
903
904impl SequenceClaimFrontier {
905 #[must_use]
907 pub const fn ledger(&self) -> SequenceLedger {
908 self.ledger
909 }
910
911 #[must_use]
913 pub fn movable_claims(&self) -> &[MovableSequenceClaim] {
914 &self.movable_claims
915 }
916
917 #[must_use]
919 pub fn immutable_candidates(&self) -> &[ImmutableSequenceCandidate] {
920 &self.immutable_candidates
921 }
922
923 #[must_use]
925 pub const fn products(&self) -> &SequenceProductRanges {
926 &self.products
927 }
928
929 #[must_use]
931 pub const fn recovery(&self) -> Option<RecoverySequenceBlock> {
932 self.recovery
933 }
934}
935
936#[derive(Clone, Copy, Debug, PartialEq, Eq)]
938pub enum OrderDirectOwner {
939 ActiveBindingTerminal(BindingTerminalOwner),
941 MembershipExit {
943 participant_index: ParticipantId,
945 },
946}
947
948#[derive(Clone, Copy, Debug, PartialEq, Eq)]
950pub struct MovableOrderClaim {
951 pub transaction_order: TransactionOrder,
953 pub owner: OrderDirectOwner,
955}
956
957#[derive(Clone, Debug, PartialEq, Eq)]
962pub struct ImmutableOrderCandidateMajorRestore {
963 pub transaction_order: TransactionOrder,
965 pub candidate_keys: Vec<super::AdmissionOrder>,
967}
968
969#[derive(Clone, Debug, PartialEq, Eq)]
971pub struct ImmutableOrderCandidateMajor {
972 transaction_order: TransactionOrder,
973 candidate_keys: Vec<super::AdmissionOrder>,
974}
975
976impl ImmutableOrderCandidateMajor {
977 #[must_use]
979 pub const fn transaction_order(&self) -> TransactionOrder {
980 self.transaction_order
981 }
982
983 #[must_use]
985 pub fn candidate_keys(&self) -> &[super::AdmissionOrder] {
986 &self.candidate_keys
987 }
988}
989
990#[derive(Clone, Copy, Debug, PartialEq, Eq)]
992pub struct RecoveryOrderActiveBindingRestore {
993 pub transaction_order: TransactionOrder,
995 pub owner: BindingTerminalOwner,
997}
998
999#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1001pub struct RecoveryOrderBlockRestore {
1002 pub active_binding: Option<RecoveryOrderActiveBindingRestore>,
1004 pub recovery_operation_order: TransactionOrder,
1006 pub replacement_terminal_order: TransactionOrder,
1008}
1009
1010#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1012pub struct RecoveryOrderBlock {
1013 active_binding: Option<RecoveryOrderActiveBindingRestore>,
1014 recovery_operation_order: TransactionOrder,
1015 replacement_terminal_order: TransactionOrder,
1016 participant_index: ParticipantId,
1017 marker_delivery_seq: DeliverySeq,
1018 recovered_binding_epoch: BindingEpoch,
1019}
1020
1021impl RecoveryOrderBlock {
1022 #[must_use]
1024 pub const fn active_binding(self) -> Option<RecoveryOrderActiveBindingRestore> {
1025 self.active_binding
1026 }
1027
1028 #[must_use]
1030 pub const fn recovery_operation_order(self) -> TransactionOrder {
1031 self.recovery_operation_order
1032 }
1033
1034 #[must_use]
1036 pub const fn replacement_terminal_order(self) -> TransactionOrder {
1037 self.replacement_terminal_order
1038 }
1039
1040 #[must_use]
1042 pub const fn participant_index(self) -> ParticipantId {
1043 self.participant_index
1044 }
1045
1046 #[must_use]
1048 pub const fn marker_delivery_seq(self) -> DeliverySeq {
1049 self.marker_delivery_seq
1050 }
1051
1052 #[must_use]
1054 pub const fn recovered_binding_epoch(self) -> BindingEpoch {
1055 self.recovered_binding_epoch
1056 }
1057}
1058
1059#[derive(Clone, Debug, Default, PartialEq, Eq)]
1061pub struct OrderClaimFrontierRestore {
1062 pub movable_claims: Vec<MovableOrderClaim>,
1064 pub immutable_candidates: Vec<ImmutableOrderCandidateMajorRestore>,
1066 pub recovery: Option<RecoveryOrderBlockRestore>,
1068}
1069
1070#[derive(Clone, Debug, PartialEq, Eq)]
1072pub struct OrderClaimFrontier {
1073 ledger: OrderLedger,
1074 movable_claims: Vec<MovableOrderClaim>,
1075 immutable_candidates: Vec<ImmutableOrderCandidateMajor>,
1076 recovery: Option<RecoveryOrderBlock>,
1077}
1078
1079impl OrderClaimFrontier {
1080 #[must_use]
1082 pub const fn ledger(&self) -> OrderLedger {
1083 self.ledger
1084 }
1085
1086 #[must_use]
1088 pub fn movable_claims(&self) -> &[MovableOrderClaim] {
1089 &self.movable_claims
1090 }
1091
1092 #[must_use]
1094 pub fn immutable_candidates(&self) -> &[ImmutableOrderCandidateMajor] {
1095 &self.immutable_candidates
1096 }
1097
1098 #[must_use]
1100 pub const fn recovery(&self) -> Option<RecoveryOrderBlock> {
1101 self.recovery
1102 }
1103}
1104
1105#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1106enum RecoveryClaimPhase {
1107 PreFate,
1108 PostFate,
1109 RecoveredBound,
1110}
1111
1112#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1120pub struct RecoveryClaimProvenance {
1121 participant_index: ParticipantId,
1122 marker_delivery_seq: DeliverySeq,
1123 prior_binding_epoch: BindingEpoch,
1124 current_binding_epoch: BindingEpoch,
1125 phase: RecoveryClaimPhase,
1126}
1127
1128impl RecoveryClaimProvenance {
1129 #[must_use]
1132 pub const fn from_stored_edge(edge: super::StoredEdge) -> Option<Self> {
1133 match edge {
1134 super::StoredEdge::MarkerDelivery(delivery) => Some(Self {
1135 participant_index: delivery.participant_id(),
1136 marker_delivery_seq: delivery.marker_delivery_seq(),
1137 prior_binding_epoch: delivery.binding_epoch(),
1138 current_binding_epoch: delivery.binding_epoch(),
1139 phase: RecoveryClaimPhase::PreFate,
1140 }),
1141 super::StoredEdge::ParticipantCursorProgress(progress) => {
1142 let Some(marker_delivery_seq) = progress.marker_delivery_seq() else {
1143 return None;
1144 };
1145 Some(Self {
1146 participant_index: progress.participant_id(),
1147 marker_delivery_seq,
1148 prior_binding_epoch: progress.binding_epoch(),
1149 current_binding_epoch: progress.binding_epoch(),
1150 phase: RecoveryClaimPhase::PreFate,
1151 })
1152 }
1153 super::StoredEdge::DetachedCredentialRecovery(recovery) => Some(Self {
1154 participant_index: recovery.participant_id(),
1155 marker_delivery_seq: recovery.marker_delivery_seq(),
1156 prior_binding_epoch: recovery.prior_binding_epoch(),
1157 current_binding_epoch: recovery.prior_binding_epoch(),
1158 phase: RecoveryClaimPhase::PostFate,
1159 }),
1160 _ => None,
1161 }
1162 }
1163
1164 #[must_use]
1166 pub const fn participant_index(self) -> ParticipantId {
1167 self.participant_index
1168 }
1169
1170 #[must_use]
1172 pub const fn marker_delivery_seq(self) -> DeliverySeq {
1173 self.marker_delivery_seq
1174 }
1175
1176 #[must_use]
1178 pub const fn prior_binding_epoch(self) -> BindingEpoch {
1179 self.prior_binding_epoch
1180 }
1181}
1182
1183#[derive(Clone, Debug, PartialEq, Eq)]
1185pub struct ClaimFrontiersRestore {
1186 pub conversation_id: ConversationId,
1188 pub active_identities: Vec<FrontierParticipant>,
1190 pub identity_slot_limit: u64,
1192 pub retained_floor: u128,
1194 pub retained_record_limit: u64,
1196 pub retained_records: Vec<RetainedCausalRecord>,
1198 pub active_marker_anchors: Vec<DeliverySeq>,
1204 pub historical_marker_deliveries: Vec<HistoricalMarkerDeliveryFactRestore>,
1210 pub historical_causal_facts: Vec<HistoricalCausalFactRestore>,
1212 pub sequence: SequenceClaimFrontierRestore,
1214 pub order: OrderClaimFrontierRestore,
1216 pub recovery_marker_delivery_seq: Option<DeliverySeq>,
1222}
1223
1224#[derive(Debug)]
1231pub(super) struct ClaimFrontiersPrevalidated {
1232 conversation_id: ConversationId,
1233 active_identities: ActiveIdentityRanks,
1234 identity_slot_limit: u64,
1235 retained_floor: u128,
1236 retained_records: Vec<RetainedCausalRecord>,
1237 marker_records: Vec<RetainedCausalRecord>,
1238 historical_marker_deliveries: Vec<HistoricalMarkerDeliveryAuthority>,
1239 historical_causal_authorities: Vec<HistoricalCausalAuthority>,
1240 binding_origins: Vec<BindingOrigin>,
1241 sequence_restore: SequenceClaimFrontierRestore,
1242 order_restore: OrderClaimFrontierRestore,
1243 recovery_marker_delivery_seq: Option<DeliverySeq>,
1244 sequence_ledger: SequenceLedger,
1245 order_ledger: OrderLedger,
1246 issued_marker_record: Option<MarkerRecordRequest>,
1247}
1248
1249#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1250enum MarkerRecordUse {
1251 Planned(FrontierBinding),
1252 Delivered(FrontierBinding),
1253 Recovered {
1254 prior_binding_epoch: BindingEpoch,
1255 recovered_binding_epoch: BindingEpoch,
1256 },
1257}
1258
1259#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1261pub(super) struct MarkerRecordRequest {
1262 participant_index: ParticipantId,
1263 marker_delivery_seq: DeliverySeq,
1264 use_kind: MarkerRecordUse,
1265}
1266
1267impl MarkerRecordRequest {
1268 pub(super) const fn planned(
1270 participant_index: ParticipantId,
1271 marker_delivery_seq: DeliverySeq,
1272 target: FrontierBinding,
1273 ) -> Self {
1274 Self {
1275 participant_index,
1276 marker_delivery_seq,
1277 use_kind: MarkerRecordUse::Planned(target),
1278 }
1279 }
1280
1281 pub(super) const fn delivered(
1283 participant_index: ParticipantId,
1284 marker_delivery_seq: DeliverySeq,
1285 target: FrontierBinding,
1286 ) -> Self {
1287 Self {
1288 participant_index,
1289 marker_delivery_seq,
1290 use_kind: MarkerRecordUse::Delivered(target),
1291 }
1292 }
1293
1294 pub(super) const fn recovered(
1296 participant_index: ParticipantId,
1297 marker_delivery_seq: DeliverySeq,
1298 prior_binding_epoch: BindingEpoch,
1299 recovered_binding_epoch: BindingEpoch,
1300 ) -> Self {
1301 Self {
1302 participant_index,
1303 marker_delivery_seq,
1304 use_kind: MarkerRecordUse::Recovered {
1305 prior_binding_epoch,
1306 recovered_binding_epoch,
1307 },
1308 }
1309 }
1310}
1311
1312#[derive(Debug)]
1314pub(super) struct MarkerDrainCore {
1315 frontiers: ClaimFrontiers,
1316 candidate: ValidatedMarkerCandidate,
1317 record: ValidatedMarkerRecord,
1318}
1319
1320impl MarkerDrainCore {
1321 pub(super) fn into_parts(
1324 self,
1325 ) -> (
1326 ClaimFrontiers,
1327 ValidatedMarkerCandidate,
1328 ValidatedMarkerRecord,
1329 ) {
1330 (self.frontiers, self.candidate, self.record)
1331 }
1332}
1333
1334#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1336pub(super) enum MarkerDrainCoreError {
1337 NoCandidate,
1339 BindingTerminalFirst,
1341 SequenceNotNext,
1343 CausalMajorNotAllocated,
1345 MissingOrderCandidate,
1347 ResultingLedger,
1349}
1350
1351#[derive(Debug, PartialEq, Eq)]
1353pub struct ClaimFrontiers {
1354 conversation_id: ConversationId,
1355 active_identities: ActiveIdentityRanks,
1356 identity_slot_limit: u64,
1357 retained_floor: u128,
1358 retained_records: Vec<RetainedCausalRecord>,
1359 marker_records: Vec<RetainedCausalRecord>,
1360 fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1361 sequence: SequenceClaimFrontier,
1362 order: OrderClaimFrontier,
1363}
1364
1365#[derive(Debug, PartialEq, Eq)]
1371pub struct InitialEnrollmentFrontierCommit<F> {
1372 operation: InitialEnrollmentOperationCommit<F>,
1373 frontiers: ClaimFrontiers,
1374 closure_accounting: ClosureAccounting,
1375 attached_charge: ResourceVector,
1376}
1377
1378impl<F> InitialEnrollmentFrontierCommit<F> {
1379 #[must_use]
1381 pub const fn operation(&self) -> &InitialEnrollmentOperationCommit<F> {
1382 &self.operation
1383 }
1384
1385 #[must_use]
1387 pub const fn frontiers(&self) -> &ClaimFrontiers {
1388 &self.frontiers
1389 }
1390
1391 #[must_use]
1393 pub const fn closure_accounting(&self) -> ClosureAccounting {
1394 self.closure_accounting
1395 }
1396
1397 #[must_use]
1399 pub const fn attached_charge(&self) -> ResourceVector {
1400 self.attached_charge
1401 }
1402
1403 #[allow(
1405 dead_code,
1406 reason = "the next conversation event body consumes this sealed operation/frontier unit"
1407 )]
1408 pub(in crate::lifecycle) fn into_conversation_parts(
1409 self,
1410 ) -> (
1411 InitialEnrollmentOperationCommit<F>,
1412 ClaimFrontiers,
1413 ClosureAccounting,
1414 ResourceVector,
1415 ) {
1416 (
1417 self.operation,
1418 self.frontiers,
1419 self.closure_accounting,
1420 self.attached_charge,
1421 )
1422 }
1423}
1424
1425#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1427pub enum InitialEnrollmentFrontierError {
1428 AttachedChargeMismatch {
1430 expected: ResourceVector,
1432 actual: ResourceVector,
1434 },
1435 EnrollmentShape,
1437 LedgerShape,
1439 ClosureProjection,
1441 PositionOverflow,
1443 FrontierInvariant,
1445}
1446
1447#[derive(Debug, PartialEq, Eq)]
1449pub struct InitialEnrollmentFrontierFailure<F> {
1450 operation: InitialEnrollmentOperationCommit<F>,
1451 error: InitialEnrollmentFrontierError,
1452}
1453
1454impl<F> InitialEnrollmentFrontierFailure<F> {
1455 #[must_use]
1457 pub const fn error(&self) -> InitialEnrollmentFrontierError {
1458 self.error
1459 }
1460
1461 #[allow(
1463 dead_code,
1464 reason = "the conversation decision layer recovers or terminalizes the speculative enrollment"
1465 )]
1466 pub(in crate::lifecycle) fn into_operation(self) -> InitialEnrollmentOperationCommit<F> {
1467 self.operation
1468 }
1469}
1470
1471#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1473pub enum PrepareLeaveAuthorityError {
1474 Conversation,
1476 Identity,
1478 Binding,
1480 ImmutablePrefix,
1482 PendingCandidate,
1484 MembershipExitClaim,
1486 ActiveBindingClaim,
1488 OrderCapacity,
1490 ResultingOrderLedger,
1492}
1493
1494#[derive(Debug, PartialEq, Eq)]
1500pub struct ValidatedMarkerCandidate {
1501 conversation_id: ConversationId,
1502 candidate: MarkerCandidateAuthority,
1503 seal: MarkerAuthoritySeal,
1504}
1505
1506#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1509pub(super) struct FencedMarkerSourceRecord {
1510 pub(super) conversation_id: ConversationId,
1511 pub(super) delivery_seq: DeliverySeq,
1512 pub(super) admission_order: super::AdmissionOrder,
1513 pub(super) participant_id: ParticipantId,
1514 pub(super) provenance: MarkerProvenance,
1515 pub(super) target_binding: FrontierBinding,
1516}
1517
1518#[derive(Debug, PartialEq, Eq)]
1523pub struct ValidatedMarkerRecord {
1524 conversation_id: ConversationId,
1525 record: RetainedCausalRecord,
1526 provenance: MarkerProvenance,
1527 target_binding: FrontierBinding,
1528 occurrence: MarkerRecordOccurrence,
1529 seal: MarkerAuthoritySeal,
1530}
1531
1532#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1539pub(super) enum MarkerRecordOccurrence {
1540 Undelivered,
1542 Delivered,
1544}
1545
1546#[derive(Debug, PartialEq, Eq)]
1547enum MarkerAuthoritySeal {
1548 Validated,
1549}
1550
1551#[doc(hidden)]
1557pub fn with_validated_marker_record_type<F>(probe: F)
1558where
1559 F: FnOnce(ValidatedMarkerRecord),
1560{
1561 drop(probe);
1562}
1563
1564impl ValidatedMarkerRecord {
1565 pub(super) const fn consume(self) {
1567 match self.seal {
1568 MarkerAuthoritySeal::Validated => {}
1569 }
1570 }
1571
1572 #[must_use]
1574 pub const fn conversation_id(&self) -> ConversationId {
1575 self.conversation_id
1576 }
1577
1578 #[must_use]
1580 pub const fn participant_id(&self) -> ParticipantId {
1581 self.record.admission_order.participant_index()
1582 }
1583
1584 #[must_use]
1586 pub const fn delivery_seq(&self) -> DeliverySeq {
1587 self.record.delivery_seq
1588 }
1589
1590 #[must_use]
1592 pub const fn admission_order(&self) -> super::AdmissionOrder {
1593 self.record.admission_order
1594 }
1595
1596 #[must_use]
1598 pub const fn provenance(&self) -> MarkerProvenance {
1599 self.provenance
1600 }
1601
1602 #[must_use]
1604 pub const fn target_binding(&self) -> FrontierBinding {
1605 self.target_binding
1606 }
1607
1608 #[must_use]
1610 pub const fn binding_epoch(&self) -> BindingEpoch {
1611 binding_epoch(self.target_binding)
1612 }
1613
1614 #[must_use]
1616 pub(super) const fn occurrence(&self) -> MarkerRecordOccurrence {
1617 self.occurrence
1618 }
1619
1620 #[cfg(any(test, feature = "test-support"))]
1622 pub(super) const fn delivered_for_test(mut self) -> Self {
1623 self.occurrence = MarkerRecordOccurrence::Delivered;
1624 self
1625 }
1626}
1627
1628impl ValidatedMarkerCandidate {
1629 pub(super) const fn consume(self) {
1631 match self.seal {
1632 MarkerAuthoritySeal::Validated => {}
1633 }
1634 }
1635
1636 #[must_use]
1638 pub(super) const fn conversation_id(&self) -> ConversationId {
1639 self.conversation_id
1640 }
1641
1642 #[must_use]
1644 pub(super) const fn participant_index(&self) -> ParticipantId {
1645 self.candidate.admission_order.participant_index()
1646 }
1647
1648 #[must_use]
1650 pub(super) const fn participant_id(&self) -> ParticipantId {
1651 self.participant_index()
1652 }
1653
1654 #[must_use]
1656 pub(super) const fn delivery_seq(&self) -> DeliverySeq {
1657 self.candidate.delivery_seq
1658 }
1659
1660 #[must_use]
1662 pub(super) const fn target_binding(&self) -> FrontierBinding {
1663 self.candidate.target_binding
1664 }
1665
1666 #[must_use]
1668 pub(super) const fn provenance(&self) -> MarkerProvenance {
1669 self.candidate.provenance
1670 }
1671
1672 pub(super) const fn abandoned_after(&self) -> DeliverySeq {
1673 self.candidate.abandoned_after
1674 }
1675
1676 pub(super) const fn abandoned_through(&self) -> DeliverySeq {
1677 self.candidate.abandoned_through
1678 }
1679
1680 pub(super) const fn physical_floor_at_decision(&self) -> DeliverySeq {
1681 self.candidate.physical_floor_at_decision
1682 }
1683}
1684
1685#[derive(Clone, Copy)]
1686struct InitialEnrollmentFrontierShape {
1687 conversation_id: ConversationId,
1688 binding_epoch: BindingEpoch,
1689 identity_slot_limit: u64,
1690 retained_floor: u128,
1691 attached: AttachedLifecycleRecord,
1692 order_ledger: OrderLedger,
1693 sequence_ledger: SequenceLedger,
1694 closure_accounting: ClosureAccounting,
1695}
1696
1697#[derive(Clone, Copy)]
1698struct InitialFrontierPositions {
1699 terminal_sequence: DeliverySeq,
1700 exit_sequence: DeliverySeq,
1701 product_sequence: DeliverySeq,
1702 active_terminal_order: TransactionOrder,
1703 exit_order: TransactionOrder,
1704}
1705
1706fn initial_enrollment_frontier_shape<F>(
1707 operation: &InitialEnrollmentOperationCommit<F>,
1708 attached_charge: ResourceVector,
1709) -> Result<InitialEnrollmentFrontierShape, InitialEnrollmentFrontierError> {
1710 let projection = operation.closure_projection();
1711 let expected_charge = projection.resulting_retained_charge();
1712 if attached_charge != expected_charge {
1713 return Err(InitialEnrollmentFrontierError::AttachedChargeMismatch {
1714 expected: expected_charge,
1715 actual: attached_charge,
1716 });
1717 }
1718 let enrollment = operation.enrollment();
1719 let attached = enrollment.attached;
1720 let BindingState::Bound(binding) = enrollment.binding_state else {
1721 return Err(InitialEnrollmentFrontierError::EnrollmentShape);
1722 };
1723 if enrollment.member.participant_id() != 0
1724 || enrollment.member.cursor() != 0
1725 || binding.participant_id != 0
1726 || attached.participant_id() != 0
1727 || binding.conversation_id != enrollment.member.conversation_id()
1728 || attached.conversation_id() != binding.conversation_id
1729 || attached.binding_epoch() != binding.binding_epoch
1730 || projection.participant_index() != 0
1731 || projection.binding_epoch() != binding.binding_epoch
1732 || projection.identity_slots() == 0
1733 {
1734 return Err(InitialEnrollmentFrontierError::EnrollmentShape);
1735 }
1736 let admission_order = attached.admission_order();
1737 let order_ledger = operation.order().resulting();
1738 let sequence_ledger = operation.sequence().resulting();
1739 let order_claims = order_ledger.claims();
1740 let sequence_claims = sequence_ledger.claims();
1741 if operation.order().major() != admission_order.transaction_order()
1742 || !matches!(order_ledger.high(), OrderHigh::Allocated(value) if value == admission_order.transaction_order())
1743 || order_claims.active_binding_terminals() != 1
1744 || order_claims.membership_exits() != 1
1745 || order_claims.recovery_operation()
1746 || order_claims.recovery_replacement_terminal()
1747 || sequence_ledger.high_watermark() != attached.delivery_seq()
1748 || sequence_claims.live_members() != 1
1749 || sequence_claims.binding_terminals() != 1
1750 || sequence_claims.markers() != 0
1751 || sequence_claims.recovery() != RecoverySequenceReserve::None
1752 || admission_order.candidate_phase() != CandidatePhase::AttachLifecycle
1753 {
1754 return Err(InitialEnrollmentFrontierError::LedgerShape);
1755 }
1756 Ok(InitialEnrollmentFrontierShape {
1757 conversation_id: binding.conversation_id,
1758 binding_epoch: binding.binding_epoch,
1759 identity_slot_limit: projection.identity_slots(),
1760 retained_floor: projection.resulting_floor(),
1761 attached,
1762 order_ledger,
1763 sequence_ledger,
1764 closure_accounting: validate_initial_enrollment_closure(operation, projection, attached)?,
1765 })
1766}
1767
1768fn validate_initial_enrollment_closure<F>(
1769 operation: &InitialEnrollmentOperationCommit<F>,
1770 projection: &InitialEnrollmentClosureProjection,
1771 attached: AttachedLifecycleRecord,
1772) -> Result<ClosureAccounting, InitialEnrollmentFrontierError> {
1773 let accounting = projection.resulting_closure_accounting();
1774 let state_matches = match accounting.state() {
1775 ClosureState::Clear => {
1776 projection.debt().is_zero()
1777 && projection.remaining_recovery_claim() == ResourceVector::default()
1778 }
1779 ClosureState::Owed {
1780 debt,
1781 edge: StoredEdge::ObserverProjection(observer),
1782 } => {
1783 debt.value() == projection.debt()
1784 && observer == ObserverProjection::new(attached.delivery_seq())
1785 && projection.remaining_recovery_claim() == accounting.edge_k_remaining()
1786 }
1787 ClosureState::Owed { .. } => false,
1788 };
1789 if projection.resulting_floor() != 1
1790 || projection.resulting_floor() != operation.observer_floor().cap_floor()
1791 || operation.observer_floor().observer_progress() != 0
1792 || projection.recovery_quartet() != RecoveryQuartetStatus::None
1793 || !projection.new_marker_candidates().is_empty()
1794 || accounting.marker_capacity_credits() != 0
1795 || accounting.marker_anchors() != 0
1796 || accounting.edge_sequence_claims() != 0
1797 || accounting.edge_order_position_claims() != 0
1798 || accounting.baseline() != projection.resulting_baseline()
1799 || !state_matches
1800 {
1801 Err(InitialEnrollmentFrontierError::ClosureProjection)
1802 } else {
1803 Ok(accounting)
1804 }
1805}
1806
1807fn initial_frontier_positions(
1808 attached: AttachedLifecycleRecord,
1809) -> Result<InitialFrontierPositions, InitialEnrollmentFrontierError> {
1810 let terminal_sequence = attached
1811 .delivery_seq()
1812 .checked_add(1)
1813 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1814 let exit_sequence = terminal_sequence
1815 .checked_add(1)
1816 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1817 let product_sequence = exit_sequence
1818 .checked_add(1)
1819 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1820 let active_terminal_order = attached
1821 .admission_order()
1822 .transaction_order()
1823 .checked_add(1)
1824 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1825 let exit_order = active_terminal_order
1826 .checked_add(1)
1827 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1828 Ok(InitialFrontierPositions {
1829 terminal_sequence,
1830 exit_sequence,
1831 product_sequence,
1832 active_terminal_order,
1833 exit_order,
1834 })
1835}
1836
1837fn initial_sequence_frontier(
1838 shape: &InitialEnrollmentFrontierShape,
1839 positions: InitialFrontierPositions,
1840 terminal: BindingTerminalOwner,
1841) -> SequenceClaimFrontier {
1842 SequenceClaimFrontier {
1843 ledger: shape.sequence_ledger,
1844 movable_claims: alloc::vec![
1845 MovableSequenceClaim {
1846 delivery_seq: positions.terminal_sequence,
1847 owner: SequenceDirectOwner::BindingTerminal(terminal),
1848 },
1849 MovableSequenceClaim {
1850 delivery_seq: positions.exit_sequence,
1851 owner: SequenceDirectOwner::MembershipExit {
1852 participant_index: 0,
1853 },
1854 },
1855 ],
1856 immutable_candidates: Vec::new(),
1857 products: SequenceProductRanges {
1858 live_times_terminal: alloc::vec![TerminalProductRange {
1859 start: positions.product_sequence,
1860 length: 1,
1861 terminal,
1862 }],
1863 live_times_replacement_terminal: None,
1864 other_live_times_exit: Vec::new(),
1865 },
1866 recovery: None,
1867 }
1868}
1869
1870fn initial_order_frontier(
1871 shape: &InitialEnrollmentFrontierShape,
1872 positions: InitialFrontierPositions,
1873 terminal: BindingTerminalOwner,
1874) -> OrderClaimFrontier {
1875 OrderClaimFrontier {
1876 ledger: shape.order_ledger,
1877 movable_claims: alloc::vec![
1878 MovableOrderClaim {
1879 transaction_order: positions.active_terminal_order,
1880 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
1881 },
1882 MovableOrderClaim {
1883 transaction_order: positions.exit_order,
1884 owner: OrderDirectOwner::MembershipExit {
1885 participant_index: 0,
1886 },
1887 },
1888 ],
1889 immutable_candidates: Vec::new(),
1890 recovery: None,
1891 }
1892}
1893
1894fn build_initial_enrollment_frontiers(
1895 shape: &InitialEnrollmentFrontierShape,
1896) -> Result<ClaimFrontiers, InitialEnrollmentFrontierError> {
1897 let positions = initial_frontier_positions(shape.attached)?;
1898 let terminal = BindingTerminalOwner {
1899 participant_index: 0,
1900 binding_epoch: shape.binding_epoch,
1901 };
1902 let sequence = initial_sequence_frontier(shape, positions, terminal);
1903 let order = initial_order_frontier(shape, positions, terminal);
1904 validate_cross_counter(&sequence, &order)
1905 .map_err(|_| InitialEnrollmentFrontierError::FrontierInvariant)?;
1906 Ok(ClaimFrontiers {
1907 conversation_id: shape.conversation_id,
1908 active_identities: ActiveIdentityRanks {
1909 participants: alloc::vec![FrontierParticipant::new(
1910 0,
1911 0,
1912 FrontierBinding::Bound(shape.binding_epoch),
1913 )],
1914 },
1915 identity_slot_limit: shape.identity_slot_limit,
1916 retained_floor: shape.retained_floor,
1917 retained_records: alloc::vec![RetainedCausalRecord {
1918 delivery_seq: shape.attached.delivery_seq(),
1919 admission_order: shape.attached.admission_order(),
1920 kind: RetainedCausalRecordKind::AttachLifecycle {
1921 participant_index: 0,
1922 binding_epoch: shape.binding_epoch,
1923 },
1924 }],
1925 marker_records: Vec::new(),
1926 fenced_marker_issued: None,
1927 sequence,
1928 order,
1929 })
1930}
1931
1932#[derive(Clone, Copy)]
1933enum LeaveSequenceUnit {
1934 Direct(MovableSequenceClaim),
1935 TerminalProduct(TerminalProductRange),
1936 ReplacementProduct(ReplacementTerminalProductRange),
1937 ExitProduct(ExitProductRange),
1938 Recovery(RecoverySequenceBlock),
1939}
1940
1941impl LeaveSequenceUnit {
1942 fn original_start(self) -> DeliverySeq {
1943 match self {
1944 Self::Direct(claim) => claim.delivery_seq,
1945 Self::TerminalProduct(range) => range.start,
1946 Self::ReplacementProduct(range) => range.start,
1947 Self::ExitProduct(range) => range.start,
1948 Self::Recovery(block) => block_start_validated_sequence(block),
1949 }
1950 }
1951}
1952
1953fn allocate_leave_sequence_range(
1954 cursor: &mut Option<DeliverySeq>,
1955 length: u64,
1956) -> Result<DeliverySeq, LeaveCommitError> {
1957 let Some(start) = *cursor else {
1958 return Err(LeaveCommitError::ResultingFrontier);
1959 };
1960 let end = u128::from(start)
1961 .checked_add(u128::from(length))
1962 .and_then(|value| value.checked_sub(1))
1963 .ok_or(LeaveCommitError::ResultingFrontier)?;
1964 if end > u128::from(u64::MAX) {
1965 return Err(LeaveCommitError::ResultingFrontier);
1966 }
1967 *cursor = u64::try_from(end + 1).ok();
1968 Ok(start)
1969}
1970
1971#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1973pub(in crate::lifecycle) enum LiveFrontierTransitionError {
1974 Authority,
1976 Precedence,
1978 RecordPosition,
1980 Exhausted,
1982 ResultingFrontier,
1984}
1985
1986fn select_retained_marker_records(records: &[RetainedCausalRecord]) -> Vec<RetainedCausalRecord> {
1987 records
1988 .iter()
1989 .copied()
1990 .filter(|record| {
1991 matches!(
1992 record.kind,
1993 RetainedCausalRecordKind::CompactionMarker { .. }
1994 )
1995 })
1996 .collect()
1997}
1998
1999impl ClaimFrontiers {
2000 pub fn from_initial_enrollment<F>(
2015 operation: InitialEnrollmentOperationCommit<F>,
2016 attached_charge: ResourceVector,
2017 ) -> Result<InitialEnrollmentFrontierCommit<F>, Box<InitialEnrollmentFrontierFailure<F>>> {
2018 let shape = match initial_enrollment_frontier_shape(&operation, attached_charge) {
2019 Ok(shape) => shape,
2020 Err(error) => {
2021 return Err(Box::new(InitialEnrollmentFrontierFailure {
2022 operation,
2023 error,
2024 }));
2025 }
2026 };
2027 let closure_accounting = shape.closure_accounting;
2028 let frontiers = match build_initial_enrollment_frontiers(&shape) {
2029 Ok(frontiers) => frontiers,
2030 Err(error) => {
2031 return Err(Box::new(InitialEnrollmentFrontierFailure {
2032 operation,
2033 error,
2034 }));
2035 }
2036 };
2037 Ok(InitialEnrollmentFrontierCommit {
2038 operation,
2039 frontiers,
2040 closure_accounting,
2041 attached_charge,
2042 })
2043 }
2044
2045 pub fn restore(
2058 restore: ClaimFrontiersRestore,
2059 sequence_ledger: SequenceLedger,
2060 order_ledger: OrderLedger,
2061 ) -> Result<Self, ParticipantStateCorruptReason> {
2062 let history = ValidatedConversationHistory::empty();
2063 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)?
2064 .finish(None)
2065 }
2066
2067 #[cfg(any(test, feature = "test-support"))]
2070 pub(super) fn prevalidate(
2071 restore: ClaimFrontiersRestore,
2072 sequence_ledger: SequenceLedger,
2073 order_ledger: OrderLedger,
2074 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2075 let history = ValidatedConversationHistory::empty();
2076 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)
2077 }
2078
2079 pub(super) fn prevalidate_with_history(
2080 restore: ClaimFrontiersRestore,
2081 sequence_ledger: SequenceLedger,
2082 order_ledger: OrderLedger,
2083 history: &ValidatedConversationHistory,
2084 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2085 validate_sequence_numeric(&restore.sequence, sequence_ledger).map_err(corrupt_frontier)?;
2086 validate_order_numeric(&restore.order, order_ledger).map_err(corrupt_frontier)?;
2087 validate_unique_candidate_keys(
2088 &restore.sequence.immutable_candidates,
2089 &restore.retained_records,
2090 )?;
2091 validate_bounded_shape(&restore, sequence_ledger).map_err(corrupt_frontier)?;
2092 let active_identities = ActiveIdentityRanks::try_new(
2093 restore.active_identities,
2094 sequence_ledger.high_watermark(),
2095 restore.identity_slot_limit,
2096 )
2097 .map_err(corrupt_frontier)?;
2098 let retained_records = validated_retained_records(
2099 restore.retained_records,
2100 restore.retained_floor,
2101 restore.retained_record_limit,
2102 restore.identity_slot_limit,
2103 sequence_ledger,
2104 )
2105 .map_err(corrupt_frontier)?;
2106 let historical_causal_authorities = validated_historical_authorities(
2107 restore.historical_causal_facts,
2108 restore.conversation_id,
2109 restore.identity_slot_limit,
2110 sequence_ledger,
2111 history,
2112 )
2113 .map_err(corrupt_frontier)?;
2114 let retained_marker_records = select_retained_marker_records(&retained_records);
2115 let marker_records = validated_active_marker_records(
2116 &retained_marker_records,
2117 restore.active_marker_anchors,
2118 restore.identity_slot_limit,
2119 sequence_ledger,
2120 )
2121 .map_err(corrupt_frontier)?;
2122 let historical_marker_deliveries = validated_historical_marker_deliveries(
2123 restore.historical_marker_deliveries,
2124 restore.conversation_id,
2125 &active_identities,
2126 &MarkerDeliverySources {
2127 retained: &retained_records,
2128 historical: &historical_causal_authorities,
2129 candidates: &restore.sequence.immutable_candidates,
2130 },
2131 restore.retained_record_limit,
2132 sequence_ledger,
2133 )
2134 .map_err(corrupt_frontier)?;
2135 BindingOriginValidation {
2136 conversation_id: restore.conversation_id,
2137 active: &active_identities,
2138 origins: &history.binding_origins,
2139 retained_records: &retained_records,
2140 causal_authorities: &history.causal_authorities,
2141 historical_marker_deliveries: &historical_marker_deliveries,
2142 total: history.total,
2143 ledger: sequence_ledger,
2144 }
2145 .validate()
2146 .map_err(corrupt_frontier)?;
2147 validate_sequence_candidates(
2148 &active_identities,
2149 &restore.sequence.immutable_candidates,
2150 restore.retained_floor,
2151 &retained_records,
2152 &historical_causal_authorities,
2153 sequence_ledger,
2154 )
2155 .map_err(corrupt_frontier)?;
2156 validate_marker_credit_owners(
2157 &restore.sequence.immutable_candidates,
2158 &marker_records,
2159 restore.identity_slot_limit,
2160 sequence_ledger,
2161 )
2162 .map_err(corrupt_frontier)?;
2163 Ok(ClaimFrontiersPrevalidated {
2164 conversation_id: restore.conversation_id,
2165 active_identities,
2166 identity_slot_limit: restore.identity_slot_limit,
2167 retained_floor: restore.retained_floor,
2168 retained_records,
2169 marker_records,
2170 historical_marker_deliveries,
2171 historical_causal_authorities,
2172 binding_origins: history.binding_origins.clone(),
2173 sequence_restore: restore.sequence,
2174 order_restore: restore.order,
2175 recovery_marker_delivery_seq: restore.recovery_marker_delivery_seq,
2176 sequence_ledger,
2177 order_ledger,
2178 issued_marker_record: None,
2179 })
2180 }
2181
2182 #[must_use]
2184 pub const fn active_identities(&self) -> &ActiveIdentityRanks {
2185 &self.active_identities
2186 }
2187
2188 #[must_use]
2190 pub const fn identity_slot_limit(&self) -> u64 {
2191 self.identity_slot_limit
2192 }
2193
2194 #[must_use]
2196 pub const fn conversation_id(&self) -> ConversationId {
2197 self.conversation_id
2198 }
2199
2200 #[must_use]
2202 pub const fn retained_floor(&self) -> u128 {
2203 self.retained_floor
2204 }
2205
2206 #[must_use]
2208 pub fn retained_records(&self) -> &[RetainedCausalRecord] {
2209 &self.retained_records
2210 }
2211
2212 #[must_use]
2214 pub fn retained_marker_records(&self) -> &[RetainedCausalRecord] {
2215 &self.marker_records
2216 }
2217
2218 #[must_use]
2226 pub fn project_offered_marker_progress(
2227 &self,
2228 participant_id: ParticipantId,
2229 binding_epoch: BindingEpoch,
2230 marker_delivery_seq: DeliverySeq,
2231 event: Event,
2232 ) -> Option<ParticipantCursorProgress> {
2233 let record = self
2234 .marker_records
2235 .iter()
2236 .find(|record| record.delivery_seq == marker_delivery_seq)
2237 .copied()?;
2238 let RetainedCausalRecordKind::CompactionMarker {
2239 participant_index,
2240 provenance,
2241 } = record.kind
2242 else {
2243 return None;
2244 };
2245 if participant_index != participant_id {
2246 return None;
2247 }
2248 let participant = active_participant(&self.active_identities, participant_id)?;
2249 let target_binding = FrontierBinding::Bound(binding_epoch);
2250 if participant.binding != target_binding {
2251 return None;
2252 }
2253 let authority = ValidatedMarkerRecord {
2254 conversation_id: self.conversation_id,
2255 record,
2256 provenance,
2257 target_binding,
2258 occurrence: MarkerRecordOccurrence::Undelivered,
2259 seal: MarkerAuthoritySeal::Validated,
2260 };
2261 MarkerDelivery::from_validated_record(&authority)
2262 .delivered_progress(event)
2263 .ok()
2264 }
2265
2266 pub(super) fn fenced_marker_source(
2269 &self,
2270 recovery: super::DetachedCredentialRecovery,
2271 ) -> Option<FencedMarkerSourceRecord> {
2272 if self.fenced_marker_issued.is_some() || recovery.conversation_id() != self.conversation_id
2273 {
2274 return None;
2275 }
2276 let marker_delivery_seq = recovery.marker_delivery_seq();
2277 let participant_id = recovery.participant_id();
2278 let prior_binding_epoch = recovery.prior_binding_epoch();
2279 let record = self
2280 .marker_records
2281 .iter()
2282 .find(|record| record.delivery_seq == marker_delivery_seq)
2283 .copied()?;
2284 let RetainedCausalRecordKind::CompactionMarker {
2285 participant_index,
2286 provenance,
2287 } = record.kind
2288 else {
2289 return None;
2290 };
2291 let participant = active_participant(&self.active_identities, participant_id)?;
2292 let target_binding = FrontierBinding::Detached(prior_binding_epoch);
2293 if participant_index != participant_id || participant.binding != target_binding {
2294 return None;
2295 }
2296 Some(FencedMarkerSourceRecord {
2297 conversation_id: self.conversation_id,
2298 delivery_seq: record.delivery_seq,
2299 admission_order: record.admission_order,
2300 participant_id,
2301 provenance,
2302 target_binding,
2303 })
2304 }
2305
2306 pub(in crate::lifecycle) fn take_fenced_marker_record(
2315 &mut self,
2316 recovery: super::DetachedCredentialRecovery,
2317 ) -> Option<ValidatedMarkerRecord> {
2318 let source = self.fenced_marker_source(recovery)?;
2319 self.fenced_marker_issued = Some((
2320 source.participant_id,
2321 source.delivery_seq,
2322 recovery.prior_binding_epoch(),
2323 ));
2324 Some(ValidatedMarkerRecord {
2325 conversation_id: source.conversation_id,
2326 record: RetainedCausalRecord {
2327 delivery_seq: source.delivery_seq,
2328 admission_order: source.admission_order,
2329 kind: RetainedCausalRecordKind::CompactionMarker {
2330 participant_index: source.participant_id,
2331 provenance: source.provenance,
2332 },
2333 },
2334 provenance: source.provenance,
2335 target_binding: source.target_binding,
2336 occurrence: MarkerRecordOccurrence::Delivered,
2337 seal: MarkerAuthoritySeal::Validated,
2338 })
2339 }
2340
2341 pub(in crate::lifecycle) const fn reinstall_fenced_marker_record(
2344 &mut self,
2345 record: ValidatedMarkerRecord,
2346 ) {
2347 self.fenced_marker_issued = None;
2348 record.consume();
2349 }
2350
2351 #[must_use]
2353 pub const fn sequence(&self) -> &SequenceClaimFrontier {
2354 &self.sequence
2355 }
2356
2357 #[must_use]
2359 pub const fn order(&self) -> &OrderClaimFrontier {
2360 &self.order
2361 }
2362
2363 #[cfg(test)]
2364 pub(in crate::lifecycle) fn cross_counter_valid_for_test(&self) -> bool {
2365 validate_cross_counter(&self.sequence, &self.order).is_ok()
2366 }
2367
2368 pub(in crate::lifecycle) fn apply_live_transition(
2374 self,
2375 active_identities: Vec<FrontierParticipant>,
2376 appended_records: &[RetainedCausalRecord],
2377 sequence_ledger: SequenceLedger,
2378 order_ledger: OrderLedger,
2379 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2380 if !self.sequence.immutable_candidates.is_empty()
2381 || self.sequence.recovery.is_some()
2382 || !self.order.immutable_candidates.is_empty()
2383 || self.order.recovery.is_some()
2384 {
2385 return Err(Box::new((self, LiveFrontierTransitionError::Precedence)));
2386 }
2387 let Ok(active) = ActiveIdentityRanks::try_new(
2388 active_identities,
2389 sequence_ledger.high_watermark(),
2390 self.identity_slot_limit,
2391 ) else {
2392 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2393 };
2394 let first_sequence = self.sequence.ledger.high_watermark().checked_add(1);
2395 if first_sequence.is_none_or(|first| {
2396 appended_records.iter().enumerate().any(|(index, record)| {
2397 u64::try_from(index)
2398 .ok()
2399 .and_then(|offset| first.checked_add(offset))
2400 != Some(record.delivery_seq)
2401 })
2402 }) || appended_records
2403 .last()
2404 .is_some_and(|record| record.delivery_seq != sequence_ledger.high_watermark())
2405 {
2406 return Err(Box::new((
2407 self,
2408 LiveFrontierTransitionError::RecordPosition,
2409 )));
2410 }
2411 let (sequence, order) =
2412 match rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger) {
2413 Ok(frontiers) => frontiers,
2414 Err(error) => return Err(Box::new((self, error))),
2415 };
2416 let Self {
2417 conversation_id,
2418 identity_slot_limit,
2419 retained_floor,
2420 mut retained_records,
2421 marker_records,
2422 fenced_marker_issued,
2423 ..
2424 } = self;
2425 retained_records.extend_from_slice(appended_records);
2426 Ok(Self {
2427 conversation_id,
2428 active_identities: active,
2429 identity_slot_limit,
2430 retained_floor,
2431 retained_records,
2432 marker_records,
2433 fenced_marker_issued,
2434 sequence,
2435 order,
2436 })
2437 }
2438
2439 pub(in crate::lifecycle) fn apply_pending_binding_terminal(
2443 mut self,
2444 participant_id: ParticipantId,
2445 binding_epoch: BindingEpoch,
2446 delivery_seq: DeliverySeq,
2447 admission_order: super::AdmissionOrder,
2448 order_ledger: OrderLedger,
2449 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2450 if !self.sequence.immutable_candidates.is_empty()
2451 || self.sequence.recovery.is_some()
2452 || !self.order.immutable_candidates.is_empty()
2453 || self.order.recovery.is_some()
2454 {
2455 return Err(Box::new((self, LiveFrontierTransitionError::Precedence)));
2456 }
2457 let mut participants = self.active_identities.participants().to_vec();
2458 let Some(participant) = participants
2459 .iter_mut()
2460 .find(|participant| participant.participant_index() == participant_id)
2461 else {
2462 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2463 };
2464 if participant.binding() != FrontierBinding::Bound(binding_epoch) {
2465 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2466 }
2467 *participant = FrontierParticipant::new(
2468 participant_id,
2469 participant.cursor(),
2470 FrontierBinding::Detached(binding_epoch),
2471 );
2472 let Ok(active) = ActiveIdentityRanks::try_new(
2473 participants,
2474 self.sequence.ledger.high_watermark(),
2475 self.identity_slot_limit,
2476 ) else {
2477 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2478 };
2479 let pending_owner = BindingTerminalOwner {
2480 participant_index: participant_id,
2481 binding_epoch,
2482 };
2483 let (sequence, order) = match rebuild_pending_terminal_frontiers(
2484 &active,
2485 pending_owner,
2486 delivery_seq,
2487 admission_order,
2488 self.sequence.ledger,
2489 order_ledger,
2490 ) {
2491 Ok(frontiers) => frontiers,
2492 Err(error) => return Err(Box::new((self, error))),
2493 };
2494 self.active_identities = active;
2495 self.sequence = sequence;
2496 self.order = order;
2497 Ok(self)
2498 }
2499
2500 pub(in crate::lifecycle) fn apply_live_fenced_attach(
2508 self,
2509 participant: FrontierParticipant,
2510 prior_binding_epoch: BindingEpoch,
2511 appended_records: &[RetainedCausalRecord],
2512 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2513 let Some(current) = self
2514 .active_identities
2515 .participants()
2516 .iter()
2517 .find(|current| current.participant_index() == participant.participant_index())
2518 .copied()
2519 else {
2520 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2521 };
2522 let Some(sequence_recovery) = self.sequence.recovery else {
2523 return Err(Box::new((self, LiveFrontierTransitionError::Precedence)));
2524 };
2525 let Some(order_recovery) = self.order.recovery else {
2526 return Err(Box::new((self, LiveFrontierTransitionError::Precedence)));
2527 };
2528 if !Self::fenced_recovery_authority_matches(
2529 current,
2530 participant,
2531 prior_binding_epoch,
2532 sequence_recovery,
2533 order_recovery,
2534 ) {
2535 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2536 }
2537 let finalizes_pending = appended_records.len() == 2;
2538 let pending_terminal_matches = self.pending_fenced_terminal_matches(appended_records);
2539 let candidates_empty = self.sequence.immutable_candidates.is_empty()
2540 && self.order.immutable_candidates.is_empty();
2541 if !(candidates_empty || finalizes_pending && pending_terminal_matches) {
2542 return Err(Box::new((self, LiveFrontierTransitionError::Precedence)));
2543 }
2544 if !Self::fenced_records_match(
2545 participant,
2546 sequence_recovery,
2547 order_recovery,
2548 appended_records,
2549 pending_terminal_matches,
2550 ) {
2551 return Err(Box::new((
2552 self,
2553 LiveFrontierTransitionError::RecordPosition,
2554 )));
2555 }
2556 if !self.has_recovery_marker(participant) {
2557 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2558 }
2559 if self.has_duplicate_appended_record(appended_records) {
2560 return Err(Box::new((
2561 self,
2562 LiveFrontierTransitionError::RecordPosition,
2563 )));
2564 }
2565 let Some((sequence_ledger, order_ledger)) =
2566 apply_fenced_ledgers(self.sequence.ledger, self.order.ledger, finalizes_pending)
2567 else {
2568 return Err(Box::new((
2569 self,
2570 LiveFrontierTransitionError::ResultingFrontier,
2571 )));
2572 };
2573 let mut active = self.active_identities.participants().to_vec();
2574 let Some(current) = active
2575 .iter_mut()
2576 .find(|current| current.participant_index() == participant.participant_index())
2577 else {
2578 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2579 };
2580 *current = participant;
2581 let Ok(active) = ActiveIdentityRanks::try_new(
2582 active,
2583 sequence_ledger.high_watermark(),
2584 self.identity_slot_limit,
2585 ) else {
2586 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2587 };
2588 let Ok((sequence, order)) =
2589 rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger)
2590 else {
2591 return Err(Box::new((
2592 self,
2593 LiveFrontierTransitionError::ResultingFrontier,
2594 )));
2595 };
2596 let mut resulting = self;
2597 resulting.active_identities = active;
2598 resulting
2599 .retained_records
2600 .extend_from_slice(appended_records);
2601 resulting
2602 .retained_records
2603 .sort_unstable_by_key(|record| record.delivery_seq);
2604 resulting
2605 .marker_records
2606 .retain(|record| record.delivery_seq != participant.cursor());
2607 resulting.sequence = sequence;
2608 resulting.order = order;
2609 Ok(resulting)
2610 }
2611
2612 fn fenced_recovery_authority_matches(
2613 current: FrontierParticipant,
2614 participant: FrontierParticipant,
2615 prior_binding_epoch: BindingEpoch,
2616 sequence_recovery: RecoverySequenceBlock,
2617 order_recovery: RecoveryOrderBlock,
2618 ) -> bool {
2619 let participant_matches = sequence_recovery.participant_index()
2620 == participant.participant_index()
2621 && order_recovery.participant_index() == participant.participant_index();
2622 let marker_matches = sequence_recovery.marker_delivery_seq() == participant.cursor()
2623 && order_recovery.marker_delivery_seq() == participant.cursor();
2624 let prior_epoch_matches = sequence_recovery.recovered_binding_epoch()
2625 == prior_binding_epoch
2626 && order_recovery.recovered_binding_epoch() == prior_binding_epoch;
2627 participant_matches
2628 && marker_matches
2629 && prior_epoch_matches
2630 && current.binding() == FrontierBinding::Detached(prior_binding_epoch)
2631 && current.cursor() <= participant.cursor()
2632 && matches!(participant.binding(), FrontierBinding::Bound(_))
2633 }
2634
2635 fn fenced_records_match(
2636 participant: FrontierParticipant,
2637 sequence_recovery: RecoverySequenceBlock,
2638 order_recovery: RecoveryOrderBlock,
2639 appended_records: &[RetainedCausalRecord],
2640 pending_terminal_matches: bool,
2641 ) -> bool {
2642 let Some(attached) = appended_records.last().copied() else {
2643 return false;
2644 };
2645 let FrontierBinding::Bound(recovered_binding_epoch) = participant.binding() else {
2646 return false;
2647 };
2648 let attached_matches = attached.delivery_seq == sequence_recovery.recovery_attach_seq()
2649 && attached.admission_order.transaction_order()
2650 == order_recovery.recovery_operation_order()
2651 && attached.kind
2652 == (RetainedCausalRecordKind::AttachLifecycle {
2653 participant_index: participant.participant_index(),
2654 binding_epoch: recovered_binding_epoch,
2655 });
2656 if !attached_matches {
2657 return false;
2658 }
2659 let prefix = &appended_records[..appended_records.len() - 1];
2660 match (
2661 sequence_recovery.terminal(),
2662 order_recovery.active_binding(),
2663 prefix,
2664 ) {
2665 (None, None, []) => true,
2666 (None, None, [_]) => pending_terminal_matches,
2667 (Some(sequence_terminal), Some(order_terminal), [terminal]) => {
2668 sequence_terminal.owner == order_terminal.owner
2669 && terminal.delivery_seq == sequence_terminal.delivery_seq
2670 && terminal.admission_order.transaction_order()
2671 == order_terminal.transaction_order
2672 && terminal.kind
2673 == RetainedCausalRecordKind::BindingTerminal(sequence_terminal.owner)
2674 }
2675 _ => false,
2676 }
2677 }
2678
2679 fn pending_fenced_terminal_matches(&self, appended_records: &[RetainedCausalRecord]) -> bool {
2680 let [terminal, _] = appended_records else {
2681 return false;
2682 };
2683 let RetainedCausalRecordKind::BindingTerminal(owner) = terminal.kind else {
2684 return false;
2685 };
2686 let sequence_matches = self.sequence.immutable_candidates.iter().any(|candidate| {
2687 matches!(
2688 candidate,
2689 ImmutableSequenceCandidate::BindingTerminal {
2690 delivery_seq,
2691 admission_order,
2692 owner: candidate_owner,
2693 } if *delivery_seq == terminal.delivery_seq
2694 && *admission_order == terminal.admission_order
2695 && *candidate_owner == owner
2696 )
2697 });
2698 let order_matches = self.order.immutable_candidates.iter().any(|candidate| {
2699 candidate.transaction_order == terminal.admission_order.transaction_order()
2700 && candidate.candidate_keys.as_slice() == [terminal.admission_order]
2701 });
2702 sequence_matches && order_matches
2703 }
2704
2705 fn has_recovery_marker(&self, participant: FrontierParticipant) -> bool {
2706 self.retained_records.iter().any(|record| {
2707 record.delivery_seq == participant.cursor()
2708 && matches!(
2709 record.kind,
2710 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
2711 if participant_index == participant.participant_index()
2712 )
2713 })
2714 }
2715
2716 fn has_duplicate_appended_record(&self, appended_records: &[RetainedCausalRecord]) -> bool {
2717 appended_records.iter().any(|row| {
2718 self.retained_records
2719 .iter()
2720 .any(|retained| retained.delivery_seq == row.delivery_seq)
2721 })
2722 }
2723
2724 pub(in crate::lifecycle) fn apply_live_identity(
2727 mut self,
2728 participant: FrontierParticipant,
2729 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2730 let Some(current) = self
2731 .active_identities
2732 .participants
2733 .iter_mut()
2734 .find(|current| current.participant_index == participant.participant_index)
2735 else {
2736 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2737 };
2738 if participant.cursor < current.cursor
2739 || participant.cursor > self.sequence.ledger.high_watermark()
2740 {
2741 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2742 }
2743 *current = participant;
2744 Ok(self)
2745 }
2746
2747 pub fn project_ordinary_record(
2763 self,
2764 input: OrdinaryRecordProjectionInput,
2765 ) -> Result<OrdinaryRecordProjectionDecision, Box<OrdinaryRecordProjectionFailure>> {
2766 let (
2767 request,
2768 receiving_binding_epoch,
2769 encoded_record_charge,
2770 retained_charges,
2771 observer_progress,
2772 closure_accounting,
2773 limits,
2774 ) = input.as_parts();
2775 if request.conversation_id != self.conversation_id {
2776 return Err(projection_failure(
2777 self,
2778 input,
2779 OrdinaryProjectionError::Conversation,
2780 ));
2781 }
2782 let unaccepted_marker_anchors = ordinary_unaccepted_marker_anchors(&self);
2783 let kernel = match project_ordinary_fixed_point(&OrdinaryProjectionFacts {
2784 request: request.clone(),
2785 receiving_binding_epoch,
2786 encoded_record_charge,
2787 retained_records: &self.retained_records,
2788 retained_charges,
2789 active_marker_credit_records: &self.marker_records,
2790 unaccepted_marker_anchors: &unaccepted_marker_anchors,
2791 active_identities: self.active_identities.participants(),
2792 identity_slot_limit: self.identity_slot_limit,
2793 current_floor: self.retained_floor,
2794 observer_progress,
2795 order_ledger: self.order.ledger,
2796 sequence_ledger: self.sequence.ledger,
2797 immutable_candidates: &self.sequence.immutable_candidates,
2798 closure_accounting,
2799 remaining_recovery_claim: closure_accounting.edge_k_remaining(),
2800 limits,
2801 }) {
2802 Ok(value) => value,
2803 Err(error) => return Err(projection_failure(self, input, error)),
2804 };
2805 match kernel {
2806 OrdinaryProjectionKernelDecision::DrainFirst(prefix) => Ok(
2807 OrdinaryRecordProjectionDecision::DrainFirst(Box::new(OrdinaryRecordDrainFirst {
2808 frontiers: self,
2809 input,
2810 candidate: prefix.candidate(),
2811 })),
2812 ),
2813 OrdinaryProjectionKernelDecision::Projected(projected) => {
2814 let observer_floor = match check_observer_floor(
2815 ObserverCheckedOperation::RecordAdmission(request.clone()),
2816 observer_progress,
2817 projected.floor().resulting_floor,
2818 ) {
2819 ObserverFloorDecision::Eligible(permit) => permit,
2820 ObserverFloorDecision::Respond(_) => {
2821 return Err(projection_failure(
2822 self,
2823 input,
2824 OrdinaryProjectionError::ObserverSelectorInvariant,
2825 ));
2826 }
2827 };
2828 let closure = match check_remaining_closure(
2829 &ClosureCheckedEnvelope::RecordAdmission(request.clone()),
2830 closure_accounting,
2831 false,
2832 0,
2833 projected.required_capacity(),
2834 ) {
2835 RemainingClosureDecision::Eligible(permit) => *permit,
2836 RemainingClosureDecision::Respond(_) => {
2837 return Err(projection_failure(
2838 self,
2839 input,
2840 OrdinaryProjectionError::ClosureSelectorInvariant,
2841 ));
2842 }
2843 };
2844 match self.apply_ordinary_projection(*projected, observer_floor, closure) {
2845 Ok(projected) => Ok(OrdinaryRecordProjectionDecision::Projected(Box::new(
2846 projected,
2847 ))),
2848 Err(failure) => {
2849 let (frontiers, error) = *failure;
2850 Err(projection_failure(frontiers, input, error))
2851 }
2852 }
2853 }
2854 }
2855 }
2856
2857 fn apply_ordinary_projection(
2858 mut self,
2859 projected: OrdinaryFixedPointPlan,
2860 observer_floor: super::ObserverFloorPermit,
2861 closure: super::RemainingClosurePermit,
2862 ) -> Result<ProjectedOrdinaryRecord, Box<(Self, OrdinaryProjectionError)>> {
2863 let Ok(marker_count) = u64::try_from(projected.marker_candidates().len()) else {
2864 return Err(Box::new((
2865 self,
2866 OrdinaryProjectionError::SequenceRelocation,
2867 )));
2868 };
2869 let Some(sequence_delta) = marker_count.checked_add(1) else {
2870 return Err(Box::new((
2871 self,
2872 OrdinaryProjectionError::SequenceRelocation,
2873 )));
2874 };
2875 if let Err(error) = preflight_ordinary_sequence_owners(&self.sequence, sequence_delta) {
2876 return Err(Box::new((self, error)));
2877 }
2878 if let Err(error) = preflight_ordinary_order_owners(&self.order) {
2879 return Err(Box::new((self, error)));
2880 }
2881 let (
2882 floor,
2883 retained_charge,
2884 baseline,
2885 accounting,
2886 required_capacity,
2887 order,
2888 sequence,
2889 caller_record,
2890 caller_charge,
2891 retained_records,
2892 retained_charges,
2893 new_marker_candidates,
2894 ) = projected.into_parts();
2895
2896 let prior_sequence_ledger = self.sequence.ledger;
2897 let prior_order_ledger = self.order.ledger;
2898 relay_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
2899 relay_ordinary_order_owners(&mut self.order);
2900
2901 self.sequence.ledger = sequence.resulting();
2902 self.sequence.immutable_candidates.extend(
2903 new_marker_candidates
2904 .iter()
2905 .copied()
2906 .map(ImmutableSequenceCandidate::Marker),
2907 );
2908 self.order.ledger = order.resulting();
2909 if !new_marker_candidates.is_empty() {
2910 self.order
2911 .immutable_candidates
2912 .push(ImmutableOrderCandidateMajor {
2913 transaction_order: order.major(),
2914 candidate_keys: new_marker_candidates
2915 .iter()
2916 .map(|candidate| candidate.admission_order)
2917 .collect(),
2918 });
2919 }
2920 if validate_cross_counter(&self.sequence, &self.order).is_err() {
2921 self.sequence.ledger = prior_sequence_ledger;
2922 self.sequence.immutable_candidates.clear();
2923 self.order.ledger = prior_order_ledger;
2924 self.order.immutable_candidates.clear();
2925 rollback_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
2926 rollback_ordinary_order_owners(&mut self.order);
2927 return Err(Box::new((
2928 self,
2929 OrdinaryProjectionError::SequenceRelocation,
2930 )));
2931 }
2932 self.retained_floor = floor.resulting_floor;
2933 self.retained_records = retained_records;
2934 self.marker_records
2935 .retain(|record| u128::from(record.delivery_seq) >= floor.resulting_floor);
2936
2937 Ok(ProjectedOrdinaryRecord {
2938 frontiers: self,
2939 floor,
2940 retained_charge,
2941 baseline,
2942 accounting,
2943 required_capacity,
2944 order,
2945 sequence,
2946 observer_floor,
2947 closure,
2948 caller_record,
2949 caller_charge,
2950 retained_charges,
2951 new_marker_candidates,
2952 })
2953 }
2954
2955 pub fn planned_settled_leave_admission_order<F>(
2967 &self,
2968 member: &LiveMember<F>,
2969 binding_state: BindingState,
2970 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
2971 let (participant_id, ended_binding_epoch) =
2972 validate_settled_leave_prestate(self, member, binding_state)?;
2973 let selection = select_leave_order(&self.order, participant_id, ended_binding_epoch, None)?;
2974 Ok(super::AdmissionOrder::new(
2975 selection.selected_major,
2976 CandidatePhase::MembershipExit,
2977 participant_id,
2978 ))
2979 }
2980
2981 pub fn prepare_settled_leave_authority<F>(
2994 mut self,
2995 member: &LiveMember<F>,
2996 binding_state: BindingState,
2997 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
2998 let (participant_id, ended_binding_epoch) =
2999 validate_settled_leave_prestate(&self, member, binding_state)?;
3000 let left_transaction_order =
3001 consume_leave_order_lane(&mut self.order, participant_id, ended_binding_epoch, None)?;
3002 Ok(PreparedLeaveAuthority::settled(
3003 self,
3004 member.conversation_id(),
3005 participant_id,
3006 ended_binding_epoch,
3007 left_transaction_order,
3008 ))
3009 }
3010
3011 pub fn planned_pending_leave_admission_order<F>(
3019 &self,
3020 member: &LiveMember<F>,
3021 pending: PendingFinalization,
3022 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
3023 let (participant_id, expected_order) =
3024 validate_pending_leave_prestate(self, member, pending)?;
3025 let selection =
3026 select_leave_order(&self.order, participant_id, None, Some(expected_order))?;
3027 Ok(super::AdmissionOrder::new(
3028 selection.selected_major,
3029 CandidatePhase::MembershipExit,
3030 participant_id,
3031 ))
3032 }
3033
3034 pub fn prepare_pending_leave_authority<F>(
3048 mut self,
3049 member: &LiveMember<F>,
3050 pending: PendingFinalization,
3051 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
3052 let (participant_id, expected_order) =
3053 validate_pending_leave_prestate(&self, member, pending)?;
3054 let left_transaction_order =
3055 consume_leave_order_lane(&mut self.order, participant_id, None, Some(expected_order))?;
3056 Ok(PreparedLeaveAuthority::pending(
3057 self,
3058 member.conversation_id(),
3059 participant_id,
3060 pending.binding_epoch(),
3061 expected_order,
3062 left_transaction_order,
3063 ))
3064 }
3065
3066 #[allow(
3075 clippy::too_many_lines,
3076 reason = "the atomic Leave relay keeps membership, both ledgers, products, recovery, and retained rows visibly in one checked transition"
3077 )]
3078 pub(super) fn finish_leave_claims(
3079 mut self,
3080 participant_id: ParticipantId,
3081 ended_binding_epoch: Option<BindingEpoch>,
3082 committed_terminal: Option<CommittedBindingTerminal>,
3083 left_delivery_seq: DeliverySeq,
3084 left_transaction_order: TransactionOrder,
3085 ) -> Result<Self, LeaveCommitError> {
3086 let prior_high = self.sequence.ledger.high_watermark();
3087 let first_appended = prior_high
3088 .checked_add(1)
3089 .ok_or(LeaveCommitError::SequenceAuthority)?;
3090 let expected_left = if committed_terminal.is_some() {
3091 first_appended
3092 .checked_add(1)
3093 .ok_or(LeaveCommitError::SequenceAuthority)?
3094 } else {
3095 first_appended
3096 };
3097 if left_delivery_seq != expected_left {
3098 return Err(LeaveCommitError::SequenceAuthority);
3099 }
3100
3101 match (
3102 committed_terminal,
3103 self.sequence.immutable_candidates.as_slice(),
3104 ) {
3105 (
3106 Some(terminal),
3107 [
3108 ImmutableSequenceCandidate::BindingTerminal {
3109 delivery_seq,
3110 admission_order,
3111 owner,
3112 },
3113 ],
3114 ) if *delivery_seq == first_appended
3115 && terminal.delivery_seq() == first_appended
3116 && *admission_order == terminal.admission_order()
3117 && owner.participant_index == participant_id
3118 && owner.binding_epoch == terminal.binding_epoch() => {}
3119 (None, []) => {}
3120 (Some(_) | None, _) => return Err(LeaveCommitError::SequenceAuthority),
3121 }
3122
3123 let Some(active_index) = self
3124 .active_identities
3125 .participants
3126 .iter()
3127 .position(|participant| participant.participant_index == participant_id)
3128 else {
3129 return Err(LeaveCommitError::ResultingFrontier);
3130 };
3131 self.active_identities.participants.remove(active_index);
3132 self.marker_records.retain(|record| {
3133 !matches!(
3134 record.kind,
3135 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
3136 if participant_index == participant_id
3137 )
3138 });
3139 let resulting_live = usize_to_u64(self.active_identities.participants.len());
3140
3141 let mut exit_consumed = false;
3142 let mut terminal_consumed = ended_binding_epoch.is_none();
3143 let mut units = Vec::new();
3144 for claim in self.sequence.movable_claims.iter().copied() {
3145 match claim.owner {
3146 SequenceDirectOwner::MembershipExit {
3147 participant_index: owner,
3148 } if owner == participant_id => {
3149 if exit_consumed {
3150 return Err(LeaveCommitError::ResultingFrontier);
3151 }
3152 exit_consumed = true;
3153 }
3154 SequenceDirectOwner::BindingTerminal(owner)
3155 if owner.participant_index == participant_id
3156 && Some(owner.binding_epoch) == ended_binding_epoch =>
3157 {
3158 if terminal_consumed {
3159 return Err(LeaveCommitError::ResultingFrontier);
3160 }
3161 terminal_consumed = true;
3162 }
3163 SequenceDirectOwner::MembershipExit { .. }
3164 | SequenceDirectOwner::BindingTerminal(_) => {
3165 units.push(LeaveSequenceUnit::Direct(claim));
3166 }
3167 }
3168 }
3169 if !exit_consumed || !terminal_consumed {
3170 return Err(LeaveCommitError::ResultingFrontier);
3171 }
3172
3173 let recovery_owned = self
3174 .sequence
3175 .recovery
3176 .is_some_and(|block| block.participant_index == participant_id);
3177 for range in self.sequence.products.live_times_terminal.iter().copied() {
3178 if range.terminal.participant_index != participant_id && resulting_live != 0 {
3179 units.push(LeaveSequenceUnit::TerminalProduct(TerminalProductRange {
3180 start: range.start,
3181 length: resulting_live,
3182 terminal: range.terminal,
3183 }));
3184 }
3185 }
3186 if let Some(range) = self.sequence.products.live_times_replacement_terminal
3187 && !recovery_owned
3188 && resulting_live != 0
3189 {
3190 units.push(LeaveSequenceUnit::ReplacementProduct(
3191 ReplacementTerminalProductRange {
3192 start: range.start,
3193 length: resulting_live,
3194 participant_index: range.participant_index,
3195 marker_delivery_seq: range.marker_delivery_seq,
3196 prior_binding_epoch: range.prior_binding_epoch,
3197 },
3198 ));
3199 }
3200 let resulting_other = resulting_live.saturating_sub(1);
3201 if resulting_other != 0 {
3202 for range in self.sequence.products.other_live_times_exit.iter().copied() {
3203 if range.exit_participant != participant_id {
3204 units.push(LeaveSequenceUnit::ExitProduct(ExitProductRange {
3205 start: range.start,
3206 length: resulting_other,
3207 exit_participant: range.exit_participant,
3208 }));
3209 }
3210 }
3211 }
3212 if let Some(recovery) = self.sequence.recovery
3213 && !recovery_owned
3214 {
3215 units.push(LeaveSequenceUnit::Recovery(recovery));
3216 }
3217 units.sort_by_key(|unit| unit.original_start());
3218
3219 let mut cursor = left_delivery_seq.checked_add(1);
3220 let mut movable_claims = Vec::new();
3221 let mut terminal_products = Vec::new();
3222 let mut replacement_product = None;
3223 let mut exit_products = Vec::new();
3224 let mut recovery = None;
3225 for unit in units {
3226 match unit {
3227 LeaveSequenceUnit::Direct(mut claim) => {
3228 claim.delivery_seq = allocate_leave_sequence_range(&mut cursor, 1)?;
3229 movable_claims.push(claim);
3230 }
3231 LeaveSequenceUnit::TerminalProduct(mut range) => {
3232 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3233 terminal_products.push(range);
3234 }
3235 LeaveSequenceUnit::ReplacementProduct(mut range) => {
3236 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3237 replacement_product = Some(range);
3238 }
3239 LeaveSequenceUnit::ExitProduct(mut range) => {
3240 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3241 exit_products.push(range);
3242 }
3243 LeaveSequenceUnit::Recovery(mut block) => {
3244 let length = 2 + u64::from(block.terminal.is_some());
3245 let start = allocate_leave_sequence_range(&mut cursor, length)?;
3246 if let Some(mut terminal) = block.terminal {
3247 terminal.delivery_seq = start;
3248 block.terminal = Some(terminal);
3249 block.recovery_attach_seq = start
3250 .checked_add(1)
3251 .ok_or(LeaveCommitError::ResultingFrontier)?;
3252 } else {
3253 block.recovery_attach_seq = start;
3254 }
3255 block.replacement_terminal_seq = block
3256 .recovery_attach_seq
3257 .checked_add(1)
3258 .ok_or(LeaveCommitError::ResultingFrontier)?;
3259 recovery = Some(block);
3260 }
3261 }
3262 }
3263 movable_claims.sort_by_key(|claim| claim.delivery_seq);
3264 terminal_products.sort_by_key(|range| range.start);
3265 exit_products.sort_by_key(|range| range.start);
3266 let terminal_count = usize_to_u64(
3267 movable_claims
3268 .iter()
3269 .filter(|claim| matches!(claim.owner, SequenceDirectOwner::BindingTerminal(_)))
3270 .count(),
3271 ) + u64::from(recovery.is_some_and(|block| block.terminal.is_some()));
3272 let recovery_reserve = if recovery.is_some() {
3273 RecoverySequenceReserve::DetachedCredentialRecovery
3274 } else {
3275 RecoverySequenceReserve::None
3276 };
3277 let ledger = SequenceLedger::try_new(
3278 left_delivery_seq,
3279 SequenceClaims::new(resulting_live, terminal_count, 0, recovery_reserve),
3280 )
3281 .map_err(|_| LeaveCommitError::ResultingFrontier)?;
3282 self.sequence = SequenceClaimFrontier {
3283 ledger,
3284 movable_claims,
3285 immutable_candidates: Vec::new(),
3286 products: SequenceProductRanges {
3287 live_times_terminal: terminal_products,
3288 live_times_replacement_terminal: replacement_product,
3289 other_live_times_exit: exit_products,
3290 },
3291 recovery,
3292 };
3293
3294 if let Some(terminal) = committed_terminal {
3295 self.retained_records.push(RetainedCausalRecord {
3296 delivery_seq: terminal.delivery_seq(),
3297 admission_order: terminal.admission_order(),
3298 kind: RetainedCausalRecordKind::BindingTerminal(BindingTerminalOwner {
3299 participant_index: participant_id,
3300 binding_epoch: terminal.binding_epoch(),
3301 }),
3302 });
3303 }
3304 self.retained_records.push(RetainedCausalRecord {
3305 delivery_seq: left_delivery_seq,
3306 admission_order: super::AdmissionOrder::new(
3307 left_transaction_order,
3308 CandidatePhase::MembershipExit,
3309 participant_id,
3310 ),
3311 kind: RetainedCausalRecordKind::MembershipExit {
3312 participant_index: participant_id,
3313 },
3314 });
3315 self.retained_records
3316 .sort_by_key(|record| record.delivery_seq);
3317
3318 let order_claims = self.order.ledger.claims();
3319 if order_claims.membership_exits() != resulting_live
3320 || self.order.recovery.is_some() != self.sequence.recovery.is_some()
3321 || validate_cross_counter(&self.sequence, &self.order).is_err()
3322 {
3323 return Err(LeaveCommitError::ResultingFrontier);
3324 }
3325 Ok(self)
3326 }
3327
3328 fn marker_candidate(&self, delivery_seq: DeliverySeq) -> Option<ValidatedMarkerCandidate> {
3329 self.sequence
3330 .immutable_candidates
3331 .iter()
3332 .find_map(|candidate| match candidate {
3333 ImmutableSequenceCandidate::Marker(candidate)
3334 if candidate.delivery_seq == delivery_seq =>
3335 {
3336 Some(ValidatedMarkerCandidate {
3337 conversation_id: self.conversation_id,
3338 candidate: *candidate,
3339 seal: MarkerAuthoritySeal::Validated,
3340 })
3341 }
3342 _ => None,
3343 })
3344 }
3345
3346 pub(super) fn drain_next_marker_core(
3355 mut self,
3356 ) -> Result<MarkerDrainCore, MarkerDrainCoreError> {
3357 let Some(first) = self.sequence.immutable_candidates.first().copied() else {
3358 return Err(MarkerDrainCoreError::NoCandidate);
3359 };
3360 let ImmutableSequenceCandidate::Marker(marker) = first else {
3361 return Err(MarkerDrainCoreError::BindingTerminalFirst);
3362 };
3363 let expected_sequence = self
3364 .sequence
3365 .ledger
3366 .high_watermark()
3367 .checked_add(1)
3368 .ok_or(MarkerDrainCoreError::SequenceNotNext)?;
3369 if marker.delivery_seq != expected_sequence {
3370 return Err(MarkerDrainCoreError::SequenceNotNext);
3371 }
3372 if order_is_above_high(
3373 marker.admission_order.transaction_order(),
3374 self.order.ledger.high(),
3375 ) {
3376 return Err(MarkerDrainCoreError::CausalMajorNotAllocated);
3377 }
3378 let candidate = self
3379 .marker_candidate(marker.delivery_seq)
3380 .ok_or(MarkerDrainCoreError::NoCandidate)?;
3381
3382 let key = marker.admission_order;
3383 let Some(group_index) = self
3384 .order
3385 .immutable_candidates
3386 .iter()
3387 .position(|group| group.candidate_keys.contains(&key))
3388 else {
3389 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3390 };
3391 let group = &mut self.order.immutable_candidates[group_index];
3392 let Ok(key_index) = group.candidate_keys.binary_search(&key) else {
3393 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3394 };
3395 group.candidate_keys.remove(key_index);
3396 if group.candidate_keys.is_empty() {
3397 self.order.immutable_candidates.remove(group_index);
3398 }
3399 self.sequence.immutable_candidates.remove(0);
3400
3401 let claims = self.sequence.ledger.claims();
3402 let markers = claims
3403 .markers()
3404 .checked_sub(1)
3405 .ok_or(MarkerDrainCoreError::ResultingLedger)?;
3406 self.sequence.ledger = SequenceLedger::try_new(
3407 expected_sequence,
3408 super::SequenceClaims::new(
3409 claims.live_members(),
3410 claims.binding_terminals(),
3411 markers,
3412 claims.recovery(),
3413 ),
3414 )
3415 .map_err(|_| MarkerDrainCoreError::ResultingLedger)?;
3416
3417 let record = RetainedCausalRecord {
3418 delivery_seq: marker.delivery_seq,
3419 admission_order: marker.admission_order,
3420 kind: RetainedCausalRecordKind::CompactionMarker {
3421 participant_index: marker.admission_order.participant_index(),
3422 provenance: marker.provenance,
3423 },
3424 };
3425 self.marker_records.push(record);
3426 self.retained_records.push(record);
3427 Ok(MarkerDrainCore {
3428 candidate,
3429 record: ValidatedMarkerRecord {
3430 conversation_id: self.conversation_id,
3431 record,
3432 provenance: marker.provenance,
3433 target_binding: marker.target_binding,
3434 occurrence: MarkerRecordOccurrence::Undelivered,
3435 seal: MarkerAuthoritySeal::Validated,
3436 },
3437 frontiers: self,
3438 })
3439 }
3440}
3441
3442fn apply_fenced_ledgers(
3443 sequence: SequenceLedger,
3444 order: OrderLedger,
3445 finalizes_pending: bool,
3446) -> Option<(SequenceLedger, OrderLedger)> {
3447 let sequence = if finalizes_pending {
3448 sequence.apply_fenced_recovery_finalizing_pending()
3449 } else {
3450 sequence.apply_fenced_recovery()
3451 }
3452 .ok()?;
3453 let order = if finalizes_pending {
3454 order.apply_fenced_recovery_finalizing_pending()
3455 } else {
3456 order.apply_fenced_recovery()
3457 }
3458 .ok()?;
3459 Some((sequence, order))
3460}
3461
3462fn rebuild_unreserved_frontiers(
3463 active: &ActiveIdentityRanks,
3464 sequence_ledger: SequenceLedger,
3465 order_ledger: OrderLedger,
3466) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
3467 let terminal_owners: Vec<_> = active
3468 .participants()
3469 .iter()
3470 .filter_map(|participant| match participant.binding() {
3471 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
3472 participant_index: participant.participant_index(),
3473 binding_epoch,
3474 }),
3475 FrontierBinding::Detached(_) => None,
3476 })
3477 .collect();
3478 let live_count = active.len();
3479 let terminal_count =
3480 u64::try_from(terminal_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3481 let sequence_claims = sequence_ledger.claims();
3482 let order_claims = order_ledger.claims();
3483 if sequence_claims.live_members() != live_count
3484 || sequence_claims.binding_terminals() != terminal_count
3485 || sequence_claims.markers() != 0
3486 || sequence_claims.recovery() != RecoverySequenceReserve::None
3487 || order_claims.active_binding_terminals() != terminal_count
3488 || order_claims.membership_exits() != live_count
3489 || order_claims.recovery_operation()
3490 || order_claims.recovery_replacement_terminal()
3491 {
3492 return Err(LiveFrontierTransitionError::ResultingFrontier);
3493 }
3494
3495 let sequence = rebuild_unreserved_sequence(active, &terminal_owners, sequence_ledger)?;
3496 let order = rebuild_unreserved_order(active, &terminal_owners, order_ledger)?;
3497 validate_cross_counter(&sequence, &order)
3498 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
3499 Ok((sequence, order))
3500}
3501
3502fn rebuild_unreserved_sequence(
3503 active: &ActiveIdentityRanks,
3504 terminal_owners: &[BindingTerminalOwner],
3505 sequence_ledger: SequenceLedger,
3506) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
3507 let live_count = active.len();
3508 let mut sequence_cursor = sequence_ledger
3509 .high_watermark()
3510 .checked_add(1)
3511 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3512 let mut movable_sequence = Vec::new();
3513 for terminal in terminal_owners {
3514 movable_sequence.push(MovableSequenceClaim {
3515 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3516 owner: SequenceDirectOwner::BindingTerminal(*terminal),
3517 });
3518 }
3519 for participant in active.participants() {
3520 movable_sequence.push(MovableSequenceClaim {
3521 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3522 owner: SequenceDirectOwner::MembershipExit {
3523 participant_index: participant.participant_index(),
3524 },
3525 });
3526 }
3527 let mut terminal_products = Vec::new();
3528 for terminal in terminal_owners {
3529 terminal_products.push(TerminalProductRange {
3530 start: take_live_sequence(&mut sequence_cursor, live_count)?,
3531 length: live_count,
3532 terminal: *terminal,
3533 });
3534 }
3535 let exit_product_length = live_count.saturating_sub(1);
3536 let mut exit_products = Vec::new();
3537 for participant in active.participants() {
3538 exit_products.push(ExitProductRange {
3539 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
3540 length: exit_product_length,
3541 exit_participant: participant.participant_index(),
3542 });
3543 }
3544 let sequence_end = u128::from(sequence_ledger.high_watermark())
3545 .checked_add(sequence_ledger.required_reserve())
3546 .and_then(|value| value.checked_add(1))
3547 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3548 if u128::from(sequence_cursor) != sequence_end {
3549 return Err(LiveFrontierTransitionError::ResultingFrontier);
3550 }
3551 Ok(SequenceClaimFrontier {
3552 ledger: sequence_ledger,
3553 movable_claims: movable_sequence,
3554 immutable_candidates: Vec::new(),
3555 products: SequenceProductRanges {
3556 live_times_terminal: terminal_products,
3557 live_times_replacement_terminal: None,
3558 other_live_times_exit: exit_products,
3559 },
3560 recovery: None,
3561 })
3562}
3563
3564fn rebuild_unreserved_order(
3565 active: &ActiveIdentityRanks,
3566 terminal_owners: &[BindingTerminalOwner],
3567 order_ledger: OrderLedger,
3568) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
3569 let order_claims = order_ledger.claims();
3570 let order_start = order_frontier_start(order_ledger.high());
3571 let mut order_cursor =
3572 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3573 let mut movable_order = Vec::new();
3574 for terminal in terminal_owners.iter().copied() {
3575 movable_order.push(MovableOrderClaim {
3576 transaction_order: take_live_order(&mut order_cursor)?,
3577 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
3578 });
3579 }
3580 for participant in active.participants() {
3581 movable_order.push(MovableOrderClaim {
3582 transaction_order: take_live_order(&mut order_cursor)?,
3583 owner: OrderDirectOwner::MembershipExit {
3584 participant_index: participant.participant_index(),
3585 },
3586 });
3587 }
3588 if u128::from(order_cursor) != order_start + order_claims.total() {
3589 return Err(LiveFrontierTransitionError::ResultingFrontier);
3590 }
3591 Ok(OrderClaimFrontier {
3592 ledger: order_ledger,
3593 movable_claims: movable_order,
3594 immutable_candidates: Vec::new(),
3595 recovery: None,
3596 })
3597}
3598
3599fn rebuild_pending_terminal_frontiers(
3600 active: &ActiveIdentityRanks,
3601 pending_owner: BindingTerminalOwner,
3602 delivery_seq: DeliverySeq,
3603 admission_order: super::AdmissionOrder,
3604 sequence_ledger: SequenceLedger,
3605 order_ledger: OrderLedger,
3606) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
3607 if admission_order.candidate_phase() != CandidatePhase::BindingTerminal
3608 || admission_order.participant_index() != pending_owner.participant_index
3609 || admission_order.transaction_order()
3610 != match order_ledger.high() {
3611 OrderHigh::Allocated(high) => high,
3612 OrderHigh::Empty => return Err(LiveFrontierTransitionError::RecordPosition),
3613 }
3614 || sequence_ledger.high_watermark().checked_add(1) != Some(delivery_seq)
3615 {
3616 return Err(LiveFrontierTransitionError::RecordPosition);
3617 }
3618 let bound_owners: Vec<_> = active
3619 .participants()
3620 .iter()
3621 .filter_map(|participant| match participant.binding() {
3622 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
3623 participant_index: participant.participant_index(),
3624 binding_epoch,
3625 }),
3626 FrontierBinding::Detached(_) => None,
3627 })
3628 .collect();
3629 let mut all_terminal_owners = bound_owners.clone();
3630 all_terminal_owners.push(pending_owner);
3631 all_terminal_owners.sort_unstable_by_key(|owner| owner.participant_index);
3632 let live_count = active.len();
3633 let bound_count =
3634 u64::try_from(bound_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3635 let terminal_count = u64::try_from(all_terminal_owners.len())
3636 .map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3637 let sequence_claims = sequence_ledger.claims();
3638 let order_claims = order_ledger.claims();
3639 if sequence_claims.live_members() != live_count
3640 || sequence_claims.binding_terminals() != terminal_count
3641 || sequence_claims.markers() != 0
3642 || sequence_claims.recovery() != RecoverySequenceReserve::None
3643 || order_claims.active_binding_terminals() != bound_count
3644 || order_claims.membership_exits() != live_count
3645 || order_claims.recovery_operation()
3646 || order_claims.recovery_replacement_terminal()
3647 {
3648 return Err(LiveFrontierTransitionError::ResultingFrontier);
3649 }
3650
3651 let sequence = rebuild_pending_terminal_sequence(
3652 active,
3653 pending_owner,
3654 delivery_seq,
3655 admission_order,
3656 sequence_ledger,
3657 &bound_owners,
3658 &all_terminal_owners,
3659 )?;
3660 let order =
3661 rebuild_pending_terminal_order(active, admission_order, order_ledger, bound_owners)?;
3662 validate_cross_counter(&sequence, &order)
3663 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
3664 Ok((sequence, order))
3665}
3666
3667fn rebuild_pending_terminal_sequence(
3668 active: &ActiveIdentityRanks,
3669 pending_owner: BindingTerminalOwner,
3670 delivery_seq: DeliverySeq,
3671 admission_order: super::AdmissionOrder,
3672 sequence_ledger: SequenceLedger,
3673 bound_owners: &[BindingTerminalOwner],
3674 all_terminal_owners: &[BindingTerminalOwner],
3675) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
3676 let live_count = active.len();
3677 let mut sequence_cursor = delivery_seq;
3678 let candidate_sequence = take_live_sequence(&mut sequence_cursor, 1)?;
3679 let mut movable_sequence = Vec::new();
3680 for terminal in bound_owners {
3681 movable_sequence.push(MovableSequenceClaim {
3682 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3683 owner: SequenceDirectOwner::BindingTerminal(*terminal),
3684 });
3685 }
3686 for participant in active.participants() {
3687 movable_sequence.push(MovableSequenceClaim {
3688 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3689 owner: SequenceDirectOwner::MembershipExit {
3690 participant_index: participant.participant_index(),
3691 },
3692 });
3693 }
3694 let mut terminal_products = Vec::new();
3695 for terminal in all_terminal_owners {
3696 terminal_products.push(TerminalProductRange {
3697 start: take_live_sequence(&mut sequence_cursor, live_count)?,
3698 length: live_count,
3699 terminal: *terminal,
3700 });
3701 }
3702 let exit_product_length = live_count.saturating_sub(1);
3703 let mut exit_products = Vec::new();
3704 for participant in active.participants() {
3705 exit_products.push(ExitProductRange {
3706 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
3707 length: exit_product_length,
3708 exit_participant: participant.participant_index(),
3709 });
3710 }
3711 let sequence_end = u128::from(sequence_ledger.high_watermark())
3712 .checked_add(sequence_ledger.required_reserve())
3713 .and_then(|value| value.checked_add(1))
3714 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3715 if u128::from(sequence_cursor) != sequence_end {
3716 return Err(LiveFrontierTransitionError::ResultingFrontier);
3717 }
3718 Ok(SequenceClaimFrontier {
3719 ledger: sequence_ledger,
3720 movable_claims: movable_sequence,
3721 immutable_candidates: alloc::vec![ImmutableSequenceCandidate::BindingTerminal {
3722 delivery_seq: candidate_sequence,
3723 admission_order,
3724 owner: pending_owner,
3725 }],
3726 products: SequenceProductRanges {
3727 live_times_terminal: terminal_products,
3728 live_times_replacement_terminal: None,
3729 other_live_times_exit: exit_products,
3730 },
3731 recovery: None,
3732 })
3733}
3734
3735fn rebuild_pending_terminal_order(
3736 active: &ActiveIdentityRanks,
3737 admission_order: super::AdmissionOrder,
3738 order_ledger: OrderLedger,
3739 bound_owners: Vec<BindingTerminalOwner>,
3740) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
3741 let order_claims = order_ledger.claims();
3742 let order_start = order_frontier_start(order_ledger.high());
3743 let mut order_cursor =
3744 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3745 let mut movable_order = Vec::new();
3746 for terminal in bound_owners {
3747 movable_order.push(MovableOrderClaim {
3748 transaction_order: take_live_order(&mut order_cursor)?,
3749 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
3750 });
3751 }
3752 for participant in active.participants() {
3753 movable_order.push(MovableOrderClaim {
3754 transaction_order: take_live_order(&mut order_cursor)?,
3755 owner: OrderDirectOwner::MembershipExit {
3756 participant_index: participant.participant_index(),
3757 },
3758 });
3759 }
3760 if u128::from(order_cursor) != order_start + order_claims.total() {
3761 return Err(LiveFrontierTransitionError::ResultingFrontier);
3762 }
3763 Ok(OrderClaimFrontier {
3764 ledger: order_ledger,
3765 movable_claims: movable_order,
3766 immutable_candidates: alloc::vec![ImmutableOrderCandidateMajor {
3767 transaction_order: admission_order.transaction_order(),
3768 candidate_keys: alloc::vec![admission_order],
3769 }],
3770 recovery: None,
3771 })
3772}
3773
3774fn take_live_sequence(
3775 cursor: &mut DeliverySeq,
3776 length: u64,
3777) -> Result<DeliverySeq, LiveFrontierTransitionError> {
3778 let start = *cursor;
3779 *cursor = cursor
3780 .checked_add(length)
3781 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3782 Ok(start)
3783}
3784
3785fn take_live_order(
3786 cursor: &mut TransactionOrder,
3787) -> Result<TransactionOrder, LiveFrontierTransitionError> {
3788 let value = *cursor;
3789 *cursor = cursor
3790 .checked_add(1)
3791 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3792 Ok(value)
3793}
3794
3795fn ordinary_unaccepted_marker_anchors(frontiers: &ClaimFrontiers) -> Vec<DeliverySeq> {
3796 frontiers
3797 .marker_records
3798 .iter()
3799 .filter_map(|record| {
3800 let RetainedCausalRecordKind::CompactionMarker {
3801 participant_index, ..
3802 } = record.kind
3803 else {
3804 return None;
3805 };
3806 active_participant(&frontiers.active_identities, participant_index)
3807 .is_some_and(|participant| participant.cursor < record.delivery_seq)
3808 .then_some(record.delivery_seq)
3809 })
3810 .collect()
3811}
3812
3813fn projection_failure(
3814 frontiers: ClaimFrontiers,
3815 input: OrdinaryRecordProjectionInput,
3816 error: OrdinaryProjectionError,
3817) -> Box<OrdinaryRecordProjectionFailure> {
3818 Box::new(OrdinaryRecordProjectionFailure {
3819 frontiers,
3820 input,
3821 error,
3822 })
3823}
3824
3825fn preflight_ordinary_sequence_owners(
3826 sequence: &SequenceClaimFrontier,
3827 delta: u64,
3828) -> Result<(), OrdinaryProjectionError> {
3829 if !sequence.immutable_candidates.is_empty() {
3830 return Err(OrdinaryProjectionError::SequenceRelocation);
3831 }
3832 for claim in &sequence.movable_claims {
3833 claim
3834 .delivery_seq
3835 .checked_add(delta)
3836 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3837 }
3838 for range in &sequence.products.live_times_terminal {
3839 range
3840 .start
3841 .checked_add(delta)
3842 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3843 }
3844 if let Some(range) = &sequence.products.live_times_replacement_terminal {
3845 range
3846 .start
3847 .checked_add(delta)
3848 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3849 }
3850 for range in &sequence.products.other_live_times_exit {
3851 range
3852 .start
3853 .checked_add(delta)
3854 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3855 }
3856 if let Some(recovery) = &sequence.recovery {
3857 if let Some(terminal) = &recovery.terminal {
3858 terminal
3859 .delivery_seq
3860 .checked_add(delta)
3861 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3862 }
3863 recovery
3864 .recovery_attach_seq
3865 .checked_add(delta)
3866 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3867 recovery
3868 .replacement_terminal_seq
3869 .checked_add(delta)
3870 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
3871 }
3872 Ok(())
3873}
3874
3875fn relay_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
3876 for claim in &mut sequence.movable_claims {
3877 claim.delivery_seq = claim.delivery_seq.wrapping_add(delta);
3878 }
3879 for range in &mut sequence.products.live_times_terminal {
3880 range.start = range.start.wrapping_add(delta);
3881 }
3882 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
3883 range.start = range.start.wrapping_add(delta);
3884 }
3885 for range in &mut sequence.products.other_live_times_exit {
3886 range.start = range.start.wrapping_add(delta);
3887 }
3888 if let Some(recovery) = &mut sequence.recovery {
3889 if let Some(terminal) = &mut recovery.terminal {
3890 terminal.delivery_seq = terminal.delivery_seq.wrapping_add(delta);
3891 }
3892 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_add(delta);
3893 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_add(delta);
3894 }
3895}
3896
3897fn rollback_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
3898 for claim in &mut sequence.movable_claims {
3899 claim.delivery_seq = claim.delivery_seq.wrapping_sub(delta);
3900 }
3901 for range in &mut sequence.products.live_times_terminal {
3902 range.start = range.start.wrapping_sub(delta);
3903 }
3904 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
3905 range.start = range.start.wrapping_sub(delta);
3906 }
3907 for range in &mut sequence.products.other_live_times_exit {
3908 range.start = range.start.wrapping_sub(delta);
3909 }
3910 if let Some(recovery) = &mut sequence.recovery {
3911 if let Some(terminal) = &mut recovery.terminal {
3912 terminal.delivery_seq = terminal.delivery_seq.wrapping_sub(delta);
3913 }
3914 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_sub(delta);
3915 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_sub(delta);
3916 }
3917}
3918
3919fn preflight_ordinary_order_owners(
3920 order: &OrderClaimFrontier,
3921) -> Result<(), OrdinaryProjectionError> {
3922 if !order.immutable_candidates.is_empty() {
3923 return Err(OrdinaryProjectionError::OrderRelocation);
3924 }
3925 for claim in &order.movable_claims {
3926 claim
3927 .transaction_order
3928 .checked_add(1)
3929 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
3930 }
3931 if let Some(recovery) = &order.recovery {
3932 if let Some(active_binding) = &recovery.active_binding {
3933 active_binding
3934 .transaction_order
3935 .checked_add(1)
3936 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
3937 }
3938 recovery
3939 .recovery_operation_order
3940 .checked_add(1)
3941 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
3942 recovery
3943 .replacement_terminal_order
3944 .checked_add(1)
3945 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
3946 }
3947 Ok(())
3948}
3949
3950fn relay_ordinary_order_owners(order: &mut OrderClaimFrontier) {
3951 for claim in &mut order.movable_claims {
3952 claim.transaction_order = claim.transaction_order.wrapping_add(1);
3953 }
3954 if let Some(recovery) = &mut order.recovery {
3955 if let Some(active_binding) = &mut recovery.active_binding {
3956 active_binding.transaction_order = active_binding.transaction_order.wrapping_add(1);
3957 }
3958 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_add(1);
3959 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_add(1);
3960 }
3961}
3962
3963fn rollback_ordinary_order_owners(order: &mut OrderClaimFrontier) {
3964 for claim in &mut order.movable_claims {
3965 claim.transaction_order = claim.transaction_order.wrapping_sub(1);
3966 }
3967 if let Some(recovery) = &mut order.recovery {
3968 if let Some(active_binding) = &mut recovery.active_binding {
3969 active_binding.transaction_order = active_binding.transaction_order.wrapping_sub(1);
3970 }
3971 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_sub(1);
3972 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_sub(1);
3973 }
3974}
3975
3976fn validate_leave_identity<F>(
3977 frontiers: &ClaimFrontiers,
3978 member: &LiveMember<F>,
3979) -> Result<(), PrepareLeaveAuthorityError> {
3980 if member.conversation_id() != frontiers.conversation_id {
3981 return Err(PrepareLeaveAuthorityError::Conversation);
3982 }
3983 let Some(participant) =
3984 active_participant(&frontiers.active_identities, member.participant_id())
3985 else {
3986 return Err(PrepareLeaveAuthorityError::Identity);
3987 };
3988 if participant.cursor != member.cursor() {
3989 return Err(PrepareLeaveAuthorityError::Identity);
3990 }
3991 Ok(())
3992}
3993
3994fn validate_settled_leave_prestate<F>(
3995 frontiers: &ClaimFrontiers,
3996 member: &LiveMember<F>,
3997 binding_state: BindingState,
3998) -> Result<(ParticipantId, Option<BindingEpoch>), PrepareLeaveAuthorityError> {
3999 let participant_id = member.participant_id();
4000 validate_leave_identity(frontiers, member)?;
4001 if !frontiers.order.immutable_candidates.is_empty() {
4002 return Err(PrepareLeaveAuthorityError::ImmutablePrefix);
4003 }
4004 let ended_binding_epoch = match binding_state {
4005 BindingState::Detached => {
4006 let Some(participant) =
4007 active_participant(&frontiers.active_identities, participant_id)
4008 else {
4009 return Err(PrepareLeaveAuthorityError::Identity);
4010 };
4011 if !matches!(participant.binding, FrontierBinding::Detached(_)) {
4012 return Err(PrepareLeaveAuthorityError::Binding);
4013 }
4014 None
4015 }
4016 BindingState::Bound(binding)
4017 if binding.conversation_id == frontiers.conversation_id
4018 && binding.participant_id == participant_id =>
4019 {
4020 let Some(participant) =
4021 active_participant(&frontiers.active_identities, participant_id)
4022 else {
4023 return Err(PrepareLeaveAuthorityError::Identity);
4024 };
4025 if participant.binding != FrontierBinding::Bound(binding.binding_epoch) {
4026 return Err(PrepareLeaveAuthorityError::Binding);
4027 }
4028 Some(binding.binding_epoch)
4029 }
4030 BindingState::Bound(_) | BindingState::PendingFinalization(_) => {
4031 return Err(PrepareLeaveAuthorityError::Binding);
4032 }
4033 };
4034 Ok((participant_id, ended_binding_epoch))
4035}
4036
4037fn validate_pending_leave_prestate<F>(
4038 frontiers: &ClaimFrontiers,
4039 member: &LiveMember<F>,
4040 pending: PendingFinalization,
4041) -> Result<(ParticipantId, super::AdmissionOrder), PrepareLeaveAuthorityError> {
4042 let participant_id = member.participant_id();
4043 validate_leave_identity(frontiers, member)?;
4044 if pending.conversation_id() != frontiers.conversation_id
4045 || pending.participant_id() != participant_id
4046 {
4047 return Err(PrepareLeaveAuthorityError::Binding);
4048 }
4049 let Some(participant) = active_participant(&frontiers.active_identities, participant_id) else {
4050 return Err(PrepareLeaveAuthorityError::Identity);
4051 };
4052 if participant.binding != FrontierBinding::Detached(pending.binding_epoch()) {
4053 return Err(PrepareLeaveAuthorityError::Binding);
4054 }
4055 let expected_order = pending.admission_order();
4056 let exact_sequence_candidate = matches!(
4057 frontiers.sequence.immutable_candidates.as_slice(),
4058 [ImmutableSequenceCandidate::BindingTerminal {
4059 admission_order,
4060 owner,
4061 ..
4062 }] if *admission_order == expected_order
4063 && owner.participant_index == participant_id
4064 && owner.binding_epoch == pending.binding_epoch()
4065 );
4066 let exact_order_candidate = matches!(
4067 frontiers.order.immutable_candidates.as_slice(),
4068 [ImmutableOrderCandidateMajor {
4069 transaction_order,
4070 candidate_keys,
4071 }] if *transaction_order == expected_order.transaction_order()
4072 && candidate_keys.as_slice() == [expected_order]
4073 );
4074 if !exact_sequence_candidate || !exact_order_candidate {
4075 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4076 }
4077 Ok((participant_id, expected_order))
4078}
4079
4080#[derive(Clone, Copy)]
4081enum LeaveRelayUnit {
4082 Direct(MovableOrderClaim),
4083 Recovery(RecoveryOrderBlock),
4084}
4085
4086impl LeaveRelayUnit {
4087 fn start(self) -> TransactionOrder {
4088 match self {
4089 Self::Direct(claim) => claim.transaction_order,
4090 Self::Recovery(block) => block_start_validated_order(block),
4091 }
4092 }
4093
4094 const fn len(self) -> u64 {
4095 match self {
4096 Self::Direct(_) => 1,
4097 Self::Recovery(block) => match block.active_binding {
4098 Some(_) => 3,
4099 None => 2,
4100 },
4101 }
4102 }
4103}
4104
4105struct LeaveOrderSelection {
4106 units: Vec<LeaveRelayUnit>,
4107 selected_major: TransactionOrder,
4108}
4109
4110fn select_leave_order(
4111 order: &OrderClaimFrontier,
4112 participant_id: ParticipantId,
4113 ended_binding_epoch: Option<BindingEpoch>,
4114 pending_order: Option<super::AdmissionOrder>,
4115) -> Result<LeaveOrderSelection, PrepareLeaveAuthorityError> {
4116 let Some(exit_index) = order.movable_claims.iter().position(|claim| {
4117 claim.owner
4118 == OrderDirectOwner::MembershipExit {
4119 participant_index: participant_id,
4120 }
4121 }) else {
4122 return Err(PrepareLeaveAuthorityError::MembershipExitClaim);
4123 };
4124 let exit_claim = order.movable_claims[exit_index];
4125 if pending_order
4126 .is_some_and(|pending| pending.transaction_order() >= exit_claim.transaction_order)
4127 {
4128 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4129 }
4130 let active_index = matching_active_claim(order, participant_id, ended_binding_epoch)?;
4131 let mut units = Vec::new();
4132 for (index, claim) in order.movable_claims.iter().copied().enumerate() {
4133 if index != exit_index && Some(index) != active_index {
4134 units.push(LeaveRelayUnit::Direct(claim));
4135 }
4136 }
4137 if let Some(recovery) = order
4138 .recovery
4139 .filter(|recovery| recovery.participant_index != participant_id)
4140 {
4141 units.push(LeaveRelayUnit::Recovery(recovery));
4142 }
4143 units.sort_by_key(|unit| unit.start());
4144 let surviving_handles: u128 = units.iter().map(|unit| u128::from(unit.len())).sum();
4145 let exit_major = exit_claim.transaction_order;
4146 let later_handle_fits = u128::from(u64::MAX - exit_major) >= surviving_handles;
4147 let selected_major = if later_handle_fits {
4148 exit_major
4149 } else if pending_order.is_some() {
4150 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4151 } else {
4152 u64::try_from(order_frontier_start(order.ledger.high()))
4153 .map_err(|_| PrepareLeaveAuthorityError::OrderCapacity)?
4154 };
4155 Ok(LeaveOrderSelection {
4156 units,
4157 selected_major,
4158 })
4159}
4160
4161fn matching_active_claim(
4162 order: &OrderClaimFrontier,
4163 participant_id: ParticipantId,
4164 ended_binding_epoch: Option<BindingEpoch>,
4165) -> Result<Option<usize>, PrepareLeaveAuthorityError> {
4166 let Some(binding_epoch) = ended_binding_epoch else {
4167 return Ok(None);
4168 };
4169 let expected = OrderDirectOwner::ActiveBindingTerminal(BindingTerminalOwner {
4170 participant_index: participant_id,
4171 binding_epoch,
4172 });
4173 order
4174 .movable_claims
4175 .iter()
4176 .position(|claim| claim.owner == expected)
4177 .map(Some)
4178 .ok_or(PrepareLeaveAuthorityError::ActiveBindingClaim)
4179}
4180
4181fn relay_leave_order_units(
4182 units: Vec<LeaveRelayUnit>,
4183 selected_major: TransactionOrder,
4184) -> Result<(Vec<MovableOrderClaim>, Option<RecoveryOrderBlock>), PrepareLeaveAuthorityError> {
4185 let mut cursor = selected_major.checked_add(1);
4186 let mut movable_claims = Vec::new();
4187 let mut recovery = None;
4188 for unit in units {
4189 match unit {
4190 LeaveRelayUnit::Direct(mut claim) => {
4191 let Some(position) = cursor else {
4192 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4193 };
4194 claim.transaction_order = position;
4195 movable_claims.push(claim);
4196 cursor = position.checked_add(1);
4197 }
4198 LeaveRelayUnit::Recovery(block) => {
4199 let (relayed, next) = relay_recovery_block(block, cursor)?;
4200 recovery = Some(relayed);
4201 cursor = next;
4202 }
4203 }
4204 }
4205 movable_claims.sort_by_key(|claim| claim.transaction_order);
4206 Ok((movable_claims, recovery))
4207}
4208
4209const fn relay_recovery_block(
4210 block: RecoveryOrderBlock,
4211 cursor: Option<TransactionOrder>,
4212) -> Result<(RecoveryOrderBlock, Option<TransactionOrder>), PrepareLeaveAuthorityError> {
4213 let mut next = cursor;
4214 let active_binding = if let Some(mut active) = block.active_binding {
4215 let Some(position) = next else {
4216 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4217 };
4218 active.transaction_order = position;
4219 next = position.checked_add(1);
4220 Some(active)
4221 } else {
4222 None
4223 };
4224 let Some(recovery_operation_order) = next else {
4225 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4226 };
4227 let Some(replacement_terminal_order) = recovery_operation_order.checked_add(1) else {
4228 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4229 };
4230 Ok((
4231 RecoveryOrderBlock {
4232 active_binding,
4233 recovery_operation_order,
4234 replacement_terminal_order,
4235 participant_index: block.participant_index,
4236 marker_delivery_seq: block.marker_delivery_seq,
4237 recovered_binding_epoch: block.recovered_binding_epoch,
4238 },
4239 replacement_terminal_order.checked_add(1),
4240 ))
4241}
4242
4243fn leave_resulting_order_ledger(
4244 selected_major: TransactionOrder,
4245 movable_claims: &[MovableOrderClaim],
4246 recovery: Option<RecoveryOrderBlock>,
4247) -> Result<OrderLedger, PrepareLeaveAuthorityError> {
4248 let active_binding_terminals =
4249 usize_to_u64(
4250 movable_claims
4251 .iter()
4252 .filter(|claim| matches!(claim.owner, OrderDirectOwner::ActiveBindingTerminal(_)))
4253 .count(),
4254 ) + u64::from(recovery.is_some_and(|block| block.active_binding.is_some()));
4255 let membership_exits = usize_to_u64(
4256 movable_claims
4257 .iter()
4258 .filter(|claim| matches!(claim.owner, OrderDirectOwner::MembershipExit { .. }))
4259 .count(),
4260 );
4261 let has_recovery = recovery.is_some();
4262 let resulting_claims = OrderClaims::new(
4263 active_binding_terminals,
4264 membership_exits,
4265 has_recovery,
4266 has_recovery,
4267 )
4268 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)?;
4269 OrderLedger::try_new(OrderHigh::Allocated(selected_major), resulting_claims)
4270 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)
4271}
4272
4273fn consume_leave_order_lane(
4274 order: &mut OrderClaimFrontier,
4275 participant_id: ParticipantId,
4276 ended_binding_epoch: Option<BindingEpoch>,
4277 pending_order: Option<super::AdmissionOrder>,
4278) -> Result<TransactionOrder, PrepareLeaveAuthorityError> {
4279 let selection = select_leave_order(order, participant_id, ended_binding_epoch, pending_order)?;
4280 let (movable_claims, recovery) =
4281 relay_leave_order_units(selection.units, selection.selected_major)?;
4282 let ledger = leave_resulting_order_ledger(selection.selected_major, &movable_claims, recovery)?;
4283 *order = OrderClaimFrontier {
4284 ledger,
4285 movable_claims,
4286 immutable_candidates: Vec::new(),
4287 recovery,
4288 };
4289 Ok(selection.selected_major)
4290}
4291
4292impl ClaimFrontiersPrevalidated {
4293 #[must_use]
4295 pub(super) const fn conversation_id(&self) -> ConversationId {
4296 self.conversation_id
4297 }
4298
4299 pub(super) fn take_marker_record(
4305 &mut self,
4306 request: MarkerRecordRequest,
4307 ) -> Option<ValidatedMarkerRecord> {
4308 if self.issued_marker_record.is_some() {
4309 return None;
4310 }
4311 let record = self
4312 .retained_records
4313 .iter()
4314 .find(|record| record.delivery_seq == request.marker_delivery_seq)
4315 .copied()?;
4316 let RetainedCausalRecordKind::CompactionMarker {
4317 participant_index,
4318 provenance,
4319 } = record.kind
4320 else {
4321 return None;
4322 };
4323 if participant_index != request.participant_index {
4324 return None;
4325 }
4326 let participant = active_participant(&self.active_identities, participant_index)?;
4327 let historical_delivery = self
4328 .historical_marker_deliveries
4329 .iter()
4330 .find(|authority| authority.marker_delivery_seq == request.marker_delivery_seq);
4331 let (target_binding, occurrence) = match request.use_kind {
4332 MarkerRecordUse::Planned(target) => {
4333 if participant.binding != target || historical_delivery.is_some() {
4334 return None;
4335 }
4336 (target, MarkerRecordOccurrence::Undelivered)
4337 }
4338 MarkerRecordUse::Delivered(target) => {
4339 let delivered_binding_epoch = binding_epoch(target);
4340 if participant.binding != target
4341 || !historical_delivery.is_some_and(|authority| {
4342 authority.participant_index == participant_index
4343 && authority.delivered_binding_epoch == delivered_binding_epoch
4344 })
4345 {
4346 return None;
4347 }
4348 (target, MarkerRecordOccurrence::Delivered)
4349 }
4350 MarkerRecordUse::Recovered {
4351 prior_binding_epoch,
4352 recovered_binding_epoch,
4353 } => {
4354 if participant.binding != FrontierBinding::Detached(recovered_binding_epoch)
4355 || !historical_delivery.is_some_and(|authority| {
4356 authority.participant_index == participant_index
4357 && authority.delivered_binding_epoch == prior_binding_epoch
4358 })
4359 {
4360 return None;
4361 }
4362 (
4363 FrontierBinding::Detached(prior_binding_epoch),
4364 MarkerRecordOccurrence::Delivered,
4365 )
4366 }
4367 };
4368 self.issued_marker_record = Some(request);
4369 Some(ValidatedMarkerRecord {
4370 conversation_id: self.conversation_id,
4371 record,
4372 provenance,
4373 target_binding,
4374 occurrence,
4375 seal: MarkerAuthoritySeal::Validated,
4376 })
4377 }
4378
4379 pub(super) fn finish(
4382 self,
4383 edge: Option<super::StoredEdge>,
4384 ) -> Result<ClaimFrontiers, ParticipantStateCorruptReason> {
4385 self.validate_current_marker_edge(edge)?;
4386 self.validate_historical_delivery_consumers(edge)?;
4387 let recovery_provenance = self.resolve_recovery_provenance(edge)?;
4388 let sequence = restore_sequence_frontier(
4389 &self.active_identities,
4390 self.sequence_restore,
4391 self.retained_floor,
4392 self.sequence_ledger,
4393 recovery_provenance,
4394 &self.retained_records,
4395 &self.historical_causal_authorities,
4396 )
4397 .map_err(corrupt_frontier)?;
4398 let order = restore_order_frontier(
4399 &self.active_identities,
4400 self.order_restore,
4401 self.order_ledger,
4402 recovery_provenance,
4403 )
4404 .map_err(corrupt_frontier)?;
4405 validate_cross_counter(&sequence, &order).map_err(corrupt_frontier)?;
4406 Ok(ClaimFrontiers {
4407 conversation_id: self.conversation_id,
4408 active_identities: self.active_identities,
4409 identity_slot_limit: self.identity_slot_limit,
4410 retained_floor: self.retained_floor,
4411 retained_records: self.retained_records,
4412 marker_records: self.marker_records,
4413 fenced_marker_issued: None,
4414 sequence,
4415 order,
4416 })
4417 }
4418
4419 fn validate_historical_delivery_consumers(
4420 &self,
4421 edge: Option<super::StoredEdge>,
4422 ) -> Result<(), ParticipantStateCorruptReason> {
4423 for history in &self.historical_marker_deliveries {
4424 let recovered_origin = self.binding_origins.iter().any(|origin| {
4425 origin.participant_id() == history.participant_index
4426 && origin.recovered_marker()
4427 == Some((history.marker_delivery_seq, history.delivered_binding_epoch))
4428 });
4429 let current_marker_edge = match edge {
4430 Some(super::StoredEdge::ParticipantCursorProgress(progress)) => {
4431 progress.participant_id() == history.participant_index
4432 && progress.marker_delivery_seq() == Some(history.marker_delivery_seq)
4433 && progress.binding_epoch() == history.delivered_binding_epoch
4434 }
4435 Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) => {
4436 recovery.participant_id() == history.participant_index
4437 && recovery.marker_delivery_seq() == history.marker_delivery_seq
4438 && recovery.prior_binding_epoch() == history.delivered_binding_epoch
4439 }
4440 Some(
4441 super::StoredEdge::ObserverProjection(_)
4442 | super::StoredEdge::PhysicalCompaction(_)
4443 | super::StoredEdge::MarkerDelivery(_)
4444 | super::StoredEdge::DetachedMarkerRelease(_)
4445 | super::StoredEdge::DetachedCursorRelease(_),
4446 )
4447 | None => false,
4448 };
4449 if !recovered_origin && !current_marker_edge {
4450 return Err(self.marker_corruption(history.marker_delivery_seq));
4451 }
4452 }
4453 Ok(())
4454 }
4455
4456 fn validate_current_marker_edge(
4457 &self,
4458 edge: Option<super::StoredEdge>,
4459 ) -> Result<(), ParticipantStateCorruptReason> {
4460 if let Some(MarkerRecordRequest {
4461 participant_index,
4462 marker_delivery_seq,
4463 use_kind:
4464 MarkerRecordUse::Recovered {
4465 recovered_binding_epoch,
4466 ..
4467 },
4468 }) = self.issued_marker_record
4469 {
4470 let recovered_edge_matches = matches!(
4471 edge,
4472 Some(super::StoredEdge::DetachedCursorRelease(release))
4473 if release.participant_id() == participant_index
4474 && release.last_dead_binding_epoch() == recovered_binding_epoch
4475 );
4476 if !recovered_edge_matches {
4477 return Err(self.marker_corruption(marker_delivery_seq));
4478 }
4479 return Ok(());
4480 }
4481 let Some(context) = edge.and_then(marker_edge_context) else {
4482 return Ok(());
4483 };
4484 if self
4485 .issued_marker_record
4486 .is_none_or(|request| request.marker_delivery_seq != context.marker_delivery_seq)
4487 || !self.marker_context_matches(context)
4488 {
4489 return Err(self.marker_corruption(context.marker_delivery_seq));
4490 }
4491 Ok(())
4492 }
4493
4494 fn resolve_recovery_provenance(
4495 &self,
4496 edge: Option<super::StoredEdge>,
4497 ) -> Result<Option<RecoveryClaimProvenance>, ParticipantStateCorruptReason> {
4498 let Some(marker_delivery_seq) = self.recovery_marker_delivery_seq else {
4499 return Ok(None);
4500 };
4501
4502 if let Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) = edge {
4503 return self
4504 .resolve_postfate_recovery_provenance(marker_delivery_seq, recovery)
4505 .map(Some);
4506 }
4507
4508 let candidate =
4509 self.sequence_restore.immutable_candidates.iter().find_map(
4510 |candidate| match candidate {
4511 ImmutableSequenceCandidate::Marker(marker)
4512 if marker.delivery_seq == marker_delivery_seq
4513 && matches!(marker.target_binding, FrontierBinding::Bound(_)) =>
4514 {
4515 Some(*marker)
4516 }
4517 _ => None,
4518 },
4519 );
4520 let retained = self.retained_records.iter().find_map(|record| {
4521 let RetainedCausalRecordKind::CompactionMarker {
4522 participant_index, ..
4523 } = record.kind
4524 else {
4525 return None;
4526 };
4527 if record.delivery_seq != marker_delivery_seq {
4528 return None;
4529 }
4530 let participant = active_participant(&self.active_identities, participant_index)?;
4531 let FrontierBinding::Bound(binding_epoch) = participant.binding else {
4532 return None;
4533 };
4534 let historical = self.historical_marker_deliveries.iter().find(|history| {
4535 history.participant_index == participant_index
4536 && history.marker_delivery_seq == marker_delivery_seq
4537 })?;
4538 Some((
4539 participant_index,
4540 binding_epoch,
4541 historical.delivered_binding_epoch,
4542 ))
4543 });
4544 let provenance = match (candidate, retained) {
4545 (Some(marker), None) => RecoveryClaimProvenance {
4546 participant_index: marker.admission_order.participant_index(),
4547 marker_delivery_seq,
4548 prior_binding_epoch: binding_epoch(marker.target_binding),
4549 current_binding_epoch: binding_epoch(marker.target_binding),
4550 phase: RecoveryClaimPhase::PreFate,
4551 },
4552 (None, Some((participant_index, current_binding_epoch, prior_binding_epoch))) => {
4553 let phase = if current_binding_epoch == prior_binding_epoch {
4554 let Some(context) = edge.and_then(marker_edge_context) else {
4555 return Err(self.marker_corruption(marker_delivery_seq));
4556 };
4557 if context.marker_delivery_seq != marker_delivery_seq
4558 || context.participant_index != participant_index
4559 || context.binding_epoch != prior_binding_epoch
4560 || context.target_binding != FrontierBinding::Bound(prior_binding_epoch)
4561 {
4562 return Err(self.marker_corruption(marker_delivery_seq));
4563 }
4564 RecoveryClaimPhase::PreFate
4565 } else {
4566 let recovered_origin_matches = self.binding_origins.iter().any(|origin| {
4567 origin.participant_id() == participant_index
4568 && origin.binding_epoch() == current_binding_epoch
4569 && origin.recovered_marker()
4570 == Some((marker_delivery_seq, prior_binding_epoch))
4571 });
4572 if !recovered_origin_matches
4573 || !matches!(
4574 edge,
4575 Some(
4576 super::StoredEdge::ObserverProjection(_)
4577 | super::StoredEdge::PhysicalCompaction(_)
4578 )
4579 )
4580 {
4581 return Err(self.marker_corruption(marker_delivery_seq));
4582 }
4583 RecoveryClaimPhase::RecoveredBound
4584 };
4585 RecoveryClaimProvenance {
4586 participant_index,
4587 marker_delivery_seq,
4588 prior_binding_epoch,
4589 current_binding_epoch,
4590 phase,
4591 }
4592 }
4593 (Some(_), Some(_)) | (None, None) => {
4594 return Err(self.marker_corruption(marker_delivery_seq));
4595 }
4596 };
4597 Ok(Some(provenance))
4598 }
4599
4600 fn resolve_postfate_recovery_provenance(
4601 &self,
4602 marker_delivery_seq: DeliverySeq,
4603 recovery: super::DetachedCredentialRecovery,
4604 ) -> Result<RecoveryClaimProvenance, ParticipantStateCorruptReason> {
4605 let provenance = RecoveryClaimProvenance {
4606 participant_index: recovery.participant_id(),
4607 marker_delivery_seq: recovery.marker_delivery_seq(),
4608 prior_binding_epoch: recovery.prior_binding_epoch(),
4609 current_binding_epoch: recovery.prior_binding_epoch(),
4610 phase: RecoveryClaimPhase::PostFate,
4611 };
4612 let context = MarkerEdgeContext {
4613 participant_index: provenance.participant_index,
4614 marker_delivery_seq: provenance.marker_delivery_seq,
4615 binding_epoch: provenance.prior_binding_epoch,
4616 target_binding: FrontierBinding::Detached(provenance.prior_binding_epoch),
4617 };
4618 if marker_delivery_seq != provenance.marker_delivery_seq
4619 || self
4620 .issued_marker_record
4621 .is_none_or(|request| request.marker_delivery_seq != marker_delivery_seq)
4622 || !self.marker_context_matches(context)
4623 {
4624 return Err(self.marker_corruption(marker_delivery_seq));
4625 }
4626 Ok(provenance)
4627 }
4628
4629 fn marker_context_matches(&self, context: MarkerEdgeContext) -> bool {
4630 self.marker_records.iter().any(|record| {
4631 matches!(
4632 record.kind,
4633 RetainedCausalRecordKind::CompactionMarker {
4634 participant_index,
4635 ..
4636 } if participant_index == context.participant_index
4637 ) && record.delivery_seq == context.marker_delivery_seq
4638 && active_participant(&self.active_identities, context.participant_index)
4639 .is_some_and(|participant| participant.binding == context.target_binding)
4640 })
4641 }
4642
4643 fn marker_corruption(&self, delivery_seq: DeliverySeq) -> ParticipantStateCorruptReason {
4644 corrupt_frontier(sequence_error(
4645 sequence_ordinal(self.sequence_ledger, delivery_seq),
4646 ClaimFrontierInvalidReason::RecoveryBlock,
4647 ))
4648 }
4649}
4650
4651#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4652struct MarkerEdgeContext {
4653 participant_index: ParticipantId,
4654 marker_delivery_seq: DeliverySeq,
4655 binding_epoch: BindingEpoch,
4656 target_binding: FrontierBinding,
4657}
4658
4659fn marker_edge_context(edge: super::StoredEdge) -> Option<MarkerEdgeContext> {
4660 match edge {
4661 super::StoredEdge::MarkerDelivery(delivery) => Some(MarkerEdgeContext {
4662 participant_index: delivery.participant_id(),
4663 marker_delivery_seq: delivery.marker_delivery_seq(),
4664 binding_epoch: delivery.binding_epoch(),
4665 target_binding: FrontierBinding::Bound(delivery.binding_epoch()),
4666 }),
4667 super::StoredEdge::ParticipantCursorProgress(progress) => {
4668 let marker_delivery_seq = progress.marker_delivery_seq()?;
4669 Some(MarkerEdgeContext {
4670 participant_index: progress.participant_id(),
4671 marker_delivery_seq,
4672 binding_epoch: progress.binding_epoch(),
4673 target_binding: FrontierBinding::Bound(progress.binding_epoch()),
4674 })
4675 }
4676 super::StoredEdge::DetachedCredentialRecovery(recovery) => Some(MarkerEdgeContext {
4677 participant_index: recovery.participant_id(),
4678 marker_delivery_seq: recovery.marker_delivery_seq(),
4679 binding_epoch: recovery.prior_binding_epoch(),
4680 target_binding: FrontierBinding::Detached(recovery.prior_binding_epoch()),
4681 }),
4682 super::StoredEdge::DetachedMarkerRelease(release) => Some(MarkerEdgeContext {
4683 participant_index: release.participant_id(),
4684 marker_delivery_seq: release.marker_delivery_seq(),
4685 binding_epoch: release.last_dead_binding_epoch(),
4686 target_binding: FrontierBinding::Detached(release.last_dead_binding_epoch()),
4687 }),
4688 super::StoredEdge::ObserverProjection(_)
4689 | super::StoredEdge::PhysicalCompaction(_)
4690 | super::StoredEdge::DetachedCursorRelease(_) => None,
4691 }
4692}
4693
4694#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4695#[repr(u8)]
4696enum SequenceClass {
4697 Exit = 0,
4698 Terminal = 1,
4699 Marker = 2,
4700 RecoveryAttach = 3,
4701 RecoveryReplacementTerminal = 4,
4702 LiveTimesTerminal = 5,
4703 LiveTimesReplacementTerminal = 6,
4704 OtherLiveTimesExit = 7,
4705}
4706
4707#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4708#[repr(u8)]
4709enum OrderClass {
4710 ActiveBindingTerminal = 0,
4711 MembershipExit = 1,
4712 RecoveryOperation = 2,
4713 RecoveryReplacementTerminal = 3,
4714}
4715
4716#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4717struct NumericSegment<C> {
4718 start: u128,
4719 length: u128,
4720 class: Option<C>,
4721 immutable: bool,
4722}
4723
4724fn first_duplicate_candidate_key(
4725 candidates: &[ImmutableSequenceCandidate],
4726 retained_records: &[RetainedCausalRecord],
4727) -> Option<super::AdmissionOrder> {
4728 let mut keys: Vec<_> = candidates
4729 .iter()
4730 .map(|candidate| candidate.admission_order())
4731 .chain(retained_records.iter().map(|record| record.admission_order))
4732 .collect();
4733 keys.sort_unstable();
4734 let mut previous = None;
4735 for key in keys {
4736 if previous == Some(key) {
4737 return Some(key);
4738 }
4739 previous = Some(key);
4740 }
4741 None
4742}
4743
4744fn validate_unique_candidate_keys(
4745 candidates: &[ImmutableSequenceCandidate],
4746 retained_records: &[RetainedCausalRecord],
4747) -> Result<(), ParticipantStateCorruptReason> {
4748 let Some(order) = first_duplicate_candidate_key(candidates, retained_records) else {
4749 return Ok(());
4750 };
4751 Err(ParticipantStateCorruptReason::DuplicateCandidateKey {
4752 transaction_order: order.transaction_order(),
4753 candidate_phase: order.candidate_phase(),
4754 participant_index: order.participant_index(),
4755 })
4756}
4757
4758fn validate_sequence_numeric(
4759 restore: &SequenceClaimFrontierRestore,
4760 ledger: SequenceLedger,
4761) -> Result<(), ClaimFrontierError> {
4762 let mut segments = sequence_segments(restore);
4763 validate_numeric_segments(
4764 ClaimFrontierCounter::DeliverySequence,
4765 u128::from(ledger.high_watermark()) + 1,
4766 ledger.required_reserve(),
4767 &mut segments,
4768 &sequence_expected_counts(ledger),
4769 )
4770}
4771
4772fn validate_order_numeric(
4773 restore: &OrderClaimFrontierRestore,
4774 ledger: OrderLedger,
4775) -> Result<(), ClaimFrontierError> {
4776 let mut segments = order_segments(restore, ledger.high());
4777 validate_numeric_segments(
4778 ClaimFrontierCounter::TransactionOrder,
4779 order_frontier_start(ledger.high()),
4780 order_frontier_candidate_count(restore, ledger.high()) + ledger.claims().total(),
4781 &mut segments,
4782 &order_expected_counts(ledger),
4783 )
4784}
4785
4786fn validate_bounded_shape(
4787 restore: &ClaimFrontiersRestore,
4788 sequence_ledger: SequenceLedger,
4789) -> Result<(), ClaimFrontierError> {
4790 let identity_limit = u128::from(restore.identity_slot_limit);
4791 let twice_identity_limit = identity_limit.saturating_mul(2);
4792 let order_candidate_keys = restore
4793 .order
4794 .immutable_candidates
4795 .iter()
4796 .fold(0_u128, |count, candidate| {
4797 count.saturating_add(usize_to_u128(candidate.candidate_keys.len()))
4798 });
4799 let bounded = usize_to_u128(restore.active_identities.len()) <= identity_limit
4800 && usize_to_u128(restore.sequence.movable_claims.len()) <= twice_identity_limit
4801 && usize_to_u128(restore.sequence.immutable_candidates.len()) <= twice_identity_limit
4802 && usize_to_u128(restore.sequence.products.live_times_terminal.len()) <= identity_limit
4803 && usize_to_u128(restore.sequence.products.other_live_times_exit.len()) <= identity_limit
4804 && usize_to_u128(restore.historical_marker_deliveries.len())
4805 <= u128::from(restore.retained_record_limit)
4806 && usize_to_u128(restore.historical_causal_facts.len()) <= twice_identity_limit
4807 && usize_to_u128(restore.order.movable_claims.len()) <= twice_identity_limit
4808 && usize_to_u128(restore.order.immutable_candidates.len()) <= twice_identity_limit
4809 && order_candidate_keys <= twice_identity_limit;
4810 if bounded {
4811 Ok(())
4812 } else {
4813 Err(sequence_error(
4814 sequence_ledger.required_reserve(),
4815 ClaimFrontierInvalidReason::LogicalOwner,
4816 ))
4817 }
4818}
4819
4820fn validated_retained_records(
4821 mut records: Vec<RetainedCausalRecord>,
4822 retained_floor: u128,
4823 retained_record_limit: u64,
4824 identity_slot_limit: u64,
4825 ledger: SequenceLedger,
4826) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
4827 let high_end = u128::from(ledger.high_watermark()) + 1;
4828 let expected_count = high_end.saturating_sub(retained_floor);
4829 if retained_floor > high_end
4830 || usize_to_u128(records.len()) > u128::from(retained_record_limit)
4831 || usize_to_u128(records.len()) != expected_count
4832 {
4833 return Err(sequence_error(
4834 usize_to_u128(records.len()).min(expected_count),
4835 ClaimFrontierInvalidReason::LogicalOwner,
4836 ));
4837 }
4838 records.sort_by_key(|record| record.delivery_seq);
4839 let mut previous_admission_order = None;
4840 for (record_index, record) in records.iter().enumerate() {
4841 let (participant_index, valid_kind) = match record.kind {
4842 RetainedCausalRecordKind::BindingTerminal(owner) => (
4843 owner.participant_index,
4844 record.admission_order.candidate_phase() == CandidatePhase::BindingTerminal
4845 && record.admission_order.participant_index() == owner.participant_index,
4846 ),
4847 RetainedCausalRecordKind::MembershipExit { participant_index } => (
4848 participant_index,
4849 record.admission_order.candidate_phase() == CandidatePhase::MembershipExit
4850 && record.admission_order.participant_index() == participant_index,
4851 ),
4852 RetainedCausalRecordKind::AttachLifecycle {
4853 participant_index, ..
4854 } => (
4855 participant_index,
4856 record.admission_order.candidate_phase() == CandidatePhase::AttachLifecycle
4857 && record.admission_order.participant_index() == participant_index,
4858 ),
4859 RetainedCausalRecordKind::OrdinaryRecord { participant_index } => (
4860 participant_index,
4861 record.admission_order.candidate_phase() == CandidatePhase::OrdinaryRecord
4862 && record.admission_order.participant_index() == participant_index,
4863 ),
4864 RetainedCausalRecordKind::CompactionMarker {
4865 participant_index,
4866 provenance,
4867 } => (
4868 participant_index,
4869 record.admission_order.candidate_phase() == CandidatePhase::CompactionMarker
4870 && record.admission_order.participant_index() == participant_index
4871 && marker_provenance_targets(provenance, participant_index),
4872 ),
4873 };
4874 let expected_sequence = retained_floor + rank_index(record_index);
4875 if participant_index >= identity_slot_limit
4876 || !valid_kind
4877 || previous_admission_order.is_some_and(|previous| previous >= record.admission_order)
4878 || u128::from(record.delivery_seq) != expected_sequence
4879 {
4880 return Err(sequence_error(
4881 rank_index(record_index),
4882 ClaimFrontierInvalidReason::CandidateKey,
4883 ));
4884 }
4885 previous_admission_order = Some(record.admission_order);
4886 }
4887 Ok(records)
4888}
4889
4890fn validated_active_marker_records(
4891 retained_markers: &[RetainedCausalRecord],
4892 mut active_marker_anchors: Vec<DeliverySeq>,
4893 identity_slot_limit: u64,
4894 ledger: SequenceLedger,
4895) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
4896 if usize_to_u128(active_marker_anchors.len()) > u128::from(identity_slot_limit) {
4897 return Err(sequence_error(
4898 ledger.required_reserve(),
4899 ClaimFrontierInvalidReason::LogicalOwner,
4900 ));
4901 }
4902 active_marker_anchors.sort_unstable();
4903 let mut previous_sequence = None;
4904 let mut owners = Vec::new();
4905 let mut active_records = Vec::new();
4906 for delivery_seq in active_marker_anchors {
4907 let Some(record) = retained_markers
4908 .iter()
4909 .find(|record| record.delivery_seq == delivery_seq)
4910 .copied()
4911 else {
4912 return Err(sequence_error(
4913 sequence_ordinal(ledger, delivery_seq),
4914 ClaimFrontierInvalidReason::LogicalOwner,
4915 ));
4916 };
4917 let RetainedCausalRecordKind::CompactionMarker {
4918 participant_index, ..
4919 } = record.kind
4920 else {
4921 return Err(sequence_error(
4922 sequence_ordinal(ledger, delivery_seq),
4923 ClaimFrontierInvalidReason::LogicalOwner,
4924 ));
4925 };
4926 if previous_sequence == Some(delivery_seq) || owners.contains(&participant_index) {
4927 return Err(sequence_error(
4928 sequence_ordinal(ledger, delivery_seq),
4929 ClaimFrontierInvalidReason::LogicalOwner,
4930 ));
4931 }
4932 previous_sequence = Some(delivery_seq);
4933 owners.push(participant_index);
4934 active_records.push(record);
4935 }
4936 Ok(active_records)
4937}
4938
4939fn validated_historical_marker_deliveries(
4940 mut facts: Vec<HistoricalMarkerDeliveryFactRestore>,
4941 conversation_id: ConversationId,
4942 active: &ActiveIdentityRanks,
4943 sources: &MarkerDeliverySources<'_>,
4944 retained_record_limit: u64,
4945 ledger: SequenceLedger,
4946) -> Result<Vec<HistoricalMarkerDeliveryAuthority>, ClaimFrontierError> {
4947 let MarkerDeliverySources {
4948 retained: retained_records,
4949 historical: historical_causal_authorities,
4950 candidates: immutable_candidates,
4951 } = *sources;
4952 if usize_to_u128(facts.len()) > u128::from(retained_record_limit) {
4953 return Err(sequence_error(
4954 ledger.required_reserve(),
4955 ClaimFrontierInvalidReason::LogicalOwner,
4956 ));
4957 }
4958 facts.sort_by_key(|fact| fact.marker_delivery_seq);
4959 let mut previous_sequence = None;
4960 let mut authorities = Vec::new();
4961 for fact in facts {
4962 let matching_record = retained_records.iter().find(|record| {
4963 record.delivery_seq == fact.marker_delivery_seq
4964 && matches!(
4965 record.kind,
4966 RetainedCausalRecordKind::CompactionMarker {
4967 participant_index,
4968 ..
4969 } if participant_index == fact.participant_index
4970 )
4971 });
4972 let current_bound =
4973 active_participant(active, fact.participant_index).is_some_and(|participant| {
4974 participant.binding == FrontierBinding::Bound(fact.delivered_binding_epoch)
4975 });
4976 let terminal_order = historical_causal_authorities
4977 .iter()
4978 .find_map(|authority| match authority.kind {
4979 HistoricalCausalKind::BindingTerminal(owner)
4980 if binding_terminal_matches_delivery(owner, fact) =>
4981 {
4982 Some(authority.admission_order)
4983 }
4984 HistoricalCausalKind::BindingTerminal(_)
4985 | HistoricalCausalKind::MembershipExit(_) => None,
4986 })
4987 .or_else(|| {
4988 retained_records
4989 .iter()
4990 .find_map(|record| match record.kind {
4991 RetainedCausalRecordKind::BindingTerminal(owner)
4992 if binding_terminal_matches_delivery(owner, fact) =>
4993 {
4994 Some(record.admission_order)
4995 }
4996 RetainedCausalRecordKind::BindingTerminal(_)
4997 | RetainedCausalRecordKind::MembershipExit { .. }
4998 | RetainedCausalRecordKind::AttachLifecycle { .. }
4999 | RetainedCausalRecordKind::OrdinaryRecord { .. }
5000 | RetainedCausalRecordKind::CompactionMarker { .. } => None,
5001 })
5002 })
5003 .or_else(|| {
5004 immutable_candidates
5005 .iter()
5006 .find_map(|candidate| match candidate {
5007 ImmutableSequenceCandidate::BindingTerminal {
5008 admission_order,
5009 owner,
5010 ..
5011 } if binding_terminal_matches_delivery(*owner, fact) => {
5012 Some(*admission_order)
5013 }
5014 ImmutableSequenceCandidate::BindingTerminal { .. }
5015 | ImmutableSequenceCandidate::Marker(_) => None,
5016 })
5017 });
5018 let historical_epoch_is_backed = matching_record.is_some_and(|marker_record| {
5019 current_bound
5020 || terminal_order
5021 .is_some_and(|terminal_order| terminal_order > marker_record.admission_order)
5022 });
5023 if fact.conversation_id != conversation_id
5024 || previous_sequence == Some(fact.marker_delivery_seq)
5025 || !historical_epoch_is_backed
5026 {
5027 return Err(sequence_error(
5028 sequence_ordinal(ledger, fact.marker_delivery_seq),
5029 ClaimFrontierInvalidReason::LogicalOwner,
5030 ));
5031 }
5032 previous_sequence = Some(fact.marker_delivery_seq);
5033 authorities.push(HistoricalMarkerDeliveryAuthority {
5034 participant_index: fact.participant_index,
5035 marker_delivery_seq: fact.marker_delivery_seq,
5036 delivered_binding_epoch: fact.delivered_binding_epoch,
5037 });
5038 }
5039 Ok(authorities)
5040}
5041
5042fn binding_terminal_matches_delivery(
5043 owner: BindingTerminalOwner,
5044 fact: HistoricalMarkerDeliveryFactRestore,
5045) -> bool {
5046 (owner.participant_index, owner.binding_epoch)
5047 == (fact.participant_index, fact.delivered_binding_epoch)
5048}
5049
5050struct MarkerDeliverySources<'a> {
5051 retained: &'a [RetainedCausalRecord],
5052 historical: &'a [HistoricalCausalAuthority],
5053 candidates: &'a [ImmutableSequenceCandidate],
5054}
5055
5056struct BindingOriginValidation<'a> {
5057 conversation_id: ConversationId,
5058 active: &'a ActiveIdentityRanks,
5059 origins: &'a [BindingOrigin],
5060 retained_records: &'a [RetainedCausalRecord],
5061 causal_authorities: &'a [HistoricalCausalAuthority],
5062 historical_marker_deliveries: &'a [HistoricalMarkerDeliveryAuthority],
5063 total: bool,
5064 ledger: SequenceLedger,
5065}
5066
5067impl BindingOriginValidation<'_> {
5068 fn validate(&self) -> Result<(), ClaimFrontierError> {
5069 if !self.total {
5070 return if self.origins.is_empty() {
5071 Ok(())
5072 } else {
5073 Err(self.logical_owner_error())
5074 };
5075 }
5076 if self.origins.len() != self.active.participants.len() {
5077 return Err(self.logical_owner_error());
5078 }
5079 for participant in &self.active.participants {
5080 let mut matching = self
5081 .origins
5082 .iter()
5083 .filter(|origin| origin.participant_id() == participant.participant_index);
5084 let Some(origin) = matching.next() else {
5085 return Err(self.logical_owner_error());
5086 };
5087 if matching.next().is_some() {
5088 return Err(self.logical_owner_error());
5089 }
5090 self.validate_origin(*participant, *origin)?;
5091 }
5092 Ok(())
5093 }
5094
5095 fn validate_origin(
5096 &self,
5097 participant: FrontierParticipant,
5098 origin: BindingOrigin,
5099 ) -> Result<(), ClaimFrontierError> {
5100 let current_epoch = binding_epoch(participant.binding);
5101 let attached = origin.attached();
5102 if origin.conversation_id() != self.conversation_id
5103 || origin.binding_epoch() != current_epoch
5104 || attached.conversation_id() != self.conversation_id
5105 || attached.participant_id() != participant.participant_index
5106 || attached.binding_epoch() != current_epoch
5107 || attached.admission_order().candidate_phase() != CandidatePhase::AttachLifecycle
5108 {
5109 return Err(self.logical_owner_error());
5110 }
5111 let mut retained_attach_for_binding = self.retained_records.iter().filter(|record| {
5112 matches!(
5113 record.kind,
5114 RetainedCausalRecordKind::AttachLifecycle {
5115 participant_index,
5116 binding_epoch,
5117 } if participant_index == participant.participant_index
5118 && binding_epoch == current_epoch
5119 )
5120 });
5121 let retained_attach_matches = retained_attach_for_binding.clone().any(|record| {
5122 record.delivery_seq == attached.delivery_seq()
5123 && record.admission_order == attached.admission_order()
5124 });
5125 if retained_attach_for_binding.next().is_some() && !retained_attach_matches {
5126 return Err(self.logical_owner_error());
5127 }
5128 if let Some((marker_delivery_seq, prior_binding_epoch)) = origin.recovered_marker() {
5129 let generation_is_next = prior_binding_epoch
5130 .capability_generation
5131 .get()
5132 .checked_add(1)
5133 == Some(current_epoch.capability_generation.get());
5134 let marker_history_matches = self.historical_marker_deliveries.iter().any(|history| {
5135 history.participant_index == participant.participant_index
5136 && history.marker_delivery_seq == marker_delivery_seq
5137 && history.delivered_binding_epoch == prior_binding_epoch
5138 });
5139 if !generation_is_next || !marker_history_matches {
5140 return Err(sequence_error(
5141 sequence_ordinal(self.ledger, marker_delivery_seq),
5142 ClaimFrontierInvalidReason::RecoveryBlock,
5143 ));
5144 }
5145 } else if matches!(participant.binding, FrontierBinding::Detached(_))
5146 && !binding_terminal_exists(
5147 participant.participant_index,
5148 current_epoch,
5149 self.retained_records,
5150 self.causal_authorities,
5151 )
5152 {
5153 return Err(self.logical_owner_error());
5154 }
5155 Ok(())
5156 }
5157
5158 const fn logical_owner_error(&self) -> ClaimFrontierError {
5159 sequence_error(
5160 self.ledger.required_reserve(),
5161 ClaimFrontierInvalidReason::LogicalOwner,
5162 )
5163 }
5164}
5165
5166fn binding_terminal_exists(
5167 participant_index: ParticipantId,
5168 binding_epoch: BindingEpoch,
5169 retained_records: &[RetainedCausalRecord],
5170 historical_causal_authorities: &[HistoricalCausalAuthority],
5171) -> bool {
5172 retained_records.iter().any(|record| {
5173 matches!(
5174 record.kind,
5175 RetainedCausalRecordKind::BindingTerminal(owner)
5176 if owner.participant_index == participant_index
5177 && owner.binding_epoch == binding_epoch
5178 )
5179 }) || historical_causal_authorities.iter().any(|authority| {
5180 matches!(
5181 authority.kind,
5182 HistoricalCausalKind::BindingTerminal(owner)
5183 if owner.participant_index == participant_index
5184 && owner.binding_epoch == binding_epoch
5185 )
5186 })
5187}
5188
5189fn validate_marker_credit_owners(
5190 candidates: &[ImmutableSequenceCandidate],
5191 marker_records: &[RetainedCausalRecord],
5192 identity_slot_limit: u64,
5193 ledger: SequenceLedger,
5194) -> Result<(), ClaimFrontierError> {
5195 let mut owners = Vec::new();
5196 for record in marker_records {
5197 let RetainedCausalRecordKind::CompactionMarker {
5198 participant_index, ..
5199 } = record.kind
5200 else {
5201 continue;
5202 };
5203 if owners.contains(&participant_index) {
5204 return Err(sequence_error(
5205 ledger.required_reserve(),
5206 ClaimFrontierInvalidReason::LogicalOwner,
5207 ));
5208 }
5209 owners.push(participant_index);
5210 }
5211 for candidate in candidates {
5212 let ImmutableSequenceCandidate::Marker(marker) = candidate else {
5213 continue;
5214 };
5215 let participant_index = marker.admission_order.participant_index();
5216 if owners.contains(&participant_index) {
5217 return Err(sequence_error(
5218 sequence_ordinal(ledger, marker.delivery_seq),
5219 ClaimFrontierInvalidReason::LogicalOwner,
5220 ));
5221 }
5222 owners.push(participant_index);
5223 }
5224 if usize_to_u128(owners.len()) > u128::from(identity_slot_limit) {
5225 return Err(sequence_error(
5226 ledger.required_reserve(),
5227 ClaimFrontierInvalidReason::LogicalOwner,
5228 ));
5229 }
5230 Ok(())
5231}
5232
5233fn validated_historical_authorities(
5234 facts: Vec<HistoricalCausalFactRestore>,
5235 conversation_id: ConversationId,
5236 identity_slot_limit: u64,
5237 ledger: SequenceLedger,
5238 history: &ValidatedConversationHistory,
5239) -> Result<Vec<HistoricalCausalAuthority>, ClaimFrontierError> {
5240 if usize_to_u128(facts.len()) > u128::from(identity_slot_limit).saturating_mul(2) {
5241 return Err(sequence_error(
5242 ledger.required_reserve(),
5243 ClaimFrontierInvalidReason::LogicalOwner,
5244 ));
5245 }
5246 let authorities: Vec<_> = facts
5247 .into_iter()
5248 .map(HistoricalCausalAuthority::from_restore)
5249 .collect();
5250 let mut seen = Vec::new();
5251 for authority in &authorities {
5252 let (participant_index, phase) = match authority.kind {
5253 HistoricalCausalKind::BindingTerminal(owner) => {
5254 (owner.participant_index, CandidatePhase::BindingTerminal)
5255 }
5256 HistoricalCausalKind::MembershipExit(participant_index) => {
5257 (participant_index, CandidatePhase::MembershipExit)
5258 }
5259 };
5260 if authority.conversation_id != conversation_id
5261 || participant_index >= identity_slot_limit
5262 || authority.admission_order.participant_index() != participant_index
5263 || authority.admission_order.candidate_phase() != phase
5264 || seen.contains(authority)
5265 || !history.causal_authorities.contains(authority)
5266 {
5267 return Err(sequence_error(
5268 ledger.required_reserve(),
5269 ClaimFrontierInvalidReason::LogicalOwner,
5270 ));
5271 }
5272 seen.push(*authority);
5273 }
5274 Ok(authorities)
5275}
5276
5277const fn corrupt_frontier(error: ClaimFrontierError) -> ParticipantStateCorruptReason {
5278 ParticipantStateCorruptReason::ClaimFrontierInvalid {
5279 counter: match error.counter {
5280 ClaimFrontierCounter::DeliverySequence => ClaimCounter::DeliverySeq,
5281 ClaimFrontierCounter::TransactionOrder => ClaimCounter::TransactionOrder,
5282 },
5283 first_bad_position: error.first_bad_position,
5284 }
5285}
5286
5287fn sequence_segments(restore: &SequenceClaimFrontierRestore) -> Vec<NumericSegment<SequenceClass>> {
5288 let mut segments = Vec::new();
5289 for claim in &restore.movable_claims {
5290 segments.push(NumericSegment {
5291 start: u128::from(claim.delivery_seq),
5292 length: 1,
5293 class: Some(match claim.owner {
5294 SequenceDirectOwner::MembershipExit { .. } => SequenceClass::Exit,
5295 SequenceDirectOwner::BindingTerminal(_) => SequenceClass::Terminal,
5296 }),
5297 immutable: false,
5298 });
5299 }
5300 for candidate in &restore.immutable_candidates {
5301 segments.push(NumericSegment {
5302 start: u128::from(candidate.delivery_seq()),
5303 length: 1,
5304 class: Some(sequence_candidate_class(*candidate)),
5305 immutable: true,
5306 });
5307 }
5308 for range in &restore.products.live_times_terminal {
5309 segments.push(NumericSegment {
5310 start: u128::from(range.start),
5311 length: u128::from(range.length),
5312 class: Some(SequenceClass::LiveTimesTerminal),
5313 immutable: false,
5314 });
5315 }
5316 if let Some(range) = restore.products.live_times_replacement_terminal {
5317 segments.push(NumericSegment {
5318 start: u128::from(range.start),
5319 length: u128::from(range.length),
5320 class: Some(SequenceClass::LiveTimesReplacementTerminal),
5321 immutable: false,
5322 });
5323 }
5324 for range in &restore.products.other_live_times_exit {
5325 segments.push(NumericSegment {
5326 start: u128::from(range.start),
5327 length: u128::from(range.length),
5328 class: Some(SequenceClass::OtherLiveTimesExit),
5329 immutable: false,
5330 });
5331 }
5332 if let Some(recovery) = restore.recovery {
5333 if let Some(terminal) = recovery.terminal {
5334 segments.push(NumericSegment {
5335 start: u128::from(terminal.delivery_seq),
5336 length: 1,
5337 class: Some(SequenceClass::Terminal),
5338 immutable: false,
5339 });
5340 }
5341 segments.push(NumericSegment {
5342 start: u128::from(recovery.recovery_attach_seq),
5343 length: 1,
5344 class: Some(SequenceClass::RecoveryAttach),
5345 immutable: false,
5346 });
5347 segments.push(NumericSegment {
5348 start: u128::from(recovery.replacement_terminal_seq),
5349 length: 1,
5350 class: Some(SequenceClass::RecoveryReplacementTerminal),
5351 immutable: false,
5352 });
5353 }
5354 segments
5355}
5356
5357fn order_segments(
5358 restore: &OrderClaimFrontierRestore,
5359 high: OrderHigh,
5360) -> Vec<NumericSegment<OrderClass>> {
5361 let mut segments = Vec::new();
5362 for claim in &restore.movable_claims {
5363 segments.push(NumericSegment {
5364 start: u128::from(claim.transaction_order),
5365 length: 1,
5366 class: Some(match claim.owner {
5367 OrderDirectOwner::ActiveBindingTerminal(_) => OrderClass::ActiveBindingTerminal,
5368 OrderDirectOwner::MembershipExit { .. } => OrderClass::MembershipExit,
5369 }),
5370 immutable: false,
5371 });
5372 }
5373 for candidate in restore
5374 .immutable_candidates
5375 .iter()
5376 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
5377 {
5378 segments.push(NumericSegment {
5379 start: u128::from(candidate.transaction_order),
5380 length: 1,
5381 class: None,
5382 immutable: true,
5383 });
5384 }
5385 if let Some(recovery) = restore.recovery {
5386 if let Some(active_binding) = recovery.active_binding {
5387 segments.push(NumericSegment {
5388 start: u128::from(active_binding.transaction_order),
5389 length: 1,
5390 class: Some(OrderClass::ActiveBindingTerminal),
5391 immutable: false,
5392 });
5393 }
5394 segments.push(NumericSegment {
5395 start: u128::from(recovery.recovery_operation_order),
5396 length: 1,
5397 class: Some(OrderClass::RecoveryOperation),
5398 immutable: false,
5399 });
5400 segments.push(NumericSegment {
5401 start: u128::from(recovery.replacement_terminal_order),
5402 length: 1,
5403 class: Some(OrderClass::RecoveryReplacementTerminal),
5404 immutable: false,
5405 });
5406 }
5407 segments
5408}
5409
5410fn restore_sequence_frontier(
5411 active: &ActiveIdentityRanks,
5412 restore: SequenceClaimFrontierRestore,
5413 retained_floor: u128,
5414 ledger: SequenceLedger,
5415 recovery_provenance: Option<RecoveryClaimProvenance>,
5416 retained_records: &[RetainedCausalRecord],
5417 historical_causal_authorities: &[HistoricalCausalAuthority],
5418) -> Result<SequenceClaimFrontier, ClaimFrontierError> {
5419 let mut segments = sequence_segments(&restore);
5420
5421 let expected_counts = sequence_expected_counts(ledger);
5422 validate_numeric_segments(
5423 ClaimFrontierCounter::DeliverySequence,
5424 u128::from(ledger.high_watermark()) + 1,
5425 ledger.required_reserve(),
5426 &mut segments,
5427 &expected_counts,
5428 )?;
5429 validate_sequence_recovery(
5430 active,
5431 restore.recovery,
5432 recovery_provenance,
5433 ledger,
5434 ledger.required_reserve(),
5435 )?;
5436 validate_sequence_candidates(
5437 active,
5438 &restore.immutable_candidates,
5439 retained_floor,
5440 retained_records,
5441 historical_causal_authorities,
5442 ledger,
5443 )?;
5444 let terminal_owners = validate_sequence_direct_owners(
5445 active,
5446 &restore.movable_claims,
5447 &restore.immutable_candidates,
5448 restore.recovery,
5449 ledger,
5450 )?;
5451 let products = validate_sequence_products(
5452 active,
5453 restore.products,
5454 &terminal_owners,
5455 restore.recovery,
5456 recovery_provenance,
5457 ledger,
5458 )?;
5459 let recovery = restore
5460 .recovery
5461 .zip(recovery_provenance)
5462 .map(|(value, provenance)| RecoverySequenceBlock {
5463 terminal: value.terminal,
5464 recovery_attach_seq: value.recovery_attach_seq,
5465 replacement_terminal_seq: value.replacement_terminal_seq,
5466 participant_index: provenance.participant_index,
5467 marker_delivery_seq: provenance.marker_delivery_seq,
5468 recovered_binding_epoch: provenance.prior_binding_epoch,
5469 });
5470
5471 let mut movable_claims = restore.movable_claims;
5472 movable_claims.sort_by_key(|claim| claim.delivery_seq);
5473 let mut immutable_candidates = restore.immutable_candidates;
5474 immutable_candidates.sort_by_key(|candidate| candidate.delivery_seq());
5475
5476 Ok(SequenceClaimFrontier {
5477 ledger,
5478 movable_claims,
5479 immutable_candidates,
5480 products,
5481 recovery,
5482 })
5483}
5484
5485fn restore_order_frontier(
5486 active: &ActiveIdentityRanks,
5487 restore: OrderClaimFrontierRestore,
5488 ledger: OrderLedger,
5489 recovery_provenance: Option<RecoveryClaimProvenance>,
5490) -> Result<OrderClaimFrontier, ClaimFrontierError> {
5491 let mut segments = order_segments(&restore, ledger.high());
5492
5493 let candidate_count = order_frontier_candidate_count(&restore, ledger.high());
5494 let expected_length = candidate_count + ledger.claims().total();
5495 let expected_counts = order_expected_counts(ledger);
5496 validate_numeric_segments(
5497 ClaimFrontierCounter::TransactionOrder,
5498 order_frontier_start(ledger.high()),
5499 expected_length,
5500 &mut segments,
5501 &expected_counts,
5502 )?;
5503 validate_order_recovery(
5504 active,
5505 restore.recovery,
5506 recovery_provenance,
5507 ledger,
5508 expected_length,
5509 )?;
5510 let immutable_candidates = validate_order_candidates(&restore.immutable_candidates, ledger)?;
5511 validate_order_direct_owners(active, &restore.movable_claims, restore.recovery, ledger)?;
5512 let recovery = restore
5513 .recovery
5514 .zip(recovery_provenance)
5515 .map(|(value, provenance)| RecoveryOrderBlock {
5516 active_binding: value.active_binding,
5517 recovery_operation_order: value.recovery_operation_order,
5518 replacement_terminal_order: value.replacement_terminal_order,
5519 participant_index: provenance.participant_index,
5520 marker_delivery_seq: provenance.marker_delivery_seq,
5521 recovered_binding_epoch: provenance.prior_binding_epoch,
5522 });
5523
5524 let mut movable_claims = restore.movable_claims;
5525 movable_claims.sort_by_key(|claim| claim.transaction_order);
5526
5527 Ok(OrderClaimFrontier {
5528 ledger,
5529 movable_claims,
5530 immutable_candidates,
5531 recovery,
5532 })
5533}
5534
5535fn validate_numeric_segments<C: Copy + Into<usize>>(
5536 counter: ClaimFrontierCounter,
5537 first_value: u128,
5538 expected_length: u128,
5539 segments: &mut [NumericSegment<C>],
5540 expected_counts: &[u128],
5541) -> Result<(), ClaimFrontierError> {
5542 segments.sort_by_key(|segment| segment.start);
5543 let mut events = numeric_events(counter, first_value, segments)?;
5544 let emitted = scan_numeric_events(counter, first_value, &mut events)?;
5545 if emitted != expected_length {
5546 return Err(frontier_error(
5547 counter,
5548 emitted.min(expected_length),
5549 ClaimFrontierInvalidReason::AggregateLedger,
5550 ));
5551 }
5552 validate_immutable_prefix(counter, first_value, segments)?;
5553 validate_segment_class_counts(counter, expected_length, segments, expected_counts)
5554}
5555
5556fn numeric_events<C>(
5557 counter: ClaimFrontierCounter,
5558 first_value: u128,
5559 segments: &[NumericSegment<C>],
5560) -> Result<Vec<(u128, i8)>, ClaimFrontierError> {
5561 let mut events = Vec::new();
5562 let counter_limit = u128::from(u64::MAX) + 1;
5563 for segment in segments {
5564 if segment.length == 0 {
5565 continue;
5566 }
5567 let Some(end) = segment.start.checked_add(segment.length) else {
5568 return Err(frontier_error(
5569 counter,
5570 counter_limit.saturating_sub(first_value),
5571 ClaimFrontierInvalidReason::NumericPosition,
5572 ));
5573 };
5574 if segment.start < first_value {
5575 return Err(frontier_error(
5576 counter,
5577 0,
5578 ClaimFrontierInvalidReason::NumericPosition,
5579 ));
5580 }
5581 if end > counter_limit {
5582 return Err(frontier_error(
5583 counter,
5584 counter_limit.saturating_sub(first_value),
5585 ClaimFrontierInvalidReason::NumericPosition,
5586 ));
5587 }
5588 events.push((segment.start, 1_i8));
5589 events.push((end, -1_i8));
5590 }
5591 Ok(events)
5592}
5593
5594fn scan_numeric_events(
5595 counter: ClaimFrontierCounter,
5596 first_value: u128,
5597 events: &mut [(u128, i8)],
5598) -> Result<u128, ClaimFrontierError> {
5599 events.sort_unstable_by_key(|event| event.0);
5600 let mut event_index = 0_usize;
5601 let mut coordinate = first_value;
5602 let mut coverage = 0_i128;
5603 let mut emitted = 0_u128;
5604 while let Some((event_coordinate, _)) = events.get(event_index).copied() {
5605 if event_coordinate > coordinate {
5606 if coverage == 0 {
5607 return Err(frontier_error(
5608 counter,
5609 emitted,
5610 ClaimFrontierInvalidReason::NumericPosition,
5611 ));
5612 }
5613 if coverage > 1 {
5614 return Err(frontier_error(
5615 counter,
5616 emitted.saturating_add(1),
5617 ClaimFrontierInvalidReason::NumericPosition,
5618 ));
5619 }
5620 emitted = emitted.saturating_add(event_coordinate - coordinate);
5621 coordinate = event_coordinate;
5622 }
5623 while let Some((same_coordinate, delta)) = events.get(event_index).copied() {
5624 if same_coordinate != coordinate {
5625 break;
5626 }
5627 coverage += i128::from(delta);
5628 event_index += 1;
5629 }
5630 }
5631 if coverage != 0 {
5632 return Err(frontier_error(
5633 counter,
5634 emitted,
5635 ClaimFrontierInvalidReason::NumericPosition,
5636 ));
5637 }
5638 Ok(emitted)
5639}
5640
5641fn validate_immutable_prefix<C>(
5642 counter: ClaimFrontierCounter,
5643 first_value: u128,
5644 segments: &[NumericSegment<C>],
5645) -> Result<(), ClaimFrontierError> {
5646 let mut first_movable = None;
5647 for segment in segments.iter().filter(|segment| segment.length != 0) {
5648 if segment.immutable {
5649 if let Some(first_movable) = first_movable {
5650 return Err(frontier_error(
5651 counter,
5652 first_movable,
5653 ClaimFrontierInvalidReason::NumericPosition,
5654 ));
5655 }
5656 } else if first_movable.is_none() {
5657 first_movable = Some(segment.start - first_value);
5658 }
5659 }
5660 Ok(())
5661}
5662
5663fn validate_segment_class_counts<C: Copy + Into<usize>>(
5664 counter: ClaimFrontierCounter,
5665 expected_length: u128,
5666 segments: &[NumericSegment<C>],
5667 expected_counts: &[u128],
5668) -> Result<(), ClaimFrontierError> {
5669 let mut actual_counts = core::iter::repeat_n(0_u128, expected_counts.len()).collect::<Vec<_>>();
5670 let mut class_ordinal = 0_u128;
5671 for segment in segments {
5672 if segment.length == 0 {
5673 continue;
5674 }
5675 if let Some(class) = segment.class {
5676 let index = class.into();
5677 let prior = actual_counts[index];
5678 let Some(resulting) = prior.checked_add(segment.length) else {
5679 return Err(frontier_error(
5680 counter,
5681 class_ordinal,
5682 ClaimFrontierInvalidReason::AggregateLedger,
5683 ));
5684 };
5685 if resulting > expected_counts[index] {
5686 return Err(frontier_error(
5687 counter,
5688 class_ordinal + expected_counts[index].saturating_sub(prior),
5689 ClaimFrontierInvalidReason::AggregateLedger,
5690 ));
5691 }
5692 actual_counts[index] = resulting;
5693 }
5694 class_ordinal += segment.length;
5695 }
5696 if actual_counts != expected_counts {
5697 return Err(frontier_error(
5698 counter,
5699 expected_length,
5700 ClaimFrontierInvalidReason::AggregateLedger,
5701 ));
5702 }
5703 Ok(())
5704}
5705
5706#[cfg(test)]
5707pub(super) fn validate_numeric_union_for_test(
5708 first_value: u128,
5709 expected_length: u128,
5710 ranges: &[(u128, u128)],
5711) -> Result<(), ClaimFrontierError> {
5712 let mut segments: Vec<_> = ranges
5713 .iter()
5714 .map(|(start, length)| NumericSegment {
5715 start: *start,
5716 length: *length,
5717 class: Some(SequenceClass::Exit),
5718 immutable: false,
5719 })
5720 .collect();
5721 validate_numeric_segments(
5722 ClaimFrontierCounter::DeliverySequence,
5723 first_value,
5724 expected_length,
5725 &mut segments,
5726 &[expected_length, 0, 0, 0, 0, 0, 0, 0],
5727 )
5728}
5729
5730impl From<SequenceClass> for usize {
5731 fn from(value: SequenceClass) -> Self {
5732 value as Self
5733 }
5734}
5735
5736impl From<OrderClass> for usize {
5737 fn from(value: OrderClass) -> Self {
5738 value as Self
5739 }
5740}
5741
5742const fn sequence_candidate_class(candidate: ImmutableSequenceCandidate) -> SequenceClass {
5743 match candidate {
5744 ImmutableSequenceCandidate::BindingTerminal { .. } => SequenceClass::Terminal,
5745 ImmutableSequenceCandidate::Marker(marker) => match marker.current_owner {
5746 MarkerSequenceOwner::Marker => SequenceClass::Marker,
5747 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::LiveTimesTerminal) => {
5748 SequenceClass::LiveTimesTerminal
5749 }
5750 MarkerSequenceOwner::ConditionalProduct(
5751 SequenceProductClass::LiveTimesReplacementTerminal,
5752 ) => SequenceClass::LiveTimesReplacementTerminal,
5753 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::OtherLiveTimesExit) => {
5754 SequenceClass::OtherLiveTimesExit
5755 }
5756 },
5757 }
5758}
5759
5760fn sequence_expected_counts(ledger: SequenceLedger) -> [u128; 8] {
5761 let budget = ledger.budget();
5762 [
5763 u128::from(budget.e),
5764 u128::from(budget.t),
5765 u128::from(budget.m),
5766 u128::from(budget.rs),
5767 u128::from(budget.rt),
5768 budget.l_times_t,
5769 budget.l_times_rt,
5770 budget.l_other_times_e,
5771 ]
5772}
5773
5774fn order_expected_counts(ledger: OrderLedger) -> [u128; 4] {
5775 let claims = ledger.claims();
5776 [
5777 u128::from(claims.active_binding_terminals()),
5778 u128::from(claims.membership_exits()),
5779 u128::from(claims.recovery_operation()),
5780 u128::from(claims.recovery_replacement_terminal()),
5781 ]
5782}
5783
5784fn validate_sequence_recovery(
5785 active: &ActiveIdentityRanks,
5786 recovery: Option<RecoverySequenceBlockRestore>,
5787 provenance: Option<RecoveryClaimProvenance>,
5788 ledger: SequenceLedger,
5789 frontier_length: u128,
5790) -> Result<(), ClaimFrontierError> {
5791 let expected = ledger.claims().recovery();
5792 match (expected, recovery, provenance) {
5793 (RecoverySequenceReserve::None, None, None) => Ok(()),
5794 (RecoverySequenceReserve::DetachedCredentialRecovery, None, _)
5795 | (RecoverySequenceReserve::DetachedCredentialRecovery, Some(_), None)
5796 | (RecoverySequenceReserve::None, None, Some(_)) => Err(sequence_error(
5797 frontier_length,
5798 ClaimFrontierInvalidReason::RecoveryBlock,
5799 )),
5800 (RecoverySequenceReserve::None, Some(block), _) => Err(sequence_error(
5801 sequence_ordinal(ledger, block_start_sequence(block)),
5802 ClaimFrontierInvalidReason::RecoveryBlock,
5803 )),
5804 (RecoverySequenceReserve::DetachedCredentialRecovery, Some(block), Some(provenance)) => {
5805 let block_ordinal = sequence_ordinal(ledger, block_start_sequence(block));
5806 let expected_recovery_attach = block
5807 .terminal
5808 .map_or(Some(block.recovery_attach_seq), |terminal| {
5809 terminal.delivery_seq.checked_add(1)
5810 });
5811 if expected_recovery_attach != Some(block.recovery_attach_seq) {
5812 return Err(sequence_error(
5813 block_ordinal + 1,
5814 ClaimFrontierInvalidReason::RecoveryBlock,
5815 ));
5816 }
5817 if block.recovery_attach_seq.checked_add(1) != Some(block.replacement_terminal_seq) {
5818 return Err(sequence_error(
5819 block_ordinal + u128::from(block.terminal.is_some()) + 1,
5820 ClaimFrontierInvalidReason::RecoveryBlock,
5821 ));
5822 }
5823 let Some(participant) = active_participant(active, provenance.participant_index) else {
5824 return Err(sequence_error(
5825 block_ordinal,
5826 ClaimFrontierInvalidReason::LogicalOwner,
5827 ));
5828 };
5829 let expected_binding = match provenance.phase {
5830 RecoveryClaimPhase::PreFate => {
5831 FrontierBinding::Bound(provenance.prior_binding_epoch)
5832 }
5833 RecoveryClaimPhase::PostFate => {
5834 FrontierBinding::Detached(provenance.prior_binding_epoch)
5835 }
5836 RecoveryClaimPhase::RecoveredBound => {
5837 FrontierBinding::Bound(provenance.current_binding_epoch)
5838 }
5839 };
5840 if participant.binding != expected_binding {
5841 return Err(sequence_error(
5842 block_ordinal,
5843 ClaimFrontierInvalidReason::LogicalOwner,
5844 ));
5845 }
5846 let terminal_valid = match (provenance.phase, block.terminal) {
5847 (RecoveryClaimPhase::PreFate, Some(terminal)) => {
5848 terminal.owner.participant_index == provenance.participant_index
5849 && terminal.owner.binding_epoch == provenance.prior_binding_epoch
5850 }
5851 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
5852 _ => false,
5853 };
5854 if !terminal_valid {
5855 return Err(sequence_error(
5856 block_ordinal,
5857 ClaimFrontierInvalidReason::RecoveryBlock,
5858 ));
5859 }
5860 Ok(())
5861 }
5862 }
5863}
5864
5865fn validate_sequence_candidates(
5866 active: &ActiveIdentityRanks,
5867 candidates: &[ImmutableSequenceCandidate],
5868 retained_floor: u128,
5869 retained_records: &[RetainedCausalRecord],
5870 historical_causal_authorities: &[HistoricalCausalAuthority],
5871 ledger: SequenceLedger,
5872) -> Result<(), ClaimFrontierError> {
5873 let mut seen_keys = Vec::new();
5874 let mut previous_sequence = None;
5875 let mut previous_order = retained_records.last().map(|record| record.admission_order);
5876 for candidate in candidates {
5877 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
5878 let order = candidate.admission_order();
5879 if previous_sequence.is_some_and(|previous| previous >= candidate.delivery_seq())
5880 || previous_order.is_some_and(|previous| previous >= order)
5881 || seen_keys.contains(&order)
5882 {
5883 return Err(sequence_error(
5884 ordinal,
5885 ClaimFrontierInvalidReason::CandidateKey,
5886 ));
5887 }
5888 previous_sequence = Some(candidate.delivery_seq());
5889 previous_order = Some(order);
5890 seen_keys.push(order);
5891 match candidate {
5892 ImmutableSequenceCandidate::BindingTerminal { owner, .. } => {
5893 if order.candidate_phase() != CandidatePhase::BindingTerminal
5894 || order.participant_index() != owner.participant_index
5895 || !terminal_matches_active(active, *owner)
5896 {
5897 return Err(sequence_error(
5898 ordinal,
5899 ClaimFrontierInvalidReason::CandidateKey,
5900 ));
5901 }
5902 }
5903 ImmutableSequenceCandidate::Marker(marker) => {
5904 let Some(participant) = active_participant(active, order.participant_index())
5905 else {
5906 return Err(sequence_error(
5907 ordinal,
5908 ClaimFrontierInvalidReason::LogicalOwner,
5909 ));
5910 };
5911 if order.candidate_phase() != CandidatePhase::CompactionMarker
5912 || marker.current_owner != MarkerSequenceOwner::Marker
5913 || marker.target_binding != participant.binding
5914 || marker.abandoned_after != participant.cursor
5915 || marker.abandoned_after > marker.abandoned_through
5916 || u128::from(marker.physical_floor_at_decision) != retained_floor
5917 || u128::from(marker.physical_floor_at_decision)
5918 > u128::from(marker.abandoned_through) + 1
5919 || marker.abandoned_through >= marker.delivery_seq
5920 || !marker_provenance_targets(marker.provenance, order.participant_index())
5921 || !marker_has_causal_authority(
5922 *marker,
5923 retained_records,
5924 historical_causal_authorities,
5925 )
5926 {
5927 return Err(sequence_error(
5928 ordinal,
5929 ClaimFrontierInvalidReason::CandidateKey,
5930 ));
5931 }
5932 }
5933 }
5934 }
5935 Ok(())
5936}
5937
5938#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5939enum TerminalOccurrenceKind {
5940 Movable,
5941 Candidate,
5942}
5943
5944#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5945struct TerminalOccurrence {
5946 owner: BindingTerminalOwner,
5947 ordinal: u128,
5948 kind: TerminalOccurrenceKind,
5949}
5950
5951fn validate_sequence_direct_owners(
5952 active: &ActiveIdentityRanks,
5953 movable: &[MovableSequenceClaim],
5954 candidates: &[ImmutableSequenceCandidate],
5955 recovery: Option<RecoverySequenceBlockRestore>,
5956 ledger: SequenceLedger,
5957) -> Result<Vec<BindingTerminalOwner>, ClaimFrontierError> {
5958 let (mut exit_owners, terminal_occurrences) =
5959 collect_sequence_direct_owners(active, movable, candidates, recovery, ledger)?;
5960 validate_sequence_exit_owners(active, &mut exit_owners, ledger)?;
5961 validate_sequence_terminal_owners(active, &terminal_occurrences, ledger)?;
5962 let mut owners: Vec<_> = terminal_occurrences
5963 .into_iter()
5964 .map(|occurrence| occurrence.owner)
5965 .collect();
5966 owners.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
5967 Ok(owners)
5968}
5969
5970fn collect_sequence_direct_owners(
5971 active: &ActiveIdentityRanks,
5972 movable: &[MovableSequenceClaim],
5973 candidates: &[ImmutableSequenceCandidate],
5974 recovery: Option<RecoverySequenceBlockRestore>,
5975 ledger: SequenceLedger,
5976) -> Result<(Vec<ParticipantId>, Vec<TerminalOccurrence>), ClaimFrontierError> {
5977 let mut ordered_movable = movable.to_vec();
5978 ordered_movable.sort_by_key(|claim| claim.delivery_seq);
5979 let mut exit_owners = Vec::new();
5980 let mut terminal_occurrences = Vec::new();
5981 for claim in ordered_movable {
5982 let ordinal = sequence_ordinal(ledger, claim.delivery_seq);
5983 match claim.owner {
5984 SequenceDirectOwner::MembershipExit { participant_index } => {
5985 if !active.contains(participant_index) || exit_owners.contains(&participant_index) {
5986 return Err(sequence_error(
5987 ordinal,
5988 ClaimFrontierInvalidReason::LogicalOwner,
5989 ));
5990 }
5991 exit_owners.push(participant_index);
5992 }
5993 SequenceDirectOwner::BindingTerminal(owner) => {
5994 if !terminal_matches_bound(active, owner) {
5995 return Err(sequence_error(
5996 ordinal,
5997 ClaimFrontierInvalidReason::LogicalOwner,
5998 ));
5999 }
6000 push_terminal_occurrence(
6001 &mut terminal_occurrences,
6002 owner,
6003 ordinal,
6004 TerminalOccurrenceKind::Movable,
6005 )?;
6006 }
6007 }
6008 }
6009 for candidate in candidates {
6010 if let ImmutableSequenceCandidate::BindingTerminal { owner, .. } = candidate {
6011 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
6012 push_terminal_occurrence(
6013 &mut terminal_occurrences,
6014 *owner,
6015 ordinal,
6016 TerminalOccurrenceKind::Candidate,
6017 )?;
6018 }
6019 }
6020 if let Some(terminal) = recovery.and_then(|block| block.terminal) {
6021 let ordinal = sequence_ordinal(ledger, terminal.delivery_seq);
6022 push_terminal_occurrence(
6023 &mut terminal_occurrences,
6024 terminal.owner,
6025 ordinal,
6026 TerminalOccurrenceKind::Movable,
6027 )?;
6028 }
6029 Ok((exit_owners, terminal_occurrences))
6030}
6031
6032fn push_terminal_occurrence(
6033 occurrences: &mut Vec<TerminalOccurrence>,
6034 owner: BindingTerminalOwner,
6035 ordinal: u128,
6036 kind: TerminalOccurrenceKind,
6037) -> Result<(), ClaimFrontierError> {
6038 if occurrences
6039 .iter()
6040 .any(|occurrence| occurrence.owner == owner)
6041 {
6042 return Err(sequence_error(
6043 ordinal,
6044 ClaimFrontierInvalidReason::LogicalOwner,
6045 ));
6046 }
6047 occurrences.push(TerminalOccurrence {
6048 owner,
6049 ordinal,
6050 kind,
6051 });
6052 Ok(())
6053}
6054
6055fn validate_sequence_exit_owners(
6056 active: &ActiveIdentityRanks,
6057 exit_owners: &mut [ParticipantId],
6058 ledger: SequenceLedger,
6059) -> Result<(), ClaimFrontierError> {
6060 exit_owners.sort_unstable();
6061 if exit_owners.len() != active.participants.len()
6062 || !exit_owners.iter().copied().eq(active
6063 .participants
6064 .iter()
6065 .map(|participant| participant.participant_index))
6066 {
6067 return Err(sequence_error(
6068 ledger.required_reserve(),
6069 ClaimFrontierInvalidReason::LogicalOwner,
6070 ));
6071 }
6072 Ok(())
6073}
6074
6075fn validate_sequence_terminal_owners(
6076 active: &ActiveIdentityRanks,
6077 terminal_occurrences: &[TerminalOccurrence],
6078 ledger: SequenceLedger,
6079) -> Result<(), ClaimFrontierError> {
6080 for participant in &active.participants {
6081 let matching: Vec<_> = terminal_occurrences
6082 .iter()
6083 .filter(|occurrence| {
6084 occurrence.owner.participant_index == participant.participant_index
6085 })
6086 .copied()
6087 .collect();
6088 match participant.binding {
6089 FrontierBinding::Bound(epoch) => {
6090 if !matches!(matching.as_slice(), [occurrence] if occurrence.owner.binding_epoch == epoch)
6091 {
6092 return Err(sequence_error(
6093 matching.first().map_or_else(
6094 || ledger.required_reserve(),
6095 |occurrence| occurrence.ordinal,
6096 ),
6097 ClaimFrontierInvalidReason::LogicalOwner,
6098 ));
6099 }
6100 }
6101 FrontierBinding::Detached(epoch) => {
6102 if matching.len() > 1
6103 || matching.first().is_some_and(|occurrence| {
6104 occurrence.owner.binding_epoch != epoch
6105 || occurrence.kind != TerminalOccurrenceKind::Candidate
6106 })
6107 {
6108 return Err(sequence_error(
6109 matching.first().map_or_else(
6110 || ledger.required_reserve(),
6111 |occurrence| occurrence.ordinal,
6112 ),
6113 ClaimFrontierInvalidReason::LogicalOwner,
6114 ));
6115 }
6116 }
6117 }
6118 }
6119 Ok(())
6120}
6121
6122fn validate_sequence_products(
6123 active: &ActiveIdentityRanks,
6124 restore: SequenceProductRangesRestore,
6125 terminal_owners: &[BindingTerminalOwner],
6126 recovery: Option<RecoverySequenceBlockRestore>,
6127 recovery_provenance: Option<RecoveryClaimProvenance>,
6128 ledger: SequenceLedger,
6129) -> Result<SequenceProductRanges, ClaimFrontierError> {
6130 let live_count = usize_to_u64(active.participants.len());
6131 let other_count = live_count.saturating_sub(1);
6132 let live_times_terminal = validate_terminal_product_ranges(
6133 restore.live_times_terminal,
6134 terminal_owners,
6135 live_count,
6136 ledger,
6137 )?;
6138 let live_times_replacement_terminal = validate_replacement_product_range(
6139 restore.live_times_replacement_terminal,
6140 recovery,
6141 recovery_provenance,
6142 live_count,
6143 ledger,
6144 )?;
6145 let other_live_times_exit =
6146 validate_exit_product_ranges(active, restore.other_live_times_exit, other_count, ledger)?;
6147 Ok(SequenceProductRanges {
6148 live_times_terminal,
6149 live_times_replacement_terminal,
6150 other_live_times_exit,
6151 })
6152}
6153
6154fn validate_terminal_product_ranges(
6155 mut ranges: Vec<TerminalProductRangeRestore>,
6156 terminal_owners: &[BindingTerminalOwner],
6157 live_count: u64,
6158 ledger: SequenceLedger,
6159) -> Result<Vec<TerminalProductRange>, ClaimFrontierError> {
6160 ranges.sort_by_key(|range| range.start);
6161 let mut seen_terminals = Vec::new();
6162 let mut live_times_terminal = Vec::new();
6163 for range in ranges {
6164 let ordinal = sequence_ordinal(ledger, range.start);
6165 if range.length != live_count
6166 || !terminal_owners.contains(&range.terminal)
6167 || seen_terminals.contains(&range.terminal)
6168 {
6169 return Err(sequence_error(
6170 ordinal,
6171 ClaimFrontierInvalidReason::ProductRange,
6172 ));
6173 }
6174 seen_terminals.push(range.terminal);
6175 live_times_terminal.push(TerminalProductRange {
6176 start: range.start,
6177 length: range.length,
6178 terminal: range.terminal,
6179 });
6180 }
6181 if seen_terminals.len() != terminal_owners.len() {
6182 return Err(sequence_error(
6183 ledger.required_reserve(),
6184 ClaimFrontierInvalidReason::ProductRange,
6185 ));
6186 }
6187 Ok(live_times_terminal)
6188}
6189
6190fn validate_replacement_product_range(
6191 range: Option<ReplacementTerminalProductRangeRestore>,
6192 recovery: Option<RecoverySequenceBlockRestore>,
6193 recovery_provenance: Option<RecoveryClaimProvenance>,
6194 live_count: u64,
6195 ledger: SequenceLedger,
6196) -> Result<Option<ReplacementTerminalProductRange>, ClaimFrontierError> {
6197 let validated = match (range, recovery, recovery_provenance) {
6198 (None, None, None) => None,
6199 (Some(range), Some(_), Some(provenance)) if range.length == live_count => {
6200 Some(ReplacementTerminalProductRange {
6201 start: range.start,
6202 length: range.length,
6203 participant_index: provenance.participant_index,
6204 marker_delivery_seq: provenance.marker_delivery_seq,
6205 prior_binding_epoch: provenance.prior_binding_epoch,
6206 })
6207 }
6208 (Some(range), _, _) => {
6209 return Err(sequence_error(
6210 sequence_ordinal(ledger, range.start),
6211 ClaimFrontierInvalidReason::ProductRange,
6212 ));
6213 }
6214 (None, Some(_), _) | (None, None, Some(_)) => {
6215 return Err(sequence_error(
6216 ledger.required_reserve(),
6217 ClaimFrontierInvalidReason::ProductRange,
6218 ));
6219 }
6220 };
6221 Ok(validated)
6222}
6223
6224fn validate_exit_product_ranges(
6225 active: &ActiveIdentityRanks,
6226 mut ranges: Vec<ExitProductRangeRestore>,
6227 other_count: u64,
6228 ledger: SequenceLedger,
6229) -> Result<Vec<ExitProductRange>, ClaimFrontierError> {
6230 ranges.sort_by_key(|range| range.start);
6231 let mut seen_exits = Vec::new();
6232 let mut other_live_times_exit = Vec::new();
6233 if other_count == 0 && !ranges.is_empty() {
6234 return Err(sequence_error(
6235 ledger.required_reserve(),
6236 ClaimFrontierInvalidReason::ProductRange,
6237 ));
6238 }
6239 for range in ranges {
6240 let ordinal = sequence_ordinal(ledger, range.start);
6241 if range.length != other_count
6242 || !active.contains(range.exit_participant)
6243 || seen_exits.contains(&range.exit_participant)
6244 {
6245 return Err(sequence_error(
6246 ordinal,
6247 ClaimFrontierInvalidReason::ProductRange,
6248 ));
6249 }
6250 seen_exits.push(range.exit_participant);
6251 other_live_times_exit.push(ExitProductRange {
6252 start: range.start,
6253 length: range.length,
6254 exit_participant: range.exit_participant,
6255 });
6256 }
6257 seen_exits.sort_unstable();
6258 let expected_exit_ranges = if other_count == 0 {
6259 0
6260 } else {
6261 active.participants.len()
6262 };
6263 if seen_exits.len() != expected_exit_ranges
6264 || !seen_exits.iter().copied().eq(active
6265 .participants
6266 .iter()
6267 .take(expected_exit_ranges)
6268 .map(|participant| participant.participant_index))
6269 {
6270 return Err(sequence_error(
6271 ledger.required_reserve(),
6272 ClaimFrontierInvalidReason::ProductRange,
6273 ));
6274 }
6275 Ok(other_live_times_exit)
6276}
6277
6278fn validate_order_recovery(
6279 active: &ActiveIdentityRanks,
6280 recovery: Option<RecoveryOrderBlockRestore>,
6281 provenance: Option<RecoveryClaimProvenance>,
6282 ledger: OrderLedger,
6283 frontier_length: u128,
6284) -> Result<(), ClaimFrontierError> {
6285 let claims = ledger.claims();
6286 let expected = claims.recovery_operation() && claims.recovery_replacement_terminal();
6287 match (expected, recovery, provenance) {
6288 (false, None, None) => Ok(()),
6289 (true, None, _) | (true, Some(_), None) | (false, None, Some(_)) => Err(order_error(
6290 frontier_length,
6291 ClaimFrontierInvalidReason::RecoveryBlock,
6292 )),
6293 (false, Some(block), _) => Err(order_error(
6294 order_ordinal(ledger, block_start_order(block)),
6295 ClaimFrontierInvalidReason::RecoveryBlock,
6296 )),
6297 (true, Some(block), Some(provenance)) => {
6298 let block_ordinal = order_ordinal(ledger, block_start_order(block));
6299 let expected_recovery_operation = block
6300 .active_binding
6301 .map_or(Some(block.recovery_operation_order), |active_binding| {
6302 active_binding.transaction_order.checked_add(1)
6303 });
6304 if expected_recovery_operation != Some(block.recovery_operation_order) {
6305 return Err(order_error(
6306 block_ordinal + 1,
6307 ClaimFrontierInvalidReason::RecoveryBlock,
6308 ));
6309 }
6310 if block.recovery_operation_order.checked_add(1)
6311 != Some(block.replacement_terminal_order)
6312 {
6313 return Err(order_error(
6314 block_ordinal + u128::from(block.active_binding.is_some()) + 1,
6315 ClaimFrontierInvalidReason::RecoveryBlock,
6316 ));
6317 }
6318 let Some(participant) = active_participant(active, provenance.participant_index) else {
6319 return Err(order_error(
6320 block_ordinal,
6321 ClaimFrontierInvalidReason::LogicalOwner,
6322 ));
6323 };
6324 let expected_binding = match provenance.phase {
6325 RecoveryClaimPhase::PreFate => {
6326 FrontierBinding::Bound(provenance.prior_binding_epoch)
6327 }
6328 RecoveryClaimPhase::PostFate => {
6329 FrontierBinding::Detached(provenance.prior_binding_epoch)
6330 }
6331 RecoveryClaimPhase::RecoveredBound => {
6332 FrontierBinding::Bound(provenance.current_binding_epoch)
6333 }
6334 };
6335 if participant.binding != expected_binding {
6336 return Err(order_error(
6337 block_ordinal,
6338 ClaimFrontierInvalidReason::LogicalOwner,
6339 ));
6340 }
6341 let active_binding_valid = match (provenance.phase, block.active_binding) {
6342 (RecoveryClaimPhase::PreFate, Some(active_binding)) => {
6343 active_binding.owner.participant_index == provenance.participant_index
6344 && active_binding.owner.binding_epoch == provenance.prior_binding_epoch
6345 }
6346 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
6347 _ => false,
6348 };
6349 if !active_binding_valid {
6350 return Err(order_error(
6351 block_ordinal,
6352 ClaimFrontierInvalidReason::RecoveryBlock,
6353 ));
6354 }
6355 Ok(())
6356 }
6357 }
6358}
6359
6360fn validate_order_candidates(
6361 restore: &[ImmutableOrderCandidateMajorRestore],
6362 ledger: OrderLedger,
6363) -> Result<Vec<ImmutableOrderCandidateMajor>, ClaimFrontierError> {
6364 let mut groups = restore.to_vec();
6365 groups.sort_by_key(|group| group.transaction_order);
6366 let mut seen_keys = Vec::new();
6367 let mut previous_major = None;
6368 let mut validated = Vec::new();
6369 for group in groups {
6370 let ordinal = order_ordinal(ledger, group.transaction_order);
6371 let below_allocated_high = matches!(
6372 ledger.high(),
6373 OrderHigh::Allocated(high) if group.transaction_order < high
6374 );
6375 if group.candidate_keys.is_empty()
6376 || previous_major == Some(group.transaction_order)
6377 || below_allocated_high
6378 {
6379 return Err(order_error(
6380 ordinal,
6381 ClaimFrontierInvalidReason::CandidateKey,
6382 ));
6383 }
6384 previous_major = Some(group.transaction_order);
6385 let mut previous = None;
6386 for key in &group.candidate_keys {
6387 if key.transaction_order() != group.transaction_order
6388 || previous.is_some_and(|previous| previous >= *key)
6389 || seen_keys.contains(key)
6390 {
6391 return Err(order_error(
6392 ordinal,
6393 ClaimFrontierInvalidReason::CandidateKey,
6394 ));
6395 }
6396 previous = Some(*key);
6397 seen_keys.push(*key);
6398 }
6399 validated.push(ImmutableOrderCandidateMajor {
6400 transaction_order: group.transaction_order,
6401 candidate_keys: group.candidate_keys,
6402 });
6403 }
6404 Ok(validated)
6405}
6406
6407fn validate_order_direct_owners(
6408 active: &ActiveIdentityRanks,
6409 movable: &[MovableOrderClaim],
6410 recovery: Option<RecoveryOrderBlockRestore>,
6411 ledger: OrderLedger,
6412) -> Result<(), ClaimFrontierError> {
6413 let mut ordered = movable.to_vec();
6414 ordered.sort_by_key(|claim| claim.transaction_order);
6415 let mut exits = Vec::new();
6416 let mut terminals = Vec::new();
6417 for claim in ordered {
6418 let ordinal = order_ordinal(ledger, claim.transaction_order);
6419 match claim.owner {
6420 OrderDirectOwner::MembershipExit { participant_index } => {
6421 if !active.contains(participant_index) || exits.contains(&participant_index) {
6422 return Err(order_error(
6423 ordinal,
6424 ClaimFrontierInvalidReason::LogicalOwner,
6425 ));
6426 }
6427 exits.push(participant_index);
6428 }
6429 OrderDirectOwner::ActiveBindingTerminal(owner) => {
6430 if !terminal_matches_bound(active, owner) || terminals.contains(&owner) {
6431 return Err(order_error(
6432 ordinal,
6433 ClaimFrontierInvalidReason::LogicalOwner,
6434 ));
6435 }
6436 terminals.push(owner);
6437 }
6438 }
6439 }
6440 if let Some(active_binding) = recovery.and_then(|block| block.active_binding) {
6441 let ordinal = order_ordinal(ledger, active_binding.transaction_order);
6442 if !terminal_matches_bound(active, active_binding.owner)
6443 || terminals.contains(&active_binding.owner)
6444 {
6445 return Err(order_error(
6446 ordinal,
6447 ClaimFrontierInvalidReason::LogicalOwner,
6448 ));
6449 }
6450 terminals.push(active_binding.owner);
6451 }
6452 exits.sort_unstable();
6453 if exits.len() != active.participants.len()
6454 || !exits.iter().copied().eq(active
6455 .participants
6456 .iter()
6457 .map(|participant| participant.participant_index))
6458 {
6459 return Err(order_error(
6460 ledger.claims().total(),
6461 ClaimFrontierInvalidReason::LogicalOwner,
6462 ));
6463 }
6464 terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6465 if usize_to_u128(terminals.len()) != u128::from(ledger.claims().active_binding_terminals()) {
6466 return Err(order_error(
6467 ledger.claims().total(),
6468 ClaimFrontierInvalidReason::LogicalOwner,
6469 ));
6470 }
6471 Ok(())
6472}
6473
6474fn validate_cross_counter(
6475 sequence: &SequenceClaimFrontier,
6476 order: &OrderClaimFrontier,
6477) -> Result<(), ClaimFrontierError> {
6478 match (sequence.recovery, order.recovery) {
6479 (None, None) => {}
6480 (Some(sequence_block), Some(order_block))
6481 if sequence_block.participant_index == order_block.participant_index
6482 && sequence_block.marker_delivery_seq == order_block.marker_delivery_seq
6483 && sequence_block.recovered_binding_epoch
6484 == order_block.recovered_binding_epoch
6485 && sequence_block.terminal.map(|terminal| terminal.owner)
6486 == order_block.active_binding.map(|active| active.owner) => {}
6487 (Some(sequence_block), _) => {
6488 return Err(sequence_error(
6489 sequence_ordinal(
6490 sequence.ledger,
6491 block_start_validated_sequence(sequence_block),
6492 ),
6493 ClaimFrontierInvalidReason::RecoveryBlock,
6494 ));
6495 }
6496 (None, Some(order_block)) => {
6497 return Err(order_error(
6498 order_ordinal(order.ledger, block_start_validated_order(order_block)),
6499 ClaimFrontierInvalidReason::RecoveryBlock,
6500 ));
6501 }
6502 }
6503
6504 let mut order_candidate_keys = Vec::new();
6505 for group in &order.immutable_candidates {
6506 order_candidate_keys.extend(group.candidate_keys.iter().copied());
6507 }
6508 for candidate in &sequence.immutable_candidates {
6509 let key = candidate.admission_order();
6510 if !order_candidate_keys.contains(&key) {
6511 return Err(sequence_error(
6512 sequence_ordinal(sequence.ledger, candidate.delivery_seq()),
6513 ClaimFrontierInvalidReason::CandidateKey,
6514 ));
6515 }
6516 }
6517 for group in &order.immutable_candidates {
6518 for key in &group.candidate_keys {
6519 if !sequence
6520 .immutable_candidates
6521 .iter()
6522 .any(|candidate| candidate.admission_order() == *key)
6523 {
6524 return Err(order_error(
6525 order_ordinal(order.ledger, group.transaction_order),
6526 ClaimFrontierInvalidReason::CandidateKey,
6527 ));
6528 }
6529 }
6530 }
6531
6532 let mut sequence_movable_terminals = Vec::new();
6533 for claim in &sequence.movable_claims {
6534 if let SequenceDirectOwner::BindingTerminal(owner) = claim.owner {
6535 sequence_movable_terminals.push(owner);
6536 }
6537 }
6538 if let Some(terminal) = sequence.recovery.and_then(|block| block.terminal) {
6539 sequence_movable_terminals.push(terminal.owner);
6540 }
6541 let mut order_movable_terminals = Vec::new();
6542 for claim in &order.movable_claims {
6543 if let OrderDirectOwner::ActiveBindingTerminal(owner) = claim.owner {
6544 order_movable_terminals.push(owner);
6545 }
6546 }
6547 if let Some(active_binding) = order.recovery.and_then(|block| block.active_binding) {
6548 order_movable_terminals.push(active_binding.owner);
6549 }
6550 sequence_movable_terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6551 order_movable_terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6552 if sequence_movable_terminals != order_movable_terminals {
6553 return Err(sequence_error(
6554 sequence.ledger.required_reserve(),
6555 ClaimFrontierInvalidReason::LogicalOwner,
6556 ));
6557 }
6558 Ok(())
6559}
6560
6561const fn marker_provenance_targets(provenance: MarkerProvenance, target: ParticipantId) -> bool {
6562 match provenance {
6563 MarkerProvenance::NonProductM => true,
6564 MarkerProvenance::TerminalProduct {
6565 affected_participant,
6566 ..
6567 } => affected_participant == target,
6568 MarkerProvenance::ExitProduct {
6569 exit_participant,
6570 remaining_participant,
6571 } => exit_participant != remaining_participant && remaining_participant == target,
6572 }
6573}
6574
6575fn marker_has_causal_authority(
6576 marker: MarkerCandidateAuthority,
6577 records: &[RetainedCausalRecord],
6578 historical: &[HistoricalCausalAuthority],
6579) -> bool {
6580 if marker.provenance == MarkerProvenance::NonProductM {
6581 return true;
6582 }
6583 let retained_match = records.iter().any(|record| {
6584 if record.admission_order.transaction_order() != marker.admission_order.transaction_order()
6585 {
6586 return false;
6587 }
6588 match marker.provenance {
6589 MarkerProvenance::NonProductM => true,
6590 MarkerProvenance::TerminalProduct {
6591 terminal: TerminalProductSource::Binding(owner),
6592 ..
6593 } => matches!(
6594 record.kind,
6595 RetainedCausalRecordKind::BindingTerminal(actual) if actual == owner
6596 ),
6597 MarkerProvenance::TerminalProduct {
6598 terminal:
6599 TerminalProductSource::RecoveryReplacement {
6600 participant_index,
6601 binding_epoch,
6602 },
6603 ..
6604 } => matches!(
6605 record.kind,
6606 RetainedCausalRecordKind::BindingTerminal(owner)
6607 if owner.participant_index == participant_index
6608 && owner.binding_epoch == binding_epoch
6609 ),
6610 MarkerProvenance::ExitProduct {
6611 exit_participant, ..
6612 } => matches!(
6613 record.kind,
6614 RetainedCausalRecordKind::MembershipExit { participant_index }
6615 if participant_index == exit_participant
6616 ),
6617 }
6618 });
6619 retained_match
6620 || historical
6621 .iter()
6622 .any(|authority| match (marker.provenance, authority.kind) {
6623 (
6624 MarkerProvenance::TerminalProduct {
6625 terminal: TerminalProductSource::Binding(expected),
6626 ..
6627 },
6628 HistoricalCausalKind::BindingTerminal(owner),
6629 ) => {
6630 owner == expected
6631 && authority.admission_order.transaction_order()
6632 == marker.admission_order.transaction_order()
6633 }
6634 (
6635 MarkerProvenance::TerminalProduct {
6636 terminal:
6637 TerminalProductSource::RecoveryReplacement {
6638 participant_index,
6639 binding_epoch,
6640 },
6641 ..
6642 },
6643 HistoricalCausalKind::BindingTerminal(owner),
6644 ) => {
6645 owner.participant_index == participant_index
6646 && owner.binding_epoch == binding_epoch
6647 && authority.admission_order.transaction_order()
6648 == marker.admission_order.transaction_order()
6649 }
6650 (
6651 MarkerProvenance::ExitProduct {
6652 exit_participant, ..
6653 },
6654 HistoricalCausalKind::MembershipExit(participant_index),
6655 ) => {
6656 participant_index == exit_participant
6657 && authority.admission_order.transaction_order()
6658 == marker.admission_order.transaction_order()
6659 }
6660 _ => false,
6661 })
6662}
6663
6664fn terminal_matches_active(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
6665 active_participant(active, owner.participant_index)
6666 .is_some_and(|participant| binding_epoch(participant.binding) == owner.binding_epoch)
6667}
6668
6669fn terminal_matches_bound(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
6670 active_participant(active, owner.participant_index).is_some_and(|participant| {
6671 participant.binding == FrontierBinding::Bound(owner.binding_epoch)
6672 })
6673}
6674
6675fn active_participant(
6676 active: &ActiveIdentityRanks,
6677 participant_index: ParticipantId,
6678) -> Option<FrontierParticipant> {
6679 active
6680 .participants
6681 .binary_search_by_key(&participant_index, |participant| {
6682 participant.participant_index
6683 })
6684 .ok()
6685 .and_then(|index| active.participants.get(index))
6686 .copied()
6687}
6688
6689const fn binding_epoch(binding: FrontierBinding) -> BindingEpoch {
6690 match binding {
6691 FrontierBinding::Bound(epoch) | FrontierBinding::Detached(epoch) => epoch,
6692 }
6693}
6694
6695fn block_start_sequence(block: RecoverySequenceBlockRestore) -> DeliverySeq {
6696 block
6697 .terminal
6698 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
6699}
6700
6701fn block_start_validated_sequence(block: RecoverySequenceBlock) -> DeliverySeq {
6702 block
6703 .terminal
6704 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
6705}
6706
6707fn block_start_order(block: RecoveryOrderBlockRestore) -> TransactionOrder {
6708 block
6709 .active_binding
6710 .map_or(block.recovery_operation_order, |active_binding| {
6711 active_binding.transaction_order
6712 })
6713}
6714
6715fn block_start_validated_order(block: RecoveryOrderBlock) -> TransactionOrder {
6716 block
6717 .active_binding
6718 .map_or(block.recovery_operation_order, |active_binding| {
6719 active_binding.transaction_order
6720 })
6721}
6722
6723fn order_frontier_start(high: OrderHigh) -> u128 {
6724 match high {
6725 OrderHigh::Empty => 0,
6726 OrderHigh::Allocated(high) => u128::from(high) + 1,
6727 }
6728}
6729
6730const fn order_is_above_high(value: TransactionOrder, high: OrderHigh) -> bool {
6731 match high {
6732 OrderHigh::Empty => true,
6733 OrderHigh::Allocated(high) => value > high,
6734 }
6735}
6736
6737fn order_frontier_candidate_count(restore: &OrderClaimFrontierRestore, high: OrderHigh) -> u128 {
6738 usize_to_u128(
6739 restore
6740 .immutable_candidates
6741 .iter()
6742 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
6743 .count(),
6744 )
6745}
6746
6747fn sequence_ordinal(ledger: SequenceLedger, value: DeliverySeq) -> u128 {
6748 u128::from(value).saturating_sub(u128::from(ledger.high_watermark()) + 1)
6749}
6750
6751fn order_ordinal(ledger: OrderLedger, value: TransactionOrder) -> u128 {
6752 u128::from(value).saturating_sub(order_frontier_start(ledger.high()))
6753}
6754
6755fn checked_rank_value(start: DeliverySeq, active_rank: usize) -> Option<DeliverySeq> {
6756 let rank = u64::try_from(active_rank).ok()?;
6757 start.checked_add(rank)
6758}
6759
6760fn usize_to_u64(value: usize) -> u64 {
6761 u64::try_from(value).map_or(u64::MAX, core::convert::identity)
6762}
6763
6764fn usize_to_u128(value: usize) -> u128 {
6765 u64::try_from(value).map_or(u128::MAX, u128::from)
6766}
6767
6768fn rank_index(rank: usize) -> u128 {
6769 usize_to_u128(rank)
6770}
6771
6772const fn frontier_error(
6773 counter: ClaimFrontierCounter,
6774 first_bad_position: u128,
6775 reason: ClaimFrontierInvalidReason,
6776) -> ClaimFrontierError {
6777 ClaimFrontierError {
6778 counter,
6779 first_bad_position,
6780 reason,
6781 }
6782}
6783
6784const fn sequence_error(
6785 first_bad_position: u128,
6786 reason: ClaimFrontierInvalidReason,
6787) -> ClaimFrontierError {
6788 frontier_error(
6789 ClaimFrontierCounter::DeliverySequence,
6790 first_bad_position,
6791 reason,
6792 )
6793}
6794
6795const fn order_error(
6796 first_bad_position: u128,
6797 reason: ClaimFrontierInvalidReason,
6798) -> ClaimFrontierError {
6799 frontier_error(
6800 ClaimFrontierCounter::TransactionOrder,
6801 first_bad_position,
6802 reason,
6803 )
6804}