1use alloc::{boxed::Box, vec::Vec};
10
11use crate::{
12 algebra::ResourceVector,
13 outcome::CandidatePhase,
14 wire::{
15 BindingEpoch, ClosureCheckedEnvelope, DeliverySeq, OrderAllocatingEnvelope,
16 RecordAdmission, RecordAdmissionEnvelope, RecordAdmissionFaultClass,
17 RecordAdmissionProtocolFault, RecordAdmissionResponse, RecordCommitted,
18 SequenceAllocatingEnvelope, TransactionOrder,
19 },
20};
21
22use super::{
23 super::{
24 AdmissionOrder, BindingRequiredLookupResult, BindingState, CapacityCounter, ClaimFrontiers,
25 ClosureAccounting, ClosureState, ConnectionConversationCapacityCommit,
26 ConnectionConversationTracking, ImmutableSequenceCandidate, ObserverCheckedOperation,
27 ObserverFloorDecision, ObserverFloorPermit, OrderAdmissionError, OrderAllocation,
28 ParticipantBindingRequest, PresentedIdentity, RemainingClosureDecision,
29 RemainingClosurePermit, RequiredCapacityPlan, RequiredCapacityPlanError,
30 SemanticConnectionCapacityDecision, SequenceAdmission, SequenceAdmissionError, StoredEdge,
31 admit_sequence, allocate_order, check_observer_floor, check_record_size,
32 check_remaining_closure, lookup_binding_required, select_semantic_connection_capacity,
33 },
34 OrdinaryProjectionError, OrdinaryProjectionLimits, OrdinaryRecordProjectionDecision,
35 OrdinaryRecordProjectionInput, ProjectedOrdinaryRecord, RetainedRecordCharge,
36};
37
38#[derive(Debug)]
44pub struct RecordAdmissionPrestate<'a, EF, V, LF> {
45 request: RecordAdmission,
46 presented_identity: PresentedIdentity<'a, EF, V, LF>,
47 binding: &'a BindingState,
48 receiving_binding_epoch: BindingEpoch,
49 connection_tracking: ConnectionConversationTracking,
50 connection_capacity: CapacityCounter,
51 closure_accounting: ClosureAccounting,
52 max_ordinary_record_charge: ResourceVector,
53 frontiers: ClaimFrontiers,
54 retained_charges: Vec<RetainedRecordCharge>,
55 observer_progress: DeliverySeq,
56 projection_limits: OrdinaryProjectionLimits,
57}
58
59impl<'a, EF, V, LF> RecordAdmissionPrestate<'a, EF, V, LF> {
60 #[allow(clippy::too_many_arguments)]
63 #[must_use]
64 pub const fn new(
65 request: RecordAdmission,
66 presented_identity: PresentedIdentity<'a, EF, V, LF>,
67 binding: &'a BindingState,
68 receiving_binding_epoch: BindingEpoch,
69 connection_tracking: ConnectionConversationTracking,
70 connection_capacity: CapacityCounter,
71 closure_accounting: ClosureAccounting,
72 max_ordinary_record_charge: ResourceVector,
73 frontiers: ClaimFrontiers,
74 retained_charges: Vec<RetainedRecordCharge>,
75 observer_progress: DeliverySeq,
76 projection_limits: OrdinaryProjectionLimits,
77 ) -> Self {
78 Self {
79 request,
80 presented_identity,
81 binding,
82 receiving_binding_epoch,
83 connection_tracking,
84 connection_capacity,
85 closure_accounting,
86 max_ordinary_record_charge,
87 frontiers,
88 retained_charges,
89 observer_progress,
90 projection_limits,
91 }
92 }
93
94 #[must_use]
96 pub const fn request(&self) -> &RecordAdmission {
97 &self.request
98 }
99
100 #[must_use]
102 pub const fn binding(&self) -> &BindingState {
103 self.binding
104 }
105
106 #[must_use]
108 pub const fn receiving_binding_epoch(&self) -> BindingEpoch {
109 self.receiving_binding_epoch
110 }
111
112 #[must_use]
114 pub const fn frontiers(&self) -> &ClaimFrontiers {
115 &self.frontiers
116 }
117
118 #[must_use]
120 pub const fn connection_capacity(&self) -> CapacityCounter {
121 self.connection_capacity
122 }
123
124 #[must_use]
126 pub const fn closure_accounting(&self) -> ClosureAccounting {
127 self.closure_accounting
128 }
129
130 #[must_use]
132 pub const fn observer_progress(&self) -> DeliverySeq {
133 self.observer_progress
134 }
135
136 pub(super) fn into_live_owner_parts(
137 self,
138 ) -> (
139 RecordAdmission,
140 ClaimFrontiers,
141 ClosureAccounting,
142 Vec<RetainedRecordCharge>,
143 ) {
144 (
145 self.request,
146 self.frontiers,
147 self.closure_accounting,
148 self.retained_charges,
149 )
150 }
151
152 #[must_use]
154 pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
155 &self.retained_charges
156 }
157}
158
159#[derive(Debug)]
161pub struct UnchangedRecordAdmission<'a, EF, V, LF> {
162 prestate: RecordAdmissionPrestate<'a, EF, V, LF>,
163 encoded_record_charge: ResourceVector,
164}
165
166impl<'a, EF, V, LF> UnchangedRecordAdmission<'a, EF, V, LF> {
167 const fn new(
168 prestate: RecordAdmissionPrestate<'a, EF, V, LF>,
169 encoded_record_charge: ResourceVector,
170 ) -> Self {
171 Self {
172 prestate,
173 encoded_record_charge,
174 }
175 }
176
177 #[must_use]
179 pub const fn prestate(&self) -> &RecordAdmissionPrestate<'a, EF, V, LF> {
180 &self.prestate
181 }
182
183 #[must_use]
185 pub const fn encoded_record_charge(&self) -> ResourceVector {
186 self.encoded_record_charge
187 }
188
189 #[must_use]
191 pub fn into_parts(self) -> (RecordAdmissionPrestate<'a, EF, V, LF>, ResourceVector) {
192 (self.prestate, self.encoded_record_charge)
193 }
194}
195
196#[derive(Debug)]
198pub struct RecordAdmissionRefusal<'a, EF, V, LF> {
199 response: RecordAdmissionResponse,
200 unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
201}
202
203impl<'a, EF, V, LF> RecordAdmissionRefusal<'a, EF, V, LF> {
204 #[must_use]
206 pub const fn response(&self) -> &RecordAdmissionResponse {
207 &self.response
208 }
209
210 #[must_use]
212 pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
213 &self.unchanged
214 }
215
216 #[must_use]
218 pub fn into_parts(
219 self,
220 ) -> (
221 RecordAdmissionResponse,
222 UnchangedRecordAdmission<'a, EF, V, LF>,
223 ) {
224 (self.response, self.unchanged)
225 }
226}
227
228#[derive(Debug)]
230pub struct RecordAdmissionDrainFirst<'a, EF, V, LF> {
231 candidate: ImmutableSequenceCandidate,
232 unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
233}
234
235impl<'a, EF, V, LF> RecordAdmissionDrainFirst<'a, EF, V, LF> {
236 #[must_use]
238 pub const fn candidate(&self) -> ImmutableSequenceCandidate {
239 self.candidate
240 }
241
242 #[must_use]
244 pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
245 &self.unchanged
246 }
247
248 #[must_use]
250 pub fn into_parts(
251 self,
252 ) -> (
253 ImmutableSequenceCandidate,
254 UnchangedRecordAdmission<'a, EF, V, LF>,
255 ) {
256 (self.candidate, self.unchanged)
257 }
258}
259
260#[derive(Clone, Debug, PartialEq, Eq)]
262pub struct CommittedOrdinaryRecord {
263 request: RecordAdmission,
264 admission_order: AdmissionOrder,
265 delivery_seq: DeliverySeq,
266 encoded_record_charge: ResourceVector,
267}
268
269impl CommittedOrdinaryRecord {
270 #[must_use]
272 pub const fn request(&self) -> &RecordAdmission {
273 &self.request
274 }
275
276 #[must_use]
278 pub const fn admission_order(&self) -> AdmissionOrder {
279 self.admission_order
280 }
281
282 #[must_use]
284 pub const fn delivery_seq(&self) -> DeliverySeq {
285 self.delivery_seq
286 }
287
288 #[must_use]
290 pub const fn encoded_record_charge(&self) -> ResourceVector {
291 self.encoded_record_charge
292 }
293
294 const fn new(
295 request: RecordAdmission,
296 transaction_order: TransactionOrder,
297 delivery_seq: DeliverySeq,
298 encoded_record_charge: ResourceVector,
299 ) -> Self {
300 let participant_id = request.participant_id;
301 Self {
302 request,
303 admission_order: AdmissionOrder::new(
304 transaction_order,
305 CandidatePhase::OrdinaryRecord,
306 participant_id,
307 ),
308 delivery_seq,
309 encoded_record_charge,
310 }
311 }
312}
313
314#[derive(Debug, PartialEq, Eq)]
316pub struct RecordAdmissionCommit {
317 outcome: RecordCommitted,
318 record: CommittedOrdinaryRecord,
319 connection_capacity: ConnectionConversationCapacityCommit,
320 projection: Box<ProjectedOrdinaryRecord>,
321}
322
323#[derive(Debug)]
332pub struct RecordAdmissionPersistenceParts {
333 pub outcome: RecordCommitted,
335 pub record: CommittedOrdinaryRecord,
337 pub connection_capacity: ConnectionConversationCapacityCommit,
339 pub order: OrderAllocation,
341 pub sequence: SequenceAdmission,
343 pub observer_floor: ObserverFloorPermit,
345 pub closure: RemainingClosurePermit,
347 pub frontiers: ClaimFrontiers,
349 pub floor: crate::algebra::FloorComputation,
351 pub retained_charge: crate::algebra::WideResourceVector,
353 pub baseline: crate::algebra::WideResourceVector,
355 pub accounting: ClosureAccounting,
357 pub required_capacity: RequiredCapacityPlan,
359 pub caller_record: super::super::RetainedCausalRecord,
361 pub caller_charge: RetainedRecordCharge,
363 pub retained_charges: Vec<RetainedRecordCharge>,
365 pub marker_candidates: Vec<super::super::MarkerCandidateAuthority>,
367}
368
369impl RecordAdmissionCommit {
370 #[must_use]
372 pub const fn outcome(&self) -> &RecordCommitted {
373 &self.outcome
374 }
375
376 #[must_use]
378 pub const fn record(&self) -> &CommittedOrdinaryRecord {
379 &self.record
380 }
381
382 #[must_use]
384 pub const fn connection_capacity(&self) -> ConnectionConversationCapacityCommit {
385 self.connection_capacity
386 }
387
388 #[must_use]
390 pub const fn order(&self) -> OrderAllocation {
391 self.projection.order()
392 }
393
394 #[must_use]
396 pub const fn sequence(&self) -> SequenceAdmission {
397 self.projection.sequence()
398 }
399
400 #[must_use]
402 pub const fn observer_floor(&self) -> ObserverFloorPermit {
403 self.projection.observer_floor()
404 }
405
406 #[must_use]
408 pub const fn closure(&self) -> &RemainingClosurePermit {
409 &self.projection.closure
410 }
411
412 #[must_use]
414 pub const fn projection(&self) -> &ProjectedOrdinaryRecord {
415 &self.projection
416 }
417
418 #[must_use]
421 pub fn into_persistence_parts(self) -> RecordAdmissionPersistenceParts {
422 let ProjectedOrdinaryRecord {
423 frontiers,
424 floor,
425 retained_charge,
426 baseline,
427 accounting,
428 required_capacity,
429 order,
430 sequence,
431 observer_floor,
432 closure,
433 caller_record,
434 caller_charge,
435 retained_charges,
436 new_marker_candidates,
437 } = *self.projection;
438 RecordAdmissionPersistenceParts {
439 outcome: self.outcome,
440 record: self.record,
441 connection_capacity: self.connection_capacity,
442 order,
443 sequence,
444 observer_floor,
445 closure,
446 frontiers,
447 floor,
448 retained_charge,
449 baseline,
450 accounting,
451 required_capacity,
452 caller_record,
453 caller_charge,
454 retained_charges,
455 marker_candidates: new_marker_candidates,
456 }
457 }
458}
459
460#[derive(Clone, Debug, PartialEq, Eq)]
462pub enum RecordAdmissionFault {
463 Projection(OrdinaryProjectionError),
465 Order(OrderAdmissionError),
467 Sequence(SequenceAdmissionError),
469 RequiredCapacity(RequiredCapacityPlanError),
471 RefusalInvariant,
473}
474
475impl RecordAdmissionFault {
476 #[must_use]
483 pub const fn class(&self) -> RecordAdmissionFaultClass {
484 match self {
485 Self::Projection(_) => RecordAdmissionFaultClass::Projection,
486 Self::Order(_) => RecordAdmissionFaultClass::Order,
487 Self::Sequence(_) => RecordAdmissionFaultClass::Sequence,
488 Self::RequiredCapacity(_) => RecordAdmissionFaultClass::RequiredCapacity,
489 Self::RefusalInvariant => RecordAdmissionFaultClass::RefusalInvariant,
490 }
491 }
492}
493
494#[derive(Debug)]
496pub struct RecordAdmissionFailure<'a, EF, V, LF> {
497 fault: RecordAdmissionFault,
498 unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
499}
500
501impl<'a, EF, V, LF> RecordAdmissionFailure<'a, EF, V, LF> {
502 #[must_use]
504 pub const fn fault(&self) -> &RecordAdmissionFault {
505 &self.fault
506 }
507
508 #[must_use]
510 pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
511 &self.unchanged
512 }
513
514 #[must_use]
516 pub fn into_parts(
517 self,
518 ) -> (
519 RecordAdmissionFault,
520 UnchangedRecordAdmission<'a, EF, V, LF>,
521 ) {
522 (self.fault, self.unchanged)
523 }
524
525 #[must_use]
541 pub fn into_terminal_refusal(
542 self,
543 ) -> (
544 RecordAdmissionResponse,
545 RecordAdmissionFault,
546 UnchangedRecordAdmission<'a, EF, V, LF>,
547 ) {
548 let request = self.unchanged.prestate().request();
549 let response = RecordAdmissionResponse::from_protocol_fault(RecordAdmissionProtocolFault {
550 request: RecordAdmissionEnvelope {
551 conversation_id: request.conversation_id,
552 participant_id: request.participant_id,
553 capability_generation: request.capability_generation,
554 record_admission_attempt_token: request.record_admission_attempt_token,
555 },
556 class: self.fault.class(),
557 });
558 (response, self.fault, self.unchanged)
559 }
560}
561
562#[derive(Debug)]
564pub enum RecordAdmissionDecision<'a, EF, V, LF> {
565 Respond(Box<RecordAdmissionRefusal<'a, EF, V, LF>>),
567 DrainFirst(Box<RecordAdmissionDrainFirst<'a, EF, V, LF>>),
569 Commit(Box<RecordAdmissionCommit>),
571 Fault(Box<RecordAdmissionFailure<'a, EF, V, LF>>),
573}
574
575#[must_use]
586pub fn classify_record_admission_binding<EF, V, LF>(
587 presented_identity: PresentedIdentity<'_, EF, V, LF>,
588 binding: &BindingState,
589 receiving_binding_epoch: BindingEpoch,
590 request: &RecordAdmission,
591) -> Option<RecordAdmissionResponse> {
592 let lookup_request = ParticipantBindingRequest::RecordAdmission(request.clone());
593 match lookup_binding_required(
594 presented_identity,
595 binding,
596 Some(receiving_binding_epoch),
597 &lookup_request,
598 ) {
599 BindingRequiredLookupResult::Retired(value) => {
600 Some(RecordAdmissionResponse::from_retired(value))
601 }
602 BindingRequiredLookupResult::ParticipantUnknown(value) => {
603 Some(RecordAdmissionResponse::from_participant_unknown(value))
604 }
605 BindingRequiredLookupResult::StaleAuthority(value) => {
606 Some(RecordAdmissionResponse::from_stale_authority(value))
607 }
608 BindingRequiredLookupResult::NoBinding(value) => {
609 Some(RecordAdmissionResponse::from_no_binding(value))
610 }
611 BindingRequiredLookupResult::Authorized { .. } => None,
612 }
613}
614
615#[must_use]
622#[allow(
623 clippy::too_many_lines,
624 reason = "the operation keeps the frozen total selector order visible in one function"
625)]
626pub fn apply_record_admission<EF, V, LF>(
627 input: RecordAdmissionPrestate<'_, EF, V, LF>,
628 encoded_record_charge: ResourceVector,
629) -> RecordAdmissionDecision<'_, EF, V, LF> {
630 let envelope = record_envelope(&input.request);
631
632 let lookup_request = ParticipantBindingRequest::RecordAdmission(input.request.clone());
633 match lookup_binding_required(
634 input.presented_identity,
635 input.binding,
636 Some(input.receiving_binding_epoch),
637 &lookup_request,
638 ) {
639 BindingRequiredLookupResult::Retired(value) => {
640 return refused(
641 input,
642 encoded_record_charge,
643 RecordAdmissionResponse::from_retired(value),
644 );
645 }
646 BindingRequiredLookupResult::ParticipantUnknown(value) => {
647 return refused(
648 input,
649 encoded_record_charge,
650 RecordAdmissionResponse::from_participant_unknown(value),
651 );
652 }
653 BindingRequiredLookupResult::StaleAuthority(value) => {
654 return refused(
655 input,
656 encoded_record_charge,
657 RecordAdmissionResponse::from_stale_authority(value),
658 );
659 }
660 BindingRequiredLookupResult::NoBinding(value) => {
661 return refused(
662 input,
663 encoded_record_charge,
664 RecordAdmissionResponse::from_no_binding(value),
665 );
666 }
667 BindingRequiredLookupResult::Authorized { .. } => {}
668 }
669
670 let connection_capacity = match select_semantic_connection_capacity(
671 input.connection_tracking,
672 input.connection_capacity,
673 ) {
674 SemanticConnectionCapacityDecision::Commit(value) => value,
675 SemanticConnectionCapacityDecision::Respond { limit } => {
676 let response =
677 RecordAdmissionResponse::connection_conversation_capacity_exceeded(envelope, limit);
678 return refused(input, encoded_record_charge, response);
679 }
680 };
681
682 let size = match check_record_size(
683 envelope.clone(),
684 encoded_record_charge,
685 input.max_ordinary_record_charge,
686 ) {
687 super::super::RecordSizeDecision::Eligible(value) => value,
688 super::super::RecordSizeDecision::Respond(value) => {
689 return refused(
690 input,
691 encoded_record_charge,
692 RecordAdmissionResponse::record_too_large(value),
693 );
694 }
695 };
696
697 if input.frontiers.sequence().immutable_candidates().is_empty()
698 && !matches!(input.closure_accounting.state(), ClosureState::Clear)
699 {
700 return match nonzero_debt_response(
701 &envelope,
702 &input.frontiers,
703 input.closure_accounting,
704 input.observer_progress,
705 input.projection_limits,
706 ) {
707 Ok(response) => refused(input, encoded_record_charge, response),
708 Err(operation_fault) => fault(input, encoded_record_charge, operation_fault),
709 };
710 }
711
712 let RecordAdmissionPrestate {
713 request,
714 presented_identity,
715 binding,
716 receiving_binding_epoch,
717 connection_tracking,
718 connection_capacity: original_connection_capacity,
719 closure_accounting,
720 max_ordinary_record_charge,
721 frontiers,
722 retained_charges,
723 observer_progress,
724 projection_limits,
725 } = input;
726 let shell = RecordAdmissionProjectionShell {
727 request,
728 presented_identity,
729 binding,
730 connection_tracking,
731 connection_capacity: original_connection_capacity,
732 max_ordinary_record_charge,
733 };
734 let projection_input = OrdinaryRecordProjectionInput::new(
735 envelope.clone(),
736 receiving_binding_epoch,
737 size.encoded_record_charge(),
738 retained_charges,
739 observer_progress,
740 closure_accounting,
741 projection_limits,
742 );
743 let projected = match frontiers.project_ordinary_record(projection_input) {
744 Ok(OrdinaryRecordProjectionDecision::DrainFirst(value)) => {
745 let candidate = value.candidate();
746 let (frontiers, projection_input) = value.into_unchanged_parts();
747 let unchanged = UnchangedRecordAdmission::new(
748 shell.rebuild(frontiers, projection_input),
749 encoded_record_charge,
750 );
751 return RecordAdmissionDecision::DrainFirst(Box::new(RecordAdmissionDrainFirst {
752 candidate,
753 unchanged,
754 }));
755 }
756 Ok(OrdinaryRecordProjectionDecision::Projected(value)) => value,
757 Err(failure) => {
758 let (frontiers, projection_input, error) = failure.into_parts();
759 let prestate = shell.rebuild(frontiers, projection_input);
760 return match projection_failure(error, &envelope, closure_accounting) {
761 Ok(response) => refused(prestate, encoded_record_charge, response),
762 Err(operation_fault) => fault(prestate, encoded_record_charge, operation_fault),
763 };
764 }
765 };
766
767 let order = projected.order();
768 let sequence = projected.sequence();
769 let delivery_seq = sequence.resulting().high_watermark();
770 let record = CommittedOrdinaryRecord::new(
771 shell.request,
772 order.major(),
773 delivery_seq,
774 size.encoded_record_charge(),
775 );
776 RecordAdmissionDecision::Commit(Box::new(RecordAdmissionCommit {
777 outcome: RecordCommitted::new(envelope, delivery_seq),
778 record,
779 connection_capacity,
780 projection: projected,
781 }))
782}
783
784struct RecordAdmissionProjectionShell<'a, EF, V, LF> {
785 request: RecordAdmission,
786 presented_identity: PresentedIdentity<'a, EF, V, LF>,
787 binding: &'a BindingState,
788 connection_tracking: ConnectionConversationTracking,
789 connection_capacity: CapacityCounter,
790 max_ordinary_record_charge: ResourceVector,
791}
792
793impl<'a, EF, V, LF> RecordAdmissionProjectionShell<'a, EF, V, LF> {
794 fn rebuild(
795 self,
796 frontiers: ClaimFrontiers,
797 projection: OrdinaryRecordProjectionInput,
798 ) -> RecordAdmissionPrestate<'a, EF, V, LF> {
799 let (
800 _envelope,
801 receiving_binding_epoch,
802 _encoded_record_charge,
803 retained_charges,
804 observer_progress,
805 closure_accounting,
806 projection_limits,
807 ) = projection.into_parts();
808 RecordAdmissionPrestate {
809 request: self.request,
810 presented_identity: self.presented_identity,
811 binding: self.binding,
812 receiving_binding_epoch,
813 connection_tracking: self.connection_tracking,
814 connection_capacity: self.connection_capacity,
815 closure_accounting,
816 max_ordinary_record_charge: self.max_ordinary_record_charge,
817 frontiers,
818 retained_charges,
819 observer_progress,
820 projection_limits,
821 }
822 }
823}
824
825fn refused<EF, V, LF>(
826 prestate: RecordAdmissionPrestate<'_, EF, V, LF>,
827 encoded_record_charge: ResourceVector,
828 response: RecordAdmissionResponse,
829) -> RecordAdmissionDecision<'_, EF, V, LF> {
830 RecordAdmissionDecision::Respond(Box::new(RecordAdmissionRefusal {
831 response,
832 unchanged: UnchangedRecordAdmission::new(prestate, encoded_record_charge),
833 }))
834}
835
836fn fault<EF, V, LF>(
837 prestate: RecordAdmissionPrestate<'_, EF, V, LF>,
838 encoded_record_charge: ResourceVector,
839 operation_fault: RecordAdmissionFault,
840) -> RecordAdmissionDecision<'_, EF, V, LF> {
841 RecordAdmissionDecision::Fault(Box::new(RecordAdmissionFailure {
842 fault: operation_fault,
843 unchanged: UnchangedRecordAdmission::new(prestate, encoded_record_charge),
844 }))
845}
846
847fn nonzero_debt_response(
848 envelope: &RecordAdmissionEnvelope,
849 frontiers: &ClaimFrontiers,
850 accounting: ClosureAccounting,
851 observer_progress: DeliverySeq,
852 limits: OrdinaryProjectionLimits,
853) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
854 let order = match allocate_order(
855 OrderAllocatingEnvelope::RecordAdmission(envelope.clone()),
856 frontiers.order().ledger(),
857 frontiers.order().ledger().plan_ordinary_record(),
858 ) {
859 Ok(value) => value,
860 Err(error) => return order_failure(error),
861 };
862 let sequence_plan = match frontiers.sequence().ledger().plan_ordinary_record(0) {
863 Ok(value) => value,
864 Err(error) => return sequence_failure(error),
865 };
866 if let Err(error) = admit_sequence(
867 SequenceAllocatingEnvelope::RecordAdmission(envelope.clone()),
868 sequence_plan,
869 ) {
870 return sequence_failure(error);
871 }
872 match check_observer_floor(
873 ObserverCheckedOperation::RecordAdmission(envelope.clone()),
874 observer_progress,
875 frontiers.retained_floor(),
876 ) {
877 ObserverFloorDecision::Eligible(_) => {}
878 ObserverFloorDecision::Respond(value) => {
879 return Ok(RecordAdmissionResponse::from_observer_backpressure(value));
880 }
881 }
882 let required = match RequiredCapacityPlan::ordinary(
883 accounting.baseline(),
884 limits.mandatory_bound(),
885 accounting.edge_k_remaining(),
886 ) {
887 Ok(value) => value,
888 Err(error) => {
889 return Err(RecordAdmissionFault::RequiredCapacity(error));
890 }
891 };
892 let delivered_marker_awaiting_ack = matches!(
893 accounting.state(),
894 ClosureState::Owed {
895 edge: StoredEdge::ParticipantCursorProgress(progress),
896 ..
897 } if progress.marker_delivery_seq().is_some()
898 );
899 match check_remaining_closure(
900 &ClosureCheckedEnvelope::RecordAdmission(envelope.clone()),
901 accounting,
902 delivered_marker_awaiting_ack,
903 0,
904 required,
905 ) {
906 RemainingClosureDecision::Respond(value) => {
907 Ok(RecordAdmissionResponse::from_marker_closure_capacity_exceeded(value))
908 }
909 RemainingClosureDecision::Eligible(_) => {
910 let _ = order;
911 Err(RecordAdmissionFault::RefusalInvariant)
912 }
913 }
914}
915
916fn projection_failure(
917 error: OrdinaryProjectionError,
918 envelope: &RecordAdmissionEnvelope,
919 accounting: ClosureAccounting,
920) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
921 match error {
922 OrdinaryProjectionError::Order(error) => order_failure(error),
923 OrdinaryProjectionError::Sequence(error) => sequence_failure(error),
924 OrdinaryProjectionError::ObserverBackpressure {
925 cap_floor,
926 observer_progress,
927 } => match check_observer_floor(
928 ObserverCheckedOperation::RecordAdmission(envelope.clone()),
929 observer_progress,
930 cap_floor,
931 ) {
932 ObserverFloorDecision::Respond(value) => {
933 Ok(RecordAdmissionResponse::from_observer_backpressure(value))
934 }
935 ObserverFloorDecision::Eligible(_) => Err(RecordAdmissionFault::Projection(
936 OrdinaryProjectionError::ObserverBackpressure {
937 cap_floor,
938 observer_progress,
939 },
940 )),
941 },
942 OrdinaryProjectionError::Capacity { required, .. }
943 | OrdinaryProjectionError::MarkerAnchorCapacity { required, .. } => {
944 capacity_failure(required, envelope, accounting)
945 }
946 other => Err(RecordAdmissionFault::Projection(other)),
947 }
948}
949
950fn capacity_failure(
951 required: crate::algebra::WideResourceVector,
952 envelope: &RecordAdmissionEnvelope,
953 accounting: ClosureAccounting,
954) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
955 let required_capacity = match RequiredCapacityPlan::from_successors(&[required]) {
956 Ok(value) => value,
957 Err(error) => {
958 return Err(RecordAdmissionFault::RequiredCapacity(error));
959 }
960 };
961 match check_remaining_closure(
962 &ClosureCheckedEnvelope::RecordAdmission(envelope.clone()),
963 accounting,
964 false,
965 0,
966 required_capacity,
967 ) {
968 RemainingClosureDecision::Respond(value) => {
969 Ok(RecordAdmissionResponse::from_marker_closure_capacity_exceeded(value))
970 }
971 RemainingClosureDecision::Eligible(_) => Err(RecordAdmissionFault::RefusalInvariant),
972 }
973}
974
975fn order_failure(
976 error: OrderAdmissionError,
977) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
978 match error {
979 OrderAdmissionError::Exhausted(value) => Ok(
980 RecordAdmissionResponse::from_conversation_order_exhausted(value),
981 ),
982 other => Err(RecordAdmissionFault::Order(other)),
983 }
984}
985
986fn sequence_failure(
987 error: SequenceAdmissionError,
988) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
989 match error {
990 SequenceAdmissionError::Exhausted(value) => {
991 Ok(RecordAdmissionResponse::from_conversation_sequence_exhausted(value))
992 }
993 other => Err(RecordAdmissionFault::Sequence(other)),
994 }
995}
996
997const fn record_envelope(request: &RecordAdmission) -> RecordAdmissionEnvelope {
998 RecordAdmissionEnvelope {
999 conversation_id: request.conversation_id,
1000 participant_id: request.participant_id,
1001 capability_generation: request.capability_generation,
1002 record_admission_attempt_token: request.record_admission_attempt_token,
1003 }
1004}