liminal-server 0.24.9

//! 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::{
    RegistrySets, RegistrySnapshot, RegistrySnapshotError, SNAPSHOT_KEY,
};
use super::routed_stream::{read_head, read_page};
use super::tests_registry_snapshot_fixture::{KEY, Recording, TestResult, flat_ids, memory};

/// Codec 2.0 (DUR-027 R4): `covered_next_seq` 1,000, live {7} and sweeping
/// {9}, byte for byte. The digest is blake3 over
/// `b"liminal-registry-snapshot-v2"` and the 56 bytes before it, computed
/// outside the codec with `b3sum`.
const VECTOR: [u8; 88] = [
    b'L', b'S', b'R', b'S', // magic
    0x00, 0x02, 0x00, 0x00, // version 2.0
    0x02, // 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, // live count 1
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07, // live id 7
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, // sweeping count 1
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x09, // sweeping id 9
    0x0a, 0x07, 0xec, 0xb5, 0xb2, 0x88, 0xba, 0x28, 0x22, 0x39, 0x6f, 0x0d, 0xfb, 0x94, 0xd1, 0x46,
    0xc0, 0xcd, 0x42, 0x99, 0x67, 0xf3, 0x6c, 0xe0, 0x8b, 0x56, 0x10, 0x62, 0xc5, 0xbc, 0x2c,
    0xd9, // digest
];

/// Codec 1.0, written before retirement (DUR-019): `covered_next_seq` 1,000
/// and the single live id 7. The digest is blake3 over
/// `b"liminal-registry-snapshot-v1"` and the 40 bytes before it, computed
/// outside the codec with `b3sum`. It still decodes; nothing writes it.
const VECTOR_V1: [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) {
        Ok(snapshot) => Err(format!("malformed snapshot decoded: {snapshot:?}").into()),
        Err(refusal) => Ok(refusal),
    }
}

fn sets(
    live: impl IntoIterator<Item = u64>,
    sweeping: impl IntoIterator<Item = u64>,
) -> RegistrySets {
    RegistrySets {
        live: live.into_iter().collect(),
        sweeping: sweeping.into_iter().collect(),
    }
}

#[test]
fn registry_snapshot_roundtrips_empty_one_and_a_thousand_ids_in_each_set()
-> Result<(), Box<dyn Error>> {
    let thousand: BTreeSet<u64> = (0..999_u64).map(|id| id * 3).chain([u64::MAX]).collect();
    let other: BTreeSet<u64> = (0..1_000_u64).map(|id| id * 3 + 1).collect();
    assert_eq!(thousand.len(), 1_000);
    for (covered_next_seq, written) in [
        (0, RegistrySets::default()),
        (1, sets([7], [])),
        (1, sets([], [7])),
        (100_000, sets(thousand.clone(), [])),
        (100_000, sets(thousand, other)),
    ] {
        let bytes = RegistrySnapshot::encode(covered_next_seq, &written)?;
        assert_eq!(
            bytes.len(),
            72 + 8 * (written.live.len() + written.sweeping.len())
        );
        let decoded = RegistrySnapshot::decode(&bytes)?;
        assert_eq!(decoded.covered_next_seq, covered_next_seq);
        assert_eq!(decoded.sets, written);
    }
    Ok(())
}

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

/// A 1.0 snapshot reads as its live ids with nothing sweeping.
#[test]
fn registry_snapshot_decodes_the_pinned_v1_vector_as_live_ids() -> Result<(), Box<dyn Error>> {
    assert_eq!(
        RegistrySnapshot::decode(&VECTOR_V1)?,
        RegistrySnapshot {
            covered_next_seq: 1_000,
            sets: sets([7], []),
        }
    );
    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: 2, minor: 1 }
    );
    assert_eq!(
        refusal(&flipped(&VECTOR_V1, 7, 1)?)?,
        RegistrySnapshotError::Version { major: 1, minor: 1 }
    );
    // Each codec names its own row schema: 2.0 under 2, 1.0 under 1.
    assert_eq!(
        refusal(&flipped(&VECTOR, 8, 3)?)?,
        RegistrySnapshotError::RowSchema {
            expected: 2,
            actual: 1
        }
    );
    assert_eq!(
        refusal(&flipped(&VECTOR_V1, 8, 3)?)?,
        RegistrySnapshotError::RowSchema {
            expected: 1,
            actual: 2
        }
    );
    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>> {
    // The sweeping count says none: the frame is 80 bytes, not 88.
    assert_eq!(
        refusal(&flipped(&VECTOR, 47, 1)?)?,
        RegistrySnapshotError::Length {
            expected: 80,
            actual: 88
        }
    );
    assert_eq!(
        refusal(&flipped(&VECTOR_V1, 31, 1)?)?,
        RegistrySnapshotError::Length {
            expected: 64,
            actual: 72
        }
    );
    assert_eq!(
        refusal(&VECTOR[..87])?,
        RegistrySnapshotError::Length {
            expected: 88,
            actual: 87
        }
    );
    let mut extended = VECTOR.to_vec();
    extended.push(0);
    assert_eq!(
        refusal(&extended)?,
        RegistrySnapshotError::Length {
            expected: 88,
            actual: 89
        }
    );
    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. Both runs are checked.
#[test]
fn registry_snapshot_refuses_ids_out_of_strictly_ascending_order() -> Result<(), Box<dyn Error>> {
    let live = RegistrySnapshot::encode(2, &sets([3, 9], []))?;
    // The second live id's low byte: 9 ^ 0x0a = 3, equal to the first id.
    assert_eq!(
        refusal(&flipped(&live, 47, 0x0a)?)?,
        RegistrySnapshotError::Order {
            index: 1,
            previous: 3,
            id: 3
        }
    );
    // The first live id's high byte: it rises above the second.
    assert_eq!(
        refusal(&flipped(&live, 32, 0x80)?)?,
        RegistrySnapshotError::Order {
            index: 1,
            previous: 0x8000_0000_0000_0003,
            id: 9
        }
    );
    // The second sweeping id's low byte, after the empty live run and the
    // sweeping count.
    let sweeping = RegistrySnapshot::encode(2, &sets([], [3, 9]))?;
    assert_eq!(
        refusal(&flipped(&sweeping, 55, 0x0a)?)?,
        RegistrySnapshotError::Order {
            index: 1,
            previous: 3,
            id: 3
        }
    );
    Ok(())
}

/// An id live and sweeping at once is refused before the digest, by name;
/// the codec refuses to write one too.
#[test]
fn registry_snapshot_refuses_an_id_in_both_sets() -> Result<(), Box<dyn Error>> {
    // The sweeping id's low byte: 9 ^ 0x0e = 7, the live id.
    assert_eq!(
        refusal(&flipped(&VECTOR, 55, 0x0e)?)?,
        RegistrySnapshotError::BothSets { id: 7 }
    );
    assert_eq!(
        RegistrySnapshot::encode(1_000, &sets([7], [7])),
        Err(RegistrySnapshotError::BothSets { id: 7 })
    );
    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>> {
    // The covered head, a live id that stays in order, a sweeping id that
    // stays apart from the live one, and the digest itself.
    for at in [23, 39, 55, 56, 87] {
        assert_eq!(
            refusal(&flipped(&VECTOR, at, 2)?)?,
            RegistrySnapshotError::Digest
        );
    }
    Ok(())
}

#[test]
fn registry_snapshot_refuses_every_truncation_and_single_byte_change() -> Result<(), Box<dyn Error>>
{
    for vector in [VECTOR.as_slice(), VECTOR_V1.as_slice()] {
        for end in 0..vector.len() {
            assert!(RegistrySnapshot::decode(&vector[..end]).is_err());
        }
        for at in 0..vector.len() {
            for bit in 0..8 {
                let changed = flipped(vector, at, 1 << bit)?;
                assert!(
                    RegistrySnapshot::decode(&changed).is_err(),
                    "byte {at} bit {bit} of a {}-byte vector changed and still decoded",
                    vector.len()
                );
            }
        }
    }
    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, 3)?)?,
        refusal(&flipped(&VECTOR, 9, 1)?)?,
        refusal(&flipped(&VECTOR, 55, 0x0e)?)?,
        refusal(&flipped(&VECTOR, 87, 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(())
}