liminal-server 0.24.5

//! 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
//! the live and the sweeping conversation ids that registry rows
//! `0..covered_next_seq` leave, in the style of the checkpoint locator codec
//! (`restore_checkpoint_locator.rs`): big-endian, versioned, and closed by a
//! blake3 digest. Codec 2.0 (DUR-027 R4) carries both sets; a 1.0 snapshot,
//! written before retirement, reads as live ids with nothing sweeping.
//!
//! [`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, SCHEMA_VERSION_V1};

/// 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";
/// The codec written: live then sweeping ids, under row schema 2.
const VERSION: (u16, u16) = (2, 0);
/// The codec of a snapshot written before retirement: live ids only, under
/// row schema 1.
const VERSION_V1: (u16, u16) = (1, 0);
/// The snapshot item for rows `0..covered_next_seq` naming `sets`, 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,
    sets: &RegistrySets,
    expected: Option<Hash>,
) -> Result<(RoutedBatchItem, Hash), OperationLogError> {
    let value = RegistrySnapshot::encode(covered_next_seq, sets)
        .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,
    sets: &RegistrySets,
) -> Result<Hash, OperationLogError> {
    let (item, hash) = snapshot_item(covered_next_seq, sets, 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 of a 2.0 snapshot.
const DIGEST_DOMAIN: &[u8] = b"liminal-registry-snapshot-v2";
/// Domain prefixed to every byte before the digest of a 1.0 snapshot.
const DIGEST_DOMAIN_V1: &[u8] = b"liminal-registry-snapshot-v1";
/// Magic, version, row schema, padding, covered head and live 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;
/// The sweeping count's width, after the live run of a 2.0 snapshot.
const SWEEPING_COUNT_BYTES: usize = 8;

/// The registry's two sets: conversations live, and conversations whose
/// retirement committed and whose sweep has not (DUR-027 R4).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct RegistrySets {
    /// Registered and not retired.
    pub(super) live: BTreeSet<u64>,
    /// Retired and not yet swept.
    pub(super) sweeping: BTreeSet<u64>,
}

#[cfg(test)]
impl RegistrySets {
    /// Live ids with nothing sweeping, as every install before retirement.
    pub(super) fn of_live(live: BTreeSet<u64>) -> Self {
        Self {
            live,
            sweeping: BTreeSet::new(),
        }
    }
}

/// 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,
    /// The live and sweeping ids those rows leave.
    pub(super) sets: RegistrySets,
}

impl RegistrySnapshot {
    /// Encodes the canonical 2.0 snapshot bytes: `72 + 8 x (live + sweeping)`
    /// long, the live run then the sweeping count and run.
    ///
    /// # Errors
    ///
    /// Refuses an id in both sets, and sets whose count or length cannot be
    /// framed.
    pub(super) fn encode(
        covered_next_seq: u64,
        sets: &RegistrySets,
    ) -> Result<Vec<u8>, RegistrySnapshotError> {
        if let Some(id) = sets.live.intersection(&sets.sweeping).next() {
            return Err(RegistrySnapshotError::BothSets { id: *id });
        }
        let ids = sets.live.len().saturating_add(sets.sweeping.len());
        let too_large = RegistrySnapshotError::TooLarge { count: ids };
        let live = u64::try_from(sets.live.len()).map_err(|_| too_large.clone())?;
        let sweeping = u64::try_from(sets.sweeping.len()).map_err(|_| too_large.clone())?;
        let length = ids
            .checked_mul(ID_BYTES)
            .and_then(|bytes| bytes.checked_add(EMPTY_BYTES + SWEEPING_COUNT_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(SCHEMA_VERSION);
        bytes.extend_from_slice(&[0; PADDING_BYTES]);
        bytes.extend_from_slice(&covered_next_seq.to_be_bytes());
        bytes.extend_from_slice(&live.to_be_bytes());
        for id in &sets.live {
            bytes.extend_from_slice(&id.to_be_bytes());
        }
        bytes.extend_from_slice(&sweeping.to_be_bytes());
        for id in &sets.sweeping {
            bytes.extend_from_slice(&id.to_be_bytes());
        }
        let sealed = digest(DIGEST_DOMAIN, &bytes);
        bytes.extend_from_slice(&sealed);
        Ok(bytes)
    }

    /// Reads one canonical snapshot of either codec, refusing every
    /// malformed byte by name.
    ///
    /// The checks run in a fixed order: length, magic, version, row schema,
    /// padding, counts against length, id order, an id in both sets, 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]) -> 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)?);
        let (row_schema, domain) = match (major, minor) {
            VERSION => (SCHEMA_VERSION, DIGEST_DOMAIN),
            VERSION_V1 => (SCHEMA_VERSION_V1, DIGEST_DOMAIN_V1),
            _ => return Err(RegistrySnapshotError::Version { major, minor }),
        };
        let [schema] = take::<1>(input, ROW_SCHEMA_AT)?;
        if schema != row_schema {
            return Err(RegistrySnapshotError::RowSchema {
                expected: row_schema,
                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 live_count = u64::from_be_bytes(take(input, COUNT_AT)?);
        let live_end = run_end(HEADER_BYTES, live_count);
        let (sweeping_count, expected) = if (major, minor) == VERSION_V1 {
            (
                0,
                live_end.and_then(|end| end.checked_add(width(DIGEST_BYTES))),
            )
        } else {
            let count = live_end
                .and_then(|end| usize::try_from(end).ok())
                .and_then(|end| take::<SWEEPING_COUNT_BYTES>(input, end).ok())
                .map(u64::from_be_bytes);
            let Some(count) = count else {
                return Err(length(
                    live_end.map_or(u64::MAX, |end| {
                        end.saturating_add(width(SWEEPING_COUNT_BYTES))
                    }),
                    input,
                ));
            };
            let end = live_end
                .and_then(|end| end.checked_add(width(SWEEPING_COUNT_BYTES)))
                .and_then(|start| run_end_at(start, count));
            (
                count,
                end.and_then(|end| end.checked_add(width(DIGEST_BYTES))),
            )
        };
        let expected = expected.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 live_bytes =
            usize::try_from(live_count.saturating_mul(width(ID_BYTES))).unwrap_or(usize::MAX);
        let live_run = body
            .get(HEADER_BYTES..HEADER_BYTES.saturating_add(live_bytes))
            .unwrap_or_default();
        let live = ascending_ids(live_run)?;
        let sweeping = if sweeping_count == 0 {
            BTreeSet::new()
        } else {
            let start = HEADER_BYTES
                .saturating_add(live_bytes)
                .saturating_add(SWEEPING_COUNT_BYTES);
            ascending_ids(body.get(start..).unwrap_or_default())?
        };
        if let Some(id) = live.intersection(&sweeping).next() {
            return Err(RegistrySnapshotError::BothSets { id: *id });
        }
        if digest(domain, body).as_slice() != stored {
            return Err(RegistrySnapshotError::Digest);
        }
        Ok(Self {
            covered_next_seq,
            sets: RegistrySets { live, sweeping },
        })
    }
}

/// Where a run of `count` ids starting at byte `start` ends.
fn run_end(start: usize, count: u64) -> Option<u64> {
    run_end_at(width(start), count)
}

fn run_end_at(start: u64, count: u64) -> Option<u64> {
    count
        .checked_mul(width(ID_BYTES))
        .and_then(|bytes| bytes.checked_add(start))
}

/// The digest over the snapshot domain and every byte before the digest.
fn digest(domain: &[u8], body: &[u8]) -> [u8; DIGEST_BYTES] {
    let mut hash = blake3::Hasher::new();
    hash.update(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 the length its counts frame.
    #[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,
    },
    /// An id stands in both the live and the sweeping run.
    #[error("registry snapshot at {at}: id {id} is both live and sweeping", at = SnapshotAt)]
    BothSets {
        /// The id named twice.
        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,
    },
}