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