liminal_protocol/lifecycle/
cursor_facts.rs

1use alloc::{collections::BTreeMap, vec::Vec};
2
3use crate::algebra::{FloorComputation, floor_transition};
4use crate::wire::{
5    AckCommitted, AckGap, AckNoOp, AckRegression, BindingEpoch, ConversationId, DeliverySeq,
6    ParticipantAck, ParticipantAckEnvelope, ParticipantId, ParticipantIndex,
7};
8
9use super::edge::ClosureDebt;
10
11/// Participant-scoped cursor progress key mandated by extraction Fix 2.
12#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
13pub struct CursorProgressKey {
14    /// Permanent participant index.
15    pub participant_index: ParticipantIndex,
16    /// Requested cumulative boundary.
17    pub boundary: DeliverySeq,
18}
19
20/// Durable cursor-progress fact state.
21#[derive(Clone, Copy, Debug, PartialEq, Eq)]
22pub enum CursorProgressFact {
23    /// Boundary remains independently fireable for this participant.
24    Pending,
25    /// Boundary was covered by this participant's committed cumulative ack.
26    Consumed,
27}
28
29/// Variable participant-scoped cursor facts; no fixed occurrence array exists.
30#[derive(Clone, Debug, Default, PartialEq, Eq)]
31pub struct CursorProgressFacts {
32    facts: BTreeMap<CursorProgressKey, CursorProgressFact>,
33}
34
35/// Validated durable delivery obligations for one participant ack frontier.
36///
37/// The outbox supplies its sorted live-obligation index, while the protocol
38/// owns participant/frontier binding and endpoint membership. Internal
39/// conversation-sequence gaps are legal; a forward ack endpoint must occur in
40/// this exact recipient index.
41///
42/// This testimony is move-only and cannot be assembled from independent public
43/// fields:
44///
45/// ```compile_fail
46/// use liminal_protocol::lifecycle::RecipientAckObligations;
47///
48/// fn clone_testimony(value: &RecipientAckObligations) -> RecipientAckObligations {
49///     value.clone()
50/// }
51/// ```
52#[derive(Debug, PartialEq, Eq)]
53pub struct RecipientAckObligations {
54    participant_id: ParticipantId,
55    acknowledged_through: DeliverySeq,
56    delivery_sequences: Vec<DeliverySeq>,
57}
58
59/// Malformed durable recipient-obligation testimony.
60#[derive(Clone, Copy, Debug, PartialEq, Eq)]
61pub enum RecipientAckObligationsError {
62    /// An indexed obligation is already at or below the durable ack frontier.
63    NotLive {
64        /// Durable participant acknowledgement frontier.
65        acknowledged_through: DeliverySeq,
66        /// Invalid obligation endpoint.
67        delivery_seq: DeliverySeq,
68    },
69    /// Obligation endpoints are not strictly increasing and duplicate-free.
70    NotStrictlyIncreasing {
71        /// Previous obligation in the supplied index.
72        previous: DeliverySeq,
73        /// Current non-increasing obligation.
74        current: DeliverySeq,
75    },
76}
77
78/// The testimony belongs to another participant or durable ack prestate.
79#[derive(Clone, Copy, Debug, PartialEq, Eq)]
80pub enum RecipientAckObligationsContextError {
81    /// The testimony names another permanent participant.
82    Participant {
83        /// Participant required by the authorized request.
84        expected: ParticipantId,
85        /// Participant carried by the testimony.
86        actual: ParticipantId,
87    },
88    /// The testimony was sealed against another durable ack frontier.
89    AcknowledgedThrough {
90        /// Cursor required by the authorized request prestate.
91        expected: DeliverySeq,
92        /// Cursor carried by the testimony.
93        actual: DeliverySeq,
94    },
95}
96
97impl RecipientAckObligations {
98    /// Validates one recipient's complete current live-obligation index.
99    ///
100    /// # Errors
101    ///
102    /// Returns [`RecipientAckObligationsError`] when an obligation is already
103    /// discharged or the index is not strictly increasing and duplicate-free.
104    pub fn try_new(
105        participant_id: ParticipantId,
106        acknowledged_through: DeliverySeq,
107        delivery_sequences: Vec<DeliverySeq>,
108    ) -> Result<Self, RecipientAckObligationsError> {
109        let mut previous = None;
110        for delivery_seq in &delivery_sequences {
111            if *delivery_seq <= acknowledged_through {
112                return Err(RecipientAckObligationsError::NotLive {
113                    acknowledged_through,
114                    delivery_seq: *delivery_seq,
115                });
116            }
117            if let Some(previous) = previous
118                && *delivery_seq <= previous
119            {
120                return Err(RecipientAckObligationsError::NotStrictlyIncreasing {
121                    previous,
122                    current: *delivery_seq,
123                });
124            }
125            previous = Some(*delivery_seq);
126        }
127        Ok(Self {
128            participant_id,
129            acknowledged_through,
130            delivery_sequences,
131        })
132    }
133
134    pub(in crate::lifecycle) fn contains_endpoint(
135        &self,
136        participant_id: ParticipantId,
137        acknowledged_through: DeliverySeq,
138        endpoint: DeliverySeq,
139    ) -> Result<bool, RecipientAckObligationsContextError> {
140        if self.participant_id != participant_id {
141            return Err(RecipientAckObligationsContextError::Participant {
142                expected: participant_id,
143                actual: self.participant_id,
144            });
145        }
146        if self.acknowledged_through != acknowledged_through {
147            return Err(RecipientAckObligationsContextError::AcknowledgedThrough {
148                expected: acknowledged_through,
149                actual: self.acknowledged_through,
150            });
151        }
152        Ok(self.delivery_sequences.binary_search(&endpoint).is_ok())
153    }
154}
155
156/// One currently bound participant's durable cumulative-cursor state.
157///
158/// All fields are private so cursor advancement can occur only through the
159/// monotonic cumulative-ack transition below.
160#[derive(Clone, Copy, Debug, PartialEq, Eq)]
161pub struct BoundParticipantCursor {
162    participant_id: ParticipantId,
163    active_binding_epoch: BindingEpoch,
164    cursor: DeliverySeq,
165}
166
167impl BoundParticipantCursor {
168    /// Creates one bound participant cursor at its already-durable position.
169    #[must_use]
170    pub const fn new(
171        participant_id: ParticipantId,
172        active_binding_epoch: BindingEpoch,
173        cursor: DeliverySeq,
174    ) -> Self {
175        Self {
176            participant_id,
177            active_binding_epoch,
178            cursor,
179        }
180    }
181
182    /// Returns the participant's permanent index.
183    #[must_use]
184    pub const fn participant_index(self) -> ParticipantIndex {
185        self.participant_id
186    }
187
188    /// Returns the participant's permanent identifier.
189    #[must_use]
190    pub const fn participant_id(self) -> ParticipantId {
191        self.participant_id
192    }
193
194    /// Returns the binding epoch authorized to advance this cursor.
195    #[must_use]
196    pub const fn active_binding_epoch(self) -> BindingEpoch {
197        self.active_binding_epoch
198    }
199
200    /// Returns the durable cumulative cursor.
201    #[must_use]
202    pub const fn cursor(self) -> DeliverySeq {
203        self.cursor
204    }
205
206    const fn advance_to(&mut self, boundary: DeliverySeq) {
207        if boundary > self.cursor {
208            self.cursor = boundary;
209        }
210    }
211}
212
213/// Construction failure for a participant-scoped nonzero-debt episode.
214#[derive(Clone, Copy, Debug, PartialEq, Eq)]
215pub enum CursorEpisodeBuildError {
216    /// Two cursor states name the same permanent participant identifier/index.
217    DuplicateParticipant {
218        /// Repeated permanent participant identifier and index.
219        participant_id: ParticipantId,
220    },
221    /// Hard observer progress cannot be beyond the candidate watermark.
222    ObserverBeyondHighWatermark {
223        /// Durable observer progress.
224        observer_progress: DeliverySeq,
225        /// Candidate high watermark `H'`.
226        candidate_high_watermark: DeliverySeq,
227    },
228    /// A bound cursor cannot acknowledge beyond the candidate watermark.
229    CursorBeyondHighWatermark {
230        /// Permanent participant identifier/index.
231        participant_id: ParticipantId,
232        /// Durable participant cursor.
233        cursor: DeliverySeq,
234        /// Candidate high watermark `H'`.
235        candidate_high_watermark: DeliverySeq,
236    },
237    /// The supplied first-retained floor is beyond checked `H' + 1`.
238    FloorBeyondRetainedEnd {
239        /// Supplied current floor `F`.
240        current_floor: u128,
241        /// Checked one-past retained end `H' + 1`.
242        retained_end: u128,
243    },
244    /// The append-free ack envelope selected a floor below the actual base.
245    CapacityFloorBelowBase {
246        /// Supplied committing-class `cap_floor`.
247        cap_floor: u128,
248        /// `max(F, preferred_floor)` for the initial episode state.
249        base_floor: u128,
250    },
251    /// The supplied capacity floor would overtake hard observer retention.
252    CapacityFloorBeyondObserver {
253        /// Supplied committing-class `cap_floor`.
254        cap_floor: u128,
255        /// Greatest floor allowed by hard observer retention, `o + 1`.
256        observer_limit: u128,
257    },
258}
259
260/// Authority failure before a cumulative-ack transition.
261#[derive(Clone, Copy, Debug, PartialEq, Eq)]
262pub enum CumulativeAckAuthorizationError {
263    /// The episode does not contain the selected permanent participant index.
264    ParticipantIndexUnknown,
265    /// The request names another conversation.
266    ConversationMismatch,
267    /// The request identifier does not match the selected participant index.
268    ParticipantMismatch,
269    /// The request generation does not match the active binding epoch.
270    GenerationMismatch,
271    /// The receiving connection does not own the participant's active epoch.
272    BindingEpochMismatch,
273    /// Durable recipient-obligation testimony belongs to another prestate.
274    ObligationContext {
275        /// Exact testimony mismatch.
276        error: RecipientAckObligationsContextError,
277    },
278    /// A fixed wire-outcome constructor rejected an already-proven cursor relation.
279    CursorRelationInvariant,
280}
281
282/// Exhaustive outcome of an authority-checked normal cumulative ack.
283#[derive(Clone, Debug, PartialEq, Eq)]
284pub enum CumulativeAckOutcome {
285    /// The cursor advanced and its participant-scoped fact was consumed.
286    Committed(AckCommitted),
287    /// The request exactly repeated the durable cursor.
288    NoOp(AckNoOp),
289    /// The forward endpoint lacks the availability testimony required by the selector.
290    Gap(AckGap),
291    /// The request boundary was below the durable cursor.
292    Regression(AckRegression),
293}
294
295/// Participant-scoped cursor accounting for one provably nonzero-debt episode.
296///
297/// The wrapper requires [`ClosureDebt`], whose constructor rejects componentwise
298/// zero. Unlike the frozen document's defective fixed occurrence array, its
299/// progress facts are variable and keyed by `(participant_index, boundary)` as
300/// mandated by `docs/design/LP-EXTRACTION-GOAL.md` Fix 2. It also owns the hard
301/// observer position, candidate watermark, retained-suffix range, and the
302/// append-free ack class's capacity floor. Every committed ack recomputes
303/// `F' = max(F, min(m, o) + 1, cap_floor)` from those durable values.
304#[derive(Clone, Debug, PartialEq, Eq)]
305pub struct NonzeroDebtCursorEpisode {
306    conversation_id: ConversationId,
307    debt: ClosureDebt,
308    observer_progress: DeliverySeq,
309    candidate_high_watermark: DeliverySeq,
310    cap_floor: u128,
311    floor: FloorComputation,
312    participants: BTreeMap<ParticipantIndex, BoundParticipantCursor>,
313    facts: CursorProgressFacts,
314}
315
316/// Deterministic cursor-fact serialization failure.
317#[derive(Clone, Copy, Debug, PartialEq, Eq)]
318pub enum CursorFactEncodeError {
319    /// Participant count cannot fit the variable format's `u32` count.
320    TooManyParticipants,
321    /// Fact count cannot fit the variable format's `u32` count.
322    TooManyFacts,
323    /// Encoded byte length overflowed the platform allocation domain.
324    LengthOverflow,
325}
326
327impl CursorProgressFacts {
328    /// Creates an empty participant-scoped fact map.
329    #[must_use]
330    pub const fn new() -> Self {
331        Self {
332            facts: BTreeMap::new(),
333        }
334    }
335
336    /// Inserts an independently fireable `(participant_index, boundary)` fact.
337    ///
338    /// Returns `true` only when the exact pair was not already present.
339    pub fn record(&mut self, key: CursorProgressKey) -> bool {
340        if self.facts.contains_key(&key) {
341            return false;
342        }
343        self.facts.insert(key, CursorProgressFact::Pending);
344        true
345    }
346
347    /// Marks every pending boundary at or below `through` consumed for one
348    /// participant, without touching another participant's identical boundary.
349    pub fn consume_through(
350        &mut self,
351        participant_index: ParticipantIndex,
352        through: DeliverySeq,
353    ) -> Vec<CursorProgressKey> {
354        let mut consumed = Vec::new();
355        for (key, fact) in &mut self.facts {
356            if key.participant_index == participant_index
357                && key.boundary <= through
358                && *fact == CursorProgressFact::Pending
359            {
360                *fact = CursorProgressFact::Consumed;
361                consumed.push(*key);
362            }
363        }
364        consumed
365    }
366
367    /// Returns one exact fact state.
368    #[must_use]
369    pub fn get(&self, key: CursorProgressKey) -> Option<CursorProgressFact> {
370        self.facts.get(&key).copied()
371    }
372
373    /// Number of distinct participant/boundary pairs.
374    #[must_use]
375    pub fn len(&self) -> usize {
376        self.facts.len()
377    }
378
379    /// Returns whether no cursor fact exists.
380    #[must_use]
381    pub fn is_empty(&self) -> bool {
382        self.facts.is_empty()
383    }
384
385    /// Deterministically serializes the variable map in key order.
386    ///
387    /// Format: `count:u32`, followed by `participant_index:u64`,
388    /// `boundary:u64`, and `state:u8` (`0` Pending, `1` Consumed) per fact.
389    /// This is lifecycle storage, not the participant network frame format.
390    ///
391    /// # Errors
392    ///
393    /// Returns [`CursorFactEncodeError`] if count or allocation length cannot be
394    /// represented.
395    pub fn encode(&self) -> Result<Vec<u8>, CursorFactEncodeError> {
396        let count =
397            u32::try_from(self.facts.len()).map_err(|_| CursorFactEncodeError::TooManyFacts)?;
398        let body_len = self
399            .facts
400            .len()
401            .checked_mul(17)
402            .ok_or(CursorFactEncodeError::LengthOverflow)?;
403        let capacity = 4_usize
404            .checked_add(body_len)
405            .ok_or(CursorFactEncodeError::LengthOverflow)?;
406        let mut bytes = Vec::with_capacity(capacity);
407        bytes.extend_from_slice(&count.to_be_bytes());
408        for (key, fact) in &self.facts {
409            bytes.extend_from_slice(&key.participant_index.to_be_bytes());
410            bytes.extend_from_slice(&key.boundary.to_be_bytes());
411            bytes.push(match fact {
412                CursorProgressFact::Pending => 0,
413                CursorProgressFact::Consumed => 1,
414            });
415        }
416        Ok(bytes)
417    }
418}
419
420impl NonzeroDebtCursorEpisode {
421    /// Creates one nonzero-debt episode with retained-suffix and floor state.
422    ///
423    /// `candidate_high_watermark` is `H'`; the retained suffix is the inclusive
424    /// sequence range from `current_floor` through `H'`, or empty when the floor
425    /// is checked `H' + 1`. `cap_floor` is the committing append-free ack
426    /// class's actual capacity floor at this initial state.
427    ///
428    /// # Errors
429    ///
430    /// Returns [`CursorEpisodeBuildError`] for a duplicate participant, a
431    /// cursor/observer beyond `H'`, an invalid retained range, or a capacity
432    /// floor outside the initial base and observer bounds.
433    #[allow(clippy::too_many_arguments)]
434    pub fn new(
435        conversation_id: ConversationId,
436        debt: ClosureDebt,
437        observer_progress: DeliverySeq,
438        candidate_high_watermark: DeliverySeq,
439        current_floor: u128,
440        cap_floor: u128,
441        participants: Vec<BoundParticipantCursor>,
442    ) -> Result<Self, CursorEpisodeBuildError> {
443        if observer_progress > candidate_high_watermark {
444            return Err(CursorEpisodeBuildError::ObserverBeyondHighWatermark {
445                observer_progress,
446                candidate_high_watermark,
447            });
448        }
449        let retained_end = u128::from(candidate_high_watermark) + 1;
450        if current_floor > retained_end {
451            return Err(CursorEpisodeBuildError::FloorBeyondRetainedEnd {
452                current_floor,
453                retained_end,
454            });
455        }
456
457        let mut indexed = BTreeMap::new();
458        for participant in participants {
459            if participant.cursor > candidate_high_watermark {
460                return Err(CursorEpisodeBuildError::CursorBeyondHighWatermark {
461                    participant_id: participant.participant_id,
462                    cursor: participant.cursor,
463                    candidate_high_watermark,
464                });
465            }
466            if indexed.contains_key(&participant.participant_id) {
467                return Err(CursorEpisodeBuildError::DuplicateParticipant {
468                    participant_id: participant.participant_id,
469                });
470            }
471            indexed.insert(participant.participant_id, participant);
472        }
473
474        let minimum_member_cursor = indexed.values().map(|participant| participant.cursor).min();
475        let floor = floor_transition(
476            current_floor,
477            minimum_member_cursor,
478            candidate_high_watermark,
479            observer_progress,
480            cap_floor,
481        );
482        let base_floor = current_floor.max(floor.preferred_floor);
483        if cap_floor < base_floor {
484            return Err(CursorEpisodeBuildError::CapacityFloorBelowBase {
485                cap_floor,
486                base_floor,
487            });
488        }
489        let observer_limit = u128::from(observer_progress) + 1;
490        if cap_floor > observer_limit {
491            return Err(CursorEpisodeBuildError::CapacityFloorBeyondObserver {
492                cap_floor,
493                observer_limit,
494            });
495        }
496
497        Ok(Self {
498            conversation_id,
499            debt,
500            observer_progress,
501            candidate_high_watermark,
502            cap_floor,
503            floor,
504            participants: indexed,
505            facts: CursorProgressFacts::new(),
506        })
507    }
508
509    #[allow(clippy::too_many_arguments)]
510    pub(super) fn restore(
511        conversation_id: ConversationId,
512        debt: ClosureDebt,
513        observer_progress: DeliverySeq,
514        candidate_high_watermark: DeliverySeq,
515        current_floor: u128,
516        cap_floor: u128,
517        participants: Vec<BoundParticipantCursor>,
518        facts: Vec<(CursorProgressKey, CursorProgressFact)>,
519    ) -> Option<Self> {
520        let mut episode = Self::new(
521            conversation_id,
522            debt,
523            observer_progress,
524            candidate_high_watermark,
525            current_floor,
526            cap_floor,
527            participants,
528        )
529        .ok()?;
530        if episode.floor.resulting_floor != current_floor {
531            return None;
532        }
533        for (key, fact) in facts {
534            let participant = episode.participants.get(&key.participant_index)?;
535            if key.boundary > candidate_high_watermark
536                || (fact == CursorProgressFact::Consumed && key.boundary > participant.cursor)
537                || (fact == CursorProgressFact::Pending && key.boundary <= participant.cursor)
538                || episode.facts.facts.insert(key, fact).is_some()
539            {
540                return None;
541            }
542        }
543        Some(episode)
544    }
545
546    /// Returns the conversation owning this aggregate for internal operation
547    /// prestate validation.
548    pub(super) const fn conversation_id(&self) -> ConversationId {
549        self.conversation_id
550    }
551
552    /// Returns the exact nonzero debt proving this episode is active.
553    #[must_use]
554    pub const fn debt(&self) -> ClosureDebt {
555        self.debt
556    }
557
558    /// Returns hard observer progress `o` used by every ack floor transition.
559    #[must_use]
560    pub const fn observer_progress(&self) -> DeliverySeq {
561        self.observer_progress
562    }
563
564    /// Returns the candidate high watermark `H'` and retained-suffix end.
565    #[must_use]
566    pub const fn candidate_high_watermark(&self) -> DeliverySeq {
567        self.candidate_high_watermark
568    }
569
570    /// Returns the append-free ack transaction class's current `cap_floor`.
571    #[must_use]
572    pub const fn cap_floor(&self) -> u128 {
573        self.cap_floor
574    }
575
576    /// Returns the latest reproducible document floor computation.
577    #[must_use]
578    pub const fn floor_computation(&self) -> FloorComputation {
579        self.floor
580    }
581
582    /// Returns the first retained sequence, or `None` for an empty suffix.
583    #[must_use]
584    pub fn retained_suffix_start(&self) -> Option<DeliverySeq> {
585        if self.floor.resulting_floor > u128::from(self.candidate_high_watermark) {
586            return None;
587        }
588        DeliverySeq::try_from(self.floor.resulting_floor).ok()
589    }
590
591    /// Returns whether one durable sequence remains in the retained suffix.
592    #[must_use]
593    pub fn retains(&self, delivery_seq: DeliverySeq) -> bool {
594        u128::from(delivery_seq) >= self.floor.resulting_floor
595            && delivery_seq <= self.candidate_high_watermark
596    }
597
598    /// Returns one indexed bound participant cursor.
599    #[must_use]
600    pub fn participant(
601        &self,
602        participant_index: ParticipantIndex,
603    ) -> Option<BoundParticipantCursor> {
604        self.participants.get(&participant_index).copied()
605    }
606
607    /// Returns the participant-scoped cursor fact map.
608    #[must_use]
609    pub const fn facts(&self) -> &CursorProgressFacts {
610        &self.facts
611    }
612
613    /// Applies one authority-checked cumulative normal acknowledgement.
614    ///
615    /// `receiving_binding_epoch` identifies the connection epoch on which the
616    /// request arrived. This scalar entry point preserves the Unit 1 contiguous
617    /// suffix selector; Unit 2 production uses
618    /// [`Self::acknowledge_with_obligations`] so restart-safe durable recipient
619    /// testimony, rather than volatile offered progress, decides endpoint
620    /// availability. A successful advance records and consumes the fact keyed by
621    /// the selected participant index and requested boundary; any lower pending
622    /// facts for that participant are consumed in the same transition.
623    /// It then recomputes the physical floor from the post-ack minimum cursor,
624    /// hard observer progress, current floor, and append-free ack `cap_floor`.
625    /// Another participant's equal boundary is never touched.
626    ///
627    /// # Errors
628    ///
629    /// Returns [`CumulativeAckAuthorizationError`] before any mutation when the
630    /// conversation, participant, generation, or active binding epoch differs.
631    pub fn acknowledge(
632        &mut self,
633        participant_index: ParticipantIndex,
634        receiving_binding_epoch: BindingEpoch,
635        request: &ParticipantAck,
636        contiguously_available_through: DeliverySeq,
637    ) -> Result<CumulativeAckOutcome, CumulativeAckAuthorizationError> {
638        let available_through = contiguously_available_through.min(self.candidate_high_watermark);
639        self.acknowledge_by_endpoint(
640            participant_index,
641            receiving_binding_epoch,
642            request,
643            move |_, _, endpoint| Ok(endpoint <= available_through),
644        )
645    }
646
647    /// Applies the durable per-recipient obligation endpoint rule.
648    ///
649    /// Conversation-global sequence gaps are skipped as non-obligations. A
650    /// forward request commits only when its endpoint exists in `obligations`;
651    /// ending on a sender-excluded or otherwise absent sequence returns
652    /// `AckGap`. The testimony must name the selected participant and current
653    /// durable cursor exactly.
654    ///
655    /// # Errors
656    ///
657    /// Returns [`CumulativeAckAuthorizationError`] before mutation when request
658    /// authority or testimony context differs from this episode.
659    pub fn acknowledge_with_obligations(
660        &mut self,
661        participant_index: ParticipantIndex,
662        receiving_binding_epoch: BindingEpoch,
663        request: &ParticipantAck,
664        obligations: &RecipientAckObligations,
665    ) -> Result<CumulativeAckOutcome, CumulativeAckAuthorizationError> {
666        self.acknowledge_by_endpoint(
667            participant_index,
668            receiving_binding_epoch,
669            request,
670            |participant_id, acknowledged_through, endpoint| {
671                obligations
672                    .contains_endpoint(participant_id, acknowledged_through, endpoint)
673                    .map_err(|error| CumulativeAckAuthorizationError::ObligationContext { error })
674            },
675        )
676    }
677
678    fn acknowledge_by_endpoint<F>(
679        &mut self,
680        participant_index: ParticipantIndex,
681        receiving_binding_epoch: BindingEpoch,
682        request: &ParticipantAck,
683        endpoint_is_available: F,
684    ) -> Result<CumulativeAckOutcome, CumulativeAckAuthorizationError>
685    where
686        F: FnOnce(
687            ParticipantId,
688            DeliverySeq,
689            DeliverySeq,
690        ) -> Result<bool, CumulativeAckAuthorizationError>,
691    {
692        let Some(participant) = self.participants.get(&participant_index).copied() else {
693            return Err(CumulativeAckAuthorizationError::ParticipantIndexUnknown);
694        };
695        if request.conversation_id != self.conversation_id {
696            return Err(CumulativeAckAuthorizationError::ConversationMismatch);
697        }
698        if request.participant_id != participant.participant_id {
699            return Err(CumulativeAckAuthorizationError::ParticipantMismatch);
700        }
701        if request.capability_generation != participant.active_binding_epoch.capability_generation {
702            return Err(CumulativeAckAuthorizationError::GenerationMismatch);
703        }
704        if receiving_binding_epoch != participant.active_binding_epoch {
705            return Err(CumulativeAckAuthorizationError::BindingEpochMismatch);
706        }
707
708        let current_cursor = participant.cursor;
709        let through_seq = request.through_seq;
710        let envelope = ParticipantAckEnvelope {
711            conversation_id: request.conversation_id,
712            participant_id: request.participant_id,
713            capability_generation: request.capability_generation,
714            through_seq,
715        };
716        let endpoint_available =
717            endpoint_is_available(participant.participant_id, current_cursor, through_seq)?;
718
719        if through_seq < current_cursor {
720            return AckRegression::new(envelope, current_cursor)
721                .map(CumulativeAckOutcome::Regression)
722                .ok_or(CumulativeAckAuthorizationError::CursorRelationInvariant);
723        }
724        if through_seq == current_cursor {
725            return Ok(CumulativeAckOutcome::NoOp(AckNoOp::participant_ack(
726                envelope,
727            )));
728        }
729        if through_seq > self.candidate_high_watermark || !endpoint_available {
730            return AckGap::new(envelope, current_cursor)
731                .map(CumulativeAckOutcome::Gap)
732                .ok_or(CumulativeAckAuthorizationError::CursorRelationInvariant);
733        }
734
735        let Some(stored) = self.participants.get_mut(&participant_index) else {
736            return Err(CumulativeAckAuthorizationError::ParticipantIndexUnknown);
737        };
738        stored.advance_to(through_seq);
739        let key = CursorProgressKey {
740            participant_index,
741            boundary: through_seq,
742        };
743        self.facts.record(key);
744        let _ = self.facts.consume_through(participant_index, through_seq);
745
746        let minimum_member_cursor = self
747            .participants
748            .values()
749            .map(|participant| participant.cursor)
750            .min();
751        self.floor = floor_transition(
752            self.floor.resulting_floor,
753            minimum_member_cursor,
754            self.candidate_high_watermark,
755            self.observer_progress,
756            self.cap_floor,
757        );
758        // `cap_floor` is defined from `f >= base_floor`, so for the next
759        // append-free ack its durable lower bound is the just-committed floor.
760        self.cap_floor = self.floor.resulting_floor;
761
762        Ok(CumulativeAckOutcome::Committed(AckCommitted::new(envelope)))
763    }
764
765    /// Deterministically serializes the active debt, bound cursors, and facts.
766    ///
767    /// Format: conversation id, debt entries/bytes as two u128 values, observer
768    /// progress and `H'` as two u64 values, current `F'` and `cap_floor` as two
769    /// u128 values, participant count as u32, participants in index order, then
770    /// fact count as u32 and facts in `(participant_index, boundary)` order. The
771    /// retained suffix is reproduced exactly by `[F', H']`. A participant is
772    /// five u64 values: its one canonical identifier/index, server incarnation,
773    /// connection ordinal, generation, and cursor. A fact retains the 17-byte
774    /// format documented by [`CursorProgressFacts::encode`]. This is lifecycle
775    /// storage, not the participant network frame format.
776    ///
777    /// # Errors
778    ///
779    /// Returns [`CursorFactEncodeError`] if either count or the exact allocation
780    /// length cannot be represented.
781    pub fn encode(&self) -> Result<Vec<u8>, CursorFactEncodeError> {
782        let participant_count = u32::try_from(self.participants.len())
783            .map_err(|_| CursorFactEncodeError::TooManyParticipants)?;
784        let fact_count =
785            u32::try_from(self.facts.len()).map_err(|_| CursorFactEncodeError::TooManyFacts)?;
786        let participant_bytes = self
787            .participants
788            .len()
789            .checked_mul(40)
790            .ok_or(CursorFactEncodeError::LengthOverflow)?;
791        let fact_bytes = self
792            .facts
793            .len()
794            .checked_mul(17)
795            .ok_or(CursorFactEncodeError::LengthOverflow)?;
796        let capacity = 96_usize
797            .checked_add(participant_bytes)
798            .and_then(|length| length.checked_add(fact_bytes))
799            .ok_or(CursorFactEncodeError::LengthOverflow)?;
800
801        let debt = self.debt.value();
802        let mut bytes = Vec::with_capacity(capacity);
803        bytes.extend_from_slice(&self.conversation_id.to_be_bytes());
804        bytes.extend_from_slice(&debt.entries.to_be_bytes());
805        bytes.extend_from_slice(&debt.bytes.to_be_bytes());
806        bytes.extend_from_slice(&self.observer_progress.to_be_bytes());
807        bytes.extend_from_slice(&self.candidate_high_watermark.to_be_bytes());
808        bytes.extend_from_slice(&self.floor.resulting_floor.to_be_bytes());
809        bytes.extend_from_slice(&self.cap_floor.to_be_bytes());
810        bytes.extend_from_slice(&participant_count.to_be_bytes());
811        for participant in self.participants.values() {
812            bytes.extend_from_slice(&participant.participant_id.to_be_bytes());
813            bytes.extend_from_slice(
814                &participant
815                    .active_binding_epoch
816                    .connection_incarnation
817                    .server_incarnation
818                    .to_be_bytes(),
819            );
820            bytes.extend_from_slice(
821                &participant
822                    .active_binding_epoch
823                    .connection_incarnation
824                    .connection_ordinal
825                    .to_be_bytes(),
826            );
827            bytes.extend_from_slice(
828                &participant
829                    .active_binding_epoch
830                    .capability_generation
831                    .get()
832                    .to_be_bytes(),
833            );
834            bytes.extend_from_slice(&participant.cursor.to_be_bytes());
835        }
836        bytes.extend_from_slice(&fact_count.to_be_bytes());
837        for (key, fact) in &self.facts.facts {
838            bytes.extend_from_slice(&key.participant_index.to_be_bytes());
839            bytes.extend_from_slice(&key.boundary.to_be_bytes());
840            bytes.push(match fact {
841                CursorProgressFact::Pending => 0,
842                CursorProgressFact::Consumed => 1,
843            });
844        }
845        Ok(bytes)
846    }
847}
liminal_protocol/lifecycle/cursor_facts.rs

liminal_protocol/lifecycle/
cursor_facts.rs
