liminal_protocol/lifecycle/operations/
participant_ack.rs

1use alloc::boxed::Box;
2
3use crate::wire::{
4    AckCommitted, AckGap, AckRegression, BindingEpoch, ConversationId, DeliverySeq, Generation,
5    ParticipantAck, ParticipantAckEnvelope, ParticipantAckResponse, ParticipantId,
6};
7
8use super::super::{
9    BindingRequiredLookupResult, BindingState, LiveMember, ObserverProgressProjection,
10    ParticipantBindingRequest, PresentedIdentity, RecipientAckObligations,
11    RecipientAckObligationsContextError, SealedBindingFateToken, lookup_binding_required,
12    membership::{LiveMemberCursorUpdate, LiveMemberCursorUpdateError},
13};
14
15/// Atomic zero-debt participant-ack commit.
16///
17/// The cursor update is intentionally opaque. Only [`Self::apply_to`] can
18/// validate and apply it to a [`LiveMember`], while [`Self::outcome`] exposes
19/// the crate-owned wire success for persistence and response encoding.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub struct ParticipantAckCommit {
22    outcome: AckCommitted,
23    cursor_update: LiveMemberCursorUpdate,
24    binding_epoch: BindingEpoch,
25    previous_cursor: DeliverySeq,
26}
27
28impl ParticipantAckCommit {
29    /// Borrows the exact committed wire outcome.
30    #[must_use]
31    pub const fn outcome(&self) -> &AckCommitted {
32        &self.outcome
33    }
34
35    /// Projects the exact committed ack boundary into hard observer progress.
36    #[must_use]
37    pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
38        let request = self.outcome.request();
39        ObserverProgressProjection::sourced(
40            request.conversation_id,
41            request.through_seq,
42            super::super::checkpoint_codec::ObserverProjectionSource::Ack {
43                binding: super::super::ActiveBinding {
44                    conversation_id: request.conversation_id,
45                    participant_id: request.participant_id,
46                    binding_epoch: self.binding_epoch,
47                },
48                previous_cursor: self.previous_cursor,
49            },
50        )
51    }
52
53    /// Replays this exact selected cursor transition into one sealed fate token.
54    ///
55    /// # Errors
56    ///
57    /// Returns the unchanged token when its conversation, participant, binding
58    /// epoch, or cursor prestate differs from this committed acknowledgement.
59    pub fn progress_binding_fate_token(
60        &self,
61        token: SealedBindingFateToken,
62    ) -> Result<SealedBindingFateToken, Box<SealedBindingFateToken>> {
63        let request = self.outcome.request();
64        token.participant_ack_progressed(
65            request.conversation_id,
66            request.participant_id,
67            self.binding_epoch,
68            self.previous_cursor,
69            request.through_seq,
70        )
71    }
72
73    /// Applies this commit to either its exact old cursor or its already-written
74    /// resulting cursor.
75    ///
76    /// Replaying after a crash is idempotent: the old prestate advances once,
77    /// while the exact new prestate returns the same [`AckCommitted`] without a
78    /// second mutation.
79    ///
80    /// # Errors
81    ///
82    /// Returns [`ParticipantAckCommitError`] if the supplied member differs in
83    /// conversation, participant, generation, or cursor prestate.
84    pub fn apply_to<F>(
85        self,
86        member: &mut LiveMember<F>,
87    ) -> Result<AckCommitted, ParticipantAckCommitError> {
88        member
89            .apply_cursor_update(self.cursor_update)
90            .map_err(ParticipantAckCommitError::from_member_error)?;
91        Ok(self.outcome)
92    }
93}
94
95/// Failure while applying an already-selected participant-ack commit.
96#[derive(Clone, Copy, Debug, PartialEq, Eq)]
97pub enum ParticipantAckCommitError {
98    /// Commit belongs to another conversation.
99    Conversation {
100        /// Conversation captured by the commit.
101        expected: ConversationId,
102        /// Conversation carried by the supplied member.
103        actual: ConversationId,
104    },
105    /// Commit belongs to another participant.
106    Participant {
107        /// Participant captured by the commit.
108        expected: ParticipantId,
109        /// Participant carried by the supplied member.
110        actual: ParticipantId,
111    },
112    /// Commit belongs to another credential generation.
113    Generation {
114        /// Generation captured by the commit.
115        expected: Generation,
116        /// Generation carried by the supplied member.
117        actual: Generation,
118    },
119    /// A malformed internal update did not strictly advance its source cursor.
120    NonAdvancing {
121        /// Cursor from which the update was selected.
122        from_cursor: DeliverySeq,
123        /// Proposed resulting cursor.
124        resulting_cursor: DeliverySeq,
125    },
126    /// Supplied member is neither the exact old nor exact committed prestate.
127    CursorPrestate {
128        /// Cursor from which the commit was selected.
129        expected_from_cursor: DeliverySeq,
130        /// Cursor produced by the commit.
131        resulting_cursor: DeliverySeq,
132        /// Cursor carried by the supplied member.
133        actual_cursor: DeliverySeq,
134    },
135}
136
137impl ParticipantAckCommitError {
138    const fn from_member_error(error: LiveMemberCursorUpdateError) -> Self {
139        match error {
140            LiveMemberCursorUpdateError::Conversation { expected, actual } => {
141                Self::Conversation { expected, actual }
142            }
143            LiveMemberCursorUpdateError::Participant { expected, actual } => {
144                Self::Participant { expected, actual }
145            }
146            LiveMemberCursorUpdateError::Generation { expected, actual } => {
147                Self::Generation { expected, actual }
148            }
149            LiveMemberCursorUpdateError::NonAdvancing {
150                from_cursor,
151                resulting_cursor,
152            } => Self::NonAdvancing {
153                from_cursor,
154                resulting_cursor,
155            },
156            LiveMemberCursorUpdateError::CursorPrestate {
157                expected_from_cursor,
158                resulting_cursor,
159                actual_cursor,
160            } => Self::CursorPrestate {
161                expected_from_cursor,
162                resulting_cursor,
163                actual_cursor,
164            },
165        }
166    }
167}
168
169/// Total zero-debt participant-ack decision.
170#[derive(Clone, Debug, PartialEq, Eq)]
171pub enum ParticipantAckDecision {
172    /// Exact lookup, regression, no-op, or gap response; membership is unchanged.
173    Respond(ParticipantAckResponse),
174    /// Exact committed response paired with its sole cursor update authority.
175    Commit(ParticipantAckCommit),
176}
177
178/// Applies the complete zero-debt participant-ack selector.
179///
180/// Existing shared lookup enforces retired, unknown, stale-authority, and
181/// exact-binding precedence, including the receiving [`BindingEpoch`]. An
182/// authorized request then selects `AckRegression`, `AckNoOp`, `AckGap`, or
183/// `AckCommitted` in that exact relation order. Observer, order, sequence, and
184/// closure checks are absent because none is selectable for zero-debt ack.
185#[must_use]
186pub fn apply_participant_ack<EF, V, LF>(
187    presented_identity: PresentedIdentity<'_, EF, V, LF>,
188    binding: &BindingState,
189    receiving_binding_epoch: BindingEpoch,
190    request: &ParticipantAck,
191    contiguously_available_through: DeliverySeq,
192) -> ParticipantAckDecision {
193    let lookup_request = ParticipantBindingRequest::ParticipantAck(request.clone());
194    let member = match lookup_binding_required(
195        presented_identity,
196        binding,
197        Some(receiving_binding_epoch),
198        &lookup_request,
199    ) {
200        BindingRequiredLookupResult::Retired(outcome) => {
201            return ParticipantAckDecision::Respond(ParticipantAckResponse::from_retired(outcome));
202        }
203        BindingRequiredLookupResult::ParticipantUnknown(outcome) => {
204            return ParticipantAckDecision::Respond(
205                ParticipantAckResponse::from_participant_unknown(outcome),
206            );
207        }
208        BindingRequiredLookupResult::StaleAuthority(outcome) => {
209            return ParticipantAckDecision::Respond(ParticipantAckResponse::from_stale_authority(
210                outcome,
211            ));
212        }
213        BindingRequiredLookupResult::NoBinding(outcome) => {
214            return ParticipantAckDecision::Respond(ParticipantAckResponse::from_no_binding(
215                outcome,
216            ));
217        }
218        BindingRequiredLookupResult::Authorized { member, .. } => member,
219    };
220
221    let current_cursor = member.cursor();
222    if request.through_seq < current_cursor
223        && let Some(outcome) = AckRegression::new(ack_envelope(request), current_cursor)
224    {
225        return ParticipantAckDecision::Respond(ParticipantAckResponse::ack_regression(outcome));
226    }
227    if request.through_seq == current_cursor {
228        return ParticipantAckDecision::Respond(ParticipantAckResponse::ack_no_op(ack_envelope(
229            request,
230        )));
231    }
232    if request.through_seq > contiguously_available_through
233        && let Some(outcome) = AckGap::new(ack_envelope(request), current_cursor)
234    {
235        return ParticipantAckDecision::Respond(ParticipantAckResponse::ack_gap(outcome));
236    }
237
238    let outcome = AckCommitted::new(ack_envelope(request));
239    ParticipantAckDecision::Commit(ParticipantAckCommit {
240        binding_epoch: receiving_binding_epoch,
241        previous_cursor: current_cursor,
242        cursor_update: LiveMemberCursorUpdate::new(
243            request.conversation_id,
244            request.participant_id,
245            request.capability_generation,
246            current_cursor,
247            request.through_seq,
248        ),
249        outcome,
250    })
251}
252
253/// Applies the Unit 2 durable-obligation endpoint rule after shared lookup.
254///
255/// Lookup precedence and relation ordering are identical to
256/// [`apply_participant_ack`]. Unlike its Unit 1 scalar boundary, this selector
257/// requires the requested forward endpoint to exist in the recipient's sealed
258/// committed-obligation index. Internal conversation sequence gaps are skipped;
259/// ending on a non-obligation is exactly `AckGap`.
260///
261/// # Errors
262///
263/// Returns [`RecipientAckObligationsContextError`] if the testimony belongs to
264/// another participant or durable cursor prestate. Such disagreement is an
265/// invariant fault, not a fabricated protocol refusal.
266pub fn apply_participant_ack_with_obligations<EF, V, LF>(
267    presented_identity: PresentedIdentity<'_, EF, V, LF>,
268    binding: &BindingState,
269    receiving_binding_epoch: BindingEpoch,
270    request: &ParticipantAck,
271    obligations: &RecipientAckObligations,
272) -> Result<ParticipantAckDecision, RecipientAckObligationsContextError> {
273    let lookup_request = ParticipantBindingRequest::ParticipantAck(request.clone());
274    let member = match lookup_binding_required(
275        presented_identity,
276        binding,
277        Some(receiving_binding_epoch),
278        &lookup_request,
279    ) {
280        BindingRequiredLookupResult::Retired(outcome) => {
281            return Ok(ParticipantAckDecision::Respond(
282                ParticipantAckResponse::from_retired(outcome),
283            ));
284        }
285        BindingRequiredLookupResult::ParticipantUnknown(outcome) => {
286            return Ok(ParticipantAckDecision::Respond(
287                ParticipantAckResponse::from_participant_unknown(outcome),
288            ));
289        }
290        BindingRequiredLookupResult::StaleAuthority(outcome) => {
291            return Ok(ParticipantAckDecision::Respond(
292                ParticipantAckResponse::from_stale_authority(outcome),
293            ));
294        }
295        BindingRequiredLookupResult::NoBinding(outcome) => {
296            return Ok(ParticipantAckDecision::Respond(
297                ParticipantAckResponse::from_no_binding(outcome),
298            ));
299        }
300        BindingRequiredLookupResult::Authorized { member, .. } => member,
301    };
302
303    let current_cursor = member.cursor();
304    let endpoint_is_obligation = obligations.contains_endpoint(
305        request.participant_id,
306        current_cursor,
307        request.through_seq,
308    )?;
309    if request.through_seq < current_cursor
310        && let Some(outcome) = AckRegression::new(ack_envelope(request), current_cursor)
311    {
312        return Ok(ParticipantAckDecision::Respond(
313            ParticipantAckResponse::ack_regression(outcome),
314        ));
315    }
316    if request.through_seq == current_cursor {
317        return Ok(ParticipantAckDecision::Respond(
318            ParticipantAckResponse::ack_no_op(ack_envelope(request)),
319        ));
320    }
321    if !endpoint_is_obligation
322        && let Some(outcome) = AckGap::new(ack_envelope(request), current_cursor)
323    {
324        return Ok(ParticipantAckDecision::Respond(
325            ParticipantAckResponse::ack_gap(outcome),
326        ));
327    }
328
329    let outcome = AckCommitted::new(ack_envelope(request));
330    Ok(ParticipantAckDecision::Commit(ParticipantAckCommit {
331        binding_epoch: receiving_binding_epoch,
332        previous_cursor: current_cursor,
333        cursor_update: LiveMemberCursorUpdate::new(
334            request.conversation_id,
335            request.participant_id,
336            request.capability_generation,
337            current_cursor,
338            request.through_seq,
339        ),
340        outcome,
341    }))
342}
343
344const fn ack_envelope(request: &ParticipantAck) -> ParticipantAckEnvelope {
345    ParticipantAckEnvelope {
346        conversation_id: request.conversation_id,
347        participant_id: request.participant_id,
348        capability_generation: request.capability_generation,
349        through_seq: request.through_seq,
350    }
351}
liminal_protocol/lifecycle/operations/participant_ack.rs

liminal_protocol/lifecycle/operations/
participant_ack.rs
