liminal_protocol/lifecycle/operations/
marker_drain.rs

1//! Consuming mandatory marker-candidate drain.
2//!
3//! The operation owns the only fresh `MarkerDelivery` producer. It consumes the
4//! coupled claim frontiers, delegates the exact H/M/order-key transition to the
5//! sealed frontier core, and returns one opaque atomic commit containing the
6//! resulting frontiers, exact current closure state, planned marker successor,
7//! and retained-record state.
8
9use alloc::vec::Vec;
10
11use crate::wire::{ParticipantDelivery, ParticipantRecord};
12
13use super::super::{
14    ClaimFrontiers, ClosureAccounting, ClosureState, Event, MarkerDelivery, ObserverProjection,
15    StoredEdge,
16    claim_frontier::{MarkerDrainCoreError, ValidatedMarkerRecord},
17};
18use super::RetainedRecordCharge;
19
20/// Exact invariant fault selected by mandatory marker drain.
21#[derive(Clone, Copy, Debug, PartialEq, Eq)]
22pub enum MarkerDrainError {
23    /// No immutable candidate is currently owed.
24    NoCandidate,
25    /// A binding terminal has global precedence over marker work.
26    BindingTerminalFirst,
27    /// The first marker does not own exactly `H + 1`.
28    SequenceNotNext,
29    /// The current closure edge cannot coexist with this marker append.
30    CurrentEdgeMismatch,
31    /// Marker drain attempted to allocate rather than reuse its causal major.
32    CausalMajorNotAllocated,
33    /// Cross-counter validation promised an order key that is now absent.
34    MissingOrderCandidate,
35    /// Consuming `M` did not yield a valid post-append sequence ledger.
36    ResultingLedger,
37    /// Frontier core returned candidate and retained-record authorities that disagree.
38    AuthorityMismatch,
39    /// Canonical marker-row charge does not name the selected retained row.
40    MarkerChargeKey,
41    /// Every retained durable row has exactly one entry of charge.
42    MarkerEntryCharge,
43    /// The successor closure accounting failed validation.
44    ResultingAccounting,
45}
46
47/// Move-only typed wire projection of one protocol-selected compaction marker.
48///
49/// Construction remains private to [`drain_next_marker`]. The projection is
50/// inseparable from its [`MarkerDrainCommit`] until the commit is consumed, so
51/// storage cannot assemble a delivery from unrelated raw marker fields.
52///
53/// Raw fields cannot be assembled outside the protocol selector:
54///
55/// ```compile_fail
56/// use liminal_protocol::lifecycle::MarkerDeliveryProjection;
57/// use liminal_protocol::wire::ParticipantDelivery;
58///
59/// fn fabricate(delivery: ParticipantDelivery) -> MarkerDeliveryProjection {
60///     MarkerDeliveryProjection { delivery }
61/// }
62/// ```
63///
64/// `Debug` output is not a parser or an alternate construction path:
65///
66/// ```compile_fail
67/// use liminal_protocol::lifecycle::MarkerDeliveryProjection;
68///
69/// let _: MarkerDeliveryProjection = "MarkerDeliveryProjection { ... }".parse().unwrap();
70/// ```
71#[derive(Debug, PartialEq, Eq)]
72pub struct MarkerDeliveryProjection {
73    delivery: ParticipantDelivery,
74}
75
76impl MarkerDeliveryProjection {
77    /// Borrows the exact typed participant delivery selected by marker drain.
78    #[must_use]
79    pub const fn delivery(&self) -> &ParticipantDelivery {
80        &self.delivery
81    }
82
83    /// Consumes the projection into the wire delivery owned by outbox production.
84    #[must_use]
85    pub fn into_delivery(self) -> ParticipantDelivery {
86        self.delivery
87    }
88}
89
90/// Complete atomic marker-drain commit.
91///
92/// No field can be supplied independently: the claim-frontier transition,
93/// executable successor and retained marker record share one sealed predecessor
94/// and must be persisted together. The retained-record authority is never
95/// exposed independently of the selected successor: doing so would let an
96/// undelivered DMR append masquerade as delivered DCR recovery provenance.
97///
98/// ```compile_fail
99/// use liminal_protocol::lifecycle::MarkerDrainCommit;
100///
101/// fn splice(commit: MarkerDrainCommit) {
102///     let _ = commit.record();
103/// }
104/// ```
105#[derive(Debug, PartialEq, Eq)]
106pub struct MarkerDrainCommit {
107    frontiers: ClaimFrontiers,
108    closure_accounting: ClosureAccounting,
109    retained_charges: Vec<RetainedRecordCharge>,
110    marker_successor: StoredEdge,
111    delivery_projection: MarkerDeliveryProjection,
112    record: ValidatedMarkerRecord,
113}
114
115impl MarkerDrainCommit {
116    /// Borrows the resulting coupled sequence/order frontiers.
117    #[must_use]
118    pub const fn frontiers(&self) -> &ClaimFrontiers {
119        &self.frontiers
120    }
121
122    /// Returns the exact closure state after the append occurrence.
123    #[must_use]
124    pub const fn closure(&self) -> ClosureState {
125        self.closure_accounting.state()
126    }
127
128    /// Returns the complete closure accounting after the append occurrence.
129    #[must_use]
130    pub const fn closure_accounting(&self) -> ClosureAccounting {
131        self.closure_accounting
132    }
133
134    /// Borrows exact keyed charges for the complete retained suffix.
135    #[must_use]
136    pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
137        &self.retained_charges
138    }
139
140    /// Returns the exact marker edge selected once any strict OP/PC completes.
141    ///
142    /// A bound target selects [`StoredEdge::MarkerDelivery`]; a target whose
143    /// epoch already died selects [`StoredEdge::DetachedMarkerRelease`].
144    #[must_use]
145    pub const fn marker_successor(&self) -> StoredEdge {
146        self.marker_successor
147    }
148
149    /// Borrows the typed `HistoryCompacted` projection coupled to this commit.
150    #[must_use]
151    pub const fn delivery_projection(&self) -> &MarkerDeliveryProjection {
152        &self.delivery_projection
153    }
154
155    /// Consumes the opaque transaction into its persistable protocol values.
156    ///
157    /// The retained marker is already present in the returned frontiers. Its
158    /// executable validation token remains coupled to this commit and is
159    /// deliberately consumed here rather than returned as a fourth value.
160    #[must_use]
161    pub fn into_parts(
162        self,
163    ) -> (
164        ClaimFrontiers,
165        ClosureAccounting,
166        Vec<RetainedRecordCharge>,
167        StoredEdge,
168        MarkerDeliveryProjection,
169    ) {
170        self.record.consume();
171        (
172            self.frontiers,
173            self.closure_accounting,
174            self.retained_charges,
175            self.marker_successor,
176            self.delivery_projection,
177        )
178    }
179
180    /// Extracts the occurrence token for crate-internal adversarial tests.
181    #[cfg(test)]
182    pub(super) fn into_record_for_test(self) -> ValidatedMarkerRecord {
183        self.record
184    }
185}
186
187/// Consumes the globally first marker candidate into one atomic durable commit.
188///
189/// This operation never accepts a raw candidate token, sequence, participant,
190/// epoch, marker count, or order key. All are derived from the coupled
191/// frontiers, and the candidate authority is consumed while deriving the exact
192/// bound-delivery or detached-release successor. An already-current OP/PC
193/// remains strict through its typed marker-append transition.
194///
195/// # Errors
196///
197/// Returns [`MarkerDrainError`] when the mandatory prefix is absent, belongs to
198/// a binding terminal, is not the exact next sequence, targets a detached
199/// binding, conflicts with the current closure edge, lacks its
200/// already-allocated causal order key, or cannot produce a valid resulting
201/// sequence ledger.
202pub fn drain_next_marker(
203    frontiers: ClaimFrontiers,
204    current_accounting: ClosureAccounting,
205    mut retained_charges: Vec<RetainedRecordCharge>,
206    marker_charge: RetainedRecordCharge,
207) -> Result<MarkerDrainCommit, MarkerDrainError> {
208    let core = frontiers.drain_next_marker_core().map_err(map_core_error)?;
209    let (frontiers, candidate, record) = core.into_parts();
210    let candidate_conversation_id = candidate.conversation_id();
211    let candidate_participant = candidate.participant_id();
212    let candidate_sequence = candidate.delivery_seq();
213    let candidate_target = candidate.target_binding();
214    let candidate_provenance = candidate.provenance();
215    let delivery_projection = MarkerDeliveryProjection {
216        delivery: ParticipantDelivery {
217            conversation_id: candidate_conversation_id,
218            delivery_seq: candidate_sequence,
219            record: ParticipantRecord::HistoryCompacted {
220                affected_participant_id: candidate_participant,
221                abandoned_after: candidate.abandoned_after(),
222                abandoned_through: candidate.abandoned_through(),
223                physical_floor_at_decision: candidate.physical_floor_at_decision(),
224            },
225        },
226    };
227    let marker_successor = MarkerDelivery::successor_from_validated_candidate(candidate);
228    if record.conversation_id() != candidate_conversation_id
229        || record.participant_id() != candidate_participant
230        || record.delivery_seq() != candidate_sequence
231        || record.target_binding() != candidate_target
232        || record.provenance() != candidate_provenance
233    {
234        return Err(MarkerDrainError::AuthorityMismatch);
235    }
236    if marker_charge.delivery_seq() != record.delivery_seq()
237        || marker_charge.admission_order() != record.admission_order()
238    {
239        return Err(MarkerDrainError::MarkerChargeKey);
240    }
241    if marker_charge.encoded_charge().entries != 1 {
242        return Err(MarkerDrainError::MarkerEntryCharge);
243    }
244    let closure = apply_marker_append(current_accounting.state(), candidate_sequence)?;
245    let closure_accounting = ClosureAccounting::try_new(
246        closure,
247        current_accounting.marker_capacity_credits(),
248        current_accounting.marker_anchors(),
249        current_accounting.edge_sequence_claims(),
250        current_accounting.edge_order_position_claims(),
251        current_accounting.edge_k_remaining(),
252        current_accounting.baseline(),
253        current_accounting.configured_cap(),
254        current_accounting.episode_churn_used(),
255        current_accounting.episode_churn_limit(),
256    )
257    .map_err(|_| MarkerDrainError::ResultingAccounting)?;
258    retained_charges.push(marker_charge);
259    if retained_charges.len() != frontiers.retained_records().len()
260        || !retained_charges
261            .iter()
262            .zip(frontiers.retained_records())
263            .all(|(charge, retained)| {
264                charge.delivery_seq() == retained.delivery_seq
265                    && charge.admission_order() == retained.admission_order
266            })
267    {
268        return Err(MarkerDrainError::MarkerChargeKey);
269    }
270    Ok(MarkerDrainCommit {
271        frontiers,
272        closure_accounting,
273        retained_charges,
274        marker_successor,
275        delivery_projection,
276        record,
277    })
278}
279
280fn apply_marker_append(
281    current: ClosureState,
282    marker_delivery_seq: u64,
283) -> Result<ClosureState, MarkerDrainError> {
284    let event = Event::marker_appended(marker_delivery_seq, marker_delivery_seq);
285    match current {
286        ClosureState::Clear => Ok(ClosureState::Clear),
287        ClosureState::Owed {
288            debt,
289            edge: StoredEdge::ObserverProjection(projection),
290        } => {
291            let successor = projection
292                .later_projection_after_marker(
293                    &event,
294                    debt,
295                    ObserverProjection::new(marker_delivery_seq),
296                )
297                .ok_or(MarkerDrainError::CurrentEdgeMismatch)?;
298            projection
299                .marker_appended(debt, event, successor)
300                .map_err(|_| MarkerDrainError::CurrentEdgeMismatch)
301        }
302        ClosureState::Owed {
303            debt,
304            edge: StoredEdge::PhysicalCompaction(compaction),
305        } => compaction
306            .marker_appended(debt, event)
307            .map_err(|_| MarkerDrainError::CurrentEdgeMismatch),
308        ClosureState::Owed { .. } => Err(MarkerDrainError::CurrentEdgeMismatch),
309    }
310}
311
312const fn map_core_error(error: MarkerDrainCoreError) -> MarkerDrainError {
313    match error {
314        MarkerDrainCoreError::NoCandidate => MarkerDrainError::NoCandidate,
315        MarkerDrainCoreError::BindingTerminalFirst => MarkerDrainError::BindingTerminalFirst,
316        MarkerDrainCoreError::SequenceNotNext => MarkerDrainError::SequenceNotNext,
317        MarkerDrainCoreError::CausalMajorNotAllocated => MarkerDrainError::CausalMajorNotAllocated,
318        MarkerDrainCoreError::MissingOrderCandidate => MarkerDrainError::MissingOrderCandidate,
319        MarkerDrainCoreError::ResultingLedger => MarkerDrainError::ResultingLedger,
320    }
321}
liminal_protocol/lifecycle/operations/marker_drain.rs

liminal_protocol/lifecycle/operations/
marker_drain.rs
