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

liminal_protocol/lifecycle/operations/
participant_ack.rs
