1use alloc::{boxed::Box, vec, vec::Vec};
8
9use crate::{algebra::ResourceVector, wire::RecordAdmission};
10
11use super::super::{
12 AttachCommit, AttachTransition, BindingState, ClaimFrontiers, ClosureAccounting, ClosureDebt,
13 CommittedDetachTransition, DebtCompletion, DetachCell, DetachedCredentialRecovery, Event,
14 FencedAttachCommit, FrontierBinding, FrontierParticipant, IdentityState,
15 InitialEnrollmentFrontierCommit, LeaveCommitError, LeaveCommitParameters, LiveMember,
16 MarkerAckCommit, NonzeroParticipantAckCommit, ObserverProgressProjection, OrderLedger,
17 ParticipantAckCommit, PendingFinalization, PendingLeaveCommitParameters,
18 PrepareLeaveAuthorityError, RetainedCausalRecord, RetainedCausalRecordKind, SequenceLedger,
19 StoredEdge, VerifiedLeaveRequest,
20 claim_frontier::{BindingTerminalOwner, FencedMarkerSourceRecord, LiveFrontierTransitionError},
21 commit_leave, commit_pending_leave,
22};
23use super::{
24 InitialEnrollmentOperationCommit, MarkerDeliveryProjection, MarkerDrainCommit,
25 RecordAdmissionPersistenceParts, RetainedRecordCharge, UnchangedRecordAdmission,
26};
27
28mod binding_fate_transition;
29mod ledger;
30mod state;
31pub(super) use binding_fate_transition::BindingFateOwnerPlan;
32use ledger::{
33 detach_order, detach_sequence, detached_attach_order, detached_attach_sequence,
34 enrollment_order, enrollment_sequence, superseding_attach_order, superseding_attach_sequence,
35};
36use state::{
37 accounting_after_fenced_attach, accounting_after_leave, accounting_after_marker_ack,
38 accounting_after_marker_crossings, accounting_after_rows, retained_attached, retained_terminal,
39};
40
41#[derive(Debug, PartialEq, Eq)]
66pub struct LiveFrontierOwner {
67 frontiers: ClaimFrontiers,
68 closure_accounting: ClosureAccounting,
69 retained_charges: Vec<RetainedRecordCharge>,
70 retained_record_limit: u64,
71}
72
73impl LiveFrontierOwner {
74 #[must_use]
76 pub fn from_initial_enrollment<F>(
77 initial: InitialEnrollmentFrontierCommit<F>,
78 retained_record_limit: u64,
79 ) -> (InitialEnrollmentOperationCommit<F>, Self) {
80 let (operation, frontiers, closure_accounting, attached_charge) =
81 initial.into_conversation_parts();
82 let attached = operation.enrollment().attached;
83 let retained_charges = vec![RetainedRecordCharge::new(
84 attached.delivery_seq(),
85 attached.admission_order(),
86 attached_charge,
87 )];
88 (
89 operation,
90 Self {
91 frontiers,
92 closure_accounting,
93 retained_charges,
94 retained_record_limit,
95 },
96 )
97 }
98
99 #[cfg(any(test, feature = "test-support"))]
100 pub(in crate::lifecycle) const fn from_test_parts(
101 frontiers: ClaimFrontiers,
102 closure_accounting: ClosureAccounting,
103 retained_charges: Vec<RetainedRecordCharge>,
104 retained_record_limit: u64,
105 ) -> Self {
106 Self {
107 frontiers,
108 closure_accounting,
109 retained_charges,
110 retained_record_limit,
111 }
112 }
113
114 #[cfg(any(test, feature = "test-support"))]
122 pub fn with_pending_finalizer_test_capacity(
123 mut self,
124 finalizer_rows: u64,
125 finalizer_charge: ResourceVector,
126 ) -> Result<Self, &'static str> {
127 let retained_count = u64::try_from(self.retained_charges.len())
128 .map_err(|_| "retained record count exceeds u64")?;
129 self.retained_record_limit = retained_count
130 .checked_add(finalizer_rows)
131 .ok_or("pending finalizer retained-record capacity overflow")?;
132 let current = self.closure_accounting;
133 let configured = current.configured_cap();
134 let configured = ResourceVector::new(
135 configured
136 .entries
137 .checked_add(finalizer_charge.entries)
138 .ok_or("pending finalizer entry capacity overflow")?,
139 configured
140 .bytes
141 .checked_add(finalizer_charge.bytes)
142 .ok_or("pending finalizer byte capacity overflow")?,
143 );
144 self.closure_accounting = ClosureAccounting::try_new(
145 current.state(),
146 current.marker_capacity_credits(),
147 current.marker_anchors(),
148 current.edge_sequence_claims(),
149 current.edge_order_position_claims(),
150 current.edge_k_remaining(),
151 current.baseline(),
152 configured,
153 current.episode_churn_used(),
154 current.episode_churn_limit(),
155 )
156 .map_err(|_| "pending finalizer closure capacity extension refused")?;
157 Ok(self)
158 }
159
160 #[must_use]
162 pub const fn frontiers(&self) -> &ClaimFrontiers {
163 &self.frontiers
164 }
165
166 #[must_use]
168 pub const fn closure_accounting(&self) -> ClosureAccounting {
169 self.closure_accounting
170 }
171
172 #[must_use]
174 pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
175 &self.retained_charges
176 }
177
178 #[must_use]
180 pub const fn retained_record_limit(&self) -> u64 {
181 self.retained_record_limit
182 }
183
184 #[must_use]
205 pub fn reconcile_orphaned_marker_anchors(&mut self) -> u64 {
206 let derived = self
207 .frontiers
208 .unaccepted_marker_anchor_count()
209 .saturating_add(self.frontiers.pending_marker_candidate_count());
210 let stored = self.closure_accounting.marker_anchors();
211 let Some(orphaned) = stored.checked_sub(derived) else {
212 return 0;
213 };
214 if orphaned == 0 {
215 return 0;
216 }
217 match accounting_after_marker_crossings(self.closure_accounting, orphaned) {
218 Some(accounting) => {
219 self.closure_accounting = accounting;
220 orphaned
221 }
222 None => 0,
223 }
224 }
225
226 #[must_use]
228 pub fn into_parts(
229 self,
230 ) -> (
231 ClaimFrontiers,
232 ClosureAccounting,
233 Vec<RetainedRecordCharge>,
234 u64,
235 ) {
236 (
237 self.frontiers,
238 self.closure_accounting,
239 self.retained_charges,
240 self.retained_record_limit,
241 )
242 }
243
244 #[must_use]
247 pub fn from_unchanged_record_admission<EF, V, LF>(
248 unchanged: UnchangedRecordAdmission<'_, EF, V, LF>,
249 retained_record_limit: u64,
250 ) -> (Self, RecordAdmission, ResourceVector) {
251 let (prestate, encoded_record_charge) = unchanged.into_parts();
252 let (request, frontiers, closure_accounting, retained_charges) =
253 prestate.into_live_owner_parts();
254 (
255 Self {
256 frontiers,
257 closure_accounting,
258 retained_charges,
259 retained_record_limit,
260 },
261 request,
262 encoded_record_charge,
263 )
264 }
265
266 #[must_use]
269 pub fn from_record_admission_persistence(
270 persistence: RecordAdmissionPersistenceParts,
271 retained_record_limit: u64,
272 ) -> Self {
273 Self {
274 frontiers: persistence.frontiers,
275 closure_accounting: persistence.accounting,
276 retained_charges: persistence.retained_charges,
277 retained_record_limit,
278 }
279 }
280
281 #[must_use]
283 pub fn from_marker_drain(
284 commit: MarkerDrainCommit,
285 retained_record_limit: u64,
286 ) -> (Self, StoredEdge, MarkerDeliveryProjection) {
287 let (frontiers, closure_accounting, retained_charges, successor, projection) =
288 commit.into_parts();
289 (
290 Self {
291 frontiers,
292 closure_accounting,
293 retained_charges,
294 retained_record_limit,
295 },
296 successor,
297 projection,
298 )
299 }
300
301 pub(super) fn commit_binding_terminal_candidate(
302 self,
303 active_binding: super::super::ActiveBinding,
304 admission_order: super::super::AdmissionOrder,
305 delivery_seq: crate::wire::DeliverySeq,
306 charge: RetainedRecordCharge,
307 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
308 let mut active = self.frontiers.active_identities().participants().to_vec();
309 let Some(participant) = active
310 .iter_mut()
311 .find(|participant| participant.participant_index() == active_binding.participant_id)
312 else {
313 return Err(Box::new((self, LiveFrontierError::Authority)));
314 };
315 if participant.binding() != FrontierBinding::Bound(active_binding.binding_epoch) {
316 return Err(Box::new((self, LiveFrontierError::Authority)));
317 }
318 *participant = FrontierParticipant::new(
319 participant.participant_index(),
320 participant.cursor(),
321 FrontierBinding::Detached(active_binding.binding_epoch),
322 );
323 let row = RetainedCausalRecord {
324 delivery_seq,
325 admission_order,
326 kind: RetainedCausalRecordKind::BindingTerminal(super::super::BindingTerminalOwner {
327 participant_index: active_binding.participant_id,
328 binding_epoch: active_binding.binding_epoch,
329 }),
330 };
331 let Some(sequence) = detach_sequence(self.frontiers.sequence().ledger(), delivery_seq)
332 else {
333 return Err(Box::new((self, LiveFrontierError::Frontier)));
334 };
335 let Some(order) = detach_order(
336 self.frontiers.order().ledger(),
337 admission_order.transaction_order(),
338 ) else {
339 return Err(Box::new((self, LiveFrontierError::Frontier)));
340 };
341 match transition(self, (), active, &[row], vec![charge], sequence, order) {
342 Ok(committed) => {
343 let ((), owner) = committed.into_parts();
344 Ok(owner)
345 }
346 Err(failure) => {
347 let error = failure.error();
348 let ((), owner) = failure.into_parts();
349 Err(Box::new((owner, error)))
350 }
351 }
352 }
353
354 pub(super) fn pend_binding_terminal_candidate(
355 self,
356 active_binding: super::super::ActiveBinding,
357 admission_order: super::super::AdmissionOrder,
358 delivery_seq: crate::wire::DeliverySeq,
359 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
360 let Some(order) = detach_order(
361 self.frontiers.order().ledger(),
362 admission_order.transaction_order(),
363 ) else {
364 return Err(Box::new((self, LiveFrontierError::Frontier)));
365 };
366 let Self {
367 frontiers,
368 closure_accounting,
369 retained_charges,
370 retained_record_limit,
371 } = self;
372 match frontiers.apply_pending_binding_terminal(
373 active_binding.participant_id,
374 active_binding.binding_epoch,
375 delivery_seq,
376 admission_order,
377 order,
378 ) {
379 Ok(frontiers) => Ok(Self {
380 frontiers,
381 closure_accounting,
382 retained_charges,
383 retained_record_limit,
384 }),
385 Err(failure) => {
386 let (frontiers, error) = *failure;
387 Err(Box::new((
388 Self {
389 frontiers,
390 closure_accounting,
391 retained_charges,
392 retained_record_limit,
393 },
394 map_frontier_error(error),
395 )))
396 }
397 }
398 }
399
400 pub fn drain_pending_terminal(
428 self,
429 pending: PendingFinalization,
430 terminal_charge: RetainedRecordCharge,
431 ) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
432 let Some(expected_sequence) = self
433 .frontiers
434 .sequence()
435 .ledger()
436 .high_watermark()
437 .checked_add(1)
438 else {
439 return drain_refusal(self, LiveFrontierError::Frontier);
440 };
441 if terminal_charge.delivery_seq() != expected_sequence
442 || terminal_charge.admission_order() != pending.admission_order()
443 || terminal_charge.encoded_charge().entries != 1
444 {
445 return drain_refusal(self, LiveFrontierError::RetainedCharge);
446 }
447 if self
448 .retained_charges
449 .len()
450 .checked_add(1)
451 .and_then(|len| u64::try_from(len).ok())
452 .is_none()
453 {
454 return drain_refusal(self, LiveFrontierError::RetainedRecordLimit);
455 }
456 let Some(accounting) = accounting_after_rows(self.closure_accounting, &[terminal_charge])
457 else {
458 return drain_refusal(self, LiveFrontierError::ClosureAccounting);
459 };
460 let Self {
461 frontiers,
462 closure_accounting,
463 mut retained_charges,
464 retained_record_limit,
465 } = self;
466 let expected_owner = BindingTerminalOwner {
467 participant_index: pending.participant_id(),
468 binding_epoch: pending.binding_epoch(),
469 };
470 let (frontiers, record) = match frontiers
471 .drain_first_binding_terminal(expected_owner, pending.admission_order())
472 {
473 Ok(drained) => drained,
474 Err(failure) => {
475 let (frontiers, error) = *failure;
476 return drain_refusal(
477 Self {
478 frontiers,
479 closure_accounting,
480 retained_charges,
481 retained_record_limit,
482 },
483 map_frontier_error(error),
484 );
485 }
486 };
487 retained_charges.push(terminal_charge);
488 let projection =
489 ObserverProgressProjection::new(pending.conversation_id(), record.delivery_seq);
490 Ok(DrainedPendingTerminal {
491 owner: Self {
492 frontiers,
493 closure_accounting: accounting,
494 retained_charges,
495 retained_record_limit,
496 },
497 projection,
498 })
499 }
500
501 pub fn retain_fenced_marker_source(
508 self,
509 recovery: DetachedCredentialRecovery,
510 ) -> Result<RetainedFencedMarkerSource, Box<FencedMarkerSourceRetentionRefused>> {
511 let Some(source) = self.frontiers.fenced_marker_source(recovery) else {
512 return Err(Box::new(FencedMarkerSourceRetentionRefused {
513 owner: self,
514 recovery,
515 }));
516 };
517 Ok(RetainedFencedMarkerSource {
518 owner: self,
519 recovery,
520 expectation: FencedMarkerSourceExpectation { source },
521 })
522 }
523
524 #[must_use]
532 pub fn mint_fenced_attach(
533 mut self,
534 marker_source_sequence: u64,
535 recovery: DetachedCredentialRecovery,
536 debt: ClosureDebt,
537 event: Event,
538 successor: DebtCompletion,
539 ) -> MintFencedAttachResult {
540 let Some(record) = self.frontiers.take_fenced_marker_record(recovery) else {
541 return MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
542 owner: self,
543 marker_source_sequence,
544 recovery,
545 debt,
546 event,
547 successor,
548 reason: FencedAttachMintRefusalReason::MarkerAuthority,
549 }));
550 };
551 match recovery.fenced_attach(record, debt, event, successor) {
552 Ok(proof) => MintFencedAttachResult::Minted(Box::new(MintedFencedAttach {
553 owner_without_marker_authority: self,
554 proof,
555 })),
556 Err(refusal) => {
557 self.frontiers
558 .reinstall_fenced_marker_record((*refusal).into_record());
559 MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
560 owner: self,
561 marker_source_sequence,
562 recovery,
563 debt,
564 event,
565 successor,
566 reason: FencedAttachMintRefusalReason::ProofInputs,
567 }))
568 }
569 }
570 }
571}
572
573#[derive(Debug, PartialEq, Eq)]
578pub struct DrainedPendingTerminal {
579 owner: LiveFrontierOwner,
580 projection: ObserverProgressProjection,
581}
582
583impl DrainedPendingTerminal {
584 #[must_use]
586 pub fn into_parts(self) -> (LiveFrontierOwner, ObserverProgressProjection) {
587 (self.owner, self.projection)
588 }
589}
590
591#[derive(Debug, PartialEq, Eq)]
593pub struct PendingTerminalDrainRefused {
594 owner: LiveFrontierOwner,
595 error: LiveFrontierError,
596}
597
598impl PendingTerminalDrainRefused {
599 #[must_use]
601 pub const fn error(&self) -> LiveFrontierError {
602 self.error
603 }
604
605 #[must_use]
607 pub fn into_owner(self) -> LiveFrontierOwner {
608 self.owner
609 }
610}
611
612fn drain_refusal(
613 owner: LiveFrontierOwner,
614 error: LiveFrontierError,
615) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
616 Err(Box::new(PendingTerminalDrainRefused { owner, error }))
617}
618
619#[derive(Clone, Copy, Debug, PartialEq, Eq)]
621pub struct FencedMarkerSourceExpectation {
622 source: FencedMarkerSourceRecord,
623}
624
625impl FencedMarkerSourceExpectation {
626 #[must_use]
628 pub const fn conversation_id(self) -> u64 {
629 self.source.conversation_id
630 }
631
632 #[must_use]
634 pub const fn marker_delivery_seq(self) -> u64 {
635 self.source.delivery_seq
636 }
637
638 #[must_use]
640 pub const fn admission_order(self) -> super::super::AdmissionOrder {
641 self.source.admission_order
642 }
643
644 #[must_use]
646 pub const fn participant_id(self) -> u64 {
647 self.source.participant_id
648 }
649
650 #[must_use]
652 pub const fn provenance(self) -> super::super::MarkerProvenance {
653 self.source.provenance
654 }
655
656 #[must_use]
658 pub const fn target_binding(self) -> FrontierBinding {
659 self.source.target_binding
660 }
661}
662
663#[derive(Debug, PartialEq, Eq)]
665pub struct RetainedFencedMarkerSource {
666 owner: LiveFrontierOwner,
667 recovery: DetachedCredentialRecovery,
668 expectation: FencedMarkerSourceExpectation,
669}
670
671impl RetainedFencedMarkerSource {
672 #[must_use]
674 pub const fn expectation(&self) -> FencedMarkerSourceExpectation {
675 self.expectation
676 }
677
678 #[must_use]
680 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
681 (self.owner, self.recovery)
682 }
683}
684
685#[derive(Debug, PartialEq, Eq)]
687pub struct FencedMarkerSourceRetentionRefused {
688 owner: LiveFrontierOwner,
689 recovery: DetachedCredentialRecovery,
690}
691
692impl FencedMarkerSourceRetentionRefused {
693 #[must_use]
695 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
696 (self.owner, self.recovery)
697 }
698}
699
700#[derive(Clone, Copy, Debug, PartialEq, Eq)]
702pub enum FencedAttachMintRefusalReason {
703 MarkerAuthority,
706 ProofInputs,
708}
709
710#[derive(Debug, PartialEq, Eq)]
712pub struct MintedFencedAttach {
713 owner_without_marker_authority: LiveFrontierOwner,
714 proof: FencedAttachCommit,
715}
716
717impl MintedFencedAttach {
718 #[must_use]
721 pub fn into_parts(self) -> (LiveFrontierOwner, FencedAttachCommit) {
722 (self.owner_without_marker_authority, self.proof)
723 }
724}
725
726#[derive(Debug, PartialEq, Eq)]
728pub struct MintFencedAttachRefused {
729 owner: LiveFrontierOwner,
730 marker_source_sequence: u64,
731 recovery: DetachedCredentialRecovery,
732 debt: ClosureDebt,
733 event: Event,
734 successor: DebtCompletion,
735 reason: FencedAttachMintRefusalReason,
736}
737
738impl MintFencedAttachRefused {
739 #[must_use]
741 pub const fn reason(&self) -> FencedAttachMintRefusalReason {
742 self.reason
743 }
744
745 #[must_use]
747 pub fn into_parts(
748 self,
749 ) -> (
750 LiveFrontierOwner,
751 u64,
752 DetachedCredentialRecovery,
753 ClosureDebt,
754 Event,
755 DebtCompletion,
756 ) {
757 (
758 self.owner,
759 self.marker_source_sequence,
760 self.recovery,
761 self.debt,
762 self.event,
763 self.successor,
764 )
765 }
766}
767
768#[derive(Debug, PartialEq, Eq)]
770pub enum MintFencedAttachResult {
771 Minted(Box<MintedFencedAttach>),
773 MintRefused(Box<MintFencedAttachRefused>),
775}
776
777#[derive(Debug, PartialEq, Eq)]
779pub struct LiveLeaveCommit<EF, V, LF> {
780 identity: IdentityState<EF, V, LF>,
781 owner: LiveFrontierOwner,
782}
783
784impl<EF, V, LF> LiveLeaveCommit<EF, V, LF> {
785 #[must_use]
787 pub const fn observer_progress_projection(&self) -> Option<ObserverProgressProjection> {
788 let IdentityState::Retired(retired) = &self.identity else {
789 return None;
790 };
791 let committed = retired.committed_result();
792 Some(ObserverProgressProjection::new(
793 committed.conversation_id(),
794 committed.left_delivery_seq(),
795 ))
796 }
797
798 #[must_use]
800 pub fn into_parts(self) -> (IdentityState<EF, V, LF>, LiveFrontierOwner) {
801 (self.identity, self.owner)
802 }
803}
804
805#[derive(Clone, Debug, PartialEq, Eq)]
807pub enum LiveLeaveError {
808 Prepare(PrepareLeaveAuthorityError),
810 Commit(LeaveCommitError),
812 RetainedCharge,
814 RetainedRecordLimit,
816 ClosureAccounting,
818 Identity,
820}
821
822pub fn commit_settled_leave_frontier<EF, V, LF, D>(
830 owner: LiveFrontierOwner,
831 member: LiveMember<EF>,
832 binding: BindingState,
833 detach_cell: DetachCell<D>,
834 verified: VerifiedLeaveRequest<V, LF>,
835 left_delivery_seq: u64,
836 left_charge: RetainedRecordCharge,
837) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
838 let LiveFrontierOwner {
839 frontiers,
840 closure_accounting,
841 mut retained_charges,
842 retained_record_limit,
843 } = owner;
844 let retired_marker_charge =
845 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
846 let authority = frontiers
847 .prepare_settled_leave_authority(&member, binding)
848 .map_err(LiveLeaveError::Prepare)?;
849 let commit = commit_leave(
850 member,
851 binding,
852 detach_cell,
853 verified,
854 authority,
855 LeaveCommitParameters { left_delivery_seq },
856 )
857 .map_err(LiveLeaveError::Commit)?;
858 let (identity, frontiers) = commit.into_parts();
859 let IdentityState::Retired(retired) = &identity else {
860 return Err(LiveLeaveError::Identity);
861 };
862 if left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
863 || left_charge.admission_order() != retired.left_admission_order()
864 || left_charge.encoded_charge().entries != 1
865 {
866 return Err(LiveLeaveError::RetainedCharge);
867 }
868 retained_charges.push(left_charge);
869 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
870 let retained_len = u64::try_from(frontiers.retained_records().len())
871 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
872 if retained_len > retained_record_limit
873 || retained_charges.len() != frontiers.retained_records().len()
874 {
875 return Err(LiveLeaveError::RetainedRecordLimit);
876 }
877 let closure_accounting =
878 accounting_after_leave(closure_accounting, &[left_charge], retired_marker_charge)
879 .ok_or(LiveLeaveError::ClosureAccounting)?;
880 Ok(LiveLeaveCommit {
881 identity,
882 owner: LiveFrontierOwner {
883 frontiers,
884 closure_accounting,
885 retained_charges,
886 retained_record_limit,
887 },
888 })
889}
890
891pub fn commit_pending_leave_frontier<EF, V, LF, D>(
900 owner: LiveFrontierOwner,
901 member: LiveMember<EF>,
902 pending: PendingFinalization,
903 detach_cell: DetachCell<D>,
904 verified: VerifiedLeaveRequest<V, LF>,
905 parameters: PendingLeaveCommitParameters,
906 charges: [RetainedRecordCharge; 2],
907) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
908 let [terminal_charge, left_charge] = charges;
909 let terminal_delivery_seq = parameters.terminal_delivery_seq;
910 let LiveFrontierOwner {
911 frontiers,
912 closure_accounting,
913 mut retained_charges,
914 retained_record_limit,
915 } = owner;
916 let retired_marker_charge =
917 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
918 let authority = frontiers
919 .prepare_pending_leave_authority(&member, pending)
920 .map_err(LiveLeaveError::Prepare)?;
921 let commit = commit_pending_leave(
922 member,
923 pending,
924 detach_cell,
925 verified,
926 authority,
927 parameters,
928 )
929 .map_err(LiveLeaveError::Commit)?;
930 let (identity, frontiers) = commit.into_parts();
931 let IdentityState::Retired(retired) = &identity else {
932 return Err(LiveLeaveError::Identity);
933 };
934 if retired.committed_result().prior_terminal_delivery_seq() != Some(terminal_delivery_seq)
935 || terminal_charge.delivery_seq() != terminal_delivery_seq
936 || terminal_charge.admission_order() != pending.admission_order()
937 || terminal_charge.encoded_charge().entries != 1
938 || left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
939 || left_charge.admission_order() != retired.left_admission_order()
940 || left_charge.encoded_charge().entries != 1
941 {
942 return Err(LiveLeaveError::RetainedCharge);
943 }
944 retained_charges.extend([terminal_charge, left_charge]);
945 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
946 let retained_len = u64::try_from(frontiers.retained_records().len())
947 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
948 if retained_len > retained_record_limit {
949 return Err(LiveLeaveError::RetainedRecordLimit);
950 }
951 if retained_charges.len() != frontiers.retained_records().len() {
952 return Err(LiveLeaveError::RetainedCharge);
953 }
954 let closure_accounting = accounting_after_leave(
955 closure_accounting,
956 &[terminal_charge, left_charge],
957 retired_marker_charge,
958 )
959 .ok_or(LiveLeaveError::ClosureAccounting)?;
960 Ok(LiveLeaveCommit {
961 identity,
962 owner: LiveFrontierOwner {
963 frontiers,
964 closure_accounting,
965 retained_charges,
966 retained_record_limit,
967 },
968 })
969}
970
971fn retired_marker_charge(
972 frontiers: &ClaimFrontiers,
973 retained_charges: &[RetainedRecordCharge],
974 participant_id: crate::wire::ParticipantId,
975) -> Result<Option<RetainedRecordCharge>, LiveLeaveError> {
976 let marker_sequence = frontiers
977 .retained_marker_records()
978 .iter()
979 .find_map(|record| {
980 matches!(
981 record.kind,
982 RetainedCausalRecordKind::CompactionMarker {
983 participant_index,
984 ..
985 } if participant_index == participant_id
986 )
987 .then_some(record.delivery_seq)
988 });
989 let Some(marker_sequence) = marker_sequence else {
990 return Ok(None);
991 };
992 retained_charges
993 .iter()
994 .copied()
995 .find(|charge| charge.delivery_seq() == marker_sequence)
996 .map(Some)
997 .ok_or(LiveLeaveError::RetainedCharge)
998}
999
1000#[derive(Debug, PartialEq, Eq)]
1002pub struct AttachFrontierCharges {
1003 terminal: Option<RetainedRecordCharge>,
1004 attached: RetainedRecordCharge,
1005 seal: LiveTransitionInputSeal,
1006}
1007
1008#[derive(Debug, PartialEq, Eq)]
1009enum LiveTransitionInputSeal {
1010 Validated,
1011}
1012
1013impl AttachFrontierCharges {
1014 #[must_use]
1016 pub const fn new(
1017 terminal: Option<RetainedRecordCharge>,
1018 attached: RetainedRecordCharge,
1019 ) -> Self {
1020 Self {
1021 terminal,
1022 attached,
1023 seal: LiveTransitionInputSeal::Validated,
1024 }
1025 }
1026
1027 const fn into_parts(self) -> (Option<RetainedRecordCharge>, RetainedRecordCharge) {
1028 let Self {
1029 terminal,
1030 attached,
1031 seal,
1032 } = self;
1033 match seal {
1034 LiveTransitionInputSeal::Validated => (terminal, attached),
1035 }
1036 }
1037}
1038
1039#[derive(Debug, PartialEq, Eq)]
1041pub struct LiveFrontierCommit<T> {
1042 operation: T,
1043 owner: LiveFrontierOwner,
1044}
1045
1046impl<T> LiveFrontierCommit<T> {
1047 #[must_use]
1049 pub const fn operation(&self) -> &T {
1050 &self.operation
1051 }
1052
1053 #[must_use]
1055 pub const fn owner(&self) -> &LiveFrontierOwner {
1056 &self.owner
1057 }
1058
1059 #[must_use]
1061 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1062 (self.operation, self.owner)
1063 }
1064}
1065
1066#[derive(Debug, PartialEq, Eq)]
1068pub struct LiveFrontierFailure<T> {
1069 error: LiveFrontierError,
1070 operation: T,
1071 owner: LiveFrontierOwner,
1072}
1073
1074impl<T> LiveFrontierFailure<T> {
1075 #[must_use]
1077 pub const fn error(&self) -> LiveFrontierError {
1078 self.error
1079 }
1080
1081 #[must_use]
1083 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1084 (self.operation, self.owner)
1085 }
1086}
1087
1088#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1090pub enum LiveFrontierError {
1091 Authority,
1093 Precedence,
1095 RetainedCharge,
1097 RetainedRecordLimit,
1099 Frontier,
1101 ClosureAccounting,
1103}
1104
1105pub type LiveFrontierResult<T> = Result<LiveFrontierCommit<T>, Box<LiveFrontierFailure<T>>>;
1107
1108pub fn apply_enrollment_frontier<F>(
1114 owner: LiveFrontierOwner,
1115 operation: super::super::EnrollmentCommit<F>,
1116 attached_charge: RetainedRecordCharge,
1117) -> LiveFrontierResult<super::super::EnrollmentCommit<F>> {
1118 let attached = operation.attached;
1119 if attached.conversation_id() != owner.frontiers.conversation_id() {
1120 return failure(owner, operation, LiveFrontierError::Authority);
1121 }
1122 let participant_id = attached.participant_id();
1123 let mut active = owner.frontiers.active_identities().participants().to_vec();
1124 if active
1125 .iter()
1126 .any(|participant| participant.participant_index() == participant_id)
1127 {
1128 return failure(owner, operation, LiveFrontierError::Authority);
1129 }
1130 active.push(FrontierParticipant::new(
1131 participant_id,
1132 operation.member.cursor(),
1133 FrontierBinding::Bound(attached.binding_epoch()),
1134 ));
1135 active.sort_unstable_by_key(|participant| participant.participant_index());
1136 let rows = [retained_attached(attached)];
1137 let Some(sequence) =
1138 enrollment_sequence(owner.frontiers.sequence().ledger(), attached.delivery_seq())
1139 else {
1140 return failure(owner, operation, LiveFrontierError::Frontier);
1141 };
1142 let Some(order) = enrollment_order(
1143 owner.frontiers.order().ledger(),
1144 attached.admission_order().transaction_order(),
1145 ) else {
1146 return failure(owner, operation, LiveFrontierError::Frontier);
1147 };
1148 transition(
1149 owner,
1150 operation,
1151 active,
1152 &rows,
1153 vec![attached_charge],
1154 sequence,
1155 order,
1156 )
1157}
1158
1159pub fn apply_attach_frontier<F, V>(
1166 owner: LiveFrontierOwner,
1167 operation: AttachCommit<F, V>,
1168 charges: AttachFrontierCharges,
1169) -> LiveFrontierResult<AttachCommit<F, V>> {
1170 let (terminal_charge, attached_charge) = charges.into_parts();
1171 let attached = operation.attached;
1172 if attached.conversation_id() != owner.frontiers.conversation_id() {
1173 return failure(owner, operation, LiveFrontierError::Authority);
1174 }
1175 let mut active = owner.frontiers.active_identities().participants().to_vec();
1176 let Some(participant) = active
1177 .iter_mut()
1178 .find(|participant| participant.participant_index() == attached.participant_id())
1179 else {
1180 return failure(owner, operation, LiveFrontierError::Authority);
1181 };
1182 *participant = FrontierParticipant::new(
1183 participant.participant_index(),
1184 operation.member.cursor(),
1185 FrontierBinding::Bound(attached.binding_epoch()),
1186 );
1187 let current_sequence = owner.frontiers.sequence().ledger();
1188 let current_order = owner.frontiers.order().ledger();
1189 let (rows, keyed_charges, sequence, order) = match operation.transition {
1190 AttachTransition::Detached => {
1191 if terminal_charge.is_some() {
1192 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1193 }
1194 let Some(sequence) =
1195 detached_attach_sequence(current_sequence, attached.delivery_seq())
1196 else {
1197 return failure(owner, operation, LiveFrontierError::Frontier);
1198 };
1199 let Some(order) = detached_attach_order(
1200 current_order,
1201 attached.admission_order().transaction_order(),
1202 ) else {
1203 return failure(owner, operation, LiveFrontierError::Frontier);
1204 };
1205 (
1206 vec![retained_attached(attached)],
1207 vec![attached_charge],
1208 sequence,
1209 order,
1210 )
1211 }
1212 AttachTransition::Superseded { terminal } => {
1213 let Some(terminal_charge) = terminal_charge else {
1214 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1215 };
1216 let rows = vec![
1217 retained_terminal(terminal.into()),
1218 retained_attached(attached),
1219 ];
1220 let Some(sequence) = superseding_attach_sequence(current_sequence, &rows) else {
1221 return failure(owner, operation, LiveFrontierError::Frontier);
1222 };
1223 let Some(order) = superseding_attach_order(
1224 current_order,
1225 attached.admission_order().transaction_order(),
1226 ) else {
1227 return failure(owner, operation, LiveFrontierError::Frontier);
1228 };
1229 (
1230 rows,
1231 vec![terminal_charge, attached_charge],
1232 sequence,
1233 order,
1234 )
1235 }
1236 AttachTransition::FencedRecovery {
1237 prior_binding_epoch,
1238 composed_terminal,
1239 next_closure_state,
1240 } => {
1241 return apply_fenced_attach_frontier(
1242 owner,
1243 operation,
1244 terminal_charge,
1245 attached_charge,
1246 prior_binding_epoch,
1247 composed_terminal,
1248 next_closure_state,
1249 );
1250 }
1251 };
1252 transition(
1253 owner,
1254 operation,
1255 active,
1256 &rows,
1257 keyed_charges,
1258 sequence,
1259 order,
1260 )
1261}
1262
1263pub fn apply_detach_frontier<EF, V>(
1269 owner: LiveFrontierOwner,
1270 operation: CommittedDetachTransition<EF, V>,
1271 terminal_charge: RetainedRecordCharge,
1272) -> LiveFrontierResult<CommittedDetachTransition<EF, V>> {
1273 let terminal = operation.terminal();
1274 if terminal.conversation_id() != owner.frontiers.conversation_id() {
1275 return failure(owner, operation, LiveFrontierError::Authority);
1276 }
1277 let mut active = owner.frontiers.active_identities().participants().to_vec();
1278 let Some(participant) = active
1279 .iter_mut()
1280 .find(|participant| participant.participant_index() == terminal.participant_id())
1281 else {
1282 return failure(owner, operation, LiveFrontierError::Authority);
1283 };
1284 *participant = FrontierParticipant::new(
1285 participant.participant_index(),
1286 operation.member().cursor(),
1287 FrontierBinding::Detached(terminal.binding_epoch()),
1288 );
1289 let row = retained_terminal(terminal.into());
1290 let Some(sequence) = detach_sequence(owner.frontiers.sequence().ledger(), row.delivery_seq)
1291 else {
1292 return failure(owner, operation, LiveFrontierError::Frontier);
1293 };
1294 let Some(order) = detach_order(
1295 owner.frontiers.order().ledger(),
1296 row.admission_order.transaction_order(),
1297 ) else {
1298 return failure(owner, operation, LiveFrontierError::Frontier);
1299 };
1300 transition(
1301 owner,
1302 operation,
1303 active,
1304 &[row],
1305 vec![terminal_charge],
1306 sequence,
1307 order,
1308 )
1309}
1310
1311fn crossed_marker_anchors(
1321 owner: &LiveFrontierOwner,
1322 participant_id: crate::wire::ParticipantId,
1323 previous_cursor: crate::wire::DeliverySeq,
1324 through_seq: crate::wire::DeliverySeq,
1325) -> u64 {
1326 owner
1327 .frontiers
1328 .retained_marker_records()
1329 .iter()
1330 .filter(|record| {
1331 matches!(
1332 record.kind,
1333 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1334 if participant_index == participant_id
1335 ) && previous_cursor < record.delivery_seq
1336 && record.delivery_seq <= through_seq
1337 })
1338 .fold(0_u64, |count, _| count.saturating_add(1))
1339}
1340
1341pub fn apply_participant_ack_frontier(
1355 mut owner: LiveFrontierOwner,
1356 operation: ParticipantAckCommit,
1357) -> LiveFrontierResult<ParticipantAckCommit> {
1358 let request = operation.outcome().request();
1359 let Some(current) = owner
1360 .frontiers
1361 .active_identities()
1362 .participants()
1363 .iter()
1364 .find(|participant| participant.participant_index() == request.participant_id)
1365 .copied()
1366 else {
1367 return failure(owner, operation, LiveFrontierError::Authority);
1368 };
1369 let crossed = crossed_marker_anchors(
1370 &owner,
1371 request.participant_id,
1372 current.cursor(),
1373 request.through_seq,
1374 );
1375 let accounting = if crossed == 0 {
1376 None
1377 } else {
1378 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1379 Some(accounting) => Some(accounting),
1380 None => {
1381 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1382 }
1383 }
1384 };
1385 let participant = FrontierParticipant::new(
1386 request.participant_id,
1387 request.through_seq,
1388 current.binding(),
1389 );
1390 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1391 Ok(frontiers) => frontiers,
1392 Err(frontier_failure) => {
1393 let (frontiers, error) = *frontier_failure;
1394 owner.frontiers = frontiers;
1395 return failure(owner, operation, map_frontier_error(error));
1396 }
1397 };
1398 if let Some(accounting) = accounting {
1399 owner.closure_accounting = accounting;
1400 }
1401 Ok(LiveFrontierCommit { operation, owner })
1402}
1403
1404pub fn apply_nonzero_participant_ack_frontier(
1414 mut owner: LiveFrontierOwner,
1415 operation: NonzeroParticipantAckCommit,
1416) -> LiveFrontierResult<NonzeroParticipantAckCommit> {
1417 let request = operation.outcome().request();
1418 let Some(current) = owner
1419 .frontiers
1420 .active_identities()
1421 .participants()
1422 .iter()
1423 .find(|participant| participant.participant_index() == request.participant_id)
1424 .copied()
1425 else {
1426 return failure(owner, operation, LiveFrontierError::Authority);
1427 };
1428 let crossed = crossed_marker_anchors(
1431 &owner,
1432 request.participant_id,
1433 current.cursor(),
1434 request.through_seq,
1435 );
1436 let accounting = if crossed == 0 {
1437 None
1438 } else {
1439 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1440 Some(accounting) => Some(accounting),
1441 None => {
1442 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1443 }
1444 }
1445 };
1446 let participant = FrontierParticipant::new(
1447 request.participant_id,
1448 request.through_seq,
1449 current.binding(),
1450 );
1451 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1452 Ok(frontiers) => frontiers,
1453 Err(frontier_failure) => {
1454 let (frontiers, error) = *frontier_failure;
1455 owner.frontiers = frontiers;
1456 return failure(owner, operation, map_frontier_error(error));
1457 }
1458 };
1459 if let Some(accounting) = accounting {
1460 owner.closure_accounting = accounting;
1461 }
1462 Ok(LiveFrontierCommit { operation, owner })
1463}
1464
1465pub fn apply_marker_ack_frontier(
1471 mut owner: LiveFrontierOwner,
1472 operation: MarkerAckCommit,
1473) -> LiveFrontierResult<MarkerAckCommit> {
1474 let request = operation.outcome().request();
1475 if !owner
1476 .frontiers
1477 .retained_marker_records()
1478 .iter()
1479 .any(|record| {
1480 record.delivery_seq == request.marker_delivery_seq
1481 && matches!(
1482 record.kind,
1483 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1484 if participant_index == request.participant_id
1485 )
1486 })
1487 {
1488 return failure(owner, operation, LiveFrontierError::Authority);
1489 }
1490 let Some(current) = owner
1491 .frontiers
1492 .active_identities()
1493 .participants()
1494 .iter()
1495 .find(|participant| participant.participant_index() == request.participant_id)
1496 .copied()
1497 else {
1498 return failure(owner, operation, LiveFrontierError::Authority);
1499 };
1500 let Some(accounting) = accounting_after_marker_ack(owner.closure_accounting) else {
1501 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1502 };
1503 let participant = FrontierParticipant::new(
1504 request.participant_id,
1505 request.marker_delivery_seq,
1506 current.binding(),
1507 );
1508 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1509 Ok(frontiers) => frontiers,
1510 Err(frontier_failure) => {
1511 let (frontiers, error) = *frontier_failure;
1512 owner.frontiers = frontiers;
1513 return failure(owner, operation, map_frontier_error(error));
1514 }
1515 };
1516 owner.closure_accounting = accounting;
1517 Ok(LiveFrontierCommit { operation, owner })
1518}
1519
1520fn apply_fenced_attach_frontier<F, V>(
1521 owner: LiveFrontierOwner,
1522 operation: AttachCommit<F, V>,
1523 terminal_charge: Option<RetainedRecordCharge>,
1524 attached_charge: RetainedRecordCharge,
1525 prior_binding_epoch: crate::wire::BindingEpoch,
1526 composed_terminal: Option<super::super::CommittedBindingTerminal>,
1527 next_closure_state: super::super::ClosureState,
1528) -> LiveFrontierResult<AttachCommit<F, V>> {
1529 let attached = operation.attached;
1530 let (rows, charges) = match (composed_terminal, terminal_charge) {
1531 (None, None) => (vec![retained_attached(attached)], vec![attached_charge]),
1532 (Some(terminal), Some(terminal_charge)) => (
1533 vec![retained_terminal(terminal), retained_attached(attached)],
1534 vec![terminal_charge, attached_charge],
1535 ),
1536 (None, Some(_)) | (Some(_), None) => {
1537 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1538 }
1539 };
1540 let participant = FrontierParticipant::new(
1541 attached.participant_id(),
1542 operation.member.cursor(),
1543 FrontierBinding::Bound(attached.binding_epoch()),
1544 );
1545 fenced_attach_transition(
1546 owner,
1547 operation,
1548 participant,
1549 prior_binding_epoch,
1550 next_closure_state,
1551 &rows,
1552 charges,
1553 )
1554}
1555
1556fn fenced_attach_transition<T>(
1557 mut owner: LiveFrontierOwner,
1558 operation: T,
1559 participant: FrontierParticipant,
1560 prior_binding_epoch: crate::wire::BindingEpoch,
1561 next_closure_state: super::super::ClosureState,
1562 rows: &[RetainedCausalRecord],
1563 charges: Vec<RetainedRecordCharge>,
1564) -> LiveFrontierResult<T> {
1565 if rows.len() != charges.len()
1566 || rows.iter().zip(&charges).any(|(row, charge)| {
1567 row.delivery_seq != charge.delivery_seq()
1568 || row.admission_order != charge.admission_order()
1569 || charge.encoded_charge().entries != 1
1570 })
1571 {
1572 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1573 }
1574 let resulting_len = owner
1575 .frontiers
1576 .retained_records()
1577 .len()
1578 .checked_add(rows.len());
1579 if resulting_len
1580 .and_then(|len| u64::try_from(len).ok())
1581 .is_none_or(|len| len > owner.retained_record_limit)
1582 {
1583 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1584 }
1585 let Some(accounting) =
1586 accounting_after_fenced_attach(owner.closure_accounting, &charges, next_closure_state)
1587 else {
1588 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1589 };
1590 owner.frontiers =
1591 match owner
1592 .frontiers
1593 .apply_live_fenced_attach(participant, prior_binding_epoch, rows)
1594 {
1595 Ok(frontiers) => frontiers,
1596 Err(frontier_failure) => {
1597 let (frontiers, error) = *frontier_failure;
1598 owner.frontiers = frontiers;
1599 return failure(owner, operation, map_frontier_error(error));
1600 }
1601 };
1602 owner.retained_charges.extend(charges);
1603 owner
1604 .retained_charges
1605 .sort_unstable_by_key(|charge| charge.delivery_seq());
1606 owner.closure_accounting = accounting;
1607 Ok(LiveFrontierCommit { operation, owner })
1608}
1609
1610fn transition<T>(
1611 mut owner: LiveFrontierOwner,
1612 operation: T,
1613 active: Vec<FrontierParticipant>,
1614 rows: &[RetainedCausalRecord],
1615 charges: Vec<RetainedRecordCharge>,
1616 sequence: SequenceLedger,
1617 order: OrderLedger,
1618) -> LiveFrontierResult<T> {
1619 if rows.len() != charges.len()
1620 || rows.iter().zip(&charges).any(|(row, charge)| {
1621 row.delivery_seq != charge.delivery_seq()
1622 || row.admission_order != charge.admission_order()
1623 || charge.encoded_charge().entries != 1
1624 })
1625 {
1626 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1627 }
1628 let resulting_len = owner
1629 .frontiers
1630 .retained_records()
1631 .len()
1632 .checked_add(rows.len());
1633 if resulting_len
1634 .and_then(|len| u64::try_from(len).ok())
1635 .is_none_or(|len| len > owner.retained_record_limit)
1636 {
1637 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1638 }
1639 let Some(accounting) = accounting_after_rows(owner.closure_accounting, &charges) else {
1640 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1641 };
1642 owner.frontiers = match owner
1643 .frontiers
1644 .apply_live_transition(active, rows, sequence, order)
1645 {
1646 Ok(frontiers) => frontiers,
1647 Err(frontier_failure) => {
1648 let (frontiers, error) = *frontier_failure;
1649 owner.frontiers = frontiers;
1650 return failure(owner, operation, map_frontier_error(error));
1651 }
1652 };
1653 owner.retained_charges.extend(charges);
1654 owner.closure_accounting = accounting;
1655 Ok(LiveFrontierCommit { operation, owner })
1656}
1657
1658const fn map_frontier_error(error: LiveFrontierTransitionError) -> LiveFrontierError {
1659 match error {
1660 LiveFrontierTransitionError::Authority => LiveFrontierError::Authority,
1661 LiveFrontierTransitionError::Precedence => LiveFrontierError::Precedence,
1662 LiveFrontierTransitionError::RecordPosition
1663 | LiveFrontierTransitionError::Exhausted
1664 | LiveFrontierTransitionError::ResultingFrontier => LiveFrontierError::Frontier,
1665 }
1666}
1667
1668fn failure<T, U>(
1669 owner: LiveFrontierOwner,
1670 operation: T,
1671 error: LiveFrontierError,
1672) -> Result<U, Box<LiveFrontierFailure<T>>> {
1673 Err(Box::new(LiveFrontierFailure {
1674 error,
1675 operation,
1676 owner,
1677 }))
1678}