liminal_protocol/lifecycle/operations/
record_admission.rs

1//! Total ordinary-record admission after transport and capability negotiation.
2//!
3//! The operation preserves the frozen selector order and owns the validated
4//! [`ClaimFrontiers`] value it projects. A server supplies transport facts,
5//! exact durability charges, and signed limits; lookup, capacity, size, global
6//! candidate precedence, order/sequence admission, observer retention, closure
7//! capacity, and the final record/outcome are all selected by protocol APIs.
8
9use alloc::{boxed::Box, vec::Vec};
10
11use crate::{
12    algebra::ResourceVector,
13    outcome::CandidatePhase,
14    wire::{
15        BindingEpoch, ClosureCheckedEnvelope, DeliverySeq, OrderAllocatingEnvelope,
16        RecordAdmission, RecordAdmissionEnvelope, RecordAdmissionResponse, RecordCommitted,
17        SequenceAllocatingEnvelope, TransactionOrder,
18    },
19};
20
21use super::{
22    super::{
23        AdmissionOrder, BindingRequiredLookupResult, BindingState, CapacityCounter, ClaimFrontiers,
24        ClosureAccounting, ClosureState, ConnectionConversationCapacityCommit,
25        ConnectionConversationTracking, ImmutableSequenceCandidate, ObserverCheckedOperation,
26        ObserverFloorDecision, ObserverFloorPermit, OrderAdmissionError, OrderAllocation,
27        ParticipantBindingRequest, PresentedIdentity, RemainingClosureDecision,
28        RemainingClosurePermit, RequiredCapacityPlan, RequiredCapacityPlanError,
29        SemanticConnectionCapacityDecision, SequenceAdmission, SequenceAdmissionError, StoredEdge,
30        admit_sequence, allocate_order, check_observer_floor, check_record_size,
31        check_remaining_closure, lookup_binding_required, select_semantic_connection_capacity,
32    },
33    OrdinaryProjectionError, OrdinaryProjectionLimits, OrdinaryRecordProjectionDecision,
34    OrdinaryRecordProjectionInput, ProjectedOrdinaryRecord, RetainedRecordCharge,
35};
36
37/// Complete unchanged durable prestate consumed by ordinary admission.
38///
39/// Order/sequence ledgers, retained rows, marker owners, and participant cursors
40/// are intentionally absent as independent fields: the owned [`ClaimFrontiers`]
41/// carries them as one validated authority.
42#[derive(Debug)]
43pub struct RecordAdmissionPrestate<'a, EF, V, LF> {
44    request: RecordAdmission,
45    presented_identity: PresentedIdentity<'a, EF, V, LF>,
46    binding: &'a BindingState,
47    receiving_binding_epoch: BindingEpoch,
48    connection_tracking: ConnectionConversationTracking,
49    connection_capacity: CapacityCounter,
50    closure_accounting: ClosureAccounting,
51    max_ordinary_record_charge: ResourceVector,
52    frontiers: ClaimFrontiers,
53    retained_charges: Vec<RetainedRecordCharge>,
54    observer_progress: DeliverySeq,
55    projection_limits: OrdinaryProjectionLimits,
56}
57
58impl<'a, EF, V, LF> RecordAdmissionPrestate<'a, EF, V, LF> {
59    /// Captures the exact request, lookup/capacity state, complete validated
60    /// frontiers, factual retained charges, observer state, and signed limits.
61    #[allow(clippy::too_many_arguments)]
62    #[must_use]
63    pub const fn new(
64        request: RecordAdmission,
65        presented_identity: PresentedIdentity<'a, EF, V, LF>,
66        binding: &'a BindingState,
67        receiving_binding_epoch: BindingEpoch,
68        connection_tracking: ConnectionConversationTracking,
69        connection_capacity: CapacityCounter,
70        closure_accounting: ClosureAccounting,
71        max_ordinary_record_charge: ResourceVector,
72        frontiers: ClaimFrontiers,
73        retained_charges: Vec<RetainedRecordCharge>,
74        observer_progress: DeliverySeq,
75        projection_limits: OrdinaryProjectionLimits,
76    ) -> Self {
77        Self {
78            request,
79            presented_identity,
80            binding,
81            receiving_binding_epoch,
82            connection_tracking,
83            connection_capacity,
84            closure_accounting,
85            max_ordinary_record_charge,
86            frontiers,
87            retained_charges,
88            observer_progress,
89            projection_limits,
90        }
91    }
92
93    /// Borrows the exact payload-bearing request.
94    #[must_use]
95    pub const fn request(&self) -> &RecordAdmission {
96        &self.request
97    }
98
99    /// Borrows the unchanged authoritative binding state used by lookup.
100    #[must_use]
101    pub const fn binding(&self) -> &BindingState {
102        self.binding
103    }
104
105    /// Returns the unchanged receiving binding epoch.
106    #[must_use]
107    pub const fn receiving_binding_epoch(&self) -> BindingEpoch {
108        self.receiving_binding_epoch
109    }
110
111    /// Borrows the unchanged validated claim-frontier aggregate.
112    #[must_use]
113    pub const fn frontiers(&self) -> &ClaimFrontiers {
114        &self.frontiers
115    }
116
117    /// Returns the unchanged semantic connection-capacity counter.
118    #[must_use]
119    pub const fn connection_capacity(&self) -> CapacityCounter {
120        self.connection_capacity
121    }
122
123    /// Returns the unchanged closure-accounting snapshot.
124    #[must_use]
125    pub const fn closure_accounting(&self) -> ClosureAccounting {
126        self.closure_accounting
127    }
128
129    /// Returns the unchanged hard-observer progress.
130    #[must_use]
131    pub const fn observer_progress(&self) -> DeliverySeq {
132        self.observer_progress
133    }
134
135    pub(super) fn into_live_owner_parts(
136        self,
137    ) -> (
138        RecordAdmission,
139        ClaimFrontiers,
140        ClosureAccounting,
141        Vec<RetainedRecordCharge>,
142    ) {
143        (
144            self.request,
145            self.frontiers,
146            self.closure_accounting,
147            self.retained_charges,
148        )
149    }
150
151    /// Borrows exact keyed durable charges for the retained suffix.
152    #[must_use]
153    pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
154        &self.retained_charges
155    }
156}
157
158/// Complete unchanged operation state returned by every noncommit decision.
159#[derive(Debug)]
160pub struct UnchangedRecordAdmission<'a, EF, V, LF> {
161    prestate: RecordAdmissionPrestate<'a, EF, V, LF>,
162    encoded_record_charge: ResourceVector,
163}
164
165impl<'a, EF, V, LF> UnchangedRecordAdmission<'a, EF, V, LF> {
166    const fn new(
167        prestate: RecordAdmissionPrestate<'a, EF, V, LF>,
168        encoded_record_charge: ResourceVector,
169    ) -> Self {
170        Self {
171            prestate,
172            encoded_record_charge,
173        }
174    }
175
176    /// Borrows the exact reusable prestate.
177    #[must_use]
178    pub const fn prestate(&self) -> &RecordAdmissionPrestate<'a, EF, V, LF> {
179        &self.prestate
180    }
181
182    /// Returns the unchanged encoded caller-record charge.
183    #[must_use]
184    pub const fn encoded_record_charge(&self) -> ResourceVector {
185        self.encoded_record_charge
186    }
187
188    /// Recovers all input needed to replay the pure operation.
189    #[must_use]
190    pub fn into_parts(self) -> (RecordAdmissionPrestate<'a, EF, V, LF>, ResourceVector) {
191        (self.prestate, self.encoded_record_charge)
192    }
193}
194
195/// Exact wire response paired with the unchanged replayable aggregate.
196#[derive(Debug)]
197pub struct RecordAdmissionRefusal<'a, EF, V, LF> {
198    response: RecordAdmissionResponse,
199    unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
200}
201
202impl<'a, EF, V, LF> RecordAdmissionRefusal<'a, EF, V, LF> {
203    /// Borrows the selected request-bound wire response.
204    #[must_use]
205    pub const fn response(&self) -> &RecordAdmissionResponse {
206        &self.response
207    }
208
209    /// Borrows the unchanged replayable aggregate.
210    #[must_use]
211    pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
212        &self.unchanged
213    }
214
215    /// Recovers the response and complete unchanged operation state.
216    #[must_use]
217    pub fn into_parts(
218        self,
219    ) -> (
220        RecordAdmissionResponse,
221        UnchangedRecordAdmission<'a, EF, V, LF>,
222    ) {
223        (self.response, self.unchanged)
224    }
225}
226
227/// Earlier immutable candidate paired with the unchanged replayable aggregate.
228#[derive(Debug)]
229pub struct RecordAdmissionDrainFirst<'a, EF, V, LF> {
230    candidate: ImmutableSequenceCandidate,
231    unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
232}
233
234impl<'a, EF, V, LF> RecordAdmissionDrainFirst<'a, EF, V, LF> {
235    /// Returns the exact lowest immutable candidate selected by the frontier.
236    #[must_use]
237    pub const fn candidate(&self) -> ImmutableSequenceCandidate {
238        self.candidate
239    }
240
241    /// Borrows the unchanged replayable aggregate.
242    #[must_use]
243    pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
244        &self.unchanged
245    }
246
247    /// Recovers the candidate and complete unchanged operation state.
248    #[must_use]
249    pub fn into_parts(
250        self,
251    ) -> (
252        ImmutableSequenceCandidate,
253        UnchangedRecordAdmission<'a, EF, V, LF>,
254    ) {
255        (self.candidate, self.unchanged)
256    }
257}
258
259/// Ordinary record selected for one successful atomic transaction.
260#[derive(Clone, Debug, PartialEq, Eq)]
261pub struct CommittedOrdinaryRecord {
262    request: RecordAdmission,
263    admission_order: AdmissionOrder,
264    delivery_seq: DeliverySeq,
265    encoded_record_charge: ResourceVector,
266}
267
268impl CommittedOrdinaryRecord {
269    /// Borrows the exact request including its opaque payload.
270    #[must_use]
271    pub const fn request(&self) -> &RecordAdmission {
272        &self.request
273    }
274
275    /// Returns the protocol-derived phase-3 admission key.
276    #[must_use]
277    pub const fn admission_order(&self) -> AdmissionOrder {
278        self.admission_order
279    }
280
281    /// Returns the assigned gap-free delivery sequence.
282    #[must_use]
283    pub const fn delivery_seq(&self) -> DeliverySeq {
284        self.delivery_seq
285    }
286
287    /// Returns the exact encoded charge that passed static size admission.
288    #[must_use]
289    pub const fn encoded_record_charge(&self) -> ResourceVector {
290        self.encoded_record_charge
291    }
292
293    const fn new(
294        request: RecordAdmission,
295        transaction_order: TransactionOrder,
296        delivery_seq: DeliverySeq,
297        encoded_record_charge: ResourceVector,
298    ) -> Self {
299        let participant_id = request.participant_id;
300        Self {
301            request,
302            admission_order: AdmissionOrder::new(
303                transaction_order,
304                CandidatePhase::OrdinaryRecord,
305                participant_id,
306            ),
307            delivery_seq,
308            encoded_record_charge,
309        }
310    }
311}
312
313/// Atomic ordinary-record commit selected by every shared admission gate.
314#[derive(Debug, PartialEq, Eq)]
315pub struct RecordAdmissionCommit {
316    outcome: RecordCommitted,
317    record: CommittedOrdinaryRecord,
318    connection_capacity: ConnectionConversationCapacityCommit,
319    projection: Box<ProjectedOrdinaryRecord>,
320}
321
322/// Exact successful record-admission parts for one atomic persistence commit.
323///
324/// The durable conversation writer consumes every field of this value in one
325/// atomic transaction. Each field is an owned authority moved out of the
326/// selected [`RecordAdmissionCommit`]: nothing here is cloned from, or leaves
327/// a second reachable copy inside, the consumed commit. [`ClaimFrontiers`]
328/// deliberately implements neither `Clone` nor `Copy`, so the complete
329/// resulting frontier authority exists exactly once.
330#[derive(Debug)]
331pub struct RecordAdmissionPersistenceParts {
332    /// Exact payload-bearing response.
333    pub outcome: RecordCommitted,
334    /// Exact payload-bearing durable caller record.
335    pub record: CommittedOrdinaryRecord,
336    /// Resulting semantic connection-capacity state.
337    pub connection_capacity: ConnectionConversationCapacityCommit,
338    /// Admitted caller-major allocation.
339    pub order: OrderAllocation,
340    /// Admitted caller/marker sequence allocation.
341    pub sequence: SequenceAdmission,
342    /// Shared observer-floor permit.
343    pub observer_floor: ObserverFloorPermit,
344    /// Shared remaining-closure permit.
345    pub closure: RemainingClosurePermit,
346    /// Complete resulting coupled claim frontiers.
347    pub frontiers: ClaimFrontiers,
348    /// Complete preferred/cap/resulting floor transition.
349    pub floor: crate::algebra::FloorComputation,
350    /// Exact physical retained occupancy.
351    pub retained_charge: crate::algebra::WideResourceVector,
352    /// Exact resulting closure baseline.
353    pub baseline: crate::algebra::WideResourceVector,
354    /// Exact resulting closure accounting.
355    pub accounting: ClosureAccounting,
356    /// Exact ordinary required-capacity envelope.
357    pub required_capacity: RequiredCapacityPlan,
358    /// Exact causal caller-row key and kind.
359    pub caller_record: super::super::RetainedCausalRecord,
360    /// Exact keyed caller-row charge.
361    pub caller_charge: RetainedRecordCharge,
362    /// One exact keyed charge per retained poststate row.
363    pub retained_charges: Vec<RetainedRecordCharge>,
364    /// Canonically ordered newly owed markers.
365    pub marker_candidates: Vec<super::super::MarkerCandidateAuthority>,
366}
367
368impl RecordAdmissionCommit {
369    /// Borrows the exact committed wire outcome.
370    #[must_use]
371    pub const fn outcome(&self) -> &RecordCommitted {
372        &self.outcome
373    }
374
375    /// Borrows the exact payload-bearing durable record.
376    #[must_use]
377    pub const fn record(&self) -> &CommittedOrdinaryRecord {
378        &self.record
379    }
380
381    /// Returns resulting semantic connection capacity.
382    #[must_use]
383    pub const fn connection_capacity(&self) -> ConnectionConversationCapacityCommit {
384        self.connection_capacity
385    }
386
387    /// Returns the admitted caller-major allocation.
388    #[must_use]
389    pub const fn order(&self) -> OrderAllocation {
390        self.projection.order()
391    }
392
393    /// Returns the admitted caller sequence and complete reserve.
394    #[must_use]
395    pub const fn sequence(&self) -> SequenceAdmission {
396        self.projection.sequence()
397    }
398
399    /// Returns the exact stage-11 observer permit.
400    #[must_use]
401    pub const fn observer_floor(&self) -> ObserverFloorPermit {
402        self.projection.observer_floor()
403    }
404
405    /// Returns the exact stage-12 closure permit.
406    #[must_use]
407    pub const fn closure(&self) -> &RemainingClosurePermit {
408        &self.projection.closure
409    }
410
411    /// Borrows the complete projected frontier/retention/accounting poststate.
412    #[must_use]
413    pub const fn projection(&self) -> &ProjectedOrdinaryRecord {
414        &self.projection
415    }
416
417    /// Transfers the exact successful persistence parts without cloning or
418    /// dropping any frontier, accounting, row, or marker authority.
419    #[must_use]
420    pub fn into_persistence_parts(self) -> RecordAdmissionPersistenceParts {
421        let ProjectedOrdinaryRecord {
422            frontiers,
423            floor,
424            retained_charge,
425            baseline,
426            accounting,
427            required_capacity,
428            order,
429            sequence,
430            observer_floor,
431            closure,
432            caller_record,
433            caller_charge,
434            retained_charges,
435            new_marker_candidates,
436        } = *self.projection;
437        RecordAdmissionPersistenceParts {
438            outcome: self.outcome,
439            record: self.record,
440            connection_capacity: self.connection_capacity,
441            order,
442            sequence,
443            observer_floor,
444            closure,
445            frontiers,
446            floor,
447            retained_charge,
448            baseline,
449            accounting,
450            required_capacity,
451            caller_record,
452            caller_charge,
453            retained_charges,
454            marker_candidates: new_marker_candidates,
455        }
456    }
457}
458
459/// Internal durable/configuration fault, distinct from every wire outcome.
460#[derive(Clone, Debug, PartialEq, Eq)]
461pub enum RecordAdmissionFault {
462    /// The consuming ordinary fixed point rejected inconsistent durable facts.
463    Projection(OrdinaryProjectionError),
464    /// Nonzero-debt precedence planning failed without wire exhaustion.
465    Order(OrderAdmissionError),
466    /// Nonzero-debt precedence planning failed without wire exhaustion.
467    Sequence(SequenceAdmissionError),
468    /// A capacity maximum could not be rebuilt through the shared selector.
469    RequiredCapacity(RequiredCapacityPlanError),
470    /// A fixed-point refusal failed to reproduce through its shared selector.
471    RefusalInvariant,
472}
473
474/// Internal fault paired with the unchanged replayable aggregate.
475#[derive(Debug)]
476pub struct RecordAdmissionFailure<'a, EF, V, LF> {
477    fault: RecordAdmissionFault,
478    unchanged: UnchangedRecordAdmission<'a, EF, V, LF>,
479}
480
481impl<'a, EF, V, LF> RecordAdmissionFailure<'a, EF, V, LF> {
482    /// Borrows the selected internal fault.
483    #[must_use]
484    pub const fn fault(&self) -> &RecordAdmissionFault {
485        &self.fault
486    }
487
488    /// Borrows the unchanged replayable aggregate.
489    #[must_use]
490    pub const fn unchanged(&self) -> &UnchangedRecordAdmission<'a, EF, V, LF> {
491        &self.unchanged
492    }
493
494    /// Recovers the fault and complete unchanged operation state.
495    #[must_use]
496    pub fn into_parts(
497        self,
498    ) -> (
499        RecordAdmissionFault,
500        UnchangedRecordAdmission<'a, EF, V, LF>,
501    ) {
502        (self.fault, self.unchanged)
503    }
504}
505
506/// Exhaustive ordinary-record operation result.
507#[derive(Debug)]
508pub enum RecordAdmissionDecision<'a, EF, V, LF> {
509    /// Exact lookup or admission response; no durable mutation is authorized.
510    Respond(Box<RecordAdmissionRefusal<'a, EF, V, LF>>),
511    /// A globally earlier immutable candidate must drain before retry.
512    DrainFirst(Box<RecordAdmissionDrainFirst<'a, EF, V, LF>>),
513    /// Every gate passed and all resulting state may commit atomically.
514    Commit(Box<RecordAdmissionCommit>),
515    /// Durable/configuration state violated an internal invariant.
516    Fault(Box<RecordAdmissionFailure<'a, EF, V, LF>>),
517}
518
519/// Classifies one ordinary record admission through the shared
520/// binding-required lookup WITHOUT consuming any claim-frontier authority.
521///
522/// This is the frontier-free prefix of [`apply_record_admission`]: exactly
523/// the frozen stage 2-5 rows (`Retired`, `ParticipantUnknown`,
524/// `StaleAuthority`, `NoBinding`) minted through the same sealed
525/// request-bound constructors the total selector uses. `None` means the
526/// presented authority is fully authorized and the operation must continue
527/// through the frontier-consuming total selector — a caller without that
528/// authority must fail closed rather than fabricate a later-stage outcome.
529#[must_use]
530pub fn classify_record_admission_binding<EF, V, LF>(
531    presented_identity: PresentedIdentity<'_, EF, V, LF>,
532    binding: &BindingState,
533    receiving_binding_epoch: BindingEpoch,
534    request: &RecordAdmission,
535) -> Option<RecordAdmissionResponse> {
536    let lookup_request = ParticipantBindingRequest::RecordAdmission(request.clone());
537    match lookup_binding_required(
538        presented_identity,
539        binding,
540        Some(receiving_binding_epoch),
541        &lookup_request,
542    ) {
543        BindingRequiredLookupResult::Retired(value) => {
544            Some(RecordAdmissionResponse::from_retired(value))
545        }
546        BindingRequiredLookupResult::ParticipantUnknown(value) => {
547            Some(RecordAdmissionResponse::from_participant_unknown(value))
548        }
549        BindingRequiredLookupResult::StaleAuthority(value) => {
550            Some(RecordAdmissionResponse::from_stale_authority(value))
551        }
552        BindingRequiredLookupResult::NoBinding(value) => {
553            Some(RecordAdmissionResponse::from_no_binding(value))
554        }
555        BindingRequiredLookupResult::Authorized { .. } => None,
556    }
557}
558
559/// Applies frozen stages 4-12 and constructs the exact phase-13 record commit.
560///
561/// Binding-required lookup and receiving-epoch validation precede semantic
562/// connection capacity and static size. A pre-owned candidate then drains
563/// before optional allocation. Order/sequence exhaustion precede observer and
564/// closure outcomes. Only the final commit owns changed frontiers or counters.
565#[must_use]
566#[allow(
567    clippy::too_many_lines,
568    reason = "the operation keeps the frozen total selector order visible in one function"
569)]
570pub fn apply_record_admission<EF, V, LF>(
571    input: RecordAdmissionPrestate<'_, EF, V, LF>,
572    encoded_record_charge: ResourceVector,
573) -> RecordAdmissionDecision<'_, EF, V, LF> {
574    let envelope = record_envelope(&input.request);
575
576    let lookup_request = ParticipantBindingRequest::RecordAdmission(input.request.clone());
577    match lookup_binding_required(
578        input.presented_identity,
579        input.binding,
580        Some(input.receiving_binding_epoch),
581        &lookup_request,
582    ) {
583        BindingRequiredLookupResult::Retired(value) => {
584            return refused(
585                input,
586                encoded_record_charge,
587                RecordAdmissionResponse::from_retired(value),
588            );
589        }
590        BindingRequiredLookupResult::ParticipantUnknown(value) => {
591            return refused(
592                input,
593                encoded_record_charge,
594                RecordAdmissionResponse::from_participant_unknown(value),
595            );
596        }
597        BindingRequiredLookupResult::StaleAuthority(value) => {
598            return refused(
599                input,
600                encoded_record_charge,
601                RecordAdmissionResponse::from_stale_authority(value),
602            );
603        }
604        BindingRequiredLookupResult::NoBinding(value) => {
605            return refused(
606                input,
607                encoded_record_charge,
608                RecordAdmissionResponse::from_no_binding(value),
609            );
610        }
611        BindingRequiredLookupResult::Authorized { .. } => {}
612    }
613
614    let connection_capacity = match select_semantic_connection_capacity(
615        input.connection_tracking,
616        input.connection_capacity,
617    ) {
618        SemanticConnectionCapacityDecision::Commit(value) => value,
619        SemanticConnectionCapacityDecision::Respond { limit } => {
620            let response =
621                RecordAdmissionResponse::connection_conversation_capacity_exceeded(envelope, limit);
622            return refused(input, encoded_record_charge, response);
623        }
624    };
625
626    let size = match check_record_size(
627        envelope.clone(),
628        encoded_record_charge,
629        input.max_ordinary_record_charge,
630    ) {
631        super::super::RecordSizeDecision::Eligible(value) => value,
632        super::super::RecordSizeDecision::Respond(value) => {
633            return refused(
634                input,
635                encoded_record_charge,
636                RecordAdmissionResponse::record_too_large(value),
637            );
638        }
639    };
640
641    if input.frontiers.sequence().immutable_candidates().is_empty()
642        && !matches!(input.closure_accounting.state(), ClosureState::Clear)
643    {
644        return match nonzero_debt_response(
645            &envelope,
646            &input.frontiers,
647            input.closure_accounting,
648            input.observer_progress,
649            input.projection_limits,
650        ) {
651            Ok(response) => refused(input, encoded_record_charge, response),
652            Err(operation_fault) => fault(input, encoded_record_charge, operation_fault),
653        };
654    }
655
656    let RecordAdmissionPrestate {
657        request,
658        presented_identity,
659        binding,
660        receiving_binding_epoch,
661        connection_tracking,
662        connection_capacity: original_connection_capacity,
663        closure_accounting,
664        max_ordinary_record_charge,
665        frontiers,
666        retained_charges,
667        observer_progress,
668        projection_limits,
669    } = input;
670    let shell = RecordAdmissionProjectionShell {
671        request,
672        presented_identity,
673        binding,
674        connection_tracking,
675        connection_capacity: original_connection_capacity,
676        max_ordinary_record_charge,
677    };
678    let projection_input = OrdinaryRecordProjectionInput::new(
679        envelope.clone(),
680        receiving_binding_epoch,
681        size.encoded_record_charge(),
682        retained_charges,
683        observer_progress,
684        closure_accounting,
685        projection_limits,
686    );
687    let projected = match frontiers.project_ordinary_record(projection_input) {
688        Ok(OrdinaryRecordProjectionDecision::DrainFirst(value)) => {
689            let candidate = value.candidate();
690            let (frontiers, projection_input) = value.into_unchanged_parts();
691            let unchanged = UnchangedRecordAdmission::new(
692                shell.rebuild(frontiers, projection_input),
693                encoded_record_charge,
694            );
695            return RecordAdmissionDecision::DrainFirst(Box::new(RecordAdmissionDrainFirst {
696                candidate,
697                unchanged,
698            }));
699        }
700        Ok(OrdinaryRecordProjectionDecision::Projected(value)) => value,
701        Err(failure) => {
702            let (frontiers, projection_input, error) = failure.into_parts();
703            let prestate = shell.rebuild(frontiers, projection_input);
704            return match projection_failure(error, &envelope, closure_accounting) {
705                Ok(response) => refused(prestate, encoded_record_charge, response),
706                Err(operation_fault) => fault(prestate, encoded_record_charge, operation_fault),
707            };
708        }
709    };
710
711    let order = projected.order();
712    let sequence = projected.sequence();
713    let delivery_seq = sequence.resulting().high_watermark();
714    let record = CommittedOrdinaryRecord::new(
715        shell.request,
716        order.major(),
717        delivery_seq,
718        size.encoded_record_charge(),
719    );
720    RecordAdmissionDecision::Commit(Box::new(RecordAdmissionCommit {
721        outcome: RecordCommitted::new(envelope, delivery_seq),
722        record,
723        connection_capacity,
724        projection: projected,
725    }))
726}
727
728struct RecordAdmissionProjectionShell<'a, EF, V, LF> {
729    request: RecordAdmission,
730    presented_identity: PresentedIdentity<'a, EF, V, LF>,
731    binding: &'a BindingState,
732    connection_tracking: ConnectionConversationTracking,
733    connection_capacity: CapacityCounter,
734    max_ordinary_record_charge: ResourceVector,
735}
736
737impl<'a, EF, V, LF> RecordAdmissionProjectionShell<'a, EF, V, LF> {
738    fn rebuild(
739        self,
740        frontiers: ClaimFrontiers,
741        projection: OrdinaryRecordProjectionInput,
742    ) -> RecordAdmissionPrestate<'a, EF, V, LF> {
743        let (
744            _envelope,
745            receiving_binding_epoch,
746            _encoded_record_charge,
747            retained_charges,
748            observer_progress,
749            closure_accounting,
750            projection_limits,
751        ) = projection.into_parts();
752        RecordAdmissionPrestate {
753            request: self.request,
754            presented_identity: self.presented_identity,
755            binding: self.binding,
756            receiving_binding_epoch,
757            connection_tracking: self.connection_tracking,
758            connection_capacity: self.connection_capacity,
759            closure_accounting,
760            max_ordinary_record_charge: self.max_ordinary_record_charge,
761            frontiers,
762            retained_charges,
763            observer_progress,
764            projection_limits,
765        }
766    }
767}
768
769fn refused<EF, V, LF>(
770    prestate: RecordAdmissionPrestate<'_, EF, V, LF>,
771    encoded_record_charge: ResourceVector,
772    response: RecordAdmissionResponse,
773) -> RecordAdmissionDecision<'_, EF, V, LF> {
774    RecordAdmissionDecision::Respond(Box::new(RecordAdmissionRefusal {
775        response,
776        unchanged: UnchangedRecordAdmission::new(prestate, encoded_record_charge),
777    }))
778}
779
780fn fault<EF, V, LF>(
781    prestate: RecordAdmissionPrestate<'_, EF, V, LF>,
782    encoded_record_charge: ResourceVector,
783    operation_fault: RecordAdmissionFault,
784) -> RecordAdmissionDecision<'_, EF, V, LF> {
785    RecordAdmissionDecision::Fault(Box::new(RecordAdmissionFailure {
786        fault: operation_fault,
787        unchanged: UnchangedRecordAdmission::new(prestate, encoded_record_charge),
788    }))
789}
790
791fn nonzero_debt_response(
792    envelope: &RecordAdmissionEnvelope,
793    frontiers: &ClaimFrontiers,
794    accounting: ClosureAccounting,
795    observer_progress: DeliverySeq,
796    limits: OrdinaryProjectionLimits,
797) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
798    let order = match allocate_order(
799        OrderAllocatingEnvelope::RecordAdmission(envelope.clone()),
800        frontiers.order().ledger(),
801        frontiers.order().ledger().plan_ordinary_record(),
802    ) {
803        Ok(value) => value,
804        Err(error) => return order_failure(error),
805    };
806    let sequence_plan = match frontiers.sequence().ledger().plan_ordinary_record(0) {
807        Ok(value) => value,
808        Err(error) => return sequence_failure(error),
809    };
810    if let Err(error) = admit_sequence(
811        SequenceAllocatingEnvelope::RecordAdmission(envelope.clone()),
812        sequence_plan,
813    ) {
814        return sequence_failure(error);
815    }
816    match check_observer_floor(
817        ObserverCheckedOperation::RecordAdmission(envelope.clone()),
818        observer_progress,
819        frontiers.retained_floor(),
820    ) {
821        ObserverFloorDecision::Eligible(_) => {}
822        ObserverFloorDecision::Respond(value) => {
823            return Ok(RecordAdmissionResponse::from_observer_backpressure(value));
824        }
825    }
826    let required = match RequiredCapacityPlan::ordinary(
827        accounting.baseline(),
828        limits.mandatory_bound(),
829        accounting.edge_k_remaining(),
830    ) {
831        Ok(value) => value,
832        Err(error) => {
833            return Err(RecordAdmissionFault::RequiredCapacity(error));
834        }
835    };
836    let delivered_marker_awaiting_ack = matches!(
837        accounting.state(),
838        ClosureState::Owed {
839            edge: StoredEdge::ParticipantCursorProgress(progress),
840            ..
841        } if progress.marker_delivery_seq().is_some()
842    );
843    match check_remaining_closure(
844        &ClosureCheckedEnvelope::RecordAdmission(envelope.clone()),
845        accounting,
846        delivered_marker_awaiting_ack,
847        0,
848        required,
849    ) {
850        RemainingClosureDecision::Respond(value) => {
851            Ok(RecordAdmissionResponse::from_marker_closure_capacity_exceeded(value))
852        }
853        RemainingClosureDecision::Eligible(_) => {
854            let _ = order;
855            Err(RecordAdmissionFault::RefusalInvariant)
856        }
857    }
858}
859
860fn projection_failure(
861    error: OrdinaryProjectionError,
862    envelope: &RecordAdmissionEnvelope,
863    accounting: ClosureAccounting,
864) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
865    match error {
866        OrdinaryProjectionError::Order(error) => order_failure(error),
867        OrdinaryProjectionError::Sequence(error) => sequence_failure(error),
868        OrdinaryProjectionError::ObserverBackpressure {
869            cap_floor,
870            observer_progress,
871        } => match check_observer_floor(
872            ObserverCheckedOperation::RecordAdmission(envelope.clone()),
873            observer_progress,
874            cap_floor,
875        ) {
876            ObserverFloorDecision::Respond(value) => {
877                Ok(RecordAdmissionResponse::from_observer_backpressure(value))
878            }
879            ObserverFloorDecision::Eligible(_) => Err(RecordAdmissionFault::Projection(
880                OrdinaryProjectionError::ObserverBackpressure {
881                    cap_floor,
882                    observer_progress,
883                },
884            )),
885        },
886        OrdinaryProjectionError::Capacity { required, .. }
887        | OrdinaryProjectionError::MarkerAnchorCapacity { required, .. } => {
888            capacity_failure(required, envelope, accounting)
889        }
890        other => Err(RecordAdmissionFault::Projection(other)),
891    }
892}
893
894fn capacity_failure(
895    required: crate::algebra::WideResourceVector,
896    envelope: &RecordAdmissionEnvelope,
897    accounting: ClosureAccounting,
898) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
899    let required_capacity = match RequiredCapacityPlan::from_successors(&[required]) {
900        Ok(value) => value,
901        Err(error) => {
902            return Err(RecordAdmissionFault::RequiredCapacity(error));
903        }
904    };
905    match check_remaining_closure(
906        &ClosureCheckedEnvelope::RecordAdmission(envelope.clone()),
907        accounting,
908        false,
909        0,
910        required_capacity,
911    ) {
912        RemainingClosureDecision::Respond(value) => {
913            Ok(RecordAdmissionResponse::from_marker_closure_capacity_exceeded(value))
914        }
915        RemainingClosureDecision::Eligible(_) => Err(RecordAdmissionFault::RefusalInvariant),
916    }
917}
918
919fn order_failure(
920    error: OrderAdmissionError,
921) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
922    match error {
923        OrderAdmissionError::Exhausted(value) => Ok(
924            RecordAdmissionResponse::from_conversation_order_exhausted(value),
925        ),
926        other => Err(RecordAdmissionFault::Order(other)),
927    }
928}
929
930fn sequence_failure(
931    error: SequenceAdmissionError,
932) -> Result<RecordAdmissionResponse, RecordAdmissionFault> {
933    match error {
934        SequenceAdmissionError::Exhausted(value) => {
935            Ok(RecordAdmissionResponse::from_conversation_sequence_exhausted(value))
936        }
937        other => Err(RecordAdmissionFault::Sequence(other)),
938    }
939}
940
941const fn record_envelope(request: &RecordAdmission) -> RecordAdmissionEnvelope {
942    RecordAdmissionEnvelope {
943        conversation_id: request.conversation_id,
944        participant_id: request.participant_id,
945        capability_generation: request.capability_generation,
946        record_admission_attempt_token: request.record_admission_attempt_token,
947    }
948}
liminal_protocol/lifecycle/operations/record_admission.rs

liminal_protocol/lifecycle/operations/
record_admission.rs
