liminal_protocol/lifecycle/operations/
marker_ack.rs

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

liminal_protocol/lifecycle/operations/
marker_ack.rs
