liminal_protocol/lifecycle/operations/
participant_ack.rs

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

liminal_protocol/lifecycle/operations/
participant_ack.rs
