liminal_protocol/lifecycle/operations/
marker_ack.rs

1use alloc::boxed::Box;
2
3use crate::wire::{
4    BindingEpoch, ConversationId, DeliverySeq, Generation, MarkerAck, MarkerAckCommitted,
5    MarkerAckEnvelope, MarkerAckResponse, ParticipantId,
6};
7
8use super::{
9    super::{
10        BindingRequiredLookupResult, BindingState, LiveMember, ObserverProgressProjection,
11        ParticipantBindingRequest, PresentedIdentity, SealedBindingFateToken,
12        lookup_binding_required,
13        membership::{LiveMemberCursorUpdate, LiveMemberCursorUpdateError},
14    },
15    marker_proof::{
16        MarkerProofDecision, MarkerProofInput, MarkerProofPermit, MarkerProofState,
17        select_marker_proof,
18    },
19};
20
21/// Atomic zero-debt marker-ack commit.
22///
23/// The retained marker permit proves delivery to the exact authoritative
24/// binding epoch. The cursor update is opaque and can be applied only through
25/// [`Self::apply_to`]. Nonzero-debt edge completion is deliberately outside
26/// this operation.
27#[derive(Clone, Debug, PartialEq, Eq)]
28pub struct MarkerAckCommit {
29    outcome: MarkerAckCommitted,
30    proof: MarkerProofPermit,
31    cursor_update: LiveMemberCursorUpdate,
32}
33
34impl MarkerAckCommit {
35    /// Borrows the exact committed wire outcome.
36    #[must_use]
37    pub const fn outcome(&self) -> &MarkerAckCommitted {
38        &self.outcome
39    }
40
41    /// Projects the exact committed marker boundary into hard observer progress.
42    #[must_use]
43    pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
44        let request = self.outcome.request();
45        ObserverProgressProjection::new(request.conversation_id, request.marker_delivery_seq)
46    }
47
48    /// Returns the exact canonical request selected into this commit.
49    #[must_use]
50    pub const fn canonical_request(&self) -> MarkerAck {
51        let request = self.outcome.request();
52        MarkerAck {
53            conversation_id: request.conversation_id,
54            participant_id: request.participant_id,
55            capability_generation: request.capability_generation,
56            marker_delivery_seq: request.marker_delivery_seq,
57        }
58    }
59
60    /// Returns the receiving binding epoch authorized by shared lookup.
61    #[must_use]
62    pub const fn receiving_binding_epoch(&self) -> BindingEpoch {
63        self.proof.proof_binding_epoch()
64    }
65
66    /// Returns the exact marker sequence proven offered to the binding.
67    #[must_use]
68    pub const fn offered_marker_delivery_seq(&self) -> DeliverySeq {
69        self.proof.expected_marker_delivery_seq()
70    }
71
72    /// Returns the binding epoch on which the marker was proven offered.
73    #[must_use]
74    pub const fn delivered_binding_epoch(&self) -> BindingEpoch {
75        self.proof.proof_binding_epoch()
76    }
77
78    /// Returns the durable cursor prestate checked by this transition.
79    #[must_use]
80    pub const fn from_cursor(&self) -> DeliverySeq {
81        self.cursor_update.previous_cursor()
82    }
83
84    /// Returns the exact post-transition cursor for replay audit.
85    #[must_use]
86    pub const fn resulting_cursor(&self) -> DeliverySeq {
87        self.cursor_update.resulting_cursor()
88    }
89
90    /// Borrows the exact delivered-marker authority retained by this commit.
91    #[must_use]
92    pub const fn proof(&self) -> &MarkerProofPermit {
93        &self.proof
94    }
95
96    /// Replays this exact selected marker transition into one sealed fate token.
97    ///
98    /// The ordinary-ack path has always progressed the token in step with the
99    /// member cursor (`ParticipantAckCommit::progress_binding_fate_token`); the
100    /// marker path never did, so a marker-ack advanced the member and stranded
101    /// the token behind it, and the NEXT ordinary ack — live or rebuilt from
102    /// durable rows on boot — was refused against a frozen prestate.
103    ///
104    /// # Errors
105    ///
106    /// Returns the unchanged token when its conversation, participant, binding
107    /// epoch, or cursor prestate differs from this committed marker-ack.
108    /// Replaying a marker-ack the token has already consumed is NOT an error:
109    /// it returns the token untouched, matching [`Self::apply_to`]'s own
110    /// idempotence.
111    pub fn progress_binding_fate_token(
112        &self,
113        token: SealedBindingFateToken,
114    ) -> Result<SealedBindingFateToken, Box<SealedBindingFateToken>> {
115        let request = self.outcome.request();
116        token.marker_ack_progressed(
117            request.conversation_id,
118            request.participant_id,
119            self.receiving_binding_epoch(),
120            self.from_cursor(),
121            self.resulting_cursor(),
122        )
123    }
124
125    /// Applies this commit to either its exact old cursor or its already-written
126    /// resulting cursor.
127    ///
128    /// Replaying after a crash is idempotent: the old prestate advances once,
129    /// while the exact new prestate returns the same [`MarkerAckCommitted`]
130    /// without another mutation.
131    ///
132    /// # Errors
133    ///
134    /// Returns [`MarkerAckCommitError`] if the supplied member differs in
135    /// conversation, participant, generation, or cursor prestate.
136    pub fn apply_to<F>(
137        self,
138        member: &mut LiveMember<F>,
139    ) -> Result<MarkerAckCommitted, MarkerAckCommitError> {
140        member
141            .apply_cursor_update(self.cursor_update)
142            .map_err(MarkerAckCommitError::from_member_error)?;
143        Ok(self.outcome)
144    }
145}
146
147/// Failure while applying an already-selected marker-ack commit.
148#[derive(Clone, Copy, Debug, PartialEq, Eq)]
149pub enum MarkerAckCommitError {
150    /// Commit belongs to another conversation.
151    Conversation {
152        /// Conversation captured by the commit.
153        expected: ConversationId,
154        /// Conversation carried by the supplied member.
155        actual: ConversationId,
156    },
157    /// Commit belongs to another participant.
158    Participant {
159        /// Participant captured by the commit.
160        expected: ParticipantId,
161        /// Participant carried by the supplied member.
162        actual: ParticipantId,
163    },
164    /// Commit belongs to another credential generation.
165    Generation {
166        /// Generation captured by the commit.
167        expected: Generation,
168        /// Generation carried by the supplied member.
169        actual: Generation,
170    },
171    /// A malformed internal update did not strictly advance its source cursor.
172    NonAdvancing {
173        /// Cursor from which the update was selected.
174        from_cursor: DeliverySeq,
175        /// Proposed resulting cursor.
176        resulting_cursor: DeliverySeq,
177    },
178    /// Supplied member is neither the exact old nor exact committed prestate.
179    CursorPrestate {
180        /// Cursor from which the commit was selected.
181        expected_from_cursor: DeliverySeq,
182        /// Cursor produced by the commit.
183        resulting_cursor: DeliverySeq,
184        /// Cursor carried by the supplied member.
185        actual_cursor: DeliverySeq,
186    },
187}
188
189impl MarkerAckCommitError {
190    const fn from_member_error(error: LiveMemberCursorUpdateError) -> Self {
191        match error {
192            LiveMemberCursorUpdateError::Conversation { expected, actual } => {
193                Self::Conversation { expected, actual }
194            }
195            LiveMemberCursorUpdateError::Participant { expected, actual } => {
196                Self::Participant { expected, actual }
197            }
198            LiveMemberCursorUpdateError::Generation { expected, actual } => {
199                Self::Generation { expected, actual }
200            }
201            LiveMemberCursorUpdateError::NonAdvancing {
202                from_cursor,
203                resulting_cursor,
204            } => Self::NonAdvancing {
205                from_cursor,
206                resulting_cursor,
207            },
208            LiveMemberCursorUpdateError::CursorPrestate {
209                expected_from_cursor,
210                resulting_cursor,
211                actual_cursor,
212            } => Self::CursorPrestate {
213                expected_from_cursor,
214                resulting_cursor,
215                actual_cursor,
216            },
217        }
218    }
219}
220
221/// Total zero-debt marker-ack decision.
222#[derive(Clone, Debug, PartialEq, Eq)]
223pub enum MarkerAckDecision {
224    /// Authority or marker-proof response; membership is unchanged.
225    Respond(MarkerAckResponse),
226    /// Exact committed response paired with its proof and sole cursor update.
227    Commit(MarkerAckCommit),
228}
229
230/// Applies the complete zero-debt marker-ack selector.
231///
232/// Existing shared lookup enforces retired, unknown, stale-authority, and
233/// exact-binding precedence. Only authorized requests enter marker proof. The
234/// proof snapshot's cursor and proof epoch are derived here from the verified
235/// member and binding; callers provide only the remaining durable marker
236/// facts. An exact accepted marker at the member cursor returns `AckNoOp`.
237/// Every other refusal is nonmutating, while an exact delivered marker returns
238/// a replay-safe [`MarkerAckCommit`].
239#[must_use]
240pub fn apply_marker_ack<EF, V, LF>(
241    presented_identity: PresentedIdentity<'_, EF, V, LF>,
242    binding: &BindingState,
243    receiving_binding_epoch: BindingEpoch,
244    request: &MarkerAck,
245    marker_state: &MarkerProofState,
246) -> MarkerAckDecision {
247    let lookup_request = ParticipantBindingRequest::MarkerAck(request.clone());
248    let (member, active_binding) = match lookup_binding_required(
249        presented_identity,
250        binding,
251        Some(receiving_binding_epoch),
252        &lookup_request,
253    ) {
254        BindingRequiredLookupResult::Retired(outcome) => {
255            return MarkerAckDecision::Respond(MarkerAckResponse::from_retired(outcome));
256        }
257        BindingRequiredLookupResult::ParticipantUnknown(outcome) => {
258            return MarkerAckDecision::Respond(MarkerAckResponse::from_participant_unknown(
259                outcome,
260            ));
261        }
262        BindingRequiredLookupResult::StaleAuthority(outcome) => {
263            return MarkerAckDecision::Respond(MarkerAckResponse::from_stale_authority(outcome));
264        }
265        BindingRequiredLookupResult::NoBinding(outcome) => {
266            return MarkerAckDecision::Respond(MarkerAckResponse::from_no_binding(outcome));
267        }
268        BindingRequiredLookupResult::Authorized { member, binding } => (member, binding),
269    };
270
271    let exact_state = MarkerProofState::new(
272        member.cursor(),
273        marker_state.accepted_marker_at_cursor(),
274        marker_state.expected_marker_delivery_seq(),
275        active_binding.binding_epoch,
276        marker_state.delivered_to_proof_epoch(),
277    );
278    match select_marker_proof(&exact_state, MarkerProofInput::marker_ack(request)) {
279        MarkerProofDecision::AckNoOp(outcome) => {
280            MarkerAckDecision::Respond(MarkerAckResponse::from_ack_no_op(outcome))
281        }
282        MarkerProofDecision::MarkerMismatch(outcome) => {
283            MarkerAckDecision::Respond(MarkerAckResponse::from_marker_mismatch(outcome))
284        }
285        MarkerProofDecision::MarkerNotDelivered(outcome) => {
286            MarkerAckDecision::Respond(MarkerAckResponse::from_marker_not_delivered(outcome))
287        }
288        MarkerProofDecision::Permit(proof) => {
289            let envelope = marker_ack_envelope(request);
290            MarkerAckDecision::Commit(MarkerAckCommit {
291                outcome: MarkerAckCommitted::new(envelope),
292                proof,
293                cursor_update: LiveMemberCursorUpdate::new(
294                    request.conversation_id,
295                    request.participant_id,
296                    request.capability_generation,
297                    member.cursor(),
298                    request.marker_delivery_seq,
299                ),
300            })
301        }
302    }
303}
304
305const fn marker_ack_envelope(request: &MarkerAck) -> MarkerAckEnvelope {
306    MarkerAckEnvelope {
307        conversation_id: request.conversation_id,
308        participant_id: request.participant_id,
309        capability_generation: request.capability_generation,
310        marker_delivery_seq: request.marker_delivery_seq,
311    }
312}
liminal_protocol/lifecycle/operations/marker_ack.rs

liminal_protocol/lifecycle/operations/
marker_ack.rs
