armdb 0.14.3

//! Regression: an 8-byte zero gap left by a force flush at in-page offset 4088.
//!
//! A durability barrier on a page-aligned shard (`direct_io` or encryption) pads
//! the partial page with zeros and keeps appending at the next page. When the
//! partial page holds 4088 bytes, the padding is 8 bytes — shorter than an entry
//! header. A scanner that recognises padding only by an all-zero 16-byte header
//! then reads 8 zeros plus the next entry's GSN as a header and fails with
//! `CrcMismatch` (recovery, compaction) or builds a bogus hint.
#![cfg(feature = "var-collections")]

use std::path::Path;

use armdb::{Config, VarTree};
use tempfile::tempdir;

const HDR: usize = 16;
const KEY_LEN: usize = 8;

#[derive(Clone, Copy, Debug)]
enum Mode {
    /// Plain + `direct_io`: page-aligned writes without encryption.
    DirectIo,
    /// Encrypted (page-aligned via the cipher), buffered I/O.
    #[cfg(feature = "encryption")]
    Encrypted,
    /// Plain buffered I/O: byte-granular flushes, never pads (control).
    Plain,
}

fn config(mode: Mode) -> Config {
    let mut c = Config::test();
    c.shard_count = 1;
    // One force-advance barrier per write.
    c.sync_batch_size = 1;
    match mode {
        Mode::DirectIo => c.direct_io = true,
        #[cfg(feature = "encryption")]
        Mode::Encrypted => c.encryption_key = Some([0x42; 32]),
        Mode::Plain => {}
    }
    c
}

fn page_aligned_modes() -> Vec<Mode> {
    vec![
        Mode::DirectIo,
        #[cfg(feature = "encryption")]
        Mode::Encrypted,
    ]
}

fn key(i: u64) -> [u8; KEY_LEN] {
    i.to_be_bytes()
}

/// Value length whose entry occupies exactly `entry` bytes on disk.
fn value_for_entry(entry: usize, byte: u8) -> Vec<u8> {
    vec![byte; entry - HDR - KEY_LEN]
}

/// First entry occupies `first_entry` bytes → padding = 4096 - first_entry.
fn write_and_reopen(mode: Mode, first_entry: usize) {
    let dir = tempdir().unwrap();
    let first = value_for_entry(first_entry, 0xAA);
    {
        let t = VarTree::<[u8; KEY_LEN]>::open(dir.path(), config(mode)).unwrap();
        t.put(&key(1), &first).unwrap();
        t.put(&key(2), &[0xBB; 100]).unwrap();
        t.put(&key(3), &[0xCC; 200]).unwrap();
        t.close().unwrap();
    }
    let t = VarTree::<[u8; KEY_LEN]>::open(dir.path(), config(mode))
        .unwrap_or_else(|e| panic!("{mode:?}, entry {first_entry}: reopen failed: {e:?}"));
    assert_eq!(t.len(), 3, "{mode:?}, entry {first_entry}");
    assert_eq!(t.get(&key(1)).unwrap().as_ref(), &first[..]);
    assert_eq!(t.get(&key(2)).unwrap().as_ref(), &[0xBB; 100][..]);
    assert_eq!(t.get(&key(3)).unwrap().as_ref(), &[0xCC; 200][..]);
}

#[test]
fn reopen_across_8_byte_padding() {
    for mode in page_aligned_modes() {
        write_and_reopen(mode, 4088);
    }
}

#[test]
fn reopen_across_16_and_24_byte_padding() {
    for mode in page_aligned_modes() {
        write_and_reopen(mode, 4080);
        write_and_reopen(mode, 4072);
    }
}

#[test]
fn plain_buffered_entry_4088_reopens() {
    write_and_reopen(Mode::Plain, 4088);
}

/// A barrier that leaves an 8-byte gap, followed by enough writes to rotate the
/// file into an immutable one: compaction and hint generation must walk the gap.
fn rotate_and_compact(mode: Mode) {
    let dir = tempdir().unwrap();
    let mut cfg = config(mode);
    cfg.max_file_size = 64 * 1024;
    cfg.write_buffer_size = 16 * 1024;
    cfg.compaction_threshold = 0.1;

    let first = value_for_entry(4088, 0xAA);
    let last = |i: u64| vec![(i % 251) as u8; 1000];
    {
        let t = VarTree::<[u8; KEY_LEN]>::open(dir.path(), cfg.clone()).unwrap();
        t.put(&key(100), &first).unwrap();
        for i in 0..200u64 {
            t.put(&key(i % 5), &last(i)).unwrap();
        }
        t.put(&key(100), &[0xDD; 300]).unwrap();
        t.compact()
            .unwrap_or_else(|e| panic!("{mode:?}: compact failed: {e:?}"));
        assert_eq!(t.get(&key(100)).unwrap().as_ref(), &[0xDD; 300][..]);
        t.close().unwrap();
    }

    let t = VarTree::<[u8; KEY_LEN]>::open(dir.path(), cfg)
        .unwrap_or_else(|e| panic!("{mode:?}: reopen failed: {e:?}"));
    // Recovery writes a hint for every immutable file that lacks one, by the
    // same scan rotation uses; a scan that trips on the gap would skip it.
    let shard = dir.path().join("shard_000");
    let ids = data_file_ids(&shard);
    assert!(
        ids.len() > 1,
        "{mode:?}: expected rotated files, got {ids:?}"
    );
    for id in &ids[..ids.len() - 1] {
        assert!(has_hint(&shard, *id), "{mode:?}: no hint for {id:06}.data");
    }
    assert_eq!(t.len(), 6);
    assert_eq!(t.get(&key(100)).unwrap().as_ref(), &[0xDD; 300][..]);
    for k in 0..5u64 {
        // The last write to key k was i = 195 + k.
        assert_eq!(t.get(&key(k)).unwrap().as_ref(), &last(195 + k)[..]);
    }
}

fn data_file_ids(shard_dir: &Path) -> Vec<u32> {
    let mut ids: Vec<u32> = std::fs::read_dir(shard_dir)
        .unwrap()
        .filter_map(|e| {
            let name = e.ok()?.file_name().to_string_lossy().into_owned();
            name.strip_suffix(".data")?.parse().ok()
        })
        .collect();
    ids.sort_unstable();
    ids
}

fn has_hint(shard_dir: &Path, file_id: u32) -> bool {
    shard_dir.join(format!("{file_id:06}.hint")).exists()
}

#[test]
fn rotate_and_compact_across_8_byte_padding() {
    for mode in page_aligned_modes() {
        rotate_and_compact(mode);
    }
}