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