liminal-server 0.20.1

//! Validated heads and bounded point reads for routed stream rows.

use std::collections::TryReserveError;

use haematite::tree::Hash;
use liminal::durability::{DurabilityError, DurableStore, StoredEntry};

use super::routed_stream_codec::{self, CodecError};

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum RoutedStreamRefusal {
    #[error("checkpoint framing refused: {0}")]
    Checkpoint(#[source] Box<super::restore_checkpoint::CheckpointError>),
    #[error("checkpoint locator refused: {0}")]
    CheckpointLocator(#[source] super::restore_checkpoint_locator::CheckpointLocatorError),
    #[error("checkpoint belongs to another operation route")]
    CheckpointRouteMismatch,
    #[error("checkpoint positions regress behind its durable locator")]
    CheckpointPositionRegression,
    #[error("checkpoint content-addressed body differs from stored bytes")]
    CheckpointBodyMismatch,
    #[error("routed stream framing refused: {0}")]
    Codec(#[source] CodecError),
    #[error("routed stream append is empty")]
    EmptyAppend,
    #[error("routed stream sequence exceeds its encoded width")]
    SequenceOverflow,
    #[error("routed stream head mismatch: expected {expected}, actual {actual}")]
    HeadMismatch { expected: u64, actual: u64 },
    #[error("routed stream key collides with another staged row")]
    KeyCollision,
    #[error("operation batch staging was already refused")]
    BatchRefused,
    #[error("routed stream read limit is zero")]
    InvalidLimit,
    #[error("routed stream offset {offset} exceeds head {head}")]
    OffsetBeyondHead { offset: u64, head: u64 },
    #[error("routed stream head belongs to another route or stream")]
    HeadIdentityMismatch,
    #[error("routed stream {stream:?} has no validated durable head")]
    MissingHead { stream: Vec<u8> },
    #[error("routed stream {stream:?} is missing row {sequence} below its validated head")]
    MissingRow { stream: Vec<u8>, sequence: u64 },
    #[error("routed stream page allocation refused")]
    Allocation(#[source] TryReserveError),
}

#[derive(Debug, thiserror::Error)]
pub(super) enum RoutedStreamError {
    #[error(transparent)]
    Store(#[from] DurabilityError),
    #[error(transparent)]
    Refusal(#[from] RoutedStreamRefusal),
}

impl From<CodecError> for RoutedStreamError {
    fn from(error: CodecError) -> Self {
        Self::Refusal(RoutedStreamRefusal::Codec(error))
    }
}

/// The counter and original hash of an existing, validated head.
#[derive(Debug)]
pub(super) struct RoutedStreamHead {
    route: String,
    stream: String,
    pub(super) next_sequence: u64,
    pub(super) expected: Hash,
}

/// Reads one exact head without interpreting absence as recovery authority.
///
/// # Errors
/// Preserves a store failure and refuses malformed head framing.
pub(super) async fn read_head(
    store: &dyn DurableStore,
    route: &str,
    stream: &str,
) -> Result<Option<RoutedStreamHead>, RoutedStreamError> {
    let key = routed_stream_codec::head_key(stream)?;
    store
        .get_routed(route, &key)
        .await?
        .map(|bytes| {
            Ok(RoutedStreamHead {
                route: route.to_owned(),
                stream: stream.to_owned(),
                next_sequence: routed_stream_codec::decode_head(&bytes)?,
                expected: Hash::of(&bytes),
            })
        })
        .transpose()
}

/// Reads only the requested rows bounded by the validated head.
///
/// # Errors
/// Refuses a wrong head association, invalid bounds, missing row or corrupt row.
pub(super) async fn read_page(
    store: &dyn DurableStore,
    route: &str,
    stream: &str,
    head: &RoutedStreamHead,
    offset: u64,
    limit: usize,
) -> Result<Vec<StoredEntry>, RoutedStreamError> {
    if head.route != route || head.stream != stream {
        return Err(RoutedStreamRefusal::HeadIdentityMismatch.into());
    }
    let (sequences, mut rows) = prepare_page(head.next_sequence, offset, limit)?;
    for sequence in sequences {
        let key = routed_stream_codec::row_key(stream, sequence)?;
        let bytes = store.get_routed(route, &key).await?.ok_or_else(|| {
            RoutedStreamRefusal::MissingRow {
                stream: stream.as_bytes().to_vec(),
                sequence,
            }
        })?;
        rows.push(routed_stream_codec::decode_row(&bytes, sequence)?);
    }
    Ok(rows)
}

/// Validates a bounded page and reserves only its selected rows.
///
/// # Errors
/// Refuses invalid bounds, unrepresentable counts and allocation failures.
pub(super) fn prepare_page(
    next_sequence: u64,
    offset: u64,
    limit: usize,
) -> Result<(std::ops::Range<u64>, Vec<StoredEntry>), RoutedStreamRefusal> {
    if limit == 0 {
        return Err(RoutedStreamRefusal::InvalidLimit);
    }
    let remaining =
        next_sequence
            .checked_sub(offset)
            .ok_or(RoutedStreamRefusal::OffsetBeyondHead {
                offset,
                head: next_sequence,
            })?;
    let limit = u64::try_from(limit).map_err(|_| RoutedStreamRefusal::SequenceOverflow)?;
    let count = remaining.min(limit);
    let capacity = usize::try_from(count).map_err(|_| RoutedStreamRefusal::SequenceOverflow)?;
    let mut rows = Vec::new();
    rows.try_reserve_exact(capacity)
        .map_err(RoutedStreamRefusal::Allocation)?;
    let end = offset
        .checked_add(count)
        .ok_or(RoutedStreamRefusal::SequenceOverflow)?;
    Ok((offset..end, rows))
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod refusal_tests;