1use alloc::{boxed::Box, vec, vec::Vec};
8
9use crate::{algebra::ResourceVector, wire::RecordAdmission};
10
11use super::super::{
12 AttachCommit, AttachTransition, BindingState, ClaimFrontiers, ClosureAccounting, ClosureDebt,
13 CommittedDetachTransition, DebtCompletion, DetachCell, DetachedCredentialRecovery, Event,
14 FencedAttachCommit, FrontierBinding, FrontierParticipant, IdentityState,
15 InitialEnrollmentFrontierCommit, LeaveCommitError, LeaveCommitParameters, LiveMember,
16 MarkerAckCommit, NonzeroParticipantAckCommit, ObserverProgressProjection, OrderLedger,
17 ParticipantAckCommit, PendingFinalization, PendingLeaveCommitParameters,
18 PrepareLeaveAuthorityError, RetainedCausalRecord, RetainedCausalRecordKind, SequenceLedger,
19 StoredEdge, VerifiedLeaveRequest,
20 claim_frontier::{BindingTerminalOwner, FencedMarkerSourceRecord, LiveFrontierTransitionError},
21 commit_leave, commit_pending_leave,
22};
23use super::{
24 InitialEnrollmentOperationCommit, MarkerDeliveryProjection, MarkerDrainCommit,
25 RecordAdmissionPersistenceParts, RetainedRecordCharge, UnchangedRecordAdmission,
26};
27
28mod binding_fate_transition;
29mod ledger;
30mod state;
31pub(super) use binding_fate_transition::BindingFateOwnerPlan;
32use ledger::{
33 detach_order, detach_sequence, detached_attach_order, detached_attach_sequence,
34 enrollment_order, enrollment_sequence, superseding_attach_order, superseding_attach_sequence,
35};
36use state::{
37 accounting_after_fenced_attach, accounting_after_leave, accounting_after_marker_ack,
38 accounting_after_rows, retained_attached, retained_terminal,
39};
40
41#[derive(Debug, PartialEq, Eq)]
66pub struct LiveFrontierOwner {
67 frontiers: ClaimFrontiers,
68 closure_accounting: ClosureAccounting,
69 retained_charges: Vec<RetainedRecordCharge>,
70 retained_record_limit: u64,
71}
72
73impl LiveFrontierOwner {
74 #[must_use]
76 pub fn from_initial_enrollment<F>(
77 initial: InitialEnrollmentFrontierCommit<F>,
78 retained_record_limit: u64,
79 ) -> (InitialEnrollmentOperationCommit<F>, Self) {
80 let (operation, frontiers, closure_accounting, attached_charge) =
81 initial.into_conversation_parts();
82 let attached = operation.enrollment().attached;
83 let retained_charges = vec![RetainedRecordCharge::new(
84 attached.delivery_seq(),
85 attached.admission_order(),
86 attached_charge,
87 )];
88 (
89 operation,
90 Self {
91 frontiers,
92 closure_accounting,
93 retained_charges,
94 retained_record_limit,
95 },
96 )
97 }
98
99 #[cfg(any(test, feature = "test-support"))]
100 pub(in crate::lifecycle) const fn from_test_parts(
101 frontiers: ClaimFrontiers,
102 closure_accounting: ClosureAccounting,
103 retained_charges: Vec<RetainedRecordCharge>,
104 retained_record_limit: u64,
105 ) -> Self {
106 Self {
107 frontiers,
108 closure_accounting,
109 retained_charges,
110 retained_record_limit,
111 }
112 }
113
114 #[cfg(any(test, feature = "test-support"))]
122 pub fn with_pending_finalizer_test_capacity(
123 mut self,
124 finalizer_rows: u64,
125 finalizer_charge: ResourceVector,
126 ) -> Result<Self, &'static str> {
127 let retained_count = u64::try_from(self.retained_charges.len())
128 .map_err(|_| "retained record count exceeds u64")?;
129 self.retained_record_limit = retained_count
130 .checked_add(finalizer_rows)
131 .ok_or("pending finalizer retained-record capacity overflow")?;
132 let current = self.closure_accounting;
133 let configured = current.configured_cap();
134 let configured = ResourceVector::new(
135 configured
136 .entries
137 .checked_add(finalizer_charge.entries)
138 .ok_or("pending finalizer entry capacity overflow")?,
139 configured
140 .bytes
141 .checked_add(finalizer_charge.bytes)
142 .ok_or("pending finalizer byte capacity overflow")?,
143 );
144 self.closure_accounting = ClosureAccounting::try_new(
145 current.state(),
146 current.marker_capacity_credits(),
147 current.marker_anchors(),
148 current.edge_sequence_claims(),
149 current.edge_order_position_claims(),
150 current.edge_k_remaining(),
151 current.baseline(),
152 configured,
153 current.episode_churn_used(),
154 current.episode_churn_limit(),
155 )
156 .map_err(|_| "pending finalizer closure capacity extension refused")?;
157 Ok(self)
158 }
159
160 #[must_use]
162 pub const fn frontiers(&self) -> &ClaimFrontiers {
163 &self.frontiers
164 }
165
166 #[must_use]
168 pub const fn closure_accounting(&self) -> ClosureAccounting {
169 self.closure_accounting
170 }
171
172 #[must_use]
174 pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
175 &self.retained_charges
176 }
177
178 #[must_use]
180 pub const fn retained_record_limit(&self) -> u64 {
181 self.retained_record_limit
182 }
183
184 #[must_use]
186 pub fn into_parts(
187 self,
188 ) -> (
189 ClaimFrontiers,
190 ClosureAccounting,
191 Vec<RetainedRecordCharge>,
192 u64,
193 ) {
194 (
195 self.frontiers,
196 self.closure_accounting,
197 self.retained_charges,
198 self.retained_record_limit,
199 )
200 }
201
202 #[must_use]
205 pub fn from_unchanged_record_admission<EF, V, LF>(
206 unchanged: UnchangedRecordAdmission<'_, EF, V, LF>,
207 retained_record_limit: u64,
208 ) -> (Self, RecordAdmission, ResourceVector) {
209 let (prestate, encoded_record_charge) = unchanged.into_parts();
210 let (request, frontiers, closure_accounting, retained_charges) =
211 prestate.into_live_owner_parts();
212 (
213 Self {
214 frontiers,
215 closure_accounting,
216 retained_charges,
217 retained_record_limit,
218 },
219 request,
220 encoded_record_charge,
221 )
222 }
223
224 #[must_use]
227 pub fn from_record_admission_persistence(
228 persistence: RecordAdmissionPersistenceParts,
229 retained_record_limit: u64,
230 ) -> Self {
231 Self {
232 frontiers: persistence.frontiers,
233 closure_accounting: persistence.accounting,
234 retained_charges: persistence.retained_charges,
235 retained_record_limit,
236 }
237 }
238
239 #[must_use]
241 pub fn from_marker_drain(
242 commit: MarkerDrainCommit,
243 retained_record_limit: u64,
244 ) -> (Self, StoredEdge, MarkerDeliveryProjection) {
245 let (frontiers, closure_accounting, retained_charges, successor, projection) =
246 commit.into_parts();
247 (
248 Self {
249 frontiers,
250 closure_accounting,
251 retained_charges,
252 retained_record_limit,
253 },
254 successor,
255 projection,
256 )
257 }
258
259 pub(super) fn commit_binding_terminal_candidate(
260 self,
261 active_binding: super::super::ActiveBinding,
262 admission_order: super::super::AdmissionOrder,
263 delivery_seq: crate::wire::DeliverySeq,
264 charge: RetainedRecordCharge,
265 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
266 let mut active = self.frontiers.active_identities().participants().to_vec();
267 let Some(participant) = active
268 .iter_mut()
269 .find(|participant| participant.participant_index() == active_binding.participant_id)
270 else {
271 return Err(Box::new((self, LiveFrontierError::Authority)));
272 };
273 if participant.binding() != FrontierBinding::Bound(active_binding.binding_epoch) {
274 return Err(Box::new((self, LiveFrontierError::Authority)));
275 }
276 *participant = FrontierParticipant::new(
277 participant.participant_index(),
278 participant.cursor(),
279 FrontierBinding::Detached(active_binding.binding_epoch),
280 );
281 let row = RetainedCausalRecord {
282 delivery_seq,
283 admission_order,
284 kind: RetainedCausalRecordKind::BindingTerminal(super::super::BindingTerminalOwner {
285 participant_index: active_binding.participant_id,
286 binding_epoch: active_binding.binding_epoch,
287 }),
288 };
289 let Some(sequence) = detach_sequence(self.frontiers.sequence().ledger(), delivery_seq)
290 else {
291 return Err(Box::new((self, LiveFrontierError::Frontier)));
292 };
293 let Some(order) = detach_order(
294 self.frontiers.order().ledger(),
295 admission_order.transaction_order(),
296 ) else {
297 return Err(Box::new((self, LiveFrontierError::Frontier)));
298 };
299 match transition(self, (), active, &[row], vec![charge], sequence, order) {
300 Ok(committed) => {
301 let ((), owner) = committed.into_parts();
302 Ok(owner)
303 }
304 Err(failure) => {
305 let error = failure.error();
306 let ((), owner) = failure.into_parts();
307 Err(Box::new((owner, error)))
308 }
309 }
310 }
311
312 pub(super) fn pend_binding_terminal_candidate(
313 self,
314 active_binding: super::super::ActiveBinding,
315 admission_order: super::super::AdmissionOrder,
316 delivery_seq: crate::wire::DeliverySeq,
317 ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
318 let Some(order) = detach_order(
319 self.frontiers.order().ledger(),
320 admission_order.transaction_order(),
321 ) else {
322 return Err(Box::new((self, LiveFrontierError::Frontier)));
323 };
324 let Self {
325 frontiers,
326 closure_accounting,
327 retained_charges,
328 retained_record_limit,
329 } = self;
330 match frontiers.apply_pending_binding_terminal(
331 active_binding.participant_id,
332 active_binding.binding_epoch,
333 delivery_seq,
334 admission_order,
335 order,
336 ) {
337 Ok(frontiers) => Ok(Self {
338 frontiers,
339 closure_accounting,
340 retained_charges,
341 retained_record_limit,
342 }),
343 Err(failure) => {
344 let (frontiers, error) = *failure;
345 Err(Box::new((
346 Self {
347 frontiers,
348 closure_accounting,
349 retained_charges,
350 retained_record_limit,
351 },
352 map_frontier_error(error),
353 )))
354 }
355 }
356 }
357
358 pub fn drain_pending_terminal(
386 self,
387 pending: PendingFinalization,
388 terminal_charge: RetainedRecordCharge,
389 ) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
390 let Some(expected_sequence) = self
391 .frontiers
392 .sequence()
393 .ledger()
394 .high_watermark()
395 .checked_add(1)
396 else {
397 return drain_refusal(self, LiveFrontierError::Frontier);
398 };
399 if terminal_charge.delivery_seq() != expected_sequence
400 || terminal_charge.admission_order() != pending.admission_order()
401 || terminal_charge.encoded_charge().entries != 1
402 {
403 return drain_refusal(self, LiveFrontierError::RetainedCharge);
404 }
405 if self
406 .retained_charges
407 .len()
408 .checked_add(1)
409 .and_then(|len| u64::try_from(len).ok())
410 .is_none()
411 {
412 return drain_refusal(self, LiveFrontierError::RetainedRecordLimit);
413 }
414 let Some(accounting) = accounting_after_rows(self.closure_accounting, &[terminal_charge])
415 else {
416 return drain_refusal(self, LiveFrontierError::ClosureAccounting);
417 };
418 let Self {
419 frontiers,
420 closure_accounting,
421 mut retained_charges,
422 retained_record_limit,
423 } = self;
424 let expected_owner = BindingTerminalOwner {
425 participant_index: pending.participant_id(),
426 binding_epoch: pending.binding_epoch(),
427 };
428 let (frontiers, record) = match frontiers
429 .drain_first_binding_terminal(expected_owner, pending.admission_order())
430 {
431 Ok(drained) => drained,
432 Err(failure) => {
433 let (frontiers, error) = *failure;
434 return drain_refusal(
435 Self {
436 frontiers,
437 closure_accounting,
438 retained_charges,
439 retained_record_limit,
440 },
441 map_frontier_error(error),
442 );
443 }
444 };
445 retained_charges.push(terminal_charge);
446 let projection =
447 ObserverProgressProjection::new(pending.conversation_id(), record.delivery_seq);
448 Ok(DrainedPendingTerminal {
449 owner: Self {
450 frontiers,
451 closure_accounting: accounting,
452 retained_charges,
453 retained_record_limit,
454 },
455 projection,
456 })
457 }
458
459 pub fn retain_fenced_marker_source(
466 self,
467 recovery: DetachedCredentialRecovery,
468 ) -> Result<RetainedFencedMarkerSource, Box<FencedMarkerSourceRetentionRefused>> {
469 let Some(source) = self.frontiers.fenced_marker_source(recovery) else {
470 return Err(Box::new(FencedMarkerSourceRetentionRefused {
471 owner: self,
472 recovery,
473 }));
474 };
475 Ok(RetainedFencedMarkerSource {
476 owner: self,
477 recovery,
478 expectation: FencedMarkerSourceExpectation { source },
479 })
480 }
481
482 #[must_use]
490 pub fn mint_fenced_attach(
491 mut self,
492 marker_source_sequence: u64,
493 recovery: DetachedCredentialRecovery,
494 debt: ClosureDebt,
495 event: Event,
496 successor: DebtCompletion,
497 ) -> MintFencedAttachResult {
498 let Some(record) = self.frontiers.take_fenced_marker_record(recovery) else {
499 return MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
500 owner: self,
501 marker_source_sequence,
502 recovery,
503 debt,
504 event,
505 successor,
506 reason: FencedAttachMintRefusalReason::MarkerAuthority,
507 }));
508 };
509 match recovery.fenced_attach(record, debt, event, successor) {
510 Ok(proof) => MintFencedAttachResult::Minted(Box::new(MintedFencedAttach {
511 owner_without_marker_authority: self,
512 proof,
513 })),
514 Err(refusal) => {
515 self.frontiers
516 .reinstall_fenced_marker_record((*refusal).into_record());
517 MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
518 owner: self,
519 marker_source_sequence,
520 recovery,
521 debt,
522 event,
523 successor,
524 reason: FencedAttachMintRefusalReason::ProofInputs,
525 }))
526 }
527 }
528 }
529}
530
531#[derive(Debug, PartialEq, Eq)]
536pub struct DrainedPendingTerminal {
537 owner: LiveFrontierOwner,
538 projection: ObserverProgressProjection,
539}
540
541impl DrainedPendingTerminal {
542 #[must_use]
544 pub fn into_parts(self) -> (LiveFrontierOwner, ObserverProgressProjection) {
545 (self.owner, self.projection)
546 }
547}
548
549#[derive(Debug, PartialEq, Eq)]
551pub struct PendingTerminalDrainRefused {
552 owner: LiveFrontierOwner,
553 error: LiveFrontierError,
554}
555
556impl PendingTerminalDrainRefused {
557 #[must_use]
559 pub const fn error(&self) -> LiveFrontierError {
560 self.error
561 }
562
563 #[must_use]
565 pub fn into_owner(self) -> LiveFrontierOwner {
566 self.owner
567 }
568}
569
570fn drain_refusal(
571 owner: LiveFrontierOwner,
572 error: LiveFrontierError,
573) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
574 Err(Box::new(PendingTerminalDrainRefused { owner, error }))
575}
576
577#[derive(Clone, Copy, Debug, PartialEq, Eq)]
579pub struct FencedMarkerSourceExpectation {
580 source: FencedMarkerSourceRecord,
581}
582
583impl FencedMarkerSourceExpectation {
584 #[must_use]
586 pub const fn conversation_id(self) -> u64 {
587 self.source.conversation_id
588 }
589
590 #[must_use]
592 pub const fn marker_delivery_seq(self) -> u64 {
593 self.source.delivery_seq
594 }
595
596 #[must_use]
598 pub const fn admission_order(self) -> super::super::AdmissionOrder {
599 self.source.admission_order
600 }
601
602 #[must_use]
604 pub const fn participant_id(self) -> u64 {
605 self.source.participant_id
606 }
607
608 #[must_use]
610 pub const fn provenance(self) -> super::super::MarkerProvenance {
611 self.source.provenance
612 }
613
614 #[must_use]
616 pub const fn target_binding(self) -> FrontierBinding {
617 self.source.target_binding
618 }
619}
620
621#[derive(Debug, PartialEq, Eq)]
623pub struct RetainedFencedMarkerSource {
624 owner: LiveFrontierOwner,
625 recovery: DetachedCredentialRecovery,
626 expectation: FencedMarkerSourceExpectation,
627}
628
629impl RetainedFencedMarkerSource {
630 #[must_use]
632 pub const fn expectation(&self) -> FencedMarkerSourceExpectation {
633 self.expectation
634 }
635
636 #[must_use]
638 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
639 (self.owner, self.recovery)
640 }
641}
642
643#[derive(Debug, PartialEq, Eq)]
645pub struct FencedMarkerSourceRetentionRefused {
646 owner: LiveFrontierOwner,
647 recovery: DetachedCredentialRecovery,
648}
649
650impl FencedMarkerSourceRetentionRefused {
651 #[must_use]
653 pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
654 (self.owner, self.recovery)
655 }
656}
657
658#[derive(Clone, Copy, Debug, PartialEq, Eq)]
660pub enum FencedAttachMintRefusalReason {
661 MarkerAuthority,
664 ProofInputs,
666}
667
668#[derive(Debug, PartialEq, Eq)]
670pub struct MintedFencedAttach {
671 owner_without_marker_authority: LiveFrontierOwner,
672 proof: FencedAttachCommit,
673}
674
675impl MintedFencedAttach {
676 #[must_use]
679 pub fn into_parts(self) -> (LiveFrontierOwner, FencedAttachCommit) {
680 (self.owner_without_marker_authority, self.proof)
681 }
682}
683
684#[derive(Debug, PartialEq, Eq)]
686pub struct MintFencedAttachRefused {
687 owner: LiveFrontierOwner,
688 marker_source_sequence: u64,
689 recovery: DetachedCredentialRecovery,
690 debt: ClosureDebt,
691 event: Event,
692 successor: DebtCompletion,
693 reason: FencedAttachMintRefusalReason,
694}
695
696impl MintFencedAttachRefused {
697 #[must_use]
699 pub const fn reason(&self) -> FencedAttachMintRefusalReason {
700 self.reason
701 }
702
703 #[must_use]
705 pub fn into_parts(
706 self,
707 ) -> (
708 LiveFrontierOwner,
709 u64,
710 DetachedCredentialRecovery,
711 ClosureDebt,
712 Event,
713 DebtCompletion,
714 ) {
715 (
716 self.owner,
717 self.marker_source_sequence,
718 self.recovery,
719 self.debt,
720 self.event,
721 self.successor,
722 )
723 }
724}
725
726#[derive(Debug, PartialEq, Eq)]
728pub enum MintFencedAttachResult {
729 Minted(Box<MintedFencedAttach>),
731 MintRefused(Box<MintFencedAttachRefused>),
733}
734
735#[derive(Debug, PartialEq, Eq)]
737pub struct LiveLeaveCommit<EF, V, LF> {
738 identity: IdentityState<EF, V, LF>,
739 owner: LiveFrontierOwner,
740}
741
742impl<EF, V, LF> LiveLeaveCommit<EF, V, LF> {
743 #[must_use]
745 pub const fn observer_progress_projection(&self) -> Option<ObserverProgressProjection> {
746 let IdentityState::Retired(retired) = &self.identity else {
747 return None;
748 };
749 let committed = retired.committed_result();
750 Some(ObserverProgressProjection::new(
751 committed.conversation_id(),
752 committed.left_delivery_seq(),
753 ))
754 }
755
756 #[must_use]
758 pub fn into_parts(self) -> (IdentityState<EF, V, LF>, LiveFrontierOwner) {
759 (self.identity, self.owner)
760 }
761}
762
763#[derive(Clone, Debug, PartialEq, Eq)]
765pub enum LiveLeaveError {
766 Prepare(PrepareLeaveAuthorityError),
768 Commit(LeaveCommitError),
770 RetainedCharge,
772 RetainedRecordLimit,
774 ClosureAccounting,
776 Identity,
778}
779
780pub fn commit_settled_leave_frontier<EF, V, LF, D>(
788 owner: LiveFrontierOwner,
789 member: LiveMember<EF>,
790 binding: BindingState,
791 detach_cell: DetachCell<D>,
792 verified: VerifiedLeaveRequest<V, LF>,
793 left_delivery_seq: u64,
794 left_charge: RetainedRecordCharge,
795) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
796 let LiveFrontierOwner {
797 frontiers,
798 closure_accounting,
799 mut retained_charges,
800 retained_record_limit,
801 } = owner;
802 let retired_marker_charge =
803 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
804 let authority = frontiers
805 .prepare_settled_leave_authority(&member, binding)
806 .map_err(LiveLeaveError::Prepare)?;
807 let commit = commit_leave(
808 member,
809 binding,
810 detach_cell,
811 verified,
812 authority,
813 LeaveCommitParameters { left_delivery_seq },
814 )
815 .map_err(LiveLeaveError::Commit)?;
816 let (identity, frontiers) = commit.into_parts();
817 let IdentityState::Retired(retired) = &identity else {
818 return Err(LiveLeaveError::Identity);
819 };
820 if left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
821 || left_charge.admission_order() != retired.left_admission_order()
822 || left_charge.encoded_charge().entries != 1
823 {
824 return Err(LiveLeaveError::RetainedCharge);
825 }
826 retained_charges.push(left_charge);
827 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
828 let retained_len = u64::try_from(frontiers.retained_records().len())
829 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
830 if retained_len > retained_record_limit
831 || retained_charges.len() != frontiers.retained_records().len()
832 {
833 return Err(LiveLeaveError::RetainedRecordLimit);
834 }
835 let closure_accounting =
836 accounting_after_leave(closure_accounting, &[left_charge], retired_marker_charge)
837 .ok_or(LiveLeaveError::ClosureAccounting)?;
838 Ok(LiveLeaveCommit {
839 identity,
840 owner: LiveFrontierOwner {
841 frontiers,
842 closure_accounting,
843 retained_charges,
844 retained_record_limit,
845 },
846 })
847}
848
849pub fn commit_pending_leave_frontier<EF, V, LF, D>(
858 owner: LiveFrontierOwner,
859 member: LiveMember<EF>,
860 pending: PendingFinalization,
861 detach_cell: DetachCell<D>,
862 verified: VerifiedLeaveRequest<V, LF>,
863 parameters: PendingLeaveCommitParameters,
864 charges: [RetainedRecordCharge; 2],
865) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
866 let [terminal_charge, left_charge] = charges;
867 let terminal_delivery_seq = parameters.terminal_delivery_seq;
868 let LiveFrontierOwner {
869 frontiers,
870 closure_accounting,
871 mut retained_charges,
872 retained_record_limit,
873 } = owner;
874 let retired_marker_charge =
875 retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
876 let authority = frontiers
877 .prepare_pending_leave_authority(&member, pending)
878 .map_err(LiveLeaveError::Prepare)?;
879 let commit = commit_pending_leave(
880 member,
881 pending,
882 detach_cell,
883 verified,
884 authority,
885 parameters,
886 )
887 .map_err(LiveLeaveError::Commit)?;
888 let (identity, frontiers) = commit.into_parts();
889 let IdentityState::Retired(retired) = &identity else {
890 return Err(LiveLeaveError::Identity);
891 };
892 if retired.committed_result().prior_terminal_delivery_seq() != Some(terminal_delivery_seq)
893 || terminal_charge.delivery_seq() != terminal_delivery_seq
894 || terminal_charge.admission_order() != pending.admission_order()
895 || terminal_charge.encoded_charge().entries != 1
896 || left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
897 || left_charge.admission_order() != retired.left_admission_order()
898 || left_charge.encoded_charge().entries != 1
899 {
900 return Err(LiveLeaveError::RetainedCharge);
901 }
902 retained_charges.extend([terminal_charge, left_charge]);
903 retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
904 let retained_len = u64::try_from(frontiers.retained_records().len())
905 .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
906 if retained_len > retained_record_limit {
907 return Err(LiveLeaveError::RetainedRecordLimit);
908 }
909 if retained_charges.len() != frontiers.retained_records().len() {
910 return Err(LiveLeaveError::RetainedCharge);
911 }
912 let closure_accounting = accounting_after_leave(
913 closure_accounting,
914 &[terminal_charge, left_charge],
915 retired_marker_charge,
916 )
917 .ok_or(LiveLeaveError::ClosureAccounting)?;
918 Ok(LiveLeaveCommit {
919 identity,
920 owner: LiveFrontierOwner {
921 frontiers,
922 closure_accounting,
923 retained_charges,
924 retained_record_limit,
925 },
926 })
927}
928
929fn retired_marker_charge(
930 frontiers: &ClaimFrontiers,
931 retained_charges: &[RetainedRecordCharge],
932 participant_id: crate::wire::ParticipantId,
933) -> Result<Option<RetainedRecordCharge>, LiveLeaveError> {
934 let marker_sequence = frontiers
935 .retained_marker_records()
936 .iter()
937 .find_map(|record| {
938 matches!(
939 record.kind,
940 RetainedCausalRecordKind::CompactionMarker {
941 participant_index,
942 ..
943 } if participant_index == participant_id
944 )
945 .then_some(record.delivery_seq)
946 });
947 let Some(marker_sequence) = marker_sequence else {
948 return Ok(None);
949 };
950 retained_charges
951 .iter()
952 .copied()
953 .find(|charge| charge.delivery_seq() == marker_sequence)
954 .map(Some)
955 .ok_or(LiveLeaveError::RetainedCharge)
956}
957
958#[derive(Debug, PartialEq, Eq)]
960pub struct AttachFrontierCharges {
961 terminal: Option<RetainedRecordCharge>,
962 attached: RetainedRecordCharge,
963 seal: LiveTransitionInputSeal,
964}
965
966#[derive(Debug, PartialEq, Eq)]
967enum LiveTransitionInputSeal {
968 Validated,
969}
970
971impl AttachFrontierCharges {
972 #[must_use]
974 pub const fn new(
975 terminal: Option<RetainedRecordCharge>,
976 attached: RetainedRecordCharge,
977 ) -> Self {
978 Self {
979 terminal,
980 attached,
981 seal: LiveTransitionInputSeal::Validated,
982 }
983 }
984
985 const fn into_parts(self) -> (Option<RetainedRecordCharge>, RetainedRecordCharge) {
986 let Self {
987 terminal,
988 attached,
989 seal,
990 } = self;
991 match seal {
992 LiveTransitionInputSeal::Validated => (terminal, attached),
993 }
994 }
995}
996
997#[derive(Debug, PartialEq, Eq)]
999pub struct LiveFrontierCommit<T> {
1000 operation: T,
1001 owner: LiveFrontierOwner,
1002}
1003
1004impl<T> LiveFrontierCommit<T> {
1005 #[must_use]
1007 pub const fn operation(&self) -> &T {
1008 &self.operation
1009 }
1010
1011 #[must_use]
1013 pub const fn owner(&self) -> &LiveFrontierOwner {
1014 &self.owner
1015 }
1016
1017 #[must_use]
1019 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1020 (self.operation, self.owner)
1021 }
1022}
1023
1024#[derive(Debug, PartialEq, Eq)]
1026pub struct LiveFrontierFailure<T> {
1027 error: LiveFrontierError,
1028 operation: T,
1029 owner: LiveFrontierOwner,
1030}
1031
1032impl<T> LiveFrontierFailure<T> {
1033 #[must_use]
1035 pub const fn error(&self) -> LiveFrontierError {
1036 self.error
1037 }
1038
1039 #[must_use]
1041 pub fn into_parts(self) -> (T, LiveFrontierOwner) {
1042 (self.operation, self.owner)
1043 }
1044}
1045
1046#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1048pub enum LiveFrontierError {
1049 Authority,
1051 Precedence,
1053 RetainedCharge,
1055 RetainedRecordLimit,
1057 Frontier,
1059 ClosureAccounting,
1061}
1062
1063pub type LiveFrontierResult<T> = Result<LiveFrontierCommit<T>, Box<LiveFrontierFailure<T>>>;
1065
1066pub fn apply_enrollment_frontier<F>(
1072 owner: LiveFrontierOwner,
1073 operation: super::super::EnrollmentCommit<F>,
1074 attached_charge: RetainedRecordCharge,
1075) -> LiveFrontierResult<super::super::EnrollmentCommit<F>> {
1076 let attached = operation.attached;
1077 if attached.conversation_id() != owner.frontiers.conversation_id() {
1078 return failure(owner, operation, LiveFrontierError::Authority);
1079 }
1080 let participant_id = attached.participant_id();
1081 let mut active = owner.frontiers.active_identities().participants().to_vec();
1082 if active
1083 .iter()
1084 .any(|participant| participant.participant_index() == participant_id)
1085 {
1086 return failure(owner, operation, LiveFrontierError::Authority);
1087 }
1088 active.push(FrontierParticipant::new(
1089 participant_id,
1090 operation.member.cursor(),
1091 FrontierBinding::Bound(attached.binding_epoch()),
1092 ));
1093 active.sort_unstable_by_key(|participant| participant.participant_index());
1094 let rows = [retained_attached(attached)];
1095 let Some(sequence) =
1096 enrollment_sequence(owner.frontiers.sequence().ledger(), attached.delivery_seq())
1097 else {
1098 return failure(owner, operation, LiveFrontierError::Frontier);
1099 };
1100 let Some(order) = enrollment_order(
1101 owner.frontiers.order().ledger(),
1102 attached.admission_order().transaction_order(),
1103 ) else {
1104 return failure(owner, operation, LiveFrontierError::Frontier);
1105 };
1106 transition(
1107 owner,
1108 operation,
1109 active,
1110 &rows,
1111 vec![attached_charge],
1112 sequence,
1113 order,
1114 )
1115}
1116
1117pub fn apply_attach_frontier<F, V>(
1124 owner: LiveFrontierOwner,
1125 operation: AttachCommit<F, V>,
1126 charges: AttachFrontierCharges,
1127) -> LiveFrontierResult<AttachCommit<F, V>> {
1128 let (terminal_charge, attached_charge) = charges.into_parts();
1129 let attached = operation.attached;
1130 if attached.conversation_id() != owner.frontiers.conversation_id() {
1131 return failure(owner, operation, LiveFrontierError::Authority);
1132 }
1133 let mut active = owner.frontiers.active_identities().participants().to_vec();
1134 let Some(participant) = active
1135 .iter_mut()
1136 .find(|participant| participant.participant_index() == attached.participant_id())
1137 else {
1138 return failure(owner, operation, LiveFrontierError::Authority);
1139 };
1140 *participant = FrontierParticipant::new(
1141 participant.participant_index(),
1142 operation.member.cursor(),
1143 FrontierBinding::Bound(attached.binding_epoch()),
1144 );
1145 let current_sequence = owner.frontiers.sequence().ledger();
1146 let current_order = owner.frontiers.order().ledger();
1147 let (rows, keyed_charges, sequence, order) = match operation.transition {
1148 AttachTransition::Detached => {
1149 if terminal_charge.is_some() {
1150 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1151 }
1152 let Some(sequence) =
1153 detached_attach_sequence(current_sequence, attached.delivery_seq())
1154 else {
1155 return failure(owner, operation, LiveFrontierError::Frontier);
1156 };
1157 let Some(order) = detached_attach_order(
1158 current_order,
1159 attached.admission_order().transaction_order(),
1160 ) else {
1161 return failure(owner, operation, LiveFrontierError::Frontier);
1162 };
1163 (
1164 vec![retained_attached(attached)],
1165 vec![attached_charge],
1166 sequence,
1167 order,
1168 )
1169 }
1170 AttachTransition::Superseded { terminal } => {
1171 let Some(terminal_charge) = terminal_charge else {
1172 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1173 };
1174 let rows = vec![
1175 retained_terminal(terminal.into()),
1176 retained_attached(attached),
1177 ];
1178 let Some(sequence) = superseding_attach_sequence(current_sequence, &rows) else {
1179 return failure(owner, operation, LiveFrontierError::Frontier);
1180 };
1181 let Some(order) = superseding_attach_order(
1182 current_order,
1183 attached.admission_order().transaction_order(),
1184 ) else {
1185 return failure(owner, operation, LiveFrontierError::Frontier);
1186 };
1187 (
1188 rows,
1189 vec![terminal_charge, attached_charge],
1190 sequence,
1191 order,
1192 )
1193 }
1194 AttachTransition::FencedRecovery {
1195 prior_binding_epoch,
1196 composed_terminal,
1197 next_closure_state,
1198 } => {
1199 return apply_fenced_attach_frontier(
1200 owner,
1201 operation,
1202 terminal_charge,
1203 attached_charge,
1204 prior_binding_epoch,
1205 composed_terminal,
1206 next_closure_state,
1207 );
1208 }
1209 };
1210 transition(
1211 owner,
1212 operation,
1213 active,
1214 &rows,
1215 keyed_charges,
1216 sequence,
1217 order,
1218 )
1219}
1220
1221pub fn apply_detach_frontier<EF, V>(
1227 owner: LiveFrontierOwner,
1228 operation: CommittedDetachTransition<EF, V>,
1229 terminal_charge: RetainedRecordCharge,
1230) -> LiveFrontierResult<CommittedDetachTransition<EF, V>> {
1231 let terminal = operation.terminal();
1232 if terminal.conversation_id() != owner.frontiers.conversation_id() {
1233 return failure(owner, operation, LiveFrontierError::Authority);
1234 }
1235 let mut active = owner.frontiers.active_identities().participants().to_vec();
1236 let Some(participant) = active
1237 .iter_mut()
1238 .find(|participant| participant.participant_index() == terminal.participant_id())
1239 else {
1240 return failure(owner, operation, LiveFrontierError::Authority);
1241 };
1242 *participant = FrontierParticipant::new(
1243 participant.participant_index(),
1244 operation.member().cursor(),
1245 FrontierBinding::Detached(terminal.binding_epoch()),
1246 );
1247 let row = retained_terminal(terminal.into());
1248 let Some(sequence) = detach_sequence(owner.frontiers.sequence().ledger(), row.delivery_seq)
1249 else {
1250 return failure(owner, operation, LiveFrontierError::Frontier);
1251 };
1252 let Some(order) = detach_order(
1253 owner.frontiers.order().ledger(),
1254 row.admission_order.transaction_order(),
1255 ) else {
1256 return failure(owner, operation, LiveFrontierError::Frontier);
1257 };
1258 transition(
1259 owner,
1260 operation,
1261 active,
1262 &[row],
1263 vec![terminal_charge],
1264 sequence,
1265 order,
1266 )
1267}
1268
1269pub fn apply_participant_ack_frontier(
1275 mut owner: LiveFrontierOwner,
1276 operation: ParticipantAckCommit,
1277) -> LiveFrontierResult<ParticipantAckCommit> {
1278 let request = operation.outcome().request();
1279 let Some(current) = owner
1280 .frontiers
1281 .active_identities()
1282 .participants()
1283 .iter()
1284 .find(|participant| participant.participant_index() == request.participant_id)
1285 .copied()
1286 else {
1287 return failure(owner, operation, LiveFrontierError::Authority);
1288 };
1289 let participant = FrontierParticipant::new(
1290 request.participant_id,
1291 request.through_seq,
1292 current.binding(),
1293 );
1294 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1295 Ok(frontiers) => frontiers,
1296 Err(frontier_failure) => {
1297 let (frontiers, error) = *frontier_failure;
1298 owner.frontiers = frontiers;
1299 return failure(owner, operation, map_frontier_error(error));
1300 }
1301 };
1302 Ok(LiveFrontierCommit { operation, owner })
1303}
1304
1305pub fn apply_nonzero_participant_ack_frontier(
1315 mut owner: LiveFrontierOwner,
1316 operation: NonzeroParticipantAckCommit,
1317) -> LiveFrontierResult<NonzeroParticipantAckCommit> {
1318 let request = operation.outcome().request();
1319 let Some(current) = owner
1320 .frontiers
1321 .active_identities()
1322 .participants()
1323 .iter()
1324 .find(|participant| participant.participant_index() == request.participant_id)
1325 .copied()
1326 else {
1327 return failure(owner, operation, LiveFrontierError::Authority);
1328 };
1329 let participant = FrontierParticipant::new(
1330 request.participant_id,
1331 request.through_seq,
1332 current.binding(),
1333 );
1334 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1335 Ok(frontiers) => frontiers,
1336 Err(frontier_failure) => {
1337 let (frontiers, error) = *frontier_failure;
1338 owner.frontiers = frontiers;
1339 return failure(owner, operation, map_frontier_error(error));
1340 }
1341 };
1342 Ok(LiveFrontierCommit { operation, owner })
1343}
1344
1345pub fn apply_marker_ack_frontier(
1351 mut owner: LiveFrontierOwner,
1352 operation: MarkerAckCommit,
1353) -> LiveFrontierResult<MarkerAckCommit> {
1354 let request = operation.outcome().request();
1355 if !owner
1356 .frontiers
1357 .retained_marker_records()
1358 .iter()
1359 .any(|record| {
1360 record.delivery_seq == request.marker_delivery_seq
1361 && matches!(
1362 record.kind,
1363 RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
1364 if participant_index == request.participant_id
1365 )
1366 })
1367 {
1368 return failure(owner, operation, LiveFrontierError::Authority);
1369 }
1370 let Some(current) = owner
1371 .frontiers
1372 .active_identities()
1373 .participants()
1374 .iter()
1375 .find(|participant| participant.participant_index() == request.participant_id)
1376 .copied()
1377 else {
1378 return failure(owner, operation, LiveFrontierError::Authority);
1379 };
1380 let Some(accounting) = accounting_after_marker_ack(owner.closure_accounting) else {
1381 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1382 };
1383 let participant = FrontierParticipant::new(
1384 request.participant_id,
1385 request.marker_delivery_seq,
1386 current.binding(),
1387 );
1388 owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
1389 Ok(frontiers) => frontiers,
1390 Err(frontier_failure) => {
1391 let (frontiers, error) = *frontier_failure;
1392 owner.frontiers = frontiers;
1393 return failure(owner, operation, map_frontier_error(error));
1394 }
1395 };
1396 owner.closure_accounting = accounting;
1397 Ok(LiveFrontierCommit { operation, owner })
1398}
1399
1400fn apply_fenced_attach_frontier<F, V>(
1401 owner: LiveFrontierOwner,
1402 operation: AttachCommit<F, V>,
1403 terminal_charge: Option<RetainedRecordCharge>,
1404 attached_charge: RetainedRecordCharge,
1405 prior_binding_epoch: crate::wire::BindingEpoch,
1406 composed_terminal: Option<super::super::CommittedBindingTerminal>,
1407 next_closure_state: super::super::ClosureState,
1408) -> LiveFrontierResult<AttachCommit<F, V>> {
1409 let attached = operation.attached;
1410 let (rows, charges) = match (composed_terminal, terminal_charge) {
1411 (None, None) => (vec![retained_attached(attached)], vec![attached_charge]),
1412 (Some(terminal), Some(terminal_charge)) => (
1413 vec![retained_terminal(terminal), retained_attached(attached)],
1414 vec![terminal_charge, attached_charge],
1415 ),
1416 (None, Some(_)) | (Some(_), None) => {
1417 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1418 }
1419 };
1420 let participant = FrontierParticipant::new(
1421 attached.participant_id(),
1422 operation.member.cursor(),
1423 FrontierBinding::Bound(attached.binding_epoch()),
1424 );
1425 fenced_attach_transition(
1426 owner,
1427 operation,
1428 participant,
1429 prior_binding_epoch,
1430 next_closure_state,
1431 &rows,
1432 charges,
1433 )
1434}
1435
1436fn fenced_attach_transition<T>(
1437 mut owner: LiveFrontierOwner,
1438 operation: T,
1439 participant: FrontierParticipant,
1440 prior_binding_epoch: crate::wire::BindingEpoch,
1441 next_closure_state: super::super::ClosureState,
1442 rows: &[RetainedCausalRecord],
1443 charges: Vec<RetainedRecordCharge>,
1444) -> LiveFrontierResult<T> {
1445 if rows.len() != charges.len()
1446 || rows.iter().zip(&charges).any(|(row, charge)| {
1447 row.delivery_seq != charge.delivery_seq()
1448 || row.admission_order != charge.admission_order()
1449 || charge.encoded_charge().entries != 1
1450 })
1451 {
1452 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1453 }
1454 let resulting_len = owner
1455 .frontiers
1456 .retained_records()
1457 .len()
1458 .checked_add(rows.len());
1459 if resulting_len
1460 .and_then(|len| u64::try_from(len).ok())
1461 .is_none_or(|len| len > owner.retained_record_limit)
1462 {
1463 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1464 }
1465 let Some(accounting) =
1466 accounting_after_fenced_attach(owner.closure_accounting, &charges, next_closure_state)
1467 else {
1468 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1469 };
1470 owner.frontiers =
1471 match owner
1472 .frontiers
1473 .apply_live_fenced_attach(participant, prior_binding_epoch, rows)
1474 {
1475 Ok(frontiers) => frontiers,
1476 Err(frontier_failure) => {
1477 let (frontiers, error) = *frontier_failure;
1478 owner.frontiers = frontiers;
1479 return failure(owner, operation, map_frontier_error(error));
1480 }
1481 };
1482 owner.retained_charges.extend(charges);
1483 owner
1484 .retained_charges
1485 .sort_unstable_by_key(|charge| charge.delivery_seq());
1486 owner.closure_accounting = accounting;
1487 Ok(LiveFrontierCommit { operation, owner })
1488}
1489
1490fn transition<T>(
1491 mut owner: LiveFrontierOwner,
1492 operation: T,
1493 active: Vec<FrontierParticipant>,
1494 rows: &[RetainedCausalRecord],
1495 charges: Vec<RetainedRecordCharge>,
1496 sequence: SequenceLedger,
1497 order: OrderLedger,
1498) -> LiveFrontierResult<T> {
1499 if rows.len() != charges.len()
1500 || rows.iter().zip(&charges).any(|(row, charge)| {
1501 row.delivery_seq != charge.delivery_seq()
1502 || row.admission_order != charge.admission_order()
1503 || charge.encoded_charge().entries != 1
1504 })
1505 {
1506 return failure(owner, operation, LiveFrontierError::RetainedCharge);
1507 }
1508 let resulting_len = owner
1509 .frontiers
1510 .retained_records()
1511 .len()
1512 .checked_add(rows.len());
1513 if resulting_len
1514 .and_then(|len| u64::try_from(len).ok())
1515 .is_none_or(|len| len > owner.retained_record_limit)
1516 {
1517 return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
1518 }
1519 let Some(accounting) = accounting_after_rows(owner.closure_accounting, &charges) else {
1520 return failure(owner, operation, LiveFrontierError::ClosureAccounting);
1521 };
1522 owner.frontiers = match owner
1523 .frontiers
1524 .apply_live_transition(active, rows, sequence, order)
1525 {
1526 Ok(frontiers) => frontiers,
1527 Err(frontier_failure) => {
1528 let (frontiers, error) = *frontier_failure;
1529 owner.frontiers = frontiers;
1530 return failure(owner, operation, map_frontier_error(error));
1531 }
1532 };
1533 owner.retained_charges.extend(charges);
1534 owner.closure_accounting = accounting;
1535 Ok(LiveFrontierCommit { operation, owner })
1536}
1537
1538const fn map_frontier_error(error: LiveFrontierTransitionError) -> LiveFrontierError {
1539 match error {
1540 LiveFrontierTransitionError::Authority => LiveFrontierError::Authority,
1541 LiveFrontierTransitionError::Precedence => LiveFrontierError::Precedence,
1542 LiveFrontierTransitionError::RecordPosition
1543 | LiveFrontierTransitionError::Exhausted
1544 | LiveFrontierTransitionError::ResultingFrontier => LiveFrontierError::Frontier,
1545 }
1546}
1547
1548fn failure<T, U>(
1549 owner: LiveFrontierOwner,
1550 operation: T,
1551 error: LiveFrontierError,
1552) -> Result<U, Box<LiveFrontierFailure<T>>> {
1553 Err(Box::new(LiveFrontierFailure {
1554 error,
1555 operation,
1556 owner,
1557 }))
1558}