liminal-server 0.23.1

//! Durable snapshot of the conversation registry and its canonical codec.
//!
//! The snapshot is one routed value on the registry's own route
//! ([`REGISTRY_STREAM_KEY`]) under the local key [`SNAPSHOT_KEY`]. It names
//! every distinct conversation id carried by registry rows
//! `0..covered_next_seq`, in the style of the checkpoint locator codec
//! (`restore_checkpoint_locator.rs`): big-endian, versioned, and closed by a
//! blake3 digest.
//!
//! [`write_registry_snapshot`] is the one writer of a snapshot outside a
//! registration: the migration of an install written before the snapshot
//! existed calls it, and so does the boot-history fixture. A registration
//! that crosses `CHECKPOINT_EVERY` stages [`snapshot_item`] into its own
//! batch instead (DUR-019 R2).

use std::collections::BTreeSet;
use std::fmt;

use haematite::tree::Hash;
use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, RoutedBatchItem};

use super::log::OperationLogError;
use super::operation_batch::OperationBatch;
use super::registry::{REGISTRY_STREAM_KEY, SCHEMA_VERSION};

/// The snapshot's local key, as text for refusals.
const SNAPSHOT_KEY_TEXT: &str = "registry/snapshot/v1";
/// Local key of the snapshot under the registry route.
pub(super) const SNAPSHOT_KEY: &[u8] = SNAPSHOT_KEY_TEXT.as_bytes();
const MAGIC: [u8; 4] = *b"LSRS";
const VERSION: (u16, u16) = (1, 0);
/// The snapshot item for rows `0..covered_next_seq` naming `ids`, staged
/// with `expected` as its compare-and-swap witness, and the hash of its
/// bytes.
///
/// # Errors
///
/// Refuses an id set too large to frame.
pub(super) fn snapshot_item(
    covered_next_seq: u64,
    ids: &BTreeSet<u64>,
    expected: Option<Hash>,
) -> Result<(RoutedBatchItem, Hash), OperationLogError> {
    let value = RegistrySnapshot::encode(SCHEMA_VERSION, covered_next_seq, ids)
        .map_err(OperationLogError::RegistrySnapshot)?;
    let hash = Hash::of(&value);
    Ok((
        RoutedBatchItem {
            key: SNAPSHOT_KEY.to_vec(),
            expected,
            value,
            ttl: None,
        },
        hash,
    ))
}

/// Writes the snapshot alone, as one batch on the registry route, and
/// answers the hash of the bytes it wrote.
///
/// The batch stages no row and no routed head, so it changes no registry
/// row: it is the migration's write (expected `None`, proved by the read
/// that found no snapshot) and the boot-history fixture's.
///
/// # Errors
///
/// Refuses an unframeable id set, and returns the store's refusal or
/// unknown outcome unchanged.
pub(super) fn write_registry_snapshot(
    store: &dyn DurableStore,
    expected: Option<Hash>,
    covered_next_seq: u64,
    ids: &BTreeSet<u64>,
) -> Result<Hash, OperationLogError> {
    let (item, hash) = snapshot_item(covered_next_seq, ids, expected)?;
    let mut batch = OperationBatch::new(REGISTRY_STREAM_KEY);
    batch.stage(item)?;
    block_on(batch.commit(store, || {}))??;
    Ok(hash)
}

/// Domain prefixed to every byte before the digest.
const DIGEST_DOMAIN: &[u8] = b"liminal-registry-snapshot-v1";
/// Magic, version, row schema, padding, covered head and id count.
const HEADER_BYTES: usize = 32;
const DIGEST_BYTES: usize = 32;
/// The length of a snapshot that names no id.
const EMPTY_BYTES: usize = HEADER_BYTES + DIGEST_BYTES;
const ID_BYTES: usize = 8;
const PADDING_BYTES: usize = 7;
const ROW_SCHEMA_AT: usize = 8;
const PADDING_AT: usize = 9;
const COVERED_AT: usize = 16;
const COUNT_AT: usize = 24;

/// One decoded registry snapshot.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct RegistrySnapshot {
    /// Exclusive: the snapshot covers registry rows `0..covered_next_seq`.
    pub(super) covered_next_seq: u64,
    /// Every distinct conversation id those rows name.
    pub(super) ids: BTreeSet<u64>,
}

impl RegistrySnapshot {
    /// Encodes the canonical snapshot bytes: `64 + 8 x count` long.
    ///
    /// # Errors
    ///
    /// Refuses an id set whose count or length cannot be framed.
    pub(super) fn encode(
        row_schema_version: u8,
        covered_next_seq: u64,
        ids: &BTreeSet<u64>,
    ) -> Result<Vec<u8>, RegistrySnapshotError> {
        let too_large = RegistrySnapshotError::TooLarge { count: ids.len() };
        let count = u64::try_from(ids.len()).map_err(|_| too_large.clone())?;
        let length = ids
            .len()
            .checked_mul(ID_BYTES)
            .and_then(|bytes| bytes.checked_add(EMPTY_BYTES))
            .ok_or(too_large)?;
        let mut bytes = Vec::with_capacity(length);
        bytes.extend_from_slice(&MAGIC);
        bytes.extend_from_slice(&VERSION.0.to_be_bytes());
        bytes.extend_from_slice(&VERSION.1.to_be_bytes());
        bytes.push(row_schema_version);
        bytes.extend_from_slice(&[0; PADDING_BYTES]);
        bytes.extend_from_slice(&covered_next_seq.to_be_bytes());
        bytes.extend_from_slice(&count.to_be_bytes());
        for id in ids {
            bytes.extend_from_slice(&id.to_be_bytes());
        }
        let sealed = digest(&bytes);
        bytes.extend_from_slice(&sealed);
        Ok(bytes)
    }

    /// Reads one canonical snapshot, refusing every malformed byte by name.
    ///
    /// The checks run in a fixed order: length, magic, version, row schema,
    /// padding, count against length, id order, then the digest. Each refusal
    /// is final; a caller never falls back to walking the registry stream.
    ///
    /// # Errors
    ///
    /// Returns the [`RegistrySnapshotError`] variant naming the first check
    /// the bytes fail.
    pub(super) fn decode(
        input: &[u8],
        row_schema_version: u8,
    ) -> Result<Self, RegistrySnapshotError> {
        if input.len() < EMPTY_BYTES {
            return Err(length(width(EMPTY_BYTES), input));
        }
        if input.get(..MAGIC.len()) != Some(MAGIC.as_slice()) {
            return Err(RegistrySnapshotError::Magic);
        }
        let major = u16::from_be_bytes(take(input, 4)?);
        let minor = u16::from_be_bytes(take(input, 6)?);
        if (major, minor) != VERSION {
            return Err(RegistrySnapshotError::Version { major, minor });
        }
        let [schema] = take::<1>(input, ROW_SCHEMA_AT)?;
        if schema != row_schema_version {
            return Err(RegistrySnapshotError::RowSchema {
                expected: row_schema_version,
                actual: schema,
            });
        }
        if take::<PADDING_BYTES>(input, PADDING_AT)? != [0; PADDING_BYTES] {
            return Err(RegistrySnapshotError::Padding);
        }
        let covered_next_seq = u64::from_be_bytes(take(input, COVERED_AT)?);
        let count = u64::from_be_bytes(take(input, COUNT_AT)?);
        let expected = count
            .checked_mul(width(ID_BYTES))
            .and_then(|bytes| bytes.checked_add(width(EMPTY_BYTES)))
            .unwrap_or(u64::MAX);
        if width(input.len()) != expected {
            return Err(length(expected, input));
        }
        let (body, stored) = input
            .split_at_checked(input.len().saturating_sub(DIGEST_BYTES))
            .ok_or_else(|| length(expected, input))?;
        let ids = ascending_ids(body.get(HEADER_BYTES..).unwrap_or_default())?;
        if digest(body).as_slice() != stored {
            return Err(RegistrySnapshotError::Digest);
        }
        Ok(Self {
            covered_next_seq,
            ids,
        })
    }
}

/// The digest over the snapshot domain and every byte before the digest.
fn digest(body: &[u8]) -> [u8; DIGEST_BYTES] {
    let mut hash = blake3::Hasher::new();
    hash.update(DIGEST_DOMAIN);
    hash.update(body);
    *hash.finalize().as_bytes()
}

/// Decodes the id run, refusing any id not strictly above its predecessor.
fn ascending_ids(run: &[u8]) -> Result<BTreeSet<u64>, RegistrySnapshotError> {
    let mut ids = BTreeSet::new();
    let mut previous: Option<u64> = None;
    for (index, chunk) in run.chunks_exact(ID_BYTES).enumerate() {
        let word: [u8; ID_BYTES] = chunk
            .try_into()
            .map_err(|_| RegistrySnapshotError::Length {
                expected: width(ID_BYTES),
                actual: width(chunk.len()),
            })?;
        let id = u64::from_be_bytes(word);
        if let Some(previous) = previous.filter(|previous| id <= *previous) {
            return Err(RegistrySnapshotError::Order {
                index,
                previous,
                id,
            });
        }
        previous = Some(id);
        ids.insert(id);
    }
    Ok(ids)
}

fn length(expected: u64, input: &[u8]) -> RegistrySnapshotError {
    RegistrySnapshotError::Length {
        expected,
        actual: width(input.len()),
    }
}

/// A byte count as a frame width; no supported target has a wider `usize`.
fn width(bytes: usize) -> u64 {
    u64::try_from(bytes).unwrap_or(u64::MAX)
}

fn take<const N: usize>(input: &[u8], start: usize) -> Result<[u8; N], RegistrySnapshotError> {
    start
        .checked_add(N)
        .and_then(|end| input.get(start..end))
        .and_then(|bytes| <[u8; N]>::try_from(bytes).ok())
        .ok_or_else(|| length(width(EMPTY_BYTES), input))
}

/// Where every snapshot refusal points: the registry route and the local key.
#[derive(Clone, Copy, Debug)]
struct SnapshotAt;

impl fmt::Display for SnapshotAt {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            formatter,
            "route {REGISTRY_STREAM_KEY} key {SNAPSHOT_KEY_TEXT}"
        )
    }
}

/// Named refusal of a registry snapshot; restore never walks around one.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub(super) enum RegistrySnapshotError {
    /// Shorter than 64 bytes, or not `64 + 8 x count`.
    #[error("registry snapshot at {at}: length differs: expected {expected}, found {actual}", at = SnapshotAt)]
    Length {
        /// The length the header requires (64 when the header itself is short).
        expected: u64,
        /// The stored length.
        actual: u64,
    },
    /// The magic is not `LSRS`.
    #[error("registry snapshot at {at}: magic differs", at = SnapshotAt)]
    Magic,
    /// The codec version is unknown.
    #[error("registry snapshot at {at}: codec {major}.{minor} is unsupported", at = SnapshotAt)]
    Version {
        /// Declared major version.
        major: u16,
        /// Declared minor version.
        minor: u16,
    },
    /// The registry row schema the snapshot was cut under is unknown.
    #[error("registry snapshot at {at}: row schema {actual} differs from {expected}", at = SnapshotAt)]
    RowSchema {
        /// The registry's row schema version.
        expected: u8,
        /// The stored row schema version.
        actual: u8,
    },
    /// A padding byte is not zero.
    #[error("registry snapshot at {at}: padding is not zero", at = SnapshotAt)]
    Padding,
    /// An id is not strictly above the one before it.
    #[error(
        "registry snapshot at {at}: id {id} at index {index} is not above {previous}",
        at = SnapshotAt
    )]
    Order {
        /// Position of the refused id in the run.
        index: usize,
        /// The id before it.
        previous: u64,
        /// The refused id.
        id: u64,
    },
    /// The bytes differ from their digest.
    #[error("registry snapshot at {at}: digest differs", at = SnapshotAt)]
    Digest,
    /// The snapshot covers rows past the registry head.
    #[error(
        "registry snapshot at {at}: covers {covered_next_seq} rows past the registry head {head}",
        at = SnapshotAt
    )]
    AheadOfStream {
        /// The snapshot's exclusive covered head.
        covered_next_seq: u64,
        /// The registry head read beside it.
        head: u64,
    },
    /// The tail behind the snapshot is 256 rows or more: every crossing
    /// carries its snapshot in its own batch, so only damage or a foreign
    /// writer leaves one.
    #[error(
        "registry snapshot at {at}: tail from {covered_next_seq} to head {head} reaches the 256-row bound",
        at = SnapshotAt
    )]
    TailBeyondBound {
        /// The snapshot's exclusive covered head.
        covered_next_seq: u64,
        /// The registry head read beside it.
        head: u64,
    },
    /// The rows read did not end at the head that bounds them.
    #[error(
        "registry snapshot at {at}: rows end at {tail_end}, not at head {expected_head}",
        at = SnapshotAt
    )]
    TailMismatch {
        /// The head the rows must reach.
        expected_head: u64,
        /// Where the rows ended.
        tail_end: u64,
    },
    /// A routed registry head stands below the flat native head.
    #[error(
        "registry snapshot at {at}: routed head {routed} stands below the native head {native}",
        at = SnapshotAt
    )]
    RoutedHeadBehindNative {
        /// The routed head.
        routed: u64,
        /// The native flat head.
        native: u64,
    },
    /// A recovered crossing batch landed its row but its snapshot does not
    /// cover the new head.
    #[error(
        "registry snapshot at {at}: the landed crossing to {head} holds a snapshot covering {covered_next_seq:?}",
        at = SnapshotAt
    )]
    CrossingUnmatched {
        /// The head the landed batch advanced to.
        head: u64,
        /// The stored snapshot's covered head, or `None` when absent.
        covered_next_seq: Option<u64>,
    },
    /// An id set too large to frame was offered for encoding.
    #[error("registry snapshot at {at}: {count} ids exceed the frame", at = SnapshotAt)]
    TooLarge {
        /// The offered id count.
        count: usize,
    },
}