liminal-server 0.23.1

//! Registry snapshot pins (DUR-019): one canonical encoding with a named
//! refusal for every malformed byte (R1); one routed batch per registration,
//! carrying the snapshot at a crossing (R2); restore from the snapshot and a
//! bounded tail, and the one-time migration of an old install (R3). The R2
//! pins stand in `tests_registry_snapshot_cut`, the R3 pins in
//! `tests_registry_snapshot_restore`, and the store fixtures they share in
//! `tests_registry_snapshot_fixture`.

use std::collections::BTreeSet;
use std::error::Error;
use std::sync::Arc;

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

use super::log::OperationLogError;
use super::registry::{ConversationRegistry, REGISTRY_STREAM_KEY};
use super::registry_snapshot::{RegistrySnapshot, RegistrySnapshotError, SNAPSHOT_KEY};
use super::routed_stream::{read_head, read_page};
use super::tests_registry_snapshot_fixture::{
    KEY, ROW_SCHEMA, Recording, TestResult, flat_ids, memory,
};

/// `covered_next_seq` 1,000 and the single id 7, byte for byte. The digest is
/// blake3 over `b"liminal-registry-snapshot-v1"` and the 40 bytes before it,
/// computed outside the codec with `b3sum`.
const VECTOR: [u8; 72] = [
    b'L', b'S', b'R', b'S', // magic
    0x00, 0x01, 0x00, 0x00, // version 1.0
    0x01, // row schema
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // padding
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0xe8, // covered_next_seq 1,000
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, // count 1
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07, // id 7
    0x56, 0x5f, 0x3e, 0x34, 0xcf, 0x61, 0xf7, 0xb4, 0x44, 0xf8, 0x34, 0x6a, 0x77, 0x4c, 0xa9, 0x47,
    0x6b, 0xad, 0xb8, 0x7f, 0x98, 0xf2, 0x7a, 0xd3, 0x8e, 0x5d, 0x9c, 0x28, 0xb7, 0x33, 0x79,
    0x22, // digest
];

fn flipped(bytes: &[u8], at: usize, mask: u8) -> Result<Vec<u8>, Box<dyn Error>> {
    let mut changed = bytes.to_vec();
    *changed
        .get_mut(at)
        .ok_or("flip position past the encoding")? ^= mask;
    Ok(changed)
}

fn refusal(bytes: &[u8]) -> Result<RegistrySnapshotError, Box<dyn Error>> {
    match RegistrySnapshot::decode(bytes, ROW_SCHEMA) {
        Ok(snapshot) => Err(format!("malformed snapshot decoded: {snapshot:?}").into()),
        Err(refusal) => Ok(refusal),
    }
}

#[test]
fn registry_snapshot_roundtrips_zero_one_and_a_thousand_ids() -> Result<(), Box<dyn Error>> {
    let thousand: BTreeSet<u64> = (0..999_u64).map(|id| id * 3).chain([u64::MAX]).collect();
    assert_eq!(thousand.len(), 1_000);
    for (covered_next_seq, ids) in [
        (0, BTreeSet::new()),
        (1, BTreeSet::from([7])),
        (100_000, thousand),
    ] {
        let bytes = RegistrySnapshot::encode(ROW_SCHEMA, covered_next_seq, &ids)?;
        assert_eq!(bytes.len(), 64 + 8 * ids.len());
        let decoded = RegistrySnapshot::decode(&bytes, ROW_SCHEMA)?;
        assert_eq!(decoded.covered_next_seq, covered_next_seq);
        assert_eq!(decoded.ids, ids);
    }
    Ok(())
}

#[test]
fn registry_snapshot_encodes_the_pinned_vector() -> Result<(), Box<dyn Error>> {
    let bytes = RegistrySnapshot::encode(ROW_SCHEMA, 1_000, &BTreeSet::from([7]))?;
    assert_eq!(bytes.as_slice(), VECTOR.as_slice());
    assert_eq!(
        RegistrySnapshot::decode(&VECTOR, ROW_SCHEMA)?,
        RegistrySnapshot {
            covered_next_seq: 1_000,
            ids: BTreeSet::from([7]),
        }
    );
    assert_eq!(SNAPSHOT_KEY, b"registry/snapshot/v1");
    Ok(())
}

#[test]
fn registry_snapshot_refuses_magic_version_schema_and_padding_by_name() -> Result<(), Box<dyn Error>>
{
    assert_eq!(
        refusal(&flipped(&VECTOR, 0, 1)?)?,
        RegistrySnapshotError::Magic
    );
    assert_eq!(
        refusal(&flipped(&VECTOR, 7, 1)?)?,
        RegistrySnapshotError::Version { major: 1, minor: 1 }
    );
    assert_eq!(
        refusal(&flipped(&VECTOR, 8, 1)?)?,
        RegistrySnapshotError::RowSchema {
            expected: 1,
            actual: 0
        }
    );
    for at in 9..16 {
        assert_eq!(
            refusal(&flipped(&VECTOR, at, 0x80)?)?,
            RegistrySnapshotError::Padding
        );
    }
    Ok(())
}

#[test]
fn registry_snapshot_refuses_a_count_or_length_that_disagree() -> Result<(), Box<dyn Error>> {
    assert_eq!(
        refusal(&flipped(&VECTOR, 31, 1)?)?,
        RegistrySnapshotError::Length {
            expected: 64,
            actual: 72
        }
    );
    assert_eq!(
        refusal(&VECTOR[..71])?,
        RegistrySnapshotError::Length {
            expected: 72,
            actual: 71
        }
    );
    let mut extended = VECTOR.to_vec();
    extended.push(0);
    assert_eq!(
        refusal(&extended)?,
        RegistrySnapshotError::Length {
            expected: 72,
            actual: 73
        }
    );
    assert_eq!(
        refusal(&VECTOR[..63])?,
        RegistrySnapshotError::Length {
            expected: 64,
            actual: 63
        }
    );
    Ok(())
}

/// The order check runs before the digest, so one flipped byte that breaks
/// the order is named as order; a decoder without the order check meets the
/// digest instead and fails this pin.
#[test]
fn registry_snapshot_refuses_ids_out_of_strictly_ascending_order() -> Result<(), Box<dyn Error>> {
    let bytes = RegistrySnapshot::encode(ROW_SCHEMA, 2, &BTreeSet::from([3, 9]))?;
    // The second id's low byte: 9 ^ 0x0a = 3, equal to the first id.
    assert_eq!(
        refusal(&flipped(&bytes, 47, 0x0a)?)?,
        RegistrySnapshotError::Order {
            index: 1,
            previous: 3,
            id: 3
        }
    );
    // The first id's high byte: it rises above the second.
    assert_eq!(
        refusal(&flipped(&bytes, 32, 0x80)?)?,
        RegistrySnapshotError::Order {
            index: 1,
            previous: 0x8000_0000_0000_0003,
            id: 9
        }
    );
    Ok(())
}

/// A decoder that skips the digest accepts each of these and fails this pin.
#[test]
fn registry_snapshot_refuses_a_digest_mismatch() -> Result<(), Box<dyn Error>> {
    let bytes = RegistrySnapshot::encode(ROW_SCHEMA, 2, &BTreeSet::from([3, 9]))?;
    // The covered head, an id that stays in order, and the digest itself.
    for at in [23, 39, 48, 79] {
        assert_eq!(
            refusal(&flipped(&bytes, at, 1)?)?,
            RegistrySnapshotError::Digest
        );
    }
    Ok(())
}

#[test]
fn registry_snapshot_refuses_every_truncation_and_single_byte_change() -> Result<(), Box<dyn Error>>
{
    for end in 0..VECTOR.len() {
        assert!(RegistrySnapshot::decode(&VECTOR[..end], ROW_SCHEMA).is_err());
    }
    for at in 0..VECTOR.len() {
        for bit in 0..8 {
            let changed = flipped(&VECTOR, at, 1 << bit)?;
            assert!(
                RegistrySnapshot::decode(&changed, ROW_SCHEMA).is_err(),
                "byte {at} bit {bit} changed and still decoded"
            );
        }
    }
    Ok(())
}

#[test]
fn registry_snapshot_refusals_name_the_route_and_key_through_the_log_error()
-> Result<(), Box<dyn Error>> {
    let refusals = [
        refusal(&VECTOR[..63])?,
        refusal(&flipped(&VECTOR, 0, 1)?)?,
        refusal(&flipped(&VECTOR, 7, 1)?)?,
        refusal(&flipped(&VECTOR, 8, 1)?)?,
        refusal(&flipped(&VECTOR, 9, 1)?)?,
        refusal(&flipped(&VECTOR, 71, 1)?)?,
    ];
    for refused in refusals {
        let text = OperationLogError::RegistrySnapshot(refused).to_string();
        assert!(text.contains(REGISTRY_STREAM_KEY), "{text}");
        assert!(text.contains("registry/snapshot/v1"), "{text}");
    }
    Ok(())
}

/// Every registry row wherever it stands: flat below the native head, then
/// routed up to the routed head.
pub(super) fn registry_rows(store: &dyn DurableStore) -> TestResult<Vec<StoredEntry>> {
    let native = block_on(store.read_stream_next_seq(KEY))??.unwrap_or(0);
    let mut rows: Vec<StoredEntry> = Vec::new();
    while u64::try_from(rows.len())? < native {
        let page = block_on(store.read_from(KEY, u64::try_from(rows.len())?, 64))??;
        if page.is_empty() {
            break;
        }
        rows.extend(page);
    }
    if let Some(head) = block_on(read_head(store, KEY, KEY))?? {
        while u64::try_from(rows.len())? < head.next_sequence {
            let page = block_on(read_page(
                store,
                KEY,
                KEY,
                &head,
                u64::try_from(rows.len())?,
                64,
            ))??;
            if page.is_empty() {
                break;
            }
            rows.extend(page);
        }
    }
    Ok(rows)
}

// ---------------------------------------------------------------------------
// R1: a refused snapshot is never walked around.
// ---------------------------------------------------------------------------

#[test]
fn registry_snapshot_digest_refusal_reaches_restore_before_any_row_read() -> TestResult {
    let memory = memory();
    flat_ids(memory.as_ref(), [3, 9].into_iter())?;
    ConversationRegistry::new(Arc::clone(&memory)).restore()?;
    let bytes = block_on(memory.get_routed(KEY, SNAPSHOT_KEY))??
        .ok_or("the migration wrote no snapshot")?;
    let broken = flipped(&bytes, bytes.len() - 1, 1)?;
    block_on(memory.apply_routed_batch(
        KEY,
        vec![RoutedBatchItem {
            key: SNAPSHOT_KEY.to_vec(),
            expected: Some(Hash::of(&bytes)),
            value: broken.clone(),
            ttl: None,
        }],
    ))??;
    let recording = Recording::new(Arc::clone(&memory));
    let refused = ConversationRegistry::new(recording.shared()).restore();
    assert!(
        matches!(
            refused,
            Err(OperationLogError::RegistrySnapshot(
                RegistrySnapshotError::Digest
            ))
        ),
        "a digest-broken snapshot was answered {refused:?}"
    );
    assert_eq!(recording.take_reads()?.row_reads, 0);
    assert_eq!(
        block_on(memory.get_routed(KEY, SNAPSHOT_KEY))??,
        Some(broken)
    );
    Ok(())
}