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, ConnectionIncarnation, ConversationId, DeliverySeq,
19 Generation, ParticipantId, 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)]
412pub(super) struct HistoricalMarkerDeliveryAuthority {
413 participant_index: ParticipantId,
414 marker_delivery_seq: DeliverySeq,
415 delivered_binding_epoch: BindingEpoch,
416}
417
418impl HistoricalMarkerDeliveryAuthority {
419 pub(super) const fn participant_index(self) -> ParticipantId {
421 self.participant_index
422 }
423
424 pub(super) const fn marker_delivery_seq(self) -> DeliverySeq {
426 self.marker_delivery_seq
427 }
428
429 pub(super) const fn delivered_binding_epoch(self) -> BindingEpoch {
431 self.delivered_binding_epoch
432 }
433}
434
435impl HistoricalCausalAuthority {
436 pub(super) const fn from_committed_terminal(terminal: super::CommittedBindingTerminal) -> Self {
437 Self {
438 conversation_id: terminal.conversation_id(),
439 admission_order: terminal.admission_order(),
440 kind: HistoricalCausalKind::BindingTerminal(BindingTerminalOwner {
441 participant_index: terminal.participant_id(),
442 binding_epoch: terminal.binding_epoch(),
443 }),
444 }
445 }
446
447 pub(super) const fn from_retired<EF, V, LF>(
448 retired: &super::RetiredIdentity<EF, V, LF>,
449 ) -> Self {
450 Self {
451 conversation_id: retired.conversation_id(),
452 admission_order: retired.left_admission_order(),
453 kind: HistoricalCausalKind::MembershipExit(retired.participant_id()),
454 }
455 }
456
457 const fn from_restore(fact: HistoricalCausalFactRestore) -> Self {
458 match fact {
459 HistoricalCausalFactRestore::BindingTerminal {
460 conversation_id,
461 participant_index,
462 binding_epoch,
463 admission_order,
464 } => Self {
465 conversation_id,
466 admission_order,
467 kind: HistoricalCausalKind::BindingTerminal(BindingTerminalOwner {
468 participant_index,
469 binding_epoch,
470 }),
471 },
472 HistoricalCausalFactRestore::MembershipExit {
473 conversation_id,
474 participant_index,
475 admission_order,
476 } => Self {
477 conversation_id,
478 admission_order,
479 kind: HistoricalCausalKind::MembershipExit(participant_index),
480 },
481 }
482 }
483
484 const fn describe(self) -> HistoricalCausalFactRestore {
487 match self.kind {
488 HistoricalCausalKind::BindingTerminal(owner) => {
489 HistoricalCausalFactRestore::BindingTerminal {
490 conversation_id: self.conversation_id,
491 participant_index: owner.participant_index,
492 binding_epoch: owner.binding_epoch,
493 admission_order: self.admission_order,
494 }
495 }
496 HistoricalCausalKind::MembershipExit(participant_index) => {
497 HistoricalCausalFactRestore::MembershipExit {
498 conversation_id: self.conversation_id,
499 participant_index,
500 admission_order: self.admission_order,
501 }
502 }
503 }
504 }
505}
506
507#[derive(Debug)]
508pub(super) struct ValidatedConversationHistory {
509 causal_authorities: Vec<HistoricalCausalAuthority>,
510 binding_origins: Vec<BindingOrigin>,
511 total: bool,
512}
513
514impl ValidatedConversationHistory {
515 pub(super) const fn empty() -> Self {
516 Self {
517 causal_authorities: Vec::new(),
518 binding_origins: Vec::new(),
519 total: false,
520 }
521 }
522
523 pub(super) const fn new(
524 causal_authorities: Vec<HistoricalCausalAuthority>,
525 binding_origins: Vec<BindingOrigin>,
526 ) -> Self {
527 Self {
528 causal_authorities,
529 binding_origins,
530 total: true,
531 }
532 }
533
534 pub(super) fn ordinary_origin(
535 &self,
536 conversation_id: ConversationId,
537 participant_index: ParticipantId,
538 binding_epoch: BindingEpoch,
539 ) -> Option<&BindingOrigin> {
540 self.binding_origins.iter().find(|origin| {
541 origin.conversation_id() == conversation_id
542 && origin.participant_id() == participant_index
543 && origin.binding_epoch() == binding_epoch
544 && origin.is_unfenced()
545 })
546 }
547}
548
549#[derive(Clone, Copy, Debug, PartialEq, Eq)]
551pub enum SequenceProductClass {
552 LiveTimesTerminal,
554 LiveTimesReplacementTerminal,
556 OtherLiveTimesExit,
558}
559
560#[derive(Clone, Copy, Debug, PartialEq, Eq)]
562pub enum MarkerSequenceOwner {
563 Marker,
565 ConditionalProduct(SequenceProductClass),
567}
568
569#[derive(Clone, Copy, Debug, PartialEq, Eq)]
571pub struct MarkerCandidateAuthority {
572 pub delivery_seq: DeliverySeq,
574 pub admission_order: super::AdmissionOrder,
576 pub target_binding: FrontierBinding,
578 pub provenance: MarkerProvenance,
580 pub abandoned_after: DeliverySeq,
582 pub abandoned_through: DeliverySeq,
584 pub physical_floor_at_decision: DeliverySeq,
586 pub current_owner: MarkerSequenceOwner,
588}
589
590#[derive(Clone, Copy, Debug, PartialEq, Eq)]
592pub enum ImmutableSequenceCandidate {
593 BindingTerminal {
595 delivery_seq: DeliverySeq,
597 admission_order: super::AdmissionOrder,
599 owner: BindingTerminalOwner,
601 },
602 Marker(MarkerCandidateAuthority),
604}
605
606impl ImmutableSequenceCandidate {
607 #[must_use]
609 pub const fn delivery_seq(self) -> DeliverySeq {
610 match self {
611 Self::BindingTerminal { delivery_seq, .. } => delivery_seq,
612 Self::Marker(candidate) => candidate.delivery_seq,
613 }
614 }
615
616 #[must_use]
618 pub const fn admission_order(self) -> super::AdmissionOrder {
619 match self {
620 Self::BindingTerminal {
621 admission_order, ..
622 } => admission_order,
623 Self::Marker(candidate) => candidate.admission_order,
624 }
625 }
626}
627
628#[derive(Clone, Copy, Debug, PartialEq, Eq)]
630pub struct TerminalProductRangeRestore {
631 pub start: DeliverySeq,
633 pub length: u64,
635 pub terminal: BindingTerminalOwner,
637}
638
639#[derive(Clone, Copy, Debug, PartialEq, Eq)]
641pub struct ReplacementTerminalProductRangeRestore {
642 pub start: DeliverySeq,
644 pub length: u64,
646}
647
648#[derive(Clone, Copy, Debug, PartialEq, Eq)]
650pub struct ExitProductRangeRestore {
651 pub start: DeliverySeq,
653 pub length: u64,
655 pub exit_participant: ParticipantId,
657}
658
659#[derive(Clone, Debug, Default, PartialEq, Eq)]
661pub struct SequenceProductRangesRestore {
662 pub live_times_terminal: Vec<TerminalProductRangeRestore>,
664 pub live_times_replacement_terminal: Option<ReplacementTerminalProductRangeRestore>,
666 pub other_live_times_exit: Vec<ExitProductRangeRestore>,
668}
669
670#[derive(Clone, Copy, Debug, PartialEq, Eq)]
672pub struct TerminalProductRange {
673 start: DeliverySeq,
674 length: u64,
675 terminal: BindingTerminalOwner,
676}
677
678impl TerminalProductRange {
679 #[must_use]
681 pub const fn start(self) -> DeliverySeq {
682 self.start
683 }
684
685 #[must_use]
687 pub const fn terminal(self) -> BindingTerminalOwner {
688 self.terminal
689 }
690
691 #[must_use]
693 pub const fn length(self) -> u64 {
694 self.length
695 }
696
697 #[must_use]
699 pub fn value_at_rank(self, active_rank: usize) -> Option<DeliverySeq> {
700 if usize_to_u128(active_rank) >= u128::from(self.length) {
701 return None;
702 }
703 checked_rank_value(self.start, active_rank)
704 }
705}
706
707#[derive(Clone, Copy, Debug, PartialEq, Eq)]
709pub struct ReplacementTerminalProductRange {
710 start: DeliverySeq,
711 length: u64,
712 participant_index: ParticipantId,
713 marker_delivery_seq: DeliverySeq,
714 prior_binding_epoch: BindingEpoch,
715}
716
717impl ReplacementTerminalProductRange {
718 #[must_use]
720 pub const fn start(self) -> DeliverySeq {
721 self.start
722 }
723
724 #[must_use]
726 pub const fn length(self) -> u64 {
727 self.length
728 }
729
730 #[must_use]
732 pub const fn participant_index(self) -> ParticipantId {
733 self.participant_index
734 }
735
736 #[must_use]
738 pub const fn marker_delivery_seq(self) -> DeliverySeq {
739 self.marker_delivery_seq
740 }
741
742 #[must_use]
744 pub const fn prior_binding_epoch(self) -> BindingEpoch {
745 self.prior_binding_epoch
746 }
747
748 #[must_use]
750 pub fn value_at_rank(self, active_rank: usize) -> Option<DeliverySeq> {
751 if usize_to_u128(active_rank) >= u128::from(self.length) {
752 return None;
753 }
754 checked_rank_value(self.start, active_rank)
755 }
756}
757
758#[derive(Clone, Copy, Debug, PartialEq, Eq)]
760pub struct ExitProductRange {
761 start: DeliverySeq,
762 length: u64,
763 exit_participant: ParticipantId,
764}
765
766impl ExitProductRange {
767 #[must_use]
769 pub const fn start(self) -> DeliverySeq {
770 self.start
771 }
772
773 #[must_use]
775 pub const fn exit_participant(self) -> ParticipantId {
776 self.exit_participant
777 }
778
779 #[must_use]
781 pub const fn length(self) -> u64 {
782 self.length
783 }
784
785 #[must_use]
789 pub fn value_for_affected_rank(
790 self,
791 active_identities: &ActiveIdentityRanks,
792 affected_rank: usize,
793 ) -> Option<DeliverySeq> {
794 let affected = active_identities.participants().get(affected_rank)?;
795 if usize_to_u128(affected_rank) >= usize_to_u128(active_identities.participants.len()) {
796 return None;
797 }
798 if affected.participant_index == self.exit_participant {
799 return None;
800 }
801 let exit_rank = active_identities
802 .participants()
803 .binary_search_by_key(&self.exit_participant, |participant| {
804 participant.participant_index
805 })
806 .ok()?;
807 let compact_rank = if affected_rank < exit_rank {
808 affected_rank
809 } else {
810 affected_rank.checked_sub(1)?
811 };
812 if usize_to_u128(compact_rank) >= u128::from(self.length) {
813 return None;
814 }
815 checked_rank_value(self.start, compact_rank)
816 }
817}
818
819#[derive(Clone, Debug, Default, PartialEq, Eq)]
821pub struct SequenceProductRanges {
822 live_times_terminal: Vec<TerminalProductRange>,
823 live_times_replacement_terminal: Option<ReplacementTerminalProductRange>,
824 other_live_times_exit: Vec<ExitProductRange>,
825}
826
827impl SequenceProductRanges {
828 #[must_use]
830 pub fn live_times_terminal(&self) -> &[TerminalProductRange] {
831 &self.live_times_terminal
832 }
833
834 #[must_use]
836 pub const fn live_times_replacement_terminal(&self) -> Option<ReplacementTerminalProductRange> {
837 self.live_times_replacement_terminal
838 }
839
840 #[must_use]
842 pub fn other_live_times_exit(&self) -> &[ExitProductRange] {
843 &self.other_live_times_exit
844 }
845}
846
847#[derive(Clone, Copy, Debug, PartialEq, Eq)]
849pub struct RecoverySequenceTerminalRestore {
850 pub delivery_seq: DeliverySeq,
852 pub owner: BindingTerminalOwner,
854}
855
856#[derive(Clone, Copy, Debug, PartialEq, Eq)]
862pub struct RecoverySequenceBlockRestore {
863 pub terminal: Option<RecoverySequenceTerminalRestore>,
865 pub recovery_attach_seq: DeliverySeq,
867 pub replacement_terminal_seq: DeliverySeq,
869}
870
871#[derive(Clone, Copy, Debug, PartialEq, Eq)]
873pub struct RecoverySequenceBlock {
874 terminal: Option<RecoverySequenceTerminalRestore>,
875 recovery_attach_seq: DeliverySeq,
876 replacement_terminal_seq: DeliverySeq,
877 participant_index: ParticipantId,
878 marker_delivery_seq: DeliverySeq,
879 recovered_binding_epoch: BindingEpoch,
880}
881
882impl RecoverySequenceBlock {
883 #[must_use]
885 pub const fn terminal(self) -> Option<RecoverySequenceTerminalRestore> {
886 self.terminal
887 }
888
889 #[must_use]
891 pub const fn recovery_attach_seq(self) -> DeliverySeq {
892 self.recovery_attach_seq
893 }
894
895 #[must_use]
897 pub const fn replacement_terminal_seq(self) -> DeliverySeq {
898 self.replacement_terminal_seq
899 }
900
901 #[must_use]
903 pub const fn participant_index(self) -> ParticipantId {
904 self.participant_index
905 }
906
907 #[must_use]
909 pub const fn marker_delivery_seq(self) -> DeliverySeq {
910 self.marker_delivery_seq
911 }
912
913 #[must_use]
915 pub const fn recovered_binding_epoch(self) -> BindingEpoch {
916 self.recovered_binding_epoch
917 }
918}
919
920#[derive(Clone, Debug, Default, PartialEq, Eq)]
922pub struct SequenceClaimFrontierRestore {
923 pub movable_claims: Vec<MovableSequenceClaim>,
925 pub immutable_candidates: Vec<ImmutableSequenceCandidate>,
927 pub products: SequenceProductRangesRestore,
929 pub recovery: Option<RecoverySequenceBlockRestore>,
931}
932
933#[derive(Clone, Debug, PartialEq, Eq)]
935pub struct SequenceClaimFrontier {
936 ledger: SequenceLedger,
937 movable_claims: Vec<MovableSequenceClaim>,
938 immutable_candidates: Vec<ImmutableSequenceCandidate>,
939 products: SequenceProductRanges,
940 recovery: Option<RecoverySequenceBlock>,
941}
942
943impl SequenceClaimFrontier {
944 #[must_use]
946 pub const fn ledger(&self) -> SequenceLedger {
947 self.ledger
948 }
949
950 #[must_use]
952 pub fn movable_claims(&self) -> &[MovableSequenceClaim] {
953 &self.movable_claims
954 }
955
956 #[must_use]
958 pub fn immutable_candidates(&self) -> &[ImmutableSequenceCandidate] {
959 &self.immutable_candidates
960 }
961
962 #[must_use]
964 pub const fn products(&self) -> &SequenceProductRanges {
965 &self.products
966 }
967
968 #[must_use]
970 pub const fn recovery(&self) -> Option<RecoverySequenceBlock> {
971 self.recovery
972 }
973}
974
975#[derive(Clone, Copy, Debug, PartialEq, Eq)]
977pub enum OrderDirectOwner {
978 ActiveBindingTerminal(BindingTerminalOwner),
980 MembershipExit {
982 participant_index: ParticipantId,
984 },
985}
986
987#[derive(Clone, Copy, Debug, PartialEq, Eq)]
989pub struct MovableOrderClaim {
990 pub transaction_order: TransactionOrder,
992 pub owner: OrderDirectOwner,
994}
995
996#[derive(Clone, Debug, PartialEq, Eq)]
1001pub struct ImmutableOrderCandidateMajorRestore {
1002 pub transaction_order: TransactionOrder,
1004 pub candidate_keys: Vec<super::AdmissionOrder>,
1006}
1007
1008#[derive(Clone, Debug, PartialEq, Eq)]
1010pub struct ImmutableOrderCandidateMajor {
1011 transaction_order: TransactionOrder,
1012 candidate_keys: Vec<super::AdmissionOrder>,
1013}
1014
1015impl ImmutableOrderCandidateMajor {
1016 #[must_use]
1018 pub const fn transaction_order(&self) -> TransactionOrder {
1019 self.transaction_order
1020 }
1021
1022 #[must_use]
1024 pub fn candidate_keys(&self) -> &[super::AdmissionOrder] {
1025 &self.candidate_keys
1026 }
1027}
1028
1029#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1031pub struct RecoveryOrderActiveBindingRestore {
1032 pub transaction_order: TransactionOrder,
1034 pub owner: BindingTerminalOwner,
1036}
1037
1038#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1040pub struct RecoveryOrderBlockRestore {
1041 pub active_binding: Option<RecoveryOrderActiveBindingRestore>,
1043 pub recovery_operation_order: TransactionOrder,
1045 pub replacement_terminal_order: TransactionOrder,
1047}
1048
1049#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1051pub struct RecoveryOrderBlock {
1052 active_binding: Option<RecoveryOrderActiveBindingRestore>,
1053 recovery_operation_order: TransactionOrder,
1054 replacement_terminal_order: TransactionOrder,
1055 participant_index: ParticipantId,
1056 marker_delivery_seq: DeliverySeq,
1057 recovered_binding_epoch: BindingEpoch,
1058}
1059
1060impl RecoveryOrderBlock {
1061 #[must_use]
1063 pub const fn active_binding(self) -> Option<RecoveryOrderActiveBindingRestore> {
1064 self.active_binding
1065 }
1066
1067 #[must_use]
1069 pub const fn recovery_operation_order(self) -> TransactionOrder {
1070 self.recovery_operation_order
1071 }
1072
1073 #[must_use]
1075 pub const fn replacement_terminal_order(self) -> TransactionOrder {
1076 self.replacement_terminal_order
1077 }
1078
1079 #[must_use]
1081 pub const fn participant_index(self) -> ParticipantId {
1082 self.participant_index
1083 }
1084
1085 #[must_use]
1087 pub const fn marker_delivery_seq(self) -> DeliverySeq {
1088 self.marker_delivery_seq
1089 }
1090
1091 #[must_use]
1093 pub const fn recovered_binding_epoch(self) -> BindingEpoch {
1094 self.recovered_binding_epoch
1095 }
1096}
1097
1098#[derive(Clone, Debug, Default, PartialEq, Eq)]
1100pub struct OrderClaimFrontierRestore {
1101 pub movable_claims: Vec<MovableOrderClaim>,
1103 pub immutable_candidates: Vec<ImmutableOrderCandidateMajorRestore>,
1105 pub recovery: Option<RecoveryOrderBlockRestore>,
1107}
1108
1109#[derive(Clone, Debug, PartialEq, Eq)]
1111pub struct OrderClaimFrontier {
1112 ledger: OrderLedger,
1113 movable_claims: Vec<MovableOrderClaim>,
1114 immutable_candidates: Vec<ImmutableOrderCandidateMajor>,
1115 recovery: Option<RecoveryOrderBlock>,
1116}
1117
1118impl OrderClaimFrontier {
1119 #[must_use]
1121 pub const fn ledger(&self) -> OrderLedger {
1122 self.ledger
1123 }
1124
1125 #[must_use]
1127 pub fn movable_claims(&self) -> &[MovableOrderClaim] {
1128 &self.movable_claims
1129 }
1130
1131 #[must_use]
1133 pub fn immutable_candidates(&self) -> &[ImmutableOrderCandidateMajor] {
1134 &self.immutable_candidates
1135 }
1136
1137 #[must_use]
1139 pub const fn recovery(&self) -> Option<RecoveryOrderBlock> {
1140 self.recovery
1141 }
1142}
1143
1144#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1145enum RecoveryClaimPhase {
1146 PreFate,
1147 PostFate,
1148 RecoveredBound,
1149}
1150
1151#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1159pub struct RecoveryClaimProvenance {
1160 participant_index: ParticipantId,
1161 marker_delivery_seq: DeliverySeq,
1162 prior_binding_epoch: BindingEpoch,
1163 current_binding_epoch: BindingEpoch,
1164 phase: RecoveryClaimPhase,
1165}
1166
1167impl RecoveryClaimProvenance {
1168 #[must_use]
1171 pub const fn from_stored_edge(edge: super::StoredEdge) -> Option<Self> {
1172 match edge {
1173 super::StoredEdge::MarkerDelivery(delivery) => Some(Self {
1174 participant_index: delivery.participant_id(),
1175 marker_delivery_seq: delivery.marker_delivery_seq(),
1176 prior_binding_epoch: delivery.binding_epoch(),
1177 current_binding_epoch: delivery.binding_epoch(),
1178 phase: RecoveryClaimPhase::PreFate,
1179 }),
1180 super::StoredEdge::ParticipantCursorProgress(progress) => {
1181 let Some(marker_delivery_seq) = progress.marker_delivery_seq() else {
1182 return None;
1183 };
1184 Some(Self {
1185 participant_index: progress.participant_id(),
1186 marker_delivery_seq,
1187 prior_binding_epoch: progress.binding_epoch(),
1188 current_binding_epoch: progress.binding_epoch(),
1189 phase: RecoveryClaimPhase::PreFate,
1190 })
1191 }
1192 super::StoredEdge::DetachedCredentialRecovery(recovery) => Some(Self {
1193 participant_index: recovery.participant_id(),
1194 marker_delivery_seq: recovery.marker_delivery_seq(),
1195 prior_binding_epoch: recovery.prior_binding_epoch(),
1196 current_binding_epoch: recovery.prior_binding_epoch(),
1197 phase: RecoveryClaimPhase::PostFate,
1198 }),
1199 _ => None,
1200 }
1201 }
1202
1203 #[must_use]
1205 pub const fn participant_index(self) -> ParticipantId {
1206 self.participant_index
1207 }
1208
1209 #[must_use]
1211 pub const fn marker_delivery_seq(self) -> DeliverySeq {
1212 self.marker_delivery_seq
1213 }
1214
1215 #[must_use]
1217 pub const fn prior_binding_epoch(self) -> BindingEpoch {
1218 self.prior_binding_epoch
1219 }
1220}
1221
1222#[derive(Clone, Debug, PartialEq, Eq)]
1224pub struct ClaimFrontiersRestore {
1225 pub conversation_id: ConversationId,
1227 pub active_identities: Vec<FrontierParticipant>,
1229 pub identity_slot_limit: u64,
1231 pub retained_floor: u128,
1233 pub retained_record_limit: u64,
1235 pub retained_records: Vec<RetainedCausalRecord>,
1237 pub active_marker_anchors: Vec<DeliverySeq>,
1243 pub historical_marker_deliveries: Vec<HistoricalMarkerDeliveryFactRestore>,
1249 pub historical_causal_facts: Vec<HistoricalCausalFactRestore>,
1251 pub sequence: SequenceClaimFrontierRestore,
1253 pub order: OrderClaimFrontierRestore,
1255 pub recovery_marker_delivery_seq: Option<DeliverySeq>,
1261}
1262
1263#[derive(Clone, Debug, PartialEq, Eq)]
1274pub(super) struct ClaimFrontiersDescription {
1275 pub(super) restore: ClaimFrontiersRestore,
1277 pub(super) sequence_ledger: SequenceLedger,
1279 pub(super) order_ledger: OrderLedger,
1281 pub(super) fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1284}
1285
1286#[derive(Debug)]
1293pub(super) struct ClaimFrontiersPrevalidated {
1294 conversation_id: ConversationId,
1295 active_identities: ActiveIdentityRanks,
1296 identity_slot_limit: u64,
1297 retained_floor: u128,
1298 retained_records: Vec<RetainedCausalRecord>,
1299 marker_records: Vec<RetainedCausalRecord>,
1300 historical_marker_deliveries: Vec<HistoricalMarkerDeliveryAuthority>,
1301 historical_causal_authorities: Vec<HistoricalCausalAuthority>,
1302 binding_origins: Vec<BindingOrigin>,
1303 sequence_restore: SequenceClaimFrontierRestore,
1304 order_restore: OrderClaimFrontierRestore,
1305 recovery_marker_delivery_seq: Option<DeliverySeq>,
1306 sequence_ledger: SequenceLedger,
1307 order_ledger: OrderLedger,
1308 issued_marker_record: Option<MarkerRecordRequest>,
1309 fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1315}
1316
1317#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1318enum MarkerRecordUse {
1319 Planned(FrontierBinding),
1320 Delivered(FrontierBinding),
1321 Recovered {
1322 prior_binding_epoch: BindingEpoch,
1323 recovered_binding_epoch: BindingEpoch,
1324 },
1325}
1326
1327#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1329pub(super) struct MarkerRecordRequest {
1330 participant_index: ParticipantId,
1331 marker_delivery_seq: DeliverySeq,
1332 use_kind: MarkerRecordUse,
1333}
1334
1335impl MarkerRecordRequest {
1336 pub(super) const fn planned(
1338 participant_index: ParticipantId,
1339 marker_delivery_seq: DeliverySeq,
1340 target: FrontierBinding,
1341 ) -> Self {
1342 Self {
1343 participant_index,
1344 marker_delivery_seq,
1345 use_kind: MarkerRecordUse::Planned(target),
1346 }
1347 }
1348
1349 pub(super) const fn delivered(
1351 participant_index: ParticipantId,
1352 marker_delivery_seq: DeliverySeq,
1353 target: FrontierBinding,
1354 ) -> Self {
1355 Self {
1356 participant_index,
1357 marker_delivery_seq,
1358 use_kind: MarkerRecordUse::Delivered(target),
1359 }
1360 }
1361
1362 pub(super) const fn recovered(
1364 participant_index: ParticipantId,
1365 marker_delivery_seq: DeliverySeq,
1366 prior_binding_epoch: BindingEpoch,
1367 recovered_binding_epoch: BindingEpoch,
1368 ) -> Self {
1369 Self {
1370 participant_index,
1371 marker_delivery_seq,
1372 use_kind: MarkerRecordUse::Recovered {
1373 prior_binding_epoch,
1374 recovered_binding_epoch,
1375 },
1376 }
1377 }
1378}
1379
1380#[derive(Debug)]
1382pub(super) struct MarkerDrainCore {
1383 frontiers: ClaimFrontiers,
1384 candidate: ValidatedMarkerCandidate,
1385 record: ValidatedMarkerRecord,
1386}
1387
1388impl MarkerDrainCore {
1389 pub(super) fn into_parts(
1392 self,
1393 ) -> (
1394 ClaimFrontiers,
1395 ValidatedMarkerCandidate,
1396 ValidatedMarkerRecord,
1397 ) {
1398 (self.frontiers, self.candidate, self.record)
1399 }
1400}
1401
1402#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1404pub(super) enum MarkerDrainCoreError {
1405 NoCandidate,
1407 BindingTerminalFirst,
1409 SequenceNotNext,
1411 CausalMajorNotAllocated,
1413 MissingOrderCandidate,
1415 ResultingLedger,
1417}
1418
1419#[derive(Debug, Eq)]
1421pub struct ClaimFrontiers {
1422 ordinary_cursor_release: Option<Box<super::storage::OrdinaryBindingFateRestore>>,
1423 conversation_id: ConversationId,
1424 active_identities: ActiveIdentityRanks,
1425 identity_slot_limit: u64,
1426 retained_floor: u128,
1427 retained_records: Vec<RetainedCausalRecord>,
1428 marker_records: Vec<RetainedCausalRecord>,
1429 fenced_marker_issued: Option<(ParticipantId, DeliverySeq, BindingEpoch)>,
1430 sequence: SequenceClaimFrontier,
1431 order: OrderClaimFrontier,
1432 historical_marker_deliveries: Vec<HistoricalMarkerDeliveryAuthority>,
1442 historical_causal_authorities: Vec<HistoricalCausalAuthority>,
1450}
1451
1452impl PartialEq for ClaimFrontiers {
1453 fn eq(&self, other: &Self) -> bool {
1454 (
1456 &self.conversation_id,
1457 &self.active_identities,
1458 &self.identity_slot_limit,
1459 &self.retained_floor,
1460 &self.retained_records,
1461 &self.marker_records,
1462 &self.fenced_marker_issued,
1463 &self.sequence,
1464 &self.order,
1465 &self.historical_marker_deliveries,
1466 &self.historical_causal_authorities,
1467 ) == (
1468 &other.conversation_id,
1469 &other.active_identities,
1470 &other.identity_slot_limit,
1471 &other.retained_floor,
1472 &other.retained_records,
1473 &other.marker_records,
1474 &other.fenced_marker_issued,
1475 &other.sequence,
1476 &other.order,
1477 &other.historical_marker_deliveries,
1478 &other.historical_causal_authorities,
1479 )
1480 }
1481}
1482
1483#[derive(Debug, PartialEq, Eq)]
1489pub struct InitialEnrollmentFrontierCommit<F> {
1490 operation: InitialEnrollmentOperationCommit<F>,
1491 frontiers: ClaimFrontiers,
1492 closure_accounting: ClosureAccounting,
1493 attached_charge: ResourceVector,
1494}
1495
1496impl<F> InitialEnrollmentFrontierCommit<F> {
1497 #[must_use]
1499 pub const fn operation(&self) -> &InitialEnrollmentOperationCommit<F> {
1500 &self.operation
1501 }
1502
1503 #[must_use]
1505 pub const fn frontiers(&self) -> &ClaimFrontiers {
1506 &self.frontiers
1507 }
1508
1509 #[must_use]
1511 pub const fn closure_accounting(&self) -> ClosureAccounting {
1512 self.closure_accounting
1513 }
1514
1515 #[must_use]
1517 pub const fn attached_charge(&self) -> ResourceVector {
1518 self.attached_charge
1519 }
1520
1521 #[allow(
1523 dead_code,
1524 reason = "the next conversation event body consumes this sealed operation/frontier unit"
1525 )]
1526 pub(in crate::lifecycle) fn into_conversation_parts(
1527 self,
1528 ) -> (
1529 InitialEnrollmentOperationCommit<F>,
1530 ClaimFrontiers,
1531 ClosureAccounting,
1532 ResourceVector,
1533 ) {
1534 (
1535 self.operation,
1536 self.frontiers,
1537 self.closure_accounting,
1538 self.attached_charge,
1539 )
1540 }
1541}
1542
1543#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1545pub enum InitialEnrollmentFrontierError {
1546 AttachedChargeMismatch {
1548 expected: ResourceVector,
1550 actual: ResourceVector,
1552 },
1553 EnrollmentShape,
1555 LedgerShape,
1557 ClosureProjection,
1559 PositionOverflow,
1561 FrontierInvariant,
1563}
1564
1565#[derive(Debug, PartialEq, Eq)]
1567pub struct InitialEnrollmentFrontierFailure<F> {
1568 operation: InitialEnrollmentOperationCommit<F>,
1569 error: InitialEnrollmentFrontierError,
1570}
1571
1572impl<F> InitialEnrollmentFrontierFailure<F> {
1573 #[must_use]
1575 pub const fn error(&self) -> InitialEnrollmentFrontierError {
1576 self.error
1577 }
1578
1579 #[allow(
1581 dead_code,
1582 reason = "the conversation decision layer recovers or terminalizes the speculative enrollment"
1583 )]
1584 pub(in crate::lifecycle) fn into_operation(self) -> InitialEnrollmentOperationCommit<F> {
1585 self.operation
1586 }
1587}
1588
1589#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1591pub enum PrepareLeaveAuthorityError {
1592 Conversation,
1594 Identity,
1596 Binding,
1598 ImmutablePrefix,
1600 PendingCandidate,
1602 MembershipExitClaim,
1604 ActiveBindingClaim,
1606 OrderCapacity,
1608 ResultingOrderLedger,
1610}
1611
1612#[derive(Debug, PartialEq, Eq)]
1618pub struct ValidatedMarkerCandidate {
1619 conversation_id: ConversationId,
1620 candidate: MarkerCandidateAuthority,
1621 seal: MarkerAuthoritySeal,
1622}
1623
1624#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1627pub(super) struct FencedMarkerSourceRecord {
1628 pub(super) conversation_id: ConversationId,
1629 pub(super) delivery_seq: DeliverySeq,
1630 pub(super) admission_order: super::AdmissionOrder,
1631 pub(super) participant_id: ParticipantId,
1632 pub(super) provenance: MarkerProvenance,
1633 pub(super) target_binding: FrontierBinding,
1634}
1635
1636#[derive(Debug, PartialEq, Eq)]
1641pub struct ValidatedMarkerRecord {
1642 conversation_id: ConversationId,
1643 record: RetainedCausalRecord,
1644 provenance: MarkerProvenance,
1645 target_binding: FrontierBinding,
1646 occurrence: MarkerRecordOccurrence,
1647 seal: MarkerAuthoritySeal,
1648}
1649
1650#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1657pub(super) enum MarkerRecordOccurrence {
1658 Undelivered,
1660 Delivered,
1662}
1663
1664#[derive(Debug, PartialEq, Eq)]
1665enum MarkerAuthoritySeal {
1666 Validated,
1667}
1668
1669#[doc(hidden)]
1675pub fn with_validated_marker_record_type<F>(probe: F)
1676where
1677 F: FnOnce(ValidatedMarkerRecord),
1678{
1679 drop(probe);
1680}
1681
1682impl ValidatedMarkerRecord {
1683 pub(super) const fn consume(self) {
1685 match self.seal {
1686 MarkerAuthoritySeal::Validated => {}
1687 }
1688 }
1689
1690 #[must_use]
1692 pub const fn conversation_id(&self) -> ConversationId {
1693 self.conversation_id
1694 }
1695
1696 #[must_use]
1698 pub const fn participant_id(&self) -> ParticipantId {
1699 self.record.admission_order.participant_index()
1700 }
1701
1702 #[must_use]
1704 pub const fn delivery_seq(&self) -> DeliverySeq {
1705 self.record.delivery_seq
1706 }
1707
1708 #[must_use]
1710 pub const fn admission_order(&self) -> super::AdmissionOrder {
1711 self.record.admission_order
1712 }
1713
1714 #[must_use]
1716 pub const fn provenance(&self) -> MarkerProvenance {
1717 self.provenance
1718 }
1719
1720 #[must_use]
1722 pub const fn target_binding(&self) -> FrontierBinding {
1723 self.target_binding
1724 }
1725
1726 #[must_use]
1728 pub const fn binding_epoch(&self) -> BindingEpoch {
1729 binding_epoch(self.target_binding)
1730 }
1731
1732 #[must_use]
1734 pub(super) const fn occurrence(&self) -> MarkerRecordOccurrence {
1735 self.occurrence
1736 }
1737
1738 #[cfg(any(test, feature = "test-support"))]
1740 pub(super) const fn delivered_for_test(mut self) -> Self {
1741 self.occurrence = MarkerRecordOccurrence::Delivered;
1742 self
1743 }
1744}
1745
1746impl ValidatedMarkerCandidate {
1747 pub(super) const fn consume(self) {
1749 match self.seal {
1750 MarkerAuthoritySeal::Validated => {}
1751 }
1752 }
1753
1754 #[must_use]
1756 pub(super) const fn conversation_id(&self) -> ConversationId {
1757 self.conversation_id
1758 }
1759
1760 #[must_use]
1762 pub(super) const fn participant_index(&self) -> ParticipantId {
1763 self.candidate.admission_order.participant_index()
1764 }
1765
1766 #[must_use]
1768 pub(super) const fn participant_id(&self) -> ParticipantId {
1769 self.participant_index()
1770 }
1771
1772 #[must_use]
1774 pub(super) const fn delivery_seq(&self) -> DeliverySeq {
1775 self.candidate.delivery_seq
1776 }
1777
1778 #[must_use]
1780 pub(super) const fn target_binding(&self) -> FrontierBinding {
1781 self.candidate.target_binding
1782 }
1783
1784 #[must_use]
1786 pub(super) const fn provenance(&self) -> MarkerProvenance {
1787 self.candidate.provenance
1788 }
1789
1790 pub(super) const fn abandoned_after(&self) -> DeliverySeq {
1791 self.candidate.abandoned_after
1792 }
1793
1794 pub(super) const fn abandoned_through(&self) -> DeliverySeq {
1795 self.candidate.abandoned_through
1796 }
1797
1798 pub(super) const fn physical_floor_at_decision(&self) -> DeliverySeq {
1799 self.candidate.physical_floor_at_decision
1800 }
1801}
1802
1803#[derive(Clone, Copy)]
1804struct InitialEnrollmentFrontierShape {
1805 conversation_id: ConversationId,
1806 binding_epoch: BindingEpoch,
1807 identity_slot_limit: u64,
1808 retained_floor: u128,
1809 attached: AttachedLifecycleRecord,
1810 order_ledger: OrderLedger,
1811 sequence_ledger: SequenceLedger,
1812 closure_accounting: ClosureAccounting,
1813}
1814
1815#[derive(Clone, Copy)]
1816struct InitialFrontierPositions {
1817 terminal_sequence: DeliverySeq,
1818 exit_sequence: DeliverySeq,
1819 product_sequence: DeliverySeq,
1820 active_terminal_order: TransactionOrder,
1821 exit_order: TransactionOrder,
1822}
1823
1824fn initial_enrollment_frontier_shape<F>(
1825 operation: &InitialEnrollmentOperationCommit<F>,
1826 attached_charge: ResourceVector,
1827) -> Result<InitialEnrollmentFrontierShape, InitialEnrollmentFrontierError> {
1828 let projection = operation.closure_projection();
1829 let expected_charge = projection.resulting_retained_charge();
1830 if attached_charge != expected_charge {
1831 return Err(InitialEnrollmentFrontierError::AttachedChargeMismatch {
1832 expected: expected_charge,
1833 actual: attached_charge,
1834 });
1835 }
1836 let enrollment = operation.enrollment();
1837 let attached = enrollment.attached;
1838 let BindingState::Bound(binding) = enrollment.binding_state else {
1839 return Err(InitialEnrollmentFrontierError::EnrollmentShape);
1840 };
1841 if enrollment.member.participant_id() != 0
1842 || enrollment.member.cursor() != 0
1843 || binding.participant_id != 0
1844 || attached.participant_id() != 0
1845 || binding.conversation_id != enrollment.member.conversation_id()
1846 || attached.conversation_id() != binding.conversation_id
1847 || attached.binding_epoch() != binding.binding_epoch
1848 || projection.participant_index() != 0
1849 || projection.binding_epoch() != binding.binding_epoch
1850 || projection.identity_slots() == 0
1851 {
1852 return Err(InitialEnrollmentFrontierError::EnrollmentShape);
1853 }
1854 let admission_order = attached.admission_order();
1855 let order_ledger = operation.order().resulting();
1856 let sequence_ledger = operation.sequence().resulting();
1857 let order_claims = order_ledger.claims();
1858 let sequence_claims = sequence_ledger.claims();
1859 if operation.order().major() != admission_order.transaction_order()
1860 || !matches!(order_ledger.high(), OrderHigh::Allocated(value) if value == admission_order.transaction_order())
1861 || order_claims.active_binding_terminals() != 1
1862 || order_claims.membership_exits() != 1
1863 || order_claims.recovery_operation()
1864 || order_claims.recovery_replacement_terminal()
1865 || sequence_ledger.high_watermark() != attached.delivery_seq()
1866 || sequence_claims.live_members() != 1
1867 || sequence_claims.binding_terminals() != 1
1868 || sequence_claims.markers() != 0
1869 || sequence_claims.recovery() != RecoverySequenceReserve::None
1870 || admission_order.candidate_phase() != CandidatePhase::AttachLifecycle
1871 {
1872 return Err(InitialEnrollmentFrontierError::LedgerShape);
1873 }
1874 Ok(InitialEnrollmentFrontierShape {
1875 conversation_id: binding.conversation_id,
1876 binding_epoch: binding.binding_epoch,
1877 identity_slot_limit: projection.identity_slots(),
1878 retained_floor: projection.resulting_floor(),
1879 attached,
1880 order_ledger,
1881 sequence_ledger,
1882 closure_accounting: validate_initial_enrollment_closure(operation, projection, attached)?,
1883 })
1884}
1885
1886fn validate_initial_enrollment_closure<F>(
1887 operation: &InitialEnrollmentOperationCommit<F>,
1888 projection: &InitialEnrollmentClosureProjection,
1889 attached: AttachedLifecycleRecord,
1890) -> Result<ClosureAccounting, InitialEnrollmentFrontierError> {
1891 let accounting = projection.resulting_closure_accounting();
1892 let state_matches = match accounting.state() {
1893 ClosureState::Clear => {
1894 projection.debt().is_zero()
1895 && projection.remaining_recovery_claim() == ResourceVector::default()
1896 }
1897 ClosureState::Owed {
1898 debt,
1899 edge: StoredEdge::ObserverProjection(observer),
1900 } => {
1901 debt.value() == projection.debt()
1902 && observer == ObserverProjection::new(attached.delivery_seq())
1903 && projection.remaining_recovery_claim() == accounting.edge_k_remaining()
1904 }
1905 ClosureState::Owed { .. } => false,
1906 };
1907 if projection.resulting_floor() != 1
1908 || projection.resulting_floor() != operation.observer_floor().cap_floor()
1909 || operation.observer_floor().observer_progress() != 0
1910 || projection.recovery_quartet() != RecoveryQuartetStatus::None
1911 || !projection.new_marker_candidates().is_empty()
1912 || accounting.marker_capacity_credits() != 0
1913 || accounting.marker_anchors() != 0
1914 || accounting.edge_sequence_claims() != 0
1915 || accounting.edge_order_position_claims() != 0
1916 || accounting.baseline() != projection.resulting_baseline()
1917 || !state_matches
1918 {
1919 Err(InitialEnrollmentFrontierError::ClosureProjection)
1920 } else {
1921 Ok(accounting)
1922 }
1923}
1924
1925fn initial_frontier_positions(
1926 attached: AttachedLifecycleRecord,
1927) -> Result<InitialFrontierPositions, InitialEnrollmentFrontierError> {
1928 let terminal_sequence = attached
1929 .delivery_seq()
1930 .checked_add(1)
1931 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1932 let exit_sequence = terminal_sequence
1933 .checked_add(1)
1934 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1935 let product_sequence = exit_sequence
1936 .checked_add(1)
1937 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1938 let active_terminal_order = attached
1939 .admission_order()
1940 .transaction_order()
1941 .checked_add(1)
1942 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1943 let exit_order = active_terminal_order
1944 .checked_add(1)
1945 .ok_or(InitialEnrollmentFrontierError::PositionOverflow)?;
1946 Ok(InitialFrontierPositions {
1947 terminal_sequence,
1948 exit_sequence,
1949 product_sequence,
1950 active_terminal_order,
1951 exit_order,
1952 })
1953}
1954
1955fn initial_sequence_frontier(
1956 shape: &InitialEnrollmentFrontierShape,
1957 positions: InitialFrontierPositions,
1958 terminal: BindingTerminalOwner,
1959) -> SequenceClaimFrontier {
1960 SequenceClaimFrontier {
1961 ledger: shape.sequence_ledger,
1962 movable_claims: alloc::vec![
1963 MovableSequenceClaim {
1964 delivery_seq: positions.terminal_sequence,
1965 owner: SequenceDirectOwner::BindingTerminal(terminal),
1966 },
1967 MovableSequenceClaim {
1968 delivery_seq: positions.exit_sequence,
1969 owner: SequenceDirectOwner::MembershipExit {
1970 participant_index: 0,
1971 },
1972 },
1973 ],
1974 immutable_candidates: Vec::new(),
1975 products: SequenceProductRanges {
1976 live_times_terminal: alloc::vec![TerminalProductRange {
1977 start: positions.product_sequence,
1978 length: 1,
1979 terminal,
1980 }],
1981 live_times_replacement_terminal: None,
1982 other_live_times_exit: Vec::new(),
1983 },
1984 recovery: None,
1985 }
1986}
1987
1988fn initial_order_frontier(
1989 shape: &InitialEnrollmentFrontierShape,
1990 positions: InitialFrontierPositions,
1991 terminal: BindingTerminalOwner,
1992) -> OrderClaimFrontier {
1993 OrderClaimFrontier {
1994 ledger: shape.order_ledger,
1995 movable_claims: alloc::vec![
1996 MovableOrderClaim {
1997 transaction_order: positions.active_terminal_order,
1998 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
1999 },
2000 MovableOrderClaim {
2001 transaction_order: positions.exit_order,
2002 owner: OrderDirectOwner::MembershipExit {
2003 participant_index: 0,
2004 },
2005 },
2006 ],
2007 immutable_candidates: Vec::new(),
2008 recovery: None,
2009 }
2010}
2011
2012fn build_initial_enrollment_frontiers(
2013 shape: &InitialEnrollmentFrontierShape,
2014) -> Result<ClaimFrontiers, InitialEnrollmentFrontierError> {
2015 let positions = initial_frontier_positions(shape.attached)?;
2016 let terminal = BindingTerminalOwner {
2017 participant_index: 0,
2018 binding_epoch: shape.binding_epoch,
2019 };
2020 let sequence = initial_sequence_frontier(shape, positions, terminal);
2021 let order = initial_order_frontier(shape, positions, terminal);
2022 validate_cross_counter(&sequence, &order)
2023 .map_err(|_| InitialEnrollmentFrontierError::FrontierInvariant)?;
2024 Ok(ClaimFrontiers {
2025 ordinary_cursor_release: None,
2026 historical_marker_deliveries: alloc::vec::Vec::new(),
2028 historical_causal_authorities: alloc::vec::Vec::new(),
2030 conversation_id: shape.conversation_id,
2031 active_identities: ActiveIdentityRanks {
2032 participants: alloc::vec![FrontierParticipant::new(
2033 0,
2034 0,
2035 FrontierBinding::Bound(shape.binding_epoch),
2036 )],
2037 },
2038 identity_slot_limit: shape.identity_slot_limit,
2039 retained_floor: shape.retained_floor,
2040 retained_records: alloc::vec![RetainedCausalRecord {
2041 delivery_seq: shape.attached.delivery_seq(),
2042 admission_order: shape.attached.admission_order(),
2043 kind: RetainedCausalRecordKind::AttachLifecycle {
2044 participant_index: 0,
2045 binding_epoch: shape.binding_epoch,
2046 },
2047 }],
2048 marker_records: Vec::new(),
2049 fenced_marker_issued: None,
2050 sequence,
2051 order,
2052 })
2053}
2054
2055#[derive(Clone, Copy)]
2056enum LeaveSequenceUnit {
2057 Direct(MovableSequenceClaim),
2058 TerminalProduct(TerminalProductRange),
2059 ReplacementProduct(ReplacementTerminalProductRange),
2060 ExitProduct(ExitProductRange),
2061 Recovery(RecoverySequenceBlock),
2062}
2063
2064impl LeaveSequenceUnit {
2065 fn original_start(self) -> DeliverySeq {
2066 match self {
2067 Self::Direct(claim) => claim.delivery_seq,
2068 Self::TerminalProduct(range) => range.start,
2069 Self::ReplacementProduct(range) => range.start,
2070 Self::ExitProduct(range) => range.start,
2071 Self::Recovery(block) => block_start_validated_sequence(block),
2072 }
2073 }
2074}
2075
2076fn allocate_leave_sequence_range(
2077 cursor: &mut Option<DeliverySeq>,
2078 length: u64,
2079) -> Result<DeliverySeq, LeaveCommitError> {
2080 let Some(start) = *cursor else {
2081 return Err(LeaveCommitError::ResultingFrontier);
2082 };
2083 let end = u128::from(start)
2084 .checked_add(u128::from(length))
2085 .and_then(|value| value.checked_sub(1))
2086 .ok_or(LeaveCommitError::ResultingFrontier)?;
2087 if end > u128::from(u64::MAX) {
2088 return Err(LeaveCommitError::ResultingFrontier);
2089 }
2090 *cursor = u64::try_from(end + 1).ok();
2091 Ok(start)
2092}
2093
2094#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2108pub enum PrecedenceCondition {
2109 BindingTerminal,
2113 MarkerDrain {
2115 settlement_epoch: DeliverySeq,
2123 },
2124 FencedRecovery,
2130 Unclassified,
2134}
2135
2136#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2138pub(in crate::lifecycle) enum LiveFrontierTransitionError {
2139 Authority,
2141 Precedence(PrecedenceCondition),
2146 RecordPosition,
2148 Exhausted,
2150 ResultingFrontier,
2152}
2153
2154fn precedence(
2156 frontiers: ClaimFrontiers,
2157 condition: PrecedenceCondition,
2158) -> Box<(ClaimFrontiers, LiveFrontierTransitionError)> {
2159 Box::new((
2160 frontiers,
2161 LiveFrontierTransitionError::Precedence(condition),
2162 ))
2163}
2164
2165fn select_retained_marker_records(records: &[RetainedCausalRecord]) -> Vec<RetainedCausalRecord> {
2166 records
2167 .iter()
2168 .copied()
2169 .filter(|record| {
2170 matches!(
2171 record.kind,
2172 RetainedCausalRecordKind::CompactionMarker { .. }
2173 )
2174 })
2175 .collect()
2176}
2177
2178impl ClaimFrontiers {
2179 pub(super) fn retain_ordinary_cursor_release(
2180 &mut self,
2181 source: Box<super::storage::OrdinaryBindingFateRestore>,
2182 ) {
2183 self.ordinary_cursor_release = Some(source);
2184 }
2185
2186 pub(super) fn ordinary_cursor_release(
2187 &self,
2188 ) -> Option<&super::storage::OrdinaryBindingFateRestore> {
2189 self.ordinary_cursor_release.as_deref()
2190 }
2191
2192 pub(super) fn historical_marker_deliveries(&self) -> &[HistoricalMarkerDeliveryAuthority] {
2198 &self.historical_marker_deliveries
2199 }
2200
2201 #[must_use]
2214 pub(in crate::lifecycle) fn describe(
2215 &self,
2216 retained_record_limit: u64,
2217 ) -> ClaimFrontiersDescription {
2218 let conversation_id = self.conversation_id;
2219 ClaimFrontiersDescription {
2220 restore: ClaimFrontiersRestore {
2221 conversation_id,
2222 active_identities: self.active_identities.participants().to_vec(),
2223 identity_slot_limit: self.identity_slot_limit,
2224 retained_floor: self.retained_floor,
2225 retained_record_limit,
2226 retained_records: self.retained_records.clone(),
2227 active_marker_anchors: self
2228 .marker_records
2229 .iter()
2230 .map(|record| record.delivery_seq)
2231 .collect(),
2232 historical_marker_deliveries: self
2233 .historical_marker_deliveries()
2234 .iter()
2235 .map(|authority| HistoricalMarkerDeliveryFactRestore {
2236 conversation_id,
2237 participant_index: authority.participant_index(),
2238 marker_delivery_seq: authority.marker_delivery_seq(),
2239 delivered_binding_epoch: authority.delivered_binding_epoch(),
2240 })
2241 .collect(),
2242 historical_causal_facts: self
2243 .historical_causal_authorities
2244 .iter()
2245 .copied()
2246 .map(HistoricalCausalAuthority::describe)
2247 .collect(),
2248 sequence: describe_sequence_frontier(&self.sequence),
2249 order: describe_order_frontier(&self.order),
2250 recovery_marker_delivery_seq: self
2251 .sequence
2252 .recovery
2253 .map(RecoverySequenceBlock::marker_delivery_seq),
2254 },
2255 sequence_ledger: self.sequence.ledger,
2256 order_ledger: self.order.ledger,
2257 fenced_marker_issued: self.fenced_marker_issued,
2258 }
2259 }
2260
2261 pub(in crate::lifecycle) fn prevalidate_description(
2283 description: ClaimFrontiersDescription,
2284 history: &ValidatedConversationHistory,
2285 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2286 let ClaimFrontiersDescription {
2287 restore,
2288 sequence_ledger,
2289 order_ledger,
2290 fenced_marker_issued,
2291 } = description;
2292 let mut prevalidated =
2293 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, history)?;
2294 prevalidated.fenced_marker_issued = fenced_marker_issued;
2295 Ok(prevalidated)
2296 }
2297
2298 pub fn from_initial_enrollment<F>(
2313 operation: InitialEnrollmentOperationCommit<F>,
2314 attached_charge: ResourceVector,
2315 ) -> Result<InitialEnrollmentFrontierCommit<F>, Box<InitialEnrollmentFrontierFailure<F>>> {
2316 let shape = match initial_enrollment_frontier_shape(&operation, attached_charge) {
2317 Ok(shape) => shape,
2318 Err(error) => {
2319 return Err(Box::new(InitialEnrollmentFrontierFailure {
2320 operation,
2321 error,
2322 }));
2323 }
2324 };
2325 let closure_accounting = shape.closure_accounting;
2326 let frontiers = match build_initial_enrollment_frontiers(&shape) {
2327 Ok(frontiers) => frontiers,
2328 Err(error) => {
2329 return Err(Box::new(InitialEnrollmentFrontierFailure {
2330 operation,
2331 error,
2332 }));
2333 }
2334 };
2335 Ok(InitialEnrollmentFrontierCommit {
2336 operation,
2337 frontiers,
2338 closure_accounting,
2339 attached_charge,
2340 })
2341 }
2342
2343 pub fn restore(
2356 restore: ClaimFrontiersRestore,
2357 sequence_ledger: SequenceLedger,
2358 order_ledger: OrderLedger,
2359 ) -> Result<Self, ParticipantStateCorruptReason> {
2360 let history = ValidatedConversationHistory::empty();
2361 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)?
2362 .finish(None)
2363 }
2364
2365 #[cfg(any(test, feature = "test-support"))]
2368 pub(super) fn prevalidate(
2369 restore: ClaimFrontiersRestore,
2370 sequence_ledger: SequenceLedger,
2371 order_ledger: OrderLedger,
2372 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2373 let history = ValidatedConversationHistory::empty();
2374 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)
2375 }
2376
2377 pub(super) fn prevalidate_with_history(
2378 restore: ClaimFrontiersRestore,
2379 sequence_ledger: SequenceLedger,
2380 order_ledger: OrderLedger,
2381 history: &ValidatedConversationHistory,
2382 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2383 validate_sequence_numeric(&restore.sequence, sequence_ledger).map_err(corrupt_frontier)?;
2384 validate_order_numeric(&restore.order, order_ledger).map_err(corrupt_frontier)?;
2385 validate_unique_candidate_keys(
2386 &restore.sequence.immutable_candidates,
2387 &restore.retained_records,
2388 )?;
2389 validate_bounded_shape(&restore, sequence_ledger).map_err(corrupt_frontier)?;
2390 let active_identities = ActiveIdentityRanks::try_new(
2391 restore.active_identities,
2392 sequence_ledger.high_watermark(),
2393 restore.identity_slot_limit,
2394 )
2395 .map_err(corrupt_frontier)?;
2396 let retained_records = validated_retained_records(
2397 restore.retained_records,
2398 restore.retained_floor,
2399 restore.retained_record_limit,
2400 restore.identity_slot_limit,
2401 sequence_ledger,
2402 )
2403 .map_err(corrupt_frontier)?;
2404 let historical_causal_authorities = validated_historical_authorities(
2405 restore.historical_causal_facts,
2406 restore.conversation_id,
2407 restore.identity_slot_limit,
2408 sequence_ledger,
2409 history,
2410 )
2411 .map_err(corrupt_frontier)?;
2412 let retained_marker_records = select_retained_marker_records(&retained_records);
2413 let marker_records = validated_active_marker_records(
2414 &retained_marker_records,
2415 restore.active_marker_anchors,
2416 restore.identity_slot_limit,
2417 sequence_ledger,
2418 )
2419 .map_err(corrupt_frontier)?;
2420 let historical_marker_deliveries = validated_historical_marker_deliveries(
2421 restore.historical_marker_deliveries,
2422 restore.conversation_id,
2423 &active_identities,
2424 &MarkerDeliverySources {
2425 retained: &retained_records,
2426 historical: &historical_causal_authorities,
2427 candidates: &restore.sequence.immutable_candidates,
2428 },
2429 restore.retained_record_limit,
2430 sequence_ledger,
2431 )
2432 .map_err(corrupt_frontier)?;
2433 BindingOriginValidation {
2434 conversation_id: restore.conversation_id,
2435 active: &active_identities,
2436 origins: &history.binding_origins,
2437 retained_records: &retained_records,
2438 causal_authorities: &history.causal_authorities,
2439 historical_marker_deliveries: &historical_marker_deliveries,
2440 total: history.total,
2441 ledger: sequence_ledger,
2442 }
2443 .validate()
2444 .map_err(corrupt_frontier)?;
2445 validate_sequence_candidates(
2446 &active_identities,
2447 &restore.sequence.immutable_candidates,
2448 restore.retained_floor,
2449 &retained_records,
2450 &historical_causal_authorities,
2451 sequence_ledger,
2452 )
2453 .map_err(corrupt_frontier)?;
2454 validate_marker_credit_owners(
2455 &restore.sequence.immutable_candidates,
2456 &marker_records,
2457 restore.identity_slot_limit,
2458 sequence_ledger,
2459 )
2460 .map_err(corrupt_frontier)?;
2461 Ok(ClaimFrontiersPrevalidated {
2462 conversation_id: restore.conversation_id,
2463 active_identities,
2464 identity_slot_limit: restore.identity_slot_limit,
2465 retained_floor: restore.retained_floor,
2466 retained_records,
2467 marker_records,
2468 historical_marker_deliveries,
2469 historical_causal_authorities,
2470 binding_origins: history.binding_origins.clone(),
2471 sequence_restore: restore.sequence,
2472 order_restore: restore.order,
2473 recovery_marker_delivery_seq: restore.recovery_marker_delivery_seq,
2474 sequence_ledger,
2475 order_ledger,
2476 issued_marker_record: None,
2477 fenced_marker_issued: None,
2478 })
2479 }
2480
2481 #[must_use]
2483 pub const fn active_identities(&self) -> &ActiveIdentityRanks {
2484 &self.active_identities
2485 }
2486
2487 #[must_use]
2489 pub const fn identity_slot_limit(&self) -> u64 {
2490 self.identity_slot_limit
2491 }
2492
2493 #[must_use]
2495 pub const fn conversation_id(&self) -> ConversationId {
2496 self.conversation_id
2497 }
2498
2499 #[must_use]
2501 pub const fn retained_floor(&self) -> u128 {
2502 self.retained_floor
2503 }
2504
2505 #[must_use]
2507 pub fn retained_records(&self) -> &[RetainedCausalRecord] {
2508 &self.retained_records
2509 }
2510
2511 pub(super) fn checkpoint_recovered_marker_record(
2512 &self,
2513 origin: BindingOrigin,
2514 ) -> Option<ValidatedMarkerRecord> {
2515 let (marker_delivery_seq, prior_binding_epoch) = origin.recovered_marker()?;
2516 if origin.conversation_id() != self.conversation_id
2517 || !self.historical_marker_deliveries.iter().any(|delivery| {
2518 delivery.participant_index == origin.participant_id()
2519 && delivery.marker_delivery_seq == marker_delivery_seq
2520 && delivery.delivered_binding_epoch == prior_binding_epoch
2521 })
2522 {
2523 return None;
2524 }
2525 let record = self
2526 .retained_records
2527 .iter()
2528 .find(|record| record.delivery_seq == marker_delivery_seq)
2529 .copied()?;
2530 let RetainedCausalRecordKind::CompactionMarker {
2531 participant_index,
2532 provenance,
2533 } = record.kind
2534 else {
2535 return None;
2536 };
2537 if participant_index != origin.participant_id() {
2538 return None;
2539 }
2540 Some(ValidatedMarkerRecord {
2541 conversation_id: self.conversation_id,
2542 record,
2543 provenance,
2544 target_binding: FrontierBinding::Detached(prior_binding_epoch),
2545 occurrence: MarkerRecordOccurrence::Delivered,
2546 seal: MarkerAuthoritySeal::Validated,
2547 })
2548 }
2549
2550 #[must_use]
2552 pub fn retained_marker_records(&self) -> &[RetainedCausalRecord] {
2553 &self.marker_records
2554 }
2555
2556 #[must_use]
2561 pub(in crate::lifecycle) fn unaccepted_marker_anchor_count(&self) -> u64 {
2562 ordinary_unaccepted_marker_anchors(self).len() as u64
2563 }
2564
2565 #[must_use]
2572 pub(in crate::lifecycle) fn pending_marker_candidate_count(&self) -> u64 {
2573 self.sequence
2574 .immutable_candidates
2575 .iter()
2576 .filter(|candidate| matches!(candidate, ImmutableSequenceCandidate::Marker(_)))
2577 .count() as u64
2578 }
2579
2580 #[must_use]
2588 pub fn project_offered_marker_progress(
2589 &self,
2590 participant_id: ParticipantId,
2591 binding_epoch: BindingEpoch,
2592 marker_delivery_seq: DeliverySeq,
2593 event: Event,
2594 ) -> Option<ParticipantCursorProgress> {
2595 let record = self
2596 .marker_records
2597 .iter()
2598 .find(|record| record.delivery_seq == marker_delivery_seq)
2599 .copied()?;
2600 let RetainedCausalRecordKind::CompactionMarker {
2601 participant_index,
2602 provenance,
2603 } = record.kind
2604 else {
2605 return None;
2606 };
2607 if participant_index != participant_id {
2608 return None;
2609 }
2610 let participant = active_participant(&self.active_identities, participant_id)?;
2611 let target_binding = FrontierBinding::Bound(binding_epoch);
2612 if participant.binding != target_binding {
2613 return None;
2614 }
2615 let authority = ValidatedMarkerRecord {
2616 conversation_id: self.conversation_id,
2617 record,
2618 provenance,
2619 target_binding,
2620 occurrence: MarkerRecordOccurrence::Undelivered,
2621 seal: MarkerAuthoritySeal::Validated,
2622 };
2623 MarkerDelivery::from_validated_record(&authority)
2624 .delivered_progress(event)
2625 .ok()
2626 }
2627
2628 pub(super) fn fenced_marker_source(
2631 &self,
2632 recovery: super::DetachedCredentialRecovery,
2633 ) -> Option<FencedMarkerSourceRecord> {
2634 if self.fenced_marker_issued.is_some() {
2635 return None;
2636 }
2637 self.unlatched_fenced_marker_source(recovery)
2638 }
2639
2640 #[must_use]
2657 pub(in crate::lifecycle) fn fenced_latch_matches(
2658 &self,
2659 edge: Option<super::StoredEdge>,
2660 ) -> bool {
2661 let Some((participant_id, marker_delivery_seq, prior_binding_epoch)) =
2662 self.fenced_marker_issued
2663 else {
2664 return true;
2665 };
2666 let Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) = edge else {
2667 return false;
2668 };
2669 recovery.participant_id() == participant_id
2670 && recovery.marker_delivery_seq() == marker_delivery_seq
2671 && recovery.prior_binding_epoch() == prior_binding_epoch
2672 && self.unlatched_fenced_marker_source(recovery).is_some()
2673 }
2674
2675 fn unlatched_fenced_marker_source(
2678 &self,
2679 recovery: super::DetachedCredentialRecovery,
2680 ) -> Option<FencedMarkerSourceRecord> {
2681 if recovery.conversation_id() != self.conversation_id {
2682 return None;
2683 }
2684 let marker_delivery_seq = recovery.marker_delivery_seq();
2685 let participant_id = recovery.participant_id();
2686 let prior_binding_epoch = recovery.prior_binding_epoch();
2687 let record = self
2688 .marker_records
2689 .iter()
2690 .find(|record| record.delivery_seq == marker_delivery_seq)
2691 .copied()?;
2692 let RetainedCausalRecordKind::CompactionMarker {
2693 participant_index,
2694 provenance,
2695 } = record.kind
2696 else {
2697 return None;
2698 };
2699 let participant = active_participant(&self.active_identities, participant_id)?;
2700 let target_binding = FrontierBinding::Detached(prior_binding_epoch);
2701 if participant_index != participant_id || participant.binding != target_binding {
2702 return None;
2703 }
2704 Some(FencedMarkerSourceRecord {
2705 conversation_id: self.conversation_id,
2706 delivery_seq: record.delivery_seq,
2707 admission_order: record.admission_order,
2708 participant_id,
2709 provenance,
2710 target_binding,
2711 })
2712 }
2713
2714 pub(in crate::lifecycle) fn take_fenced_marker_record(
2725 &mut self,
2726 recovery: super::DetachedCredentialRecovery,
2727 ) -> Option<ValidatedMarkerRecord> {
2728 let source = self.fenced_marker_source(recovery)?;
2729 self.fenced_marker_issued = Some((
2730 source.participant_id,
2731 source.delivery_seq,
2732 recovery.prior_binding_epoch(),
2733 ));
2734 Some(ValidatedMarkerRecord {
2735 conversation_id: source.conversation_id,
2736 record: RetainedCausalRecord {
2737 delivery_seq: source.delivery_seq,
2738 admission_order: source.admission_order,
2739 kind: RetainedCausalRecordKind::CompactionMarker {
2740 participant_index: source.participant_id,
2741 provenance: source.provenance,
2742 },
2743 },
2744 provenance: source.provenance,
2745 target_binding: source.target_binding,
2746 occurrence: MarkerRecordOccurrence::Delivered,
2747 seal: MarkerAuthoritySeal::Validated,
2748 })
2749 }
2750
2751 pub(in crate::lifecycle) const fn reinstall_fenced_marker_record(
2754 &mut self,
2755 record: ValidatedMarkerRecord,
2756 ) {
2757 self.fenced_marker_issued = None;
2758 record.consume();
2759 }
2760
2761 #[must_use]
2763 pub const fn sequence(&self) -> &SequenceClaimFrontier {
2764 &self.sequence
2765 }
2766
2767 #[must_use]
2769 pub const fn order(&self) -> &OrderClaimFrontier {
2770 &self.order
2771 }
2772
2773 #[cfg(test)]
2774 pub(in crate::lifecycle) fn cross_counter_valid_for_test(&self) -> bool {
2775 validate_cross_counter(&self.sequence, &self.order).is_ok()
2776 }
2777
2778 #[must_use]
2786 pub fn live_transition_blocked(&self) -> bool {
2787 !self.sequence.immutable_candidates.is_empty()
2788 || self.sequence.recovery.is_some()
2789 || !self.order.immutable_candidates.is_empty()
2790 || self.order.recovery.is_some()
2791 }
2792
2793 #[must_use]
2810 pub fn precedence_condition(&self) -> PrecedenceCondition {
2811 if self.sequence.recovery.is_some() || self.order.recovery.is_some() {
2812 return PrecedenceCondition::FencedRecovery;
2813 }
2814 match self.sequence.immutable_candidates.first() {
2815 Some(ImmutableSequenceCandidate::Marker(marker)) => PrecedenceCondition::MarkerDrain {
2816 settlement_epoch: marker.delivery_seq,
2817 },
2818 Some(ImmutableSequenceCandidate::BindingTerminal { .. }) => {
2819 PrecedenceCondition::BindingTerminal
2820 }
2821 None => PrecedenceCondition::Unclassified,
2822 }
2823 }
2824
2825 pub(in crate::lifecycle) fn apply_live_transition(
2831 self,
2832 active_identities: Vec<FrontierParticipant>,
2833 appended_records: &[RetainedCausalRecord],
2834 sequence_ledger: SequenceLedger,
2835 order_ledger: OrderLedger,
2836 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2837 if self.live_transition_blocked() {
2838 let condition = self.precedence_condition();
2839 return Err(Box::new((
2840 self,
2841 LiveFrontierTransitionError::Precedence(condition),
2842 )));
2843 }
2844 let Ok(active) = ActiveIdentityRanks::try_new(
2845 active_identities,
2846 sequence_ledger.high_watermark(),
2847 self.identity_slot_limit,
2848 ) else {
2849 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2850 };
2851 let first_sequence = self.sequence.ledger.high_watermark().checked_add(1);
2852 if first_sequence.is_none_or(|first| {
2853 appended_records.iter().enumerate().any(|(index, record)| {
2854 u64::try_from(index)
2855 .ok()
2856 .and_then(|offset| first.checked_add(offset))
2857 != Some(record.delivery_seq)
2858 })
2859 }) || appended_records
2860 .last()
2861 .is_some_and(|record| record.delivery_seq != sequence_ledger.high_watermark())
2862 {
2863 return Err(Box::new((
2864 self,
2865 LiveFrontierTransitionError::RecordPosition,
2866 )));
2867 }
2868 let (sequence, order) =
2869 match rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger) {
2870 Ok(frontiers) => frontiers,
2871 Err(error) => return Err(Box::new((self, error))),
2872 };
2873 let Self {
2874 ordinary_cursor_release,
2875 conversation_id,
2876 identity_slot_limit,
2877 retained_floor,
2878 mut retained_records,
2879 marker_records,
2880 fenced_marker_issued,
2881 historical_marker_deliveries,
2882 historical_causal_authorities,
2883 ..
2884 } = self;
2885 retained_records.extend_from_slice(appended_records);
2886 Ok(Self {
2887 ordinary_cursor_release,
2888 conversation_id,
2889 active_identities: active,
2890 identity_slot_limit,
2891 retained_floor,
2892 retained_records,
2893 marker_records,
2894 fenced_marker_issued,
2895 sequence,
2896 order,
2897 historical_marker_deliveries,
2898 historical_causal_authorities,
2899 })
2900 }
2901
2902 pub(in crate::lifecycle) fn apply_pending_binding_terminal(
2906 mut self,
2907 participant_id: ParticipantId,
2908 binding_epoch: BindingEpoch,
2909 delivery_seq: DeliverySeq,
2910 admission_order: super::AdmissionOrder,
2911 order_ledger: OrderLedger,
2912 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2913 if self.live_transition_blocked() {
2914 let condition = self.precedence_condition();
2915 return Err(Box::new((
2916 self,
2917 LiveFrontierTransitionError::Precedence(condition),
2918 )));
2919 }
2920 let mut participants = self.active_identities.participants().to_vec();
2921 let Some(participant) = participants
2922 .iter_mut()
2923 .find(|participant| participant.participant_index() == participant_id)
2924 else {
2925 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2926 };
2927 if participant.binding() != FrontierBinding::Bound(binding_epoch) {
2928 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2929 }
2930 *participant = FrontierParticipant::new(
2931 participant_id,
2932 participant.cursor(),
2933 FrontierBinding::Detached(binding_epoch),
2934 );
2935 let Ok(active) = ActiveIdentityRanks::try_new(
2936 participants,
2937 self.sequence.ledger.high_watermark(),
2938 self.identity_slot_limit,
2939 ) else {
2940 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2941 };
2942 let pending_owner = BindingTerminalOwner {
2943 participant_index: participant_id,
2944 binding_epoch,
2945 };
2946 let (sequence, order) = match rebuild_pending_terminal_frontiers(
2947 &active,
2948 pending_owner,
2949 delivery_seq,
2950 admission_order,
2951 self.sequence.ledger,
2952 order_ledger,
2953 ) {
2954 Ok(frontiers) => frontiers,
2955 Err(error) => return Err(Box::new((self, error))),
2956 };
2957 self.active_identities = active;
2958 self.sequence = sequence;
2959 self.order = order;
2960 Ok(self)
2961 }
2962
2963 pub(in crate::lifecycle) fn apply_live_fenced_attach(
2971 self,
2972 participant: FrontierParticipant,
2973 prior_binding_epoch: BindingEpoch,
2974 appended_records: &[RetainedCausalRecord],
2975 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2976 let Some(current) = self
2977 .active_identities
2978 .participants()
2979 .iter()
2980 .find(|current| current.participant_index() == participant.participant_index())
2981 .copied()
2982 else {
2983 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2984 };
2985 let Some(sequence_recovery) = self.sequence.recovery else {
2986 return Err(precedence(self, PrecedenceCondition::FencedRecovery));
2987 };
2988 let Some(order_recovery) = self.order.recovery else {
2989 return Err(precedence(self, PrecedenceCondition::FencedRecovery));
2990 };
2991 if !Self::fenced_recovery_authority_matches(
2992 current,
2993 participant,
2994 prior_binding_epoch,
2995 sequence_recovery,
2996 order_recovery,
2997 ) {
2998 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2999 }
3000 let finalizes_pending = appended_records.len() == 2;
3001 let pending_terminal_matches = self.pending_fenced_terminal_matches(appended_records);
3002 let candidates_empty = self.sequence.immutable_candidates.is_empty()
3003 && self.order.immutable_candidates.is_empty();
3004 if !(candidates_empty || finalizes_pending && pending_terminal_matches) {
3005 let condition = self.precedence_condition();
3006 return Err(precedence(self, condition));
3007 }
3008 if !Self::fenced_records_match(
3009 participant,
3010 sequence_recovery,
3011 order_recovery,
3012 appended_records,
3013 pending_terminal_matches,
3014 ) {
3015 return Err(Box::new((
3016 self,
3017 LiveFrontierTransitionError::RecordPosition,
3018 )));
3019 }
3020 if !self.has_recovery_marker(participant) {
3021 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3022 }
3023 if self.has_duplicate_appended_record(appended_records) {
3024 return Err(Box::new((
3025 self,
3026 LiveFrontierTransitionError::RecordPosition,
3027 )));
3028 }
3029 let Some((sequence_ledger, order_ledger)) =
3030 apply_fenced_ledgers(self.sequence.ledger, self.order.ledger, finalizes_pending)
3031 else {
3032 return Err(Box::new((
3033 self,
3034 LiveFrontierTransitionError::ResultingFrontier,
3035 )));
3036 };
3037 let mut active = self.active_identities.participants().to_vec();
3038 let Some(current) = active
3039 .iter_mut()
3040 .find(|current| current.participant_index() == participant.participant_index())
3041 else {
3042 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3043 };
3044 *current = participant;
3045 let Ok(active) = ActiveIdentityRanks::try_new(
3046 active,
3047 sequence_ledger.high_watermark(),
3048 self.identity_slot_limit,
3049 ) else {
3050 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3051 };
3052 let Ok((sequence, order)) =
3053 rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger)
3054 else {
3055 return Err(Box::new((
3056 self,
3057 LiveFrontierTransitionError::ResultingFrontier,
3058 )));
3059 };
3060 let mut resulting = self;
3061 resulting.active_identities = active;
3062 resulting
3063 .retained_records
3064 .extend_from_slice(appended_records);
3065 resulting
3066 .retained_records
3067 .sort_unstable_by_key(|record| record.delivery_seq);
3068 resulting
3069 .marker_records
3070 .retain(|record| record.delivery_seq != participant.cursor());
3071 resulting.spend_fenced_latch(participant, prior_binding_epoch);
3072 resulting.sequence = sequence;
3073 resulting.order = order;
3074 Ok(resulting)
3075 }
3076
3077 fn spend_fenced_latch(
3082 &mut self,
3083 participant: FrontierParticipant,
3084 prior_binding_epoch: BindingEpoch,
3085 ) {
3086 if self.fenced_marker_issued
3087 == Some((
3088 participant.participant_index(),
3089 participant.cursor(),
3090 prior_binding_epoch,
3091 ))
3092 {
3093 self.fenced_marker_issued = None;
3094 }
3095 }
3096
3097 fn fenced_recovery_authority_matches(
3098 current: FrontierParticipant,
3099 participant: FrontierParticipant,
3100 prior_binding_epoch: BindingEpoch,
3101 sequence_recovery: RecoverySequenceBlock,
3102 order_recovery: RecoveryOrderBlock,
3103 ) -> bool {
3104 let participant_matches = sequence_recovery.participant_index()
3105 == participant.participant_index()
3106 && order_recovery.participant_index() == participant.participant_index();
3107 let marker_matches = sequence_recovery.marker_delivery_seq() == participant.cursor()
3108 && order_recovery.marker_delivery_seq() == participant.cursor();
3109 let prior_epoch_matches = sequence_recovery.recovered_binding_epoch()
3110 == prior_binding_epoch
3111 && order_recovery.recovered_binding_epoch() == prior_binding_epoch;
3112 participant_matches
3113 && marker_matches
3114 && prior_epoch_matches
3115 && current.binding() == FrontierBinding::Detached(prior_binding_epoch)
3116 && current.cursor() <= participant.cursor()
3117 && matches!(participant.binding(), FrontierBinding::Bound(_))
3118 }
3119
3120 fn fenced_records_match(
3121 participant: FrontierParticipant,
3122 sequence_recovery: RecoverySequenceBlock,
3123 order_recovery: RecoveryOrderBlock,
3124 appended_records: &[RetainedCausalRecord],
3125 pending_terminal_matches: bool,
3126 ) -> bool {
3127 let Some(attached) = appended_records.last().copied() else {
3128 return false;
3129 };
3130 let FrontierBinding::Bound(recovered_binding_epoch) = participant.binding() else {
3131 return false;
3132 };
3133 let attached_matches = attached.delivery_seq == sequence_recovery.recovery_attach_seq()
3134 && attached.admission_order.transaction_order()
3135 == order_recovery.recovery_operation_order()
3136 && attached.kind
3137 == (RetainedCausalRecordKind::AttachLifecycle {
3138 participant_index: participant.participant_index(),
3139 binding_epoch: recovered_binding_epoch,
3140 });
3141 if !attached_matches {
3142 return false;
3143 }
3144 let prefix = &appended_records[..appended_records.len() - 1];
3145 match (
3146 sequence_recovery.terminal(),
3147 order_recovery.active_binding(),
3148 prefix,
3149 ) {
3150 (None, None, []) => true,
3151 (None, None, [_]) => pending_terminal_matches,
3152 (Some(sequence_terminal), Some(order_terminal), [terminal]) => {
3153 sequence_terminal.owner == order_terminal.owner
3154 && terminal.delivery_seq == sequence_terminal.delivery_seq
3155 && terminal.admission_order.transaction_order()
3156 == order_terminal.transaction_order
3157 && terminal.kind
3158 == RetainedCausalRecordKind::BindingTerminal(sequence_terminal.owner)
3159 }
3160 _ => false,
3161 }
3162 }
3163
3164 fn pending_fenced_terminal_matches(&self, appended_records: &[RetainedCausalRecord]) -> bool {
3165 let [terminal, _] = appended_records else {
3166 return false;
3167 };
3168 let RetainedCausalRecordKind::BindingTerminal(owner) = terminal.kind else {
3169 return false;
3170 };
3171 let sequence_matches = self.sequence.immutable_candidates.iter().any(|candidate| {
3172 matches!(
3173 candidate,
3174 ImmutableSequenceCandidate::BindingTerminal {
3175 delivery_seq,
3176 admission_order,
3177 owner: candidate_owner,
3178 } if *delivery_seq == terminal.delivery_seq
3179 && *admission_order == terminal.admission_order
3180 && *candidate_owner == owner
3181 )
3182 });
3183 let order_matches = self.order.immutable_candidates.iter().any(|candidate| {
3184 candidate.transaction_order == terminal.admission_order.transaction_order()
3185 && candidate.candidate_keys.as_slice() == [terminal.admission_order]
3186 });
3187 sequence_matches && order_matches
3188 }
3189
3190 fn has_recovery_marker(&self, participant: FrontierParticipant) -> bool {
3191 self.retained_records.iter().any(|record| {
3192 record.delivery_seq == participant.cursor()
3193 && matches!(
3194 record.kind,
3195 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
3196 if participant_index == participant.participant_index()
3197 )
3198 })
3199 }
3200
3201 fn has_duplicate_appended_record(&self, appended_records: &[RetainedCausalRecord]) -> bool {
3202 appended_records.iter().any(|row| {
3203 self.retained_records
3204 .iter()
3205 .any(|retained| retained.delivery_seq == row.delivery_seq)
3206 })
3207 }
3208
3209 pub(in crate::lifecycle) fn apply_live_identity(
3212 mut self,
3213 participant: FrontierParticipant,
3214 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
3215 let Some(current) = self
3216 .active_identities
3217 .participants
3218 .iter_mut()
3219 .find(|current| current.participant_index == participant.participant_index)
3220 else {
3221 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3222 };
3223 if participant.cursor < current.cursor
3224 || participant.cursor > self.sequence.ledger.high_watermark()
3225 {
3226 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3227 }
3228 *current = participant;
3229 Ok(self)
3230 }
3231
3232 pub fn project_ordinary_record(
3248 self,
3249 input: OrdinaryRecordProjectionInput,
3250 ) -> Result<OrdinaryRecordProjectionDecision, Box<OrdinaryRecordProjectionFailure>> {
3251 let (
3252 request,
3253 receiving_binding_epoch,
3254 encoded_record_charge,
3255 retained_charges,
3256 observer_progress,
3257 closure_accounting,
3258 limits,
3259 ) = input.as_parts();
3260 if request.conversation_id != self.conversation_id {
3261 return Err(projection_failure(
3262 self,
3263 input,
3264 OrdinaryProjectionError::Conversation,
3265 ));
3266 }
3267 let unaccepted_marker_anchors = ordinary_unaccepted_marker_anchors(&self);
3268 let kernel = match project_ordinary_fixed_point(&OrdinaryProjectionFacts {
3269 request: request.clone(),
3270 receiving_binding_epoch,
3271 encoded_record_charge,
3272 retained_records: &self.retained_records,
3273 retained_charges,
3274 active_marker_credit_records: &self.marker_records,
3275 unaccepted_marker_anchors: &unaccepted_marker_anchors,
3276 active_identities: self.active_identities.participants(),
3277 identity_slot_limit: self.identity_slot_limit,
3278 current_floor: self.retained_floor,
3279 observer_progress,
3280 order_ledger: self.order.ledger,
3281 sequence_ledger: self.sequence.ledger,
3282 immutable_candidates: &self.sequence.immutable_candidates,
3283 closure_accounting,
3284 remaining_recovery_claim: closure_accounting.edge_k_remaining(),
3285 limits,
3286 }) {
3287 Ok(value) => value,
3288 Err(error) => return Err(projection_failure(self, input, error)),
3289 };
3290 match kernel {
3291 OrdinaryProjectionKernelDecision::DrainFirst(prefix) => Ok(
3292 OrdinaryRecordProjectionDecision::DrainFirst(Box::new(OrdinaryRecordDrainFirst {
3293 frontiers: self,
3294 input,
3295 candidate: prefix.candidate(),
3296 })),
3297 ),
3298 OrdinaryProjectionKernelDecision::Projected(projected) => {
3299 let observer_floor = match check_observer_floor(
3300 ObserverCheckedOperation::RecordAdmission(request.clone()),
3301 observer_progress,
3302 projected.floor().resulting_floor,
3303 ) {
3304 ObserverFloorDecision::Eligible(permit) => permit,
3305 ObserverFloorDecision::Respond(_) => {
3306 return Err(projection_failure(
3307 self,
3308 input,
3309 OrdinaryProjectionError::ObserverSelectorInvariant,
3310 ));
3311 }
3312 };
3313 let closure = match check_remaining_closure(
3314 &ClosureCheckedEnvelope::RecordAdmission(request.clone()),
3315 closure_accounting,
3316 false,
3317 0,
3318 projected.required_capacity(),
3319 ) {
3320 RemainingClosureDecision::Eligible(permit) => *permit,
3321 RemainingClosureDecision::Respond(_) => {
3322 return Err(projection_failure(
3323 self,
3324 input,
3325 OrdinaryProjectionError::ClosureSelectorInvariant,
3326 ));
3327 }
3328 };
3329 match self.apply_ordinary_projection(*projected, observer_floor, closure) {
3330 Ok(projected) => Ok(OrdinaryRecordProjectionDecision::Projected(Box::new(
3331 projected,
3332 ))),
3333 Err(failure) => {
3334 let (frontiers, error) = *failure;
3335 Err(projection_failure(frontiers, input, error))
3336 }
3337 }
3338 }
3339 }
3340 }
3341
3342 fn apply_ordinary_projection(
3343 mut self,
3344 projected: OrdinaryFixedPointPlan,
3345 observer_floor: super::ObserverFloorPermit,
3346 closure: super::RemainingClosurePermit,
3347 ) -> Result<ProjectedOrdinaryRecord, Box<(Self, OrdinaryProjectionError)>> {
3348 let Ok(marker_count) = u64::try_from(projected.marker_candidates().len()) else {
3349 return Err(Box::new((
3350 self,
3351 OrdinaryProjectionError::SequenceRelocation,
3352 )));
3353 };
3354 let Some(sequence_delta) = marker_count.checked_add(1) else {
3355 return Err(Box::new((
3356 self,
3357 OrdinaryProjectionError::SequenceRelocation,
3358 )));
3359 };
3360 if let Err(error) = preflight_ordinary_sequence_owners(&self.sequence, sequence_delta) {
3361 return Err(Box::new((self, error)));
3362 }
3363 if let Err(error) = preflight_ordinary_order_owners(&self.order) {
3364 return Err(Box::new((self, error)));
3365 }
3366 let (
3367 floor,
3368 retained_charge,
3369 baseline,
3370 accounting,
3371 required_capacity,
3372 order,
3373 sequence,
3374 caller_record,
3375 caller_charge,
3376 retained_records,
3377 retained_charges,
3378 new_marker_candidates,
3379 ) = projected.into_parts();
3380
3381 let prior_sequence_ledger = self.sequence.ledger;
3382 let prior_order_ledger = self.order.ledger;
3383 relay_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
3384 relay_ordinary_order_owners(&mut self.order);
3385
3386 self.sequence.ledger = sequence.resulting();
3387 self.sequence.immutable_candidates.extend(
3388 new_marker_candidates
3389 .iter()
3390 .copied()
3391 .map(ImmutableSequenceCandidate::Marker),
3392 );
3393 self.order.ledger = order.resulting();
3394 if !new_marker_candidates.is_empty() {
3395 self.order
3396 .immutable_candidates
3397 .push(ImmutableOrderCandidateMajor {
3398 transaction_order: order.major(),
3399 candidate_keys: new_marker_candidates
3400 .iter()
3401 .map(|candidate| candidate.admission_order)
3402 .collect(),
3403 });
3404 }
3405 if validate_cross_counter(&self.sequence, &self.order).is_err() {
3406 self.sequence.ledger = prior_sequence_ledger;
3407 self.sequence.immutable_candidates.clear();
3408 self.order.ledger = prior_order_ledger;
3409 self.order.immutable_candidates.clear();
3410 rollback_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
3411 rollback_ordinary_order_owners(&mut self.order);
3412 return Err(Box::new((
3413 self,
3414 OrdinaryProjectionError::SequenceRelocation,
3415 )));
3416 }
3417 self.retained_floor = floor.resulting_floor;
3418 self.retained_records = retained_records;
3419 self.marker_records
3420 .retain(|record| u128::from(record.delivery_seq) >= floor.resulting_floor);
3421
3422 Ok(ProjectedOrdinaryRecord {
3423 frontiers: self,
3424 floor,
3425 retained_charge,
3426 baseline,
3427 accounting,
3428 required_capacity,
3429 order,
3430 sequence,
3431 observer_floor,
3432 closure,
3433 caller_record,
3434 caller_charge,
3435 retained_charges,
3436 new_marker_candidates,
3437 })
3438 }
3439
3440 pub fn planned_settled_leave_admission_order<F>(
3452 &self,
3453 member: &LiveMember<F>,
3454 binding_state: BindingState,
3455 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
3456 let (participant_id, ended_binding_epoch) =
3457 validate_settled_leave_prestate(self, member, binding_state)?;
3458 let selection = select_leave_order(&self.order, participant_id, ended_binding_epoch, None)?;
3459 Ok(super::AdmissionOrder::new(
3460 selection.selected_major,
3461 CandidatePhase::MembershipExit,
3462 participant_id,
3463 ))
3464 }
3465
3466 pub fn prepare_settled_leave_authority<F>(
3479 mut self,
3480 member: &LiveMember<F>,
3481 binding_state: BindingState,
3482 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
3483 let (participant_id, ended_binding_epoch) =
3484 validate_settled_leave_prestate(&self, member, binding_state)?;
3485 let left_transaction_order =
3486 consume_leave_order_lane(&mut self.order, participant_id, ended_binding_epoch, None)?;
3487 Ok(PreparedLeaveAuthority::settled(
3488 self,
3489 member.conversation_id(),
3490 participant_id,
3491 ended_binding_epoch,
3492 left_transaction_order,
3493 ))
3494 }
3495
3496 pub fn planned_pending_leave_admission_order<F>(
3504 &self,
3505 member: &LiveMember<F>,
3506 pending: PendingFinalization,
3507 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
3508 let (participant_id, expected_order) =
3509 validate_pending_leave_prestate(self, member, pending)?;
3510 let selection =
3511 select_leave_order(&self.order, participant_id, None, Some(expected_order))?;
3512 Ok(super::AdmissionOrder::new(
3513 selection.selected_major,
3514 CandidatePhase::MembershipExit,
3515 participant_id,
3516 ))
3517 }
3518
3519 pub fn prepare_pending_leave_authority<F>(
3533 mut self,
3534 member: &LiveMember<F>,
3535 pending: PendingFinalization,
3536 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
3537 let (participant_id, expected_order) =
3538 validate_pending_leave_prestate(&self, member, pending)?;
3539 let left_transaction_order =
3540 consume_leave_order_lane(&mut self.order, participant_id, None, Some(expected_order))?;
3541 Ok(PreparedLeaveAuthority::pending(
3542 self,
3543 member.conversation_id(),
3544 participant_id,
3545 pending.binding_epoch(),
3546 expected_order,
3547 left_transaction_order,
3548 ))
3549 }
3550
3551 #[allow(
3560 clippy::too_many_lines,
3561 reason = "the atomic Leave relay keeps membership, both ledgers, products, recovery, and retained rows visibly in one checked transition"
3562 )]
3563 pub(super) fn finish_leave_claims(
3564 mut self,
3565 participant_id: ParticipantId,
3566 ended_binding_epoch: Option<BindingEpoch>,
3567 committed_terminal: Option<CommittedBindingTerminal>,
3568 left_delivery_seq: DeliverySeq,
3569 left_transaction_order: TransactionOrder,
3570 ) -> Result<Self, LeaveCommitError> {
3571 let prior_high = self.sequence.ledger.high_watermark();
3572 let first_appended = prior_high
3573 .checked_add(1)
3574 .ok_or(LeaveCommitError::SequenceAuthority)?;
3575 let expected_left = if committed_terminal.is_some() {
3576 first_appended
3577 .checked_add(1)
3578 .ok_or(LeaveCommitError::SequenceAuthority)?
3579 } else {
3580 first_appended
3581 };
3582 if left_delivery_seq != expected_left {
3583 return Err(LeaveCommitError::SequenceAuthority);
3584 }
3585
3586 match (
3587 committed_terminal,
3588 self.sequence.immutable_candidates.as_slice(),
3589 ) {
3590 (
3591 Some(terminal),
3592 [
3593 ImmutableSequenceCandidate::BindingTerminal {
3594 delivery_seq,
3595 admission_order,
3596 owner,
3597 },
3598 ],
3599 ) if *delivery_seq == first_appended
3600 && terminal.delivery_seq() == first_appended
3601 && *admission_order == terminal.admission_order()
3602 && owner.participant_index == participant_id
3603 && owner.binding_epoch == terminal.binding_epoch() => {}
3604 (None, []) => {}
3605 (Some(_) | None, _) => return Err(LeaveCommitError::SequenceAuthority),
3606 }
3607
3608 let Some(active_index) = self
3609 .active_identities
3610 .participants
3611 .iter()
3612 .position(|participant| participant.participant_index == participant_id)
3613 else {
3614 return Err(LeaveCommitError::ResultingFrontier);
3615 };
3616 self.active_identities.participants.remove(active_index);
3617 self.marker_records.retain(|record| {
3618 !matches!(
3619 record.kind,
3620 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
3621 if participant_index == participant_id
3622 )
3623 });
3624 if self
3627 .fenced_marker_issued
3628 .is_some_and(|(latched, _, _)| latched == participant_id)
3629 {
3630 self.fenced_marker_issued = None;
3631 }
3632 let resulting_live = usize_to_u64(self.active_identities.participants.len());
3633
3634 let mut exit_consumed = false;
3635 let mut terminal_consumed = ended_binding_epoch.is_none();
3636 let mut units = Vec::new();
3637 for claim in self.sequence.movable_claims.iter().copied() {
3638 match claim.owner {
3639 SequenceDirectOwner::MembershipExit {
3640 participant_index: owner,
3641 } if owner == participant_id => {
3642 if exit_consumed {
3643 return Err(LeaveCommitError::ResultingFrontier);
3644 }
3645 exit_consumed = true;
3646 }
3647 SequenceDirectOwner::BindingTerminal(owner)
3648 if owner.participant_index == participant_id
3649 && Some(owner.binding_epoch) == ended_binding_epoch =>
3650 {
3651 if terminal_consumed {
3652 return Err(LeaveCommitError::ResultingFrontier);
3653 }
3654 terminal_consumed = true;
3655 }
3656 SequenceDirectOwner::MembershipExit { .. }
3657 | SequenceDirectOwner::BindingTerminal(_) => {
3658 units.push(LeaveSequenceUnit::Direct(claim));
3659 }
3660 }
3661 }
3662 if !exit_consumed || !terminal_consumed {
3663 return Err(LeaveCommitError::ResultingFrontier);
3664 }
3665
3666 let recovery_owned = self
3667 .sequence
3668 .recovery
3669 .is_some_and(|block| block.participant_index == participant_id);
3670 for range in self.sequence.products.live_times_terminal.iter().copied() {
3671 if range.terminal.participant_index != participant_id && resulting_live != 0 {
3672 units.push(LeaveSequenceUnit::TerminalProduct(TerminalProductRange {
3673 start: range.start,
3674 length: resulting_live,
3675 terminal: range.terminal,
3676 }));
3677 }
3678 }
3679 if let Some(range) = self.sequence.products.live_times_replacement_terminal
3680 && !recovery_owned
3681 && resulting_live != 0
3682 {
3683 units.push(LeaveSequenceUnit::ReplacementProduct(
3684 ReplacementTerminalProductRange {
3685 start: range.start,
3686 length: resulting_live,
3687 participant_index: range.participant_index,
3688 marker_delivery_seq: range.marker_delivery_seq,
3689 prior_binding_epoch: range.prior_binding_epoch,
3690 },
3691 ));
3692 }
3693 let resulting_other = resulting_live.saturating_sub(1);
3694 if resulting_other != 0 {
3695 for range in self.sequence.products.other_live_times_exit.iter().copied() {
3696 if range.exit_participant != participant_id {
3697 units.push(LeaveSequenceUnit::ExitProduct(ExitProductRange {
3698 start: range.start,
3699 length: resulting_other,
3700 exit_participant: range.exit_participant,
3701 }));
3702 }
3703 }
3704 }
3705 if let Some(recovery) = self.sequence.recovery
3706 && !recovery_owned
3707 {
3708 units.push(LeaveSequenceUnit::Recovery(recovery));
3709 }
3710 units.sort_by_key(|unit| unit.original_start());
3711
3712 let mut cursor = left_delivery_seq.checked_add(1);
3713 let mut movable_claims = Vec::new();
3714 let mut terminal_products = Vec::new();
3715 let mut replacement_product = None;
3716 let mut exit_products = Vec::new();
3717 let mut recovery = None;
3718 for unit in units {
3719 match unit {
3720 LeaveSequenceUnit::Direct(mut claim) => {
3721 claim.delivery_seq = allocate_leave_sequence_range(&mut cursor, 1)?;
3722 movable_claims.push(claim);
3723 }
3724 LeaveSequenceUnit::TerminalProduct(mut range) => {
3725 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3726 terminal_products.push(range);
3727 }
3728 LeaveSequenceUnit::ReplacementProduct(mut range) => {
3729 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3730 replacement_product = Some(range);
3731 }
3732 LeaveSequenceUnit::ExitProduct(mut range) => {
3733 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3734 exit_products.push(range);
3735 }
3736 LeaveSequenceUnit::Recovery(mut block) => {
3737 let length = 2 + u64::from(block.terminal.is_some());
3738 let start = allocate_leave_sequence_range(&mut cursor, length)?;
3739 if let Some(mut terminal) = block.terminal {
3740 terminal.delivery_seq = start;
3741 block.terminal = Some(terminal);
3742 block.recovery_attach_seq = start
3743 .checked_add(1)
3744 .ok_or(LeaveCommitError::ResultingFrontier)?;
3745 } else {
3746 block.recovery_attach_seq = start;
3747 }
3748 block.replacement_terminal_seq = block
3749 .recovery_attach_seq
3750 .checked_add(1)
3751 .ok_or(LeaveCommitError::ResultingFrontier)?;
3752 recovery = Some(block);
3753 }
3754 }
3755 }
3756 movable_claims.sort_by_key(|claim| claim.delivery_seq);
3757 terminal_products.sort_by_key(|range| range.start);
3758 exit_products.sort_by_key(|range| range.start);
3759 let terminal_count = usize_to_u64(
3760 movable_claims
3761 .iter()
3762 .filter(|claim| matches!(claim.owner, SequenceDirectOwner::BindingTerminal(_)))
3763 .count(),
3764 ) + u64::from(recovery.is_some_and(|block| block.terminal.is_some()));
3765 let recovery_reserve = if recovery.is_some() {
3766 RecoverySequenceReserve::DetachedCredentialRecovery
3767 } else {
3768 RecoverySequenceReserve::None
3769 };
3770 let ledger = SequenceLedger::try_new(
3771 left_delivery_seq,
3772 SequenceClaims::new(resulting_live, terminal_count, 0, recovery_reserve),
3773 )
3774 .map_err(|_| LeaveCommitError::ResultingFrontier)?;
3775 self.sequence = SequenceClaimFrontier {
3776 ledger,
3777 movable_claims,
3778 immutable_candidates: Vec::new(),
3779 products: SequenceProductRanges {
3780 live_times_terminal: terminal_products,
3781 live_times_replacement_terminal: replacement_product,
3782 other_live_times_exit: exit_products,
3783 },
3784 recovery,
3785 };
3786
3787 if let Some(terminal) = committed_terminal {
3788 self.retained_records.push(RetainedCausalRecord {
3789 delivery_seq: terminal.delivery_seq(),
3790 admission_order: terminal.admission_order(),
3791 kind: RetainedCausalRecordKind::BindingTerminal(BindingTerminalOwner {
3792 participant_index: participant_id,
3793 binding_epoch: terminal.binding_epoch(),
3794 }),
3795 });
3796 }
3797 self.retained_records.push(RetainedCausalRecord {
3798 delivery_seq: left_delivery_seq,
3799 admission_order: super::AdmissionOrder::new(
3800 left_transaction_order,
3801 CandidatePhase::MembershipExit,
3802 participant_id,
3803 ),
3804 kind: RetainedCausalRecordKind::MembershipExit {
3805 participant_index: participant_id,
3806 },
3807 });
3808 self.retained_records
3809 .sort_by_key(|record| record.delivery_seq);
3810
3811 let order_claims = self.order.ledger.claims();
3812 if order_claims.membership_exits() != resulting_live
3813 || self.order.recovery.is_some() != self.sequence.recovery.is_some()
3814 || validate_cross_counter(&self.sequence, &self.order).is_err()
3815 {
3816 return Err(LeaveCommitError::ResultingFrontier);
3817 }
3818 Ok(self)
3819 }
3820
3821 fn marker_candidate(&self, delivery_seq: DeliverySeq) -> Option<ValidatedMarkerCandidate> {
3822 self.sequence
3823 .immutable_candidates
3824 .iter()
3825 .find_map(|candidate| match candidate {
3826 ImmutableSequenceCandidate::Marker(candidate)
3827 if candidate.delivery_seq == delivery_seq =>
3828 {
3829 Some(ValidatedMarkerCandidate {
3830 conversation_id: self.conversation_id,
3831 candidate: *candidate,
3832 seal: MarkerAuthoritySeal::Validated,
3833 })
3834 }
3835 _ => None,
3836 })
3837 }
3838
3839 pub(super) fn drain_next_marker_core(
3848 mut self,
3849 ) -> Result<MarkerDrainCore, MarkerDrainCoreError> {
3850 let Some(first) = self.sequence.immutable_candidates.first().copied() else {
3851 return Err(MarkerDrainCoreError::NoCandidate);
3852 };
3853 let ImmutableSequenceCandidate::Marker(marker) = first else {
3854 return Err(MarkerDrainCoreError::BindingTerminalFirst);
3855 };
3856 let expected_sequence = self
3857 .sequence
3858 .ledger
3859 .high_watermark()
3860 .checked_add(1)
3861 .ok_or(MarkerDrainCoreError::SequenceNotNext)?;
3862 if marker.delivery_seq != expected_sequence {
3863 return Err(MarkerDrainCoreError::SequenceNotNext);
3864 }
3865 if order_is_above_high(
3866 marker.admission_order.transaction_order(),
3867 self.order.ledger.high(),
3868 ) {
3869 return Err(MarkerDrainCoreError::CausalMajorNotAllocated);
3870 }
3871 let candidate = self
3872 .marker_candidate(marker.delivery_seq)
3873 .ok_or(MarkerDrainCoreError::NoCandidate)?;
3874
3875 let key = marker.admission_order;
3876 let Some(group_index) = self
3877 .order
3878 .immutable_candidates
3879 .iter()
3880 .position(|group| group.candidate_keys.contains(&key))
3881 else {
3882 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3883 };
3884 let group = &mut self.order.immutable_candidates[group_index];
3885 let Ok(key_index) = group.candidate_keys.binary_search(&key) else {
3886 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3887 };
3888 group.candidate_keys.remove(key_index);
3889 if group.candidate_keys.is_empty() {
3890 self.order.immutable_candidates.remove(group_index);
3891 }
3892 self.sequence.immutable_candidates.remove(0);
3893
3894 let claims = self.sequence.ledger.claims();
3895 let markers = claims
3896 .markers()
3897 .checked_sub(1)
3898 .ok_or(MarkerDrainCoreError::ResultingLedger)?;
3899 self.sequence.ledger = SequenceLedger::try_new(
3900 expected_sequence,
3901 super::SequenceClaims::new(
3902 claims.live_members(),
3903 claims.binding_terminals(),
3904 markers,
3905 claims.recovery(),
3906 ),
3907 )
3908 .map_err(|_| MarkerDrainCoreError::ResultingLedger)?;
3909
3910 let record = RetainedCausalRecord {
3911 delivery_seq: marker.delivery_seq,
3912 admission_order: marker.admission_order,
3913 kind: RetainedCausalRecordKind::CompactionMarker {
3914 participant_index: marker.admission_order.participant_index(),
3915 provenance: marker.provenance,
3916 },
3917 };
3918 self.marker_records.push(record);
3919 self.retained_records.push(record);
3920 Ok(MarkerDrainCore {
3921 candidate,
3922 record: ValidatedMarkerRecord {
3923 conversation_id: self.conversation_id,
3924 record,
3925 provenance: marker.provenance,
3926 target_binding: marker.target_binding,
3927 occurrence: MarkerRecordOccurrence::Undelivered,
3928 seal: MarkerAuthoritySeal::Validated,
3929 },
3930 frontiers: self,
3931 })
3932 }
3933
3934 pub(in crate::lifecycle) fn drain_first_binding_terminal(
3964 mut self,
3965 expected_owner: BindingTerminalOwner,
3966 expected_order: super::AdmissionOrder,
3967 release_identity: bool,
3968 ) -> Result<(Self, RetainedCausalRecord), Box<(Self, LiveFrontierTransitionError)>> {
3969 let (delivery_seq, admission_order, owner) =
3970 match self.validate_first_terminal_candidate(expected_owner, expected_order) {
3971 Ok(candidate) => candidate,
3972 Err(error) => return Err(Box::new((self, error))),
3973 };
3974 let released = u64::from(release_identity);
3975 let claims = self.sequence.ledger.claims();
3976 let Some(sequence_ledger) = claims
3977 .binding_terminals()
3978 .checked_sub(1)
3979 .zip(claims.live_members().checked_sub(released))
3980 .and_then(|(binding_terminals, live_members)| {
3981 SequenceLedger::try_new(
3982 delivery_seq,
3983 SequenceClaims::new(
3984 live_members,
3985 binding_terminals,
3986 claims.markers(),
3987 claims.recovery(),
3988 ),
3989 )
3990 .ok()
3991 })
3992 else {
3993 return Err(Box::new((
3994 self,
3995 LiveFrontierTransitionError::ResultingFrontier,
3996 )));
3997 };
3998 let order_claims = self.order.ledger.claims();
3999 let Some(order_ledger) = order_claims
4000 .membership_exits()
4001 .checked_sub(released)
4002 .and_then(|membership_exits| {
4003 OrderClaims::new(
4004 order_claims.active_binding_terminals(),
4005 membership_exits,
4006 order_claims.recovery_operation(),
4007 order_claims.recovery_replacement_terminal(),
4008 )
4009 .ok()
4010 })
4011 .and_then(|resulting| OrderLedger::try_new(self.order.ledger.high(), resulting).ok())
4012 else {
4013 return Err(Box::new((
4014 self,
4015 LiveFrontierTransitionError::ResultingFrontier,
4016 )));
4017 };
4018 let participants: Vec<_> = self
4019 .active_identities
4020 .participants()
4021 .iter()
4022 .copied()
4023 .filter(|participant| {
4024 !release_identity || participant.participant_index != owner.participant_index
4025 })
4026 .collect();
4027 let Ok(active) = ActiveIdentityRanks::try_new(
4028 participants,
4029 sequence_ledger.high_watermark(),
4030 self.identity_slot_limit,
4031 ) else {
4032 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
4033 };
4034 let (sequence, order) =
4035 match rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger) {
4036 Ok(frontiers) => frontiers,
4037 Err(error) => return Err(Box::new((self, error))),
4038 };
4039 let record = RetainedCausalRecord {
4040 delivery_seq,
4041 admission_order,
4042 kind: RetainedCausalRecordKind::BindingTerminal(owner),
4043 };
4044 if release_identity {
4045 self.marker_records.retain(|record| {
4048 !matches!(
4049 record.kind,
4050 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
4051 if participant_index == owner.participant_index
4052 )
4053 });
4054 if self
4055 .fenced_marker_issued
4056 .is_some_and(|(latched, _, _)| latched == owner.participant_index)
4057 {
4058 self.fenced_marker_issued = None;
4059 }
4060 }
4061 self.active_identities = active;
4062 self.sequence = sequence;
4063 self.order = order;
4064 self.retained_records.push(record);
4065 Ok((self, record))
4066 }
4067
4068 fn validate_first_terminal_candidate(
4073 &self,
4074 expected_owner: BindingTerminalOwner,
4075 expected_order: super::AdmissionOrder,
4076 ) -> Result<
4077 (DeliverySeq, super::AdmissionOrder, BindingTerminalOwner),
4078 LiveFrontierTransitionError,
4079 > {
4080 if self.sequence.recovery.is_some() || self.order.recovery.is_some() {
4081 return Err(LiveFrontierTransitionError::Precedence(
4082 PrecedenceCondition::FencedRecovery,
4083 ));
4084 }
4085 let [first] = self.sequence.immutable_candidates.as_slice() else {
4086 return Err(LiveFrontierTransitionError::Precedence(
4087 self.precedence_condition(),
4088 ));
4089 };
4090 let ImmutableSequenceCandidate::BindingTerminal {
4091 delivery_seq,
4092 admission_order,
4093 owner,
4094 } = *first
4095 else {
4096 return Err(LiveFrontierTransitionError::Precedence(
4097 self.precedence_condition(),
4098 ));
4099 };
4100 if owner != expected_owner || admission_order != expected_order {
4101 return Err(LiveFrontierTransitionError::Authority);
4102 }
4103 let Some(expected_sequence) = self.sequence.ledger.high_watermark().checked_add(1) else {
4104 return Err(LiveFrontierTransitionError::Exhausted);
4105 };
4106 if delivery_seq != expected_sequence
4107 || order_is_above_high(
4108 admission_order.transaction_order(),
4109 self.order.ledger.high(),
4110 )
4111 {
4112 return Err(LiveFrontierTransitionError::RecordPosition);
4113 }
4114 let [group] = self.order.immutable_candidates.as_slice() else {
4115 return Err(LiveFrontierTransitionError::RecordPosition);
4116 };
4117 if group.candidate_keys.as_slice() != [admission_order] {
4118 return Err(LiveFrontierTransitionError::RecordPosition);
4119 }
4120 let detached_matches = self
4121 .active_identities
4122 .participants()
4123 .iter()
4124 .any(|participant| {
4125 participant.participant_index == owner.participant_index
4126 && participant.binding == FrontierBinding::Detached(owner.binding_epoch)
4127 });
4128 if !detached_matches {
4129 return Err(LiveFrontierTransitionError::Authority);
4130 }
4131 Ok((delivery_seq, admission_order, owner))
4132 }
4133}
4134
4135fn apply_fenced_ledgers(
4136 sequence: SequenceLedger,
4137 order: OrderLedger,
4138 finalizes_pending: bool,
4139) -> Option<(SequenceLedger, OrderLedger)> {
4140 let sequence = if finalizes_pending {
4141 sequence.apply_fenced_recovery_finalizing_pending()
4142 } else {
4143 sequence.apply_fenced_recovery()
4144 }
4145 .ok()?;
4146 let order = if finalizes_pending {
4147 order.apply_fenced_recovery_finalizing_pending()
4148 } else {
4149 order.apply_fenced_recovery()
4150 }
4151 .ok()?;
4152 Some((sequence, order))
4153}
4154
4155fn rebuild_unreserved_frontiers(
4156 active: &ActiveIdentityRanks,
4157 sequence_ledger: SequenceLedger,
4158 order_ledger: OrderLedger,
4159) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
4160 let terminal_owners: Vec<_> = active
4161 .participants()
4162 .iter()
4163 .filter_map(|participant| match participant.binding() {
4164 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
4165 participant_index: participant.participant_index(),
4166 binding_epoch,
4167 }),
4168 FrontierBinding::Detached(_) => None,
4169 })
4170 .collect();
4171 let live_count = active.len();
4172 let terminal_count =
4173 u64::try_from(terminal_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4174 let sequence_claims = sequence_ledger.claims();
4175 let order_claims = order_ledger.claims();
4176 if sequence_claims.live_members() != live_count
4177 || sequence_claims.binding_terminals() != terminal_count
4178 || sequence_claims.markers() != 0
4179 || sequence_claims.recovery() != RecoverySequenceReserve::None
4180 || order_claims.active_binding_terminals() != terminal_count
4181 || order_claims.membership_exits() != live_count
4182 || order_claims.recovery_operation()
4183 || order_claims.recovery_replacement_terminal()
4184 {
4185 return Err(LiveFrontierTransitionError::ResultingFrontier);
4186 }
4187
4188 let sequence = rebuild_unreserved_sequence(active, &terminal_owners, sequence_ledger)?;
4189 let order = rebuild_unreserved_order(active, &terminal_owners, order_ledger)?;
4190 validate_cross_counter(&sequence, &order)
4191 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
4192 Ok((sequence, order))
4193}
4194
4195fn rebuild_unreserved_sequence(
4196 active: &ActiveIdentityRanks,
4197 terminal_owners: &[BindingTerminalOwner],
4198 sequence_ledger: SequenceLedger,
4199) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
4200 let live_count = active.len();
4201 let mut sequence_cursor = sequence_ledger
4202 .high_watermark()
4203 .checked_add(1)
4204 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4205 let mut movable_sequence = Vec::new();
4206 for terminal in terminal_owners {
4207 movable_sequence.push(MovableSequenceClaim {
4208 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
4209 owner: SequenceDirectOwner::BindingTerminal(*terminal),
4210 });
4211 }
4212 for participant in active.participants() {
4213 movable_sequence.push(MovableSequenceClaim {
4214 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
4215 owner: SequenceDirectOwner::MembershipExit {
4216 participant_index: participant.participant_index(),
4217 },
4218 });
4219 }
4220 let mut terminal_products = Vec::new();
4221 for terminal in terminal_owners {
4222 terminal_products.push(TerminalProductRange {
4223 start: take_live_sequence(&mut sequence_cursor, live_count)?,
4224 length: live_count,
4225 terminal: *terminal,
4226 });
4227 }
4228 let exit_product_length = live_count.saturating_sub(1);
4229 let mut exit_products = Vec::new();
4230 if live_count > 1 {
4231 for participant in active.participants() {
4232 exit_products.push(ExitProductRange {
4233 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
4234 length: exit_product_length,
4235 exit_participant: participant.participant_index(),
4236 });
4237 }
4238 }
4239 let sequence_end = u128::from(sequence_ledger.high_watermark())
4240 .checked_add(sequence_ledger.required_reserve())
4241 .and_then(|value| value.checked_add(1))
4242 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4243 if u128::from(sequence_cursor) != sequence_end {
4244 return Err(LiveFrontierTransitionError::ResultingFrontier);
4245 }
4246 Ok(SequenceClaimFrontier {
4247 ledger: sequence_ledger,
4248 movable_claims: movable_sequence,
4249 immutable_candidates: Vec::new(),
4250 products: SequenceProductRanges {
4251 live_times_terminal: terminal_products,
4252 live_times_replacement_terminal: None,
4253 other_live_times_exit: exit_products,
4254 },
4255 recovery: None,
4256 })
4257}
4258
4259fn rebuild_unreserved_order(
4260 active: &ActiveIdentityRanks,
4261 terminal_owners: &[BindingTerminalOwner],
4262 order_ledger: OrderLedger,
4263) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
4264 let order_claims = order_ledger.claims();
4265 let order_start = order_frontier_start(order_ledger.high());
4266 let mut order_cursor =
4267 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4268 let mut movable_order = Vec::new();
4269 for terminal in terminal_owners.iter().copied() {
4270 movable_order.push(MovableOrderClaim {
4271 transaction_order: take_live_order(&mut order_cursor)?,
4272 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
4273 });
4274 }
4275 for participant in active.participants() {
4276 movable_order.push(MovableOrderClaim {
4277 transaction_order: take_live_order(&mut order_cursor)?,
4278 owner: OrderDirectOwner::MembershipExit {
4279 participant_index: participant.participant_index(),
4280 },
4281 });
4282 }
4283 if u128::from(order_cursor) != order_start + order_claims.total() {
4284 return Err(LiveFrontierTransitionError::ResultingFrontier);
4285 }
4286 Ok(OrderClaimFrontier {
4287 ledger: order_ledger,
4288 movable_claims: movable_order,
4289 immutable_candidates: Vec::new(),
4290 recovery: None,
4291 })
4292}
4293
4294fn rebuild_pending_terminal_frontiers(
4295 active: &ActiveIdentityRanks,
4296 pending_owner: BindingTerminalOwner,
4297 delivery_seq: DeliverySeq,
4298 admission_order: super::AdmissionOrder,
4299 sequence_ledger: SequenceLedger,
4300 order_ledger: OrderLedger,
4301) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
4302 if admission_order.candidate_phase() != CandidatePhase::BindingTerminal
4303 || admission_order.participant_index() != pending_owner.participant_index
4304 || admission_order.transaction_order()
4305 != match order_ledger.high() {
4306 OrderHigh::Allocated(high) => high,
4307 OrderHigh::Empty => return Err(LiveFrontierTransitionError::RecordPosition),
4308 }
4309 || sequence_ledger.high_watermark().checked_add(1) != Some(delivery_seq)
4310 {
4311 return Err(LiveFrontierTransitionError::RecordPosition);
4312 }
4313 let bound_owners: Vec<_> = active
4314 .participants()
4315 .iter()
4316 .filter_map(|participant| match participant.binding() {
4317 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
4318 participant_index: participant.participant_index(),
4319 binding_epoch,
4320 }),
4321 FrontierBinding::Detached(_) => None,
4322 })
4323 .collect();
4324 let mut all_terminal_owners = bound_owners.clone();
4325 all_terminal_owners.push(pending_owner);
4326 all_terminal_owners.sort_unstable_by_key(|owner| owner.participant_index);
4327 let live_count = active.len();
4328 let bound_count =
4329 u64::try_from(bound_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4330 let terminal_count = u64::try_from(all_terminal_owners.len())
4331 .map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4332 let sequence_claims = sequence_ledger.claims();
4333 let order_claims = order_ledger.claims();
4334 if sequence_claims.live_members() != live_count
4335 || sequence_claims.binding_terminals() != terminal_count
4336 || sequence_claims.markers() != 0
4337 || sequence_claims.recovery() != RecoverySequenceReserve::None
4338 || order_claims.active_binding_terminals() != bound_count
4339 || order_claims.membership_exits() != live_count
4340 || order_claims.recovery_operation()
4341 || order_claims.recovery_replacement_terminal()
4342 {
4343 return Err(LiveFrontierTransitionError::ResultingFrontier);
4344 }
4345
4346 let sequence = rebuild_pending_terminal_sequence(
4347 active,
4348 pending_owner,
4349 delivery_seq,
4350 admission_order,
4351 sequence_ledger,
4352 &bound_owners,
4353 &all_terminal_owners,
4354 )?;
4355 let order =
4356 rebuild_pending_terminal_order(active, admission_order, order_ledger, bound_owners)?;
4357 validate_cross_counter(&sequence, &order)
4358 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
4359 Ok((sequence, order))
4360}
4361
4362fn rebuild_pending_terminal_sequence(
4363 active: &ActiveIdentityRanks,
4364 pending_owner: BindingTerminalOwner,
4365 delivery_seq: DeliverySeq,
4366 admission_order: super::AdmissionOrder,
4367 sequence_ledger: SequenceLedger,
4368 bound_owners: &[BindingTerminalOwner],
4369 all_terminal_owners: &[BindingTerminalOwner],
4370) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
4371 let live_count = active.len();
4372 let mut sequence_cursor = delivery_seq;
4373 let candidate_sequence = take_live_sequence(&mut sequence_cursor, 1)?;
4374 let mut movable_sequence = Vec::new();
4375 for terminal in bound_owners {
4376 movable_sequence.push(MovableSequenceClaim {
4377 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
4378 owner: SequenceDirectOwner::BindingTerminal(*terminal),
4379 });
4380 }
4381 for participant in active.participants() {
4382 movable_sequence.push(MovableSequenceClaim {
4383 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
4384 owner: SequenceDirectOwner::MembershipExit {
4385 participant_index: participant.participant_index(),
4386 },
4387 });
4388 }
4389 let mut terminal_products = Vec::new();
4390 for terminal in all_terminal_owners {
4391 terminal_products.push(TerminalProductRange {
4392 start: take_live_sequence(&mut sequence_cursor, live_count)?,
4393 length: live_count,
4394 terminal: *terminal,
4395 });
4396 }
4397 let exit_product_length = live_count.saturating_sub(1);
4398 let mut exit_products = Vec::new();
4399 if live_count > 1 {
4400 for participant in active.participants() {
4401 exit_products.push(ExitProductRange {
4402 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
4403 length: exit_product_length,
4404 exit_participant: participant.participant_index(),
4405 });
4406 }
4407 }
4408 let sequence_end = u128::from(sequence_ledger.high_watermark())
4409 .checked_add(sequence_ledger.required_reserve())
4410 .and_then(|value| value.checked_add(1))
4411 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4412 if u128::from(sequence_cursor) != sequence_end {
4413 return Err(LiveFrontierTransitionError::ResultingFrontier);
4414 }
4415 Ok(SequenceClaimFrontier {
4416 ledger: sequence_ledger,
4417 movable_claims: movable_sequence,
4418 immutable_candidates: alloc::vec![ImmutableSequenceCandidate::BindingTerminal {
4419 delivery_seq: candidate_sequence,
4420 admission_order,
4421 owner: pending_owner,
4422 }],
4423 products: SequenceProductRanges {
4424 live_times_terminal: terminal_products,
4425 live_times_replacement_terminal: None,
4426 other_live_times_exit: exit_products,
4427 },
4428 recovery: None,
4429 })
4430}
4431
4432fn rebuild_pending_terminal_order(
4433 active: &ActiveIdentityRanks,
4434 admission_order: super::AdmissionOrder,
4435 order_ledger: OrderLedger,
4436 bound_owners: Vec<BindingTerminalOwner>,
4437) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
4438 let order_claims = order_ledger.claims();
4439 let order_start = order_frontier_start(order_ledger.high());
4440 let mut order_cursor =
4441 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4442 let mut movable_order = Vec::new();
4443 for terminal in bound_owners {
4444 movable_order.push(MovableOrderClaim {
4445 transaction_order: take_live_order(&mut order_cursor)?,
4446 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
4447 });
4448 }
4449 for participant in active.participants() {
4450 movable_order.push(MovableOrderClaim {
4451 transaction_order: take_live_order(&mut order_cursor)?,
4452 owner: OrderDirectOwner::MembershipExit {
4453 participant_index: participant.participant_index(),
4454 },
4455 });
4456 }
4457 if u128::from(order_cursor) != order_start + order_claims.total() {
4458 return Err(LiveFrontierTransitionError::ResultingFrontier);
4459 }
4460 Ok(OrderClaimFrontier {
4461 ledger: order_ledger,
4462 movable_claims: movable_order,
4463 immutable_candidates: alloc::vec![ImmutableOrderCandidateMajor {
4464 transaction_order: admission_order.transaction_order(),
4465 candidate_keys: alloc::vec![admission_order],
4466 }],
4467 recovery: None,
4468 })
4469}
4470
4471fn take_live_sequence(
4472 cursor: &mut DeliverySeq,
4473 length: u64,
4474) -> Result<DeliverySeq, LiveFrontierTransitionError> {
4475 let start = *cursor;
4476 *cursor = cursor
4477 .checked_add(length)
4478 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4479 Ok(start)
4480}
4481
4482fn take_live_order(
4483 cursor: &mut TransactionOrder,
4484) -> Result<TransactionOrder, LiveFrontierTransitionError> {
4485 let value = *cursor;
4486 *cursor = cursor
4487 .checked_add(1)
4488 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4489 Ok(value)
4490}
4491
4492fn ordinary_unaccepted_marker_anchors(frontiers: &ClaimFrontiers) -> Vec<DeliverySeq> {
4493 frontiers
4494 .marker_records
4495 .iter()
4496 .filter_map(|record| {
4497 let RetainedCausalRecordKind::CompactionMarker {
4498 participant_index, ..
4499 } = record.kind
4500 else {
4501 return None;
4502 };
4503 active_participant(&frontiers.active_identities, participant_index)
4504 .is_some_and(|participant| participant.cursor < record.delivery_seq)
4505 .then_some(record.delivery_seq)
4506 })
4507 .collect()
4508}
4509
4510fn projection_failure(
4511 frontiers: ClaimFrontiers,
4512 input: OrdinaryRecordProjectionInput,
4513 error: OrdinaryProjectionError,
4514) -> Box<OrdinaryRecordProjectionFailure> {
4515 Box::new(OrdinaryRecordProjectionFailure {
4516 frontiers,
4517 input,
4518 error,
4519 })
4520}
4521
4522fn preflight_ordinary_sequence_owners(
4523 sequence: &SequenceClaimFrontier,
4524 delta: u64,
4525) -> Result<(), OrdinaryProjectionError> {
4526 if !sequence.immutable_candidates.is_empty() {
4527 return Err(OrdinaryProjectionError::SequenceRelocation);
4528 }
4529 for claim in &sequence.movable_claims {
4530 claim
4531 .delivery_seq
4532 .checked_add(delta)
4533 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4534 }
4535 for range in &sequence.products.live_times_terminal {
4536 range
4537 .start
4538 .checked_add(delta)
4539 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4540 }
4541 if let Some(range) = &sequence.products.live_times_replacement_terminal {
4542 range
4543 .start
4544 .checked_add(delta)
4545 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4546 }
4547 for range in &sequence.products.other_live_times_exit {
4548 range
4549 .start
4550 .checked_add(delta)
4551 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4552 }
4553 if let Some(recovery) = &sequence.recovery {
4554 if let Some(terminal) = &recovery.terminal {
4555 terminal
4556 .delivery_seq
4557 .checked_add(delta)
4558 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4559 }
4560 recovery
4561 .recovery_attach_seq
4562 .checked_add(delta)
4563 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4564 recovery
4565 .replacement_terminal_seq
4566 .checked_add(delta)
4567 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4568 }
4569 Ok(())
4570}
4571
4572fn relay_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
4573 for claim in &mut sequence.movable_claims {
4574 claim.delivery_seq = claim.delivery_seq.wrapping_add(delta);
4575 }
4576 for range in &mut sequence.products.live_times_terminal {
4577 range.start = range.start.wrapping_add(delta);
4578 }
4579 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
4580 range.start = range.start.wrapping_add(delta);
4581 }
4582 for range in &mut sequence.products.other_live_times_exit {
4583 range.start = range.start.wrapping_add(delta);
4584 }
4585 if let Some(recovery) = &mut sequence.recovery {
4586 if let Some(terminal) = &mut recovery.terminal {
4587 terminal.delivery_seq = terminal.delivery_seq.wrapping_add(delta);
4588 }
4589 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_add(delta);
4590 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_add(delta);
4591 }
4592}
4593
4594fn rollback_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
4595 for claim in &mut sequence.movable_claims {
4596 claim.delivery_seq = claim.delivery_seq.wrapping_sub(delta);
4597 }
4598 for range in &mut sequence.products.live_times_terminal {
4599 range.start = range.start.wrapping_sub(delta);
4600 }
4601 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
4602 range.start = range.start.wrapping_sub(delta);
4603 }
4604 for range in &mut sequence.products.other_live_times_exit {
4605 range.start = range.start.wrapping_sub(delta);
4606 }
4607 if let Some(recovery) = &mut sequence.recovery {
4608 if let Some(terminal) = &mut recovery.terminal {
4609 terminal.delivery_seq = terminal.delivery_seq.wrapping_sub(delta);
4610 }
4611 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_sub(delta);
4612 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_sub(delta);
4613 }
4614}
4615
4616fn preflight_ordinary_order_owners(
4617 order: &OrderClaimFrontier,
4618) -> Result<(), OrdinaryProjectionError> {
4619 if !order.immutable_candidates.is_empty() {
4620 return Err(OrdinaryProjectionError::OrderRelocation);
4621 }
4622 for claim in &order.movable_claims {
4623 claim
4624 .transaction_order
4625 .checked_add(1)
4626 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4627 }
4628 if let Some(recovery) = &order.recovery {
4629 if let Some(active_binding) = &recovery.active_binding {
4630 active_binding
4631 .transaction_order
4632 .checked_add(1)
4633 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4634 }
4635 recovery
4636 .recovery_operation_order
4637 .checked_add(1)
4638 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4639 recovery
4640 .replacement_terminal_order
4641 .checked_add(1)
4642 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4643 }
4644 Ok(())
4645}
4646
4647fn relay_ordinary_order_owners(order: &mut OrderClaimFrontier) {
4648 for claim in &mut order.movable_claims {
4649 claim.transaction_order = claim.transaction_order.wrapping_add(1);
4650 }
4651 if let Some(recovery) = &mut order.recovery {
4652 if let Some(active_binding) = &mut recovery.active_binding {
4653 active_binding.transaction_order = active_binding.transaction_order.wrapping_add(1);
4654 }
4655 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_add(1);
4656 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_add(1);
4657 }
4658}
4659
4660fn rollback_ordinary_order_owners(order: &mut OrderClaimFrontier) {
4661 for claim in &mut order.movable_claims {
4662 claim.transaction_order = claim.transaction_order.wrapping_sub(1);
4663 }
4664 if let Some(recovery) = &mut order.recovery {
4665 if let Some(active_binding) = &mut recovery.active_binding {
4666 active_binding.transaction_order = active_binding.transaction_order.wrapping_sub(1);
4667 }
4668 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_sub(1);
4669 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_sub(1);
4670 }
4671}
4672
4673fn validate_leave_identity<F>(
4691 frontiers: &ClaimFrontiers,
4692 member: &LiveMember<F>,
4693 settled_detached: bool,
4694) -> Result<(), PrepareLeaveAuthorityError> {
4695 if member.conversation_id() != frontiers.conversation_id {
4696 return Err(PrepareLeaveAuthorityError::Conversation);
4697 }
4698 let Some(participant) =
4699 active_participant(&frontiers.active_identities, member.participant_id())
4700 else {
4701 return Err(PrepareLeaveAuthorityError::Identity);
4702 };
4703 let cursor_matches = if settled_detached {
4704 participant.cursor >= member.cursor()
4705 } else {
4706 participant.cursor == member.cursor()
4707 };
4708 if !cursor_matches {
4709 return Err(PrepareLeaveAuthorityError::Identity);
4710 }
4711 Ok(())
4712}
4713
4714fn validate_settled_leave_prestate<F>(
4715 frontiers: &ClaimFrontiers,
4716 member: &LiveMember<F>,
4717 binding_state: BindingState,
4718) -> Result<(ParticipantId, Option<BindingEpoch>), PrepareLeaveAuthorityError> {
4719 let participant_id = member.participant_id();
4720 validate_leave_identity(
4721 frontiers,
4722 member,
4723 matches!(binding_state, BindingState::Detached),
4724 )?;
4725 if !frontiers.order.immutable_candidates.is_empty() {
4726 return Err(PrepareLeaveAuthorityError::ImmutablePrefix);
4727 }
4728 let ended_binding_epoch = match binding_state {
4729 BindingState::Detached => {
4730 let Some(participant) =
4731 active_participant(&frontiers.active_identities, participant_id)
4732 else {
4733 return Err(PrepareLeaveAuthorityError::Identity);
4734 };
4735 if !matches!(participant.binding, FrontierBinding::Detached(_)) {
4736 return Err(PrepareLeaveAuthorityError::Binding);
4737 }
4738 None
4739 }
4740 BindingState::Bound(binding)
4741 if binding.conversation_id == frontiers.conversation_id
4742 && binding.participant_id == participant_id =>
4743 {
4744 let Some(participant) =
4745 active_participant(&frontiers.active_identities, participant_id)
4746 else {
4747 return Err(PrepareLeaveAuthorityError::Identity);
4748 };
4749 if participant.binding != FrontierBinding::Bound(binding.binding_epoch) {
4750 return Err(PrepareLeaveAuthorityError::Binding);
4751 }
4752 Some(binding.binding_epoch)
4753 }
4754 BindingState::Bound(_) | BindingState::PendingFinalization(_) => {
4755 return Err(PrepareLeaveAuthorityError::Binding);
4756 }
4757 };
4758 Ok((participant_id, ended_binding_epoch))
4759}
4760
4761fn validate_pending_leave_prestate<F>(
4762 frontiers: &ClaimFrontiers,
4763 member: &LiveMember<F>,
4764 pending: PendingFinalization,
4765) -> Result<(ParticipantId, super::AdmissionOrder), PrepareLeaveAuthorityError> {
4766 let participant_id = member.participant_id();
4767 validate_leave_identity(frontiers, member, false)?;
4770 if pending.conversation_id() != frontiers.conversation_id
4771 || pending.participant_id() != participant_id
4772 {
4773 return Err(PrepareLeaveAuthorityError::Binding);
4774 }
4775 let Some(participant) = active_participant(&frontiers.active_identities, participant_id) else {
4776 return Err(PrepareLeaveAuthorityError::Identity);
4777 };
4778 if participant.binding != FrontierBinding::Detached(pending.binding_epoch()) {
4779 return Err(PrepareLeaveAuthorityError::Binding);
4780 }
4781 let expected_order = pending.admission_order();
4782 let exact_sequence_candidate = matches!(
4783 frontiers.sequence.immutable_candidates.as_slice(),
4784 [ImmutableSequenceCandidate::BindingTerminal {
4785 admission_order,
4786 owner,
4787 ..
4788 }] if *admission_order == expected_order
4789 && owner.participant_index == participant_id
4790 && owner.binding_epoch == pending.binding_epoch()
4791 );
4792 let exact_order_candidate = matches!(
4793 frontiers.order.immutable_candidates.as_slice(),
4794 [ImmutableOrderCandidateMajor {
4795 transaction_order,
4796 candidate_keys,
4797 }] if *transaction_order == expected_order.transaction_order()
4798 && candidate_keys.as_slice() == [expected_order]
4799 );
4800 if !exact_sequence_candidate || !exact_order_candidate {
4801 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4802 }
4803 Ok((participant_id, expected_order))
4804}
4805
4806#[derive(Clone, Copy)]
4807enum LeaveRelayUnit {
4808 Direct(MovableOrderClaim),
4809 Recovery(RecoveryOrderBlock),
4810}
4811
4812impl LeaveRelayUnit {
4813 fn start(self) -> TransactionOrder {
4814 match self {
4815 Self::Direct(claim) => claim.transaction_order,
4816 Self::Recovery(block) => block_start_validated_order(block),
4817 }
4818 }
4819
4820 const fn len(self) -> u64 {
4821 match self {
4822 Self::Direct(_) => 1,
4823 Self::Recovery(block) => match block.active_binding {
4824 Some(_) => 3,
4825 None => 2,
4826 },
4827 }
4828 }
4829}
4830
4831struct LeaveOrderSelection {
4832 units: Vec<LeaveRelayUnit>,
4833 selected_major: TransactionOrder,
4834}
4835
4836fn select_leave_order(
4837 order: &OrderClaimFrontier,
4838 participant_id: ParticipantId,
4839 ended_binding_epoch: Option<BindingEpoch>,
4840 pending_order: Option<super::AdmissionOrder>,
4841) -> Result<LeaveOrderSelection, PrepareLeaveAuthorityError> {
4842 let Some(exit_index) = order.movable_claims.iter().position(|claim| {
4843 claim.owner
4844 == OrderDirectOwner::MembershipExit {
4845 participant_index: participant_id,
4846 }
4847 }) else {
4848 return Err(PrepareLeaveAuthorityError::MembershipExitClaim);
4849 };
4850 let exit_claim = order.movable_claims[exit_index];
4851 if pending_order
4852 .is_some_and(|pending| pending.transaction_order() >= exit_claim.transaction_order)
4853 {
4854 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4855 }
4856 let active_index = matching_active_claim(order, participant_id, ended_binding_epoch)?;
4857 let mut units = Vec::new();
4858 for (index, claim) in order.movable_claims.iter().copied().enumerate() {
4859 if index != exit_index && Some(index) != active_index {
4860 units.push(LeaveRelayUnit::Direct(claim));
4861 }
4862 }
4863 if let Some(recovery) = order
4864 .recovery
4865 .filter(|recovery| recovery.participant_index != participant_id)
4866 {
4867 units.push(LeaveRelayUnit::Recovery(recovery));
4868 }
4869 units.sort_by_key(|unit| unit.start());
4870 let surviving_handles: u128 = units.iter().map(|unit| u128::from(unit.len())).sum();
4871 let exit_major = exit_claim.transaction_order;
4872 let later_handle_fits = u128::from(u64::MAX - exit_major) >= surviving_handles;
4873 let selected_major = if later_handle_fits {
4874 exit_major
4875 } else if pending_order.is_some() {
4876 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4877 } else {
4878 u64::try_from(order_frontier_start(order.ledger.high()))
4879 .map_err(|_| PrepareLeaveAuthorityError::OrderCapacity)?
4880 };
4881 Ok(LeaveOrderSelection {
4882 units,
4883 selected_major,
4884 })
4885}
4886
4887fn matching_active_claim(
4888 order: &OrderClaimFrontier,
4889 participant_id: ParticipantId,
4890 ended_binding_epoch: Option<BindingEpoch>,
4891) -> Result<Option<usize>, PrepareLeaveAuthorityError> {
4892 let Some(binding_epoch) = ended_binding_epoch else {
4893 return Ok(None);
4894 };
4895 let expected = OrderDirectOwner::ActiveBindingTerminal(BindingTerminalOwner {
4896 participant_index: participant_id,
4897 binding_epoch,
4898 });
4899 order
4900 .movable_claims
4901 .iter()
4902 .position(|claim| claim.owner == expected)
4903 .map(Some)
4904 .ok_or(PrepareLeaveAuthorityError::ActiveBindingClaim)
4905}
4906
4907fn relay_leave_order_units(
4908 units: Vec<LeaveRelayUnit>,
4909 selected_major: TransactionOrder,
4910) -> Result<(Vec<MovableOrderClaim>, Option<RecoveryOrderBlock>), PrepareLeaveAuthorityError> {
4911 let mut cursor = selected_major.checked_add(1);
4912 let mut movable_claims = Vec::new();
4913 let mut recovery = None;
4914 for unit in units {
4915 match unit {
4916 LeaveRelayUnit::Direct(mut claim) => {
4917 let Some(position) = cursor else {
4918 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4919 };
4920 claim.transaction_order = position;
4921 movable_claims.push(claim);
4922 cursor = position.checked_add(1);
4923 }
4924 LeaveRelayUnit::Recovery(block) => {
4925 let (relayed, next) = relay_recovery_block(block, cursor)?;
4926 recovery = Some(relayed);
4927 cursor = next;
4928 }
4929 }
4930 }
4931 movable_claims.sort_by_key(|claim| claim.transaction_order);
4932 Ok((movable_claims, recovery))
4933}
4934
4935const fn relay_recovery_block(
4936 block: RecoveryOrderBlock,
4937 cursor: Option<TransactionOrder>,
4938) -> Result<(RecoveryOrderBlock, Option<TransactionOrder>), PrepareLeaveAuthorityError> {
4939 let mut next = cursor;
4940 let active_binding = if let Some(mut active) = block.active_binding {
4941 let Some(position) = next else {
4942 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4943 };
4944 active.transaction_order = position;
4945 next = position.checked_add(1);
4946 Some(active)
4947 } else {
4948 None
4949 };
4950 let Some(recovery_operation_order) = next else {
4951 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4952 };
4953 let Some(replacement_terminal_order) = recovery_operation_order.checked_add(1) else {
4954 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4955 };
4956 Ok((
4957 RecoveryOrderBlock {
4958 active_binding,
4959 recovery_operation_order,
4960 replacement_terminal_order,
4961 participant_index: block.participant_index,
4962 marker_delivery_seq: block.marker_delivery_seq,
4963 recovered_binding_epoch: block.recovered_binding_epoch,
4964 },
4965 replacement_terminal_order.checked_add(1),
4966 ))
4967}
4968
4969fn leave_resulting_order_ledger(
4970 selected_major: TransactionOrder,
4971 movable_claims: &[MovableOrderClaim],
4972 recovery: Option<RecoveryOrderBlock>,
4973) -> Result<OrderLedger, PrepareLeaveAuthorityError> {
4974 let active_binding_terminals =
4975 usize_to_u64(
4976 movable_claims
4977 .iter()
4978 .filter(|claim| matches!(claim.owner, OrderDirectOwner::ActiveBindingTerminal(_)))
4979 .count(),
4980 ) + u64::from(recovery.is_some_and(|block| block.active_binding.is_some()));
4981 let membership_exits = usize_to_u64(
4982 movable_claims
4983 .iter()
4984 .filter(|claim| matches!(claim.owner, OrderDirectOwner::MembershipExit { .. }))
4985 .count(),
4986 );
4987 let has_recovery = recovery.is_some();
4988 let resulting_claims = OrderClaims::new(
4989 active_binding_terminals,
4990 membership_exits,
4991 has_recovery,
4992 has_recovery,
4993 )
4994 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)?;
4995 OrderLedger::try_new(OrderHigh::Allocated(selected_major), resulting_claims)
4996 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)
4997}
4998
4999fn consume_leave_order_lane(
5000 order: &mut OrderClaimFrontier,
5001 participant_id: ParticipantId,
5002 ended_binding_epoch: Option<BindingEpoch>,
5003 pending_order: Option<super::AdmissionOrder>,
5004) -> Result<TransactionOrder, PrepareLeaveAuthorityError> {
5005 let selection = select_leave_order(order, participant_id, ended_binding_epoch, pending_order)?;
5006 let (movable_claims, recovery) =
5007 relay_leave_order_units(selection.units, selection.selected_major)?;
5008 let ledger = leave_resulting_order_ledger(selection.selected_major, &movable_claims, recovery)?;
5009 *order = OrderClaimFrontier {
5010 ledger,
5011 movable_claims,
5012 immutable_candidates: Vec::new(),
5013 recovery,
5014 };
5015 Ok(selection.selected_major)
5016}
5017
5018impl ClaimFrontiersPrevalidated {
5019 #[must_use]
5021 pub(super) const fn conversation_id(&self) -> ConversationId {
5022 self.conversation_id
5023 }
5024
5025 pub(super) fn take_marker_record(
5031 &mut self,
5032 request: MarkerRecordRequest,
5033 ) -> Option<ValidatedMarkerRecord> {
5034 if self.issued_marker_record.is_some() {
5035 return None;
5036 }
5037 let record = self
5038 .retained_records
5039 .iter()
5040 .find(|record| record.delivery_seq == request.marker_delivery_seq)
5041 .copied()?;
5042 let RetainedCausalRecordKind::CompactionMarker {
5043 participant_index,
5044 provenance,
5045 } = record.kind
5046 else {
5047 return None;
5048 };
5049 if participant_index != request.participant_index {
5050 return None;
5051 }
5052 let participant = active_participant(&self.active_identities, participant_index)?;
5053 let historical_delivery = self
5054 .historical_marker_deliveries
5055 .iter()
5056 .find(|authority| authority.marker_delivery_seq == request.marker_delivery_seq);
5057 let (target_binding, occurrence) = match request.use_kind {
5058 MarkerRecordUse::Planned(target) => {
5059 if participant.binding != target || historical_delivery.is_some() {
5060 return None;
5061 }
5062 (target, MarkerRecordOccurrence::Undelivered)
5063 }
5064 MarkerRecordUse::Delivered(target) => {
5065 let delivered_binding_epoch = binding_epoch(target);
5066 if participant.binding != target
5067 || !historical_delivery.is_some_and(|authority| {
5068 authority.participant_index == participant_index
5069 && authority.delivered_binding_epoch == delivered_binding_epoch
5070 })
5071 {
5072 return None;
5073 }
5074 (target, MarkerRecordOccurrence::Delivered)
5075 }
5076 MarkerRecordUse::Recovered {
5077 prior_binding_epoch,
5078 recovered_binding_epoch,
5079 } => {
5080 if participant.binding != FrontierBinding::Detached(recovered_binding_epoch)
5081 || !historical_delivery.is_some_and(|authority| {
5082 authority.participant_index == participant_index
5083 && authority.delivered_binding_epoch == prior_binding_epoch
5084 })
5085 {
5086 return None;
5087 }
5088 (
5089 FrontierBinding::Detached(prior_binding_epoch),
5090 MarkerRecordOccurrence::Delivered,
5091 )
5092 }
5093 };
5094 self.issued_marker_record = Some(request);
5095 Some(ValidatedMarkerRecord {
5096 conversation_id: self.conversation_id,
5097 record,
5098 provenance,
5099 target_binding,
5100 occurrence,
5101 seal: MarkerAuthoritySeal::Validated,
5102 })
5103 }
5104
5105 pub(super) fn finish(
5108 self,
5109 edge: Option<super::StoredEdge>,
5110 ) -> Result<ClaimFrontiers, ParticipantStateCorruptReason> {
5111 self.validate_current_marker_edge(edge)?;
5112 self.validate_historical_delivery_consumers(edge)?;
5113 let recovery_provenance = self.resolve_recovery_provenance(edge)?;
5114 let sequence = restore_sequence_frontier(
5115 &self.active_identities,
5116 self.sequence_restore,
5117 self.retained_floor,
5118 self.sequence_ledger,
5119 recovery_provenance,
5120 &self.retained_records,
5121 &self.historical_causal_authorities,
5122 )
5123 .map_err(corrupt_frontier)?;
5124 let order = restore_order_frontier(
5125 &self.active_identities,
5126 self.order_restore,
5127 self.order_ledger,
5128 recovery_provenance,
5129 )
5130 .map_err(corrupt_frontier)?;
5131 validate_cross_counter(&sequence, &order).map_err(corrupt_frontier)?;
5132 Ok(ClaimFrontiers {
5133 ordinary_cursor_release: None,
5134 conversation_id: self.conversation_id,
5135 active_identities: self.active_identities,
5136 identity_slot_limit: self.identity_slot_limit,
5137 retained_floor: self.retained_floor,
5138 retained_records: self.retained_records,
5139 marker_records: self.marker_records,
5140 fenced_marker_issued: self.fenced_marker_issued,
5141 sequence,
5142 order,
5143 historical_marker_deliveries: self.historical_marker_deliveries,
5144 historical_causal_authorities: self.historical_causal_authorities,
5145 })
5146 }
5147
5148 fn validate_historical_delivery_consumers(
5149 &self,
5150 edge: Option<super::StoredEdge>,
5151 ) -> Result<(), ParticipantStateCorruptReason> {
5152 for history in &self.historical_marker_deliveries {
5153 let recovered_origin = self.binding_origins.iter().any(|origin| {
5154 origin.participant_id() == history.participant_index
5155 && origin.recovered_marker()
5156 == Some((history.marker_delivery_seq, history.delivered_binding_epoch))
5157 });
5158 let current_marker_edge = match edge {
5159 Some(super::StoredEdge::ParticipantCursorProgress(progress)) => {
5160 progress.participant_id() == history.participant_index
5161 && progress.marker_delivery_seq() == Some(history.marker_delivery_seq)
5162 && progress.binding_epoch() == history.delivered_binding_epoch
5163 }
5164 Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) => {
5165 recovery.participant_id() == history.participant_index
5166 && recovery.marker_delivery_seq() == history.marker_delivery_seq
5167 && recovery.prior_binding_epoch() == history.delivered_binding_epoch
5168 }
5169 Some(
5170 super::StoredEdge::ObserverProjection(_)
5171 | super::StoredEdge::PhysicalCompaction(_)
5172 | super::StoredEdge::MarkerDelivery(_)
5173 | super::StoredEdge::DetachedMarkerRelease(_)
5174 | super::StoredEdge::DetachedCursorRelease(_),
5175 )
5176 | None => false,
5177 };
5178 if !recovered_origin && !current_marker_edge {
5179 return Err(self.marker_corruption(history.marker_delivery_seq));
5180 }
5181 }
5182 Ok(())
5183 }
5184
5185 fn validate_current_marker_edge(
5186 &self,
5187 edge: Option<super::StoredEdge>,
5188 ) -> Result<(), ParticipantStateCorruptReason> {
5189 if let Some(MarkerRecordRequest {
5190 participant_index,
5191 marker_delivery_seq,
5192 use_kind:
5193 MarkerRecordUse::Recovered {
5194 recovered_binding_epoch,
5195 ..
5196 },
5197 }) = self.issued_marker_record
5198 {
5199 let recovered_edge_matches = matches!(
5200 edge,
5201 Some(super::StoredEdge::DetachedCursorRelease(release))
5202 if release.participant_id() == participant_index
5203 && release.last_dead_binding_epoch() == recovered_binding_epoch
5204 );
5205 if !recovered_edge_matches {
5206 return Err(self.marker_corruption(marker_delivery_seq));
5207 }
5208 return Ok(());
5209 }
5210 let Some(context) = edge.and_then(marker_edge_context) else {
5211 return Ok(());
5212 };
5213 if self
5214 .issued_marker_record
5215 .is_none_or(|request| request.marker_delivery_seq != context.marker_delivery_seq)
5216 || !self.marker_context_matches(context)
5217 {
5218 return Err(self.marker_corruption(context.marker_delivery_seq));
5219 }
5220 Ok(())
5221 }
5222
5223 fn resolve_recovery_provenance(
5224 &self,
5225 edge: Option<super::StoredEdge>,
5226 ) -> Result<Option<RecoveryClaimProvenance>, ParticipantStateCorruptReason> {
5227 let Some(marker_delivery_seq) = self.recovery_marker_delivery_seq else {
5228 return Ok(None);
5229 };
5230
5231 if let Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) = edge {
5232 return self
5233 .resolve_postfate_recovery_provenance(marker_delivery_seq, recovery)
5234 .map(Some);
5235 }
5236
5237 let candidate =
5238 self.sequence_restore.immutable_candidates.iter().find_map(
5239 |candidate| match candidate {
5240 ImmutableSequenceCandidate::Marker(marker)
5241 if marker.delivery_seq == marker_delivery_seq
5242 && matches!(marker.target_binding, FrontierBinding::Bound(_)) =>
5243 {
5244 Some(*marker)
5245 }
5246 _ => None,
5247 },
5248 );
5249 let retained = self.retained_records.iter().find_map(|record| {
5250 let RetainedCausalRecordKind::CompactionMarker {
5251 participant_index, ..
5252 } = record.kind
5253 else {
5254 return None;
5255 };
5256 if record.delivery_seq != marker_delivery_seq {
5257 return None;
5258 }
5259 let participant = active_participant(&self.active_identities, participant_index)?;
5260 let FrontierBinding::Bound(binding_epoch) = participant.binding else {
5261 return None;
5262 };
5263 let historical = self.historical_marker_deliveries.iter().find(|history| {
5264 history.participant_index == participant_index
5265 && history.marker_delivery_seq == marker_delivery_seq
5266 })?;
5267 Some((
5268 participant_index,
5269 binding_epoch,
5270 historical.delivered_binding_epoch,
5271 ))
5272 });
5273 let provenance = match (candidate, retained) {
5274 (Some(marker), None) => RecoveryClaimProvenance {
5275 participant_index: marker.admission_order.participant_index(),
5276 marker_delivery_seq,
5277 prior_binding_epoch: binding_epoch(marker.target_binding),
5278 current_binding_epoch: binding_epoch(marker.target_binding),
5279 phase: RecoveryClaimPhase::PreFate,
5280 },
5281 (None, Some((participant_index, current_binding_epoch, prior_binding_epoch))) => {
5282 let phase = if current_binding_epoch == prior_binding_epoch {
5283 let Some(context) = edge.and_then(marker_edge_context) else {
5284 return Err(self.marker_corruption(marker_delivery_seq));
5285 };
5286 if context.marker_delivery_seq != marker_delivery_seq
5287 || context.participant_index != participant_index
5288 || context.binding_epoch != prior_binding_epoch
5289 || context.target_binding != FrontierBinding::Bound(prior_binding_epoch)
5290 {
5291 return Err(self.marker_corruption(marker_delivery_seq));
5292 }
5293 RecoveryClaimPhase::PreFate
5294 } else {
5295 let recovered_origin_matches = self.binding_origins.iter().any(|origin| {
5296 origin.participant_id() == participant_index
5297 && origin.binding_epoch() == current_binding_epoch
5298 && origin.recovered_marker()
5299 == Some((marker_delivery_seq, prior_binding_epoch))
5300 });
5301 if !recovered_origin_matches
5302 || !matches!(
5303 edge,
5304 Some(
5305 super::StoredEdge::ObserverProjection(_)
5306 | super::StoredEdge::PhysicalCompaction(_)
5307 )
5308 )
5309 {
5310 return Err(self.marker_corruption(marker_delivery_seq));
5311 }
5312 RecoveryClaimPhase::RecoveredBound
5313 };
5314 RecoveryClaimProvenance {
5315 participant_index,
5316 marker_delivery_seq,
5317 prior_binding_epoch,
5318 current_binding_epoch,
5319 phase,
5320 }
5321 }
5322 (Some(_), Some(_)) | (None, None) => {
5323 return Err(self.marker_corruption(marker_delivery_seq));
5324 }
5325 };
5326 Ok(Some(provenance))
5327 }
5328
5329 fn resolve_postfate_recovery_provenance(
5330 &self,
5331 marker_delivery_seq: DeliverySeq,
5332 recovery: super::DetachedCredentialRecovery,
5333 ) -> Result<RecoveryClaimProvenance, ParticipantStateCorruptReason> {
5334 let provenance = RecoveryClaimProvenance {
5335 participant_index: recovery.participant_id(),
5336 marker_delivery_seq: recovery.marker_delivery_seq(),
5337 prior_binding_epoch: recovery.prior_binding_epoch(),
5338 current_binding_epoch: recovery.prior_binding_epoch(),
5339 phase: RecoveryClaimPhase::PostFate,
5340 };
5341 let context = MarkerEdgeContext {
5342 participant_index: provenance.participant_index,
5343 marker_delivery_seq: provenance.marker_delivery_seq,
5344 binding_epoch: provenance.prior_binding_epoch,
5345 target_binding: FrontierBinding::Detached(provenance.prior_binding_epoch),
5346 };
5347 if marker_delivery_seq != provenance.marker_delivery_seq
5348 || self
5349 .issued_marker_record
5350 .is_none_or(|request| request.marker_delivery_seq != marker_delivery_seq)
5351 || !self.marker_context_matches(context)
5352 {
5353 return Err(self.marker_corruption(marker_delivery_seq));
5354 }
5355 Ok(provenance)
5356 }
5357
5358 fn marker_context_matches(&self, context: MarkerEdgeContext) -> bool {
5359 self.marker_records.iter().any(|record| {
5360 matches!(
5361 record.kind,
5362 RetainedCausalRecordKind::CompactionMarker {
5363 participant_index,
5364 ..
5365 } if participant_index == context.participant_index
5366 ) && record.delivery_seq == context.marker_delivery_seq
5367 && active_participant(&self.active_identities, context.participant_index)
5368 .is_some_and(|participant| participant.binding == context.target_binding)
5369 })
5370 }
5371
5372 fn marker_corruption(&self, delivery_seq: DeliverySeq) -> ParticipantStateCorruptReason {
5373 corrupt_frontier(sequence_error(
5374 sequence_ordinal(self.sequence_ledger, delivery_seq),
5375 ClaimFrontierInvalidReason::RecoveryBlock,
5376 ))
5377 }
5378}
5379
5380#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5381struct MarkerEdgeContext {
5382 participant_index: ParticipantId,
5383 marker_delivery_seq: DeliverySeq,
5384 binding_epoch: BindingEpoch,
5385 target_binding: FrontierBinding,
5386}
5387
5388fn marker_edge_context(edge: super::StoredEdge) -> Option<MarkerEdgeContext> {
5389 match edge {
5390 super::StoredEdge::MarkerDelivery(delivery) => Some(MarkerEdgeContext {
5391 participant_index: delivery.participant_id(),
5392 marker_delivery_seq: delivery.marker_delivery_seq(),
5393 binding_epoch: delivery.binding_epoch(),
5394 target_binding: FrontierBinding::Bound(delivery.binding_epoch()),
5395 }),
5396 super::StoredEdge::ParticipantCursorProgress(progress) => {
5397 let marker_delivery_seq = progress.marker_delivery_seq()?;
5398 Some(MarkerEdgeContext {
5399 participant_index: progress.participant_id(),
5400 marker_delivery_seq,
5401 binding_epoch: progress.binding_epoch(),
5402 target_binding: FrontierBinding::Bound(progress.binding_epoch()),
5403 })
5404 }
5405 super::StoredEdge::DetachedCredentialRecovery(recovery) => Some(MarkerEdgeContext {
5406 participant_index: recovery.participant_id(),
5407 marker_delivery_seq: recovery.marker_delivery_seq(),
5408 binding_epoch: recovery.prior_binding_epoch(),
5409 target_binding: FrontierBinding::Detached(recovery.prior_binding_epoch()),
5410 }),
5411 super::StoredEdge::DetachedMarkerRelease(release) => Some(MarkerEdgeContext {
5412 participant_index: release.participant_id(),
5413 marker_delivery_seq: release.marker_delivery_seq(),
5414 binding_epoch: release.last_dead_binding_epoch(),
5415 target_binding: FrontierBinding::Detached(release.last_dead_binding_epoch()),
5416 }),
5417 super::StoredEdge::ObserverProjection(_)
5418 | super::StoredEdge::PhysicalCompaction(_)
5419 | super::StoredEdge::DetachedCursorRelease(_) => None,
5420 }
5421}
5422
5423#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5424#[repr(u8)]
5425enum SequenceClass {
5426 Exit = 0,
5427 Terminal = 1,
5428 Marker = 2,
5429 RecoveryAttach = 3,
5430 RecoveryReplacementTerminal = 4,
5431 LiveTimesTerminal = 5,
5432 LiveTimesReplacementTerminal = 6,
5433 OtherLiveTimesExit = 7,
5434}
5435
5436#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5437#[repr(u8)]
5438enum OrderClass {
5439 ActiveBindingTerminal = 0,
5440 MembershipExit = 1,
5441 RecoveryOperation = 2,
5442 RecoveryReplacementTerminal = 3,
5443}
5444
5445#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5446struct NumericSegment<C> {
5447 start: u128,
5448 length: u128,
5449 class: Option<C>,
5450 immutable: bool,
5451}
5452
5453fn first_duplicate_candidate_key(
5454 candidates: &[ImmutableSequenceCandidate],
5455 retained_records: &[RetainedCausalRecord],
5456) -> Option<super::AdmissionOrder> {
5457 let mut keys: Vec<_> = candidates
5458 .iter()
5459 .map(|candidate| candidate.admission_order())
5460 .chain(retained_records.iter().map(|record| record.admission_order))
5461 .collect();
5462 keys.sort_unstable();
5463 let mut previous = None;
5464 for key in keys {
5465 if previous == Some(key) {
5466 return Some(key);
5467 }
5468 previous = Some(key);
5469 }
5470 None
5471}
5472
5473fn validate_unique_candidate_keys(
5474 candidates: &[ImmutableSequenceCandidate],
5475 retained_records: &[RetainedCausalRecord],
5476) -> Result<(), ParticipantStateCorruptReason> {
5477 let Some(order) = first_duplicate_candidate_key(candidates, retained_records) else {
5478 return Ok(());
5479 };
5480 Err(ParticipantStateCorruptReason::DuplicateCandidateKey {
5481 transaction_order: order.transaction_order(),
5482 candidate_phase: order.candidate_phase(),
5483 participant_index: order.participant_index(),
5484 })
5485}
5486
5487fn validate_sequence_numeric(
5488 restore: &SequenceClaimFrontierRestore,
5489 ledger: SequenceLedger,
5490) -> Result<(), ClaimFrontierError> {
5491 let mut segments = sequence_segments(restore);
5492 validate_numeric_segments(
5493 ClaimFrontierCounter::DeliverySequence,
5494 u128::from(ledger.high_watermark()) + 1,
5495 ledger.required_reserve(),
5496 &mut segments,
5497 &sequence_expected_counts(ledger),
5498 )
5499}
5500
5501fn validate_order_numeric(
5502 restore: &OrderClaimFrontierRestore,
5503 ledger: OrderLedger,
5504) -> Result<(), ClaimFrontierError> {
5505 let mut segments = order_segments(restore, ledger.high());
5506 validate_numeric_segments(
5507 ClaimFrontierCounter::TransactionOrder,
5508 order_frontier_start(ledger.high()),
5509 order_frontier_candidate_count(restore, ledger.high()) + ledger.claims().total(),
5510 &mut segments,
5511 &order_expected_counts(ledger),
5512 )
5513}
5514
5515fn validate_bounded_shape(
5516 restore: &ClaimFrontiersRestore,
5517 sequence_ledger: SequenceLedger,
5518) -> Result<(), ClaimFrontierError> {
5519 let identity_limit = u128::from(restore.identity_slot_limit);
5520 let twice_identity_limit = identity_limit.saturating_mul(2);
5521 let order_candidate_keys = restore
5522 .order
5523 .immutable_candidates
5524 .iter()
5525 .fold(0_u128, |count, candidate| {
5526 count.saturating_add(usize_to_u128(candidate.candidate_keys.len()))
5527 });
5528 let bounded = usize_to_u128(restore.active_identities.len()) <= identity_limit
5529 && usize_to_u128(restore.sequence.movable_claims.len()) <= twice_identity_limit
5530 && usize_to_u128(restore.sequence.immutable_candidates.len()) <= twice_identity_limit
5531 && usize_to_u128(restore.sequence.products.live_times_terminal.len()) <= identity_limit
5532 && usize_to_u128(restore.sequence.products.other_live_times_exit.len()) <= identity_limit
5533 && usize_to_u128(restore.historical_marker_deliveries.len())
5534 <= u128::from(restore.retained_record_limit)
5535 && usize_to_u128(restore.historical_causal_facts.len()) <= twice_identity_limit
5536 && usize_to_u128(restore.order.movable_claims.len()) <= twice_identity_limit
5537 && usize_to_u128(restore.order.immutable_candidates.len()) <= twice_identity_limit
5538 && order_candidate_keys <= twice_identity_limit;
5539 if bounded {
5540 Ok(())
5541 } else {
5542 Err(sequence_error(
5543 sequence_ledger.required_reserve(),
5544 ClaimFrontierInvalidReason::LogicalOwner,
5545 ))
5546 }
5547}
5548
5549fn validated_retained_records(
5550 mut records: Vec<RetainedCausalRecord>,
5551 retained_floor: u128,
5552 retained_record_limit: u64,
5553 identity_slot_limit: u64,
5554 ledger: SequenceLedger,
5555) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
5556 let high_end = u128::from(ledger.high_watermark()) + 1;
5557 let expected_count = high_end.saturating_sub(retained_floor);
5558 let record_bound =
5568 u128::from(retained_record_limit).saturating_add(u128::from(identity_slot_limit));
5569 if retained_floor > high_end
5570 || usize_to_u128(records.len()) > record_bound
5571 || usize_to_u128(records.len()) != expected_count
5572 {
5573 return Err(sequence_error(
5574 usize_to_u128(records.len()).min(expected_count),
5575 ClaimFrontierInvalidReason::LogicalOwner,
5576 ));
5577 }
5578 records.sort_by_key(|record| record.delivery_seq);
5579 let mut previous_admission_order = None;
5580 for (record_index, record) in records.iter().enumerate() {
5581 let (participant_index, valid_kind) = match record.kind {
5582 RetainedCausalRecordKind::BindingTerminal(owner) => (
5583 owner.participant_index,
5584 record.admission_order.candidate_phase() == CandidatePhase::BindingTerminal
5585 && record.admission_order.participant_index() == owner.participant_index,
5586 ),
5587 RetainedCausalRecordKind::MembershipExit { participant_index } => (
5588 participant_index,
5589 record.admission_order.candidate_phase() == CandidatePhase::MembershipExit
5590 && record.admission_order.participant_index() == participant_index,
5591 ),
5592 RetainedCausalRecordKind::AttachLifecycle {
5593 participant_index, ..
5594 } => (
5595 participant_index,
5596 record.admission_order.candidate_phase() == CandidatePhase::AttachLifecycle
5597 && record.admission_order.participant_index() == participant_index,
5598 ),
5599 RetainedCausalRecordKind::OrdinaryRecord { participant_index } => (
5600 participant_index,
5601 record.admission_order.candidate_phase() == CandidatePhase::OrdinaryRecord
5602 && record.admission_order.participant_index() == participant_index,
5603 ),
5604 RetainedCausalRecordKind::CompactionMarker {
5605 participant_index,
5606 provenance,
5607 } => (
5608 participant_index,
5609 record.admission_order.candidate_phase() == CandidatePhase::CompactionMarker
5610 && record.admission_order.participant_index() == participant_index
5611 && marker_provenance_targets(provenance, participant_index),
5612 ),
5613 };
5614 let expected_sequence = retained_floor + rank_index(record_index);
5615 if participant_index >= identity_slot_limit
5616 || !valid_kind
5617 || previous_admission_order.is_some_and(|previous| previous >= record.admission_order)
5618 || u128::from(record.delivery_seq) != expected_sequence
5619 {
5620 return Err(sequence_error(
5621 rank_index(record_index),
5622 ClaimFrontierInvalidReason::CandidateKey,
5623 ));
5624 }
5625 previous_admission_order = Some(record.admission_order);
5626 }
5627 Ok(records)
5628}
5629
5630fn validated_active_marker_records(
5631 retained_markers: &[RetainedCausalRecord],
5632 mut active_marker_anchors: Vec<DeliverySeq>,
5633 identity_slot_limit: u64,
5634 ledger: SequenceLedger,
5635) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
5636 if usize_to_u128(active_marker_anchors.len()) > u128::from(identity_slot_limit) {
5637 return Err(sequence_error(
5638 ledger.required_reserve(),
5639 ClaimFrontierInvalidReason::LogicalOwner,
5640 ));
5641 }
5642 active_marker_anchors.sort_unstable();
5643 let mut previous_sequence = None;
5644 let mut owners = Vec::new();
5645 let mut active_records = Vec::new();
5646 for delivery_seq in active_marker_anchors {
5647 let Some(record) = retained_markers
5648 .iter()
5649 .find(|record| record.delivery_seq == delivery_seq)
5650 .copied()
5651 else {
5652 return Err(sequence_error(
5653 sequence_ordinal(ledger, delivery_seq),
5654 ClaimFrontierInvalidReason::LogicalOwner,
5655 ));
5656 };
5657 let RetainedCausalRecordKind::CompactionMarker {
5658 participant_index, ..
5659 } = record.kind
5660 else {
5661 return Err(sequence_error(
5662 sequence_ordinal(ledger, delivery_seq),
5663 ClaimFrontierInvalidReason::LogicalOwner,
5664 ));
5665 };
5666 if previous_sequence == Some(delivery_seq) || owners.contains(&participant_index) {
5667 return Err(sequence_error(
5668 sequence_ordinal(ledger, delivery_seq),
5669 ClaimFrontierInvalidReason::LogicalOwner,
5670 ));
5671 }
5672 previous_sequence = Some(delivery_seq);
5673 owners.push(participant_index);
5674 active_records.push(record);
5675 }
5676 Ok(active_records)
5677}
5678
5679fn validated_historical_marker_deliveries(
5680 mut facts: Vec<HistoricalMarkerDeliveryFactRestore>,
5681 conversation_id: ConversationId,
5682 active: &ActiveIdentityRanks,
5683 sources: &MarkerDeliverySources<'_>,
5684 retained_record_limit: u64,
5685 ledger: SequenceLedger,
5686) -> Result<Vec<HistoricalMarkerDeliveryAuthority>, ClaimFrontierError> {
5687 let MarkerDeliverySources {
5688 retained: retained_records,
5689 historical: historical_causal_authorities,
5690 candidates: immutable_candidates,
5691 } = *sources;
5692 if usize_to_u128(facts.len()) > u128::from(retained_record_limit) {
5693 return Err(sequence_error(
5694 ledger.required_reserve(),
5695 ClaimFrontierInvalidReason::LogicalOwner,
5696 ));
5697 }
5698 facts.sort_by_key(|fact| fact.marker_delivery_seq);
5699 let mut previous_sequence = None;
5700 let mut authorities = Vec::new();
5701 for fact in facts {
5702 let matching_record = retained_records.iter().find(|record| {
5703 record.delivery_seq == fact.marker_delivery_seq
5704 && matches!(
5705 record.kind,
5706 RetainedCausalRecordKind::CompactionMarker {
5707 participant_index,
5708 ..
5709 } if participant_index == fact.participant_index
5710 )
5711 });
5712 let current_bound =
5713 active_participant(active, fact.participant_index).is_some_and(|participant| {
5714 participant.binding == FrontierBinding::Bound(fact.delivered_binding_epoch)
5715 });
5716 let terminal_order = historical_causal_authorities
5717 .iter()
5718 .find_map(|authority| match authority.kind {
5719 HistoricalCausalKind::BindingTerminal(owner)
5720 if binding_terminal_matches_delivery(owner, fact) =>
5721 {
5722 Some(authority.admission_order)
5723 }
5724 HistoricalCausalKind::BindingTerminal(_)
5725 | HistoricalCausalKind::MembershipExit(_) => None,
5726 })
5727 .or_else(|| {
5728 retained_records
5729 .iter()
5730 .find_map(|record| match record.kind {
5731 RetainedCausalRecordKind::BindingTerminal(owner)
5732 if binding_terminal_matches_delivery(owner, fact) =>
5733 {
5734 Some(record.admission_order)
5735 }
5736 RetainedCausalRecordKind::BindingTerminal(_)
5737 | RetainedCausalRecordKind::MembershipExit { .. }
5738 | RetainedCausalRecordKind::AttachLifecycle { .. }
5739 | RetainedCausalRecordKind::OrdinaryRecord { .. }
5740 | RetainedCausalRecordKind::CompactionMarker { .. } => None,
5741 })
5742 })
5743 .or_else(|| {
5744 immutable_candidates
5745 .iter()
5746 .find_map(|candidate| match candidate {
5747 ImmutableSequenceCandidate::BindingTerminal {
5748 admission_order,
5749 owner,
5750 ..
5751 } if binding_terminal_matches_delivery(*owner, fact) => {
5752 Some(*admission_order)
5753 }
5754 ImmutableSequenceCandidate::BindingTerminal { .. }
5755 | ImmutableSequenceCandidate::Marker(_) => None,
5756 })
5757 });
5758 let historical_epoch_is_backed = matching_record.is_some_and(|marker_record| {
5759 current_bound
5760 || terminal_order
5761 .is_some_and(|terminal_order| terminal_order > marker_record.admission_order)
5762 });
5763 if fact.conversation_id != conversation_id
5764 || previous_sequence == Some(fact.marker_delivery_seq)
5765 || !historical_epoch_is_backed
5766 {
5767 return Err(sequence_error(
5768 sequence_ordinal(ledger, fact.marker_delivery_seq),
5769 ClaimFrontierInvalidReason::LogicalOwner,
5770 ));
5771 }
5772 previous_sequence = Some(fact.marker_delivery_seq);
5773 authorities.push(HistoricalMarkerDeliveryAuthority {
5774 participant_index: fact.participant_index,
5775 marker_delivery_seq: fact.marker_delivery_seq,
5776 delivered_binding_epoch: fact.delivered_binding_epoch,
5777 });
5778 }
5779 Ok(authorities)
5780}
5781
5782fn binding_terminal_matches_delivery(
5783 owner: BindingTerminalOwner,
5784 fact: HistoricalMarkerDeliveryFactRestore,
5785) -> bool {
5786 (owner.participant_index, owner.binding_epoch)
5787 == (fact.participant_index, fact.delivered_binding_epoch)
5788}
5789
5790struct MarkerDeliverySources<'a> {
5791 retained: &'a [RetainedCausalRecord],
5792 historical: &'a [HistoricalCausalAuthority],
5793 candidates: &'a [ImmutableSequenceCandidate],
5794}
5795
5796struct BindingOriginValidation<'a> {
5797 conversation_id: ConversationId,
5798 active: &'a ActiveIdentityRanks,
5799 origins: &'a [BindingOrigin],
5800 retained_records: &'a [RetainedCausalRecord],
5801 causal_authorities: &'a [HistoricalCausalAuthority],
5802 historical_marker_deliveries: &'a [HistoricalMarkerDeliveryAuthority],
5803 total: bool,
5804 ledger: SequenceLedger,
5805}
5806
5807impl BindingOriginValidation<'_> {
5808 fn validate(&self) -> Result<(), ClaimFrontierError> {
5809 if !self.total {
5810 return if self.origins.is_empty() {
5811 Ok(())
5812 } else {
5813 Err(self.logical_owner_error())
5814 };
5815 }
5816 if self.origins.len() != self.active.participants.len() {
5817 return Err(self.logical_owner_error());
5818 }
5819 for participant in &self.active.participants {
5820 let mut matching = self
5821 .origins
5822 .iter()
5823 .filter(|origin| origin.participant_id() == participant.participant_index);
5824 let Some(origin) = matching.next() else {
5825 return Err(self.logical_owner_error());
5826 };
5827 if matching.next().is_some() {
5828 return Err(self.logical_owner_error());
5829 }
5830 self.validate_origin(*participant, *origin)?;
5831 }
5832 Ok(())
5833 }
5834
5835 fn validate_origin(
5836 &self,
5837 participant: FrontierParticipant,
5838 origin: BindingOrigin,
5839 ) -> Result<(), ClaimFrontierError> {
5840 let current_epoch = binding_epoch(participant.binding);
5841 let attached = origin.attached();
5842 if origin.conversation_id() != self.conversation_id
5843 || origin.binding_epoch() != current_epoch
5844 || attached.conversation_id() != self.conversation_id
5845 || attached.participant_id() != participant.participant_index
5846 || attached.binding_epoch() != current_epoch
5847 || attached.admission_order().candidate_phase() != CandidatePhase::AttachLifecycle
5848 {
5849 return Err(self.logical_owner_error());
5850 }
5851 let mut retained_attach_for_binding = self.retained_records.iter().filter(|record| {
5852 matches!(
5853 record.kind,
5854 RetainedCausalRecordKind::AttachLifecycle {
5855 participant_index,
5856 binding_epoch,
5857 } if participant_index == participant.participant_index
5858 && binding_epoch == current_epoch
5859 )
5860 });
5861 let retained_attach_matches = retained_attach_for_binding.clone().any(|record| {
5862 record.delivery_seq == attached.delivery_seq()
5863 && record.admission_order == attached.admission_order()
5864 });
5865 if retained_attach_for_binding.next().is_some() && !retained_attach_matches {
5866 return Err(self.logical_owner_error());
5867 }
5868 if let Some((marker_delivery_seq, prior_binding_epoch)) = origin.recovered_marker() {
5869 let generation_is_next = prior_binding_epoch
5870 .capability_generation
5871 .get()
5872 .checked_add(1)
5873 == Some(current_epoch.capability_generation.get());
5874 let marker_history_matches = self.historical_marker_deliveries.iter().any(|history| {
5875 history.participant_index == participant.participant_index
5876 && history.marker_delivery_seq == marker_delivery_seq
5877 && history.delivered_binding_epoch == prior_binding_epoch
5878 });
5879 if !generation_is_next || !marker_history_matches {
5880 return Err(sequence_error(
5881 sequence_ordinal(self.ledger, marker_delivery_seq),
5882 ClaimFrontierInvalidReason::RecoveryBlock,
5883 ));
5884 }
5885 } else if matches!(participant.binding, FrontierBinding::Detached(_))
5886 && !binding_terminal_exists(
5887 participant.participant_index,
5888 current_epoch,
5889 self.retained_records,
5890 self.causal_authorities,
5891 )
5892 {
5893 return Err(self.logical_owner_error());
5894 }
5895 Ok(())
5896 }
5897
5898 const fn logical_owner_error(&self) -> ClaimFrontierError {
5899 sequence_error(
5900 self.ledger.required_reserve(),
5901 ClaimFrontierInvalidReason::LogicalOwner,
5902 )
5903 }
5904}
5905
5906fn binding_terminal_exists(
5907 participant_index: ParticipantId,
5908 binding_epoch: BindingEpoch,
5909 retained_records: &[RetainedCausalRecord],
5910 historical_causal_authorities: &[HistoricalCausalAuthority],
5911) -> bool {
5912 retained_records.iter().any(|record| {
5913 matches!(
5914 record.kind,
5915 RetainedCausalRecordKind::BindingTerminal(owner)
5916 if owner.participant_index == participant_index
5917 && owner.binding_epoch == binding_epoch
5918 )
5919 }) || historical_causal_authorities.iter().any(|authority| {
5920 matches!(
5921 authority.kind,
5922 HistoricalCausalKind::BindingTerminal(owner)
5923 if owner.participant_index == participant_index
5924 && owner.binding_epoch == binding_epoch
5925 )
5926 })
5927}
5928
5929fn validate_marker_credit_owners(
5930 candidates: &[ImmutableSequenceCandidate],
5931 marker_records: &[RetainedCausalRecord],
5932 identity_slot_limit: u64,
5933 ledger: SequenceLedger,
5934) -> Result<(), ClaimFrontierError> {
5935 let mut owners = Vec::new();
5936 for record in marker_records {
5937 let RetainedCausalRecordKind::CompactionMarker {
5938 participant_index, ..
5939 } = record.kind
5940 else {
5941 continue;
5942 };
5943 if owners.contains(&participant_index) {
5944 return Err(sequence_error(
5945 ledger.required_reserve(),
5946 ClaimFrontierInvalidReason::LogicalOwner,
5947 ));
5948 }
5949 owners.push(participant_index);
5950 }
5951 for candidate in candidates {
5952 let ImmutableSequenceCandidate::Marker(marker) = candidate else {
5953 continue;
5954 };
5955 let participant_index = marker.admission_order.participant_index();
5956 if owners.contains(&participant_index) {
5957 return Err(sequence_error(
5958 sequence_ordinal(ledger, marker.delivery_seq),
5959 ClaimFrontierInvalidReason::LogicalOwner,
5960 ));
5961 }
5962 owners.push(participant_index);
5963 }
5964 if usize_to_u128(owners.len()) > u128::from(identity_slot_limit) {
5965 return Err(sequence_error(
5966 ledger.required_reserve(),
5967 ClaimFrontierInvalidReason::LogicalOwner,
5968 ));
5969 }
5970 Ok(())
5971}
5972
5973fn validated_historical_authorities(
5974 facts: Vec<HistoricalCausalFactRestore>,
5975 conversation_id: ConversationId,
5976 identity_slot_limit: u64,
5977 ledger: SequenceLedger,
5978 history: &ValidatedConversationHistory,
5979) -> Result<Vec<HistoricalCausalAuthority>, ClaimFrontierError> {
5980 if usize_to_u128(facts.len()) > u128::from(identity_slot_limit).saturating_mul(2) {
5981 return Err(sequence_error(
5982 ledger.required_reserve(),
5983 ClaimFrontierInvalidReason::LogicalOwner,
5984 ));
5985 }
5986 let authorities: Vec<_> = facts
5987 .into_iter()
5988 .map(HistoricalCausalAuthority::from_restore)
5989 .collect();
5990 let mut seen = Vec::new();
5991 for authority in &authorities {
5992 let (participant_index, phase) = match authority.kind {
5993 HistoricalCausalKind::BindingTerminal(owner) => {
5994 (owner.participant_index, CandidatePhase::BindingTerminal)
5995 }
5996 HistoricalCausalKind::MembershipExit(participant_index) => {
5997 (participant_index, CandidatePhase::MembershipExit)
5998 }
5999 };
6000 if authority.conversation_id != conversation_id
6001 || participant_index >= identity_slot_limit
6002 || authority.admission_order.participant_index() != participant_index
6003 || authority.admission_order.candidate_phase() != phase
6004 || seen.contains(authority)
6005 || !history.causal_authorities.contains(authority)
6006 {
6007 return Err(sequence_error(
6008 ledger.required_reserve(),
6009 ClaimFrontierInvalidReason::LogicalOwner,
6010 ));
6011 }
6012 seen.push(*authority);
6013 }
6014 Ok(authorities)
6015}
6016
6017const fn corrupt_frontier(error: ClaimFrontierError) -> ParticipantStateCorruptReason {
6018 ParticipantStateCorruptReason::ClaimFrontierInvalid {
6019 counter: match error.counter {
6020 ClaimFrontierCounter::DeliverySequence => ClaimCounter::DeliverySeq,
6021 ClaimFrontierCounter::TransactionOrder => ClaimCounter::TransactionOrder,
6022 },
6023 first_bad_position: error.first_bad_position,
6024 }
6025}
6026
6027fn sequence_segments(restore: &SequenceClaimFrontierRestore) -> Vec<NumericSegment<SequenceClass>> {
6028 let mut segments = Vec::new();
6029 for claim in &restore.movable_claims {
6030 segments.push(NumericSegment {
6031 start: u128::from(claim.delivery_seq),
6032 length: 1,
6033 class: Some(match claim.owner {
6034 SequenceDirectOwner::MembershipExit { .. } => SequenceClass::Exit,
6035 SequenceDirectOwner::BindingTerminal(_) => SequenceClass::Terminal,
6036 }),
6037 immutable: false,
6038 });
6039 }
6040 for candidate in &restore.immutable_candidates {
6041 segments.push(NumericSegment {
6042 start: u128::from(candidate.delivery_seq()),
6043 length: 1,
6044 class: Some(sequence_candidate_class(*candidate)),
6045 immutable: true,
6046 });
6047 }
6048 for range in &restore.products.live_times_terminal {
6049 segments.push(NumericSegment {
6050 start: u128::from(range.start),
6051 length: u128::from(range.length),
6052 class: Some(SequenceClass::LiveTimesTerminal),
6053 immutable: false,
6054 });
6055 }
6056 if let Some(range) = restore.products.live_times_replacement_terminal {
6057 segments.push(NumericSegment {
6058 start: u128::from(range.start),
6059 length: u128::from(range.length),
6060 class: Some(SequenceClass::LiveTimesReplacementTerminal),
6061 immutable: false,
6062 });
6063 }
6064 for range in &restore.products.other_live_times_exit {
6065 segments.push(NumericSegment {
6066 start: u128::from(range.start),
6067 length: u128::from(range.length),
6068 class: Some(SequenceClass::OtherLiveTimesExit),
6069 immutable: false,
6070 });
6071 }
6072 if let Some(recovery) = restore.recovery {
6073 if let Some(terminal) = recovery.terminal {
6074 segments.push(NumericSegment {
6075 start: u128::from(terminal.delivery_seq),
6076 length: 1,
6077 class: Some(SequenceClass::Terminal),
6078 immutable: false,
6079 });
6080 }
6081 segments.push(NumericSegment {
6082 start: u128::from(recovery.recovery_attach_seq),
6083 length: 1,
6084 class: Some(SequenceClass::RecoveryAttach),
6085 immutable: false,
6086 });
6087 segments.push(NumericSegment {
6088 start: u128::from(recovery.replacement_terminal_seq),
6089 length: 1,
6090 class: Some(SequenceClass::RecoveryReplacementTerminal),
6091 immutable: false,
6092 });
6093 }
6094 segments
6095}
6096
6097fn order_segments(
6098 restore: &OrderClaimFrontierRestore,
6099 high: OrderHigh,
6100) -> Vec<NumericSegment<OrderClass>> {
6101 let mut segments = Vec::new();
6102 for claim in &restore.movable_claims {
6103 segments.push(NumericSegment {
6104 start: u128::from(claim.transaction_order),
6105 length: 1,
6106 class: Some(match claim.owner {
6107 OrderDirectOwner::ActiveBindingTerminal(_) => OrderClass::ActiveBindingTerminal,
6108 OrderDirectOwner::MembershipExit { .. } => OrderClass::MembershipExit,
6109 }),
6110 immutable: false,
6111 });
6112 }
6113 for candidate in restore
6114 .immutable_candidates
6115 .iter()
6116 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
6117 {
6118 segments.push(NumericSegment {
6119 start: u128::from(candidate.transaction_order),
6120 length: 1,
6121 class: None,
6122 immutable: true,
6123 });
6124 }
6125 if let Some(recovery) = restore.recovery {
6126 if let Some(active_binding) = recovery.active_binding {
6127 segments.push(NumericSegment {
6128 start: u128::from(active_binding.transaction_order),
6129 length: 1,
6130 class: Some(OrderClass::ActiveBindingTerminal),
6131 immutable: false,
6132 });
6133 }
6134 segments.push(NumericSegment {
6135 start: u128::from(recovery.recovery_operation_order),
6136 length: 1,
6137 class: Some(OrderClass::RecoveryOperation),
6138 immutable: false,
6139 });
6140 segments.push(NumericSegment {
6141 start: u128::from(recovery.replacement_terminal_order),
6142 length: 1,
6143 class: Some(OrderClass::RecoveryReplacementTerminal),
6144 immutable: false,
6145 });
6146 }
6147 segments
6148}
6149
6150fn restore_sequence_frontier(
6151 active: &ActiveIdentityRanks,
6152 restore: SequenceClaimFrontierRestore,
6153 retained_floor: u128,
6154 ledger: SequenceLedger,
6155 recovery_provenance: Option<RecoveryClaimProvenance>,
6156 retained_records: &[RetainedCausalRecord],
6157 historical_causal_authorities: &[HistoricalCausalAuthority],
6158) -> Result<SequenceClaimFrontier, ClaimFrontierError> {
6159 let mut segments = sequence_segments(&restore);
6160
6161 let expected_counts = sequence_expected_counts(ledger);
6162 validate_numeric_segments(
6163 ClaimFrontierCounter::DeliverySequence,
6164 u128::from(ledger.high_watermark()) + 1,
6165 ledger.required_reserve(),
6166 &mut segments,
6167 &expected_counts,
6168 )?;
6169 validate_sequence_recovery(
6170 active,
6171 restore.recovery,
6172 recovery_provenance,
6173 ledger,
6174 ledger.required_reserve(),
6175 )?;
6176 validate_sequence_candidates(
6177 active,
6178 &restore.immutable_candidates,
6179 retained_floor,
6180 retained_records,
6181 historical_causal_authorities,
6182 ledger,
6183 )?;
6184 let terminal_owners = validate_sequence_direct_owners(
6185 active,
6186 &restore.movable_claims,
6187 &restore.immutable_candidates,
6188 restore.recovery,
6189 ledger,
6190 )?;
6191 let products = validate_sequence_products(
6192 active,
6193 restore.products,
6194 &terminal_owners,
6195 restore.recovery,
6196 recovery_provenance,
6197 ledger,
6198 )?;
6199 let recovery = restore
6200 .recovery
6201 .zip(recovery_provenance)
6202 .map(|(value, provenance)| RecoverySequenceBlock {
6203 terminal: value.terminal,
6204 recovery_attach_seq: value.recovery_attach_seq,
6205 replacement_terminal_seq: value.replacement_terminal_seq,
6206 participant_index: provenance.participant_index,
6207 marker_delivery_seq: provenance.marker_delivery_seq,
6208 recovered_binding_epoch: provenance.prior_binding_epoch,
6209 });
6210
6211 let mut movable_claims = restore.movable_claims;
6212 movable_claims.sort_by_key(|claim| claim.delivery_seq);
6213 let mut immutable_candidates = restore.immutable_candidates;
6214 immutable_candidates.sort_by_key(|candidate| candidate.delivery_seq());
6215
6216 Ok(SequenceClaimFrontier {
6217 ledger,
6218 movable_claims,
6219 immutable_candidates,
6220 products,
6221 recovery,
6222 })
6223}
6224
6225fn restore_order_frontier(
6226 active: &ActiveIdentityRanks,
6227 restore: OrderClaimFrontierRestore,
6228 ledger: OrderLedger,
6229 recovery_provenance: Option<RecoveryClaimProvenance>,
6230) -> Result<OrderClaimFrontier, ClaimFrontierError> {
6231 let mut segments = order_segments(&restore, ledger.high());
6232
6233 let candidate_count = order_frontier_candidate_count(&restore, ledger.high());
6234 let expected_length = candidate_count + ledger.claims().total();
6235 let expected_counts = order_expected_counts(ledger);
6236 validate_numeric_segments(
6237 ClaimFrontierCounter::TransactionOrder,
6238 order_frontier_start(ledger.high()),
6239 expected_length,
6240 &mut segments,
6241 &expected_counts,
6242 )?;
6243 validate_order_recovery(
6244 active,
6245 restore.recovery,
6246 recovery_provenance,
6247 ledger,
6248 expected_length,
6249 )?;
6250 let immutable_candidates = validate_order_candidates(&restore.immutable_candidates, ledger)?;
6251 validate_order_direct_owners(active, &restore.movable_claims, restore.recovery, ledger)?;
6252 let recovery = restore
6253 .recovery
6254 .zip(recovery_provenance)
6255 .map(|(value, provenance)| RecoveryOrderBlock {
6256 active_binding: value.active_binding,
6257 recovery_operation_order: value.recovery_operation_order,
6258 replacement_terminal_order: value.replacement_terminal_order,
6259 participant_index: provenance.participant_index,
6260 marker_delivery_seq: provenance.marker_delivery_seq,
6261 recovered_binding_epoch: provenance.prior_binding_epoch,
6262 });
6263
6264 let mut movable_claims = restore.movable_claims;
6265 movable_claims.sort_by_key(|claim| claim.transaction_order);
6266
6267 Ok(OrderClaimFrontier {
6268 ledger,
6269 movable_claims,
6270 immutable_candidates,
6271 recovery,
6272 })
6273}
6274
6275fn describe_sequence_frontier(frontier: &SequenceClaimFrontier) -> SequenceClaimFrontierRestore {
6282 SequenceClaimFrontierRestore {
6283 movable_claims: frontier.movable_claims.clone(),
6284 immutable_candidates: frontier.immutable_candidates.clone(),
6285 products: SequenceProductRangesRestore {
6286 live_times_terminal: frontier
6287 .products
6288 .live_times_terminal
6289 .iter()
6290 .map(|range| TerminalProductRangeRestore {
6291 start: range.start,
6292 length: range.length,
6293 terminal: range.terminal,
6294 })
6295 .collect(),
6296 live_times_replacement_terminal: frontier.products.live_times_replacement_terminal.map(
6297 |range| ReplacementTerminalProductRangeRestore {
6298 start: range.start,
6299 length: range.length,
6300 },
6301 ),
6302 other_live_times_exit: frontier
6303 .products
6304 .other_live_times_exit
6305 .iter()
6306 .map(|range| ExitProductRangeRestore {
6307 start: range.start,
6308 length: range.length,
6309 exit_participant: range.exit_participant,
6310 })
6311 .collect(),
6312 },
6313 recovery: frontier.recovery.map(|block| RecoverySequenceBlockRestore {
6314 terminal: block.terminal,
6315 recovery_attach_seq: block.recovery_attach_seq,
6316 replacement_terminal_seq: block.replacement_terminal_seq,
6317 }),
6318 }
6319}
6320
6321fn describe_order_frontier(frontier: &OrderClaimFrontier) -> OrderClaimFrontierRestore {
6328 OrderClaimFrontierRestore {
6329 movable_claims: frontier.movable_claims.clone(),
6330 immutable_candidates: frontier
6331 .immutable_candidates
6332 .iter()
6333 .map(|major| ImmutableOrderCandidateMajorRestore {
6334 transaction_order: major.transaction_order,
6335 candidate_keys: major.candidate_keys.clone(),
6336 })
6337 .collect(),
6338 recovery: frontier.recovery.map(|block| RecoveryOrderBlockRestore {
6339 active_binding: block.active_binding,
6340 recovery_operation_order: block.recovery_operation_order,
6341 replacement_terminal_order: block.replacement_terminal_order,
6342 }),
6343 }
6344}
6345
6346fn validate_numeric_segments<C: Copy + Into<usize>>(
6347 counter: ClaimFrontierCounter,
6348 first_value: u128,
6349 expected_length: u128,
6350 segments: &mut [NumericSegment<C>],
6351 expected_counts: &[u128],
6352) -> Result<(), ClaimFrontierError> {
6353 segments.sort_by_key(|segment| segment.start);
6354 let mut events = numeric_events(counter, first_value, segments)?;
6355 let emitted = scan_numeric_events(counter, first_value, &mut events)?;
6356 if emitted != expected_length {
6357 return Err(frontier_error(
6358 counter,
6359 emitted.min(expected_length),
6360 ClaimFrontierInvalidReason::AggregateLedger,
6361 ));
6362 }
6363 validate_immutable_prefix(counter, first_value, segments)?;
6364 validate_segment_class_counts(counter, expected_length, segments, expected_counts)
6365}
6366
6367fn numeric_events<C>(
6368 counter: ClaimFrontierCounter,
6369 first_value: u128,
6370 segments: &[NumericSegment<C>],
6371) -> Result<Vec<(u128, i8)>, ClaimFrontierError> {
6372 let mut events = Vec::new();
6373 let counter_limit = u128::from(u64::MAX) + 1;
6374 for segment in segments {
6375 if segment.length == 0 {
6376 continue;
6377 }
6378 let Some(end) = segment.start.checked_add(segment.length) else {
6379 return Err(frontier_error(
6380 counter,
6381 counter_limit.saturating_sub(first_value),
6382 ClaimFrontierInvalidReason::NumericPosition,
6383 ));
6384 };
6385 if segment.start < first_value {
6386 return Err(frontier_error(
6387 counter,
6388 0,
6389 ClaimFrontierInvalidReason::NumericPosition,
6390 ));
6391 }
6392 if end > counter_limit {
6393 return Err(frontier_error(
6394 counter,
6395 counter_limit.saturating_sub(first_value),
6396 ClaimFrontierInvalidReason::NumericPosition,
6397 ));
6398 }
6399 events.push((segment.start, 1_i8));
6400 events.push((end, -1_i8));
6401 }
6402 Ok(events)
6403}
6404
6405fn scan_numeric_events(
6406 counter: ClaimFrontierCounter,
6407 first_value: u128,
6408 events: &mut [(u128, i8)],
6409) -> Result<u128, ClaimFrontierError> {
6410 events.sort_unstable_by_key(|event| event.0);
6411 let mut event_index = 0_usize;
6412 let mut coordinate = first_value;
6413 let mut coverage = 0_i128;
6414 let mut emitted = 0_u128;
6415 while let Some((event_coordinate, _)) = events.get(event_index).copied() {
6416 if event_coordinate > coordinate {
6417 if coverage == 0 {
6418 return Err(frontier_error(
6419 counter,
6420 emitted,
6421 ClaimFrontierInvalidReason::NumericPosition,
6422 ));
6423 }
6424 if coverage > 1 {
6425 return Err(frontier_error(
6426 counter,
6427 emitted.saturating_add(1),
6428 ClaimFrontierInvalidReason::NumericPosition,
6429 ));
6430 }
6431 emitted = emitted.saturating_add(event_coordinate - coordinate);
6432 coordinate = event_coordinate;
6433 }
6434 while let Some((same_coordinate, delta)) = events.get(event_index).copied() {
6435 if same_coordinate != coordinate {
6436 break;
6437 }
6438 coverage += i128::from(delta);
6439 event_index += 1;
6440 }
6441 }
6442 if coverage != 0 {
6443 return Err(frontier_error(
6444 counter,
6445 emitted,
6446 ClaimFrontierInvalidReason::NumericPosition,
6447 ));
6448 }
6449 Ok(emitted)
6450}
6451
6452fn validate_immutable_prefix<C>(
6453 counter: ClaimFrontierCounter,
6454 first_value: u128,
6455 segments: &[NumericSegment<C>],
6456) -> Result<(), ClaimFrontierError> {
6457 let mut first_movable = None;
6458 for segment in segments.iter().filter(|segment| segment.length != 0) {
6459 if segment.immutable {
6460 if let Some(first_movable) = first_movable {
6461 return Err(frontier_error(
6462 counter,
6463 first_movable,
6464 ClaimFrontierInvalidReason::NumericPosition,
6465 ));
6466 }
6467 } else if first_movable.is_none() {
6468 first_movable = Some(segment.start - first_value);
6469 }
6470 }
6471 Ok(())
6472}
6473
6474fn validate_segment_class_counts<C: Copy + Into<usize>>(
6475 counter: ClaimFrontierCounter,
6476 expected_length: u128,
6477 segments: &[NumericSegment<C>],
6478 expected_counts: &[u128],
6479) -> Result<(), ClaimFrontierError> {
6480 let mut actual_counts = core::iter::repeat_n(0_u128, expected_counts.len()).collect::<Vec<_>>();
6481 let mut class_ordinal = 0_u128;
6482 for segment in segments {
6483 if segment.length == 0 {
6484 continue;
6485 }
6486 if let Some(class) = segment.class {
6487 let index = class.into();
6488 let prior = actual_counts[index];
6489 let Some(resulting) = prior.checked_add(segment.length) else {
6490 return Err(frontier_error(
6491 counter,
6492 class_ordinal,
6493 ClaimFrontierInvalidReason::AggregateLedger,
6494 ));
6495 };
6496 if resulting > expected_counts[index] {
6497 return Err(frontier_error(
6498 counter,
6499 class_ordinal + expected_counts[index].saturating_sub(prior),
6500 ClaimFrontierInvalidReason::AggregateLedger,
6501 ));
6502 }
6503 actual_counts[index] = resulting;
6504 }
6505 class_ordinal += segment.length;
6506 }
6507 if actual_counts != expected_counts {
6508 return Err(frontier_error(
6509 counter,
6510 expected_length,
6511 ClaimFrontierInvalidReason::AggregateLedger,
6512 ));
6513 }
6514 Ok(())
6515}
6516
6517#[cfg(test)]
6518pub(super) fn validate_numeric_union_for_test(
6519 first_value: u128,
6520 expected_length: u128,
6521 ranges: &[(u128, u128)],
6522) -> Result<(), ClaimFrontierError> {
6523 let mut segments: Vec<_> = ranges
6524 .iter()
6525 .map(|(start, length)| NumericSegment {
6526 start: *start,
6527 length: *length,
6528 class: Some(SequenceClass::Exit),
6529 immutable: false,
6530 })
6531 .collect();
6532 validate_numeric_segments(
6533 ClaimFrontierCounter::DeliverySequence,
6534 first_value,
6535 expected_length,
6536 &mut segments,
6537 &[expected_length, 0, 0, 0, 0, 0, 0, 0],
6538 )
6539}
6540
6541impl From<SequenceClass> for usize {
6542 fn from(value: SequenceClass) -> Self {
6543 value as Self
6544 }
6545}
6546
6547impl From<OrderClass> for usize {
6548 fn from(value: OrderClass) -> Self {
6549 value as Self
6550 }
6551}
6552
6553const fn sequence_candidate_class(candidate: ImmutableSequenceCandidate) -> SequenceClass {
6554 match candidate {
6555 ImmutableSequenceCandidate::BindingTerminal { .. } => SequenceClass::Terminal,
6556 ImmutableSequenceCandidate::Marker(marker) => match marker.current_owner {
6557 MarkerSequenceOwner::Marker => SequenceClass::Marker,
6558 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::LiveTimesTerminal) => {
6559 SequenceClass::LiveTimesTerminal
6560 }
6561 MarkerSequenceOwner::ConditionalProduct(
6562 SequenceProductClass::LiveTimesReplacementTerminal,
6563 ) => SequenceClass::LiveTimesReplacementTerminal,
6564 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::OtherLiveTimesExit) => {
6565 SequenceClass::OtherLiveTimesExit
6566 }
6567 },
6568 }
6569}
6570
6571fn sequence_expected_counts(ledger: SequenceLedger) -> [u128; 8] {
6572 let budget = ledger.budget();
6573 [
6574 u128::from(budget.e),
6575 u128::from(budget.t),
6576 u128::from(budget.m),
6577 u128::from(budget.rs),
6578 u128::from(budget.rt),
6579 budget.l_times_t,
6580 budget.l_times_rt,
6581 budget.l_other_times_e,
6582 ]
6583}
6584
6585fn order_expected_counts(ledger: OrderLedger) -> [u128; 4] {
6586 let claims = ledger.claims();
6587 [
6588 u128::from(claims.active_binding_terminals()),
6589 u128::from(claims.membership_exits()),
6590 u128::from(claims.recovery_operation()),
6591 u128::from(claims.recovery_replacement_terminal()),
6592 ]
6593}
6594
6595fn validate_sequence_recovery(
6596 active: &ActiveIdentityRanks,
6597 recovery: Option<RecoverySequenceBlockRestore>,
6598 provenance: Option<RecoveryClaimProvenance>,
6599 ledger: SequenceLedger,
6600 frontier_length: u128,
6601) -> Result<(), ClaimFrontierError> {
6602 let expected = ledger.claims().recovery();
6603 match (expected, recovery, provenance) {
6604 (RecoverySequenceReserve::None, None, None) => Ok(()),
6605 (RecoverySequenceReserve::DetachedCredentialRecovery, None, _)
6606 | (RecoverySequenceReserve::DetachedCredentialRecovery, Some(_), None)
6607 | (RecoverySequenceReserve::None, None, Some(_)) => Err(sequence_error(
6608 frontier_length,
6609 ClaimFrontierInvalidReason::RecoveryBlock,
6610 )),
6611 (RecoverySequenceReserve::None, Some(block), _) => Err(sequence_error(
6612 sequence_ordinal(ledger, block_start_sequence(block)),
6613 ClaimFrontierInvalidReason::RecoveryBlock,
6614 )),
6615 (RecoverySequenceReserve::DetachedCredentialRecovery, Some(block), Some(provenance)) => {
6616 let block_ordinal = sequence_ordinal(ledger, block_start_sequence(block));
6617 let expected_recovery_attach = block
6618 .terminal
6619 .map_or(Some(block.recovery_attach_seq), |terminal| {
6620 terminal.delivery_seq.checked_add(1)
6621 });
6622 if expected_recovery_attach != Some(block.recovery_attach_seq) {
6623 return Err(sequence_error(
6624 block_ordinal + 1,
6625 ClaimFrontierInvalidReason::RecoveryBlock,
6626 ));
6627 }
6628 if block.recovery_attach_seq.checked_add(1) != Some(block.replacement_terminal_seq) {
6629 return Err(sequence_error(
6630 block_ordinal + u128::from(block.terminal.is_some()) + 1,
6631 ClaimFrontierInvalidReason::RecoveryBlock,
6632 ));
6633 }
6634 let Some(participant) = active_participant(active, provenance.participant_index) else {
6635 return Err(sequence_error(
6636 block_ordinal,
6637 ClaimFrontierInvalidReason::LogicalOwner,
6638 ));
6639 };
6640 let expected_binding = match provenance.phase {
6641 RecoveryClaimPhase::PreFate => {
6642 FrontierBinding::Bound(provenance.prior_binding_epoch)
6643 }
6644 RecoveryClaimPhase::PostFate => {
6645 FrontierBinding::Detached(provenance.prior_binding_epoch)
6646 }
6647 RecoveryClaimPhase::RecoveredBound => {
6648 FrontierBinding::Bound(provenance.current_binding_epoch)
6649 }
6650 };
6651 if participant.binding != expected_binding {
6652 return Err(sequence_error(
6653 block_ordinal,
6654 ClaimFrontierInvalidReason::LogicalOwner,
6655 ));
6656 }
6657 let terminal_valid = match (provenance.phase, block.terminal) {
6658 (RecoveryClaimPhase::PreFate, Some(terminal)) => {
6659 terminal.owner.participant_index == provenance.participant_index
6660 && terminal.owner.binding_epoch == provenance.prior_binding_epoch
6661 }
6662 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
6663 _ => false,
6664 };
6665 if !terminal_valid {
6666 return Err(sequence_error(
6667 block_ordinal,
6668 ClaimFrontierInvalidReason::RecoveryBlock,
6669 ));
6670 }
6671 Ok(())
6672 }
6673 }
6674}
6675
6676fn validate_sequence_candidates(
6677 active: &ActiveIdentityRanks,
6678 candidates: &[ImmutableSequenceCandidate],
6679 retained_floor: u128,
6680 retained_records: &[RetainedCausalRecord],
6681 historical_causal_authorities: &[HistoricalCausalAuthority],
6682 ledger: SequenceLedger,
6683) -> Result<(), ClaimFrontierError> {
6684 let mut seen_keys = Vec::new();
6685 let mut previous_sequence = None;
6686 let mut previous_order = retained_records.last().map(|record| record.admission_order);
6687 for candidate in candidates {
6688 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
6689 let order = candidate.admission_order();
6690 if previous_sequence.is_some_and(|previous| previous >= candidate.delivery_seq())
6691 || previous_order.is_some_and(|previous| previous >= order)
6692 || seen_keys.contains(&order)
6693 {
6694 return Err(sequence_error(
6695 ordinal,
6696 ClaimFrontierInvalidReason::CandidateKey,
6697 ));
6698 }
6699 previous_sequence = Some(candidate.delivery_seq());
6700 previous_order = Some(order);
6701 seen_keys.push(order);
6702 match candidate {
6703 ImmutableSequenceCandidate::BindingTerminal { owner, .. } => {
6704 if order.candidate_phase() != CandidatePhase::BindingTerminal
6705 || order.participant_index() != owner.participant_index
6706 || !terminal_matches_active(active, *owner)
6707 {
6708 return Err(sequence_error(
6709 ordinal,
6710 ClaimFrontierInvalidReason::CandidateKey,
6711 ));
6712 }
6713 }
6714 ImmutableSequenceCandidate::Marker(marker) => {
6715 let Some(participant) = active_participant(active, order.participant_index())
6716 else {
6717 return Err(sequence_error(
6718 ordinal,
6719 ClaimFrontierInvalidReason::LogicalOwner,
6720 ));
6721 };
6722 if order.candidate_phase() != CandidatePhase::CompactionMarker
6723 || marker.current_owner != MarkerSequenceOwner::Marker
6724 || marker.target_binding != participant.binding
6725 || marker.abandoned_after != participant.cursor
6726 || marker.abandoned_after > marker.abandoned_through
6727 || u128::from(marker.physical_floor_at_decision) != retained_floor
6728 || u128::from(marker.physical_floor_at_decision)
6729 > u128::from(marker.abandoned_through) + 1
6730 || marker.abandoned_through >= marker.delivery_seq
6731 || !marker_provenance_targets(marker.provenance, order.participant_index())
6732 || !marker_has_causal_authority(
6733 *marker,
6734 retained_records,
6735 historical_causal_authorities,
6736 )
6737 {
6738 return Err(sequence_error(
6739 ordinal,
6740 ClaimFrontierInvalidReason::CandidateKey,
6741 ));
6742 }
6743 }
6744 }
6745 }
6746 Ok(())
6747}
6748
6749#[derive(Clone, Copy, Debug, PartialEq, Eq)]
6750enum TerminalOccurrenceKind {
6751 Movable,
6752 Candidate,
6753}
6754
6755#[derive(Clone, Copy, Debug, PartialEq, Eq)]
6756struct TerminalOccurrence {
6757 owner: BindingTerminalOwner,
6758 ordinal: u128,
6759 kind: TerminalOccurrenceKind,
6760}
6761
6762pub(in crate::lifecycle) const fn terminal_owner_sort_key(
6767 owner: &BindingTerminalOwner,
6768) -> (ParticipantId, (ConnectionIncarnation, Generation)) {
6769 (
6770 owner.participant_index,
6771 owner.binding_epoch.canonical_sort_key(),
6772 )
6773}
6774
6775fn validate_sequence_direct_owners(
6776 active: &ActiveIdentityRanks,
6777 movable: &[MovableSequenceClaim],
6778 candidates: &[ImmutableSequenceCandidate],
6779 recovery: Option<RecoverySequenceBlockRestore>,
6780 ledger: SequenceLedger,
6781) -> Result<Vec<BindingTerminalOwner>, ClaimFrontierError> {
6782 let (mut exit_owners, terminal_occurrences) =
6783 collect_sequence_direct_owners(active, movable, candidates, recovery, ledger)?;
6784 validate_sequence_exit_owners(active, &mut exit_owners, ledger)?;
6785 validate_sequence_terminal_owners(active, &terminal_occurrences, ledger)?;
6786 let mut owners: Vec<_> = terminal_occurrences
6787 .into_iter()
6788 .map(|occurrence| occurrence.owner)
6789 .collect();
6790 owners.sort_by_key(terminal_owner_sort_key);
6791 Ok(owners)
6792}
6793
6794fn collect_sequence_direct_owners(
6795 active: &ActiveIdentityRanks,
6796 movable: &[MovableSequenceClaim],
6797 candidates: &[ImmutableSequenceCandidate],
6798 recovery: Option<RecoverySequenceBlockRestore>,
6799 ledger: SequenceLedger,
6800) -> Result<(Vec<ParticipantId>, Vec<TerminalOccurrence>), ClaimFrontierError> {
6801 let mut ordered_movable = movable.to_vec();
6802 ordered_movable.sort_by_key(|claim| claim.delivery_seq);
6803 let mut exit_owners = Vec::new();
6804 let mut terminal_occurrences = Vec::new();
6805 for claim in ordered_movable {
6806 let ordinal = sequence_ordinal(ledger, claim.delivery_seq);
6807 match claim.owner {
6808 SequenceDirectOwner::MembershipExit { participant_index } => {
6809 if !active.contains(participant_index) || exit_owners.contains(&participant_index) {
6810 return Err(sequence_error(
6811 ordinal,
6812 ClaimFrontierInvalidReason::LogicalOwner,
6813 ));
6814 }
6815 exit_owners.push(participant_index);
6816 }
6817 SequenceDirectOwner::BindingTerminal(owner) => {
6818 if !terminal_matches_bound(active, owner) {
6819 return Err(sequence_error(
6820 ordinal,
6821 ClaimFrontierInvalidReason::LogicalOwner,
6822 ));
6823 }
6824 push_terminal_occurrence(
6825 &mut terminal_occurrences,
6826 owner,
6827 ordinal,
6828 TerminalOccurrenceKind::Movable,
6829 )?;
6830 }
6831 }
6832 }
6833 for candidate in candidates {
6834 if let ImmutableSequenceCandidate::BindingTerminal { owner, .. } = candidate {
6835 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
6836 push_terminal_occurrence(
6837 &mut terminal_occurrences,
6838 *owner,
6839 ordinal,
6840 TerminalOccurrenceKind::Candidate,
6841 )?;
6842 }
6843 }
6844 if let Some(terminal) = recovery.and_then(|block| block.terminal) {
6845 let ordinal = sequence_ordinal(ledger, terminal.delivery_seq);
6846 push_terminal_occurrence(
6847 &mut terminal_occurrences,
6848 terminal.owner,
6849 ordinal,
6850 TerminalOccurrenceKind::Movable,
6851 )?;
6852 }
6853 Ok((exit_owners, terminal_occurrences))
6854}
6855
6856fn push_terminal_occurrence(
6857 occurrences: &mut Vec<TerminalOccurrence>,
6858 owner: BindingTerminalOwner,
6859 ordinal: u128,
6860 kind: TerminalOccurrenceKind,
6861) -> Result<(), ClaimFrontierError> {
6862 if occurrences
6863 .iter()
6864 .any(|occurrence| occurrence.owner == owner)
6865 {
6866 return Err(sequence_error(
6867 ordinal,
6868 ClaimFrontierInvalidReason::LogicalOwner,
6869 ));
6870 }
6871 occurrences.push(TerminalOccurrence {
6872 owner,
6873 ordinal,
6874 kind,
6875 });
6876 Ok(())
6877}
6878
6879fn validate_sequence_exit_owners(
6880 active: &ActiveIdentityRanks,
6881 exit_owners: &mut [ParticipantId],
6882 ledger: SequenceLedger,
6883) -> Result<(), ClaimFrontierError> {
6884 exit_owners.sort_unstable();
6885 if exit_owners.len() != active.participants.len()
6886 || !exit_owners.iter().copied().eq(active
6887 .participants
6888 .iter()
6889 .map(|participant| participant.participant_index))
6890 {
6891 return Err(sequence_error(
6892 ledger.required_reserve(),
6893 ClaimFrontierInvalidReason::LogicalOwner,
6894 ));
6895 }
6896 Ok(())
6897}
6898
6899fn validate_sequence_terminal_owners(
6900 active: &ActiveIdentityRanks,
6901 terminal_occurrences: &[TerminalOccurrence],
6902 ledger: SequenceLedger,
6903) -> Result<(), ClaimFrontierError> {
6904 for participant in &active.participants {
6905 let matching: Vec<_> = terminal_occurrences
6906 .iter()
6907 .filter(|occurrence| {
6908 occurrence.owner.participant_index == participant.participant_index
6909 })
6910 .copied()
6911 .collect();
6912 match participant.binding {
6913 FrontierBinding::Bound(epoch) => {
6914 if !matches!(matching.as_slice(), [occurrence] if occurrence.owner.binding_epoch == epoch)
6915 {
6916 return Err(sequence_error(
6917 matching.first().map_or_else(
6918 || ledger.required_reserve(),
6919 |occurrence| occurrence.ordinal,
6920 ),
6921 ClaimFrontierInvalidReason::LogicalOwner,
6922 ));
6923 }
6924 }
6925 FrontierBinding::Detached(epoch) => {
6926 if matching.len() > 1
6927 || matching.first().is_some_and(|occurrence| {
6928 occurrence.owner.binding_epoch != epoch
6929 || occurrence.kind != TerminalOccurrenceKind::Candidate
6930 })
6931 {
6932 return Err(sequence_error(
6933 matching.first().map_or_else(
6934 || ledger.required_reserve(),
6935 |occurrence| occurrence.ordinal,
6936 ),
6937 ClaimFrontierInvalidReason::LogicalOwner,
6938 ));
6939 }
6940 }
6941 }
6942 }
6943 Ok(())
6944}
6945
6946fn validate_sequence_products(
6947 active: &ActiveIdentityRanks,
6948 restore: SequenceProductRangesRestore,
6949 terminal_owners: &[BindingTerminalOwner],
6950 recovery: Option<RecoverySequenceBlockRestore>,
6951 recovery_provenance: Option<RecoveryClaimProvenance>,
6952 ledger: SequenceLedger,
6953) -> Result<SequenceProductRanges, ClaimFrontierError> {
6954 let live_count = usize_to_u64(active.participants.len());
6955 let other_count = live_count.saturating_sub(1);
6956 let live_times_terminal = validate_terminal_product_ranges(
6957 restore.live_times_terminal,
6958 terminal_owners,
6959 live_count,
6960 ledger,
6961 )?;
6962 let live_times_replacement_terminal = validate_replacement_product_range(
6963 restore.live_times_replacement_terminal,
6964 recovery,
6965 recovery_provenance,
6966 live_count,
6967 ledger,
6968 )?;
6969 let other_live_times_exit =
6970 validate_exit_product_ranges(active, restore.other_live_times_exit, other_count, ledger)?;
6971 Ok(SequenceProductRanges {
6972 live_times_terminal,
6973 live_times_replacement_terminal,
6974 other_live_times_exit,
6975 })
6976}
6977
6978fn validate_terminal_product_ranges(
6979 mut ranges: Vec<TerminalProductRangeRestore>,
6980 terminal_owners: &[BindingTerminalOwner],
6981 live_count: u64,
6982 ledger: SequenceLedger,
6983) -> Result<Vec<TerminalProductRange>, ClaimFrontierError> {
6984 ranges.sort_by_key(|range| range.start);
6985 let mut seen_terminals = Vec::new();
6986 let mut live_times_terminal = Vec::new();
6987 for range in ranges {
6988 let ordinal = sequence_ordinal(ledger, range.start);
6989 if range.length != live_count
6990 || !terminal_owners.contains(&range.terminal)
6991 || seen_terminals.contains(&range.terminal)
6992 {
6993 return Err(sequence_error(
6994 ordinal,
6995 ClaimFrontierInvalidReason::ProductRange,
6996 ));
6997 }
6998 seen_terminals.push(range.terminal);
6999 live_times_terminal.push(TerminalProductRange {
7000 start: range.start,
7001 length: range.length,
7002 terminal: range.terminal,
7003 });
7004 }
7005 if seen_terminals.len() != terminal_owners.len() {
7006 return Err(sequence_error(
7007 ledger.required_reserve(),
7008 ClaimFrontierInvalidReason::ProductRange,
7009 ));
7010 }
7011 Ok(live_times_terminal)
7012}
7013
7014fn validate_replacement_product_range(
7015 range: Option<ReplacementTerminalProductRangeRestore>,
7016 recovery: Option<RecoverySequenceBlockRestore>,
7017 recovery_provenance: Option<RecoveryClaimProvenance>,
7018 live_count: u64,
7019 ledger: SequenceLedger,
7020) -> Result<Option<ReplacementTerminalProductRange>, ClaimFrontierError> {
7021 let validated = match (range, recovery, recovery_provenance) {
7022 (None, None, None) => None,
7023 (Some(range), Some(_), Some(provenance)) if range.length == live_count => {
7024 Some(ReplacementTerminalProductRange {
7025 start: range.start,
7026 length: range.length,
7027 participant_index: provenance.participant_index,
7028 marker_delivery_seq: provenance.marker_delivery_seq,
7029 prior_binding_epoch: provenance.prior_binding_epoch,
7030 })
7031 }
7032 (Some(range), _, _) => {
7033 return Err(sequence_error(
7034 sequence_ordinal(ledger, range.start),
7035 ClaimFrontierInvalidReason::ProductRange,
7036 ));
7037 }
7038 (None, Some(_), _) | (None, None, Some(_)) => {
7039 return Err(sequence_error(
7040 ledger.required_reserve(),
7041 ClaimFrontierInvalidReason::ProductRange,
7042 ));
7043 }
7044 };
7045 Ok(validated)
7046}
7047
7048fn validate_exit_product_ranges(
7049 active: &ActiveIdentityRanks,
7050 mut ranges: Vec<ExitProductRangeRestore>,
7051 other_count: u64,
7052 ledger: SequenceLedger,
7053) -> Result<Vec<ExitProductRange>, ClaimFrontierError> {
7054 ranges.sort_by_key(|range| range.start);
7055 let mut seen_exits = Vec::new();
7056 let mut other_live_times_exit = Vec::new();
7057 if other_count == 0 && !ranges.is_empty() {
7058 return Err(sequence_error(
7059 ledger.required_reserve(),
7060 ClaimFrontierInvalidReason::ProductRange,
7061 ));
7062 }
7063 for range in ranges {
7064 let ordinal = sequence_ordinal(ledger, range.start);
7065 if range.length != other_count
7066 || !active.contains(range.exit_participant)
7067 || seen_exits.contains(&range.exit_participant)
7068 {
7069 return Err(sequence_error(
7070 ordinal,
7071 ClaimFrontierInvalidReason::ProductRange,
7072 ));
7073 }
7074 seen_exits.push(range.exit_participant);
7075 other_live_times_exit.push(ExitProductRange {
7076 start: range.start,
7077 length: range.length,
7078 exit_participant: range.exit_participant,
7079 });
7080 }
7081 seen_exits.sort_unstable();
7082 let expected_exit_ranges = if other_count == 0 {
7083 0
7084 } else {
7085 active.participants.len()
7086 };
7087 if seen_exits.len() != expected_exit_ranges
7088 || !seen_exits.iter().copied().eq(active
7089 .participants
7090 .iter()
7091 .take(expected_exit_ranges)
7092 .map(|participant| participant.participant_index))
7093 {
7094 return Err(sequence_error(
7095 ledger.required_reserve(),
7096 ClaimFrontierInvalidReason::ProductRange,
7097 ));
7098 }
7099 Ok(other_live_times_exit)
7100}
7101
7102fn validate_order_recovery(
7103 active: &ActiveIdentityRanks,
7104 recovery: Option<RecoveryOrderBlockRestore>,
7105 provenance: Option<RecoveryClaimProvenance>,
7106 ledger: OrderLedger,
7107 frontier_length: u128,
7108) -> Result<(), ClaimFrontierError> {
7109 let claims = ledger.claims();
7110 let expected = claims.recovery_operation() && claims.recovery_replacement_terminal();
7111 match (expected, recovery, provenance) {
7112 (false, None, None) => Ok(()),
7113 (true, None, _) | (true, Some(_), None) | (false, None, Some(_)) => Err(order_error(
7114 frontier_length,
7115 ClaimFrontierInvalidReason::RecoveryBlock,
7116 )),
7117 (false, Some(block), _) => Err(order_error(
7118 order_ordinal(ledger, block_start_order(block)),
7119 ClaimFrontierInvalidReason::RecoveryBlock,
7120 )),
7121 (true, Some(block), Some(provenance)) => {
7122 let block_ordinal = order_ordinal(ledger, block_start_order(block));
7123 let expected_recovery_operation = block
7124 .active_binding
7125 .map_or(Some(block.recovery_operation_order), |active_binding| {
7126 active_binding.transaction_order.checked_add(1)
7127 });
7128 if expected_recovery_operation != Some(block.recovery_operation_order) {
7129 return Err(order_error(
7130 block_ordinal + 1,
7131 ClaimFrontierInvalidReason::RecoveryBlock,
7132 ));
7133 }
7134 if block.recovery_operation_order.checked_add(1)
7135 != Some(block.replacement_terminal_order)
7136 {
7137 return Err(order_error(
7138 block_ordinal + u128::from(block.active_binding.is_some()) + 1,
7139 ClaimFrontierInvalidReason::RecoveryBlock,
7140 ));
7141 }
7142 let Some(participant) = active_participant(active, provenance.participant_index) else {
7143 return Err(order_error(
7144 block_ordinal,
7145 ClaimFrontierInvalidReason::LogicalOwner,
7146 ));
7147 };
7148 let expected_binding = match provenance.phase {
7149 RecoveryClaimPhase::PreFate => {
7150 FrontierBinding::Bound(provenance.prior_binding_epoch)
7151 }
7152 RecoveryClaimPhase::PostFate => {
7153 FrontierBinding::Detached(provenance.prior_binding_epoch)
7154 }
7155 RecoveryClaimPhase::RecoveredBound => {
7156 FrontierBinding::Bound(provenance.current_binding_epoch)
7157 }
7158 };
7159 if participant.binding != expected_binding {
7160 return Err(order_error(
7161 block_ordinal,
7162 ClaimFrontierInvalidReason::LogicalOwner,
7163 ));
7164 }
7165 let active_binding_valid = match (provenance.phase, block.active_binding) {
7166 (RecoveryClaimPhase::PreFate, Some(active_binding)) => {
7167 active_binding.owner.participant_index == provenance.participant_index
7168 && active_binding.owner.binding_epoch == provenance.prior_binding_epoch
7169 }
7170 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
7171 _ => false,
7172 };
7173 if !active_binding_valid {
7174 return Err(order_error(
7175 block_ordinal,
7176 ClaimFrontierInvalidReason::RecoveryBlock,
7177 ));
7178 }
7179 Ok(())
7180 }
7181 }
7182}
7183
7184fn validate_order_candidates(
7185 restore: &[ImmutableOrderCandidateMajorRestore],
7186 ledger: OrderLedger,
7187) -> Result<Vec<ImmutableOrderCandidateMajor>, ClaimFrontierError> {
7188 let mut groups = restore.to_vec();
7189 groups.sort_by_key(|group| group.transaction_order);
7190 let mut seen_keys = Vec::new();
7191 let mut previous_major = None;
7192 let mut validated = Vec::new();
7193 for group in groups {
7194 let ordinal = order_ordinal(ledger, group.transaction_order);
7195 let below_allocated_high = matches!(
7196 ledger.high(),
7197 OrderHigh::Allocated(high) if group.transaction_order < high
7198 );
7199 if group.candidate_keys.is_empty()
7200 || previous_major == Some(group.transaction_order)
7201 || below_allocated_high
7202 {
7203 return Err(order_error(
7204 ordinal,
7205 ClaimFrontierInvalidReason::CandidateKey,
7206 ));
7207 }
7208 previous_major = Some(group.transaction_order);
7209 let mut previous = None;
7210 for key in &group.candidate_keys {
7211 if key.transaction_order() != group.transaction_order
7212 || previous.is_some_and(|previous| previous >= *key)
7213 || seen_keys.contains(key)
7214 {
7215 return Err(order_error(
7216 ordinal,
7217 ClaimFrontierInvalidReason::CandidateKey,
7218 ));
7219 }
7220 previous = Some(*key);
7221 seen_keys.push(*key);
7222 }
7223 validated.push(ImmutableOrderCandidateMajor {
7224 transaction_order: group.transaction_order,
7225 candidate_keys: group.candidate_keys,
7226 });
7227 }
7228 Ok(validated)
7229}
7230
7231fn validate_order_direct_owners(
7232 active: &ActiveIdentityRanks,
7233 movable: &[MovableOrderClaim],
7234 recovery: Option<RecoveryOrderBlockRestore>,
7235 ledger: OrderLedger,
7236) -> Result<(), ClaimFrontierError> {
7237 let mut ordered = movable.to_vec();
7238 ordered.sort_by_key(|claim| claim.transaction_order);
7239 let mut exits = Vec::new();
7240 let mut terminals = Vec::new();
7241 for claim in ordered {
7242 let ordinal = order_ordinal(ledger, claim.transaction_order);
7243 match claim.owner {
7244 OrderDirectOwner::MembershipExit { participant_index } => {
7245 if !active.contains(participant_index) || exits.contains(&participant_index) {
7246 return Err(order_error(
7247 ordinal,
7248 ClaimFrontierInvalidReason::LogicalOwner,
7249 ));
7250 }
7251 exits.push(participant_index);
7252 }
7253 OrderDirectOwner::ActiveBindingTerminal(owner) => {
7254 if !terminal_matches_bound(active, owner) || terminals.contains(&owner) {
7255 return Err(order_error(
7256 ordinal,
7257 ClaimFrontierInvalidReason::LogicalOwner,
7258 ));
7259 }
7260 terminals.push(owner);
7261 }
7262 }
7263 }
7264 if let Some(active_binding) = recovery.and_then(|block| block.active_binding) {
7265 let ordinal = order_ordinal(ledger, active_binding.transaction_order);
7266 if !terminal_matches_bound(active, active_binding.owner)
7267 || terminals.contains(&active_binding.owner)
7268 {
7269 return Err(order_error(
7270 ordinal,
7271 ClaimFrontierInvalidReason::LogicalOwner,
7272 ));
7273 }
7274 terminals.push(active_binding.owner);
7275 }
7276 exits.sort_unstable();
7277 if exits.len() != active.participants.len()
7278 || !exits.iter().copied().eq(active
7279 .participants
7280 .iter()
7281 .map(|participant| participant.participant_index))
7282 {
7283 return Err(order_error(
7284 ledger.claims().total(),
7285 ClaimFrontierInvalidReason::LogicalOwner,
7286 ));
7287 }
7288 terminals.sort_by_key(terminal_owner_sort_key);
7289 if usize_to_u128(terminals.len()) != u128::from(ledger.claims().active_binding_terminals()) {
7290 return Err(order_error(
7291 ledger.claims().total(),
7292 ClaimFrontierInvalidReason::LogicalOwner,
7293 ));
7294 }
7295 Ok(())
7296}
7297
7298fn validate_cross_counter(
7299 sequence: &SequenceClaimFrontier,
7300 order: &OrderClaimFrontier,
7301) -> Result<(), ClaimFrontierError> {
7302 match (sequence.recovery, order.recovery) {
7303 (None, None) => {}
7304 (Some(sequence_block), Some(order_block))
7305 if sequence_block.participant_index == order_block.participant_index
7306 && sequence_block.marker_delivery_seq == order_block.marker_delivery_seq
7307 && sequence_block.recovered_binding_epoch
7308 == order_block.recovered_binding_epoch
7309 && sequence_block.terminal.map(|terminal| terminal.owner)
7310 == order_block.active_binding.map(|active| active.owner) => {}
7311 (Some(sequence_block), _) => {
7312 return Err(sequence_error(
7313 sequence_ordinal(
7314 sequence.ledger,
7315 block_start_validated_sequence(sequence_block),
7316 ),
7317 ClaimFrontierInvalidReason::RecoveryBlock,
7318 ));
7319 }
7320 (None, Some(order_block)) => {
7321 return Err(order_error(
7322 order_ordinal(order.ledger, block_start_validated_order(order_block)),
7323 ClaimFrontierInvalidReason::RecoveryBlock,
7324 ));
7325 }
7326 }
7327
7328 let mut order_candidate_keys = Vec::new();
7329 for group in &order.immutable_candidates {
7330 order_candidate_keys.extend(group.candidate_keys.iter().copied());
7331 }
7332 for candidate in &sequence.immutable_candidates {
7333 let key = candidate.admission_order();
7334 if !order_candidate_keys.contains(&key) {
7335 return Err(sequence_error(
7336 sequence_ordinal(sequence.ledger, candidate.delivery_seq()),
7337 ClaimFrontierInvalidReason::CandidateKey,
7338 ));
7339 }
7340 }
7341 for group in &order.immutable_candidates {
7342 for key in &group.candidate_keys {
7343 if !sequence
7344 .immutable_candidates
7345 .iter()
7346 .any(|candidate| candidate.admission_order() == *key)
7347 {
7348 return Err(order_error(
7349 order_ordinal(order.ledger, group.transaction_order),
7350 ClaimFrontierInvalidReason::CandidateKey,
7351 ));
7352 }
7353 }
7354 }
7355
7356 let mut sequence_movable_terminals = Vec::new();
7357 for claim in &sequence.movable_claims {
7358 if let SequenceDirectOwner::BindingTerminal(owner) = claim.owner {
7359 sequence_movable_terminals.push(owner);
7360 }
7361 }
7362 if let Some(terminal) = sequence.recovery.and_then(|block| block.terminal) {
7363 sequence_movable_terminals.push(terminal.owner);
7364 }
7365 let mut order_movable_terminals = Vec::new();
7366 for claim in &order.movable_claims {
7367 if let OrderDirectOwner::ActiveBindingTerminal(owner) = claim.owner {
7368 order_movable_terminals.push(owner);
7369 }
7370 }
7371 if let Some(active_binding) = order.recovery.and_then(|block| block.active_binding) {
7372 order_movable_terminals.push(active_binding.owner);
7373 }
7374 sequence_movable_terminals.sort_by_key(terminal_owner_sort_key);
7375 order_movable_terminals.sort_by_key(terminal_owner_sort_key);
7376 if sequence_movable_terminals != order_movable_terminals {
7377 return Err(sequence_error(
7378 sequence.ledger.required_reserve(),
7379 ClaimFrontierInvalidReason::LogicalOwner,
7380 ));
7381 }
7382 Ok(())
7383}
7384
7385const fn marker_provenance_targets(provenance: MarkerProvenance, target: ParticipantId) -> bool {
7386 match provenance {
7387 MarkerProvenance::NonProductM => true,
7388 MarkerProvenance::TerminalProduct {
7389 affected_participant,
7390 ..
7391 } => affected_participant == target,
7392 MarkerProvenance::ExitProduct {
7393 exit_participant,
7394 remaining_participant,
7395 } => exit_participant != remaining_participant && remaining_participant == target,
7396 }
7397}
7398
7399fn marker_has_causal_authority(
7400 marker: MarkerCandidateAuthority,
7401 records: &[RetainedCausalRecord],
7402 historical: &[HistoricalCausalAuthority],
7403) -> bool {
7404 if marker.provenance == MarkerProvenance::NonProductM {
7405 return true;
7406 }
7407 let retained_match = records.iter().any(|record| {
7408 if record.admission_order.transaction_order() != marker.admission_order.transaction_order()
7409 {
7410 return false;
7411 }
7412 match marker.provenance {
7413 MarkerProvenance::NonProductM => true,
7414 MarkerProvenance::TerminalProduct {
7415 terminal: TerminalProductSource::Binding(owner),
7416 ..
7417 } => matches!(
7418 record.kind,
7419 RetainedCausalRecordKind::BindingTerminal(actual) if actual == owner
7420 ),
7421 MarkerProvenance::TerminalProduct {
7422 terminal:
7423 TerminalProductSource::RecoveryReplacement {
7424 participant_index,
7425 binding_epoch,
7426 },
7427 ..
7428 } => matches!(
7429 record.kind,
7430 RetainedCausalRecordKind::BindingTerminal(owner)
7431 if owner.participant_index == participant_index
7432 && owner.binding_epoch == binding_epoch
7433 ),
7434 MarkerProvenance::ExitProduct {
7435 exit_participant, ..
7436 } => matches!(
7437 record.kind,
7438 RetainedCausalRecordKind::MembershipExit { participant_index }
7439 if participant_index == exit_participant
7440 ),
7441 }
7442 });
7443 retained_match
7444 || historical
7445 .iter()
7446 .any(|authority| match (marker.provenance, authority.kind) {
7447 (
7448 MarkerProvenance::TerminalProduct {
7449 terminal: TerminalProductSource::Binding(expected),
7450 ..
7451 },
7452 HistoricalCausalKind::BindingTerminal(owner),
7453 ) => {
7454 owner == expected
7455 && authority.admission_order.transaction_order()
7456 == marker.admission_order.transaction_order()
7457 }
7458 (
7459 MarkerProvenance::TerminalProduct {
7460 terminal:
7461 TerminalProductSource::RecoveryReplacement {
7462 participant_index,
7463 binding_epoch,
7464 },
7465 ..
7466 },
7467 HistoricalCausalKind::BindingTerminal(owner),
7468 ) => {
7469 owner.participant_index == participant_index
7470 && owner.binding_epoch == binding_epoch
7471 && authority.admission_order.transaction_order()
7472 == marker.admission_order.transaction_order()
7473 }
7474 (
7475 MarkerProvenance::ExitProduct {
7476 exit_participant, ..
7477 },
7478 HistoricalCausalKind::MembershipExit(participant_index),
7479 ) => {
7480 participant_index == exit_participant
7481 && authority.admission_order.transaction_order()
7482 == marker.admission_order.transaction_order()
7483 }
7484 _ => false,
7485 })
7486}
7487
7488fn terminal_matches_active(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
7489 active_participant(active, owner.participant_index)
7490 .is_some_and(|participant| binding_epoch(participant.binding) == owner.binding_epoch)
7491}
7492
7493fn terminal_matches_bound(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
7494 active_participant(active, owner.participant_index).is_some_and(|participant| {
7495 participant.binding == FrontierBinding::Bound(owner.binding_epoch)
7496 })
7497}
7498
7499fn active_participant(
7500 active: &ActiveIdentityRanks,
7501 participant_index: ParticipantId,
7502) -> Option<FrontierParticipant> {
7503 active
7504 .participants
7505 .binary_search_by_key(&participant_index, |participant| {
7506 participant.participant_index
7507 })
7508 .ok()
7509 .and_then(|index| active.participants.get(index))
7510 .copied()
7511}
7512
7513const fn binding_epoch(binding: FrontierBinding) -> BindingEpoch {
7514 match binding {
7515 FrontierBinding::Bound(epoch) | FrontierBinding::Detached(epoch) => epoch,
7516 }
7517}
7518
7519fn block_start_sequence(block: RecoverySequenceBlockRestore) -> DeliverySeq {
7520 block
7521 .terminal
7522 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
7523}
7524
7525fn block_start_validated_sequence(block: RecoverySequenceBlock) -> DeliverySeq {
7526 block
7527 .terminal
7528 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
7529}
7530
7531fn block_start_order(block: RecoveryOrderBlockRestore) -> TransactionOrder {
7532 block
7533 .active_binding
7534 .map_or(block.recovery_operation_order, |active_binding| {
7535 active_binding.transaction_order
7536 })
7537}
7538
7539fn block_start_validated_order(block: RecoveryOrderBlock) -> TransactionOrder {
7540 block
7541 .active_binding
7542 .map_or(block.recovery_operation_order, |active_binding| {
7543 active_binding.transaction_order
7544 })
7545}
7546
7547fn order_frontier_start(high: OrderHigh) -> u128 {
7548 match high {
7549 OrderHigh::Empty => 0,
7550 OrderHigh::Allocated(high) => u128::from(high) + 1,
7551 }
7552}
7553
7554const fn order_is_above_high(value: TransactionOrder, high: OrderHigh) -> bool {
7555 match high {
7556 OrderHigh::Empty => true,
7557 OrderHigh::Allocated(high) => value > high,
7558 }
7559}
7560
7561fn order_frontier_candidate_count(restore: &OrderClaimFrontierRestore, high: OrderHigh) -> u128 {
7562 usize_to_u128(
7563 restore
7564 .immutable_candidates
7565 .iter()
7566 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
7567 .count(),
7568 )
7569}
7570
7571fn sequence_ordinal(ledger: SequenceLedger, value: DeliverySeq) -> u128 {
7572 u128::from(value).saturating_sub(u128::from(ledger.high_watermark()) + 1)
7573}
7574
7575fn order_ordinal(ledger: OrderLedger, value: TransactionOrder) -> u128 {
7576 u128::from(value).saturating_sub(order_frontier_start(ledger.high()))
7577}
7578
7579fn checked_rank_value(start: DeliverySeq, active_rank: usize) -> Option<DeliverySeq> {
7580 let rank = u64::try_from(active_rank).ok()?;
7581 start.checked_add(rank)
7582}
7583
7584fn usize_to_u64(value: usize) -> u64 {
7585 u64::try_from(value).unwrap_or(u64::MAX)
7586}
7587
7588fn usize_to_u128(value: usize) -> u128 {
7589 u64::try_from(value).map_or(u128::MAX, u128::from)
7590}
7591
7592fn rank_index(rank: usize) -> u128 {
7593 usize_to_u128(rank)
7594}
7595
7596const fn frontier_error(
7597 counter: ClaimFrontierCounter,
7598 first_bad_position: u128,
7599 reason: ClaimFrontierInvalidReason,
7600) -> ClaimFrontierError {
7601 ClaimFrontierError {
7602 counter,
7603 first_bad_position,
7604 reason,
7605 }
7606}
7607
7608const fn sequence_error(
7609 first_bad_position: u128,
7610 reason: ClaimFrontierInvalidReason,
7611) -> ClaimFrontierError {
7612 frontier_error(
7613 ClaimFrontierCounter::DeliverySequence,
7614 first_bad_position,
7615 reason,
7616 )
7617}
7618
7619const fn order_error(
7620 first_bad_position: u128,
7621 reason: ClaimFrontierInvalidReason,
7622) -> ClaimFrontierError {
7623 frontier_error(
7624 ClaimFrontierCounter::TransactionOrder,
7625 first_bad_position,
7626 reason,
7627 )
7628}