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