1use alloc::{boxed::Box, vec, vec::Vec};
8use core::cmp::Ordering;
9
10use crate::{algebra::ResourceVector, wire::RecordAdmission};
11
12use super::super::{
13 AttachCommit, AttachTransition, BindingState, ClaimFrontiers, ClosureAccounting, ClosureDebt,
14 CommittedDetachTransition, DebtCompletion, DetachCell, DetachedCredentialRecovery, Event,
15 FencedAttachCommit, FrontierBinding, FrontierParticipant, IdentityState,
16 InitialEnrollmentFrontierCommit, LeaveCommitError, LeaveCommitParameters, LiveMember,
17 MarkerAckCommit, NonzeroParticipantAckCommit, ObserverProgressProjection, OrderLedger,
18 ParticipantAckCommit, PendingFinalization, PendingLeaveCommitParameters,
19 PrepareLeaveAuthorityError, RetainedCausalRecord, RetainedCausalRecordKind, SequenceLedger,
20 StoredEdge, VerifiedLeaveRequest,
21 claim_frontier::{
22 BindingTerminalOwner, FencedMarkerSourceRecord, LiveFrontierTransitionError,
23 PrecedenceCondition,
24 },
25 commit_leave, commit_pending_leave,
26};
27use super::{
28 InitialEnrollmentOperationCommit, MarkerDeliveryProjection, MarkerDrainCommit,
29 RecordAdmissionPersistenceParts, RetainedRecordCharge, UnchangedRecordAdmission,
30};
31
32mod binding_fate_transition;
33mod ledger;
34mod state;
35pub(super) use binding_fate_transition::BindingFateOwnerPlan;
36use ledger::{
37 detach_order, detach_sequence, detached_attach_order, detached_attach_sequence,
38 enrollment_order, enrollment_sequence, superseding_attach_order, superseding_attach_sequence,
39};
40use state::{
41 accounting_after_fenced_attach, accounting_after_leave, accounting_after_marker_ack,
42 accounting_after_marker_crossings, accounting_after_marker_remint, accounting_after_rows,
43 retained_attached, retained_terminal,
44};
45
46#[derive(Debug, PartialEq, Eq)]
71pub struct LiveFrontierOwner {
72 frontiers: ClaimFrontiers,
73 closure_accounting: ClosureAccounting,
74 retained_charges: Vec<RetainedRecordCharge>,
75 retained_record_limit: u64,
76}
77
78#[derive(Clone, Copy, Debug, PartialEq, Eq)]
92pub enum MarkerAnchorReconcile {
93 InStep,
95 RetiredOrphans(u64),
98 RemintedDeficit(u64),
101 Refused {
104 census: u64,
106 stored: u64,
108 },
109}
110
111impl MarkerAnchorReconcile {
112 #[must_use]
114 pub const fn reconciled(self) -> u64 {
115 match self {
116 Self::InStep | Self::Refused { .. } => 0,
117 Self::RetiredOrphans(count) | Self::RemintedDeficit(count) => count,
118 }
119 }
120}
121
122impl LiveFrontierOwner {
123 #[must_use]
125 pub fn from_initial_enrollment<F>(
126 initial: InitialEnrollmentFrontierCommit<F>,
127 retained_record_limit: u64,
128 ) -> (InitialEnrollmentOperationCommit<F>, Self) {
129 let (operation, frontiers, closure_accounting, attached_charge) =
130 initial.into_conversation_parts();
131 let attached = operation.enrollment().attached;
132 let retained_charges = vec![RetainedRecordCharge::new(
133 attached.delivery_seq(),
134 attached.admission_order(),
135 attached_charge,
136 )];
137 (
138 operation,
139 Self {
140 frontiers,
141 closure_accounting,
142 retained_charges,
143 retained_record_limit,
144 },
145 )
146 }
147
148 #[cfg(any(test, feature = "test-support"))]
159 #[must_use]
160 pub const fn from_test_parts(
161 frontiers: ClaimFrontiers,
162 closure_accounting: ClosureAccounting,
163 retained_charges: Vec<RetainedRecordCharge>,
164 retained_record_limit: u64,
165 ) -> Self {
166 Self {
167 frontiers,
168 closure_accounting,
169 retained_charges,
170 retained_record_limit,
171 }
172 }
173
174 #[cfg(any(test, feature = "test-support"))]
182 pub fn with_pending_finalizer_test_capacity(
183 mut self,
184 finalizer_rows: u64,
185 finalizer_charge: ResourceVector,
186 ) -> Result<Self, &'static str> {
187 let retained_count = u64::try_from(self.retained_charges.len())
188 .map_err(|_| "retained record count exceeds u64")?;
189 self.retained_record_limit = retained_count
190 .checked_add(finalizer_rows)
191 .ok_or("pending finalizer retained-record capacity overflow")?;
192 let current = self.closure_accounting;
193 let configured = current.configured_cap();
194 let configured = ResourceVector::new(
195 configured
196 .entries
197 .checked_add(finalizer_charge.entries)
198 .ok_or("pending finalizer entry capacity overflow")?,
199 configured
200 .bytes
201 .checked_add(finalizer_charge.bytes)
202 .ok_or("pending finalizer byte capacity overflow")?,
203 );
204 self.closure_accounting = ClosureAccounting::try_new(
205 current.state(),
206 current.marker_capacity_credits(),
207 current.marker_anchors(),
208 current.edge_sequence_claims(),
209 current.edge_order_position_claims(),
210 current.edge_k_remaining(),
211 current.baseline(),
212 configured,
213 current.episode_churn_used(),
214 current.episode_churn_limit(),
215 )
216 .map_err(|_| "pending finalizer closure capacity extension refused")?;
217 Ok(self)
218 }
219
220 #[must_use]
222 pub const fn frontiers(&self) -> &ClaimFrontiers {
223 &self.frontiers
224 }
225
226 #[must_use]
228 pub const fn closure_accounting(&self) -> ClosureAccounting {
229 self.closure_accounting
230 }
231
232 #[must_use]
234 pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
235 &self.retained_charges
236 }
237
238 #[must_use]
240 pub const fn retained_record_limit(&self) -> u64 {
241 self.retained_record_limit
242 }
243
244 #[must_use]
286 pub fn reconcile_marker_anchor_ledgers(&mut self) -> MarkerAnchorReconcile {
287 let census = self
288 .frontiers
289 .unaccepted_marker_anchor_count()
290 .saturating_add(self.frontiers.pending_marker_candidate_count());
291 let stored = self.closure_accounting.marker_anchors();
292 let repaired = match census.cmp(&stored) {
293 Ordering::Equal => return MarkerAnchorReconcile::InStep,
294 Ordering::Less => stored.checked_sub(census).and_then(|orphaned| {
295 accounting_after_marker_crossings(self.closure_accounting, orphaned)
296 .map(|accounting| (accounting, MarkerAnchorReconcile::RetiredOrphans(orphaned)))
297 }),
298 Ordering::Greater => census.checked_sub(stored).and_then(|minted| {
299 accounting_after_marker_remint(self.closure_accounting, minted)
300 .map(|accounting| (accounting, MarkerAnchorReconcile::RemintedDeficit(minted)))
301 }),
302 };
303 match repaired {
304 Some((accounting, outcome)) => {
305 self.closure_accounting = accounting;
306 outcome
307 }
308 None => MarkerAnchorReconcile::Refused { census, stored },
309 }
310 }
311
312 #[must_use]
314 pub fn into_parts(
315 self,
316 ) -> (
317 ClaimFrontiers,
318 ClosureAccounting,
319 Vec<RetainedRecordCharge>,
320 u64,
321 ) {
322 (
323 self.frontiers,
324 self.closure_accounting,
325 self.retained_charges,
326 self.retained_record_limit,
327 )
328 }
329
330 #[must_use]
333 pub fn from_unchanged_record_admission<EF, V, LF>(
334 unchanged: UnchangedRecordAdmission<'_, EF, V, LF>,
335 retained_record_limit: u64,
336 ) -> (Self, RecordAdmission, ResourceVector) {
337 let (prestate, encoded_record_charge) = unchanged.into_parts();
338 let (request, frontiers, closure_accounting, retained_charges) =
339 prestate.into_live_owner_parts();
340 (
341 Self {
342 frontiers,
343 closure_accounting,
344 retained_charges,
345 retained_record_limit,
346 },
347 request,
348 encoded_record_charge,
349 )
350 }
351
352 #[must_use]
355 pub fn from_record_admission_persistence(
356 persistence: RecordAdmissionPersistenceParts,
357 retained_record_limit: u64,
358 ) -> Self {
359 Self {
360 frontiers: persistence.frontiers,
361 closure_accounting: persistence.accounting,
362 retained_charges: persistence.retained_charges,
363 retained_record_limit,
364 }
365 }
366
367 #[must_use]
369 pub fn from_marker_drain(
370 commit: MarkerDrainCommit,
371 retained_record_limit: u64,
372 ) -> (Self, StoredEdge, MarkerDeliveryProjection) {
373 let (frontiers, closure_accounting, retained_charges, successor, projection) =
374 commit.into_parts();
375 (
376 Self {
377 frontiers,
378 closure_accounting,
379 retained_charges,
380 retained_record_limit,
381 },
382 successor,
383 projection,
384 )
385 }
386
387 pub(super) fn commit_binding_terminal_candidate(
388 self,
389 active_binding: super::super::ActiveBinding,
390 admission_order: super::super::AdmissionOrder,
391 delivery_seq: crate::wire::DeliverySeq,
392 charge: RetainedRecordCharge,
393 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
394 let mut active = self.frontiers.active_identities().participants().to_vec();
395 let Some(participant) = active
396 .iter_mut()
397 .find(|participant| participant.participant_index() == active_binding.participant_id)
398 else {
399 return Err(Box::new((self, LiveFrontierError::Authority)));
400 };
401 if participant.binding() != FrontierBinding::Bound(active_binding.binding_epoch) {
402 return Err(Box::new((self, LiveFrontierError::Authority)));
403 }
404 *participant = FrontierParticipant::new(
405 participant.participant_index(),
406 participant.cursor(),
407 FrontierBinding::Detached(active_binding.binding_epoch),
408 );
409 let row = RetainedCausalRecord {
410 delivery_seq,
411 admission_order,
412 kind: RetainedCausalRecordKind::BindingTerminal(super::super::BindingTerminalOwner {
413 participant_index: active_binding.participant_id,
414 binding_epoch: active_binding.binding_epoch,
415 }),
416 };
417 let Some(sequence) = detach_sequence(self.frontiers.sequence().ledger(), delivery_seq)
418 else {
419 return Err(Box::new((self, LiveFrontierError::Frontier)));
420 };
421 let Some(order) = detach_order(
422 self.frontiers.order().ledger(),
423 admission_order.transaction_order(),
424 ) else {
425 return Err(Box::new((self, LiveFrontierError::Frontier)));
426 };
427 match transition(self, (), active, &[row], vec![charge], sequence, order) {
428 Ok(committed) => {
429 let ((), owner) = committed.into_parts();
430 Ok(owner)
431 }
432 Err(failure) => {
433 let error = failure.error();
434 let ((), owner) = failure.into_parts();
435 Err(Box::new((owner, error)))
436 }
437 }
438 }
439
440 pub(super) fn pend_binding_terminal_candidate(
441 self,
442 active_binding: super::super::ActiveBinding,
443 admission_order: super::super::AdmissionOrder,
444 delivery_seq: crate::wire::DeliverySeq,
445 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
446 let Some(order) = detach_order(
447 self.frontiers.order().ledger(),
448 admission_order.transaction_order(),
449 ) else {
450 return Err(Box::new((self, LiveFrontierError::Frontier)));
451 };
452 let Self {
453 frontiers,
454 closure_accounting,
455 retained_charges,
456 retained_record_limit,
457 } = self;
458 match frontiers.apply_pending_binding_terminal(
459 active_binding.participant_id,
460 active_binding.binding_epoch,
461 delivery_seq,
462 admission_order,
463 order,
464 ) {
465 Ok(frontiers) => Ok(Self {
466 frontiers,
467 closure_accounting,
468 retained_charges,
469 retained_record_limit,
470 }),
471 Err(failure) => {
472 let (frontiers, error) = *failure;
473 Err(Box::new((
474 Self {
475 frontiers,
476 closure_accounting,
477 retained_charges,
478 retained_record_limit,
479 },
480 map_frontier_error(error),
481 )))
482 }
483 }
484 }
485
486 pub fn drain_pending_terminal(
514 self,
515 pending: PendingFinalization,
516 terminal_charge: RetainedRecordCharge,
517 ) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
518 let Some(expected_sequence) = self
519 .frontiers
520 .sequence()
521 .ledger()
522 .high_watermark()
523 .checked_add(1)
524 else {
525 return drain_refusal(self, LiveFrontierError::Frontier);
526 };
527 if terminal_charge.delivery_seq() != expected_sequence
528 || terminal_charge.admission_order() != pending.admission_order()
529 || terminal_charge.encoded_charge().entries != 1
530 {
531 return drain_refusal(self, LiveFrontierError::RetainedCharge);
532 }
533 if self
534 .retained_charges
535 .len()
536 .checked_add(1)
537 .and_then(|len| u64::try_from(len).ok())
538 .is_none()
539 {
540 return drain_refusal(self, LiveFrontierError::RetainedRecordLimit);
541 }
542 let Some(accounting) = accounting_after_rows(self.closure_accounting, &[terminal_charge])
543 else {
544 return drain_refusal(self, LiveFrontierError::ClosureAccounting);
545 };
546 let Self {
547 frontiers,
548 closure_accounting,
549 mut retained_charges,
550 retained_record_limit,
551 } = self;
552 let expected_owner = BindingTerminalOwner {
553 participant_index: pending.participant_id(),
554 binding_epoch: pending.binding_epoch(),
555 };
556 let (frontiers, record) = match frontiers
557 .drain_first_binding_terminal(expected_owner, pending.admission_order())
558 {
559 Ok(drained) => drained,
560 Err(failure) => {
561 let (frontiers, error) = *failure;
562 return drain_refusal(
563 Self {
564 frontiers,
565 closure_accounting,
566 retained_charges,
567 retained_record_limit,
568 },
569 map_frontier_error(error),
570 );
571 }
572 };
573 retained_charges.push(terminal_charge);
574 let projection =
575 ObserverProgressProjection::new(pending.conversation_id(), record.delivery_seq);
576 Ok(DrainedPendingTerminal {
577 owner: Self {
578 frontiers,
579 closure_accounting: accounting,
580 retained_charges,
581 retained_record_limit,
582 },
583 projection,
584 })
585 }
586
587 pub fn retain_fenced_marker_source(
594 self,
595 recovery: DetachedCredentialRecovery,
596 ) -> Result<RetainedFencedMarkerSource, Box<FencedMarkerSourceRetentionRefused>> {
597 let Some(source) = self.frontiers.fenced_marker_source(recovery) else {
598 return Err(Box::new(FencedMarkerSourceRetentionRefused {
599 owner: self,
600 recovery,
601 }));
602 };
603 Ok(RetainedFencedMarkerSource {
604 owner: self,
605 recovery,
606 expectation: FencedMarkerSourceExpectation { source },
607 })
608 }
609
610 #[must_use]
618 pub fn mint_fenced_attach(
619 mut self,
620 marker_source_sequence: u64,
621 recovery: DetachedCredentialRecovery,
622 debt: ClosureDebt,
623 event: Event,
624 successor: DebtCompletion,
625 ) -> MintFencedAttachResult {
626 let Some(record) = self.frontiers.take_fenced_marker_record(recovery) else {
627 return MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
628 owner: self,
629 marker_source_sequence,
630 recovery,
631 debt,
632 event,
633 successor,
634 reason: FencedAttachMintRefusalReason::MarkerAuthority,
635 }));
636 };
637 match recovery.fenced_attach(record, debt, event, successor) {
638 Ok(proof) => MintFencedAttachResult::Minted(Box::new(MintedFencedAttach {
639 owner_without_marker_authority: self,
640 proof,
641 })),
642 Err(refusal) => {
643 self.frontiers
644 .reinstall_fenced_marker_record((*refusal).into_record());
645 MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
646 owner: self,
647 marker_source_sequence,
648 recovery,
649 debt,
650 event,
651 successor,
652 reason: FencedAttachMintRefusalReason::ProofInputs,
653 }))
654 }
655 }
656 }
657}
658
659#[derive(Debug, PartialEq, Eq)]
664pub struct DrainedPendingTerminal {
665 owner: LiveFrontierOwner,
666 projection: ObserverProgressProjection,
667}
668
669impl DrainedPendingTerminal {
670 #[must_use]
672 pub fn into_parts(self) -> (LiveFrontierOwner, ObserverProgressProjection) {
673 (self.owner, self.projection)
674 }
675}
676
677#[derive(Debug, PartialEq, Eq)]
679pub struct PendingTerminalDrainRefused {
680 owner: LiveFrontierOwner,
681 error: LiveFrontierError,
682}
683
684impl PendingTerminalDrainRefused {
685 #[must_use]
687 pub const fn error(&self) -> LiveFrontierError {
688 self.error
689 }
690
691 #[must_use]
693 pub fn into_owner(self) -> LiveFrontierOwner {
694 self.owner
695 }
696}
697
698fn drain_refusal(
699 owner: LiveFrontierOwner,
700 error: LiveFrontierError,
701) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
702 Err(Box::new(PendingTerminalDrainRefused { owner, error }))
703}
704
705#[derive(Clone, Copy, Debug, PartialEq, Eq)]
707pub struct FencedMarkerSourceExpectation {
708 source: FencedMarkerSourceRecord,
709}
710
711impl FencedMarkerSourceExpectation {
712 #[must_use]
714 pub const fn conversation_id(self) -> u64 {
715 self.source.conversation_id
716 }
717
718 #[must_use]
720 pub const fn marker_delivery_seq(self) -> u64 {
721 self.source.delivery_seq
722 }
723
724 #[must_use]
726 pub const fn admission_order(self) -> super::super::AdmissionOrder {
727 self.source.admission_order
728 }
729
730 #[must_use]
732 pub const fn participant_id(self) -> u64 {
733 self.source.participant_id
734 }
735
736 #[must_use]
738 pub const fn provenance(self) -> super::super::MarkerProvenance {
739 self.source.provenance
740 }
741
742 #[must_use]
744 pub const fn target_binding(self) -> FrontierBinding {
745 self.source.target_binding
746 }
747}
748
749#[derive(Debug, PartialEq, Eq)]
751pub struct RetainedFencedMarkerSource {
752 owner: LiveFrontierOwner,
753 recovery: DetachedCredentialRecovery,
754 expectation: FencedMarkerSourceExpectation,
755}
756
757impl RetainedFencedMarkerSource {
758 #[must_use]
760 pub const fn expectation(&self) -> FencedMarkerSourceExpectation {
761 self.expectation
762 }
763
764 #[must_use]
766 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
767 (self.owner, self.recovery)
768 }
769}
770
771#[derive(Debug, PartialEq, Eq)]
773pub struct FencedMarkerSourceRetentionRefused {
774 owner: LiveFrontierOwner,
775 recovery: DetachedCredentialRecovery,
776}
777
778impl FencedMarkerSourceRetentionRefused {
779 #[must_use]
781 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
782 (self.owner, self.recovery)
783 }
784}
785
786#[derive(Clone, Copy, Debug, PartialEq, Eq)]
788pub enum FencedAttachMintRefusalReason {
789 MarkerAuthority,
792 ProofInputs,
794}
795
796#[derive(Debug, PartialEq, Eq)]
798pub struct MintedFencedAttach {
799 owner_without_marker_authority: LiveFrontierOwner,
800 proof: FencedAttachCommit,
801}
802
803impl MintedFencedAttach {
804 #[must_use]
807 pub fn into_parts(self) -> (LiveFrontierOwner, FencedAttachCommit) {
808 (self.owner_without_marker_authority, self.proof)
809 }
810}
811
812#[derive(Debug, PartialEq, Eq)]
814pub struct MintFencedAttachRefused {
815 owner: LiveFrontierOwner,
816 marker_source_sequence: u64,
817 recovery: DetachedCredentialRecovery,
818 debt: ClosureDebt,
819 event: Event,
820 successor: DebtCompletion,
821 reason: FencedAttachMintRefusalReason,
822}
823
824impl MintFencedAttachRefused {
825 #[must_use]
827 pub const fn reason(&self) -> FencedAttachMintRefusalReason {
828 self.reason
829 }
830
831 #[must_use]
833 pub fn into_parts(
834 self,
835 ) -> (
836 LiveFrontierOwner,
837 u64,
838 DetachedCredentialRecovery,
839 ClosureDebt,
840 Event,
841 DebtCompletion,
842 ) {
843 (
844 self.owner,
845 self.marker_source_sequence,
846 self.recovery,
847 self.debt,
848 self.event,
849 self.successor,
850 )
851 }
852}
853
854#[derive(Debug, PartialEq, Eq)]
856pub enum MintFencedAttachResult {
857 Minted(Box<MintedFencedAttach>),
859 MintRefused(Box<MintFencedAttachRefused>),
861}
862
863#[derive(Debug, PartialEq, Eq)]
865pub struct LiveLeaveCommit<EF, V, LF> {
866 identity: IdentityState<EF, V, LF>,
867 owner: LiveFrontierOwner,
868}
869
870impl<EF, V, LF> LiveLeaveCommit<EF, V, LF> {
871 #[must_use]
873 pub const fn observer_progress_projection(&self) -> Option<ObserverProgressProjection> {
874 let IdentityState::Retired(retired) = &self.identity else {
875 return None;
876 };
877 let committed = retired.committed_result();
878 Some(ObserverProgressProjection::new(
879 committed.conversation_id(),
880 committed.left_delivery_seq(),
881 ))
882 }
883
884 #[must_use]
886 pub fn into_parts(self) -> (IdentityState<EF, V, LF>, LiveFrontierOwner) {
887 (self.identity, self.owner)
888 }
889}
890
891#[derive(Clone, Debug, PartialEq, Eq)]
893pub enum LiveLeaveError {
894 Prepare(PrepareLeaveAuthorityError),
896 Commit(LeaveCommitError),
898 RetainedCharge,
900 RetainedRecordLimit,
902 ClosureAccounting,
904 Identity,
906}
907
908pub fn commit_settled_leave_frontier<EF, V, LF, D>(
916 owner: LiveFrontierOwner,
917 member: LiveMember<EF>,
918 binding: BindingState,
919 detach_cell: DetachCell<D>,
920 verified: VerifiedLeaveRequest<V, LF>,
921 left_delivery_seq: u64,
922 left_charge: RetainedRecordCharge,
923) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
924 let LiveFrontierOwner {
925 frontiers,
926 closure_accounting,
927 mut retained_charges,
928 retained_record_limit,
929 } = owner;
930 let retired_marker_charge =
931 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
932 let authority = frontiers
933 .prepare_settled_leave_authority(&member, binding)
934 .map_err(LiveLeaveError::Prepare)?;
935 let commit = commit_leave(
936 member,
937 binding,
938 detach_cell,
939 verified,
940 authority,
941 LeaveCommitParameters { left_delivery_seq },
942 )
943 .map_err(LiveLeaveError::Commit)?;
944 let (identity, frontiers) = commit.into_parts();
945 let IdentityState::Retired(retired) = &identity else {
946 return Err(LiveLeaveError::Identity);
947 };
948 if left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
949 || left_charge.admission_order() != retired.left_admission_order()
950 || left_charge.encoded_charge().entries != 1
951 {
952 return Err(LiveLeaveError::RetainedCharge);
953 }
954 retained_charges.push(left_charge);
955 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
956 let retained_len = u64::try_from(frontiers.retained_records().len())
957 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
958 if retained_len > retained_record_limit
959 || retained_charges.len() != frontiers.retained_records().len()
960 {
961 return Err(LiveLeaveError::RetainedRecordLimit);
962 }
963 let closure_accounting =
964 accounting_after_leave(closure_accounting, &[left_charge], retired_marker_charge)
965 .ok_or(LiveLeaveError::ClosureAccounting)?;
966 Ok(LiveLeaveCommit {
967 identity,
968 owner: LiveFrontierOwner {
969 frontiers,
970 closure_accounting,
971 retained_charges,
972 retained_record_limit,
973 },
974 })
975}
976
977pub fn commit_pending_leave_frontier<EF, V, LF, D>(
986 owner: LiveFrontierOwner,
987 member: LiveMember<EF>,
988 pending: PendingFinalization,
989 detach_cell: DetachCell<D>,
990 verified: VerifiedLeaveRequest<V, LF>,
991 parameters: PendingLeaveCommitParameters,
992 charges: [RetainedRecordCharge; 2],
993) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
994 let [terminal_charge, left_charge] = charges;
995 let terminal_delivery_seq = parameters.terminal_delivery_seq;
996 let LiveFrontierOwner {
997 frontiers,
998 closure_accounting,
999 mut retained_charges,
1000 retained_record_limit,
1001 } = owner;
1002 let retired_marker_charge =
1003 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
1004 let authority = frontiers
1005 .prepare_pending_leave_authority(&member, pending)
1006 .map_err(LiveLeaveError::Prepare)?;
1007 let commit = commit_pending_leave(
1008 member,
1009 pending,
1010 detach_cell,
1011 verified,
1012 authority,
1013 parameters,
1014 )
1015 .map_err(LiveLeaveError::Commit)?;
1016 let (identity, frontiers) = commit.into_parts();
1017 let IdentityState::Retired(retired) = &identity else {
1018 return Err(LiveLeaveError::Identity);
1019 };
1020 if retired.committed_result().prior_terminal_delivery_seq() != Some(terminal_delivery_seq)
1021 || terminal_charge.delivery_seq() != terminal_delivery_seq
1022 || terminal_charge.admission_order() != pending.admission_order()
1023 || terminal_charge.encoded_charge().entries != 1
1024 || left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
1025 || left_charge.admission_order() != retired.left_admission_order()
1026 || left_charge.encoded_charge().entries != 1
1027 {
1028 return Err(LiveLeaveError::RetainedCharge);
1029 }
1030 retained_charges.extend([terminal_charge, left_charge]);
1031 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
1032 let retained_len = u64::try_from(frontiers.retained_records().len())
1033 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
1034 if retained_len > retained_record_limit {
1035 return Err(LiveLeaveError::RetainedRecordLimit);
1036 }
1037 if retained_charges.len() != frontiers.retained_records().len() {
1038 return Err(LiveLeaveError::RetainedCharge);
1039 }
1040 let closure_accounting = accounting_after_leave(
1041 closure_accounting,
1042 &[terminal_charge, left_charge],
1043 retired_marker_charge,
1044 )
1045 .ok_or(LiveLeaveError::ClosureAccounting)?;
1046 Ok(LiveLeaveCommit {
1047 identity,
1048 owner: LiveFrontierOwner {
1049 frontiers,
1050 closure_accounting,
1051 retained_charges,
1052 retained_record_limit,
1053 },
1054 })
1055}
1056
1057fn retired_marker_charge(
1058 frontiers: &ClaimFrontiers,
1059 retained_charges: &[RetainedRecordCharge],
1060 participant_id: crate::wire::ParticipantId,
1061) -> Result<Option<RetainedRecordCharge>, LiveLeaveError> {
1062 let marker_sequence = frontiers
1063 .retained_marker_records()
1064 .iter()
1065 .find_map(|record| {
1066 matches!(
1067 record.kind,
1068 RetainedCausalRecordKind::CompactionMarker {
1069 participant_index,
1070 ..
1071 } if participant_index == participant_id
1072 )
1073 .then_some(record.delivery_seq)
1074 });
1075 let Some(marker_sequence) = marker_sequence else {
1076 return Ok(None);
1077 };
1078 retained_charges
1079 .iter()
1080 .copied()
1081 .find(|charge| charge.delivery_seq() == marker_sequence)
1082 .map(Some)
1083 .ok_or(LiveLeaveError::RetainedCharge)
1084}
1085
1086#[derive(Debug, PartialEq, Eq)]
1088pub struct AttachFrontierCharges {
1089 terminal: Option<RetainedRecordCharge>,
1090 attached: RetainedRecordCharge,
1091 seal: LiveTransitionInputSeal,
1092}
1093
1094#[derive(Debug, PartialEq, Eq)]
1095enum LiveTransitionInputSeal {
1096 Validated,
1097}
1098
1099impl AttachFrontierCharges {
1100 #[must_use]
1102 pub const fn new(
1103 terminal: Option<RetainedRecordCharge>,
1104 attached: RetainedRecordCharge,
1105 ) -> Self {
1106 Self {
1107 terminal,
1108 attached,
1109 seal: LiveTransitionInputSeal::Validated,
1110 }
1111 }
1112
1113 const fn into_parts(self) -> (Option<RetainedRecordCharge>, RetainedRecordCharge) {
1114 let Self {
1115 terminal,
1116 attached,
1117 seal,
1118 } = self;
1119 match seal {
1120 LiveTransitionInputSeal::Validated => (terminal, attached),
1121 }
1122 }
1123}
1124
1125#[derive(Debug, PartialEq, Eq)]
1127pub struct LiveFrontierCommit<T> {
1128 operation: T,
1129 owner: LiveFrontierOwner,
1130}
1131
1132impl<T> LiveFrontierCommit<T> {
1133 #[must_use]
1135 pub const fn operation(&self) -> &T {
1136 &self.operation
1137 }
1138
1139 #[must_use]
1141 pub const fn owner(&self) -> &LiveFrontierOwner {
1142 &self.owner
1143 }
1144
1145 #[must_use]
1147 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1148 (self.operation, self.owner)
1149 }
1150}
1151
1152#[derive(Debug, PartialEq, Eq)]
1154pub struct LiveFrontierFailure<T> {
1155 error: LiveFrontierError,
1156 operation: T,
1157 owner: LiveFrontierOwner,
1158}
1159
1160impl<T> LiveFrontierFailure<T> {
1161 #[must_use]
1163 pub const fn error(&self) -> LiveFrontierError {
1164 self.error
1165 }
1166
1167 #[must_use]
1169 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1170 (self.operation, self.owner)
1171 }
1172}
1173
1174#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1176pub enum LiveFrontierError {
1177 Authority,
1179 Precedence(PrecedenceCondition),
1187 RetainedCharge,
1189 RetainedRecordLimit,
1191 Frontier,
1193 ClosureAccounting,
1195}
1196
1197pub type LiveFrontierResult<T> = Result<LiveFrontierCommit<T>, Box<LiveFrontierFailure<T>>>;
1199
1200pub fn apply_enrollment_frontier<F>(
1206 owner: LiveFrontierOwner,
1207 operation: super::super::EnrollmentCommit<F>,
1208 attached_charge: RetainedRecordCharge,
1209) -> LiveFrontierResult<super::super::EnrollmentCommit<F>> {
1210 let attached = operation.attached;
1211 if attached.conversation_id() != owner.frontiers.conversation_id() {
1212 return failure(owner, operation, LiveFrontierError::Authority);
1213 }
1214 let participant_id = attached.participant_id();
1215 let mut active = owner.frontiers.active_identities().participants().to_vec();
1216 if active
1217 .iter()
1218 .any(|participant| participant.participant_index() == participant_id)
1219 {
1220 return failure(owner, operation, LiveFrontierError::Authority);
1221 }
1222 active.push(FrontierParticipant::new(
1223 participant_id,
1224 operation.member.cursor(),
1225 FrontierBinding::Bound(attached.binding_epoch()),
1226 ));
1227 active.sort_unstable_by_key(|participant| participant.participant_index());
1228 let rows = [retained_attached(attached)];
1229 let Some(sequence) =
1230 enrollment_sequence(owner.frontiers.sequence().ledger(), attached.delivery_seq())
1231 else {
1232 return failure(owner, operation, LiveFrontierError::Frontier);
1233 };
1234 let Some(order) = enrollment_order(
1235 owner.frontiers.order().ledger(),
1236 attached.admission_order().transaction_order(),
1237 ) else {
1238 return failure(owner, operation, LiveFrontierError::Frontier);
1239 };
1240 transition(
1241 owner,
1242 operation,
1243 active,
1244 &rows,
1245 vec![attached_charge],
1246 sequence,
1247 order,
1248 )
1249}
1250
1251pub fn apply_attach_frontier<F, V>(
1258 owner: LiveFrontierOwner,
1259 operation: AttachCommit<F, V>,
1260 charges: AttachFrontierCharges,
1261) -> LiveFrontierResult<AttachCommit<F, V>> {
1262 let (terminal_charge, attached_charge) = charges.into_parts();
1263 let attached = operation.attached;
1264 if attached.conversation_id() != owner.frontiers.conversation_id() {
1265 return failure(owner, operation, LiveFrontierError::Authority);
1266 }
1267 let mut active = owner.frontiers.active_identities().participants().to_vec();
1268 let Some(participant) = active
1269 .iter_mut()
1270 .find(|participant| participant.participant_index() == attached.participant_id())
1271 else {
1272 return failure(owner, operation, LiveFrontierError::Authority);
1273 };
1274 let frontier_cursor = participant.cursor();
1275 *participant = FrontierParticipant::new(
1276 participant.participant_index(),
1277 operation.member.cursor(),
1278 FrontierBinding::Bound(attached.binding_epoch()),
1279 );
1280 let reopened = reopened_marker_anchors(
1281 &owner,
1282 attached.participant_id(),
1283 frontier_cursor,
1284 operation.member.cursor(),
1285 );
1286 with_reopened_marker_anchors(owner, operation, reopened, |owner, operation| {
1287 apply_attach_rows(owner, operation, active, terminal_charge, attached_charge)
1288 })
1289}
1290
1291fn apply_attach_rows<F, V>(
1293 owner: LiveFrontierOwner,
1294 operation: AttachCommit<F, V>,
1295 active: Vec<FrontierParticipant>,
1296 terminal_charge: Option<RetainedRecordCharge>,
1297 attached_charge: RetainedRecordCharge,
1298) -> LiveFrontierResult<AttachCommit<F, V>> {
1299 let attached = operation.attached;
1300 let current_sequence = owner.frontiers.sequence().ledger();
1301 let current_order = owner.frontiers.order().ledger();
1302 let (rows, keyed_charges, sequence, order) = match operation.transition {
1303 AttachTransition::Detached => {
1304 if terminal_charge.is_some() {
1305 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1306 }
1307 let Some(sequence) =
1308 detached_attach_sequence(current_sequence, attached.delivery_seq())
1309 else {
1310 return failure(owner, operation, LiveFrontierError::Frontier);
1311 };
1312 let Some(order) = detached_attach_order(
1313 current_order,
1314 attached.admission_order().transaction_order(),
1315 ) else {
1316 return failure(owner, operation, LiveFrontierError::Frontier);
1317 };
1318 (
1319 vec![retained_attached(attached)],
1320 vec![attached_charge],
1321 sequence,
1322 order,
1323 )
1324 }
1325 AttachTransition::Superseded { terminal } => {
1326 let Some(terminal_charge) = terminal_charge else {
1327 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1328 };
1329 let rows = vec![
1330 retained_terminal(terminal.into()),
1331 retained_attached(attached),
1332 ];
1333 let Some(sequence) = superseding_attach_sequence(current_sequence, &rows) else {
1334 return failure(owner, operation, LiveFrontierError::Frontier);
1335 };
1336 let Some(order) = superseding_attach_order(
1337 current_order,
1338 attached.admission_order().transaction_order(),
1339 ) else {
1340 return failure(owner, operation, LiveFrontierError::Frontier);
1341 };
1342 (
1343 rows,
1344 vec![terminal_charge, attached_charge],
1345 sequence,
1346 order,
1347 )
1348 }
1349 AttachTransition::FencedRecovery {
1350 prior_binding_epoch,
1351 composed_terminal,
1352 next_closure_state,
1353 } => {
1354 return apply_fenced_attach_frontier(
1355 owner,
1356 operation,
1357 terminal_charge,
1358 attached_charge,
1359 prior_binding_epoch,
1360 composed_terminal,
1361 next_closure_state,
1362 );
1363 }
1364 };
1365 transition(
1366 owner,
1367 operation,
1368 active,
1369 &rows,
1370 keyed_charges,
1371 sequence,
1372 order,
1373 )
1374}
1375
1376pub fn apply_detach_frontier<EF, V>(
1382 owner: LiveFrontierOwner,
1383 operation: CommittedDetachTransition<EF, V>,
1384 terminal_charge: RetainedRecordCharge,
1385) -> LiveFrontierResult<CommittedDetachTransition<EF, V>> {
1386 let terminal = operation.terminal();
1387 if terminal.conversation_id() != owner.frontiers.conversation_id() {
1388 return failure(owner, operation, LiveFrontierError::Authority);
1389 }
1390 let mut active = owner.frontiers.active_identities().participants().to_vec();
1391 let Some(participant) = active
1392 .iter_mut()
1393 .find(|participant| participant.participant_index() == terminal.participant_id())
1394 else {
1395 return failure(owner, operation, LiveFrontierError::Authority);
1396 };
1397 *participant = FrontierParticipant::new(
1398 participant.participant_index(),
1399 operation.member().cursor(),
1400 FrontierBinding::Detached(terminal.binding_epoch()),
1401 );
1402 let row = retained_terminal(terminal.into());
1403 let Some(sequence) = detach_sequence(owner.frontiers.sequence().ledger(), row.delivery_seq)
1404 else {
1405 return failure(owner, operation, LiveFrontierError::Frontier);
1406 };
1407 let Some(order) = detach_order(
1408 owner.frontiers.order().ledger(),
1409 row.admission_order.transaction_order(),
1410 ) else {
1411 return failure(owner, operation, LiveFrontierError::Frontier);
1412 };
1413 transition(
1414 owner,
1415 operation,
1416 active,
1417 &[row],
1418 vec![terminal_charge],
1419 sequence,
1420 order,
1421 )
1422}
1423
1424fn crossed_marker_anchors(
1434 owner: &LiveFrontierOwner,
1435 participant_id: crate::wire::ParticipantId,
1436 previous_cursor: crate::wire::DeliverySeq,
1437 through_seq: crate::wire::DeliverySeq,
1438) -> u64 {
1439 owner
1440 .frontiers
1441 .retained_marker_records()
1442 .iter()
1443 .filter(|record| {
1444 matches!(
1445 record.kind,
1446 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1447 if participant_index == participant_id
1448 ) && previous_cursor < record.delivery_seq
1449 && record.delivery_seq <= through_seq
1450 })
1451 .fold(0_u64, |count, _| count.saturating_add(1))
1452}
1453
1454pub fn apply_participant_ack_frontier(
1468 mut owner: LiveFrontierOwner,
1469 operation: ParticipantAckCommit,
1470) -> LiveFrontierResult<ParticipantAckCommit> {
1471 let request = operation.outcome().request();
1472 let Some(current) = owner
1473 .frontiers
1474 .active_identities()
1475 .participants()
1476 .iter()
1477 .find(|participant| participant.participant_index() == request.participant_id)
1478 .copied()
1479 else {
1480 return failure(owner, operation, LiveFrontierError::Authority);
1481 };
1482 let crossed = crossed_marker_anchors(
1483 &owner,
1484 request.participant_id,
1485 current.cursor(),
1486 request.through_seq,
1487 );
1488 let accounting = if crossed == 0 {
1489 None
1490 } else {
1491 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1492 Some(accounting) => Some(accounting),
1493 None => {
1494 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1495 }
1496 }
1497 };
1498 let participant = FrontierParticipant::new(
1499 request.participant_id,
1500 request.through_seq,
1501 current.binding(),
1502 );
1503 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1504 Ok(frontiers) => frontiers,
1505 Err(frontier_failure) => {
1506 let (frontiers, error) = *frontier_failure;
1507 owner.frontiers = frontiers;
1508 return failure(owner, operation, map_frontier_error(error));
1509 }
1510 };
1511 if let Some(accounting) = accounting {
1512 owner.closure_accounting = accounting;
1513 }
1514 Ok(LiveFrontierCommit { operation, owner })
1515}
1516
1517pub fn apply_nonzero_participant_ack_frontier(
1527 mut owner: LiveFrontierOwner,
1528 operation: NonzeroParticipantAckCommit,
1529) -> LiveFrontierResult<NonzeroParticipantAckCommit> {
1530 let request = operation.outcome().request();
1531 let Some(current) = owner
1532 .frontiers
1533 .active_identities()
1534 .participants()
1535 .iter()
1536 .find(|participant| participant.participant_index() == request.participant_id)
1537 .copied()
1538 else {
1539 return failure(owner, operation, LiveFrontierError::Authority);
1540 };
1541 let crossed = crossed_marker_anchors(
1544 &owner,
1545 request.participant_id,
1546 current.cursor(),
1547 request.through_seq,
1548 );
1549 let accounting = if crossed == 0 {
1550 None
1551 } else {
1552 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1553 Some(accounting) => Some(accounting),
1554 None => {
1555 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1556 }
1557 }
1558 };
1559 let participant = FrontierParticipant::new(
1560 request.participant_id,
1561 request.through_seq,
1562 current.binding(),
1563 );
1564 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1565 Ok(frontiers) => frontiers,
1566 Err(frontier_failure) => {
1567 let (frontiers, error) = *frontier_failure;
1568 owner.frontiers = frontiers;
1569 return failure(owner, operation, map_frontier_error(error));
1570 }
1571 };
1572 if let Some(accounting) = accounting {
1573 owner.closure_accounting = accounting;
1574 }
1575 Ok(LiveFrontierCommit { operation, owner })
1576}
1577
1578pub fn apply_marker_ack_frontier(
1584 mut owner: LiveFrontierOwner,
1585 operation: MarkerAckCommit,
1586) -> LiveFrontierResult<MarkerAckCommit> {
1587 let request = operation.outcome().request();
1588 if !owner
1589 .frontiers
1590 .retained_marker_records()
1591 .iter()
1592 .any(|record| {
1593 record.delivery_seq == request.marker_delivery_seq
1594 && matches!(
1595 record.kind,
1596 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1597 if participant_index == request.participant_id
1598 )
1599 })
1600 {
1601 return failure(owner, operation, LiveFrontierError::Authority);
1602 }
1603 let Some(current) = owner
1604 .frontiers
1605 .active_identities()
1606 .participants()
1607 .iter()
1608 .find(|participant| participant.participant_index() == request.participant_id)
1609 .copied()
1610 else {
1611 return failure(owner, operation, LiveFrontierError::Authority);
1612 };
1613 let Some(accounting) = accounting_after_marker_ack(owner.closure_accounting) else {
1614 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1615 };
1616 let participant = FrontierParticipant::new(
1617 request.participant_id,
1618 request.marker_delivery_seq,
1619 current.binding(),
1620 );
1621 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1622 Ok(frontiers) => frontiers,
1623 Err(frontier_failure) => {
1624 let (frontiers, error) = *frontier_failure;
1625 owner.frontiers = frontiers;
1626 return failure(owner, operation, map_frontier_error(error));
1627 }
1628 };
1629 owner.closure_accounting = accounting;
1630 Ok(LiveFrontierCommit { operation, owner })
1631}
1632
1633fn apply_fenced_attach_frontier<F, V>(
1634 owner: LiveFrontierOwner,
1635 operation: AttachCommit<F, V>,
1636 terminal_charge: Option<RetainedRecordCharge>,
1637 attached_charge: RetainedRecordCharge,
1638 prior_binding_epoch: crate::wire::BindingEpoch,
1639 composed_terminal: Option<super::super::CommittedBindingTerminal>,
1640 next_closure_state: super::super::ClosureState,
1641) -> LiveFrontierResult<AttachCommit<F, V>> {
1642 let attached = operation.attached;
1643 let (rows, charges) = match (composed_terminal, terminal_charge) {
1644 (None, None) => (vec![retained_attached(attached)], vec![attached_charge]),
1645 (Some(terminal), Some(terminal_charge)) => (
1646 vec![retained_terminal(terminal), retained_attached(attached)],
1647 vec![terminal_charge, attached_charge],
1648 ),
1649 (None, Some(_)) | (Some(_), None) => {
1650 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1651 }
1652 };
1653 let participant = FrontierParticipant::new(
1654 attached.participant_id(),
1655 operation.member.cursor(),
1656 FrontierBinding::Bound(attached.binding_epoch()),
1657 );
1658 fenced_attach_transition(
1659 owner,
1660 operation,
1661 participant,
1662 prior_binding_epoch,
1663 next_closure_state,
1664 &rows,
1665 charges,
1666 )
1667}
1668
1669fn reopened_marker_anchors(
1687 owner: &LiveFrontierOwner,
1688 participant_id: crate::wire::ParticipantId,
1689 frontier_cursor: crate::wire::DeliverySeq,
1690 member_cursor: crate::wire::DeliverySeq,
1691) -> u64 {
1692 if member_cursor >= frontier_cursor {
1693 return 0;
1694 }
1695 crossed_marker_anchors(owner, participant_id, member_cursor, frontier_cursor)
1696}
1697
1698fn with_reopened_marker_anchors<T>(
1701 mut owner: LiveFrontierOwner,
1702 operation: T,
1703 reopened: u64,
1704 apply: impl FnOnce(LiveFrontierOwner, T) -> LiveFrontierResult<T>,
1705) -> LiveFrontierResult<T> {
1706 if reopened == 0 {
1707 return apply(owner, operation);
1708 }
1709 let unchanged = owner.closure_accounting;
1710 let Some(accounting) = accounting_after_marker_remint(unchanged, reopened) else {
1711 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1712 };
1713 owner.closure_accounting = accounting;
1714 apply(owner, operation).map_err(|mut refused| {
1715 refused.owner.closure_accounting = unchanged;
1716 refused
1717 })
1718}
1719
1720fn fenced_attach_transition<T>(
1721 mut owner: LiveFrontierOwner,
1722 operation: T,
1723 participant: FrontierParticipant,
1724 prior_binding_epoch: crate::wire::BindingEpoch,
1725 next_closure_state: super::super::ClosureState,
1726 rows: &[RetainedCausalRecord],
1727 charges: Vec<RetainedRecordCharge>,
1728) -> LiveFrontierResult<T> {
1729 if rows.len() != charges.len()
1730 || rows.iter().zip(&charges).any(|(row, charge)| {
1731 row.delivery_seq != charge.delivery_seq()
1732 || row.admission_order != charge.admission_order()
1733 || charge.encoded_charge().entries != 1
1734 })
1735 {
1736 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1737 }
1738 let resulting_len = owner
1739 .frontiers
1740 .retained_records()
1741 .len()
1742 .checked_add(rows.len());
1743 if resulting_len
1744 .and_then(|len| u64::try_from(len).ok())
1745 .is_none_or(|len| len > owner.retained_record_limit)
1746 {
1747 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1748 }
1749 let Some(accounting) =
1750 accounting_after_fenced_attach(owner.closure_accounting, &charges, next_closure_state)
1751 else {
1752 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1753 };
1754 owner.frontiers =
1755 match owner
1756 .frontiers
1757 .apply_live_fenced_attach(participant, prior_binding_epoch, rows)
1758 {
1759 Ok(frontiers) => frontiers,
1760 Err(frontier_failure) => {
1761 let (frontiers, error) = *frontier_failure;
1762 owner.frontiers = frontiers;
1763 return failure(owner, operation, map_frontier_error(error));
1764 }
1765 };
1766 owner.retained_charges.extend(charges);
1767 owner
1768 .retained_charges
1769 .sort_unstable_by_key(|charge| charge.delivery_seq());
1770 owner.closure_accounting = accounting;
1771 Ok(LiveFrontierCommit { operation, owner })
1772}
1773
1774fn transition<T>(
1775 mut owner: LiveFrontierOwner,
1776 operation: T,
1777 active: Vec<FrontierParticipant>,
1778 rows: &[RetainedCausalRecord],
1779 charges: Vec<RetainedRecordCharge>,
1780 sequence: SequenceLedger,
1781 order: OrderLedger,
1782) -> LiveFrontierResult<T> {
1783 if rows.len() != charges.len()
1784 || rows.iter().zip(&charges).any(|(row, charge)| {
1785 row.delivery_seq != charge.delivery_seq()
1786 || row.admission_order != charge.admission_order()
1787 || charge.encoded_charge().entries != 1
1788 })
1789 {
1790 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1791 }
1792 let resulting_len = owner
1793 .frontiers
1794 .retained_records()
1795 .len()
1796 .checked_add(rows.len());
1797 if resulting_len
1798 .and_then(|len| u64::try_from(len).ok())
1799 .is_none_or(|len| len > owner.retained_record_limit)
1800 {
1801 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1802 }
1803 let Some(accounting) = accounting_after_rows(owner.closure_accounting, &charges) else {
1804 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1805 };
1806 owner.frontiers = match owner
1807 .frontiers
1808 .apply_live_transition(active, rows, sequence, order)
1809 {
1810 Ok(frontiers) => frontiers,
1811 Err(frontier_failure) => {
1812 let (frontiers, error) = *frontier_failure;
1813 owner.frontiers = frontiers;
1814 return failure(owner, operation, map_frontier_error(error));
1815 }
1816 };
1817 owner.retained_charges.extend(charges);
1818 owner.closure_accounting = accounting;
1819 Ok(LiveFrontierCommit { operation, owner })
1820}
1821
1822const fn map_frontier_error(error: LiveFrontierTransitionError) -> LiveFrontierError {
1823 match error {
1824 LiveFrontierTransitionError::Authority => LiveFrontierError::Authority,
1825 LiveFrontierTransitionError::Precedence(condition) => {
1826 LiveFrontierError::Precedence(condition)
1827 }
1828 LiveFrontierTransitionError::RecordPosition
1829 | LiveFrontierTransitionError::Exhausted
1830 | LiveFrontierTransitionError::ResultingFrontier => LiveFrontierError::Frontier,
1831 }
1832}
1833
1834fn failure<T, U>(
1835 owner: LiveFrontierOwner,
1836 operation: T,
1837 error: LiveFrontierError,
1838) -> Result<U, Box<LiveFrontierFailure<T>>> {
1839 Err(Box::new(LiveFrontierFailure {
1840 error,
1841 operation,
1842 owner,
1843 }))
1844}