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