coordinode-lsm-tree 7.4.1

#![expect(clippy::unwrap_used, clippy::indexing_slicing, reason = "test code")]

use super::*;
use crate::fs::StdFs;
use test_log::test;

/// A written bound round-trips: `read` returns the same `(table_id, bound)`.
#[test]
fn write_then_read_roundtrips_the_bound() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    write(&StdFs, &sst, None, 7, b"k00130", SyncMode::Normal).unwrap();
    match read(&StdFs, &sst, None).unwrap() {
        SidecarRead::Present(id, bound) => {
            assert_eq!(id, 7);
            assert_eq!(bound, b"k00130");
        }
        _ => panic!("expected Present"),
    }
}

/// The sidecar carries the table id in its own payload, so `read` returns the
/// STORED id verbatim, not one inferred from the file name. A foreign id (distinct
/// from the SST's numeric name) must round-trip, so repair's `id == table_id`
/// cross-check (which tells a sidecar for THIS table apart from a stale one left
/// behind by a reused id) compares against the true stored id.
#[test]
fn write_then_read_roundtrips_a_foreign_table_id() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    let foreign_id = 0xDEAD_BEEF_u64;
    write(&StdFs, &sst, None, foreign_id, b"k00130", SyncMode::Normal).unwrap();
    match read(&StdFs, &sst, None).unwrap() {
        SidecarRead::Present(id, bound) => {
            assert_eq!(
                id, foreign_id,
                "the stored id round-trips, not the file name"
            );
            assert_eq!(bound, b"k00130");
        }
        _ => panic!("expected Present"),
    }
}

/// An absent sidecar reads as `Missing` (the SST is unrestricted), never an error.
#[test]
fn absent_sidecar_reads_missing() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Missing
    ));
}

/// A corrupted payload (flipped byte in the body) fails the checksum and reads as
/// `Corrupt`, never a wrong `Present` bound.
#[test]
fn corrupted_payload_reads_corrupt() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    write(&StdFs, &sst, None, 7, b"k00130", SyncMode::Normal).unwrap();

    // Flip a byte in the middle of the sidecar (inside the bound region).
    let path = sidecar_path(&sst);
    let mut bytes = std::fs::read(&path).unwrap();
    let mid = bytes.len() / 2;
    bytes[mid] ^= 0xFF;
    std::fs::write(&path, &bytes).unwrap();

    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Corrupt
    ));
}

/// A truncated sidecar (shorter than header + checksum) reads as `Corrupt`.
#[test]
fn truncated_sidecar_reads_corrupt() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    write(&StdFs, &sst, None, 7, b"k00130", SyncMode::Normal).unwrap();
    let path = sidecar_path(&sst);
    std::fs::write(&path, b"short").unwrap();
    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Corrupt
    ));
}

/// An OVER-LONG sidecar (larger than a valid header + max-length bound + checksum)
/// reads as `Corrupt` and never forces a large allocation: the read is bounded, so
/// a corrupt / padded file is rejected at the size branch before it is
/// materialized.
#[test]
fn oversized_sidecar_reads_corrupt() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    write(&StdFs, &sst, None, 7, b"k00130", SyncMode::Normal).unwrap();

    // One byte past the largest a valid plaintext sidecar can be.
    let over = HEADER_LEN + u16::MAX as usize + CHECKSUM_LEN + 1;
    std::fs::write(sidecar_path(&sst), vec![0u8; over]).unwrap();

    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Corrupt
    ));
}

/// `remove` deletes the sidecar; a subsequent read is `Missing`.
#[test]
fn remove_deletes_the_sidecar() {
    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    write(&StdFs, &sst, None, 7, b"k00130", SyncMode::Normal).unwrap();
    assert!(exists(&StdFs, &sst).unwrap());
    remove(&StdFs, &sst, SyncMode::Normal);
    assert!(!exists(&StdFs, &sst).unwrap());
    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Missing
    ));
}

/// A publish whose atomic RENAME fails must leave NO artifact: the synced temp is
/// removed (best-effort) and a later `read` sees `Missing`, never a half-published
/// sidecar.
#[test]
fn failed_publish_cleans_up_the_temp() {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    let fault = FaultFs::new(StdFs);
    fault.injector().arm(
        FaultRule::new(FaultOp::Rename, Fault::Error(ErrorKind::Other)).on_path("restrict-bound"),
    );

    // The final rename that publishes the sidecar fails, so `write` errors...
    assert!(write(&fault, &sst, None, 7, b"k00130", SyncMode::Normal).is_err());
    // ...leaving no temp behind...
    assert!(
        !sidecar_tmp_path(&sst).exists(),
        "the synced temp must be cleaned up after a failed publish",
    );
    // ...and the sidecar reads as absent, never half-published.
    assert!(matches!(
        read(&StdFs, &sst, None).unwrap(),
        SidecarRead::Missing
    ));
}

/// An ENCRYPTED sidecar round-trips through the provider, and a WRONG key fails
/// the AEAD open (`Corrupt`), never a forged bound.
#[cfg(feature = "encryption")]
#[test]
fn encrypted_sidecar_roundtrips_and_rejects_a_wrong_key() {
    use crate::encryption::{Aes256GcmProvider, EncryptionProvider};
    use std::sync::Arc;

    let dir = tempfile::tempdir().unwrap();
    let sst = dir.path().join("7");
    let provider: Arc<dyn EncryptionProvider> = Arc::new(Aes256GcmProvider::new(&[3u8; 32]));
    write(
        &StdFs,
        &sst,
        Some(provider.as_ref()),
        7,
        b"k00130",
        SyncMode::Normal,
    )
    .unwrap();

    match read(&StdFs, &sst, Some(provider.as_ref())).unwrap() {
        SidecarRead::Present(id, bound) => {
            assert_eq!(id, 7);
            assert_eq!(bound, b"k00130");
        }
        _ => panic!("expected Present with the right key"),
    }

    let wrong: Arc<dyn EncryptionProvider> = Arc::new(Aes256GcmProvider::new(&[9u8; 32]));
    assert!(matches!(
        read(&StdFs, &sst, Some(wrong.as_ref())).unwrap(),
        SidecarRead::Corrupt,
    ));
}

/// The sidecars a slice publishes over two folders on a `FaultFs`, with the
/// tables `1..=3` in `/db/a` and `4` in `/db/b`.
fn batch_fixture() -> (crate::fs::FaultFs<crate::fs::MemFs>, Vec<PathBuf>) {
    let fs = crate::fs::FaultFs::new(crate::fs::MemFs::new());
    fs.create_dir_all(Path::new("/db/a")).unwrap();
    fs.create_dir_all(Path::new("/db/b")).unwrap();
    let paths = ["/db/a/1", "/db/a/2", "/db/a/3", "/db/b/4"]
        .iter()
        .map(PathBuf::from)
        .collect();
    (fs, paths)
}

fn batch<'a>(fs: &'a dyn Fs, paths: &'a [PathBuf], bounds: &'a [Vec<u8>]) -> Vec<SidecarWrite<'a>> {
    paths
        .iter()
        .zip(bounds)
        .zip(1u64..)
        .map(|((table_path, bound), table_id)| SidecarWrite {
            fs,
            table_path,
            encryption: None,
            table_id,
            bound,
        })
        .collect()
}

/// A slice's sidecars publish together: each reads back its own bound, and
/// each folder they share is synced once, not once per sidecar.
#[test]
fn write_many_syncs_each_folder_once() {
    let (fs, paths) = batch_fixture();
    let bounds: Vec<Vec<u8>> = (1..=4).map(|i| format!("k{i}").into_bytes()).collect();
    drop(fs.injector().take_syncs());

    let results = write_many(&batch(&fs, &paths, &bounds), SyncMode::Normal);
    assert!(results.iter().all(Result::is_ok), "every sidecar published");

    for ((path, bound), id) in paths.iter().zip(&bounds).zip(1u64..) {
        match read(&fs, path, None).unwrap() {
            SidecarRead::Present(read_id, read_bound) => {
                assert_eq!((id, bound), (read_id, &read_bound), "{}", path.display());
            }
            _ => panic!("{} has no sidecar", path.display()),
        }
    }
    let folders: Vec<PathBuf> = fs
        .injector()
        .take_syncs()
        .into_iter()
        .filter(|(_, kind)| matches!(kind, crate::fs::SyncKind::Directory))
        .map(|(path, _)| path)
        .collect();
    assert_eq!(
        vec![PathBuf::from("/db/a"), PathBuf::from("/db/b")],
        folders,
        "one sync per folder",
    );
}

/// A folder whose sync fails leaves none of its sidecars known durable, so
/// every write into it fails and no punch relies on it; a sidecar in another
/// folder still publishes, and no temp file is left.
#[test]
fn write_many_fails_every_sidecar_of_a_folder_whose_sync_fails() {
    use crate::fs::{Fault, FaultOp, FaultRule};
    let (fs, paths) = batch_fixture();
    let bounds: Vec<Vec<u8>> = (1..=4).map(|i| format!("k{i}").into_bytes()).collect();
    fs.injector().arm(
        FaultRule::new(
            FaultOp::SyncDirectory,
            Fault::Error(crate::io::ErrorKind::Other),
        )
        .on_path("/db/a")
        .once(),
    );

    let results = write_many(&batch(&fs, &paths, &bounds), SyncMode::Normal);
    let ok: Vec<bool> = results.iter().map(Result::is_ok).collect();
    assert_eq!(
        vec![false, false, false, true],
        ok,
        "the failed folder's three fail"
    );
    for path in &paths {
        assert!(
            !fs.exists(&sidecar_tmp_path(path)).unwrap(),
            "no temp left beside {}",
            path.display(),
        );
    }
}

/// A sidecar that cannot be written fails alone: its neighbours publish, and
/// an over-long bound is refused before anything is written for it.
#[test]
fn write_many_fails_only_the_sidecar_that_cannot_be_written() {
    use crate::fs::{Fault, FaultOp, FaultRule};
    let (fs, paths) = batch_fixture();
    let mut bounds: Vec<Vec<u8>> = (1..=4).map(|i| format!("k{i}").into_bytes()).collect();
    bounds[3] = vec![b'x'; usize::from(u16::MAX) + 1];
    fs.injector().arm(
        FaultRule::new(FaultOp::Open, Fault::Error(crate::io::ErrorKind::Other))
            .on_path("2.restrict-bound.tmp")
            .once(),
    );

    let results = write_many(&batch(&fs, &paths, &bounds), SyncMode::Normal);
    let ok: Vec<bool> = results.iter().map(Result::is_ok).collect();
    assert_eq!(vec![true, false, true, false], ok);
    assert!(matches!(
        read(&fs, &paths[1], None).unwrap(),
        SidecarRead::Missing
    ));
    assert!(matches!(
        read(&fs, &paths[3], None).unwrap(),
        SidecarRead::Missing
    ));
    assert!(matches!(
        read(&fs, &paths[2], None).unwrap(),
        SidecarRead::Present(3, _)
    ));
}