liminal_server/server/participant/production/
committed_operations.rs

1//! The B7 broker read: one conversation's committed operation rows, read
2//! through the public surface in bounded pages, the flat native prefix first
3//! and then the routed tail.
4//!
5//! Since every production commit became one routed batch on the
6//! conversation's route, a reader of the flat stream alone sees only the
7//! native prefix a store written before routing holds, and nothing a live
8//! commit staged. This read takes the same branch the server's own
9//! `OperationLog::read_committed_at` takes: below the native head a row is a
10//! flat row, at or after it a routed row under the conversation's route, read
11//! only up to the routed stream's validated head.
12//!
13//! It answers each row's stored payload bytes exactly as committed, without
14//! decoding them, so two stores that committed the same operations answer the
15//! same bytes.
16
17use std::sync::Arc;
18
19use liminal::durability::{DurabilityError, DurableStore, StoredEntry};
20
21use super::log::OperationLog;
22use super::log_error::StreamKeyRefusal;
23use super::restore_checkpoint_locator::checkpoint_conversation_route;
24use super::routed_stream::{self, RoutedStreamError, RoutedStreamRefusal};
25
26/// The most rows one page of [`read_committed_operations`] answers.
27pub const COMMITTED_OPERATION_PAGE_MAX: usize = 256;
28
29/// One committed operation row: its position in the conversation's operation
30/// log and the payload bytes stored for it.
31#[derive(Clone, Debug, PartialEq, Eq)]
32pub struct CommittedOperation {
33    /// The row's sequence in the conversation's operation log.
34    pub sequence: u64,
35    /// The stored payload bytes, exactly as committed.
36    pub payload: Vec<u8>,
37}
38
39/// Why a page of committed operations could not be read.
40#[derive(Debug, thiserror::Error)]
41#[non_exhaustive]
42pub enum CommittedOperationReadError {
43    /// The page limit is zero or above [`COMMITTED_OPERATION_PAGE_MAX`].
44    #[error("committed operation page limit {limit} is outside 1..={max}")]
45    Limit {
46        /// The limit asked for.
47        limit: usize,
48        /// The largest limit a page accepts.
49        max: usize,
50    },
51    /// The durable store refused a read.
52    #[error(transparent)]
53    Store(#[from] DurabilityError),
54    /// The conversation is retired; its log is never reopened (DUR-027 R4).
55    #[error("committed operations of conversation {conversation_id}: it is retired")]
56    Retired {
57        /// The retired conversation.
58        conversation_id: u64,
59    },
60    /// The routed tail's head or a routed row was refused.
61    #[error("committed operation routed stream refused: {0}")]
62    RoutedStream(#[source] RoutedStreamRefusal),
63    /// A flat row below the native head was absent.
64    #[error("committed operation flat row {sequence} is absent below native head {native_head}")]
65    MissingFlatRow {
66        /// The first sequence the read asked for.
67        sequence: u64,
68        /// The flat stream's exclusive head.
69        native_head: u64,
70    },
71    /// A row answered out of order.
72    #[error("committed operation read expected sequence {expected}, read {actual}")]
73    Sequence {
74        /// The sequence the page required next.
75        expected: u64,
76        /// The sequence the store answered.
77        actual: u64,
78    },
79    /// A sequence in the page does not fit its encoded width.
80    #[error("committed operation sequence overflowed after {from_sequence}")]
81    SequenceOverflow {
82        /// The page's first sequence.
83        from_sequence: u64,
84    },
85}
86
87impl From<RoutedStreamError> for CommittedOperationReadError {
88    fn from(error: RoutedStreamError) -> Self {
89        match error {
90            RoutedStreamError::Store(original) => Self::Store(original),
91            RoutedStreamError::Refusal(refusal) => Self::RoutedStream(refusal),
92        }
93    }
94}
95
96/// Reads at most `limit` committed operation rows of `conversation_id`,
97/// starting at `from_sequence`: the B7 broker read.
98///
99/// A page never crosses the native head, so a page that ends there can be
100/// short; read the next page from the sequence after the last row answered.
101/// An empty page means no committed row exists at `from_sequence`: the log
102/// ends there. Rows come in sequence order with no gap.
103///
104/// # Errors
105/// Returns [`CommittedOperationReadError::Limit`] for a limit of zero or above
106/// [`COMMITTED_OPERATION_PAGE_MAX`], [`CommittedOperationReadError::Store`]
107/// when the store refuses a read,
108/// [`CommittedOperationReadError::RoutedStream`] for a corrupt routed head, a
109/// routed row missing below that head, or a `from_sequence` past it,
110/// [`CommittedOperationReadError::MissingFlatRow`] when no flat row answers
111/// below the native head, and [`CommittedOperationReadError::Sequence`] or
112/// [`CommittedOperationReadError::SequenceOverflow`] when rows answer out of
113/// order.
114pub async fn read_committed_operations(
115    store: &Arc<dyn DurableStore>,
116    conversation_id: u64,
117    from_sequence: u64,
118    limit: usize,
119) -> Result<Vec<CommittedOperation>, CommittedOperationReadError> {
120    if limit == 0 || limit > COMMITTED_OPERATION_PAGE_MAX {
121        return Err(CommittedOperationReadError::Limit {
122            limit,
123            max: COMMITTED_OPERATION_PAGE_MAX,
124        });
125    }
126    let stream = OperationLog::new(Arc::clone(store), conversation_id)
127        .stream_key()
128        .await
129        .map_err(|refused| match refused {
130            StreamKeyRefusal::Store(store) => CommittedOperationReadError::Store(store),
131            StreamKeyRefusal::Retired { conversation_id } => {
132                CommittedOperationReadError::Retired { conversation_id }
133            }
134        })?;
135    let native_head = store.read_stream_next_seq(&stream).await?.unwrap_or(0);
136    let entries = if from_sequence < native_head {
137        read_flat(store.as_ref(), &stream, native_head, from_sequence, limit).await?
138    } else {
139        let route = checkpoint_conversation_route(conversation_id);
140        read_routed(store.as_ref(), &route, &stream, from_sequence, limit).await?
141    };
142    contiguous(entries, from_sequence)
143}
144
145/// Reads flat rows below the native head, never past it.
146async fn read_flat(
147    store: &dyn DurableStore,
148    stream: &str,
149    native_head: u64,
150    from_sequence: u64,
151    limit: usize,
152) -> Result<Vec<StoredEntry>, CommittedOperationReadError> {
153    let remaining = native_head.saturating_sub(from_sequence);
154    let bounded = usize::try_from(remaining).map_or(limit, |remaining| remaining.min(limit));
155    let entries = store.read_from(stream, from_sequence, bounded).await?;
156    if entries.is_empty() {
157        return Err(CommittedOperationReadError::MissingFlatRow {
158            sequence: from_sequence,
159            native_head,
160        });
161    }
162    Ok(entries)
163}
164
165/// Reads routed rows at or after the native head, bounded by the routed
166/// stream's validated head; no routed head means no routed row.
167async fn read_routed(
168    store: &dyn DurableStore,
169    route: &str,
170    stream: &str,
171    from_sequence: u64,
172    limit: usize,
173) -> Result<Vec<StoredEntry>, CommittedOperationReadError> {
174    let Some(head) = routed_stream::read_head(store, route, stream).await? else {
175        return Ok(Vec::new());
176    };
177    Ok(routed_stream::read_page(store, route, stream, &head, from_sequence, limit).await?)
178}
179
180/// Keeps each entry's sequence and payload, refusing any gap or reordering.
181fn contiguous(
182    entries: Vec<StoredEntry>,
183    from_sequence: u64,
184) -> Result<Vec<CommittedOperation>, CommittedOperationReadError> {
185    let mut operations = Vec::with_capacity(entries.len());
186    for (offset, entry) in entries.into_iter().enumerate() {
187        let expected = u64::try_from(offset)
188            .ok()
189            .and_then(|offset| from_sequence.checked_add(offset))
190            .ok_or(CommittedOperationReadError::SequenceOverflow { from_sequence })?;
191        if entry.sequence != expected {
192            return Err(CommittedOperationReadError::Sequence {
193                expected,
194                actual: entry.sequence,
195            });
196        }
197        operations.push(CommittedOperation {
198            sequence: entry.sequence,
199            payload: entry.payload,
200        });
201    }
202    Ok(operations)
203}
liminal_server/server/participant/production/committed_operations.rs

liminal_server/server/participant/production/
committed_operations.rs
