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 checkpoint_restore;
34pub use checkpoint_restore::{
35 CheckpointParticipant, FrontierCheckpointSource, LiveFrontierRestoreError,
36 RestoredLiveFrontierOwner,
37};
38mod ledger;
39mod state;
40pub(super) use binding_fate_transition::BindingFateOwnerPlan;
41use ledger::{
42 detach_order, detach_sequence, detached_attach_order, detached_attach_sequence,
43 enrollment_order, enrollment_sequence, superseding_attach_order, superseding_attach_sequence,
44};
45use state::{
46 accounting_after_fenced_attach, accounting_after_leave, accounting_after_marker_ack,
47 accounting_after_marker_crossings, accounting_after_marker_remint, accounting_after_rows,
48 retained_attached, retained_terminal,
49};
50
51#[derive(Debug, PartialEq, Eq)]
76pub struct LiveFrontierOwner {
77 frontiers: ClaimFrontiers,
78 closure_accounting: ClosureAccounting,
79 retained_charges: Vec<RetainedRecordCharge>,
80 retained_record_limit: u64,
81}
82
83#[derive(Clone, Copy, Debug, PartialEq, Eq)]
97pub enum MarkerAnchorReconcile {
98 InStep,
100 RetiredOrphans(u64),
103 RemintedDeficit(u64),
106 Refused {
109 census: u64,
111 stored: u64,
113 },
114}
115
116impl MarkerAnchorReconcile {
117 #[must_use]
119 pub const fn reconciled(self) -> u64 {
120 match self {
121 Self::InStep | Self::Refused { .. } => 0,
122 Self::RetiredOrphans(count) | Self::RemintedDeficit(count) => count,
123 }
124 }
125}
126
127impl LiveFrontierOwner {
128 pub(super) fn retain_ordinary_cursor_release(
129 &mut self,
130 source: Box<crate::lifecycle::storage::OrdinaryBindingFateRestore>,
131 ) {
132 self.frontiers.retain_ordinary_cursor_release(source);
133 }
134
135 #[must_use]
137 pub fn encode_closure_accounting_source(&self) -> Vec<u8> {
138 super::super::checkpoint_codec::encode_accounting(&self.closure_accounting)
139 }
140
141 pub fn decode_closure_accounting_source(
147 state: super::super::ClosureState,
148 bytes: &[u8],
149 ) -> Result<ClosureAccounting, super::super::FrontierAccountingDecodeError> {
150 super::super::checkpoint_codec::decode_accounting(state, bytes)
151 }
152
153 #[must_use]
155 pub fn encode_retained_charge_source(&self) -> Option<Vec<u8>> {
156 super::super::checkpoint_codec::encode_charges(&self.retained_charges)
157 }
158
159 pub fn decode_retained_charge_source(
165 bytes: &[u8],
166 ) -> Result<Vec<RetainedRecordCharge>, super::super::CapsuleDecodeError> {
167 super::super::checkpoint_codec::decode_charges(bytes)
168 }
169
170 #[must_use]
172 pub fn from_initial_enrollment<F>(
173 initial: InitialEnrollmentFrontierCommit<F>,
174 retained_record_limit: u64,
175 ) -> (InitialEnrollmentOperationCommit<F>, Self) {
176 let (operation, frontiers, closure_accounting, attached_charge) =
177 initial.into_conversation_parts();
178 let attached = operation.enrollment().attached;
179 let retained_charges = vec![RetainedRecordCharge::new(
180 attached.delivery_seq(),
181 attached.admission_order(),
182 attached_charge,
183 )];
184 (
185 operation,
186 Self {
187 frontiers,
188 closure_accounting,
189 retained_charges,
190 retained_record_limit,
191 },
192 )
193 }
194
195 #[cfg(any(test, feature = "test-support"))]
206 #[must_use]
207 pub const fn from_test_parts(
208 frontiers: ClaimFrontiers,
209 closure_accounting: ClosureAccounting,
210 retained_charges: Vec<RetainedRecordCharge>,
211 retained_record_limit: u64,
212 ) -> Self {
213 Self {
214 frontiers,
215 closure_accounting,
216 retained_charges,
217 retained_record_limit,
218 }
219 }
220
221 #[cfg(any(test, feature = "test-support"))]
229 pub fn with_pending_finalizer_test_capacity(
230 mut self,
231 finalizer_rows: u64,
232 finalizer_charge: ResourceVector,
233 ) -> Result<Self, &'static str> {
234 let retained_count = u64::try_from(self.retained_charges.len())
235 .map_err(|_| "retained record count exceeds u64")?;
236 self.retained_record_limit = retained_count
237 .checked_add(finalizer_rows)
238 .ok_or("pending finalizer retained-record capacity overflow")?;
239 let current = self.closure_accounting;
240 let configured = current.configured_cap();
241 let configured = ResourceVector::new(
242 configured
243 .entries
244 .checked_add(finalizer_charge.entries)
245 .ok_or("pending finalizer entry capacity overflow")?,
246 configured
247 .bytes
248 .checked_add(finalizer_charge.bytes)
249 .ok_or("pending finalizer byte capacity overflow")?,
250 );
251 self.closure_accounting = ClosureAccounting::try_new(
252 current.state(),
253 current.marker_capacity_credits(),
254 current.marker_anchors(),
255 current.edge_sequence_claims(),
256 current.edge_order_position_claims(),
257 current.edge_k_remaining(),
258 current.baseline(),
259 configured,
260 current.episode_churn_used(),
261 current.episode_churn_limit(),
262 )
263 .map_err(|_| "pending finalizer closure capacity extension refused")?;
264 Ok(self)
265 }
266
267 #[must_use]
269 pub const fn frontiers(&self) -> &ClaimFrontiers {
270 &self.frontiers
271 }
272
273 #[must_use]
275 pub const fn closure_accounting(&self) -> ClosureAccounting {
276 self.closure_accounting
277 }
278
279 #[must_use]
281 pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
282 &self.retained_charges
283 }
284
285 #[must_use]
287 pub const fn retained_record_limit(&self) -> u64 {
288 self.retained_record_limit
289 }
290
291 #[must_use]
333 pub fn reconcile_marker_anchor_ledgers(&mut self) -> MarkerAnchorReconcile {
334 let census = self
335 .frontiers
336 .unaccepted_marker_anchor_count()
337 .saturating_add(self.frontiers.pending_marker_candidate_count());
338 let stored = self.closure_accounting.marker_anchors();
339 let repaired = match census.cmp(&stored) {
340 Ordering::Equal => return MarkerAnchorReconcile::InStep,
341 Ordering::Less => stored.checked_sub(census).and_then(|orphaned| {
342 accounting_after_marker_crossings(self.closure_accounting, orphaned)
343 .map(|accounting| (accounting, MarkerAnchorReconcile::RetiredOrphans(orphaned)))
344 }),
345 Ordering::Greater => census.checked_sub(stored).and_then(|minted| {
346 accounting_after_marker_remint(self.closure_accounting, minted)
347 .map(|accounting| (accounting, MarkerAnchorReconcile::RemintedDeficit(minted)))
348 }),
349 };
350 match repaired {
351 Some((accounting, outcome)) => {
352 self.closure_accounting = accounting;
353 outcome
354 }
355 None => MarkerAnchorReconcile::Refused { census, stored },
356 }
357 }
358
359 #[must_use]
361 pub fn into_parts(
362 self,
363 ) -> (
364 ClaimFrontiers,
365 ClosureAccounting,
366 Vec<RetainedRecordCharge>,
367 u64,
368 ) {
369 (
370 self.frontiers,
371 self.closure_accounting,
372 self.retained_charges,
373 self.retained_record_limit,
374 )
375 }
376
377 #[must_use]
380 pub fn from_unchanged_record_admission<EF, V, LF>(
381 unchanged: UnchangedRecordAdmission<'_, EF, V, LF>,
382 retained_record_limit: u64,
383 ) -> (Self, RecordAdmission, ResourceVector) {
384 let (prestate, encoded_record_charge) = unchanged.into_parts();
385 let (request, frontiers, closure_accounting, retained_charges) =
386 prestate.into_live_owner_parts();
387 (
388 Self {
389 frontiers,
390 closure_accounting,
391 retained_charges,
392 retained_record_limit,
393 },
394 request,
395 encoded_record_charge,
396 )
397 }
398
399 #[must_use]
402 pub fn from_record_admission_persistence(
403 persistence: RecordAdmissionPersistenceParts,
404 retained_record_limit: u64,
405 ) -> Self {
406 Self {
407 frontiers: persistence.frontiers,
408 closure_accounting: persistence.accounting,
409 retained_charges: persistence.retained_charges,
410 retained_record_limit,
411 }
412 }
413
414 #[must_use]
416 pub fn from_marker_drain(
417 commit: MarkerDrainCommit,
418 retained_record_limit: u64,
419 ) -> (Self, StoredEdge, MarkerDeliveryProjection) {
420 let (frontiers, closure_accounting, retained_charges, successor, projection) =
421 commit.into_parts();
422 (
423 Self {
424 frontiers,
425 closure_accounting,
426 retained_charges,
427 retained_record_limit,
428 },
429 successor,
430 projection,
431 )
432 }
433
434 pub(super) fn commit_binding_terminal_candidate(
435 self,
436 active_binding: super::super::ActiveBinding,
437 admission_order: super::super::AdmissionOrder,
438 delivery_seq: crate::wire::DeliverySeq,
439 charge: RetainedRecordCharge,
440 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
441 let mut active = self.frontiers.active_identities().participants().to_vec();
442 let Some(participant) = active
443 .iter_mut()
444 .find(|participant| participant.participant_index() == active_binding.participant_id)
445 else {
446 return Err(Box::new((self, LiveFrontierError::Authority)));
447 };
448 if participant.binding() != FrontierBinding::Bound(active_binding.binding_epoch) {
449 return Err(Box::new((self, LiveFrontierError::Authority)));
450 }
451 *participant = FrontierParticipant::new(
452 participant.participant_index(),
453 participant.cursor(),
454 FrontierBinding::Detached(active_binding.binding_epoch),
455 );
456 let row = RetainedCausalRecord {
457 delivery_seq,
458 admission_order,
459 kind: RetainedCausalRecordKind::BindingTerminal(super::super::BindingTerminalOwner {
460 participant_index: active_binding.participant_id,
461 binding_epoch: active_binding.binding_epoch,
462 }),
463 };
464 let Some(sequence) = detach_sequence(self.frontiers.sequence().ledger(), delivery_seq)
465 else {
466 return Err(Box::new((self, LiveFrontierError::Frontier)));
467 };
468 let Some(order) = detach_order(
469 self.frontiers.order().ledger(),
470 admission_order.transaction_order(),
471 ) else {
472 return Err(Box::new((self, LiveFrontierError::Frontier)));
473 };
474 match transition(self, (), active, &[row], vec![charge], sequence, order) {
475 Ok(committed) => {
476 let ((), owner) = committed.into_parts();
477 Ok(owner)
478 }
479 Err(failure) => {
480 let error = failure.error();
481 let ((), owner) = failure.into_parts();
482 Err(Box::new((owner, error)))
483 }
484 }
485 }
486
487 pub(super) fn pend_binding_terminal_candidate(
488 self,
489 active_binding: super::super::ActiveBinding,
490 admission_order: super::super::AdmissionOrder,
491 delivery_seq: crate::wire::DeliverySeq,
492 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
493 let Some(order) = detach_order(
494 self.frontiers.order().ledger(),
495 admission_order.transaction_order(),
496 ) else {
497 return Err(Box::new((self, LiveFrontierError::Frontier)));
498 };
499 let Self {
500 frontiers,
501 closure_accounting,
502 retained_charges,
503 retained_record_limit,
504 } = self;
505 match frontiers.apply_pending_binding_terminal(
506 active_binding.participant_id,
507 active_binding.binding_epoch,
508 delivery_seq,
509 admission_order,
510 order,
511 ) {
512 Ok(frontiers) => Ok(Self {
513 frontiers,
514 closure_accounting,
515 retained_charges,
516 retained_record_limit,
517 }),
518 Err(failure) => {
519 let (frontiers, error) = *failure;
520 Err(Box::new((
521 Self {
522 frontiers,
523 closure_accounting,
524 retained_charges,
525 retained_record_limit,
526 },
527 map_frontier_error(error),
528 )))
529 }
530 }
531 }
532
533 pub fn drain_pending_terminal(
561 self,
562 pending: PendingFinalization,
563 terminal_charge: RetainedRecordCharge,
564 ) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
565 let Some(expected_sequence) = self
566 .frontiers
567 .sequence()
568 .ledger()
569 .high_watermark()
570 .checked_add(1)
571 else {
572 return drain_refusal(self, LiveFrontierError::Frontier);
573 };
574 if terminal_charge.delivery_seq() != expected_sequence
575 || terminal_charge.admission_order() != pending.admission_order()
576 || terminal_charge.encoded_charge().entries != 1
577 {
578 return drain_refusal(self, LiveFrontierError::RetainedCharge);
579 }
580 if self
581 .retained_charges
582 .len()
583 .checked_add(1)
584 .and_then(|len| u64::try_from(len).ok())
585 .is_none()
586 {
587 return drain_refusal(self, LiveFrontierError::RetainedRecordLimit);
588 }
589 let Some(accounting) = accounting_after_rows(self.closure_accounting, &[terminal_charge])
590 else {
591 return drain_refusal(self, LiveFrontierError::ClosureAccounting);
592 };
593 let Self {
594 frontiers,
595 closure_accounting,
596 mut retained_charges,
597 retained_record_limit,
598 } = self;
599 let expected_owner = BindingTerminalOwner {
600 participant_index: pending.participant_id(),
601 binding_epoch: pending.binding_epoch(),
602 };
603 let (frontiers, record) = match frontiers
604 .drain_first_binding_terminal(expected_owner, pending.admission_order())
605 {
606 Ok(drained) => drained,
607 Err(failure) => {
608 let (frontiers, error) = *failure;
609 return drain_refusal(
610 Self {
611 frontiers,
612 closure_accounting,
613 retained_charges,
614 retained_record_limit,
615 },
616 map_frontier_error(error),
617 );
618 }
619 };
620 retained_charges.push(terminal_charge);
621 let projection = ObserverProgressProjection::sourced(
622 pending.conversation_id(),
623 record.delivery_seq,
624 super::super::checkpoint_codec::ObserverProjectionSource::Terminal(
625 pending.commit(record.delivery_seq).storage_description(),
626 ),
627 );
628 Ok(DrainedPendingTerminal {
629 owner: Self {
630 frontiers,
631 closure_accounting: accounting,
632 retained_charges,
633 retained_record_limit,
634 },
635 projection,
636 })
637 }
638
639 pub fn retain_fenced_marker_source(
646 self,
647 recovery: DetachedCredentialRecovery,
648 ) -> Result<RetainedFencedMarkerSource, Box<FencedMarkerSourceRetentionRefused>> {
649 let Some(source) = self.frontiers.fenced_marker_source(recovery) else {
650 return Err(Box::new(FencedMarkerSourceRetentionRefused {
651 owner: self,
652 recovery,
653 }));
654 };
655 Ok(RetainedFencedMarkerSource {
656 owner: self,
657 recovery,
658 expectation: FencedMarkerSourceExpectation { source },
659 })
660 }
661
662 #[must_use]
670 pub fn mint_fenced_attach(
671 mut self,
672 marker_source_sequence: u64,
673 recovery: DetachedCredentialRecovery,
674 debt: ClosureDebt,
675 event: Event,
676 successor: DebtCompletion,
677 ) -> MintFencedAttachResult {
678 let Some(record) = self.frontiers.take_fenced_marker_record(recovery) else {
679 return MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
680 owner: self,
681 marker_source_sequence,
682 recovery,
683 debt,
684 event,
685 successor,
686 reason: FencedAttachMintRefusalReason::MarkerAuthority,
687 }));
688 };
689 let marker_source_bytes = super::super::checkpoint_codec::encode_marker_source(&record);
690 match recovery.fenced_attach(record, debt, event, successor) {
691 Ok(proof) => MintFencedAttachResult::Minted(Box::new(MintedFencedAttach {
692 owner_without_marker_authority: self,
693 proof,
694 marker_source_sequence,
695 marker_source_bytes,
696 })),
697 Err(refusal) => {
698 self.frontiers
699 .reinstall_fenced_marker_record((*refusal).into_record());
700 MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
701 owner: self,
702 marker_source_sequence,
703 recovery,
704 debt,
705 event,
706 successor,
707 reason: FencedAttachMintRefusalReason::ProofInputs,
708 }))
709 }
710 }
711 }
712}
713
714#[derive(Debug, PartialEq, Eq)]
719pub struct DrainedPendingTerminal {
720 owner: LiveFrontierOwner,
721 projection: ObserverProgressProjection,
722}
723
724impl DrainedPendingTerminal {
725 #[must_use]
727 pub fn into_parts(self) -> (LiveFrontierOwner, ObserverProgressProjection) {
728 (self.owner, self.projection)
729 }
730}
731
732#[derive(Debug, PartialEq, Eq)]
734pub struct PendingTerminalDrainRefused {
735 owner: LiveFrontierOwner,
736 error: LiveFrontierError,
737}
738
739impl PendingTerminalDrainRefused {
740 #[must_use]
742 pub const fn error(&self) -> LiveFrontierError {
743 self.error
744 }
745
746 #[must_use]
748 pub fn into_owner(self) -> LiveFrontierOwner {
749 self.owner
750 }
751}
752
753fn drain_refusal(
754 owner: LiveFrontierOwner,
755 error: LiveFrontierError,
756) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
757 Err(Box::new(PendingTerminalDrainRefused { owner, error }))
758}
759
760#[derive(Clone, Copy, Debug, PartialEq, Eq)]
762pub struct FencedMarkerSourceExpectation {
763 source: FencedMarkerSourceRecord,
764}
765
766impl FencedMarkerSourceExpectation {
767 #[must_use]
769 pub const fn conversation_id(self) -> u64 {
770 self.source.conversation_id
771 }
772
773 #[must_use]
775 pub const fn marker_delivery_seq(self) -> u64 {
776 self.source.delivery_seq
777 }
778
779 #[must_use]
781 pub const fn admission_order(self) -> super::super::AdmissionOrder {
782 self.source.admission_order
783 }
784
785 #[must_use]
787 pub const fn participant_id(self) -> u64 {
788 self.source.participant_id
789 }
790
791 #[must_use]
793 pub const fn provenance(self) -> super::super::MarkerProvenance {
794 self.source.provenance
795 }
796
797 #[must_use]
799 pub const fn target_binding(self) -> FrontierBinding {
800 self.source.target_binding
801 }
802}
803
804#[derive(Debug, PartialEq, Eq)]
806pub struct RetainedFencedMarkerSource {
807 owner: LiveFrontierOwner,
808 recovery: DetachedCredentialRecovery,
809 expectation: FencedMarkerSourceExpectation,
810}
811
812impl RetainedFencedMarkerSource {
813 #[must_use]
815 pub const fn expectation(&self) -> FencedMarkerSourceExpectation {
816 self.expectation
817 }
818
819 #[must_use]
821 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
822 (self.owner, self.recovery)
823 }
824}
825
826#[derive(Debug, PartialEq, Eq)]
828pub struct FencedMarkerSourceRetentionRefused {
829 owner: LiveFrontierOwner,
830 recovery: DetachedCredentialRecovery,
831}
832
833impl FencedMarkerSourceRetentionRefused {
834 #[must_use]
836 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
837 (self.owner, self.recovery)
838 }
839}
840
841#[derive(Clone, Copy, Debug, PartialEq, Eq)]
843pub enum FencedAttachMintRefusalReason {
844 MarkerAuthority,
847 ProofInputs,
849}
850
851#[derive(Debug, PartialEq, Eq)]
853pub struct MintedFencedAttach {
854 marker_source_sequence: u64,
855 marker_source_bytes: Vec<u8>,
856 owner_without_marker_authority: LiveFrontierOwner,
857 proof: FencedAttachCommit,
858}
859
860impl MintedFencedAttach {
861 #[must_use]
863 pub fn marker_record_source(&self) -> (u64, &[u8]) {
864 (self.marker_source_sequence, &self.marker_source_bytes)
865 }
866
867 #[must_use]
870 pub fn into_parts(self) -> (LiveFrontierOwner, FencedAttachCommit) {
871 (self.owner_without_marker_authority, self.proof)
872 }
873}
874
875#[derive(Debug, PartialEq, Eq)]
877pub struct MintFencedAttachRefused {
878 owner: LiveFrontierOwner,
879 marker_source_sequence: u64,
880 recovery: DetachedCredentialRecovery,
881 debt: ClosureDebt,
882 event: Event,
883 successor: DebtCompletion,
884 reason: FencedAttachMintRefusalReason,
885}
886
887impl MintFencedAttachRefused {
888 #[must_use]
890 pub const fn reason(&self) -> FencedAttachMintRefusalReason {
891 self.reason
892 }
893
894 #[must_use]
896 pub fn into_parts(
897 self,
898 ) -> (
899 LiveFrontierOwner,
900 u64,
901 DetachedCredentialRecovery,
902 ClosureDebt,
903 Event,
904 DebtCompletion,
905 ) {
906 (
907 self.owner,
908 self.marker_source_sequence,
909 self.recovery,
910 self.debt,
911 self.event,
912 self.successor,
913 )
914 }
915}
916
917#[derive(Debug, PartialEq, Eq)]
919pub enum MintFencedAttachResult {
920 Minted(Box<MintedFencedAttach>),
922 MintRefused(Box<MintFencedAttachRefused>),
924}
925
926#[derive(Debug, PartialEq, Eq)]
928pub struct LiveLeaveCommit<EF, V, LF> {
929 identity: IdentityState<EF, V, LF>,
930 owner: LiveFrontierOwner,
931}
932
933impl<EF, V, LF> LiveLeaveCommit<EF, V, LF> {
934 #[must_use]
936 pub const fn observer_progress_projection(&self) -> Option<ObserverProgressProjection> {
937 let IdentityState::Retired(retired) = &self.identity else {
938 return None;
939 };
940 let committed = retired.committed_result();
941 Some(ObserverProgressProjection::sourced(
942 committed.conversation_id(),
943 committed.left_delivery_seq(),
944 super::super::checkpoint_codec::ObserverProjectionSource::Leave(
945 super::super::LeaveCommittedRestore {
946 conversation_id: committed.conversation_id(),
947 leave_attempt_token: committed.leave_attempt_token(),
948 participant_id: committed.participant_id(),
949 retired_generation: committed.retired_generation(),
950 ended_binding_epoch: committed.ended_binding_epoch(),
951 prior_terminal_delivery_seq: committed.prior_terminal_delivery_seq(),
952 left_delivery_seq: committed.left_delivery_seq(),
953 },
954 ),
955 ))
956 }
957
958 #[must_use]
960 pub fn into_parts(self) -> (IdentityState<EF, V, LF>, LiveFrontierOwner) {
961 (self.identity, self.owner)
962 }
963}
964
965#[derive(Clone, Debug, PartialEq, Eq)]
967pub enum LiveLeaveError {
968 Prepare(PrepareLeaveAuthorityError),
970 Commit(LeaveCommitError),
972 RetainedCharge,
974 RetainedRecordLimit,
976 ClosureAccounting,
978 Identity,
980}
981
982pub fn commit_settled_leave_frontier<EF, V, LF, D>(
990 owner: LiveFrontierOwner,
991 member: LiveMember<EF>,
992 binding: BindingState,
993 detach_cell: DetachCell<D>,
994 verified: VerifiedLeaveRequest<V, LF>,
995 left_delivery_seq: u64,
996 left_charge: RetainedRecordCharge,
997) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
998 let LiveFrontierOwner {
999 frontiers,
1000 closure_accounting,
1001 mut retained_charges,
1002 retained_record_limit,
1003 } = owner;
1004 let retired_marker_charge =
1005 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
1006 let authority = frontiers
1007 .prepare_settled_leave_authority(&member, binding)
1008 .map_err(LiveLeaveError::Prepare)?;
1009 let commit = commit_leave(
1010 member,
1011 binding,
1012 detach_cell,
1013 verified,
1014 authority,
1015 LeaveCommitParameters { left_delivery_seq },
1016 )
1017 .map_err(LiveLeaveError::Commit)?;
1018 let (identity, frontiers) = commit.into_parts();
1019 let IdentityState::Retired(retired) = &identity else {
1020 return Err(LiveLeaveError::Identity);
1021 };
1022 if left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
1023 || left_charge.admission_order() != retired.left_admission_order()
1024 || left_charge.encoded_charge().entries != 1
1025 {
1026 return Err(LiveLeaveError::RetainedCharge);
1027 }
1028 retained_charges.push(left_charge);
1029 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
1030 let retained_len = u64::try_from(frontiers.retained_records().len())
1031 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
1032 if retained_len > retained_record_limit
1033 || retained_charges.len() != frontiers.retained_records().len()
1034 {
1035 return Err(LiveLeaveError::RetainedRecordLimit);
1036 }
1037 let closure_accounting =
1038 accounting_after_leave(closure_accounting, &[left_charge], retired_marker_charge)
1039 .ok_or(LiveLeaveError::ClosureAccounting)?;
1040 Ok(LiveLeaveCommit {
1041 identity,
1042 owner: LiveFrontierOwner {
1043 frontiers,
1044 closure_accounting,
1045 retained_charges,
1046 retained_record_limit,
1047 },
1048 })
1049}
1050
1051pub fn commit_pending_leave_frontier<EF, V, LF, D>(
1060 owner: LiveFrontierOwner,
1061 member: LiveMember<EF>,
1062 pending: PendingFinalization,
1063 detach_cell: DetachCell<D>,
1064 verified: VerifiedLeaveRequest<V, LF>,
1065 parameters: PendingLeaveCommitParameters,
1066 charges: [RetainedRecordCharge; 2],
1067) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
1068 let [terminal_charge, left_charge] = charges;
1069 let terminal_delivery_seq = parameters.terminal_delivery_seq;
1070 let LiveFrontierOwner {
1071 frontiers,
1072 closure_accounting,
1073 mut retained_charges,
1074 retained_record_limit,
1075 } = owner;
1076 let retired_marker_charge =
1077 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
1078 let authority = frontiers
1079 .prepare_pending_leave_authority(&member, pending)
1080 .map_err(LiveLeaveError::Prepare)?;
1081 let commit = commit_pending_leave(
1082 member,
1083 pending,
1084 detach_cell,
1085 verified,
1086 authority,
1087 parameters,
1088 )
1089 .map_err(LiveLeaveError::Commit)?;
1090 let (identity, frontiers) = commit.into_parts();
1091 let IdentityState::Retired(retired) = &identity else {
1092 return Err(LiveLeaveError::Identity);
1093 };
1094 if retired.committed_result().prior_terminal_delivery_seq() != Some(terminal_delivery_seq)
1095 || terminal_charge.delivery_seq() != terminal_delivery_seq
1096 || terminal_charge.admission_order() != pending.admission_order()
1097 || terminal_charge.encoded_charge().entries != 1
1098 || left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
1099 || left_charge.admission_order() != retired.left_admission_order()
1100 || left_charge.encoded_charge().entries != 1
1101 {
1102 return Err(LiveLeaveError::RetainedCharge);
1103 }
1104 retained_charges.extend([terminal_charge, left_charge]);
1105 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
1106 let retained_len = u64::try_from(frontiers.retained_records().len())
1107 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
1108 if retained_len > retained_record_limit {
1109 return Err(LiveLeaveError::RetainedRecordLimit);
1110 }
1111 if retained_charges.len() != frontiers.retained_records().len() {
1112 return Err(LiveLeaveError::RetainedCharge);
1113 }
1114 let closure_accounting = accounting_after_leave(
1115 closure_accounting,
1116 &[terminal_charge, left_charge],
1117 retired_marker_charge,
1118 )
1119 .ok_or(LiveLeaveError::ClosureAccounting)?;
1120 Ok(LiveLeaveCommit {
1121 identity,
1122 owner: LiveFrontierOwner {
1123 frontiers,
1124 closure_accounting,
1125 retained_charges,
1126 retained_record_limit,
1127 },
1128 })
1129}
1130
1131fn retired_marker_charge(
1132 frontiers: &ClaimFrontiers,
1133 retained_charges: &[RetainedRecordCharge],
1134 participant_id: crate::wire::ParticipantId,
1135) -> Result<Option<RetainedRecordCharge>, LiveLeaveError> {
1136 let marker_sequence = frontiers
1137 .retained_marker_records()
1138 .iter()
1139 .find_map(|record| {
1140 matches!(
1141 record.kind,
1142 RetainedCausalRecordKind::CompactionMarker {
1143 participant_index,
1144 ..
1145 } if participant_index == participant_id
1146 )
1147 .then_some(record.delivery_seq)
1148 });
1149 let Some(marker_sequence) = marker_sequence else {
1150 return Ok(None);
1151 };
1152 retained_charges
1153 .iter()
1154 .copied()
1155 .find(|charge| charge.delivery_seq() == marker_sequence)
1156 .map(Some)
1157 .ok_or(LiveLeaveError::RetainedCharge)
1158}
1159
1160#[derive(Debug, PartialEq, Eq)]
1162pub struct AttachFrontierCharges {
1163 terminal: Option<RetainedRecordCharge>,
1164 attached: RetainedRecordCharge,
1165 seal: LiveTransitionInputSeal,
1166}
1167
1168#[derive(Debug, PartialEq, Eq)]
1169enum LiveTransitionInputSeal {
1170 Validated,
1171}
1172
1173impl AttachFrontierCharges {
1174 #[must_use]
1176 pub const fn new(
1177 terminal: Option<RetainedRecordCharge>,
1178 attached: RetainedRecordCharge,
1179 ) -> Self {
1180 Self {
1181 terminal,
1182 attached,
1183 seal: LiveTransitionInputSeal::Validated,
1184 }
1185 }
1186
1187 const fn into_parts(self) -> (Option<RetainedRecordCharge>, RetainedRecordCharge) {
1188 let Self {
1189 terminal,
1190 attached,
1191 seal,
1192 } = self;
1193 match seal {
1194 LiveTransitionInputSeal::Validated => (terminal, attached),
1195 }
1196 }
1197}
1198
1199#[derive(Debug, PartialEq, Eq)]
1201pub struct LiveFrontierCommit<T> {
1202 operation: T,
1203 owner: LiveFrontierOwner,
1204}
1205
1206impl<T> LiveFrontierCommit<T> {
1207 #[must_use]
1209 pub const fn operation(&self) -> &T {
1210 &self.operation
1211 }
1212
1213 #[must_use]
1215 pub const fn owner(&self) -> &LiveFrontierOwner {
1216 &self.owner
1217 }
1218
1219 #[must_use]
1221 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1222 (self.operation, self.owner)
1223 }
1224}
1225
1226#[derive(Debug, PartialEq, Eq)]
1228pub struct LiveFrontierFailure<T> {
1229 error: LiveFrontierError,
1230 operation: T,
1231 owner: LiveFrontierOwner,
1232}
1233
1234impl<T> LiveFrontierFailure<T> {
1235 #[must_use]
1237 pub const fn error(&self) -> LiveFrontierError {
1238 self.error
1239 }
1240
1241 #[must_use]
1243 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1244 (self.operation, self.owner)
1245 }
1246}
1247
1248#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1250pub enum LiveFrontierError {
1251 Authority,
1253 Precedence(PrecedenceCondition),
1261 RetainedCharge,
1263 RetainedRecordLimit,
1265 Frontier,
1267 ClosureAccounting,
1269}
1270
1271pub type LiveFrontierResult<T> = Result<LiveFrontierCommit<T>, Box<LiveFrontierFailure<T>>>;
1273
1274pub fn apply_enrollment_frontier<F>(
1280 owner: LiveFrontierOwner,
1281 operation: super::super::EnrollmentCommit<F>,
1282 attached_charge: RetainedRecordCharge,
1283) -> LiveFrontierResult<super::super::EnrollmentCommit<F>> {
1284 let attached = operation.attached;
1285 if attached.conversation_id() != owner.frontiers.conversation_id() {
1286 return failure(owner, operation, LiveFrontierError::Authority);
1287 }
1288 let participant_id = attached.participant_id();
1289 let mut active = owner.frontiers.active_identities().participants().to_vec();
1290 if active
1291 .iter()
1292 .any(|participant| participant.participant_index() == participant_id)
1293 {
1294 return failure(owner, operation, LiveFrontierError::Authority);
1295 }
1296 active.push(FrontierParticipant::new(
1297 participant_id,
1298 operation.member.cursor(),
1299 FrontierBinding::Bound(attached.binding_epoch()),
1300 ));
1301 active.sort_unstable_by_key(|participant| participant.participant_index());
1302 let rows = [retained_attached(attached)];
1303 let Some(sequence) =
1304 enrollment_sequence(owner.frontiers.sequence().ledger(), attached.delivery_seq())
1305 else {
1306 return failure(owner, operation, LiveFrontierError::Frontier);
1307 };
1308 let Some(order) = enrollment_order(
1309 owner.frontiers.order().ledger(),
1310 attached.admission_order().transaction_order(),
1311 ) else {
1312 return failure(owner, operation, LiveFrontierError::Frontier);
1313 };
1314 transition(
1315 owner,
1316 operation,
1317 active,
1318 &rows,
1319 vec![attached_charge],
1320 sequence,
1321 order,
1322 )
1323}
1324
1325pub fn apply_attach_frontier<F, V>(
1332 owner: LiveFrontierOwner,
1333 operation: AttachCommit<F, V>,
1334 charges: AttachFrontierCharges,
1335) -> LiveFrontierResult<AttachCommit<F, V>> {
1336 let (terminal_charge, attached_charge) = charges.into_parts();
1337 let attached = operation.attached;
1338 if attached.conversation_id() != owner.frontiers.conversation_id() {
1339 return failure(owner, operation, LiveFrontierError::Authority);
1340 }
1341 let mut active = owner.frontiers.active_identities().participants().to_vec();
1342 let Some(participant) = active
1343 .iter_mut()
1344 .find(|participant| participant.participant_index() == attached.participant_id())
1345 else {
1346 return failure(owner, operation, LiveFrontierError::Authority);
1347 };
1348 let frontier_cursor = participant.cursor();
1349 *participant = FrontierParticipant::new(
1350 participant.participant_index(),
1351 operation.member.cursor(),
1352 FrontierBinding::Bound(attached.binding_epoch()),
1353 );
1354 let reopened = reopened_marker_anchors(
1355 &owner,
1356 attached.participant_id(),
1357 frontier_cursor,
1358 operation.member.cursor(),
1359 );
1360 with_reopened_marker_anchors(owner, operation, reopened, |owner, operation| {
1361 apply_attach_rows(owner, operation, active, terminal_charge, attached_charge)
1362 })
1363}
1364
1365fn apply_attach_rows<F, V>(
1367 owner: LiveFrontierOwner,
1368 operation: AttachCommit<F, V>,
1369 active: Vec<FrontierParticipant>,
1370 terminal_charge: Option<RetainedRecordCharge>,
1371 attached_charge: RetainedRecordCharge,
1372) -> LiveFrontierResult<AttachCommit<F, V>> {
1373 let attached = operation.attached;
1374 let current_sequence = owner.frontiers.sequence().ledger();
1375 let current_order = owner.frontiers.order().ledger();
1376 let (rows, keyed_charges, sequence, order) = match operation.transition {
1377 AttachTransition::Detached => {
1378 if terminal_charge.is_some() {
1379 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1380 }
1381 let Some(sequence) =
1382 detached_attach_sequence(current_sequence, attached.delivery_seq())
1383 else {
1384 return failure(owner, operation, LiveFrontierError::Frontier);
1385 };
1386 let Some(order) = detached_attach_order(
1387 current_order,
1388 attached.admission_order().transaction_order(),
1389 ) else {
1390 return failure(owner, operation, LiveFrontierError::Frontier);
1391 };
1392 (
1393 vec![retained_attached(attached)],
1394 vec![attached_charge],
1395 sequence,
1396 order,
1397 )
1398 }
1399 AttachTransition::Superseded { terminal } => {
1400 let Some(terminal_charge) = terminal_charge else {
1401 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1402 };
1403 let rows = vec![
1404 retained_terminal(terminal.into()),
1405 retained_attached(attached),
1406 ];
1407 let Some(sequence) = superseding_attach_sequence(current_sequence, &rows) else {
1408 return failure(owner, operation, LiveFrontierError::Frontier);
1409 };
1410 let Some(order) = superseding_attach_order(
1411 current_order,
1412 attached.admission_order().transaction_order(),
1413 ) else {
1414 return failure(owner, operation, LiveFrontierError::Frontier);
1415 };
1416 (
1417 rows,
1418 vec![terminal_charge, attached_charge],
1419 sequence,
1420 order,
1421 )
1422 }
1423 AttachTransition::FencedRecovery {
1424 prior_binding_epoch,
1425 composed_terminal,
1426 next_closure_state,
1427 } => {
1428 return apply_fenced_attach_frontier(
1429 owner,
1430 operation,
1431 terminal_charge,
1432 attached_charge,
1433 prior_binding_epoch,
1434 composed_terminal,
1435 next_closure_state,
1436 );
1437 }
1438 };
1439 transition(
1440 owner,
1441 operation,
1442 active,
1443 &rows,
1444 keyed_charges,
1445 sequence,
1446 order,
1447 )
1448}
1449
1450pub fn apply_detach_frontier<EF, V>(
1456 owner: LiveFrontierOwner,
1457 operation: CommittedDetachTransition<EF, V>,
1458 terminal_charge: RetainedRecordCharge,
1459) -> LiveFrontierResult<CommittedDetachTransition<EF, V>> {
1460 let terminal = operation.terminal();
1461 if terminal.conversation_id() != owner.frontiers.conversation_id() {
1462 return failure(owner, operation, LiveFrontierError::Authority);
1463 }
1464 let mut active = owner.frontiers.active_identities().participants().to_vec();
1465 let Some(participant) = active
1466 .iter_mut()
1467 .find(|participant| participant.participant_index() == terminal.participant_id())
1468 else {
1469 return failure(owner, operation, LiveFrontierError::Authority);
1470 };
1471 *participant = FrontierParticipant::new(
1472 participant.participant_index(),
1473 operation.member().cursor(),
1474 FrontierBinding::Detached(terminal.binding_epoch()),
1475 );
1476 let row = retained_terminal(terminal.into());
1477 let Some(sequence) = detach_sequence(owner.frontiers.sequence().ledger(), row.delivery_seq)
1478 else {
1479 return failure(owner, operation, LiveFrontierError::Frontier);
1480 };
1481 let Some(order) = detach_order(
1482 owner.frontiers.order().ledger(),
1483 row.admission_order.transaction_order(),
1484 ) else {
1485 return failure(owner, operation, LiveFrontierError::Frontier);
1486 };
1487 transition(
1488 owner,
1489 operation,
1490 active,
1491 &[row],
1492 vec![terminal_charge],
1493 sequence,
1494 order,
1495 )
1496}
1497
1498fn crossed_marker_anchors(
1508 owner: &LiveFrontierOwner,
1509 participant_id: crate::wire::ParticipantId,
1510 previous_cursor: crate::wire::DeliverySeq,
1511 through_seq: crate::wire::DeliverySeq,
1512) -> u64 {
1513 owner
1514 .frontiers
1515 .retained_marker_records()
1516 .iter()
1517 .filter(|record| {
1518 matches!(
1519 record.kind,
1520 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1521 if participant_index == participant_id
1522 ) && previous_cursor < record.delivery_seq
1523 && record.delivery_seq <= through_seq
1524 })
1525 .fold(0_u64, |count, _| count.saturating_add(1))
1526}
1527
1528pub fn apply_participant_ack_frontier(
1542 mut owner: LiveFrontierOwner,
1543 operation: ParticipantAckCommit,
1544) -> LiveFrontierResult<ParticipantAckCommit> {
1545 let request = operation.outcome().request();
1546 let Some(current) = owner
1547 .frontiers
1548 .active_identities()
1549 .participants()
1550 .iter()
1551 .find(|participant| participant.participant_index() == request.participant_id)
1552 .copied()
1553 else {
1554 return failure(owner, operation, LiveFrontierError::Authority);
1555 };
1556 let crossed = crossed_marker_anchors(
1557 &owner,
1558 request.participant_id,
1559 current.cursor(),
1560 request.through_seq,
1561 );
1562 let accounting = if crossed == 0 {
1563 None
1564 } else {
1565 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1566 Some(accounting) => Some(accounting),
1567 None => {
1568 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1569 }
1570 }
1571 };
1572 let participant = FrontierParticipant::new(
1573 request.participant_id,
1574 request.through_seq,
1575 current.binding(),
1576 );
1577 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1578 Ok(frontiers) => frontiers,
1579 Err(frontier_failure) => {
1580 let (frontiers, error) = *frontier_failure;
1581 owner.frontiers = frontiers;
1582 return failure(owner, operation, map_frontier_error(error));
1583 }
1584 };
1585 if let Some(accounting) = accounting {
1586 owner.closure_accounting = accounting;
1587 }
1588 Ok(LiveFrontierCommit { operation, owner })
1589}
1590
1591pub fn apply_nonzero_participant_ack_frontier(
1601 mut owner: LiveFrontierOwner,
1602 operation: NonzeroParticipantAckCommit,
1603) -> LiveFrontierResult<NonzeroParticipantAckCommit> {
1604 let request = operation.outcome().request();
1605 let Some(current) = owner
1606 .frontiers
1607 .active_identities()
1608 .participants()
1609 .iter()
1610 .find(|participant| participant.participant_index() == request.participant_id)
1611 .copied()
1612 else {
1613 return failure(owner, operation, LiveFrontierError::Authority);
1614 };
1615 let crossed = crossed_marker_anchors(
1618 &owner,
1619 request.participant_id,
1620 current.cursor(),
1621 request.through_seq,
1622 );
1623 let accounting = if crossed == 0 {
1624 None
1625 } else {
1626 match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
1627 Some(accounting) => Some(accounting),
1628 None => {
1629 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1630 }
1631 }
1632 };
1633 let participant = FrontierParticipant::new(
1634 request.participant_id,
1635 request.through_seq,
1636 current.binding(),
1637 );
1638 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1639 Ok(frontiers) => frontiers,
1640 Err(frontier_failure) => {
1641 let (frontiers, error) = *frontier_failure;
1642 owner.frontiers = frontiers;
1643 return failure(owner, operation, map_frontier_error(error));
1644 }
1645 };
1646 if let Some(accounting) = accounting {
1647 owner.closure_accounting = accounting;
1648 }
1649 Ok(LiveFrontierCommit { operation, owner })
1650}
1651
1652pub fn apply_marker_ack_frontier(
1658 mut owner: LiveFrontierOwner,
1659 operation: MarkerAckCommit,
1660) -> LiveFrontierResult<MarkerAckCommit> {
1661 let request = operation.outcome().request();
1662 if !owner
1663 .frontiers
1664 .retained_marker_records()
1665 .iter()
1666 .any(|record| {
1667 record.delivery_seq == request.marker_delivery_seq
1668 && matches!(
1669 record.kind,
1670 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1671 if participant_index == request.participant_id
1672 )
1673 })
1674 {
1675 return failure(owner, operation, LiveFrontierError::Authority);
1676 }
1677 let Some(current) = owner
1678 .frontiers
1679 .active_identities()
1680 .participants()
1681 .iter()
1682 .find(|participant| participant.participant_index() == request.participant_id)
1683 .copied()
1684 else {
1685 return failure(owner, operation, LiveFrontierError::Authority);
1686 };
1687 let Some(accounting) = accounting_after_marker_ack(owner.closure_accounting) else {
1688 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1689 };
1690 let participant = FrontierParticipant::new(
1691 request.participant_id,
1692 request.marker_delivery_seq,
1693 current.binding(),
1694 );
1695 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1696 Ok(frontiers) => frontiers,
1697 Err(frontier_failure) => {
1698 let (frontiers, error) = *frontier_failure;
1699 owner.frontiers = frontiers;
1700 return failure(owner, operation, map_frontier_error(error));
1701 }
1702 };
1703 owner.closure_accounting = accounting;
1704 Ok(LiveFrontierCommit { operation, owner })
1705}
1706
1707fn apply_fenced_attach_frontier<F, V>(
1708 owner: LiveFrontierOwner,
1709 operation: AttachCommit<F, V>,
1710 terminal_charge: Option<RetainedRecordCharge>,
1711 attached_charge: RetainedRecordCharge,
1712 prior_binding_epoch: crate::wire::BindingEpoch,
1713 composed_terminal: Option<super::super::CommittedBindingTerminal>,
1714 next_closure_state: super::super::ClosureState,
1715) -> LiveFrontierResult<AttachCommit<F, V>> {
1716 let attached = operation.attached;
1717 let (rows, charges) = match (composed_terminal, terminal_charge) {
1718 (None, None) => (vec![retained_attached(attached)], vec![attached_charge]),
1719 (Some(terminal), Some(terminal_charge)) => (
1720 vec![retained_terminal(terminal), retained_attached(attached)],
1721 vec![terminal_charge, attached_charge],
1722 ),
1723 (None, Some(_)) | (Some(_), None) => {
1724 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1725 }
1726 };
1727 let participant = FrontierParticipant::new(
1728 attached.participant_id(),
1729 operation.member.cursor(),
1730 FrontierBinding::Bound(attached.binding_epoch()),
1731 );
1732 fenced_attach_transition(
1733 owner,
1734 operation,
1735 participant,
1736 prior_binding_epoch,
1737 next_closure_state,
1738 &rows,
1739 charges,
1740 )
1741}
1742
1743fn reopened_marker_anchors(
1761 owner: &LiveFrontierOwner,
1762 participant_id: crate::wire::ParticipantId,
1763 frontier_cursor: crate::wire::DeliverySeq,
1764 member_cursor: crate::wire::DeliverySeq,
1765) -> u64 {
1766 if member_cursor >= frontier_cursor {
1767 return 0;
1768 }
1769 crossed_marker_anchors(owner, participant_id, member_cursor, frontier_cursor)
1770}
1771
1772fn with_reopened_marker_anchors<T>(
1775 mut owner: LiveFrontierOwner,
1776 operation: T,
1777 reopened: u64,
1778 apply: impl FnOnce(LiveFrontierOwner, T) -> LiveFrontierResult<T>,
1779) -> LiveFrontierResult<T> {
1780 if reopened == 0 {
1781 return apply(owner, operation);
1782 }
1783 let unchanged = owner.closure_accounting;
1784 let Some(accounting) = accounting_after_marker_remint(unchanged, reopened) else {
1785 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1786 };
1787 owner.closure_accounting = accounting;
1788 apply(owner, operation).map_err(|mut refused| {
1789 refused.owner.closure_accounting = unchanged;
1790 refused
1791 })
1792}
1793
1794fn fenced_attach_transition<T>(
1795 mut owner: LiveFrontierOwner,
1796 operation: T,
1797 participant: FrontierParticipant,
1798 prior_binding_epoch: crate::wire::BindingEpoch,
1799 next_closure_state: super::super::ClosureState,
1800 rows: &[RetainedCausalRecord],
1801 charges: Vec<RetainedRecordCharge>,
1802) -> LiveFrontierResult<T> {
1803 if rows.len() != charges.len()
1804 || rows.iter().zip(&charges).any(|(row, charge)| {
1805 row.delivery_seq != charge.delivery_seq()
1806 || row.admission_order != charge.admission_order()
1807 || charge.encoded_charge().entries != 1
1808 })
1809 {
1810 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1811 }
1812 let resulting_len = owner
1813 .frontiers
1814 .retained_records()
1815 .len()
1816 .checked_add(rows.len());
1817 if resulting_len
1818 .and_then(|len| u64::try_from(len).ok())
1819 .is_none_or(|len| len > owner.retained_record_limit)
1820 {
1821 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1822 }
1823 let Some(accounting) =
1824 accounting_after_fenced_attach(owner.closure_accounting, &charges, next_closure_state)
1825 else {
1826 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1827 };
1828 owner.frontiers =
1829 match owner
1830 .frontiers
1831 .apply_live_fenced_attach(participant, prior_binding_epoch, rows)
1832 {
1833 Ok(frontiers) => frontiers,
1834 Err(frontier_failure) => {
1835 let (frontiers, error) = *frontier_failure;
1836 owner.frontiers = frontiers;
1837 return failure(owner, operation, map_frontier_error(error));
1838 }
1839 };
1840 owner.retained_charges.extend(charges);
1841 owner
1842 .retained_charges
1843 .sort_unstable_by_key(|charge| charge.delivery_seq());
1844 owner.closure_accounting = accounting;
1845 Ok(LiveFrontierCommit { operation, owner })
1846}
1847
1848fn transition<T>(
1849 mut owner: LiveFrontierOwner,
1850 operation: T,
1851 active: Vec<FrontierParticipant>,
1852 rows: &[RetainedCausalRecord],
1853 charges: Vec<RetainedRecordCharge>,
1854 sequence: SequenceLedger,
1855 order: OrderLedger,
1856) -> LiveFrontierResult<T> {
1857 if rows.len() != charges.len()
1858 || rows.iter().zip(&charges).any(|(row, charge)| {
1859 row.delivery_seq != charge.delivery_seq()
1860 || row.admission_order != charge.admission_order()
1861 || charge.encoded_charge().entries != 1
1862 })
1863 {
1864 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1865 }
1866 let resulting_len = owner
1867 .frontiers
1868 .retained_records()
1869 .len()
1870 .checked_add(rows.len());
1871 if resulting_len
1872 .and_then(|len| u64::try_from(len).ok())
1873 .is_none_or(|len| len > owner.retained_record_limit)
1874 {
1875 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1876 }
1877 let Some(accounting) = accounting_after_rows(owner.closure_accounting, &charges) else {
1878 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1879 };
1880 owner.frontiers = match owner
1881 .frontiers
1882 .apply_live_transition(active, rows, sequence, order)
1883 {
1884 Ok(frontiers) => frontiers,
1885 Err(frontier_failure) => {
1886 let (frontiers, error) = *frontier_failure;
1887 owner.frontiers = frontiers;
1888 return failure(owner, operation, map_frontier_error(error));
1889 }
1890 };
1891 owner.retained_charges.extend(charges);
1892 owner.closure_accounting = accounting;
1893 Ok(LiveFrontierCommit { operation, owner })
1894}
1895
1896const fn map_frontier_error(error: LiveFrontierTransitionError) -> LiveFrontierError {
1897 match error {
1898 LiveFrontierTransitionError::Authority => LiveFrontierError::Authority,
1899 LiveFrontierTransitionError::Precedence(condition) => {
1900 LiveFrontierError::Precedence(condition)
1901 }
1902 LiveFrontierTransitionError::RecordPosition
1903 | LiveFrontierTransitionError::Exhausted
1904 | LiveFrontierTransitionError::ResultingFrontier => LiveFrontierError::Frontier,
1905 }
1906}
1907
1908fn failure<T, U>(
1909 owner: LiveFrontierOwner,
1910 operation: T,
1911 error: LiveFrontierError,
1912) -> Result<U, Box<LiveFrontierFailure<T>>> {
1913 Err(Box::new(LiveFrontierFailure {
1914 error,
1915 operation,
1916 owner,
1917 }))
1918}