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

liminal_protocol/lifecycle/operations/
marker_ack.rs
