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)]
1985pub enum PrecedenceCondition {
1986 BindingTerminal,
1990 MarkerDrain {
1992 settlement_epoch: DeliverySeq,
2000 },
2001 FencedRecovery,
2007 Unclassified,
2011}
2012
2013#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2015pub(in crate::lifecycle) enum LiveFrontierTransitionError {
2016 Authority,
2018 Precedence(PrecedenceCondition),
2023 RecordPosition,
2025 Exhausted,
2027 ResultingFrontier,
2029}
2030
2031fn precedence(
2033 frontiers: ClaimFrontiers,
2034 condition: PrecedenceCondition,
2035) -> Box<(ClaimFrontiers, LiveFrontierTransitionError)> {
2036 Box::new((
2037 frontiers,
2038 LiveFrontierTransitionError::Precedence(condition),
2039 ))
2040}
2041
2042fn select_retained_marker_records(records: &[RetainedCausalRecord]) -> Vec<RetainedCausalRecord> {
2043 records
2044 .iter()
2045 .copied()
2046 .filter(|record| {
2047 matches!(
2048 record.kind,
2049 RetainedCausalRecordKind::CompactionMarker { .. }
2050 )
2051 })
2052 .collect()
2053}
2054
2055impl ClaimFrontiers {
2056 pub fn from_initial_enrollment<F>(
2071 operation: InitialEnrollmentOperationCommit<F>,
2072 attached_charge: ResourceVector,
2073 ) -> Result<InitialEnrollmentFrontierCommit<F>, Box<InitialEnrollmentFrontierFailure<F>>> {
2074 let shape = match initial_enrollment_frontier_shape(&operation, attached_charge) {
2075 Ok(shape) => shape,
2076 Err(error) => {
2077 return Err(Box::new(InitialEnrollmentFrontierFailure {
2078 operation,
2079 error,
2080 }));
2081 }
2082 };
2083 let closure_accounting = shape.closure_accounting;
2084 let frontiers = match build_initial_enrollment_frontiers(&shape) {
2085 Ok(frontiers) => frontiers,
2086 Err(error) => {
2087 return Err(Box::new(InitialEnrollmentFrontierFailure {
2088 operation,
2089 error,
2090 }));
2091 }
2092 };
2093 Ok(InitialEnrollmentFrontierCommit {
2094 operation,
2095 frontiers,
2096 closure_accounting,
2097 attached_charge,
2098 })
2099 }
2100
2101 pub fn restore(
2114 restore: ClaimFrontiersRestore,
2115 sequence_ledger: SequenceLedger,
2116 order_ledger: OrderLedger,
2117 ) -> Result<Self, ParticipantStateCorruptReason> {
2118 let history = ValidatedConversationHistory::empty();
2119 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)?
2120 .finish(None)
2121 }
2122
2123 #[cfg(any(test, feature = "test-support"))]
2126 pub(super) fn prevalidate(
2127 restore: ClaimFrontiersRestore,
2128 sequence_ledger: SequenceLedger,
2129 order_ledger: OrderLedger,
2130 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2131 let history = ValidatedConversationHistory::empty();
2132 Self::prevalidate_with_history(restore, sequence_ledger, order_ledger, &history)
2133 }
2134
2135 pub(super) fn prevalidate_with_history(
2136 restore: ClaimFrontiersRestore,
2137 sequence_ledger: SequenceLedger,
2138 order_ledger: OrderLedger,
2139 history: &ValidatedConversationHistory,
2140 ) -> Result<ClaimFrontiersPrevalidated, ParticipantStateCorruptReason> {
2141 validate_sequence_numeric(&restore.sequence, sequence_ledger).map_err(corrupt_frontier)?;
2142 validate_order_numeric(&restore.order, order_ledger).map_err(corrupt_frontier)?;
2143 validate_unique_candidate_keys(
2144 &restore.sequence.immutable_candidates,
2145 &restore.retained_records,
2146 )?;
2147 validate_bounded_shape(&restore, sequence_ledger).map_err(corrupt_frontier)?;
2148 let active_identities = ActiveIdentityRanks::try_new(
2149 restore.active_identities,
2150 sequence_ledger.high_watermark(),
2151 restore.identity_slot_limit,
2152 )
2153 .map_err(corrupt_frontier)?;
2154 let retained_records = validated_retained_records(
2155 restore.retained_records,
2156 restore.retained_floor,
2157 restore.retained_record_limit,
2158 restore.identity_slot_limit,
2159 sequence_ledger,
2160 )
2161 .map_err(corrupt_frontier)?;
2162 let historical_causal_authorities = validated_historical_authorities(
2163 restore.historical_causal_facts,
2164 restore.conversation_id,
2165 restore.identity_slot_limit,
2166 sequence_ledger,
2167 history,
2168 )
2169 .map_err(corrupt_frontier)?;
2170 let retained_marker_records = select_retained_marker_records(&retained_records);
2171 let marker_records = validated_active_marker_records(
2172 &retained_marker_records,
2173 restore.active_marker_anchors,
2174 restore.identity_slot_limit,
2175 sequence_ledger,
2176 )
2177 .map_err(corrupt_frontier)?;
2178 let historical_marker_deliveries = validated_historical_marker_deliveries(
2179 restore.historical_marker_deliveries,
2180 restore.conversation_id,
2181 &active_identities,
2182 &MarkerDeliverySources {
2183 retained: &retained_records,
2184 historical: &historical_causal_authorities,
2185 candidates: &restore.sequence.immutable_candidates,
2186 },
2187 restore.retained_record_limit,
2188 sequence_ledger,
2189 )
2190 .map_err(corrupt_frontier)?;
2191 BindingOriginValidation {
2192 conversation_id: restore.conversation_id,
2193 active: &active_identities,
2194 origins: &history.binding_origins,
2195 retained_records: &retained_records,
2196 causal_authorities: &history.causal_authorities,
2197 historical_marker_deliveries: &historical_marker_deliveries,
2198 total: history.total,
2199 ledger: sequence_ledger,
2200 }
2201 .validate()
2202 .map_err(corrupt_frontier)?;
2203 validate_sequence_candidates(
2204 &active_identities,
2205 &restore.sequence.immutable_candidates,
2206 restore.retained_floor,
2207 &retained_records,
2208 &historical_causal_authorities,
2209 sequence_ledger,
2210 )
2211 .map_err(corrupt_frontier)?;
2212 validate_marker_credit_owners(
2213 &restore.sequence.immutable_candidates,
2214 &marker_records,
2215 restore.identity_slot_limit,
2216 sequence_ledger,
2217 )
2218 .map_err(corrupt_frontier)?;
2219 Ok(ClaimFrontiersPrevalidated {
2220 conversation_id: restore.conversation_id,
2221 active_identities,
2222 identity_slot_limit: restore.identity_slot_limit,
2223 retained_floor: restore.retained_floor,
2224 retained_records,
2225 marker_records,
2226 historical_marker_deliveries,
2227 historical_causal_authorities,
2228 binding_origins: history.binding_origins.clone(),
2229 sequence_restore: restore.sequence,
2230 order_restore: restore.order,
2231 recovery_marker_delivery_seq: restore.recovery_marker_delivery_seq,
2232 sequence_ledger,
2233 order_ledger,
2234 issued_marker_record: None,
2235 })
2236 }
2237
2238 #[must_use]
2240 pub const fn active_identities(&self) -> &ActiveIdentityRanks {
2241 &self.active_identities
2242 }
2243
2244 #[must_use]
2246 pub const fn identity_slot_limit(&self) -> u64 {
2247 self.identity_slot_limit
2248 }
2249
2250 #[must_use]
2252 pub const fn conversation_id(&self) -> ConversationId {
2253 self.conversation_id
2254 }
2255
2256 #[must_use]
2258 pub const fn retained_floor(&self) -> u128 {
2259 self.retained_floor
2260 }
2261
2262 #[must_use]
2264 pub fn retained_records(&self) -> &[RetainedCausalRecord] {
2265 &self.retained_records
2266 }
2267
2268 #[must_use]
2270 pub fn retained_marker_records(&self) -> &[RetainedCausalRecord] {
2271 &self.marker_records
2272 }
2273
2274 #[must_use]
2279 pub(in crate::lifecycle) fn unaccepted_marker_anchor_count(&self) -> u64 {
2280 ordinary_unaccepted_marker_anchors(self).len() as u64
2281 }
2282
2283 #[must_use]
2290 pub(in crate::lifecycle) fn pending_marker_candidate_count(&self) -> u64 {
2291 self.sequence
2292 .immutable_candidates
2293 .iter()
2294 .filter(|candidate| matches!(candidate, ImmutableSequenceCandidate::Marker(_)))
2295 .count() as u64
2296 }
2297
2298 #[must_use]
2306 pub fn project_offered_marker_progress(
2307 &self,
2308 participant_id: ParticipantId,
2309 binding_epoch: BindingEpoch,
2310 marker_delivery_seq: DeliverySeq,
2311 event: Event,
2312 ) -> Option<ParticipantCursorProgress> {
2313 let record = self
2314 .marker_records
2315 .iter()
2316 .find(|record| record.delivery_seq == marker_delivery_seq)
2317 .copied()?;
2318 let RetainedCausalRecordKind::CompactionMarker {
2319 participant_index,
2320 provenance,
2321 } = record.kind
2322 else {
2323 return None;
2324 };
2325 if participant_index != participant_id {
2326 return None;
2327 }
2328 let participant = active_participant(&self.active_identities, participant_id)?;
2329 let target_binding = FrontierBinding::Bound(binding_epoch);
2330 if participant.binding != target_binding {
2331 return None;
2332 }
2333 let authority = ValidatedMarkerRecord {
2334 conversation_id: self.conversation_id,
2335 record,
2336 provenance,
2337 target_binding,
2338 occurrence: MarkerRecordOccurrence::Undelivered,
2339 seal: MarkerAuthoritySeal::Validated,
2340 };
2341 MarkerDelivery::from_validated_record(&authority)
2342 .delivered_progress(event)
2343 .ok()
2344 }
2345
2346 pub(super) fn fenced_marker_source(
2349 &self,
2350 recovery: super::DetachedCredentialRecovery,
2351 ) -> Option<FencedMarkerSourceRecord> {
2352 if self.fenced_marker_issued.is_some() || recovery.conversation_id() != self.conversation_id
2353 {
2354 return None;
2355 }
2356 let marker_delivery_seq = recovery.marker_delivery_seq();
2357 let participant_id = recovery.participant_id();
2358 let prior_binding_epoch = recovery.prior_binding_epoch();
2359 let record = self
2360 .marker_records
2361 .iter()
2362 .find(|record| record.delivery_seq == marker_delivery_seq)
2363 .copied()?;
2364 let RetainedCausalRecordKind::CompactionMarker {
2365 participant_index,
2366 provenance,
2367 } = record.kind
2368 else {
2369 return None;
2370 };
2371 let participant = active_participant(&self.active_identities, participant_id)?;
2372 let target_binding = FrontierBinding::Detached(prior_binding_epoch);
2373 if participant_index != participant_id || participant.binding != target_binding {
2374 return None;
2375 }
2376 Some(FencedMarkerSourceRecord {
2377 conversation_id: self.conversation_id,
2378 delivery_seq: record.delivery_seq,
2379 admission_order: record.admission_order,
2380 participant_id,
2381 provenance,
2382 target_binding,
2383 })
2384 }
2385
2386 pub(in crate::lifecycle) fn take_fenced_marker_record(
2395 &mut self,
2396 recovery: super::DetachedCredentialRecovery,
2397 ) -> Option<ValidatedMarkerRecord> {
2398 let source = self.fenced_marker_source(recovery)?;
2399 self.fenced_marker_issued = Some((
2400 source.participant_id,
2401 source.delivery_seq,
2402 recovery.prior_binding_epoch(),
2403 ));
2404 Some(ValidatedMarkerRecord {
2405 conversation_id: source.conversation_id,
2406 record: RetainedCausalRecord {
2407 delivery_seq: source.delivery_seq,
2408 admission_order: source.admission_order,
2409 kind: RetainedCausalRecordKind::CompactionMarker {
2410 participant_index: source.participant_id,
2411 provenance: source.provenance,
2412 },
2413 },
2414 provenance: source.provenance,
2415 target_binding: source.target_binding,
2416 occurrence: MarkerRecordOccurrence::Delivered,
2417 seal: MarkerAuthoritySeal::Validated,
2418 })
2419 }
2420
2421 pub(in crate::lifecycle) const fn reinstall_fenced_marker_record(
2424 &mut self,
2425 record: ValidatedMarkerRecord,
2426 ) {
2427 self.fenced_marker_issued = None;
2428 record.consume();
2429 }
2430
2431 #[must_use]
2433 pub const fn sequence(&self) -> &SequenceClaimFrontier {
2434 &self.sequence
2435 }
2436
2437 #[must_use]
2439 pub const fn order(&self) -> &OrderClaimFrontier {
2440 &self.order
2441 }
2442
2443 #[cfg(test)]
2444 pub(in crate::lifecycle) fn cross_counter_valid_for_test(&self) -> bool {
2445 validate_cross_counter(&self.sequence, &self.order).is_ok()
2446 }
2447
2448 #[must_use]
2456 pub fn live_transition_blocked(&self) -> bool {
2457 !self.sequence.immutable_candidates.is_empty()
2458 || self.sequence.recovery.is_some()
2459 || !self.order.immutable_candidates.is_empty()
2460 || self.order.recovery.is_some()
2461 }
2462
2463 #[must_use]
2480 pub fn precedence_condition(&self) -> PrecedenceCondition {
2481 if self.sequence.recovery.is_some() || self.order.recovery.is_some() {
2482 return PrecedenceCondition::FencedRecovery;
2483 }
2484 match self.sequence.immutable_candidates.first() {
2485 Some(ImmutableSequenceCandidate::Marker(marker)) => PrecedenceCondition::MarkerDrain {
2486 settlement_epoch: marker.delivery_seq,
2487 },
2488 Some(ImmutableSequenceCandidate::BindingTerminal { .. }) => {
2489 PrecedenceCondition::BindingTerminal
2490 }
2491 None => PrecedenceCondition::Unclassified,
2492 }
2493 }
2494
2495 pub(in crate::lifecycle) fn apply_live_transition(
2501 self,
2502 active_identities: Vec<FrontierParticipant>,
2503 appended_records: &[RetainedCausalRecord],
2504 sequence_ledger: SequenceLedger,
2505 order_ledger: OrderLedger,
2506 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2507 if self.live_transition_blocked() {
2508 let condition = self.precedence_condition();
2509 return Err(Box::new((
2510 self,
2511 LiveFrontierTransitionError::Precedence(condition),
2512 )));
2513 }
2514 let Ok(active) = ActiveIdentityRanks::try_new(
2515 active_identities,
2516 sequence_ledger.high_watermark(),
2517 self.identity_slot_limit,
2518 ) else {
2519 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2520 };
2521 let first_sequence = self.sequence.ledger.high_watermark().checked_add(1);
2522 if first_sequence.is_none_or(|first| {
2523 appended_records.iter().enumerate().any(|(index, record)| {
2524 u64::try_from(index)
2525 .ok()
2526 .and_then(|offset| first.checked_add(offset))
2527 != Some(record.delivery_seq)
2528 })
2529 }) || appended_records
2530 .last()
2531 .is_some_and(|record| record.delivery_seq != sequence_ledger.high_watermark())
2532 {
2533 return Err(Box::new((
2534 self,
2535 LiveFrontierTransitionError::RecordPosition,
2536 )));
2537 }
2538 let (sequence, order) =
2539 match rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger) {
2540 Ok(frontiers) => frontiers,
2541 Err(error) => return Err(Box::new((self, error))),
2542 };
2543 let Self {
2544 conversation_id,
2545 identity_slot_limit,
2546 retained_floor,
2547 mut retained_records,
2548 marker_records,
2549 fenced_marker_issued,
2550 ..
2551 } = self;
2552 retained_records.extend_from_slice(appended_records);
2553 Ok(Self {
2554 conversation_id,
2555 active_identities: active,
2556 identity_slot_limit,
2557 retained_floor,
2558 retained_records,
2559 marker_records,
2560 fenced_marker_issued,
2561 sequence,
2562 order,
2563 })
2564 }
2565
2566 pub(in crate::lifecycle) fn apply_pending_binding_terminal(
2570 mut self,
2571 participant_id: ParticipantId,
2572 binding_epoch: BindingEpoch,
2573 delivery_seq: DeliverySeq,
2574 admission_order: super::AdmissionOrder,
2575 order_ledger: OrderLedger,
2576 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2577 if self.live_transition_blocked() {
2578 let condition = self.precedence_condition();
2579 return Err(Box::new((
2580 self,
2581 LiveFrontierTransitionError::Precedence(condition),
2582 )));
2583 }
2584 let mut participants = self.active_identities.participants().to_vec();
2585 let Some(participant) = participants
2586 .iter_mut()
2587 .find(|participant| participant.participant_index() == participant_id)
2588 else {
2589 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2590 };
2591 if participant.binding() != FrontierBinding::Bound(binding_epoch) {
2592 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2593 }
2594 *participant = FrontierParticipant::new(
2595 participant_id,
2596 participant.cursor(),
2597 FrontierBinding::Detached(binding_epoch),
2598 );
2599 let Ok(active) = ActiveIdentityRanks::try_new(
2600 participants,
2601 self.sequence.ledger.high_watermark(),
2602 self.identity_slot_limit,
2603 ) else {
2604 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2605 };
2606 let pending_owner = BindingTerminalOwner {
2607 participant_index: participant_id,
2608 binding_epoch,
2609 };
2610 let (sequence, order) = match rebuild_pending_terminal_frontiers(
2611 &active,
2612 pending_owner,
2613 delivery_seq,
2614 admission_order,
2615 self.sequence.ledger,
2616 order_ledger,
2617 ) {
2618 Ok(frontiers) => frontiers,
2619 Err(error) => return Err(Box::new((self, error))),
2620 };
2621 self.active_identities = active;
2622 self.sequence = sequence;
2623 self.order = order;
2624 Ok(self)
2625 }
2626
2627 pub(in crate::lifecycle) fn apply_live_fenced_attach(
2635 self,
2636 participant: FrontierParticipant,
2637 prior_binding_epoch: BindingEpoch,
2638 appended_records: &[RetainedCausalRecord],
2639 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2640 let Some(current) = self
2641 .active_identities
2642 .participants()
2643 .iter()
2644 .find(|current| current.participant_index() == participant.participant_index())
2645 .copied()
2646 else {
2647 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2648 };
2649 let Some(sequence_recovery) = self.sequence.recovery else {
2650 return Err(precedence(self, PrecedenceCondition::FencedRecovery));
2651 };
2652 let Some(order_recovery) = self.order.recovery else {
2653 return Err(precedence(self, PrecedenceCondition::FencedRecovery));
2654 };
2655 if !Self::fenced_recovery_authority_matches(
2656 current,
2657 participant,
2658 prior_binding_epoch,
2659 sequence_recovery,
2660 order_recovery,
2661 ) {
2662 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2663 }
2664 let finalizes_pending = appended_records.len() == 2;
2665 let pending_terminal_matches = self.pending_fenced_terminal_matches(appended_records);
2666 let candidates_empty = self.sequence.immutable_candidates.is_empty()
2667 && self.order.immutable_candidates.is_empty();
2668 if !(candidates_empty || finalizes_pending && pending_terminal_matches) {
2669 let condition = self.precedence_condition();
2670 return Err(precedence(self, condition));
2671 }
2672 if !Self::fenced_records_match(
2673 participant,
2674 sequence_recovery,
2675 order_recovery,
2676 appended_records,
2677 pending_terminal_matches,
2678 ) {
2679 return Err(Box::new((
2680 self,
2681 LiveFrontierTransitionError::RecordPosition,
2682 )));
2683 }
2684 if !self.has_recovery_marker(participant) {
2685 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2686 }
2687 if self.has_duplicate_appended_record(appended_records) {
2688 return Err(Box::new((
2689 self,
2690 LiveFrontierTransitionError::RecordPosition,
2691 )));
2692 }
2693 let Some((sequence_ledger, order_ledger)) =
2694 apply_fenced_ledgers(self.sequence.ledger, self.order.ledger, finalizes_pending)
2695 else {
2696 return Err(Box::new((
2697 self,
2698 LiveFrontierTransitionError::ResultingFrontier,
2699 )));
2700 };
2701 let mut active = self.active_identities.participants().to_vec();
2702 let Some(current) = active
2703 .iter_mut()
2704 .find(|current| current.participant_index() == participant.participant_index())
2705 else {
2706 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2707 };
2708 *current = participant;
2709 let Ok(active) = ActiveIdentityRanks::try_new(
2710 active,
2711 sequence_ledger.high_watermark(),
2712 self.identity_slot_limit,
2713 ) else {
2714 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2715 };
2716 let Ok((sequence, order)) =
2717 rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger)
2718 else {
2719 return Err(Box::new((
2720 self,
2721 LiveFrontierTransitionError::ResultingFrontier,
2722 )));
2723 };
2724 let mut resulting = self;
2725 resulting.active_identities = active;
2726 resulting
2727 .retained_records
2728 .extend_from_slice(appended_records);
2729 resulting
2730 .retained_records
2731 .sort_unstable_by_key(|record| record.delivery_seq);
2732 resulting
2733 .marker_records
2734 .retain(|record| record.delivery_seq != participant.cursor());
2735 resulting.sequence = sequence;
2736 resulting.order = order;
2737 Ok(resulting)
2738 }
2739
2740 fn fenced_recovery_authority_matches(
2741 current: FrontierParticipant,
2742 participant: FrontierParticipant,
2743 prior_binding_epoch: BindingEpoch,
2744 sequence_recovery: RecoverySequenceBlock,
2745 order_recovery: RecoveryOrderBlock,
2746 ) -> bool {
2747 let participant_matches = sequence_recovery.participant_index()
2748 == participant.participant_index()
2749 && order_recovery.participant_index() == participant.participant_index();
2750 let marker_matches = sequence_recovery.marker_delivery_seq() == participant.cursor()
2751 && order_recovery.marker_delivery_seq() == participant.cursor();
2752 let prior_epoch_matches = sequence_recovery.recovered_binding_epoch()
2753 == prior_binding_epoch
2754 && order_recovery.recovered_binding_epoch() == prior_binding_epoch;
2755 participant_matches
2756 && marker_matches
2757 && prior_epoch_matches
2758 && current.binding() == FrontierBinding::Detached(prior_binding_epoch)
2759 && current.cursor() <= participant.cursor()
2760 && matches!(participant.binding(), FrontierBinding::Bound(_))
2761 }
2762
2763 fn fenced_records_match(
2764 participant: FrontierParticipant,
2765 sequence_recovery: RecoverySequenceBlock,
2766 order_recovery: RecoveryOrderBlock,
2767 appended_records: &[RetainedCausalRecord],
2768 pending_terminal_matches: bool,
2769 ) -> bool {
2770 let Some(attached) = appended_records.last().copied() else {
2771 return false;
2772 };
2773 let FrontierBinding::Bound(recovered_binding_epoch) = participant.binding() else {
2774 return false;
2775 };
2776 let attached_matches = attached.delivery_seq == sequence_recovery.recovery_attach_seq()
2777 && attached.admission_order.transaction_order()
2778 == order_recovery.recovery_operation_order()
2779 && attached.kind
2780 == (RetainedCausalRecordKind::AttachLifecycle {
2781 participant_index: participant.participant_index(),
2782 binding_epoch: recovered_binding_epoch,
2783 });
2784 if !attached_matches {
2785 return false;
2786 }
2787 let prefix = &appended_records[..appended_records.len() - 1];
2788 match (
2789 sequence_recovery.terminal(),
2790 order_recovery.active_binding(),
2791 prefix,
2792 ) {
2793 (None, None, []) => true,
2794 (None, None, [_]) => pending_terminal_matches,
2795 (Some(sequence_terminal), Some(order_terminal), [terminal]) => {
2796 sequence_terminal.owner == order_terminal.owner
2797 && terminal.delivery_seq == sequence_terminal.delivery_seq
2798 && terminal.admission_order.transaction_order()
2799 == order_terminal.transaction_order
2800 && terminal.kind
2801 == RetainedCausalRecordKind::BindingTerminal(sequence_terminal.owner)
2802 }
2803 _ => false,
2804 }
2805 }
2806
2807 fn pending_fenced_terminal_matches(&self, appended_records: &[RetainedCausalRecord]) -> bool {
2808 let [terminal, _] = appended_records else {
2809 return false;
2810 };
2811 let RetainedCausalRecordKind::BindingTerminal(owner) = terminal.kind else {
2812 return false;
2813 };
2814 let sequence_matches = self.sequence.immutable_candidates.iter().any(|candidate| {
2815 matches!(
2816 candidate,
2817 ImmutableSequenceCandidate::BindingTerminal {
2818 delivery_seq,
2819 admission_order,
2820 owner: candidate_owner,
2821 } if *delivery_seq == terminal.delivery_seq
2822 && *admission_order == terminal.admission_order
2823 && *candidate_owner == owner
2824 )
2825 });
2826 let order_matches = self.order.immutable_candidates.iter().any(|candidate| {
2827 candidate.transaction_order == terminal.admission_order.transaction_order()
2828 && candidate.candidate_keys.as_slice() == [terminal.admission_order]
2829 });
2830 sequence_matches && order_matches
2831 }
2832
2833 fn has_recovery_marker(&self, participant: FrontierParticipant) -> bool {
2834 self.retained_records.iter().any(|record| {
2835 record.delivery_seq == participant.cursor()
2836 && matches!(
2837 record.kind,
2838 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
2839 if participant_index == participant.participant_index()
2840 )
2841 })
2842 }
2843
2844 fn has_duplicate_appended_record(&self, appended_records: &[RetainedCausalRecord]) -> bool {
2845 appended_records.iter().any(|row| {
2846 self.retained_records
2847 .iter()
2848 .any(|retained| retained.delivery_seq == row.delivery_seq)
2849 })
2850 }
2851
2852 pub(in crate::lifecycle) fn apply_live_identity(
2855 mut self,
2856 participant: FrontierParticipant,
2857 ) -> Result<Self, Box<(Self, LiveFrontierTransitionError)>> {
2858 let Some(current) = self
2859 .active_identities
2860 .participants
2861 .iter_mut()
2862 .find(|current| current.participant_index == participant.participant_index)
2863 else {
2864 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2865 };
2866 if participant.cursor < current.cursor
2867 || participant.cursor > self.sequence.ledger.high_watermark()
2868 {
2869 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
2870 }
2871 *current = participant;
2872 Ok(self)
2873 }
2874
2875 pub fn project_ordinary_record(
2891 self,
2892 input: OrdinaryRecordProjectionInput,
2893 ) -> Result<OrdinaryRecordProjectionDecision, Box<OrdinaryRecordProjectionFailure>> {
2894 let (
2895 request,
2896 receiving_binding_epoch,
2897 encoded_record_charge,
2898 retained_charges,
2899 observer_progress,
2900 closure_accounting,
2901 limits,
2902 ) = input.as_parts();
2903 if request.conversation_id != self.conversation_id {
2904 return Err(projection_failure(
2905 self,
2906 input,
2907 OrdinaryProjectionError::Conversation,
2908 ));
2909 }
2910 let unaccepted_marker_anchors = ordinary_unaccepted_marker_anchors(&self);
2911 let kernel = match project_ordinary_fixed_point(&OrdinaryProjectionFacts {
2912 request: request.clone(),
2913 receiving_binding_epoch,
2914 encoded_record_charge,
2915 retained_records: &self.retained_records,
2916 retained_charges,
2917 active_marker_credit_records: &self.marker_records,
2918 unaccepted_marker_anchors: &unaccepted_marker_anchors,
2919 active_identities: self.active_identities.participants(),
2920 identity_slot_limit: self.identity_slot_limit,
2921 current_floor: self.retained_floor,
2922 observer_progress,
2923 order_ledger: self.order.ledger,
2924 sequence_ledger: self.sequence.ledger,
2925 immutable_candidates: &self.sequence.immutable_candidates,
2926 closure_accounting,
2927 remaining_recovery_claim: closure_accounting.edge_k_remaining(),
2928 limits,
2929 }) {
2930 Ok(value) => value,
2931 Err(error) => return Err(projection_failure(self, input, error)),
2932 };
2933 match kernel {
2934 OrdinaryProjectionKernelDecision::DrainFirst(prefix) => Ok(
2935 OrdinaryRecordProjectionDecision::DrainFirst(Box::new(OrdinaryRecordDrainFirst {
2936 frontiers: self,
2937 input,
2938 candidate: prefix.candidate(),
2939 })),
2940 ),
2941 OrdinaryProjectionKernelDecision::Projected(projected) => {
2942 let observer_floor = match check_observer_floor(
2943 ObserverCheckedOperation::RecordAdmission(request.clone()),
2944 observer_progress,
2945 projected.floor().resulting_floor,
2946 ) {
2947 ObserverFloorDecision::Eligible(permit) => permit,
2948 ObserverFloorDecision::Respond(_) => {
2949 return Err(projection_failure(
2950 self,
2951 input,
2952 OrdinaryProjectionError::ObserverSelectorInvariant,
2953 ));
2954 }
2955 };
2956 let closure = match check_remaining_closure(
2957 &ClosureCheckedEnvelope::RecordAdmission(request.clone()),
2958 closure_accounting,
2959 false,
2960 0,
2961 projected.required_capacity(),
2962 ) {
2963 RemainingClosureDecision::Eligible(permit) => *permit,
2964 RemainingClosureDecision::Respond(_) => {
2965 return Err(projection_failure(
2966 self,
2967 input,
2968 OrdinaryProjectionError::ClosureSelectorInvariant,
2969 ));
2970 }
2971 };
2972 match self.apply_ordinary_projection(*projected, observer_floor, closure) {
2973 Ok(projected) => Ok(OrdinaryRecordProjectionDecision::Projected(Box::new(
2974 projected,
2975 ))),
2976 Err(failure) => {
2977 let (frontiers, error) = *failure;
2978 Err(projection_failure(frontiers, input, error))
2979 }
2980 }
2981 }
2982 }
2983 }
2984
2985 fn apply_ordinary_projection(
2986 mut self,
2987 projected: OrdinaryFixedPointPlan,
2988 observer_floor: super::ObserverFloorPermit,
2989 closure: super::RemainingClosurePermit,
2990 ) -> Result<ProjectedOrdinaryRecord, Box<(Self, OrdinaryProjectionError)>> {
2991 let Ok(marker_count) = u64::try_from(projected.marker_candidates().len()) else {
2992 return Err(Box::new((
2993 self,
2994 OrdinaryProjectionError::SequenceRelocation,
2995 )));
2996 };
2997 let Some(sequence_delta) = marker_count.checked_add(1) else {
2998 return Err(Box::new((
2999 self,
3000 OrdinaryProjectionError::SequenceRelocation,
3001 )));
3002 };
3003 if let Err(error) = preflight_ordinary_sequence_owners(&self.sequence, sequence_delta) {
3004 return Err(Box::new((self, error)));
3005 }
3006 if let Err(error) = preflight_ordinary_order_owners(&self.order) {
3007 return Err(Box::new((self, error)));
3008 }
3009 let (
3010 floor,
3011 retained_charge,
3012 baseline,
3013 accounting,
3014 required_capacity,
3015 order,
3016 sequence,
3017 caller_record,
3018 caller_charge,
3019 retained_records,
3020 retained_charges,
3021 new_marker_candidates,
3022 ) = projected.into_parts();
3023
3024 let prior_sequence_ledger = self.sequence.ledger;
3025 let prior_order_ledger = self.order.ledger;
3026 relay_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
3027 relay_ordinary_order_owners(&mut self.order);
3028
3029 self.sequence.ledger = sequence.resulting();
3030 self.sequence.immutable_candidates.extend(
3031 new_marker_candidates
3032 .iter()
3033 .copied()
3034 .map(ImmutableSequenceCandidate::Marker),
3035 );
3036 self.order.ledger = order.resulting();
3037 if !new_marker_candidates.is_empty() {
3038 self.order
3039 .immutable_candidates
3040 .push(ImmutableOrderCandidateMajor {
3041 transaction_order: order.major(),
3042 candidate_keys: new_marker_candidates
3043 .iter()
3044 .map(|candidate| candidate.admission_order)
3045 .collect(),
3046 });
3047 }
3048 if validate_cross_counter(&self.sequence, &self.order).is_err() {
3049 self.sequence.ledger = prior_sequence_ledger;
3050 self.sequence.immutable_candidates.clear();
3051 self.order.ledger = prior_order_ledger;
3052 self.order.immutable_candidates.clear();
3053 rollback_ordinary_sequence_owners(&mut self.sequence, sequence_delta);
3054 rollback_ordinary_order_owners(&mut self.order);
3055 return Err(Box::new((
3056 self,
3057 OrdinaryProjectionError::SequenceRelocation,
3058 )));
3059 }
3060 self.retained_floor = floor.resulting_floor;
3061 self.retained_records = retained_records;
3062 self.marker_records
3063 .retain(|record| u128::from(record.delivery_seq) >= floor.resulting_floor);
3064
3065 Ok(ProjectedOrdinaryRecord {
3066 frontiers: self,
3067 floor,
3068 retained_charge,
3069 baseline,
3070 accounting,
3071 required_capacity,
3072 order,
3073 sequence,
3074 observer_floor,
3075 closure,
3076 caller_record,
3077 caller_charge,
3078 retained_charges,
3079 new_marker_candidates,
3080 })
3081 }
3082
3083 pub fn planned_settled_leave_admission_order<F>(
3095 &self,
3096 member: &LiveMember<F>,
3097 binding_state: BindingState,
3098 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
3099 let (participant_id, ended_binding_epoch) =
3100 validate_settled_leave_prestate(self, member, binding_state)?;
3101 let selection = select_leave_order(&self.order, participant_id, ended_binding_epoch, None)?;
3102 Ok(super::AdmissionOrder::new(
3103 selection.selected_major,
3104 CandidatePhase::MembershipExit,
3105 participant_id,
3106 ))
3107 }
3108
3109 pub fn prepare_settled_leave_authority<F>(
3122 mut self,
3123 member: &LiveMember<F>,
3124 binding_state: BindingState,
3125 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
3126 let (participant_id, ended_binding_epoch) =
3127 validate_settled_leave_prestate(&self, member, binding_state)?;
3128 let left_transaction_order =
3129 consume_leave_order_lane(&mut self.order, participant_id, ended_binding_epoch, None)?;
3130 Ok(PreparedLeaveAuthority::settled(
3131 self,
3132 member.conversation_id(),
3133 participant_id,
3134 ended_binding_epoch,
3135 left_transaction_order,
3136 ))
3137 }
3138
3139 pub fn planned_pending_leave_admission_order<F>(
3147 &self,
3148 member: &LiveMember<F>,
3149 pending: PendingFinalization,
3150 ) -> Result<super::AdmissionOrder, PrepareLeaveAuthorityError> {
3151 let (participant_id, expected_order) =
3152 validate_pending_leave_prestate(self, member, pending)?;
3153 let selection =
3154 select_leave_order(&self.order, participant_id, None, Some(expected_order))?;
3155 Ok(super::AdmissionOrder::new(
3156 selection.selected_major,
3157 CandidatePhase::MembershipExit,
3158 participant_id,
3159 ))
3160 }
3161
3162 pub fn prepare_pending_leave_authority<F>(
3176 mut self,
3177 member: &LiveMember<F>,
3178 pending: PendingFinalization,
3179 ) -> Result<PreparedLeaveAuthority, PrepareLeaveAuthorityError> {
3180 let (participant_id, expected_order) =
3181 validate_pending_leave_prestate(&self, member, pending)?;
3182 let left_transaction_order =
3183 consume_leave_order_lane(&mut self.order, participant_id, None, Some(expected_order))?;
3184 Ok(PreparedLeaveAuthority::pending(
3185 self,
3186 member.conversation_id(),
3187 participant_id,
3188 pending.binding_epoch(),
3189 expected_order,
3190 left_transaction_order,
3191 ))
3192 }
3193
3194 #[allow(
3203 clippy::too_many_lines,
3204 reason = "the atomic Leave relay keeps membership, both ledgers, products, recovery, and retained rows visibly in one checked transition"
3205 )]
3206 pub(super) fn finish_leave_claims(
3207 mut self,
3208 participant_id: ParticipantId,
3209 ended_binding_epoch: Option<BindingEpoch>,
3210 committed_terminal: Option<CommittedBindingTerminal>,
3211 left_delivery_seq: DeliverySeq,
3212 left_transaction_order: TransactionOrder,
3213 ) -> Result<Self, LeaveCommitError> {
3214 let prior_high = self.sequence.ledger.high_watermark();
3215 let first_appended = prior_high
3216 .checked_add(1)
3217 .ok_or(LeaveCommitError::SequenceAuthority)?;
3218 let expected_left = if committed_terminal.is_some() {
3219 first_appended
3220 .checked_add(1)
3221 .ok_or(LeaveCommitError::SequenceAuthority)?
3222 } else {
3223 first_appended
3224 };
3225 if left_delivery_seq != expected_left {
3226 return Err(LeaveCommitError::SequenceAuthority);
3227 }
3228
3229 match (
3230 committed_terminal,
3231 self.sequence.immutable_candidates.as_slice(),
3232 ) {
3233 (
3234 Some(terminal),
3235 [
3236 ImmutableSequenceCandidate::BindingTerminal {
3237 delivery_seq,
3238 admission_order,
3239 owner,
3240 },
3241 ],
3242 ) if *delivery_seq == first_appended
3243 && terminal.delivery_seq() == first_appended
3244 && *admission_order == terminal.admission_order()
3245 && owner.participant_index == participant_id
3246 && owner.binding_epoch == terminal.binding_epoch() => {}
3247 (None, []) => {}
3248 (Some(_) | None, _) => return Err(LeaveCommitError::SequenceAuthority),
3249 }
3250
3251 let Some(active_index) = self
3252 .active_identities
3253 .participants
3254 .iter()
3255 .position(|participant| participant.participant_index == participant_id)
3256 else {
3257 return Err(LeaveCommitError::ResultingFrontier);
3258 };
3259 self.active_identities.participants.remove(active_index);
3260 self.marker_records.retain(|record| {
3261 !matches!(
3262 record.kind,
3263 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
3264 if participant_index == participant_id
3265 )
3266 });
3267 let resulting_live = usize_to_u64(self.active_identities.participants.len());
3268
3269 let mut exit_consumed = false;
3270 let mut terminal_consumed = ended_binding_epoch.is_none();
3271 let mut units = Vec::new();
3272 for claim in self.sequence.movable_claims.iter().copied() {
3273 match claim.owner {
3274 SequenceDirectOwner::MembershipExit {
3275 participant_index: owner,
3276 } if owner == participant_id => {
3277 if exit_consumed {
3278 return Err(LeaveCommitError::ResultingFrontier);
3279 }
3280 exit_consumed = true;
3281 }
3282 SequenceDirectOwner::BindingTerminal(owner)
3283 if owner.participant_index == participant_id
3284 && Some(owner.binding_epoch) == ended_binding_epoch =>
3285 {
3286 if terminal_consumed {
3287 return Err(LeaveCommitError::ResultingFrontier);
3288 }
3289 terminal_consumed = true;
3290 }
3291 SequenceDirectOwner::MembershipExit { .. }
3292 | SequenceDirectOwner::BindingTerminal(_) => {
3293 units.push(LeaveSequenceUnit::Direct(claim));
3294 }
3295 }
3296 }
3297 if !exit_consumed || !terminal_consumed {
3298 return Err(LeaveCommitError::ResultingFrontier);
3299 }
3300
3301 let recovery_owned = self
3302 .sequence
3303 .recovery
3304 .is_some_and(|block| block.participant_index == participant_id);
3305 for range in self.sequence.products.live_times_terminal.iter().copied() {
3306 if range.terminal.participant_index != participant_id && resulting_live != 0 {
3307 units.push(LeaveSequenceUnit::TerminalProduct(TerminalProductRange {
3308 start: range.start,
3309 length: resulting_live,
3310 terminal: range.terminal,
3311 }));
3312 }
3313 }
3314 if let Some(range) = self.sequence.products.live_times_replacement_terminal
3315 && !recovery_owned
3316 && resulting_live != 0
3317 {
3318 units.push(LeaveSequenceUnit::ReplacementProduct(
3319 ReplacementTerminalProductRange {
3320 start: range.start,
3321 length: resulting_live,
3322 participant_index: range.participant_index,
3323 marker_delivery_seq: range.marker_delivery_seq,
3324 prior_binding_epoch: range.prior_binding_epoch,
3325 },
3326 ));
3327 }
3328 let resulting_other = resulting_live.saturating_sub(1);
3329 if resulting_other != 0 {
3330 for range in self.sequence.products.other_live_times_exit.iter().copied() {
3331 if range.exit_participant != participant_id {
3332 units.push(LeaveSequenceUnit::ExitProduct(ExitProductRange {
3333 start: range.start,
3334 length: resulting_other,
3335 exit_participant: range.exit_participant,
3336 }));
3337 }
3338 }
3339 }
3340 if let Some(recovery) = self.sequence.recovery
3341 && !recovery_owned
3342 {
3343 units.push(LeaveSequenceUnit::Recovery(recovery));
3344 }
3345 units.sort_by_key(|unit| unit.original_start());
3346
3347 let mut cursor = left_delivery_seq.checked_add(1);
3348 let mut movable_claims = Vec::new();
3349 let mut terminal_products = Vec::new();
3350 let mut replacement_product = None;
3351 let mut exit_products = Vec::new();
3352 let mut recovery = None;
3353 for unit in units {
3354 match unit {
3355 LeaveSequenceUnit::Direct(mut claim) => {
3356 claim.delivery_seq = allocate_leave_sequence_range(&mut cursor, 1)?;
3357 movable_claims.push(claim);
3358 }
3359 LeaveSequenceUnit::TerminalProduct(mut range) => {
3360 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3361 terminal_products.push(range);
3362 }
3363 LeaveSequenceUnit::ReplacementProduct(mut range) => {
3364 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3365 replacement_product = Some(range);
3366 }
3367 LeaveSequenceUnit::ExitProduct(mut range) => {
3368 range.start = allocate_leave_sequence_range(&mut cursor, range.length)?;
3369 exit_products.push(range);
3370 }
3371 LeaveSequenceUnit::Recovery(mut block) => {
3372 let length = 2 + u64::from(block.terminal.is_some());
3373 let start = allocate_leave_sequence_range(&mut cursor, length)?;
3374 if let Some(mut terminal) = block.terminal {
3375 terminal.delivery_seq = start;
3376 block.terminal = Some(terminal);
3377 block.recovery_attach_seq = start
3378 .checked_add(1)
3379 .ok_or(LeaveCommitError::ResultingFrontier)?;
3380 } else {
3381 block.recovery_attach_seq = start;
3382 }
3383 block.replacement_terminal_seq = block
3384 .recovery_attach_seq
3385 .checked_add(1)
3386 .ok_or(LeaveCommitError::ResultingFrontier)?;
3387 recovery = Some(block);
3388 }
3389 }
3390 }
3391 movable_claims.sort_by_key(|claim| claim.delivery_seq);
3392 terminal_products.sort_by_key(|range| range.start);
3393 exit_products.sort_by_key(|range| range.start);
3394 let terminal_count = usize_to_u64(
3395 movable_claims
3396 .iter()
3397 .filter(|claim| matches!(claim.owner, SequenceDirectOwner::BindingTerminal(_)))
3398 .count(),
3399 ) + u64::from(recovery.is_some_and(|block| block.terminal.is_some()));
3400 let recovery_reserve = if recovery.is_some() {
3401 RecoverySequenceReserve::DetachedCredentialRecovery
3402 } else {
3403 RecoverySequenceReserve::None
3404 };
3405 let ledger = SequenceLedger::try_new(
3406 left_delivery_seq,
3407 SequenceClaims::new(resulting_live, terminal_count, 0, recovery_reserve),
3408 )
3409 .map_err(|_| LeaveCommitError::ResultingFrontier)?;
3410 self.sequence = SequenceClaimFrontier {
3411 ledger,
3412 movable_claims,
3413 immutable_candidates: Vec::new(),
3414 products: SequenceProductRanges {
3415 live_times_terminal: terminal_products,
3416 live_times_replacement_terminal: replacement_product,
3417 other_live_times_exit: exit_products,
3418 },
3419 recovery,
3420 };
3421
3422 if let Some(terminal) = committed_terminal {
3423 self.retained_records.push(RetainedCausalRecord {
3424 delivery_seq: terminal.delivery_seq(),
3425 admission_order: terminal.admission_order(),
3426 kind: RetainedCausalRecordKind::BindingTerminal(BindingTerminalOwner {
3427 participant_index: participant_id,
3428 binding_epoch: terminal.binding_epoch(),
3429 }),
3430 });
3431 }
3432 self.retained_records.push(RetainedCausalRecord {
3433 delivery_seq: left_delivery_seq,
3434 admission_order: super::AdmissionOrder::new(
3435 left_transaction_order,
3436 CandidatePhase::MembershipExit,
3437 participant_id,
3438 ),
3439 kind: RetainedCausalRecordKind::MembershipExit {
3440 participant_index: participant_id,
3441 },
3442 });
3443 self.retained_records
3444 .sort_by_key(|record| record.delivery_seq);
3445
3446 let order_claims = self.order.ledger.claims();
3447 if order_claims.membership_exits() != resulting_live
3448 || self.order.recovery.is_some() != self.sequence.recovery.is_some()
3449 || validate_cross_counter(&self.sequence, &self.order).is_err()
3450 {
3451 return Err(LeaveCommitError::ResultingFrontier);
3452 }
3453 Ok(self)
3454 }
3455
3456 fn marker_candidate(&self, delivery_seq: DeliverySeq) -> Option<ValidatedMarkerCandidate> {
3457 self.sequence
3458 .immutable_candidates
3459 .iter()
3460 .find_map(|candidate| match candidate {
3461 ImmutableSequenceCandidate::Marker(candidate)
3462 if candidate.delivery_seq == delivery_seq =>
3463 {
3464 Some(ValidatedMarkerCandidate {
3465 conversation_id: self.conversation_id,
3466 candidate: *candidate,
3467 seal: MarkerAuthoritySeal::Validated,
3468 })
3469 }
3470 _ => None,
3471 })
3472 }
3473
3474 pub(super) fn drain_next_marker_core(
3483 mut self,
3484 ) -> Result<MarkerDrainCore, MarkerDrainCoreError> {
3485 let Some(first) = self.sequence.immutable_candidates.first().copied() else {
3486 return Err(MarkerDrainCoreError::NoCandidate);
3487 };
3488 let ImmutableSequenceCandidate::Marker(marker) = first else {
3489 return Err(MarkerDrainCoreError::BindingTerminalFirst);
3490 };
3491 let expected_sequence = self
3492 .sequence
3493 .ledger
3494 .high_watermark()
3495 .checked_add(1)
3496 .ok_or(MarkerDrainCoreError::SequenceNotNext)?;
3497 if marker.delivery_seq != expected_sequence {
3498 return Err(MarkerDrainCoreError::SequenceNotNext);
3499 }
3500 if order_is_above_high(
3501 marker.admission_order.transaction_order(),
3502 self.order.ledger.high(),
3503 ) {
3504 return Err(MarkerDrainCoreError::CausalMajorNotAllocated);
3505 }
3506 let candidate = self
3507 .marker_candidate(marker.delivery_seq)
3508 .ok_or(MarkerDrainCoreError::NoCandidate)?;
3509
3510 let key = marker.admission_order;
3511 let Some(group_index) = self
3512 .order
3513 .immutable_candidates
3514 .iter()
3515 .position(|group| group.candidate_keys.contains(&key))
3516 else {
3517 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3518 };
3519 let group = &mut self.order.immutable_candidates[group_index];
3520 let Ok(key_index) = group.candidate_keys.binary_search(&key) else {
3521 return Err(MarkerDrainCoreError::MissingOrderCandidate);
3522 };
3523 group.candidate_keys.remove(key_index);
3524 if group.candidate_keys.is_empty() {
3525 self.order.immutable_candidates.remove(group_index);
3526 }
3527 self.sequence.immutable_candidates.remove(0);
3528
3529 let claims = self.sequence.ledger.claims();
3530 let markers = claims
3531 .markers()
3532 .checked_sub(1)
3533 .ok_or(MarkerDrainCoreError::ResultingLedger)?;
3534 self.sequence.ledger = SequenceLedger::try_new(
3535 expected_sequence,
3536 super::SequenceClaims::new(
3537 claims.live_members(),
3538 claims.binding_terminals(),
3539 markers,
3540 claims.recovery(),
3541 ),
3542 )
3543 .map_err(|_| MarkerDrainCoreError::ResultingLedger)?;
3544
3545 let record = RetainedCausalRecord {
3546 delivery_seq: marker.delivery_seq,
3547 admission_order: marker.admission_order,
3548 kind: RetainedCausalRecordKind::CompactionMarker {
3549 participant_index: marker.admission_order.participant_index(),
3550 provenance: marker.provenance,
3551 },
3552 };
3553 self.marker_records.push(record);
3554 self.retained_records.push(record);
3555 Ok(MarkerDrainCore {
3556 candidate,
3557 record: ValidatedMarkerRecord {
3558 conversation_id: self.conversation_id,
3559 record,
3560 provenance: marker.provenance,
3561 target_binding: marker.target_binding,
3562 occurrence: MarkerRecordOccurrence::Undelivered,
3563 seal: MarkerAuthoritySeal::Validated,
3564 },
3565 frontiers: self,
3566 })
3567 }
3568
3569 pub(in crate::lifecycle) fn drain_first_binding_terminal(
3583 mut self,
3584 expected_owner: BindingTerminalOwner,
3585 expected_order: super::AdmissionOrder,
3586 ) -> Result<(Self, RetainedCausalRecord), Box<(Self, LiveFrontierTransitionError)>> {
3587 let (delivery_seq, admission_order, owner) =
3588 match self.validate_first_terminal_candidate(expected_owner, expected_order) {
3589 Ok(candidate) => candidate,
3590 Err(error) => return Err(Box::new((self, error))),
3591 };
3592 let claims = self.sequence.ledger.claims();
3593 let Some(sequence_ledger) =
3594 claims
3595 .binding_terminals()
3596 .checked_sub(1)
3597 .and_then(|binding_terminals| {
3598 SequenceLedger::try_new(
3599 delivery_seq,
3600 SequenceClaims::new(
3601 claims.live_members(),
3602 binding_terminals,
3603 claims.markers(),
3604 claims.recovery(),
3605 ),
3606 )
3607 .ok()
3608 })
3609 else {
3610 return Err(Box::new((
3611 self,
3612 LiveFrontierTransitionError::ResultingFrontier,
3613 )));
3614 };
3615 let order_ledger = self.order.ledger;
3616 let Ok(active) = ActiveIdentityRanks::try_new(
3617 self.active_identities.participants().to_vec(),
3618 sequence_ledger.high_watermark(),
3619 self.identity_slot_limit,
3620 ) else {
3621 return Err(Box::new((self, LiveFrontierTransitionError::Authority)));
3622 };
3623 let (sequence, order) =
3624 match rebuild_unreserved_frontiers(&active, sequence_ledger, order_ledger) {
3625 Ok(frontiers) => frontiers,
3626 Err(error) => return Err(Box::new((self, error))),
3627 };
3628 let record = RetainedCausalRecord {
3629 delivery_seq,
3630 admission_order,
3631 kind: RetainedCausalRecordKind::BindingTerminal(owner),
3632 };
3633 self.active_identities = active;
3634 self.sequence = sequence;
3635 self.order = order;
3636 self.retained_records.push(record);
3637 Ok((self, record))
3638 }
3639
3640 fn validate_first_terminal_candidate(
3645 &self,
3646 expected_owner: BindingTerminalOwner,
3647 expected_order: super::AdmissionOrder,
3648 ) -> Result<
3649 (DeliverySeq, super::AdmissionOrder, BindingTerminalOwner),
3650 LiveFrontierTransitionError,
3651 > {
3652 if self.sequence.recovery.is_some() || self.order.recovery.is_some() {
3653 return Err(LiveFrontierTransitionError::Precedence(
3654 PrecedenceCondition::FencedRecovery,
3655 ));
3656 }
3657 let [first] = self.sequence.immutable_candidates.as_slice() else {
3658 return Err(LiveFrontierTransitionError::Precedence(
3659 self.precedence_condition(),
3660 ));
3661 };
3662 let ImmutableSequenceCandidate::BindingTerminal {
3663 delivery_seq,
3664 admission_order,
3665 owner,
3666 } = *first
3667 else {
3668 return Err(LiveFrontierTransitionError::Precedence(
3669 self.precedence_condition(),
3670 ));
3671 };
3672 if owner != expected_owner || admission_order != expected_order {
3673 return Err(LiveFrontierTransitionError::Authority);
3674 }
3675 let Some(expected_sequence) = self.sequence.ledger.high_watermark().checked_add(1) else {
3676 return Err(LiveFrontierTransitionError::Exhausted);
3677 };
3678 if delivery_seq != expected_sequence
3679 || order_is_above_high(
3680 admission_order.transaction_order(),
3681 self.order.ledger.high(),
3682 )
3683 {
3684 return Err(LiveFrontierTransitionError::RecordPosition);
3685 }
3686 let [group] = self.order.immutable_candidates.as_slice() else {
3687 return Err(LiveFrontierTransitionError::RecordPosition);
3688 };
3689 if group.candidate_keys.as_slice() != [admission_order] {
3690 return Err(LiveFrontierTransitionError::RecordPosition);
3691 }
3692 let detached_matches = self
3693 .active_identities
3694 .participants()
3695 .iter()
3696 .any(|participant| {
3697 participant.participant_index == owner.participant_index
3698 && participant.binding == FrontierBinding::Detached(owner.binding_epoch)
3699 });
3700 if !detached_matches {
3701 return Err(LiveFrontierTransitionError::Authority);
3702 }
3703 Ok((delivery_seq, admission_order, owner))
3704 }
3705}
3706
3707fn apply_fenced_ledgers(
3708 sequence: SequenceLedger,
3709 order: OrderLedger,
3710 finalizes_pending: bool,
3711) -> Option<(SequenceLedger, OrderLedger)> {
3712 let sequence = if finalizes_pending {
3713 sequence.apply_fenced_recovery_finalizing_pending()
3714 } else {
3715 sequence.apply_fenced_recovery()
3716 }
3717 .ok()?;
3718 let order = if finalizes_pending {
3719 order.apply_fenced_recovery_finalizing_pending()
3720 } else {
3721 order.apply_fenced_recovery()
3722 }
3723 .ok()?;
3724 Some((sequence, order))
3725}
3726
3727fn rebuild_unreserved_frontiers(
3728 active: &ActiveIdentityRanks,
3729 sequence_ledger: SequenceLedger,
3730 order_ledger: OrderLedger,
3731) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
3732 let terminal_owners: Vec<_> = active
3733 .participants()
3734 .iter()
3735 .filter_map(|participant| match participant.binding() {
3736 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
3737 participant_index: participant.participant_index(),
3738 binding_epoch,
3739 }),
3740 FrontierBinding::Detached(_) => None,
3741 })
3742 .collect();
3743 let live_count = active.len();
3744 let terminal_count =
3745 u64::try_from(terminal_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3746 let sequence_claims = sequence_ledger.claims();
3747 let order_claims = order_ledger.claims();
3748 if sequence_claims.live_members() != live_count
3749 || sequence_claims.binding_terminals() != terminal_count
3750 || sequence_claims.markers() != 0
3751 || sequence_claims.recovery() != RecoverySequenceReserve::None
3752 || order_claims.active_binding_terminals() != terminal_count
3753 || order_claims.membership_exits() != live_count
3754 || order_claims.recovery_operation()
3755 || order_claims.recovery_replacement_terminal()
3756 {
3757 return Err(LiveFrontierTransitionError::ResultingFrontier);
3758 }
3759
3760 let sequence = rebuild_unreserved_sequence(active, &terminal_owners, sequence_ledger)?;
3761 let order = rebuild_unreserved_order(active, &terminal_owners, order_ledger)?;
3762 validate_cross_counter(&sequence, &order)
3763 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
3764 Ok((sequence, order))
3765}
3766
3767fn rebuild_unreserved_sequence(
3768 active: &ActiveIdentityRanks,
3769 terminal_owners: &[BindingTerminalOwner],
3770 sequence_ledger: SequenceLedger,
3771) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
3772 let live_count = active.len();
3773 let mut sequence_cursor = sequence_ledger
3774 .high_watermark()
3775 .checked_add(1)
3776 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3777 let mut movable_sequence = Vec::new();
3778 for terminal in terminal_owners {
3779 movable_sequence.push(MovableSequenceClaim {
3780 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3781 owner: SequenceDirectOwner::BindingTerminal(*terminal),
3782 });
3783 }
3784 for participant in active.participants() {
3785 movable_sequence.push(MovableSequenceClaim {
3786 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3787 owner: SequenceDirectOwner::MembershipExit {
3788 participant_index: participant.participant_index(),
3789 },
3790 });
3791 }
3792 let mut terminal_products = Vec::new();
3793 for terminal in terminal_owners {
3794 terminal_products.push(TerminalProductRange {
3795 start: take_live_sequence(&mut sequence_cursor, live_count)?,
3796 length: live_count,
3797 terminal: *terminal,
3798 });
3799 }
3800 let exit_product_length = live_count.saturating_sub(1);
3801 let mut exit_products = Vec::new();
3802 for participant in active.participants() {
3803 exit_products.push(ExitProductRange {
3804 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
3805 length: exit_product_length,
3806 exit_participant: participant.participant_index(),
3807 });
3808 }
3809 let sequence_end = u128::from(sequence_ledger.high_watermark())
3810 .checked_add(sequence_ledger.required_reserve())
3811 .and_then(|value| value.checked_add(1))
3812 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3813 if u128::from(sequence_cursor) != sequence_end {
3814 return Err(LiveFrontierTransitionError::ResultingFrontier);
3815 }
3816 Ok(SequenceClaimFrontier {
3817 ledger: sequence_ledger,
3818 movable_claims: movable_sequence,
3819 immutable_candidates: Vec::new(),
3820 products: SequenceProductRanges {
3821 live_times_terminal: terminal_products,
3822 live_times_replacement_terminal: None,
3823 other_live_times_exit: exit_products,
3824 },
3825 recovery: None,
3826 })
3827}
3828
3829fn rebuild_unreserved_order(
3830 active: &ActiveIdentityRanks,
3831 terminal_owners: &[BindingTerminalOwner],
3832 order_ledger: OrderLedger,
3833) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
3834 let order_claims = order_ledger.claims();
3835 let order_start = order_frontier_start(order_ledger.high());
3836 let mut order_cursor =
3837 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3838 let mut movable_order = Vec::new();
3839 for terminal in terminal_owners.iter().copied() {
3840 movable_order.push(MovableOrderClaim {
3841 transaction_order: take_live_order(&mut order_cursor)?,
3842 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
3843 });
3844 }
3845 for participant in active.participants() {
3846 movable_order.push(MovableOrderClaim {
3847 transaction_order: take_live_order(&mut order_cursor)?,
3848 owner: OrderDirectOwner::MembershipExit {
3849 participant_index: participant.participant_index(),
3850 },
3851 });
3852 }
3853 if u128::from(order_cursor) != order_start + order_claims.total() {
3854 return Err(LiveFrontierTransitionError::ResultingFrontier);
3855 }
3856 Ok(OrderClaimFrontier {
3857 ledger: order_ledger,
3858 movable_claims: movable_order,
3859 immutable_candidates: Vec::new(),
3860 recovery: None,
3861 })
3862}
3863
3864fn rebuild_pending_terminal_frontiers(
3865 active: &ActiveIdentityRanks,
3866 pending_owner: BindingTerminalOwner,
3867 delivery_seq: DeliverySeq,
3868 admission_order: super::AdmissionOrder,
3869 sequence_ledger: SequenceLedger,
3870 order_ledger: OrderLedger,
3871) -> Result<(SequenceClaimFrontier, OrderClaimFrontier), LiveFrontierTransitionError> {
3872 if admission_order.candidate_phase() != CandidatePhase::BindingTerminal
3873 || admission_order.participant_index() != pending_owner.participant_index
3874 || admission_order.transaction_order()
3875 != match order_ledger.high() {
3876 OrderHigh::Allocated(high) => high,
3877 OrderHigh::Empty => return Err(LiveFrontierTransitionError::RecordPosition),
3878 }
3879 || sequence_ledger.high_watermark().checked_add(1) != Some(delivery_seq)
3880 {
3881 return Err(LiveFrontierTransitionError::RecordPosition);
3882 }
3883 let bound_owners: Vec<_> = active
3884 .participants()
3885 .iter()
3886 .filter_map(|participant| match participant.binding() {
3887 FrontierBinding::Bound(binding_epoch) => Some(BindingTerminalOwner {
3888 participant_index: participant.participant_index(),
3889 binding_epoch,
3890 }),
3891 FrontierBinding::Detached(_) => None,
3892 })
3893 .collect();
3894 let mut all_terminal_owners = bound_owners.clone();
3895 all_terminal_owners.push(pending_owner);
3896 all_terminal_owners.sort_unstable_by_key(|owner| owner.participant_index);
3897 let live_count = active.len();
3898 let bound_count =
3899 u64::try_from(bound_owners.len()).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3900 let terminal_count = u64::try_from(all_terminal_owners.len())
3901 .map_err(|_| LiveFrontierTransitionError::Exhausted)?;
3902 let sequence_claims = sequence_ledger.claims();
3903 let order_claims = order_ledger.claims();
3904 if sequence_claims.live_members() != live_count
3905 || sequence_claims.binding_terminals() != terminal_count
3906 || sequence_claims.markers() != 0
3907 || sequence_claims.recovery() != RecoverySequenceReserve::None
3908 || order_claims.active_binding_terminals() != bound_count
3909 || order_claims.membership_exits() != live_count
3910 || order_claims.recovery_operation()
3911 || order_claims.recovery_replacement_terminal()
3912 {
3913 return Err(LiveFrontierTransitionError::ResultingFrontier);
3914 }
3915
3916 let sequence = rebuild_pending_terminal_sequence(
3917 active,
3918 pending_owner,
3919 delivery_seq,
3920 admission_order,
3921 sequence_ledger,
3922 &bound_owners,
3923 &all_terminal_owners,
3924 )?;
3925 let order =
3926 rebuild_pending_terminal_order(active, admission_order, order_ledger, bound_owners)?;
3927 validate_cross_counter(&sequence, &order)
3928 .map_err(|_| LiveFrontierTransitionError::ResultingFrontier)?;
3929 Ok((sequence, order))
3930}
3931
3932fn rebuild_pending_terminal_sequence(
3933 active: &ActiveIdentityRanks,
3934 pending_owner: BindingTerminalOwner,
3935 delivery_seq: DeliverySeq,
3936 admission_order: super::AdmissionOrder,
3937 sequence_ledger: SequenceLedger,
3938 bound_owners: &[BindingTerminalOwner],
3939 all_terminal_owners: &[BindingTerminalOwner],
3940) -> Result<SequenceClaimFrontier, LiveFrontierTransitionError> {
3941 let live_count = active.len();
3942 let mut sequence_cursor = delivery_seq;
3943 let candidate_sequence = take_live_sequence(&mut sequence_cursor, 1)?;
3944 let mut movable_sequence = Vec::new();
3945 for terminal in bound_owners {
3946 movable_sequence.push(MovableSequenceClaim {
3947 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3948 owner: SequenceDirectOwner::BindingTerminal(*terminal),
3949 });
3950 }
3951 for participant in active.participants() {
3952 movable_sequence.push(MovableSequenceClaim {
3953 delivery_seq: take_live_sequence(&mut sequence_cursor, 1)?,
3954 owner: SequenceDirectOwner::MembershipExit {
3955 participant_index: participant.participant_index(),
3956 },
3957 });
3958 }
3959 let mut terminal_products = Vec::new();
3960 for terminal in all_terminal_owners {
3961 terminal_products.push(TerminalProductRange {
3962 start: take_live_sequence(&mut sequence_cursor, live_count)?,
3963 length: live_count,
3964 terminal: *terminal,
3965 });
3966 }
3967 let exit_product_length = live_count.saturating_sub(1);
3968 let mut exit_products = Vec::new();
3969 for participant in active.participants() {
3970 exit_products.push(ExitProductRange {
3971 start: take_live_sequence(&mut sequence_cursor, exit_product_length)?,
3972 length: exit_product_length,
3973 exit_participant: participant.participant_index(),
3974 });
3975 }
3976 let sequence_end = u128::from(sequence_ledger.high_watermark())
3977 .checked_add(sequence_ledger.required_reserve())
3978 .and_then(|value| value.checked_add(1))
3979 .ok_or(LiveFrontierTransitionError::Exhausted)?;
3980 if u128::from(sequence_cursor) != sequence_end {
3981 return Err(LiveFrontierTransitionError::ResultingFrontier);
3982 }
3983 Ok(SequenceClaimFrontier {
3984 ledger: sequence_ledger,
3985 movable_claims: movable_sequence,
3986 immutable_candidates: alloc::vec![ImmutableSequenceCandidate::BindingTerminal {
3987 delivery_seq: candidate_sequence,
3988 admission_order,
3989 owner: pending_owner,
3990 }],
3991 products: SequenceProductRanges {
3992 live_times_terminal: terminal_products,
3993 live_times_replacement_terminal: None,
3994 other_live_times_exit: exit_products,
3995 },
3996 recovery: None,
3997 })
3998}
3999
4000fn rebuild_pending_terminal_order(
4001 active: &ActiveIdentityRanks,
4002 admission_order: super::AdmissionOrder,
4003 order_ledger: OrderLedger,
4004 bound_owners: Vec<BindingTerminalOwner>,
4005) -> Result<OrderClaimFrontier, LiveFrontierTransitionError> {
4006 let order_claims = order_ledger.claims();
4007 let order_start = order_frontier_start(order_ledger.high());
4008 let mut order_cursor =
4009 u64::try_from(order_start).map_err(|_| LiveFrontierTransitionError::Exhausted)?;
4010 let mut movable_order = Vec::new();
4011 for terminal in bound_owners {
4012 movable_order.push(MovableOrderClaim {
4013 transaction_order: take_live_order(&mut order_cursor)?,
4014 owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
4015 });
4016 }
4017 for participant in active.participants() {
4018 movable_order.push(MovableOrderClaim {
4019 transaction_order: take_live_order(&mut order_cursor)?,
4020 owner: OrderDirectOwner::MembershipExit {
4021 participant_index: participant.participant_index(),
4022 },
4023 });
4024 }
4025 if u128::from(order_cursor) != order_start + order_claims.total() {
4026 return Err(LiveFrontierTransitionError::ResultingFrontier);
4027 }
4028 Ok(OrderClaimFrontier {
4029 ledger: order_ledger,
4030 movable_claims: movable_order,
4031 immutable_candidates: alloc::vec![ImmutableOrderCandidateMajor {
4032 transaction_order: admission_order.transaction_order(),
4033 candidate_keys: alloc::vec![admission_order],
4034 }],
4035 recovery: None,
4036 })
4037}
4038
4039fn take_live_sequence(
4040 cursor: &mut DeliverySeq,
4041 length: u64,
4042) -> Result<DeliverySeq, LiveFrontierTransitionError> {
4043 let start = *cursor;
4044 *cursor = cursor
4045 .checked_add(length)
4046 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4047 Ok(start)
4048}
4049
4050fn take_live_order(
4051 cursor: &mut TransactionOrder,
4052) -> Result<TransactionOrder, LiveFrontierTransitionError> {
4053 let value = *cursor;
4054 *cursor = cursor
4055 .checked_add(1)
4056 .ok_or(LiveFrontierTransitionError::Exhausted)?;
4057 Ok(value)
4058}
4059
4060fn ordinary_unaccepted_marker_anchors(frontiers: &ClaimFrontiers) -> Vec<DeliverySeq> {
4061 frontiers
4062 .marker_records
4063 .iter()
4064 .filter_map(|record| {
4065 let RetainedCausalRecordKind::CompactionMarker {
4066 participant_index, ..
4067 } = record.kind
4068 else {
4069 return None;
4070 };
4071 active_participant(&frontiers.active_identities, participant_index)
4072 .is_some_and(|participant| participant.cursor < record.delivery_seq)
4073 .then_some(record.delivery_seq)
4074 })
4075 .collect()
4076}
4077
4078fn projection_failure(
4079 frontiers: ClaimFrontiers,
4080 input: OrdinaryRecordProjectionInput,
4081 error: OrdinaryProjectionError,
4082) -> Box<OrdinaryRecordProjectionFailure> {
4083 Box::new(OrdinaryRecordProjectionFailure {
4084 frontiers,
4085 input,
4086 error,
4087 })
4088}
4089
4090fn preflight_ordinary_sequence_owners(
4091 sequence: &SequenceClaimFrontier,
4092 delta: u64,
4093) -> Result<(), OrdinaryProjectionError> {
4094 if !sequence.immutable_candidates.is_empty() {
4095 return Err(OrdinaryProjectionError::SequenceRelocation);
4096 }
4097 for claim in &sequence.movable_claims {
4098 claim
4099 .delivery_seq
4100 .checked_add(delta)
4101 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4102 }
4103 for range in &sequence.products.live_times_terminal {
4104 range
4105 .start
4106 .checked_add(delta)
4107 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4108 }
4109 if let Some(range) = &sequence.products.live_times_replacement_terminal {
4110 range
4111 .start
4112 .checked_add(delta)
4113 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4114 }
4115 for range in &sequence.products.other_live_times_exit {
4116 range
4117 .start
4118 .checked_add(delta)
4119 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4120 }
4121 if let Some(recovery) = &sequence.recovery {
4122 if let Some(terminal) = &recovery.terminal {
4123 terminal
4124 .delivery_seq
4125 .checked_add(delta)
4126 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4127 }
4128 recovery
4129 .recovery_attach_seq
4130 .checked_add(delta)
4131 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4132 recovery
4133 .replacement_terminal_seq
4134 .checked_add(delta)
4135 .ok_or(OrdinaryProjectionError::SequenceRelocation)?;
4136 }
4137 Ok(())
4138}
4139
4140fn relay_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
4141 for claim in &mut sequence.movable_claims {
4142 claim.delivery_seq = claim.delivery_seq.wrapping_add(delta);
4143 }
4144 for range in &mut sequence.products.live_times_terminal {
4145 range.start = range.start.wrapping_add(delta);
4146 }
4147 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
4148 range.start = range.start.wrapping_add(delta);
4149 }
4150 for range in &mut sequence.products.other_live_times_exit {
4151 range.start = range.start.wrapping_add(delta);
4152 }
4153 if let Some(recovery) = &mut sequence.recovery {
4154 if let Some(terminal) = &mut recovery.terminal {
4155 terminal.delivery_seq = terminal.delivery_seq.wrapping_add(delta);
4156 }
4157 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_add(delta);
4158 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_add(delta);
4159 }
4160}
4161
4162fn rollback_ordinary_sequence_owners(sequence: &mut SequenceClaimFrontier, delta: u64) {
4163 for claim in &mut sequence.movable_claims {
4164 claim.delivery_seq = claim.delivery_seq.wrapping_sub(delta);
4165 }
4166 for range in &mut sequence.products.live_times_terminal {
4167 range.start = range.start.wrapping_sub(delta);
4168 }
4169 if let Some(range) = &mut sequence.products.live_times_replacement_terminal {
4170 range.start = range.start.wrapping_sub(delta);
4171 }
4172 for range in &mut sequence.products.other_live_times_exit {
4173 range.start = range.start.wrapping_sub(delta);
4174 }
4175 if let Some(recovery) = &mut sequence.recovery {
4176 if let Some(terminal) = &mut recovery.terminal {
4177 terminal.delivery_seq = terminal.delivery_seq.wrapping_sub(delta);
4178 }
4179 recovery.recovery_attach_seq = recovery.recovery_attach_seq.wrapping_sub(delta);
4180 recovery.replacement_terminal_seq = recovery.replacement_terminal_seq.wrapping_sub(delta);
4181 }
4182}
4183
4184fn preflight_ordinary_order_owners(
4185 order: &OrderClaimFrontier,
4186) -> Result<(), OrdinaryProjectionError> {
4187 if !order.immutable_candidates.is_empty() {
4188 return Err(OrdinaryProjectionError::OrderRelocation);
4189 }
4190 for claim in &order.movable_claims {
4191 claim
4192 .transaction_order
4193 .checked_add(1)
4194 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4195 }
4196 if let Some(recovery) = &order.recovery {
4197 if let Some(active_binding) = &recovery.active_binding {
4198 active_binding
4199 .transaction_order
4200 .checked_add(1)
4201 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4202 }
4203 recovery
4204 .recovery_operation_order
4205 .checked_add(1)
4206 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4207 recovery
4208 .replacement_terminal_order
4209 .checked_add(1)
4210 .ok_or(OrdinaryProjectionError::OrderRelocation)?;
4211 }
4212 Ok(())
4213}
4214
4215fn relay_ordinary_order_owners(order: &mut OrderClaimFrontier) {
4216 for claim in &mut order.movable_claims {
4217 claim.transaction_order = claim.transaction_order.wrapping_add(1);
4218 }
4219 if let Some(recovery) = &mut order.recovery {
4220 if let Some(active_binding) = &mut recovery.active_binding {
4221 active_binding.transaction_order = active_binding.transaction_order.wrapping_add(1);
4222 }
4223 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_add(1);
4224 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_add(1);
4225 }
4226}
4227
4228fn rollback_ordinary_order_owners(order: &mut OrderClaimFrontier) {
4229 for claim in &mut order.movable_claims {
4230 claim.transaction_order = claim.transaction_order.wrapping_sub(1);
4231 }
4232 if let Some(recovery) = &mut order.recovery {
4233 if let Some(active_binding) = &mut recovery.active_binding {
4234 active_binding.transaction_order = active_binding.transaction_order.wrapping_sub(1);
4235 }
4236 recovery.recovery_operation_order = recovery.recovery_operation_order.wrapping_sub(1);
4237 recovery.replacement_terminal_order = recovery.replacement_terminal_order.wrapping_sub(1);
4238 }
4239}
4240
4241fn validate_leave_identity<F>(
4242 frontiers: &ClaimFrontiers,
4243 member: &LiveMember<F>,
4244) -> Result<(), PrepareLeaveAuthorityError> {
4245 if member.conversation_id() != frontiers.conversation_id {
4246 return Err(PrepareLeaveAuthorityError::Conversation);
4247 }
4248 let Some(participant) =
4249 active_participant(&frontiers.active_identities, member.participant_id())
4250 else {
4251 return Err(PrepareLeaveAuthorityError::Identity);
4252 };
4253 if participant.cursor != member.cursor() {
4254 return Err(PrepareLeaveAuthorityError::Identity);
4255 }
4256 Ok(())
4257}
4258
4259fn validate_settled_leave_prestate<F>(
4260 frontiers: &ClaimFrontiers,
4261 member: &LiveMember<F>,
4262 binding_state: BindingState,
4263) -> Result<(ParticipantId, Option<BindingEpoch>), PrepareLeaveAuthorityError> {
4264 let participant_id = member.participant_id();
4265 validate_leave_identity(frontiers, member)?;
4266 if !frontiers.order.immutable_candidates.is_empty() {
4267 return Err(PrepareLeaveAuthorityError::ImmutablePrefix);
4268 }
4269 let ended_binding_epoch = match binding_state {
4270 BindingState::Detached => {
4271 let Some(participant) =
4272 active_participant(&frontiers.active_identities, participant_id)
4273 else {
4274 return Err(PrepareLeaveAuthorityError::Identity);
4275 };
4276 if !matches!(participant.binding, FrontierBinding::Detached(_)) {
4277 return Err(PrepareLeaveAuthorityError::Binding);
4278 }
4279 None
4280 }
4281 BindingState::Bound(binding)
4282 if binding.conversation_id == frontiers.conversation_id
4283 && binding.participant_id == participant_id =>
4284 {
4285 let Some(participant) =
4286 active_participant(&frontiers.active_identities, participant_id)
4287 else {
4288 return Err(PrepareLeaveAuthorityError::Identity);
4289 };
4290 if participant.binding != FrontierBinding::Bound(binding.binding_epoch) {
4291 return Err(PrepareLeaveAuthorityError::Binding);
4292 }
4293 Some(binding.binding_epoch)
4294 }
4295 BindingState::Bound(_) | BindingState::PendingFinalization(_) => {
4296 return Err(PrepareLeaveAuthorityError::Binding);
4297 }
4298 };
4299 Ok((participant_id, ended_binding_epoch))
4300}
4301
4302fn validate_pending_leave_prestate<F>(
4303 frontiers: &ClaimFrontiers,
4304 member: &LiveMember<F>,
4305 pending: PendingFinalization,
4306) -> Result<(ParticipantId, super::AdmissionOrder), PrepareLeaveAuthorityError> {
4307 let participant_id = member.participant_id();
4308 validate_leave_identity(frontiers, member)?;
4309 if pending.conversation_id() != frontiers.conversation_id
4310 || pending.participant_id() != participant_id
4311 {
4312 return Err(PrepareLeaveAuthorityError::Binding);
4313 }
4314 let Some(participant) = active_participant(&frontiers.active_identities, participant_id) else {
4315 return Err(PrepareLeaveAuthorityError::Identity);
4316 };
4317 if participant.binding != FrontierBinding::Detached(pending.binding_epoch()) {
4318 return Err(PrepareLeaveAuthorityError::Binding);
4319 }
4320 let expected_order = pending.admission_order();
4321 let exact_sequence_candidate = matches!(
4322 frontiers.sequence.immutable_candidates.as_slice(),
4323 [ImmutableSequenceCandidate::BindingTerminal {
4324 admission_order,
4325 owner,
4326 ..
4327 }] if *admission_order == expected_order
4328 && owner.participant_index == participant_id
4329 && owner.binding_epoch == pending.binding_epoch()
4330 );
4331 let exact_order_candidate = matches!(
4332 frontiers.order.immutable_candidates.as_slice(),
4333 [ImmutableOrderCandidateMajor {
4334 transaction_order,
4335 candidate_keys,
4336 }] if *transaction_order == expected_order.transaction_order()
4337 && candidate_keys.as_slice() == [expected_order]
4338 );
4339 if !exact_sequence_candidate || !exact_order_candidate {
4340 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4341 }
4342 Ok((participant_id, expected_order))
4343}
4344
4345#[derive(Clone, Copy)]
4346enum LeaveRelayUnit {
4347 Direct(MovableOrderClaim),
4348 Recovery(RecoveryOrderBlock),
4349}
4350
4351impl LeaveRelayUnit {
4352 fn start(self) -> TransactionOrder {
4353 match self {
4354 Self::Direct(claim) => claim.transaction_order,
4355 Self::Recovery(block) => block_start_validated_order(block),
4356 }
4357 }
4358
4359 const fn len(self) -> u64 {
4360 match self {
4361 Self::Direct(_) => 1,
4362 Self::Recovery(block) => match block.active_binding {
4363 Some(_) => 3,
4364 None => 2,
4365 },
4366 }
4367 }
4368}
4369
4370struct LeaveOrderSelection {
4371 units: Vec<LeaveRelayUnit>,
4372 selected_major: TransactionOrder,
4373}
4374
4375fn select_leave_order(
4376 order: &OrderClaimFrontier,
4377 participant_id: ParticipantId,
4378 ended_binding_epoch: Option<BindingEpoch>,
4379 pending_order: Option<super::AdmissionOrder>,
4380) -> Result<LeaveOrderSelection, PrepareLeaveAuthorityError> {
4381 let Some(exit_index) = order.movable_claims.iter().position(|claim| {
4382 claim.owner
4383 == OrderDirectOwner::MembershipExit {
4384 participant_index: participant_id,
4385 }
4386 }) else {
4387 return Err(PrepareLeaveAuthorityError::MembershipExitClaim);
4388 };
4389 let exit_claim = order.movable_claims[exit_index];
4390 if pending_order
4391 .is_some_and(|pending| pending.transaction_order() >= exit_claim.transaction_order)
4392 {
4393 return Err(PrepareLeaveAuthorityError::PendingCandidate);
4394 }
4395 let active_index = matching_active_claim(order, participant_id, ended_binding_epoch)?;
4396 let mut units = Vec::new();
4397 for (index, claim) in order.movable_claims.iter().copied().enumerate() {
4398 if index != exit_index && Some(index) != active_index {
4399 units.push(LeaveRelayUnit::Direct(claim));
4400 }
4401 }
4402 if let Some(recovery) = order
4403 .recovery
4404 .filter(|recovery| recovery.participant_index != participant_id)
4405 {
4406 units.push(LeaveRelayUnit::Recovery(recovery));
4407 }
4408 units.sort_by_key(|unit| unit.start());
4409 let surviving_handles: u128 = units.iter().map(|unit| u128::from(unit.len())).sum();
4410 let exit_major = exit_claim.transaction_order;
4411 let later_handle_fits = u128::from(u64::MAX - exit_major) >= surviving_handles;
4412 let selected_major = if later_handle_fits {
4413 exit_major
4414 } else if pending_order.is_some() {
4415 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4416 } else {
4417 u64::try_from(order_frontier_start(order.ledger.high()))
4418 .map_err(|_| PrepareLeaveAuthorityError::OrderCapacity)?
4419 };
4420 Ok(LeaveOrderSelection {
4421 units,
4422 selected_major,
4423 })
4424}
4425
4426fn matching_active_claim(
4427 order: &OrderClaimFrontier,
4428 participant_id: ParticipantId,
4429 ended_binding_epoch: Option<BindingEpoch>,
4430) -> Result<Option<usize>, PrepareLeaveAuthorityError> {
4431 let Some(binding_epoch) = ended_binding_epoch else {
4432 return Ok(None);
4433 };
4434 let expected = OrderDirectOwner::ActiveBindingTerminal(BindingTerminalOwner {
4435 participant_index: participant_id,
4436 binding_epoch,
4437 });
4438 order
4439 .movable_claims
4440 .iter()
4441 .position(|claim| claim.owner == expected)
4442 .map(Some)
4443 .ok_or(PrepareLeaveAuthorityError::ActiveBindingClaim)
4444}
4445
4446fn relay_leave_order_units(
4447 units: Vec<LeaveRelayUnit>,
4448 selected_major: TransactionOrder,
4449) -> Result<(Vec<MovableOrderClaim>, Option<RecoveryOrderBlock>), PrepareLeaveAuthorityError> {
4450 let mut cursor = selected_major.checked_add(1);
4451 let mut movable_claims = Vec::new();
4452 let mut recovery = None;
4453 for unit in units {
4454 match unit {
4455 LeaveRelayUnit::Direct(mut claim) => {
4456 let Some(position) = cursor else {
4457 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4458 };
4459 claim.transaction_order = position;
4460 movable_claims.push(claim);
4461 cursor = position.checked_add(1);
4462 }
4463 LeaveRelayUnit::Recovery(block) => {
4464 let (relayed, next) = relay_recovery_block(block, cursor)?;
4465 recovery = Some(relayed);
4466 cursor = next;
4467 }
4468 }
4469 }
4470 movable_claims.sort_by_key(|claim| claim.transaction_order);
4471 Ok((movable_claims, recovery))
4472}
4473
4474const fn relay_recovery_block(
4475 block: RecoveryOrderBlock,
4476 cursor: Option<TransactionOrder>,
4477) -> Result<(RecoveryOrderBlock, Option<TransactionOrder>), PrepareLeaveAuthorityError> {
4478 let mut next = cursor;
4479 let active_binding = if let Some(mut active) = block.active_binding {
4480 let Some(position) = next else {
4481 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4482 };
4483 active.transaction_order = position;
4484 next = position.checked_add(1);
4485 Some(active)
4486 } else {
4487 None
4488 };
4489 let Some(recovery_operation_order) = next else {
4490 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4491 };
4492 let Some(replacement_terminal_order) = recovery_operation_order.checked_add(1) else {
4493 return Err(PrepareLeaveAuthorityError::OrderCapacity);
4494 };
4495 Ok((
4496 RecoveryOrderBlock {
4497 active_binding,
4498 recovery_operation_order,
4499 replacement_terminal_order,
4500 participant_index: block.participant_index,
4501 marker_delivery_seq: block.marker_delivery_seq,
4502 recovered_binding_epoch: block.recovered_binding_epoch,
4503 },
4504 replacement_terminal_order.checked_add(1),
4505 ))
4506}
4507
4508fn leave_resulting_order_ledger(
4509 selected_major: TransactionOrder,
4510 movable_claims: &[MovableOrderClaim],
4511 recovery: Option<RecoveryOrderBlock>,
4512) -> Result<OrderLedger, PrepareLeaveAuthorityError> {
4513 let active_binding_terminals =
4514 usize_to_u64(
4515 movable_claims
4516 .iter()
4517 .filter(|claim| matches!(claim.owner, OrderDirectOwner::ActiveBindingTerminal(_)))
4518 .count(),
4519 ) + u64::from(recovery.is_some_and(|block| block.active_binding.is_some()));
4520 let membership_exits = usize_to_u64(
4521 movable_claims
4522 .iter()
4523 .filter(|claim| matches!(claim.owner, OrderDirectOwner::MembershipExit { .. }))
4524 .count(),
4525 );
4526 let has_recovery = recovery.is_some();
4527 let resulting_claims = OrderClaims::new(
4528 active_binding_terminals,
4529 membership_exits,
4530 has_recovery,
4531 has_recovery,
4532 )
4533 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)?;
4534 OrderLedger::try_new(OrderHigh::Allocated(selected_major), resulting_claims)
4535 .map_err(|_| PrepareLeaveAuthorityError::ResultingOrderLedger)
4536}
4537
4538fn consume_leave_order_lane(
4539 order: &mut OrderClaimFrontier,
4540 participant_id: ParticipantId,
4541 ended_binding_epoch: Option<BindingEpoch>,
4542 pending_order: Option<super::AdmissionOrder>,
4543) -> Result<TransactionOrder, PrepareLeaveAuthorityError> {
4544 let selection = select_leave_order(order, participant_id, ended_binding_epoch, pending_order)?;
4545 let (movable_claims, recovery) =
4546 relay_leave_order_units(selection.units, selection.selected_major)?;
4547 let ledger = leave_resulting_order_ledger(selection.selected_major, &movable_claims, recovery)?;
4548 *order = OrderClaimFrontier {
4549 ledger,
4550 movable_claims,
4551 immutable_candidates: Vec::new(),
4552 recovery,
4553 };
4554 Ok(selection.selected_major)
4555}
4556
4557impl ClaimFrontiersPrevalidated {
4558 #[must_use]
4560 pub(super) const fn conversation_id(&self) -> ConversationId {
4561 self.conversation_id
4562 }
4563
4564 pub(super) fn take_marker_record(
4570 &mut self,
4571 request: MarkerRecordRequest,
4572 ) -> Option<ValidatedMarkerRecord> {
4573 if self.issued_marker_record.is_some() {
4574 return None;
4575 }
4576 let record = self
4577 .retained_records
4578 .iter()
4579 .find(|record| record.delivery_seq == request.marker_delivery_seq)
4580 .copied()?;
4581 let RetainedCausalRecordKind::CompactionMarker {
4582 participant_index,
4583 provenance,
4584 } = record.kind
4585 else {
4586 return None;
4587 };
4588 if participant_index != request.participant_index {
4589 return None;
4590 }
4591 let participant = active_participant(&self.active_identities, participant_index)?;
4592 let historical_delivery = self
4593 .historical_marker_deliveries
4594 .iter()
4595 .find(|authority| authority.marker_delivery_seq == request.marker_delivery_seq);
4596 let (target_binding, occurrence) = match request.use_kind {
4597 MarkerRecordUse::Planned(target) => {
4598 if participant.binding != target || historical_delivery.is_some() {
4599 return None;
4600 }
4601 (target, MarkerRecordOccurrence::Undelivered)
4602 }
4603 MarkerRecordUse::Delivered(target) => {
4604 let delivered_binding_epoch = binding_epoch(target);
4605 if participant.binding != target
4606 || !historical_delivery.is_some_and(|authority| {
4607 authority.participant_index == participant_index
4608 && authority.delivered_binding_epoch == delivered_binding_epoch
4609 })
4610 {
4611 return None;
4612 }
4613 (target, MarkerRecordOccurrence::Delivered)
4614 }
4615 MarkerRecordUse::Recovered {
4616 prior_binding_epoch,
4617 recovered_binding_epoch,
4618 } => {
4619 if participant.binding != FrontierBinding::Detached(recovered_binding_epoch)
4620 || !historical_delivery.is_some_and(|authority| {
4621 authority.participant_index == participant_index
4622 && authority.delivered_binding_epoch == prior_binding_epoch
4623 })
4624 {
4625 return None;
4626 }
4627 (
4628 FrontierBinding::Detached(prior_binding_epoch),
4629 MarkerRecordOccurrence::Delivered,
4630 )
4631 }
4632 };
4633 self.issued_marker_record = Some(request);
4634 Some(ValidatedMarkerRecord {
4635 conversation_id: self.conversation_id,
4636 record,
4637 provenance,
4638 target_binding,
4639 occurrence,
4640 seal: MarkerAuthoritySeal::Validated,
4641 })
4642 }
4643
4644 pub(super) fn finish(
4647 self,
4648 edge: Option<super::StoredEdge>,
4649 ) -> Result<ClaimFrontiers, ParticipantStateCorruptReason> {
4650 self.validate_current_marker_edge(edge)?;
4651 self.validate_historical_delivery_consumers(edge)?;
4652 let recovery_provenance = self.resolve_recovery_provenance(edge)?;
4653 let sequence = restore_sequence_frontier(
4654 &self.active_identities,
4655 self.sequence_restore,
4656 self.retained_floor,
4657 self.sequence_ledger,
4658 recovery_provenance,
4659 &self.retained_records,
4660 &self.historical_causal_authorities,
4661 )
4662 .map_err(corrupt_frontier)?;
4663 let order = restore_order_frontier(
4664 &self.active_identities,
4665 self.order_restore,
4666 self.order_ledger,
4667 recovery_provenance,
4668 )
4669 .map_err(corrupt_frontier)?;
4670 validate_cross_counter(&sequence, &order).map_err(corrupt_frontier)?;
4671 Ok(ClaimFrontiers {
4672 conversation_id: self.conversation_id,
4673 active_identities: self.active_identities,
4674 identity_slot_limit: self.identity_slot_limit,
4675 retained_floor: self.retained_floor,
4676 retained_records: self.retained_records,
4677 marker_records: self.marker_records,
4678 fenced_marker_issued: None,
4679 sequence,
4680 order,
4681 })
4682 }
4683
4684 fn validate_historical_delivery_consumers(
4685 &self,
4686 edge: Option<super::StoredEdge>,
4687 ) -> Result<(), ParticipantStateCorruptReason> {
4688 for history in &self.historical_marker_deliveries {
4689 let recovered_origin = self.binding_origins.iter().any(|origin| {
4690 origin.participant_id() == history.participant_index
4691 && origin.recovered_marker()
4692 == Some((history.marker_delivery_seq, history.delivered_binding_epoch))
4693 });
4694 let current_marker_edge = match edge {
4695 Some(super::StoredEdge::ParticipantCursorProgress(progress)) => {
4696 progress.participant_id() == history.participant_index
4697 && progress.marker_delivery_seq() == Some(history.marker_delivery_seq)
4698 && progress.binding_epoch() == history.delivered_binding_epoch
4699 }
4700 Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) => {
4701 recovery.participant_id() == history.participant_index
4702 && recovery.marker_delivery_seq() == history.marker_delivery_seq
4703 && recovery.prior_binding_epoch() == history.delivered_binding_epoch
4704 }
4705 Some(
4706 super::StoredEdge::ObserverProjection(_)
4707 | super::StoredEdge::PhysicalCompaction(_)
4708 | super::StoredEdge::MarkerDelivery(_)
4709 | super::StoredEdge::DetachedMarkerRelease(_)
4710 | super::StoredEdge::DetachedCursorRelease(_),
4711 )
4712 | None => false,
4713 };
4714 if !recovered_origin && !current_marker_edge {
4715 return Err(self.marker_corruption(history.marker_delivery_seq));
4716 }
4717 }
4718 Ok(())
4719 }
4720
4721 fn validate_current_marker_edge(
4722 &self,
4723 edge: Option<super::StoredEdge>,
4724 ) -> Result<(), ParticipantStateCorruptReason> {
4725 if let Some(MarkerRecordRequest {
4726 participant_index,
4727 marker_delivery_seq,
4728 use_kind:
4729 MarkerRecordUse::Recovered {
4730 recovered_binding_epoch,
4731 ..
4732 },
4733 }) = self.issued_marker_record
4734 {
4735 let recovered_edge_matches = matches!(
4736 edge,
4737 Some(super::StoredEdge::DetachedCursorRelease(release))
4738 if release.participant_id() == participant_index
4739 && release.last_dead_binding_epoch() == recovered_binding_epoch
4740 );
4741 if !recovered_edge_matches {
4742 return Err(self.marker_corruption(marker_delivery_seq));
4743 }
4744 return Ok(());
4745 }
4746 let Some(context) = edge.and_then(marker_edge_context) else {
4747 return Ok(());
4748 };
4749 if self
4750 .issued_marker_record
4751 .is_none_or(|request| request.marker_delivery_seq != context.marker_delivery_seq)
4752 || !self.marker_context_matches(context)
4753 {
4754 return Err(self.marker_corruption(context.marker_delivery_seq));
4755 }
4756 Ok(())
4757 }
4758
4759 fn resolve_recovery_provenance(
4760 &self,
4761 edge: Option<super::StoredEdge>,
4762 ) -> Result<Option<RecoveryClaimProvenance>, ParticipantStateCorruptReason> {
4763 let Some(marker_delivery_seq) = self.recovery_marker_delivery_seq else {
4764 return Ok(None);
4765 };
4766
4767 if let Some(super::StoredEdge::DetachedCredentialRecovery(recovery)) = edge {
4768 return self
4769 .resolve_postfate_recovery_provenance(marker_delivery_seq, recovery)
4770 .map(Some);
4771 }
4772
4773 let candidate =
4774 self.sequence_restore.immutable_candidates.iter().find_map(
4775 |candidate| match candidate {
4776 ImmutableSequenceCandidate::Marker(marker)
4777 if marker.delivery_seq == marker_delivery_seq
4778 && matches!(marker.target_binding, FrontierBinding::Bound(_)) =>
4779 {
4780 Some(*marker)
4781 }
4782 _ => None,
4783 },
4784 );
4785 let retained = self.retained_records.iter().find_map(|record| {
4786 let RetainedCausalRecordKind::CompactionMarker {
4787 participant_index, ..
4788 } = record.kind
4789 else {
4790 return None;
4791 };
4792 if record.delivery_seq != marker_delivery_seq {
4793 return None;
4794 }
4795 let participant = active_participant(&self.active_identities, participant_index)?;
4796 let FrontierBinding::Bound(binding_epoch) = participant.binding else {
4797 return None;
4798 };
4799 let historical = self.historical_marker_deliveries.iter().find(|history| {
4800 history.participant_index == participant_index
4801 && history.marker_delivery_seq == marker_delivery_seq
4802 })?;
4803 Some((
4804 participant_index,
4805 binding_epoch,
4806 historical.delivered_binding_epoch,
4807 ))
4808 });
4809 let provenance = match (candidate, retained) {
4810 (Some(marker), None) => RecoveryClaimProvenance {
4811 participant_index: marker.admission_order.participant_index(),
4812 marker_delivery_seq,
4813 prior_binding_epoch: binding_epoch(marker.target_binding),
4814 current_binding_epoch: binding_epoch(marker.target_binding),
4815 phase: RecoveryClaimPhase::PreFate,
4816 },
4817 (None, Some((participant_index, current_binding_epoch, prior_binding_epoch))) => {
4818 let phase = if current_binding_epoch == prior_binding_epoch {
4819 let Some(context) = edge.and_then(marker_edge_context) else {
4820 return Err(self.marker_corruption(marker_delivery_seq));
4821 };
4822 if context.marker_delivery_seq != marker_delivery_seq
4823 || context.participant_index != participant_index
4824 || context.binding_epoch != prior_binding_epoch
4825 || context.target_binding != FrontierBinding::Bound(prior_binding_epoch)
4826 {
4827 return Err(self.marker_corruption(marker_delivery_seq));
4828 }
4829 RecoveryClaimPhase::PreFate
4830 } else {
4831 let recovered_origin_matches = self.binding_origins.iter().any(|origin| {
4832 origin.participant_id() == participant_index
4833 && origin.binding_epoch() == current_binding_epoch
4834 && origin.recovered_marker()
4835 == Some((marker_delivery_seq, prior_binding_epoch))
4836 });
4837 if !recovered_origin_matches
4838 || !matches!(
4839 edge,
4840 Some(
4841 super::StoredEdge::ObserverProjection(_)
4842 | super::StoredEdge::PhysicalCompaction(_)
4843 )
4844 )
4845 {
4846 return Err(self.marker_corruption(marker_delivery_seq));
4847 }
4848 RecoveryClaimPhase::RecoveredBound
4849 };
4850 RecoveryClaimProvenance {
4851 participant_index,
4852 marker_delivery_seq,
4853 prior_binding_epoch,
4854 current_binding_epoch,
4855 phase,
4856 }
4857 }
4858 (Some(_), Some(_)) | (None, None) => {
4859 return Err(self.marker_corruption(marker_delivery_seq));
4860 }
4861 };
4862 Ok(Some(provenance))
4863 }
4864
4865 fn resolve_postfate_recovery_provenance(
4866 &self,
4867 marker_delivery_seq: DeliverySeq,
4868 recovery: super::DetachedCredentialRecovery,
4869 ) -> Result<RecoveryClaimProvenance, ParticipantStateCorruptReason> {
4870 let provenance = RecoveryClaimProvenance {
4871 participant_index: recovery.participant_id(),
4872 marker_delivery_seq: recovery.marker_delivery_seq(),
4873 prior_binding_epoch: recovery.prior_binding_epoch(),
4874 current_binding_epoch: recovery.prior_binding_epoch(),
4875 phase: RecoveryClaimPhase::PostFate,
4876 };
4877 let context = MarkerEdgeContext {
4878 participant_index: provenance.participant_index,
4879 marker_delivery_seq: provenance.marker_delivery_seq,
4880 binding_epoch: provenance.prior_binding_epoch,
4881 target_binding: FrontierBinding::Detached(provenance.prior_binding_epoch),
4882 };
4883 if marker_delivery_seq != provenance.marker_delivery_seq
4884 || self
4885 .issued_marker_record
4886 .is_none_or(|request| request.marker_delivery_seq != marker_delivery_seq)
4887 || !self.marker_context_matches(context)
4888 {
4889 return Err(self.marker_corruption(marker_delivery_seq));
4890 }
4891 Ok(provenance)
4892 }
4893
4894 fn marker_context_matches(&self, context: MarkerEdgeContext) -> bool {
4895 self.marker_records.iter().any(|record| {
4896 matches!(
4897 record.kind,
4898 RetainedCausalRecordKind::CompactionMarker {
4899 participant_index,
4900 ..
4901 } if participant_index == context.participant_index
4902 ) && record.delivery_seq == context.marker_delivery_seq
4903 && active_participant(&self.active_identities, context.participant_index)
4904 .is_some_and(|participant| participant.binding == context.target_binding)
4905 })
4906 }
4907
4908 fn marker_corruption(&self, delivery_seq: DeliverySeq) -> ParticipantStateCorruptReason {
4909 corrupt_frontier(sequence_error(
4910 sequence_ordinal(self.sequence_ledger, delivery_seq),
4911 ClaimFrontierInvalidReason::RecoveryBlock,
4912 ))
4913 }
4914}
4915
4916#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4917struct MarkerEdgeContext {
4918 participant_index: ParticipantId,
4919 marker_delivery_seq: DeliverySeq,
4920 binding_epoch: BindingEpoch,
4921 target_binding: FrontierBinding,
4922}
4923
4924fn marker_edge_context(edge: super::StoredEdge) -> Option<MarkerEdgeContext> {
4925 match edge {
4926 super::StoredEdge::MarkerDelivery(delivery) => Some(MarkerEdgeContext {
4927 participant_index: delivery.participant_id(),
4928 marker_delivery_seq: delivery.marker_delivery_seq(),
4929 binding_epoch: delivery.binding_epoch(),
4930 target_binding: FrontierBinding::Bound(delivery.binding_epoch()),
4931 }),
4932 super::StoredEdge::ParticipantCursorProgress(progress) => {
4933 let marker_delivery_seq = progress.marker_delivery_seq()?;
4934 Some(MarkerEdgeContext {
4935 participant_index: progress.participant_id(),
4936 marker_delivery_seq,
4937 binding_epoch: progress.binding_epoch(),
4938 target_binding: FrontierBinding::Bound(progress.binding_epoch()),
4939 })
4940 }
4941 super::StoredEdge::DetachedCredentialRecovery(recovery) => Some(MarkerEdgeContext {
4942 participant_index: recovery.participant_id(),
4943 marker_delivery_seq: recovery.marker_delivery_seq(),
4944 binding_epoch: recovery.prior_binding_epoch(),
4945 target_binding: FrontierBinding::Detached(recovery.prior_binding_epoch()),
4946 }),
4947 super::StoredEdge::DetachedMarkerRelease(release) => Some(MarkerEdgeContext {
4948 participant_index: release.participant_id(),
4949 marker_delivery_seq: release.marker_delivery_seq(),
4950 binding_epoch: release.last_dead_binding_epoch(),
4951 target_binding: FrontierBinding::Detached(release.last_dead_binding_epoch()),
4952 }),
4953 super::StoredEdge::ObserverProjection(_)
4954 | super::StoredEdge::PhysicalCompaction(_)
4955 | super::StoredEdge::DetachedCursorRelease(_) => None,
4956 }
4957}
4958
4959#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4960#[repr(u8)]
4961enum SequenceClass {
4962 Exit = 0,
4963 Terminal = 1,
4964 Marker = 2,
4965 RecoveryAttach = 3,
4966 RecoveryReplacementTerminal = 4,
4967 LiveTimesTerminal = 5,
4968 LiveTimesReplacementTerminal = 6,
4969 OtherLiveTimesExit = 7,
4970}
4971
4972#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4973#[repr(u8)]
4974enum OrderClass {
4975 ActiveBindingTerminal = 0,
4976 MembershipExit = 1,
4977 RecoveryOperation = 2,
4978 RecoveryReplacementTerminal = 3,
4979}
4980
4981#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4982struct NumericSegment<C> {
4983 start: u128,
4984 length: u128,
4985 class: Option<C>,
4986 immutable: bool,
4987}
4988
4989fn first_duplicate_candidate_key(
4990 candidates: &[ImmutableSequenceCandidate],
4991 retained_records: &[RetainedCausalRecord],
4992) -> Option<super::AdmissionOrder> {
4993 let mut keys: Vec<_> = candidates
4994 .iter()
4995 .map(|candidate| candidate.admission_order())
4996 .chain(retained_records.iter().map(|record| record.admission_order))
4997 .collect();
4998 keys.sort_unstable();
4999 let mut previous = None;
5000 for key in keys {
5001 if previous == Some(key) {
5002 return Some(key);
5003 }
5004 previous = Some(key);
5005 }
5006 None
5007}
5008
5009fn validate_unique_candidate_keys(
5010 candidates: &[ImmutableSequenceCandidate],
5011 retained_records: &[RetainedCausalRecord],
5012) -> Result<(), ParticipantStateCorruptReason> {
5013 let Some(order) = first_duplicate_candidate_key(candidates, retained_records) else {
5014 return Ok(());
5015 };
5016 Err(ParticipantStateCorruptReason::DuplicateCandidateKey {
5017 transaction_order: order.transaction_order(),
5018 candidate_phase: order.candidate_phase(),
5019 participant_index: order.participant_index(),
5020 })
5021}
5022
5023fn validate_sequence_numeric(
5024 restore: &SequenceClaimFrontierRestore,
5025 ledger: SequenceLedger,
5026) -> Result<(), ClaimFrontierError> {
5027 let mut segments = sequence_segments(restore);
5028 validate_numeric_segments(
5029 ClaimFrontierCounter::DeliverySequence,
5030 u128::from(ledger.high_watermark()) + 1,
5031 ledger.required_reserve(),
5032 &mut segments,
5033 &sequence_expected_counts(ledger),
5034 )
5035}
5036
5037fn validate_order_numeric(
5038 restore: &OrderClaimFrontierRestore,
5039 ledger: OrderLedger,
5040) -> Result<(), ClaimFrontierError> {
5041 let mut segments = order_segments(restore, ledger.high());
5042 validate_numeric_segments(
5043 ClaimFrontierCounter::TransactionOrder,
5044 order_frontier_start(ledger.high()),
5045 order_frontier_candidate_count(restore, ledger.high()) + ledger.claims().total(),
5046 &mut segments,
5047 &order_expected_counts(ledger),
5048 )
5049}
5050
5051fn validate_bounded_shape(
5052 restore: &ClaimFrontiersRestore,
5053 sequence_ledger: SequenceLedger,
5054) -> Result<(), ClaimFrontierError> {
5055 let identity_limit = u128::from(restore.identity_slot_limit);
5056 let twice_identity_limit = identity_limit.saturating_mul(2);
5057 let order_candidate_keys = restore
5058 .order
5059 .immutable_candidates
5060 .iter()
5061 .fold(0_u128, |count, candidate| {
5062 count.saturating_add(usize_to_u128(candidate.candidate_keys.len()))
5063 });
5064 let bounded = usize_to_u128(restore.active_identities.len()) <= identity_limit
5065 && usize_to_u128(restore.sequence.movable_claims.len()) <= twice_identity_limit
5066 && usize_to_u128(restore.sequence.immutable_candidates.len()) <= twice_identity_limit
5067 && usize_to_u128(restore.sequence.products.live_times_terminal.len()) <= identity_limit
5068 && usize_to_u128(restore.sequence.products.other_live_times_exit.len()) <= identity_limit
5069 && usize_to_u128(restore.historical_marker_deliveries.len())
5070 <= u128::from(restore.retained_record_limit)
5071 && usize_to_u128(restore.historical_causal_facts.len()) <= twice_identity_limit
5072 && usize_to_u128(restore.order.movable_claims.len()) <= twice_identity_limit
5073 && usize_to_u128(restore.order.immutable_candidates.len()) <= twice_identity_limit
5074 && order_candidate_keys <= twice_identity_limit;
5075 if bounded {
5076 Ok(())
5077 } else {
5078 Err(sequence_error(
5079 sequence_ledger.required_reserve(),
5080 ClaimFrontierInvalidReason::LogicalOwner,
5081 ))
5082 }
5083}
5084
5085fn validated_retained_records(
5086 mut records: Vec<RetainedCausalRecord>,
5087 retained_floor: u128,
5088 retained_record_limit: u64,
5089 identity_slot_limit: u64,
5090 ledger: SequenceLedger,
5091) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
5092 let high_end = u128::from(ledger.high_watermark()) + 1;
5093 let expected_count = high_end.saturating_sub(retained_floor);
5094 if retained_floor > high_end
5095 || usize_to_u128(records.len()) > u128::from(retained_record_limit)
5096 || usize_to_u128(records.len()) != expected_count
5097 {
5098 return Err(sequence_error(
5099 usize_to_u128(records.len()).min(expected_count),
5100 ClaimFrontierInvalidReason::LogicalOwner,
5101 ));
5102 }
5103 records.sort_by_key(|record| record.delivery_seq);
5104 let mut previous_admission_order = None;
5105 for (record_index, record) in records.iter().enumerate() {
5106 let (participant_index, valid_kind) = match record.kind {
5107 RetainedCausalRecordKind::BindingTerminal(owner) => (
5108 owner.participant_index,
5109 record.admission_order.candidate_phase() == CandidatePhase::BindingTerminal
5110 && record.admission_order.participant_index() == owner.participant_index,
5111 ),
5112 RetainedCausalRecordKind::MembershipExit { participant_index } => (
5113 participant_index,
5114 record.admission_order.candidate_phase() == CandidatePhase::MembershipExit
5115 && record.admission_order.participant_index() == participant_index,
5116 ),
5117 RetainedCausalRecordKind::AttachLifecycle {
5118 participant_index, ..
5119 } => (
5120 participant_index,
5121 record.admission_order.candidate_phase() == CandidatePhase::AttachLifecycle
5122 && record.admission_order.participant_index() == participant_index,
5123 ),
5124 RetainedCausalRecordKind::OrdinaryRecord { participant_index } => (
5125 participant_index,
5126 record.admission_order.candidate_phase() == CandidatePhase::OrdinaryRecord
5127 && record.admission_order.participant_index() == participant_index,
5128 ),
5129 RetainedCausalRecordKind::CompactionMarker {
5130 participant_index,
5131 provenance,
5132 } => (
5133 participant_index,
5134 record.admission_order.candidate_phase() == CandidatePhase::CompactionMarker
5135 && record.admission_order.participant_index() == participant_index
5136 && marker_provenance_targets(provenance, participant_index),
5137 ),
5138 };
5139 let expected_sequence = retained_floor + rank_index(record_index);
5140 if participant_index >= identity_slot_limit
5141 || !valid_kind
5142 || previous_admission_order.is_some_and(|previous| previous >= record.admission_order)
5143 || u128::from(record.delivery_seq) != expected_sequence
5144 {
5145 return Err(sequence_error(
5146 rank_index(record_index),
5147 ClaimFrontierInvalidReason::CandidateKey,
5148 ));
5149 }
5150 previous_admission_order = Some(record.admission_order);
5151 }
5152 Ok(records)
5153}
5154
5155fn validated_active_marker_records(
5156 retained_markers: &[RetainedCausalRecord],
5157 mut active_marker_anchors: Vec<DeliverySeq>,
5158 identity_slot_limit: u64,
5159 ledger: SequenceLedger,
5160) -> Result<Vec<RetainedCausalRecord>, ClaimFrontierError> {
5161 if usize_to_u128(active_marker_anchors.len()) > u128::from(identity_slot_limit) {
5162 return Err(sequence_error(
5163 ledger.required_reserve(),
5164 ClaimFrontierInvalidReason::LogicalOwner,
5165 ));
5166 }
5167 active_marker_anchors.sort_unstable();
5168 let mut previous_sequence = None;
5169 let mut owners = Vec::new();
5170 let mut active_records = Vec::new();
5171 for delivery_seq in active_marker_anchors {
5172 let Some(record) = retained_markers
5173 .iter()
5174 .find(|record| record.delivery_seq == delivery_seq)
5175 .copied()
5176 else {
5177 return Err(sequence_error(
5178 sequence_ordinal(ledger, delivery_seq),
5179 ClaimFrontierInvalidReason::LogicalOwner,
5180 ));
5181 };
5182 let RetainedCausalRecordKind::CompactionMarker {
5183 participant_index, ..
5184 } = record.kind
5185 else {
5186 return Err(sequence_error(
5187 sequence_ordinal(ledger, delivery_seq),
5188 ClaimFrontierInvalidReason::LogicalOwner,
5189 ));
5190 };
5191 if previous_sequence == Some(delivery_seq) || owners.contains(&participant_index) {
5192 return Err(sequence_error(
5193 sequence_ordinal(ledger, delivery_seq),
5194 ClaimFrontierInvalidReason::LogicalOwner,
5195 ));
5196 }
5197 previous_sequence = Some(delivery_seq);
5198 owners.push(participant_index);
5199 active_records.push(record);
5200 }
5201 Ok(active_records)
5202}
5203
5204fn validated_historical_marker_deliveries(
5205 mut facts: Vec<HistoricalMarkerDeliveryFactRestore>,
5206 conversation_id: ConversationId,
5207 active: &ActiveIdentityRanks,
5208 sources: &MarkerDeliverySources<'_>,
5209 retained_record_limit: u64,
5210 ledger: SequenceLedger,
5211) -> Result<Vec<HistoricalMarkerDeliveryAuthority>, ClaimFrontierError> {
5212 let MarkerDeliverySources {
5213 retained: retained_records,
5214 historical: historical_causal_authorities,
5215 candidates: immutable_candidates,
5216 } = *sources;
5217 if usize_to_u128(facts.len()) > u128::from(retained_record_limit) {
5218 return Err(sequence_error(
5219 ledger.required_reserve(),
5220 ClaimFrontierInvalidReason::LogicalOwner,
5221 ));
5222 }
5223 facts.sort_by_key(|fact| fact.marker_delivery_seq);
5224 let mut previous_sequence = None;
5225 let mut authorities = Vec::new();
5226 for fact in facts {
5227 let matching_record = retained_records.iter().find(|record| {
5228 record.delivery_seq == fact.marker_delivery_seq
5229 && matches!(
5230 record.kind,
5231 RetainedCausalRecordKind::CompactionMarker {
5232 participant_index,
5233 ..
5234 } if participant_index == fact.participant_index
5235 )
5236 });
5237 let current_bound =
5238 active_participant(active, fact.participant_index).is_some_and(|participant| {
5239 participant.binding == FrontierBinding::Bound(fact.delivered_binding_epoch)
5240 });
5241 let terminal_order = historical_causal_authorities
5242 .iter()
5243 .find_map(|authority| match authority.kind {
5244 HistoricalCausalKind::BindingTerminal(owner)
5245 if binding_terminal_matches_delivery(owner, fact) =>
5246 {
5247 Some(authority.admission_order)
5248 }
5249 HistoricalCausalKind::BindingTerminal(_)
5250 | HistoricalCausalKind::MembershipExit(_) => None,
5251 })
5252 .or_else(|| {
5253 retained_records
5254 .iter()
5255 .find_map(|record| match record.kind {
5256 RetainedCausalRecordKind::BindingTerminal(owner)
5257 if binding_terminal_matches_delivery(owner, fact) =>
5258 {
5259 Some(record.admission_order)
5260 }
5261 RetainedCausalRecordKind::BindingTerminal(_)
5262 | RetainedCausalRecordKind::MembershipExit { .. }
5263 | RetainedCausalRecordKind::AttachLifecycle { .. }
5264 | RetainedCausalRecordKind::OrdinaryRecord { .. }
5265 | RetainedCausalRecordKind::CompactionMarker { .. } => None,
5266 })
5267 })
5268 .or_else(|| {
5269 immutable_candidates
5270 .iter()
5271 .find_map(|candidate| match candidate {
5272 ImmutableSequenceCandidate::BindingTerminal {
5273 admission_order,
5274 owner,
5275 ..
5276 } if binding_terminal_matches_delivery(*owner, fact) => {
5277 Some(*admission_order)
5278 }
5279 ImmutableSequenceCandidate::BindingTerminal { .. }
5280 | ImmutableSequenceCandidate::Marker(_) => None,
5281 })
5282 });
5283 let historical_epoch_is_backed = matching_record.is_some_and(|marker_record| {
5284 current_bound
5285 || terminal_order
5286 .is_some_and(|terminal_order| terminal_order > marker_record.admission_order)
5287 });
5288 if fact.conversation_id != conversation_id
5289 || previous_sequence == Some(fact.marker_delivery_seq)
5290 || !historical_epoch_is_backed
5291 {
5292 return Err(sequence_error(
5293 sequence_ordinal(ledger, fact.marker_delivery_seq),
5294 ClaimFrontierInvalidReason::LogicalOwner,
5295 ));
5296 }
5297 previous_sequence = Some(fact.marker_delivery_seq);
5298 authorities.push(HistoricalMarkerDeliveryAuthority {
5299 participant_index: fact.participant_index,
5300 marker_delivery_seq: fact.marker_delivery_seq,
5301 delivered_binding_epoch: fact.delivered_binding_epoch,
5302 });
5303 }
5304 Ok(authorities)
5305}
5306
5307fn binding_terminal_matches_delivery(
5308 owner: BindingTerminalOwner,
5309 fact: HistoricalMarkerDeliveryFactRestore,
5310) -> bool {
5311 (owner.participant_index, owner.binding_epoch)
5312 == (fact.participant_index, fact.delivered_binding_epoch)
5313}
5314
5315struct MarkerDeliverySources<'a> {
5316 retained: &'a [RetainedCausalRecord],
5317 historical: &'a [HistoricalCausalAuthority],
5318 candidates: &'a [ImmutableSequenceCandidate],
5319}
5320
5321struct BindingOriginValidation<'a> {
5322 conversation_id: ConversationId,
5323 active: &'a ActiveIdentityRanks,
5324 origins: &'a [BindingOrigin],
5325 retained_records: &'a [RetainedCausalRecord],
5326 causal_authorities: &'a [HistoricalCausalAuthority],
5327 historical_marker_deliveries: &'a [HistoricalMarkerDeliveryAuthority],
5328 total: bool,
5329 ledger: SequenceLedger,
5330}
5331
5332impl BindingOriginValidation<'_> {
5333 fn validate(&self) -> Result<(), ClaimFrontierError> {
5334 if !self.total {
5335 return if self.origins.is_empty() {
5336 Ok(())
5337 } else {
5338 Err(self.logical_owner_error())
5339 };
5340 }
5341 if self.origins.len() != self.active.participants.len() {
5342 return Err(self.logical_owner_error());
5343 }
5344 for participant in &self.active.participants {
5345 let mut matching = self
5346 .origins
5347 .iter()
5348 .filter(|origin| origin.participant_id() == participant.participant_index);
5349 let Some(origin) = matching.next() else {
5350 return Err(self.logical_owner_error());
5351 };
5352 if matching.next().is_some() {
5353 return Err(self.logical_owner_error());
5354 }
5355 self.validate_origin(*participant, *origin)?;
5356 }
5357 Ok(())
5358 }
5359
5360 fn validate_origin(
5361 &self,
5362 participant: FrontierParticipant,
5363 origin: BindingOrigin,
5364 ) -> Result<(), ClaimFrontierError> {
5365 let current_epoch = binding_epoch(participant.binding);
5366 let attached = origin.attached();
5367 if origin.conversation_id() != self.conversation_id
5368 || origin.binding_epoch() != current_epoch
5369 || attached.conversation_id() != self.conversation_id
5370 || attached.participant_id() != participant.participant_index
5371 || attached.binding_epoch() != current_epoch
5372 || attached.admission_order().candidate_phase() != CandidatePhase::AttachLifecycle
5373 {
5374 return Err(self.logical_owner_error());
5375 }
5376 let mut retained_attach_for_binding = self.retained_records.iter().filter(|record| {
5377 matches!(
5378 record.kind,
5379 RetainedCausalRecordKind::AttachLifecycle {
5380 participant_index,
5381 binding_epoch,
5382 } if participant_index == participant.participant_index
5383 && binding_epoch == current_epoch
5384 )
5385 });
5386 let retained_attach_matches = retained_attach_for_binding.clone().any(|record| {
5387 record.delivery_seq == attached.delivery_seq()
5388 && record.admission_order == attached.admission_order()
5389 });
5390 if retained_attach_for_binding.next().is_some() && !retained_attach_matches {
5391 return Err(self.logical_owner_error());
5392 }
5393 if let Some((marker_delivery_seq, prior_binding_epoch)) = origin.recovered_marker() {
5394 let generation_is_next = prior_binding_epoch
5395 .capability_generation
5396 .get()
5397 .checked_add(1)
5398 == Some(current_epoch.capability_generation.get());
5399 let marker_history_matches = self.historical_marker_deliveries.iter().any(|history| {
5400 history.participant_index == participant.participant_index
5401 && history.marker_delivery_seq == marker_delivery_seq
5402 && history.delivered_binding_epoch == prior_binding_epoch
5403 });
5404 if !generation_is_next || !marker_history_matches {
5405 return Err(sequence_error(
5406 sequence_ordinal(self.ledger, marker_delivery_seq),
5407 ClaimFrontierInvalidReason::RecoveryBlock,
5408 ));
5409 }
5410 } else if matches!(participant.binding, FrontierBinding::Detached(_))
5411 && !binding_terminal_exists(
5412 participant.participant_index,
5413 current_epoch,
5414 self.retained_records,
5415 self.causal_authorities,
5416 )
5417 {
5418 return Err(self.logical_owner_error());
5419 }
5420 Ok(())
5421 }
5422
5423 const fn logical_owner_error(&self) -> ClaimFrontierError {
5424 sequence_error(
5425 self.ledger.required_reserve(),
5426 ClaimFrontierInvalidReason::LogicalOwner,
5427 )
5428 }
5429}
5430
5431fn binding_terminal_exists(
5432 participant_index: ParticipantId,
5433 binding_epoch: BindingEpoch,
5434 retained_records: &[RetainedCausalRecord],
5435 historical_causal_authorities: &[HistoricalCausalAuthority],
5436) -> bool {
5437 retained_records.iter().any(|record| {
5438 matches!(
5439 record.kind,
5440 RetainedCausalRecordKind::BindingTerminal(owner)
5441 if owner.participant_index == participant_index
5442 && owner.binding_epoch == binding_epoch
5443 )
5444 }) || historical_causal_authorities.iter().any(|authority| {
5445 matches!(
5446 authority.kind,
5447 HistoricalCausalKind::BindingTerminal(owner)
5448 if owner.participant_index == participant_index
5449 && owner.binding_epoch == binding_epoch
5450 )
5451 })
5452}
5453
5454fn validate_marker_credit_owners(
5455 candidates: &[ImmutableSequenceCandidate],
5456 marker_records: &[RetainedCausalRecord],
5457 identity_slot_limit: u64,
5458 ledger: SequenceLedger,
5459) -> Result<(), ClaimFrontierError> {
5460 let mut owners = Vec::new();
5461 for record in marker_records {
5462 let RetainedCausalRecordKind::CompactionMarker {
5463 participant_index, ..
5464 } = record.kind
5465 else {
5466 continue;
5467 };
5468 if owners.contains(&participant_index) {
5469 return Err(sequence_error(
5470 ledger.required_reserve(),
5471 ClaimFrontierInvalidReason::LogicalOwner,
5472 ));
5473 }
5474 owners.push(participant_index);
5475 }
5476 for candidate in candidates {
5477 let ImmutableSequenceCandidate::Marker(marker) = candidate else {
5478 continue;
5479 };
5480 let participant_index = marker.admission_order.participant_index();
5481 if owners.contains(&participant_index) {
5482 return Err(sequence_error(
5483 sequence_ordinal(ledger, marker.delivery_seq),
5484 ClaimFrontierInvalidReason::LogicalOwner,
5485 ));
5486 }
5487 owners.push(participant_index);
5488 }
5489 if usize_to_u128(owners.len()) > u128::from(identity_slot_limit) {
5490 return Err(sequence_error(
5491 ledger.required_reserve(),
5492 ClaimFrontierInvalidReason::LogicalOwner,
5493 ));
5494 }
5495 Ok(())
5496}
5497
5498fn validated_historical_authorities(
5499 facts: Vec<HistoricalCausalFactRestore>,
5500 conversation_id: ConversationId,
5501 identity_slot_limit: u64,
5502 ledger: SequenceLedger,
5503 history: &ValidatedConversationHistory,
5504) -> Result<Vec<HistoricalCausalAuthority>, ClaimFrontierError> {
5505 if usize_to_u128(facts.len()) > u128::from(identity_slot_limit).saturating_mul(2) {
5506 return Err(sequence_error(
5507 ledger.required_reserve(),
5508 ClaimFrontierInvalidReason::LogicalOwner,
5509 ));
5510 }
5511 let authorities: Vec<_> = facts
5512 .into_iter()
5513 .map(HistoricalCausalAuthority::from_restore)
5514 .collect();
5515 let mut seen = Vec::new();
5516 for authority in &authorities {
5517 let (participant_index, phase) = match authority.kind {
5518 HistoricalCausalKind::BindingTerminal(owner) => {
5519 (owner.participant_index, CandidatePhase::BindingTerminal)
5520 }
5521 HistoricalCausalKind::MembershipExit(participant_index) => {
5522 (participant_index, CandidatePhase::MembershipExit)
5523 }
5524 };
5525 if authority.conversation_id != conversation_id
5526 || participant_index >= identity_slot_limit
5527 || authority.admission_order.participant_index() != participant_index
5528 || authority.admission_order.candidate_phase() != phase
5529 || seen.contains(authority)
5530 || !history.causal_authorities.contains(authority)
5531 {
5532 return Err(sequence_error(
5533 ledger.required_reserve(),
5534 ClaimFrontierInvalidReason::LogicalOwner,
5535 ));
5536 }
5537 seen.push(*authority);
5538 }
5539 Ok(authorities)
5540}
5541
5542const fn corrupt_frontier(error: ClaimFrontierError) -> ParticipantStateCorruptReason {
5543 ParticipantStateCorruptReason::ClaimFrontierInvalid {
5544 counter: match error.counter {
5545 ClaimFrontierCounter::DeliverySequence => ClaimCounter::DeliverySeq,
5546 ClaimFrontierCounter::TransactionOrder => ClaimCounter::TransactionOrder,
5547 },
5548 first_bad_position: error.first_bad_position,
5549 }
5550}
5551
5552fn sequence_segments(restore: &SequenceClaimFrontierRestore) -> Vec<NumericSegment<SequenceClass>> {
5553 let mut segments = Vec::new();
5554 for claim in &restore.movable_claims {
5555 segments.push(NumericSegment {
5556 start: u128::from(claim.delivery_seq),
5557 length: 1,
5558 class: Some(match claim.owner {
5559 SequenceDirectOwner::MembershipExit { .. } => SequenceClass::Exit,
5560 SequenceDirectOwner::BindingTerminal(_) => SequenceClass::Terminal,
5561 }),
5562 immutable: false,
5563 });
5564 }
5565 for candidate in &restore.immutable_candidates {
5566 segments.push(NumericSegment {
5567 start: u128::from(candidate.delivery_seq()),
5568 length: 1,
5569 class: Some(sequence_candidate_class(*candidate)),
5570 immutable: true,
5571 });
5572 }
5573 for range in &restore.products.live_times_terminal {
5574 segments.push(NumericSegment {
5575 start: u128::from(range.start),
5576 length: u128::from(range.length),
5577 class: Some(SequenceClass::LiveTimesTerminal),
5578 immutable: false,
5579 });
5580 }
5581 if let Some(range) = restore.products.live_times_replacement_terminal {
5582 segments.push(NumericSegment {
5583 start: u128::from(range.start),
5584 length: u128::from(range.length),
5585 class: Some(SequenceClass::LiveTimesReplacementTerminal),
5586 immutable: false,
5587 });
5588 }
5589 for range in &restore.products.other_live_times_exit {
5590 segments.push(NumericSegment {
5591 start: u128::from(range.start),
5592 length: u128::from(range.length),
5593 class: Some(SequenceClass::OtherLiveTimesExit),
5594 immutable: false,
5595 });
5596 }
5597 if let Some(recovery) = restore.recovery {
5598 if let Some(terminal) = recovery.terminal {
5599 segments.push(NumericSegment {
5600 start: u128::from(terminal.delivery_seq),
5601 length: 1,
5602 class: Some(SequenceClass::Terminal),
5603 immutable: false,
5604 });
5605 }
5606 segments.push(NumericSegment {
5607 start: u128::from(recovery.recovery_attach_seq),
5608 length: 1,
5609 class: Some(SequenceClass::RecoveryAttach),
5610 immutable: false,
5611 });
5612 segments.push(NumericSegment {
5613 start: u128::from(recovery.replacement_terminal_seq),
5614 length: 1,
5615 class: Some(SequenceClass::RecoveryReplacementTerminal),
5616 immutable: false,
5617 });
5618 }
5619 segments
5620}
5621
5622fn order_segments(
5623 restore: &OrderClaimFrontierRestore,
5624 high: OrderHigh,
5625) -> Vec<NumericSegment<OrderClass>> {
5626 let mut segments = Vec::new();
5627 for claim in &restore.movable_claims {
5628 segments.push(NumericSegment {
5629 start: u128::from(claim.transaction_order),
5630 length: 1,
5631 class: Some(match claim.owner {
5632 OrderDirectOwner::ActiveBindingTerminal(_) => OrderClass::ActiveBindingTerminal,
5633 OrderDirectOwner::MembershipExit { .. } => OrderClass::MembershipExit,
5634 }),
5635 immutable: false,
5636 });
5637 }
5638 for candidate in restore
5639 .immutable_candidates
5640 .iter()
5641 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
5642 {
5643 segments.push(NumericSegment {
5644 start: u128::from(candidate.transaction_order),
5645 length: 1,
5646 class: None,
5647 immutable: true,
5648 });
5649 }
5650 if let Some(recovery) = restore.recovery {
5651 if let Some(active_binding) = recovery.active_binding {
5652 segments.push(NumericSegment {
5653 start: u128::from(active_binding.transaction_order),
5654 length: 1,
5655 class: Some(OrderClass::ActiveBindingTerminal),
5656 immutable: false,
5657 });
5658 }
5659 segments.push(NumericSegment {
5660 start: u128::from(recovery.recovery_operation_order),
5661 length: 1,
5662 class: Some(OrderClass::RecoveryOperation),
5663 immutable: false,
5664 });
5665 segments.push(NumericSegment {
5666 start: u128::from(recovery.replacement_terminal_order),
5667 length: 1,
5668 class: Some(OrderClass::RecoveryReplacementTerminal),
5669 immutable: false,
5670 });
5671 }
5672 segments
5673}
5674
5675fn restore_sequence_frontier(
5676 active: &ActiveIdentityRanks,
5677 restore: SequenceClaimFrontierRestore,
5678 retained_floor: u128,
5679 ledger: SequenceLedger,
5680 recovery_provenance: Option<RecoveryClaimProvenance>,
5681 retained_records: &[RetainedCausalRecord],
5682 historical_causal_authorities: &[HistoricalCausalAuthority],
5683) -> Result<SequenceClaimFrontier, ClaimFrontierError> {
5684 let mut segments = sequence_segments(&restore);
5685
5686 let expected_counts = sequence_expected_counts(ledger);
5687 validate_numeric_segments(
5688 ClaimFrontierCounter::DeliverySequence,
5689 u128::from(ledger.high_watermark()) + 1,
5690 ledger.required_reserve(),
5691 &mut segments,
5692 &expected_counts,
5693 )?;
5694 validate_sequence_recovery(
5695 active,
5696 restore.recovery,
5697 recovery_provenance,
5698 ledger,
5699 ledger.required_reserve(),
5700 )?;
5701 validate_sequence_candidates(
5702 active,
5703 &restore.immutable_candidates,
5704 retained_floor,
5705 retained_records,
5706 historical_causal_authorities,
5707 ledger,
5708 )?;
5709 let terminal_owners = validate_sequence_direct_owners(
5710 active,
5711 &restore.movable_claims,
5712 &restore.immutable_candidates,
5713 restore.recovery,
5714 ledger,
5715 )?;
5716 let products = validate_sequence_products(
5717 active,
5718 restore.products,
5719 &terminal_owners,
5720 restore.recovery,
5721 recovery_provenance,
5722 ledger,
5723 )?;
5724 let recovery = restore
5725 .recovery
5726 .zip(recovery_provenance)
5727 .map(|(value, provenance)| RecoverySequenceBlock {
5728 terminal: value.terminal,
5729 recovery_attach_seq: value.recovery_attach_seq,
5730 replacement_terminal_seq: value.replacement_terminal_seq,
5731 participant_index: provenance.participant_index,
5732 marker_delivery_seq: provenance.marker_delivery_seq,
5733 recovered_binding_epoch: provenance.prior_binding_epoch,
5734 });
5735
5736 let mut movable_claims = restore.movable_claims;
5737 movable_claims.sort_by_key(|claim| claim.delivery_seq);
5738 let mut immutable_candidates = restore.immutable_candidates;
5739 immutable_candidates.sort_by_key(|candidate| candidate.delivery_seq());
5740
5741 Ok(SequenceClaimFrontier {
5742 ledger,
5743 movable_claims,
5744 immutable_candidates,
5745 products,
5746 recovery,
5747 })
5748}
5749
5750fn restore_order_frontier(
5751 active: &ActiveIdentityRanks,
5752 restore: OrderClaimFrontierRestore,
5753 ledger: OrderLedger,
5754 recovery_provenance: Option<RecoveryClaimProvenance>,
5755) -> Result<OrderClaimFrontier, ClaimFrontierError> {
5756 let mut segments = order_segments(&restore, ledger.high());
5757
5758 let candidate_count = order_frontier_candidate_count(&restore, ledger.high());
5759 let expected_length = candidate_count + ledger.claims().total();
5760 let expected_counts = order_expected_counts(ledger);
5761 validate_numeric_segments(
5762 ClaimFrontierCounter::TransactionOrder,
5763 order_frontier_start(ledger.high()),
5764 expected_length,
5765 &mut segments,
5766 &expected_counts,
5767 )?;
5768 validate_order_recovery(
5769 active,
5770 restore.recovery,
5771 recovery_provenance,
5772 ledger,
5773 expected_length,
5774 )?;
5775 let immutable_candidates = validate_order_candidates(&restore.immutable_candidates, ledger)?;
5776 validate_order_direct_owners(active, &restore.movable_claims, restore.recovery, ledger)?;
5777 let recovery = restore
5778 .recovery
5779 .zip(recovery_provenance)
5780 .map(|(value, provenance)| RecoveryOrderBlock {
5781 active_binding: value.active_binding,
5782 recovery_operation_order: value.recovery_operation_order,
5783 replacement_terminal_order: value.replacement_terminal_order,
5784 participant_index: provenance.participant_index,
5785 marker_delivery_seq: provenance.marker_delivery_seq,
5786 recovered_binding_epoch: provenance.prior_binding_epoch,
5787 });
5788
5789 let mut movable_claims = restore.movable_claims;
5790 movable_claims.sort_by_key(|claim| claim.transaction_order);
5791
5792 Ok(OrderClaimFrontier {
5793 ledger,
5794 movable_claims,
5795 immutable_candidates,
5796 recovery,
5797 })
5798}
5799
5800fn validate_numeric_segments<C: Copy + Into<usize>>(
5801 counter: ClaimFrontierCounter,
5802 first_value: u128,
5803 expected_length: u128,
5804 segments: &mut [NumericSegment<C>],
5805 expected_counts: &[u128],
5806) -> Result<(), ClaimFrontierError> {
5807 segments.sort_by_key(|segment| segment.start);
5808 let mut events = numeric_events(counter, first_value, segments)?;
5809 let emitted = scan_numeric_events(counter, first_value, &mut events)?;
5810 if emitted != expected_length {
5811 return Err(frontier_error(
5812 counter,
5813 emitted.min(expected_length),
5814 ClaimFrontierInvalidReason::AggregateLedger,
5815 ));
5816 }
5817 validate_immutable_prefix(counter, first_value, segments)?;
5818 validate_segment_class_counts(counter, expected_length, segments, expected_counts)
5819}
5820
5821fn numeric_events<C>(
5822 counter: ClaimFrontierCounter,
5823 first_value: u128,
5824 segments: &[NumericSegment<C>],
5825) -> Result<Vec<(u128, i8)>, ClaimFrontierError> {
5826 let mut events = Vec::new();
5827 let counter_limit = u128::from(u64::MAX) + 1;
5828 for segment in segments {
5829 if segment.length == 0 {
5830 continue;
5831 }
5832 let Some(end) = segment.start.checked_add(segment.length) else {
5833 return Err(frontier_error(
5834 counter,
5835 counter_limit.saturating_sub(first_value),
5836 ClaimFrontierInvalidReason::NumericPosition,
5837 ));
5838 };
5839 if segment.start < first_value {
5840 return Err(frontier_error(
5841 counter,
5842 0,
5843 ClaimFrontierInvalidReason::NumericPosition,
5844 ));
5845 }
5846 if end > counter_limit {
5847 return Err(frontier_error(
5848 counter,
5849 counter_limit.saturating_sub(first_value),
5850 ClaimFrontierInvalidReason::NumericPosition,
5851 ));
5852 }
5853 events.push((segment.start, 1_i8));
5854 events.push((end, -1_i8));
5855 }
5856 Ok(events)
5857}
5858
5859fn scan_numeric_events(
5860 counter: ClaimFrontierCounter,
5861 first_value: u128,
5862 events: &mut [(u128, i8)],
5863) -> Result<u128, ClaimFrontierError> {
5864 events.sort_unstable_by_key(|event| event.0);
5865 let mut event_index = 0_usize;
5866 let mut coordinate = first_value;
5867 let mut coverage = 0_i128;
5868 let mut emitted = 0_u128;
5869 while let Some((event_coordinate, _)) = events.get(event_index).copied() {
5870 if event_coordinate > coordinate {
5871 if coverage == 0 {
5872 return Err(frontier_error(
5873 counter,
5874 emitted,
5875 ClaimFrontierInvalidReason::NumericPosition,
5876 ));
5877 }
5878 if coverage > 1 {
5879 return Err(frontier_error(
5880 counter,
5881 emitted.saturating_add(1),
5882 ClaimFrontierInvalidReason::NumericPosition,
5883 ));
5884 }
5885 emitted = emitted.saturating_add(event_coordinate - coordinate);
5886 coordinate = event_coordinate;
5887 }
5888 while let Some((same_coordinate, delta)) = events.get(event_index).copied() {
5889 if same_coordinate != coordinate {
5890 break;
5891 }
5892 coverage += i128::from(delta);
5893 event_index += 1;
5894 }
5895 }
5896 if coverage != 0 {
5897 return Err(frontier_error(
5898 counter,
5899 emitted,
5900 ClaimFrontierInvalidReason::NumericPosition,
5901 ));
5902 }
5903 Ok(emitted)
5904}
5905
5906fn validate_immutable_prefix<C>(
5907 counter: ClaimFrontierCounter,
5908 first_value: u128,
5909 segments: &[NumericSegment<C>],
5910) -> Result<(), ClaimFrontierError> {
5911 let mut first_movable = None;
5912 for segment in segments.iter().filter(|segment| segment.length != 0) {
5913 if segment.immutable {
5914 if let Some(first_movable) = first_movable {
5915 return Err(frontier_error(
5916 counter,
5917 first_movable,
5918 ClaimFrontierInvalidReason::NumericPosition,
5919 ));
5920 }
5921 } else if first_movable.is_none() {
5922 first_movable = Some(segment.start - first_value);
5923 }
5924 }
5925 Ok(())
5926}
5927
5928fn validate_segment_class_counts<C: Copy + Into<usize>>(
5929 counter: ClaimFrontierCounter,
5930 expected_length: u128,
5931 segments: &[NumericSegment<C>],
5932 expected_counts: &[u128],
5933) -> Result<(), ClaimFrontierError> {
5934 let mut actual_counts = core::iter::repeat_n(0_u128, expected_counts.len()).collect::<Vec<_>>();
5935 let mut class_ordinal = 0_u128;
5936 for segment in segments {
5937 if segment.length == 0 {
5938 continue;
5939 }
5940 if let Some(class) = segment.class {
5941 let index = class.into();
5942 let prior = actual_counts[index];
5943 let Some(resulting) = prior.checked_add(segment.length) else {
5944 return Err(frontier_error(
5945 counter,
5946 class_ordinal,
5947 ClaimFrontierInvalidReason::AggregateLedger,
5948 ));
5949 };
5950 if resulting > expected_counts[index] {
5951 return Err(frontier_error(
5952 counter,
5953 class_ordinal + expected_counts[index].saturating_sub(prior),
5954 ClaimFrontierInvalidReason::AggregateLedger,
5955 ));
5956 }
5957 actual_counts[index] = resulting;
5958 }
5959 class_ordinal += segment.length;
5960 }
5961 if actual_counts != expected_counts {
5962 return Err(frontier_error(
5963 counter,
5964 expected_length,
5965 ClaimFrontierInvalidReason::AggregateLedger,
5966 ));
5967 }
5968 Ok(())
5969}
5970
5971#[cfg(test)]
5972pub(super) fn validate_numeric_union_for_test(
5973 first_value: u128,
5974 expected_length: u128,
5975 ranges: &[(u128, u128)],
5976) -> Result<(), ClaimFrontierError> {
5977 let mut segments: Vec<_> = ranges
5978 .iter()
5979 .map(|(start, length)| NumericSegment {
5980 start: *start,
5981 length: *length,
5982 class: Some(SequenceClass::Exit),
5983 immutable: false,
5984 })
5985 .collect();
5986 validate_numeric_segments(
5987 ClaimFrontierCounter::DeliverySequence,
5988 first_value,
5989 expected_length,
5990 &mut segments,
5991 &[expected_length, 0, 0, 0, 0, 0, 0, 0],
5992 )
5993}
5994
5995impl From<SequenceClass> for usize {
5996 fn from(value: SequenceClass) -> Self {
5997 value as Self
5998 }
5999}
6000
6001impl From<OrderClass> for usize {
6002 fn from(value: OrderClass) -> Self {
6003 value as Self
6004 }
6005}
6006
6007const fn sequence_candidate_class(candidate: ImmutableSequenceCandidate) -> SequenceClass {
6008 match candidate {
6009 ImmutableSequenceCandidate::BindingTerminal { .. } => SequenceClass::Terminal,
6010 ImmutableSequenceCandidate::Marker(marker) => match marker.current_owner {
6011 MarkerSequenceOwner::Marker => SequenceClass::Marker,
6012 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::LiveTimesTerminal) => {
6013 SequenceClass::LiveTimesTerminal
6014 }
6015 MarkerSequenceOwner::ConditionalProduct(
6016 SequenceProductClass::LiveTimesReplacementTerminal,
6017 ) => SequenceClass::LiveTimesReplacementTerminal,
6018 MarkerSequenceOwner::ConditionalProduct(SequenceProductClass::OtherLiveTimesExit) => {
6019 SequenceClass::OtherLiveTimesExit
6020 }
6021 },
6022 }
6023}
6024
6025fn sequence_expected_counts(ledger: SequenceLedger) -> [u128; 8] {
6026 let budget = ledger.budget();
6027 [
6028 u128::from(budget.e),
6029 u128::from(budget.t),
6030 u128::from(budget.m),
6031 u128::from(budget.rs),
6032 u128::from(budget.rt),
6033 budget.l_times_t,
6034 budget.l_times_rt,
6035 budget.l_other_times_e,
6036 ]
6037}
6038
6039fn order_expected_counts(ledger: OrderLedger) -> [u128; 4] {
6040 let claims = ledger.claims();
6041 [
6042 u128::from(claims.active_binding_terminals()),
6043 u128::from(claims.membership_exits()),
6044 u128::from(claims.recovery_operation()),
6045 u128::from(claims.recovery_replacement_terminal()),
6046 ]
6047}
6048
6049fn validate_sequence_recovery(
6050 active: &ActiveIdentityRanks,
6051 recovery: Option<RecoverySequenceBlockRestore>,
6052 provenance: Option<RecoveryClaimProvenance>,
6053 ledger: SequenceLedger,
6054 frontier_length: u128,
6055) -> Result<(), ClaimFrontierError> {
6056 let expected = ledger.claims().recovery();
6057 match (expected, recovery, provenance) {
6058 (RecoverySequenceReserve::None, None, None) => Ok(()),
6059 (RecoverySequenceReserve::DetachedCredentialRecovery, None, _)
6060 | (RecoverySequenceReserve::DetachedCredentialRecovery, Some(_), None)
6061 | (RecoverySequenceReserve::None, None, Some(_)) => Err(sequence_error(
6062 frontier_length,
6063 ClaimFrontierInvalidReason::RecoveryBlock,
6064 )),
6065 (RecoverySequenceReserve::None, Some(block), _) => Err(sequence_error(
6066 sequence_ordinal(ledger, block_start_sequence(block)),
6067 ClaimFrontierInvalidReason::RecoveryBlock,
6068 )),
6069 (RecoverySequenceReserve::DetachedCredentialRecovery, Some(block), Some(provenance)) => {
6070 let block_ordinal = sequence_ordinal(ledger, block_start_sequence(block));
6071 let expected_recovery_attach = block
6072 .terminal
6073 .map_or(Some(block.recovery_attach_seq), |terminal| {
6074 terminal.delivery_seq.checked_add(1)
6075 });
6076 if expected_recovery_attach != Some(block.recovery_attach_seq) {
6077 return Err(sequence_error(
6078 block_ordinal + 1,
6079 ClaimFrontierInvalidReason::RecoveryBlock,
6080 ));
6081 }
6082 if block.recovery_attach_seq.checked_add(1) != Some(block.replacement_terminal_seq) {
6083 return Err(sequence_error(
6084 block_ordinal + u128::from(block.terminal.is_some()) + 1,
6085 ClaimFrontierInvalidReason::RecoveryBlock,
6086 ));
6087 }
6088 let Some(participant) = active_participant(active, provenance.participant_index) else {
6089 return Err(sequence_error(
6090 block_ordinal,
6091 ClaimFrontierInvalidReason::LogicalOwner,
6092 ));
6093 };
6094 let expected_binding = match provenance.phase {
6095 RecoveryClaimPhase::PreFate => {
6096 FrontierBinding::Bound(provenance.prior_binding_epoch)
6097 }
6098 RecoveryClaimPhase::PostFate => {
6099 FrontierBinding::Detached(provenance.prior_binding_epoch)
6100 }
6101 RecoveryClaimPhase::RecoveredBound => {
6102 FrontierBinding::Bound(provenance.current_binding_epoch)
6103 }
6104 };
6105 if participant.binding != expected_binding {
6106 return Err(sequence_error(
6107 block_ordinal,
6108 ClaimFrontierInvalidReason::LogicalOwner,
6109 ));
6110 }
6111 let terminal_valid = match (provenance.phase, block.terminal) {
6112 (RecoveryClaimPhase::PreFate, Some(terminal)) => {
6113 terminal.owner.participant_index == provenance.participant_index
6114 && terminal.owner.binding_epoch == provenance.prior_binding_epoch
6115 }
6116 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
6117 _ => false,
6118 };
6119 if !terminal_valid {
6120 return Err(sequence_error(
6121 block_ordinal,
6122 ClaimFrontierInvalidReason::RecoveryBlock,
6123 ));
6124 }
6125 Ok(())
6126 }
6127 }
6128}
6129
6130fn validate_sequence_candidates(
6131 active: &ActiveIdentityRanks,
6132 candidates: &[ImmutableSequenceCandidate],
6133 retained_floor: u128,
6134 retained_records: &[RetainedCausalRecord],
6135 historical_causal_authorities: &[HistoricalCausalAuthority],
6136 ledger: SequenceLedger,
6137) -> Result<(), ClaimFrontierError> {
6138 let mut seen_keys = Vec::new();
6139 let mut previous_sequence = None;
6140 let mut previous_order = retained_records.last().map(|record| record.admission_order);
6141 for candidate in candidates {
6142 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
6143 let order = candidate.admission_order();
6144 if previous_sequence.is_some_and(|previous| previous >= candidate.delivery_seq())
6145 || previous_order.is_some_and(|previous| previous >= order)
6146 || seen_keys.contains(&order)
6147 {
6148 return Err(sequence_error(
6149 ordinal,
6150 ClaimFrontierInvalidReason::CandidateKey,
6151 ));
6152 }
6153 previous_sequence = Some(candidate.delivery_seq());
6154 previous_order = Some(order);
6155 seen_keys.push(order);
6156 match candidate {
6157 ImmutableSequenceCandidate::BindingTerminal { owner, .. } => {
6158 if order.candidate_phase() != CandidatePhase::BindingTerminal
6159 || order.participant_index() != owner.participant_index
6160 || !terminal_matches_active(active, *owner)
6161 {
6162 return Err(sequence_error(
6163 ordinal,
6164 ClaimFrontierInvalidReason::CandidateKey,
6165 ));
6166 }
6167 }
6168 ImmutableSequenceCandidate::Marker(marker) => {
6169 let Some(participant) = active_participant(active, order.participant_index())
6170 else {
6171 return Err(sequence_error(
6172 ordinal,
6173 ClaimFrontierInvalidReason::LogicalOwner,
6174 ));
6175 };
6176 if order.candidate_phase() != CandidatePhase::CompactionMarker
6177 || marker.current_owner != MarkerSequenceOwner::Marker
6178 || marker.target_binding != participant.binding
6179 || marker.abandoned_after != participant.cursor
6180 || marker.abandoned_after > marker.abandoned_through
6181 || u128::from(marker.physical_floor_at_decision) != retained_floor
6182 || u128::from(marker.physical_floor_at_decision)
6183 > u128::from(marker.abandoned_through) + 1
6184 || marker.abandoned_through >= marker.delivery_seq
6185 || !marker_provenance_targets(marker.provenance, order.participant_index())
6186 || !marker_has_causal_authority(
6187 *marker,
6188 retained_records,
6189 historical_causal_authorities,
6190 )
6191 {
6192 return Err(sequence_error(
6193 ordinal,
6194 ClaimFrontierInvalidReason::CandidateKey,
6195 ));
6196 }
6197 }
6198 }
6199 }
6200 Ok(())
6201}
6202
6203#[derive(Clone, Copy, Debug, PartialEq, Eq)]
6204enum TerminalOccurrenceKind {
6205 Movable,
6206 Candidate,
6207}
6208
6209#[derive(Clone, Copy, Debug, PartialEq, Eq)]
6210struct TerminalOccurrence {
6211 owner: BindingTerminalOwner,
6212 ordinal: u128,
6213 kind: TerminalOccurrenceKind,
6214}
6215
6216fn validate_sequence_direct_owners(
6217 active: &ActiveIdentityRanks,
6218 movable: &[MovableSequenceClaim],
6219 candidates: &[ImmutableSequenceCandidate],
6220 recovery: Option<RecoverySequenceBlockRestore>,
6221 ledger: SequenceLedger,
6222) -> Result<Vec<BindingTerminalOwner>, ClaimFrontierError> {
6223 let (mut exit_owners, terminal_occurrences) =
6224 collect_sequence_direct_owners(active, movable, candidates, recovery, ledger)?;
6225 validate_sequence_exit_owners(active, &mut exit_owners, ledger)?;
6226 validate_sequence_terminal_owners(active, &terminal_occurrences, ledger)?;
6227 let mut owners: Vec<_> = terminal_occurrences
6228 .into_iter()
6229 .map(|occurrence| occurrence.owner)
6230 .collect();
6231 owners.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6232 Ok(owners)
6233}
6234
6235fn collect_sequence_direct_owners(
6236 active: &ActiveIdentityRanks,
6237 movable: &[MovableSequenceClaim],
6238 candidates: &[ImmutableSequenceCandidate],
6239 recovery: Option<RecoverySequenceBlockRestore>,
6240 ledger: SequenceLedger,
6241) -> Result<(Vec<ParticipantId>, Vec<TerminalOccurrence>), ClaimFrontierError> {
6242 let mut ordered_movable = movable.to_vec();
6243 ordered_movable.sort_by_key(|claim| claim.delivery_seq);
6244 let mut exit_owners = Vec::new();
6245 let mut terminal_occurrences = Vec::new();
6246 for claim in ordered_movable {
6247 let ordinal = sequence_ordinal(ledger, claim.delivery_seq);
6248 match claim.owner {
6249 SequenceDirectOwner::MembershipExit { participant_index } => {
6250 if !active.contains(participant_index) || exit_owners.contains(&participant_index) {
6251 return Err(sequence_error(
6252 ordinal,
6253 ClaimFrontierInvalidReason::LogicalOwner,
6254 ));
6255 }
6256 exit_owners.push(participant_index);
6257 }
6258 SequenceDirectOwner::BindingTerminal(owner) => {
6259 if !terminal_matches_bound(active, owner) {
6260 return Err(sequence_error(
6261 ordinal,
6262 ClaimFrontierInvalidReason::LogicalOwner,
6263 ));
6264 }
6265 push_terminal_occurrence(
6266 &mut terminal_occurrences,
6267 owner,
6268 ordinal,
6269 TerminalOccurrenceKind::Movable,
6270 )?;
6271 }
6272 }
6273 }
6274 for candidate in candidates {
6275 if let ImmutableSequenceCandidate::BindingTerminal { owner, .. } = candidate {
6276 let ordinal = sequence_ordinal(ledger, candidate.delivery_seq());
6277 push_terminal_occurrence(
6278 &mut terminal_occurrences,
6279 *owner,
6280 ordinal,
6281 TerminalOccurrenceKind::Candidate,
6282 )?;
6283 }
6284 }
6285 if let Some(terminal) = recovery.and_then(|block| block.terminal) {
6286 let ordinal = sequence_ordinal(ledger, terminal.delivery_seq);
6287 push_terminal_occurrence(
6288 &mut terminal_occurrences,
6289 terminal.owner,
6290 ordinal,
6291 TerminalOccurrenceKind::Movable,
6292 )?;
6293 }
6294 Ok((exit_owners, terminal_occurrences))
6295}
6296
6297fn push_terminal_occurrence(
6298 occurrences: &mut Vec<TerminalOccurrence>,
6299 owner: BindingTerminalOwner,
6300 ordinal: u128,
6301 kind: TerminalOccurrenceKind,
6302) -> Result<(), ClaimFrontierError> {
6303 if occurrences
6304 .iter()
6305 .any(|occurrence| occurrence.owner == owner)
6306 {
6307 return Err(sequence_error(
6308 ordinal,
6309 ClaimFrontierInvalidReason::LogicalOwner,
6310 ));
6311 }
6312 occurrences.push(TerminalOccurrence {
6313 owner,
6314 ordinal,
6315 kind,
6316 });
6317 Ok(())
6318}
6319
6320fn validate_sequence_exit_owners(
6321 active: &ActiveIdentityRanks,
6322 exit_owners: &mut [ParticipantId],
6323 ledger: SequenceLedger,
6324) -> Result<(), ClaimFrontierError> {
6325 exit_owners.sort_unstable();
6326 if exit_owners.len() != active.participants.len()
6327 || !exit_owners.iter().copied().eq(active
6328 .participants
6329 .iter()
6330 .map(|participant| participant.participant_index))
6331 {
6332 return Err(sequence_error(
6333 ledger.required_reserve(),
6334 ClaimFrontierInvalidReason::LogicalOwner,
6335 ));
6336 }
6337 Ok(())
6338}
6339
6340fn validate_sequence_terminal_owners(
6341 active: &ActiveIdentityRanks,
6342 terminal_occurrences: &[TerminalOccurrence],
6343 ledger: SequenceLedger,
6344) -> Result<(), ClaimFrontierError> {
6345 for participant in &active.participants {
6346 let matching: Vec<_> = terminal_occurrences
6347 .iter()
6348 .filter(|occurrence| {
6349 occurrence.owner.participant_index == participant.participant_index
6350 })
6351 .copied()
6352 .collect();
6353 match participant.binding {
6354 FrontierBinding::Bound(epoch) => {
6355 if !matches!(matching.as_slice(), [occurrence] if occurrence.owner.binding_epoch == epoch)
6356 {
6357 return Err(sequence_error(
6358 matching.first().map_or_else(
6359 || ledger.required_reserve(),
6360 |occurrence| occurrence.ordinal,
6361 ),
6362 ClaimFrontierInvalidReason::LogicalOwner,
6363 ));
6364 }
6365 }
6366 FrontierBinding::Detached(epoch) => {
6367 if matching.len() > 1
6368 || matching.first().is_some_and(|occurrence| {
6369 occurrence.owner.binding_epoch != epoch
6370 || occurrence.kind != TerminalOccurrenceKind::Candidate
6371 })
6372 {
6373 return Err(sequence_error(
6374 matching.first().map_or_else(
6375 || ledger.required_reserve(),
6376 |occurrence| occurrence.ordinal,
6377 ),
6378 ClaimFrontierInvalidReason::LogicalOwner,
6379 ));
6380 }
6381 }
6382 }
6383 }
6384 Ok(())
6385}
6386
6387fn validate_sequence_products(
6388 active: &ActiveIdentityRanks,
6389 restore: SequenceProductRangesRestore,
6390 terminal_owners: &[BindingTerminalOwner],
6391 recovery: Option<RecoverySequenceBlockRestore>,
6392 recovery_provenance: Option<RecoveryClaimProvenance>,
6393 ledger: SequenceLedger,
6394) -> Result<SequenceProductRanges, ClaimFrontierError> {
6395 let live_count = usize_to_u64(active.participants.len());
6396 let other_count = live_count.saturating_sub(1);
6397 let live_times_terminal = validate_terminal_product_ranges(
6398 restore.live_times_terminal,
6399 terminal_owners,
6400 live_count,
6401 ledger,
6402 )?;
6403 let live_times_replacement_terminal = validate_replacement_product_range(
6404 restore.live_times_replacement_terminal,
6405 recovery,
6406 recovery_provenance,
6407 live_count,
6408 ledger,
6409 )?;
6410 let other_live_times_exit =
6411 validate_exit_product_ranges(active, restore.other_live_times_exit, other_count, ledger)?;
6412 Ok(SequenceProductRanges {
6413 live_times_terminal,
6414 live_times_replacement_terminal,
6415 other_live_times_exit,
6416 })
6417}
6418
6419fn validate_terminal_product_ranges(
6420 mut ranges: Vec<TerminalProductRangeRestore>,
6421 terminal_owners: &[BindingTerminalOwner],
6422 live_count: u64,
6423 ledger: SequenceLedger,
6424) -> Result<Vec<TerminalProductRange>, ClaimFrontierError> {
6425 ranges.sort_by_key(|range| range.start);
6426 let mut seen_terminals = Vec::new();
6427 let mut live_times_terminal = Vec::new();
6428 for range in ranges {
6429 let ordinal = sequence_ordinal(ledger, range.start);
6430 if range.length != live_count
6431 || !terminal_owners.contains(&range.terminal)
6432 || seen_terminals.contains(&range.terminal)
6433 {
6434 return Err(sequence_error(
6435 ordinal,
6436 ClaimFrontierInvalidReason::ProductRange,
6437 ));
6438 }
6439 seen_terminals.push(range.terminal);
6440 live_times_terminal.push(TerminalProductRange {
6441 start: range.start,
6442 length: range.length,
6443 terminal: range.terminal,
6444 });
6445 }
6446 if seen_terminals.len() != terminal_owners.len() {
6447 return Err(sequence_error(
6448 ledger.required_reserve(),
6449 ClaimFrontierInvalidReason::ProductRange,
6450 ));
6451 }
6452 Ok(live_times_terminal)
6453}
6454
6455fn validate_replacement_product_range(
6456 range: Option<ReplacementTerminalProductRangeRestore>,
6457 recovery: Option<RecoverySequenceBlockRestore>,
6458 recovery_provenance: Option<RecoveryClaimProvenance>,
6459 live_count: u64,
6460 ledger: SequenceLedger,
6461) -> Result<Option<ReplacementTerminalProductRange>, ClaimFrontierError> {
6462 let validated = match (range, recovery, recovery_provenance) {
6463 (None, None, None) => None,
6464 (Some(range), Some(_), Some(provenance)) if range.length == live_count => {
6465 Some(ReplacementTerminalProductRange {
6466 start: range.start,
6467 length: range.length,
6468 participant_index: provenance.participant_index,
6469 marker_delivery_seq: provenance.marker_delivery_seq,
6470 prior_binding_epoch: provenance.prior_binding_epoch,
6471 })
6472 }
6473 (Some(range), _, _) => {
6474 return Err(sequence_error(
6475 sequence_ordinal(ledger, range.start),
6476 ClaimFrontierInvalidReason::ProductRange,
6477 ));
6478 }
6479 (None, Some(_), _) | (None, None, Some(_)) => {
6480 return Err(sequence_error(
6481 ledger.required_reserve(),
6482 ClaimFrontierInvalidReason::ProductRange,
6483 ));
6484 }
6485 };
6486 Ok(validated)
6487}
6488
6489fn validate_exit_product_ranges(
6490 active: &ActiveIdentityRanks,
6491 mut ranges: Vec<ExitProductRangeRestore>,
6492 other_count: u64,
6493 ledger: SequenceLedger,
6494) -> Result<Vec<ExitProductRange>, ClaimFrontierError> {
6495 ranges.sort_by_key(|range| range.start);
6496 let mut seen_exits = Vec::new();
6497 let mut other_live_times_exit = Vec::new();
6498 if other_count == 0 && !ranges.is_empty() {
6499 return Err(sequence_error(
6500 ledger.required_reserve(),
6501 ClaimFrontierInvalidReason::ProductRange,
6502 ));
6503 }
6504 for range in ranges {
6505 let ordinal = sequence_ordinal(ledger, range.start);
6506 if range.length != other_count
6507 || !active.contains(range.exit_participant)
6508 || seen_exits.contains(&range.exit_participant)
6509 {
6510 return Err(sequence_error(
6511 ordinal,
6512 ClaimFrontierInvalidReason::ProductRange,
6513 ));
6514 }
6515 seen_exits.push(range.exit_participant);
6516 other_live_times_exit.push(ExitProductRange {
6517 start: range.start,
6518 length: range.length,
6519 exit_participant: range.exit_participant,
6520 });
6521 }
6522 seen_exits.sort_unstable();
6523 let expected_exit_ranges = if other_count == 0 {
6524 0
6525 } else {
6526 active.participants.len()
6527 };
6528 if seen_exits.len() != expected_exit_ranges
6529 || !seen_exits.iter().copied().eq(active
6530 .participants
6531 .iter()
6532 .take(expected_exit_ranges)
6533 .map(|participant| participant.participant_index))
6534 {
6535 return Err(sequence_error(
6536 ledger.required_reserve(),
6537 ClaimFrontierInvalidReason::ProductRange,
6538 ));
6539 }
6540 Ok(other_live_times_exit)
6541}
6542
6543fn validate_order_recovery(
6544 active: &ActiveIdentityRanks,
6545 recovery: Option<RecoveryOrderBlockRestore>,
6546 provenance: Option<RecoveryClaimProvenance>,
6547 ledger: OrderLedger,
6548 frontier_length: u128,
6549) -> Result<(), ClaimFrontierError> {
6550 let claims = ledger.claims();
6551 let expected = claims.recovery_operation() && claims.recovery_replacement_terminal();
6552 match (expected, recovery, provenance) {
6553 (false, None, None) => Ok(()),
6554 (true, None, _) | (true, Some(_), None) | (false, None, Some(_)) => Err(order_error(
6555 frontier_length,
6556 ClaimFrontierInvalidReason::RecoveryBlock,
6557 )),
6558 (false, Some(block), _) => Err(order_error(
6559 order_ordinal(ledger, block_start_order(block)),
6560 ClaimFrontierInvalidReason::RecoveryBlock,
6561 )),
6562 (true, Some(block), Some(provenance)) => {
6563 let block_ordinal = order_ordinal(ledger, block_start_order(block));
6564 let expected_recovery_operation = block
6565 .active_binding
6566 .map_or(Some(block.recovery_operation_order), |active_binding| {
6567 active_binding.transaction_order.checked_add(1)
6568 });
6569 if expected_recovery_operation != Some(block.recovery_operation_order) {
6570 return Err(order_error(
6571 block_ordinal + 1,
6572 ClaimFrontierInvalidReason::RecoveryBlock,
6573 ));
6574 }
6575 if block.recovery_operation_order.checked_add(1)
6576 != Some(block.replacement_terminal_order)
6577 {
6578 return Err(order_error(
6579 block_ordinal + u128::from(block.active_binding.is_some()) + 1,
6580 ClaimFrontierInvalidReason::RecoveryBlock,
6581 ));
6582 }
6583 let Some(participant) = active_participant(active, provenance.participant_index) else {
6584 return Err(order_error(
6585 block_ordinal,
6586 ClaimFrontierInvalidReason::LogicalOwner,
6587 ));
6588 };
6589 let expected_binding = match provenance.phase {
6590 RecoveryClaimPhase::PreFate => {
6591 FrontierBinding::Bound(provenance.prior_binding_epoch)
6592 }
6593 RecoveryClaimPhase::PostFate => {
6594 FrontierBinding::Detached(provenance.prior_binding_epoch)
6595 }
6596 RecoveryClaimPhase::RecoveredBound => {
6597 FrontierBinding::Bound(provenance.current_binding_epoch)
6598 }
6599 };
6600 if participant.binding != expected_binding {
6601 return Err(order_error(
6602 block_ordinal,
6603 ClaimFrontierInvalidReason::LogicalOwner,
6604 ));
6605 }
6606 let active_binding_valid = match (provenance.phase, block.active_binding) {
6607 (RecoveryClaimPhase::PreFate, Some(active_binding)) => {
6608 active_binding.owner.participant_index == provenance.participant_index
6609 && active_binding.owner.binding_epoch == provenance.prior_binding_epoch
6610 }
6611 (RecoveryClaimPhase::PostFate | RecoveryClaimPhase::RecoveredBound, None) => true,
6612 _ => false,
6613 };
6614 if !active_binding_valid {
6615 return Err(order_error(
6616 block_ordinal,
6617 ClaimFrontierInvalidReason::RecoveryBlock,
6618 ));
6619 }
6620 Ok(())
6621 }
6622 }
6623}
6624
6625fn validate_order_candidates(
6626 restore: &[ImmutableOrderCandidateMajorRestore],
6627 ledger: OrderLedger,
6628) -> Result<Vec<ImmutableOrderCandidateMajor>, ClaimFrontierError> {
6629 let mut groups = restore.to_vec();
6630 groups.sort_by_key(|group| group.transaction_order);
6631 let mut seen_keys = Vec::new();
6632 let mut previous_major = None;
6633 let mut validated = Vec::new();
6634 for group in groups {
6635 let ordinal = order_ordinal(ledger, group.transaction_order);
6636 let below_allocated_high = matches!(
6637 ledger.high(),
6638 OrderHigh::Allocated(high) if group.transaction_order < high
6639 );
6640 if group.candidate_keys.is_empty()
6641 || previous_major == Some(group.transaction_order)
6642 || below_allocated_high
6643 {
6644 return Err(order_error(
6645 ordinal,
6646 ClaimFrontierInvalidReason::CandidateKey,
6647 ));
6648 }
6649 previous_major = Some(group.transaction_order);
6650 let mut previous = None;
6651 for key in &group.candidate_keys {
6652 if key.transaction_order() != group.transaction_order
6653 || previous.is_some_and(|previous| previous >= *key)
6654 || seen_keys.contains(key)
6655 {
6656 return Err(order_error(
6657 ordinal,
6658 ClaimFrontierInvalidReason::CandidateKey,
6659 ));
6660 }
6661 previous = Some(*key);
6662 seen_keys.push(*key);
6663 }
6664 validated.push(ImmutableOrderCandidateMajor {
6665 transaction_order: group.transaction_order,
6666 candidate_keys: group.candidate_keys,
6667 });
6668 }
6669 Ok(validated)
6670}
6671
6672fn validate_order_direct_owners(
6673 active: &ActiveIdentityRanks,
6674 movable: &[MovableOrderClaim],
6675 recovery: Option<RecoveryOrderBlockRestore>,
6676 ledger: OrderLedger,
6677) -> Result<(), ClaimFrontierError> {
6678 let mut ordered = movable.to_vec();
6679 ordered.sort_by_key(|claim| claim.transaction_order);
6680 let mut exits = Vec::new();
6681 let mut terminals = Vec::new();
6682 for claim in ordered {
6683 let ordinal = order_ordinal(ledger, claim.transaction_order);
6684 match claim.owner {
6685 OrderDirectOwner::MembershipExit { participant_index } => {
6686 if !active.contains(participant_index) || exits.contains(&participant_index) {
6687 return Err(order_error(
6688 ordinal,
6689 ClaimFrontierInvalidReason::LogicalOwner,
6690 ));
6691 }
6692 exits.push(participant_index);
6693 }
6694 OrderDirectOwner::ActiveBindingTerminal(owner) => {
6695 if !terminal_matches_bound(active, owner) || terminals.contains(&owner) {
6696 return Err(order_error(
6697 ordinal,
6698 ClaimFrontierInvalidReason::LogicalOwner,
6699 ));
6700 }
6701 terminals.push(owner);
6702 }
6703 }
6704 }
6705 if let Some(active_binding) = recovery.and_then(|block| block.active_binding) {
6706 let ordinal = order_ordinal(ledger, active_binding.transaction_order);
6707 if !terminal_matches_bound(active, active_binding.owner)
6708 || terminals.contains(&active_binding.owner)
6709 {
6710 return Err(order_error(
6711 ordinal,
6712 ClaimFrontierInvalidReason::LogicalOwner,
6713 ));
6714 }
6715 terminals.push(active_binding.owner);
6716 }
6717 exits.sort_unstable();
6718 if exits.len() != active.participants.len()
6719 || !exits.iter().copied().eq(active
6720 .participants
6721 .iter()
6722 .map(|participant| participant.participant_index))
6723 {
6724 return Err(order_error(
6725 ledger.claims().total(),
6726 ClaimFrontierInvalidReason::LogicalOwner,
6727 ));
6728 }
6729 terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6730 if usize_to_u128(terminals.len()) != u128::from(ledger.claims().active_binding_terminals()) {
6731 return Err(order_error(
6732 ledger.claims().total(),
6733 ClaimFrontierInvalidReason::LogicalOwner,
6734 ));
6735 }
6736 Ok(())
6737}
6738
6739fn validate_cross_counter(
6740 sequence: &SequenceClaimFrontier,
6741 order: &OrderClaimFrontier,
6742) -> Result<(), ClaimFrontierError> {
6743 match (sequence.recovery, order.recovery) {
6744 (None, None) => {}
6745 (Some(sequence_block), Some(order_block))
6746 if sequence_block.participant_index == order_block.participant_index
6747 && sequence_block.marker_delivery_seq == order_block.marker_delivery_seq
6748 && sequence_block.recovered_binding_epoch
6749 == order_block.recovered_binding_epoch
6750 && sequence_block.terminal.map(|terminal| terminal.owner)
6751 == order_block.active_binding.map(|active| active.owner) => {}
6752 (Some(sequence_block), _) => {
6753 return Err(sequence_error(
6754 sequence_ordinal(
6755 sequence.ledger,
6756 block_start_validated_sequence(sequence_block),
6757 ),
6758 ClaimFrontierInvalidReason::RecoveryBlock,
6759 ));
6760 }
6761 (None, Some(order_block)) => {
6762 return Err(order_error(
6763 order_ordinal(order.ledger, block_start_validated_order(order_block)),
6764 ClaimFrontierInvalidReason::RecoveryBlock,
6765 ));
6766 }
6767 }
6768
6769 let mut order_candidate_keys = Vec::new();
6770 for group in &order.immutable_candidates {
6771 order_candidate_keys.extend(group.candidate_keys.iter().copied());
6772 }
6773 for candidate in &sequence.immutable_candidates {
6774 let key = candidate.admission_order();
6775 if !order_candidate_keys.contains(&key) {
6776 return Err(sequence_error(
6777 sequence_ordinal(sequence.ledger, candidate.delivery_seq()),
6778 ClaimFrontierInvalidReason::CandidateKey,
6779 ));
6780 }
6781 }
6782 for group in &order.immutable_candidates {
6783 for key in &group.candidate_keys {
6784 if !sequence
6785 .immutable_candidates
6786 .iter()
6787 .any(|candidate| candidate.admission_order() == *key)
6788 {
6789 return Err(order_error(
6790 order_ordinal(order.ledger, group.transaction_order),
6791 ClaimFrontierInvalidReason::CandidateKey,
6792 ));
6793 }
6794 }
6795 }
6796
6797 let mut sequence_movable_terminals = Vec::new();
6798 for claim in &sequence.movable_claims {
6799 if let SequenceDirectOwner::BindingTerminal(owner) = claim.owner {
6800 sequence_movable_terminals.push(owner);
6801 }
6802 }
6803 if let Some(terminal) = sequence.recovery.and_then(|block| block.terminal) {
6804 sequence_movable_terminals.push(terminal.owner);
6805 }
6806 let mut order_movable_terminals = Vec::new();
6807 for claim in &order.movable_claims {
6808 if let OrderDirectOwner::ActiveBindingTerminal(owner) = claim.owner {
6809 order_movable_terminals.push(owner);
6810 }
6811 }
6812 if let Some(active_binding) = order.recovery.and_then(|block| block.active_binding) {
6813 order_movable_terminals.push(active_binding.owner);
6814 }
6815 sequence_movable_terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6816 order_movable_terminals.sort_by_key(|owner| (owner.participant_index, owner.binding_epoch));
6817 if sequence_movable_terminals != order_movable_terminals {
6818 return Err(sequence_error(
6819 sequence.ledger.required_reserve(),
6820 ClaimFrontierInvalidReason::LogicalOwner,
6821 ));
6822 }
6823 Ok(())
6824}
6825
6826const fn marker_provenance_targets(provenance: MarkerProvenance, target: ParticipantId) -> bool {
6827 match provenance {
6828 MarkerProvenance::NonProductM => true,
6829 MarkerProvenance::TerminalProduct {
6830 affected_participant,
6831 ..
6832 } => affected_participant == target,
6833 MarkerProvenance::ExitProduct {
6834 exit_participant,
6835 remaining_participant,
6836 } => exit_participant != remaining_participant && remaining_participant == target,
6837 }
6838}
6839
6840fn marker_has_causal_authority(
6841 marker: MarkerCandidateAuthority,
6842 records: &[RetainedCausalRecord],
6843 historical: &[HistoricalCausalAuthority],
6844) -> bool {
6845 if marker.provenance == MarkerProvenance::NonProductM {
6846 return true;
6847 }
6848 let retained_match = records.iter().any(|record| {
6849 if record.admission_order.transaction_order() != marker.admission_order.transaction_order()
6850 {
6851 return false;
6852 }
6853 match marker.provenance {
6854 MarkerProvenance::NonProductM => true,
6855 MarkerProvenance::TerminalProduct {
6856 terminal: TerminalProductSource::Binding(owner),
6857 ..
6858 } => matches!(
6859 record.kind,
6860 RetainedCausalRecordKind::BindingTerminal(actual) if actual == owner
6861 ),
6862 MarkerProvenance::TerminalProduct {
6863 terminal:
6864 TerminalProductSource::RecoveryReplacement {
6865 participant_index,
6866 binding_epoch,
6867 },
6868 ..
6869 } => matches!(
6870 record.kind,
6871 RetainedCausalRecordKind::BindingTerminal(owner)
6872 if owner.participant_index == participant_index
6873 && owner.binding_epoch == binding_epoch
6874 ),
6875 MarkerProvenance::ExitProduct {
6876 exit_participant, ..
6877 } => matches!(
6878 record.kind,
6879 RetainedCausalRecordKind::MembershipExit { participant_index }
6880 if participant_index == exit_participant
6881 ),
6882 }
6883 });
6884 retained_match
6885 || historical
6886 .iter()
6887 .any(|authority| match (marker.provenance, authority.kind) {
6888 (
6889 MarkerProvenance::TerminalProduct {
6890 terminal: TerminalProductSource::Binding(expected),
6891 ..
6892 },
6893 HistoricalCausalKind::BindingTerminal(owner),
6894 ) => {
6895 owner == expected
6896 && authority.admission_order.transaction_order()
6897 == marker.admission_order.transaction_order()
6898 }
6899 (
6900 MarkerProvenance::TerminalProduct {
6901 terminal:
6902 TerminalProductSource::RecoveryReplacement {
6903 participant_index,
6904 binding_epoch,
6905 },
6906 ..
6907 },
6908 HistoricalCausalKind::BindingTerminal(owner),
6909 ) => {
6910 owner.participant_index == participant_index
6911 && owner.binding_epoch == binding_epoch
6912 && authority.admission_order.transaction_order()
6913 == marker.admission_order.transaction_order()
6914 }
6915 (
6916 MarkerProvenance::ExitProduct {
6917 exit_participant, ..
6918 },
6919 HistoricalCausalKind::MembershipExit(participant_index),
6920 ) => {
6921 participant_index == exit_participant
6922 && authority.admission_order.transaction_order()
6923 == marker.admission_order.transaction_order()
6924 }
6925 _ => false,
6926 })
6927}
6928
6929fn terminal_matches_active(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
6930 active_participant(active, owner.participant_index)
6931 .is_some_and(|participant| binding_epoch(participant.binding) == owner.binding_epoch)
6932}
6933
6934fn terminal_matches_bound(active: &ActiveIdentityRanks, owner: BindingTerminalOwner) -> bool {
6935 active_participant(active, owner.participant_index).is_some_and(|participant| {
6936 participant.binding == FrontierBinding::Bound(owner.binding_epoch)
6937 })
6938}
6939
6940fn active_participant(
6941 active: &ActiveIdentityRanks,
6942 participant_index: ParticipantId,
6943) -> Option<FrontierParticipant> {
6944 active
6945 .participants
6946 .binary_search_by_key(&participant_index, |participant| {
6947 participant.participant_index
6948 })
6949 .ok()
6950 .and_then(|index| active.participants.get(index))
6951 .copied()
6952}
6953
6954const fn binding_epoch(binding: FrontierBinding) -> BindingEpoch {
6955 match binding {
6956 FrontierBinding::Bound(epoch) | FrontierBinding::Detached(epoch) => epoch,
6957 }
6958}
6959
6960fn block_start_sequence(block: RecoverySequenceBlockRestore) -> DeliverySeq {
6961 block
6962 .terminal
6963 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
6964}
6965
6966fn block_start_validated_sequence(block: RecoverySequenceBlock) -> DeliverySeq {
6967 block
6968 .terminal
6969 .map_or(block.recovery_attach_seq, |terminal| terminal.delivery_seq)
6970}
6971
6972fn block_start_order(block: RecoveryOrderBlockRestore) -> TransactionOrder {
6973 block
6974 .active_binding
6975 .map_or(block.recovery_operation_order, |active_binding| {
6976 active_binding.transaction_order
6977 })
6978}
6979
6980fn block_start_validated_order(block: RecoveryOrderBlock) -> TransactionOrder {
6981 block
6982 .active_binding
6983 .map_or(block.recovery_operation_order, |active_binding| {
6984 active_binding.transaction_order
6985 })
6986}
6987
6988fn order_frontier_start(high: OrderHigh) -> u128 {
6989 match high {
6990 OrderHigh::Empty => 0,
6991 OrderHigh::Allocated(high) => u128::from(high) + 1,
6992 }
6993}
6994
6995const fn order_is_above_high(value: TransactionOrder, high: OrderHigh) -> bool {
6996 match high {
6997 OrderHigh::Empty => true,
6998 OrderHigh::Allocated(high) => value > high,
6999 }
7000}
7001
7002fn order_frontier_candidate_count(restore: &OrderClaimFrontierRestore, high: OrderHigh) -> u128 {
7003 usize_to_u128(
7004 restore
7005 .immutable_candidates
7006 .iter()
7007 .filter(|candidate| order_is_above_high(candidate.transaction_order, high))
7008 .count(),
7009 )
7010}
7011
7012fn sequence_ordinal(ledger: SequenceLedger, value: DeliverySeq) -> u128 {
7013 u128::from(value).saturating_sub(u128::from(ledger.high_watermark()) + 1)
7014}
7015
7016fn order_ordinal(ledger: OrderLedger, value: TransactionOrder) -> u128 {
7017 u128::from(value).saturating_sub(order_frontier_start(ledger.high()))
7018}
7019
7020fn checked_rank_value(start: DeliverySeq, active_rank: usize) -> Option<DeliverySeq> {
7021 let rank = u64::try_from(active_rank).ok()?;
7022 start.checked_add(rank)
7023}
7024
7025fn usize_to_u64(value: usize) -> u64 {
7026 u64::try_from(value).map_or(u64::MAX, core::convert::identity)
7027}
7028
7029fn usize_to_u128(value: usize) -> u128 {
7030 u64::try_from(value).map_or(u128::MAX, u128::from)
7031}
7032
7033fn rank_index(rank: usize) -> u128 {
7034 usize_to_u128(rank)
7035}
7036
7037const fn frontier_error(
7038 counter: ClaimFrontierCounter,
7039 first_bad_position: u128,
7040 reason: ClaimFrontierInvalidReason,
7041) -> ClaimFrontierError {
7042 ClaimFrontierError {
7043 counter,
7044 first_bad_position,
7045 reason,
7046 }
7047}
7048
7049const fn sequence_error(
7050 first_bad_position: u128,
7051 reason: ClaimFrontierInvalidReason,
7052) -> ClaimFrontierError {
7053 frontier_error(
7054 ClaimFrontierCounter::DeliverySequence,
7055 first_bad_position,
7056 reason,
7057 )
7058}
7059
7060const fn order_error(
7061 first_bad_position: u128,
7062 reason: ClaimFrontierInvalidReason,
7063) -> ClaimFrontierError {
7064 frontier_error(
7065 ClaimFrontierCounter::TransactionOrder,
7066 first_bad_position,
7067 reason,
7068 )
7069}