liminal_protocol/lifecycle/operations/
binding_terminal.rs

1use alloc::boxed::Box;
2
3use crate::{
4    algebra::ResourceVector,
5    lifecycle::{ActiveBinding, AdmissionOrder},
6    wire::{BindingEpoch, ConversationId, DeliverySeq, ParticipantId, TransactionOrder},
7};
8
9use super::{LiveFrontierError, LiveFrontierOwner, RetainedRecordCharge};
10use crate::lifecycle::{CommittedBindingTerminalPosition, PendingBindingTerminalPosition};
11
12/// Canonical durability encoding required for one binding-terminal candidate.
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub enum BindingTerminalEncoding {
15    /// Canonical JSON encoding of one schema-v3 participant lifecycle row.
16    ParticipantLifecycleV3CanonicalJson,
17}
18
19/// Closed terminal row class selected before canonical encoding.
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
21pub enum BindingTerminalCauseClass {
22    /// A clean or orderly close produces `Detached`.
23    Detached,
24    /// An unexpected loss or refusal produces `Died`.
25    Died,
26}
27
28/// Sealed exact identity exposed to the canonical server encoder.
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
30pub struct CandidateTerminalKey {
31    active_binding: ActiveBinding,
32    cause_class: BindingTerminalCauseClass,
33    admission_order: AdmissionOrder,
34    delivery_seq: DeliverySeq,
35}
36
37impl CandidateTerminalKey {
38    /// Returns the owning conversation.
39    #[must_use]
40    pub const fn conversation_id(self) -> ConversationId {
41        self.active_binding.conversation_id
42    }
43
44    /// Returns the permanent participant identifier.
45    #[must_use]
46    pub const fn participant_id(self) -> ParticipantId {
47        self.active_binding.participant_id
48    }
49
50    /// Returns the exact ended binding epoch.
51    #[must_use]
52    pub const fn binding_epoch(self) -> BindingEpoch {
53        self.active_binding.binding_epoch
54    }
55
56    /// Returns the closed terminal row class.
57    #[must_use]
58    pub const fn cause_class(self) -> BindingTerminalCauseClass {
59        self.cause_class
60    }
61
62    /// Returns the canonical binding-terminal admission key.
63    #[must_use]
64    pub const fn admission_order(self) -> AdmissionOrder {
65        self.admission_order
66    }
67
68    /// Returns the candidate delivery sequence.
69    #[must_use]
70    pub const fn delivery_seq(self) -> DeliverySeq {
71        self.delivery_seq
72    }
73
74    /// Keys one canonical schema-v3 encoded charge to this sealed candidate.
75    #[must_use]
76    pub const fn bind_v3_charge(
77        self,
78        encoded_charge: ResourceVector,
79    ) -> BindingTerminalCandidateCharge {
80        BindingTerminalCandidateCharge {
81            conversation_id: self.conversation_id(),
82            participant_id: self.participant_id(),
83            binding_epoch: self.binding_epoch(),
84            admission_order: self.admission_order,
85            delivery_seq: self.delivery_seq,
86            encoding: BindingTerminalEncoding::ParticipantLifecycleV3CanonicalJson,
87            charge: RetainedRecordCharge::new(
88                self.delivery_seq,
89                self.admission_order,
90                encoded_charge,
91            ),
92        }
93    }
94}
95
96/// Canonical schema-v3 charge keyed to one sealed binding-terminal candidate.
97#[derive(Clone, Copy, Debug, PartialEq, Eq)]
98pub struct BindingTerminalCandidateCharge {
99    conversation_id: ConversationId,
100    participant_id: ParticipantId,
101    binding_epoch: BindingEpoch,
102    admission_order: AdmissionOrder,
103    delivery_seq: DeliverySeq,
104    encoding: BindingTerminalEncoding,
105    charge: RetainedRecordCharge,
106}
107
108impl BindingTerminalCandidateCharge {
109    /// Returns the exact retained-row charge after successful admission.
110    #[must_use]
111    pub const fn retained_charge(self) -> RetainedRecordCharge {
112        self.charge
113    }
114}
115
116/// Typed reason the non-mutating prepare stage refused authority or positions.
117#[derive(Clone, Copy, Debug, PartialEq, Eq)]
118pub enum BindingTerminalPrepareError {
119    /// The active binding does not belong to this coupled owner.
120    Authority,
121    /// The supplied transaction order is not the next checked order.
122    TransactionOrder,
123    /// The supplied delivery sequence is not the next checked sequence.
124    DeliverySequence,
125    /// Hard observer progress exceeds the current sequence high watermark.
126    ObserverProgress,
127}
128
129/// Prepare refusal preserving the unchanged complete owner.
130#[derive(Debug, PartialEq, Eq)]
131pub struct BindingTerminalPrepareRefused {
132    owner: LiveFrontierOwner,
133    error: BindingTerminalPrepareError,
134}
135
136impl BindingTerminalPrepareRefused {
137    /// Returns the typed refusal reason.
138    #[must_use]
139    pub const fn error(&self) -> BindingTerminalPrepareError {
140        self.error
141    }
142
143    /// Recovers the unchanged coupled owner.
144    #[must_use]
145    pub fn into_owner(self) -> LiveFrontierOwner {
146        self.owner
147    }
148}
149
150/// Non-mutating prepared terminal admission with its sealed canonical key.
151#[derive(Debug, PartialEq, Eq)]
152pub struct PreparedBindingTerminal {
153    owner: LiveFrontierOwner,
154    key: CandidateTerminalKey,
155    hard_observer_progress: DeliverySeq,
156}
157
158impl PreparedBindingTerminal {
159    /// Returns the sealed key required by the canonical v3 encoder.
160    #[must_use]
161    pub const fn candidate_key(&self) -> CandidateTerminalKey {
162        self.key
163    }
164
165    /// Returns the hard-observer baseline captured by non-mutating preparation.
166    #[must_use]
167    pub const fn hard_observer_progress(&self) -> DeliverySeq {
168        self.hard_observer_progress
169    }
170
171    /// Admits only the exact keyed v3 charge produced for this candidate.
172    #[must_use]
173    pub fn admit(self, candidate: BindingTerminalCandidateCharge) -> BindingTerminalAdmission {
174        if candidate.conversation_id != self.key.conversation_id()
175            || candidate.participant_id != self.key.participant_id()
176            || candidate.binding_epoch != self.key.binding_epoch()
177            || candidate.admission_order != self.key.admission_order()
178            || candidate.delivery_seq != self.key.delivery_seq()
179            || candidate.encoding != BindingTerminalEncoding::ParticipantLifecycleV3CanonicalJson
180            || candidate.charge.delivery_seq() != self.key.delivery_seq()
181            || candidate.charge.admission_order() != self.key.admission_order()
182            || candidate.charge.encoded_charge().entries != 1
183        {
184            return admit_refusal(self.owner, BindingTerminalAdmitError::CandidateCharge);
185        }
186
187        let retained_len = u64::try_from(self.owner.retained_charges().len());
188        let has_capacity = retained_len
189            .ok()
190            .and_then(|len| len.checked_add(1))
191            .is_some_and(|len| len <= self.owner.retained_record_limit());
192        if has_capacity {
193            return match self.owner.commit_binding_terminal_candidate(
194                self.key.active_binding,
195                self.key.admission_order,
196                self.key.delivery_seq,
197                candidate.charge,
198            ) {
199                Ok(owner) => BindingTerminalAdmission::Commit(BindingTerminalCommit {
200                    owner,
201                    position: CommittedBindingTerminalPosition::new(
202                        self.key.admission_order.transaction_order(),
203                        self.key.delivery_seq,
204                    ),
205                }),
206                Err(failure) => {
207                    let (owner, error) = *failure;
208                    admit_refusal(owner, map_live_frontier_error(error))
209                }
210            };
211        }
212        if self.hard_observer_progress < self.key.delivery_seq {
213            return match self.owner.pend_binding_terminal_candidate(
214                self.key.active_binding,
215                self.key.admission_order,
216                self.key.delivery_seq,
217            ) {
218                Ok(owner) => BindingTerminalAdmission::Pending(BindingTerminalPending {
219                    owner,
220                    position: PendingBindingTerminalPosition::new(
221                        self.key.admission_order.transaction_order(),
222                    ),
223                    blocked_at_observer: self.hard_observer_progress,
224                }),
225                Err(failure) => {
226                    let (owner, error) = *failure;
227                    admit_refusal(owner, map_live_frontier_error(error))
228                }
229            };
230        }
231        admit_refusal(self.owner, BindingTerminalAdmitError::RetainedRecordLimit)
232    }
233
234    /// Recovers the unchanged owner when canonical encoding cannot produce a charge.
235    #[must_use]
236    pub fn into_owner(self) -> LiveFrontierOwner {
237        self.owner
238    }
239}
240
241/// Successful committed terminal admission.
242#[derive(Debug, PartialEq, Eq)]
243pub struct BindingTerminalCommit {
244    owner: LiveFrontierOwner,
245    position: CommittedBindingTerminalPosition,
246}
247
248impl BindingTerminalCommit {
249    /// Consumes the admission into its transitioned owner and committed position.
250    #[must_use]
251    pub fn into_parts(self) -> (LiveFrontierOwner, CommittedBindingTerminalPosition) {
252        (self.owner, self.position)
253    }
254}
255
256/// Successful observer-blocked pending terminal admission.
257#[derive(Debug, PartialEq, Eq)]
258pub struct BindingTerminalPending {
259    owner: LiveFrontierOwner,
260    position: PendingBindingTerminalPosition,
261    blocked_at_observer: DeliverySeq,
262}
263
264impl BindingTerminalPending {
265    /// Returns the exact hard-observer baseline persisted with the pending source.
266    #[must_use]
267    pub const fn blocked_at_observer(&self) -> DeliverySeq {
268        self.blocked_at_observer
269    }
270
271    /// Consumes the admission into its transitioned owner and pending position.
272    #[must_use]
273    pub fn into_parts(self) -> (LiveFrontierOwner, PendingBindingTerminalPosition) {
274        (self.owner, self.position)
275    }
276}
277
278/// Typed reason the keyed candidate could not be admitted.
279#[derive(Clone, Copy, Debug, PartialEq, Eq)]
280pub enum BindingTerminalAdmitError {
281    /// Candidate identity, encoding, charge key, or entry count disagrees.
282    CandidateCharge,
283    /// The retained causal-row cap could not admit this candidate.
284    RetainedRecordLimit,
285    /// The selected binding no longer belongs to the coupled owner.
286    Authority,
287    /// An earlier immutable or recovery transition has precedence.
288    Precedence,
289    /// Claim arithmetic or exact owner reconstruction failed.
290    Frontier,
291    /// Resulting closure accounting exceeded its signed capacity.
292    ClosureAccounting,
293}
294
295/// Admit refusal preserving the unchanged coupled owner.
296#[derive(Debug, PartialEq, Eq)]
297pub struct BindingTerminalAdmitRefused {
298    owner: LiveFrontierOwner,
299    error: BindingTerminalAdmitError,
300}
301
302impl BindingTerminalAdmitRefused {
303    /// Returns the typed refusal reason.
304    #[must_use]
305    pub const fn error(&self) -> BindingTerminalAdmitError {
306        self.error
307    }
308
309    /// Recovers the unchanged owner.
310    #[must_use]
311    pub fn into_owner(self) -> LiveFrontierOwner {
312        self.owner
313    }
314}
315
316/// Exhaustive terminal-admission result.
317#[derive(Debug, PartialEq, Eq)]
318pub enum BindingTerminalAdmission {
319    /// The exact candidate committed and transformed the complete owner.
320    Commit(BindingTerminalCommit),
321    /// Observer progress blocked the candidate after consuming only its order.
322    Pending(BindingTerminalPending),
323    /// Candidate validation or a protocol invariant refused unchanged.
324    Refused(Box<BindingTerminalAdmitRefused>),
325}
326
327fn admit_refusal(
328    owner: LiveFrontierOwner,
329    error: BindingTerminalAdmitError,
330) -> BindingTerminalAdmission {
331    BindingTerminalAdmission::Refused(Box::new(BindingTerminalAdmitRefused { owner, error }))
332}
333
334const fn map_live_frontier_error(error: LiveFrontierError) -> BindingTerminalAdmitError {
335    match error {
336        LiveFrontierError::Authority => BindingTerminalAdmitError::Authority,
337        LiveFrontierError::Precedence => BindingTerminalAdmitError::Precedence,
338        LiveFrontierError::RetainedCharge => BindingTerminalAdmitError::CandidateCharge,
339        LiveFrontierError::RetainedRecordLimit => BindingTerminalAdmitError::RetainedRecordLimit,
340        LiveFrontierError::Frontier => BindingTerminalAdmitError::Frontier,
341        LiveFrontierError::ClosureAccounting => BindingTerminalAdmitError::ClosureAccounting,
342    }
343}
344
345impl LiveFrontierOwner {
346    /// Validates one active binding and checked candidate positions without mutation.
347    ///
348    /// # Errors
349    ///
350    /// Returns the unchanged owner when binding authority, checked order or sequence,
351    /// or hard-observer progress disagrees with the coupled protocol state.
352    pub fn prepare_binding_terminal(
353        self,
354        active_binding: ActiveBinding,
355        cause_class: BindingTerminalCauseClass,
356        next_transaction_order: TransactionOrder,
357        next_delivery_sequence: DeliverySeq,
358        hard_observer_progress: DeliverySeq,
359    ) -> Result<PreparedBindingTerminal, Box<BindingTerminalPrepareRefused>> {
360        let authority_matches = active_binding.conversation_id
361            == self.frontiers().conversation_id()
362            && self
363                .frontiers()
364                .active_identities()
365                .participants()
366                .iter()
367                .any(|participant| {
368                    participant.participant_index() == active_binding.participant_id
369                        && participant.binding()
370                            == crate::lifecycle::FrontierBinding::Bound(
371                                active_binding.binding_epoch,
372                            )
373                });
374        if !authority_matches {
375            return prepare_refusal(self, BindingTerminalPrepareError::Authority);
376        }
377        let expected_order = match self.frontiers().order().ledger().high() {
378            crate::lifecycle::OrderHigh::Empty => Some(0),
379            crate::lifecycle::OrderHigh::Allocated(high) => high.checked_add(1),
380        };
381        if expected_order != Some(next_transaction_order) {
382            return prepare_refusal(self, BindingTerminalPrepareError::TransactionOrder);
383        }
384        if self
385            .frontiers()
386            .sequence()
387            .ledger()
388            .high_watermark()
389            .checked_add(1)
390            != Some(next_delivery_sequence)
391        {
392            return prepare_refusal(self, BindingTerminalPrepareError::DeliverySequence);
393        }
394        if hard_observer_progress > self.frontiers().sequence().ledger().high_watermark() {
395            return prepare_refusal(self, BindingTerminalPrepareError::ObserverProgress);
396        }
397        Ok(PreparedBindingTerminal {
398            owner: self,
399            key: CandidateTerminalKey {
400                active_binding,
401                cause_class,
402                admission_order: AdmissionOrder::binding_terminal(
403                    next_transaction_order,
404                    active_binding.participant_id,
405                ),
406                delivery_seq: next_delivery_sequence,
407            },
408            hard_observer_progress,
409        })
410    }
411}
412
413fn prepare_refusal(
414    owner: LiveFrontierOwner,
415    error: BindingTerminalPrepareError,
416) -> Result<PreparedBindingTerminal, Box<BindingTerminalPrepareRefused>> {
417    Err(Box::new(BindingTerminalPrepareRefused { owner, error }))
418}
liminal_protocol/lifecycle/operations/binding_terminal.rs

liminal_protocol/lifecycle/operations/
binding_terminal.rs
