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