liminal_server/server/participant/production/

committed_operations.rs

//! 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)
}