liminal_server/server/participant/production/
routed_stream.rs

1//! Validated heads and bounded point reads for routed stream rows.
2
3use std::collections::TryReserveError;
4
5use haematite::tree::Hash;
6use liminal::durability::{DurabilityError, DurableStore, StoredEntry};
7
8use super::routed_stream_codec::{self, CodecError};
9
10#[derive(Debug, thiserror::Error)]
11#[non_exhaustive]
12pub enum RoutedStreamRefusal {
13    #[error("checkpoint framing refused: {0}")]
14    Checkpoint(#[source] Box<super::restore_checkpoint::CheckpointError>),
15    #[error("checkpoint locator refused: {0}")]
16    CheckpointLocator(#[source] super::restore_checkpoint_locator::CheckpointLocatorError),
17    #[error("checkpoint belongs to another operation route")]
18    CheckpointRouteMismatch,
19    #[error("checkpoint positions regress behind its durable locator")]
20    CheckpointPositionRegression,
21    #[error("checkpoint content-addressed body differs from stored bytes")]
22    CheckpointBodyMismatch,
23    #[error("routed stream framing refused: {0}")]
24    Codec(#[source] CodecError),
25    #[error("routed stream append is empty")]
26    EmptyAppend,
27    #[error("routed stream sequence exceeds its encoded width")]
28    SequenceOverflow,
29    #[error("routed stream head mismatch: expected {expected}, actual {actual}")]
30    HeadMismatch { expected: u64, actual: u64 },
31    #[error("routed stream key collides with another staged row")]
32    KeyCollision,
33    #[error("operation batch staging was already refused")]
34    BatchRefused,
35    #[error("routed stream read limit is zero")]
36    InvalidLimit,
37    #[error("routed stream offset {offset} exceeds head {head}")]
38    OffsetBeyondHead { offset: u64, head: u64 },
39    #[error("routed stream head belongs to another route or stream")]
40    HeadIdentityMismatch,
41    #[error("routed stream {stream:?} has no validated durable head")]
42    MissingHead { stream: Vec<u8> },
43    #[error("routed stream {stream:?} is missing row {sequence} below its validated head")]
44    MissingRow { stream: Vec<u8>, sequence: u64 },
45    #[error("routed stream page allocation refused")]
46    Allocation(#[source] TryReserveError),
47}
48
49#[derive(Debug, thiserror::Error)]
50pub(super) enum RoutedStreamError {
51    #[error(transparent)]
52    Store(#[from] DurabilityError),
53    #[error(transparent)]
54    Refusal(#[from] RoutedStreamRefusal),
55}
56
57impl From<CodecError> for RoutedStreamError {
58    fn from(error: CodecError) -> Self {
59        Self::Refusal(RoutedStreamRefusal::Codec(error))
60    }
61}
62
63/// The counter and original hash of an existing, validated head.
64#[derive(Debug)]
65pub(super) struct RoutedStreamHead {
66    route: String,
67    stream: String,
68    pub(super) next_sequence: u64,
69    pub(super) expected: Hash,
70}
71
72/// Reads one exact head without interpreting absence as recovery authority.
73///
74/// # Errors
75/// Preserves a store failure and refuses malformed head framing.
76pub(super) async fn read_head(
77    store: &dyn DurableStore,
78    route: &str,
79    stream: &str,
80) -> Result<Option<RoutedStreamHead>, RoutedStreamError> {
81    let key = routed_stream_codec::head_key(stream)?;
82    store
83        .get_routed(route, &key)
84        .await?
85        .map(|bytes| {
86            Ok(RoutedStreamHead {
87                route: route.to_owned(),
88                stream: stream.to_owned(),
89                next_sequence: routed_stream_codec::decode_head(&bytes)?,
90                expected: Hash::of(&bytes),
91            })
92        })
93        .transpose()
94}
95
96/// Reads only the requested rows bounded by the validated head.
97///
98/// # Errors
99/// Refuses a wrong head association, invalid bounds, missing row or corrupt row.
100pub(super) async fn read_page(
101    store: &dyn DurableStore,
102    route: &str,
103    stream: &str,
104    head: &RoutedStreamHead,
105    offset: u64,
106    limit: usize,
107) -> Result<Vec<StoredEntry>, RoutedStreamError> {
108    if head.route != route || head.stream != stream {
109        return Err(RoutedStreamRefusal::HeadIdentityMismatch.into());
110    }
111    let (sequences, mut rows) = prepare_page(head.next_sequence, offset, limit)?;
112    for sequence in sequences {
113        let key = routed_stream_codec::row_key(stream, sequence)?;
114        let bytes = store.get_routed(route, &key).await?.ok_or_else(|| {
115            RoutedStreamRefusal::MissingRow {
116                stream: stream.as_bytes().to_vec(),
117                sequence,
118            }
119        })?;
120        rows.push(routed_stream_codec::decode_row(&bytes, sequence)?);
121    }
122    Ok(rows)
123}
124
125/// Validates a bounded page and reserves only its selected rows.
126///
127/// # Errors
128/// Refuses invalid bounds, unrepresentable counts and allocation failures.
129pub(super) fn prepare_page(
130    next_sequence: u64,
131    offset: u64,
132    limit: usize,
133) -> Result<(std::ops::Range<u64>, Vec<StoredEntry>), RoutedStreamRefusal> {
134    if limit == 0 {
135        return Err(RoutedStreamRefusal::InvalidLimit);
136    }
137    let remaining =
138        next_sequence
139            .checked_sub(offset)
140            .ok_or(RoutedStreamRefusal::OffsetBeyondHead {
141                offset,
142                head: next_sequence,
143            })?;
144    let limit = u64::try_from(limit).map_err(|_| RoutedStreamRefusal::SequenceOverflow)?;
145    let count = remaining.min(limit);
146    let capacity = usize::try_from(count).map_err(|_| RoutedStreamRefusal::SequenceOverflow)?;
147    let mut rows = Vec::new();
148    rows.try_reserve_exact(capacity)
149        .map_err(RoutedStreamRefusal::Allocation)?;
150    let end = offset
151        .checked_add(count)
152        .ok_or(RoutedStreamRefusal::SequenceOverflow)?;
153    Ok((offset..end, rows))
154}
155
156#[cfg(test)]
157mod tests;
158
159#[cfg(test)]
160mod refusal_tests;
liminal_server/server/participant/production/routed_stream.rs

liminal_server/server/participant/production/
routed_stream.rs
