armdb 0.14.1

#[cfg(feature = "var-collections")]
use armdb::VarTree;
use armdb::{Config, ConstTree};
use std::sync::Arc;
use std::thread;
use tempfile::tempdir;

#[test]
fn compaction_two_passes_under_write_churn() {
    const KEY_COUNT: u64 = 128;
    const WRITE_ROUNDS: u64 = 32;
    const COMPACTION_PASSES: usize = 64;

    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.shard_count = 1;
    config.hints = Some(true);
    config.max_file_size = 4096;
    config.write_buffer_size = 4096;
    config.compaction_threshold = 0.0;
    let tree = Arc::new(ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap());

    for key in 0..KEY_COUNT {
        tree.put(&key.to_be_bytes(), &0u64.to_be_bytes()).unwrap();
    }
    tree.flush().unwrap();

    let writer_tree = Arc::clone(&tree);
    let writer = thread::spawn(move || {
        for round in 1..=WRITE_ROUNDS {
            for key in 0..KEY_COUNT {
                writer_tree
                    .put(&key.to_be_bytes(), &round.to_be_bytes())
                    .unwrap();
            }
        }
    });

    let first_tree = Arc::clone(&tree);
    let first = thread::spawn(move || {
        for _ in 0..COMPACTION_PASSES {
            first_tree.compact().unwrap();
        }
    });
    let second_tree = Arc::clone(&tree);
    let second = thread::spawn(move || {
        for _ in 0..COMPACTION_PASSES {
            second_tree.compact().unwrap();
        }
    });

    writer.join().unwrap();
    first.join().unwrap();
    second.join().unwrap();
    tree.flush().unwrap();

    for key in 0..KEY_COUNT {
        assert_eq!(
            tree.get(&key.to_be_bytes()),
            Some(WRITE_ROUNDS.to_be_bytes())
        );
    }

    drop(tree);
    let reopened = ConstTree::<[u8; 8], 8>::open(dir.path(), {
        let mut config = Config::test();
        config.shard_count = 1;
        config.hints = Some(true);
        config.max_file_size = 4096;
        config.write_buffer_size = 4096;
        config.compaction_threshold = 0.0;
        config
    })
    .unwrap();
    for key in 0..KEY_COUNT {
        assert_eq!(
            reopened.get(&key.to_be_bytes()),
            Some(WRITE_ROUNDS.to_be_bytes())
        );
    }
    reopened.sync_hints().unwrap();

    let shard_dir = dir.path().join("shard_000");
    for entry in std::fs::read_dir(shard_dir).unwrap() {
        let path = entry.unwrap().path();
        let name = path.file_name().unwrap().to_string_lossy();
        assert!(!name.ends_with(".tmp"), "orphan temporary file: {name}");
        if name.ends_with(".data") {
            assert!(
                path.with_extension("hint").exists(),
                "immutable data file {name} must have a hint"
            );
        }
    }
}

#[test]
fn test_compaction_with_concurrent_reads() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.shard_count = 2;
    config.max_file_size = 4096;
    config.write_buffer_size = 4096;
    config.compaction_threshold = 0.1;

    let tree = Arc::new(ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap());

    // Write 1000 entries
    for i in 0..1000u64 {
        let key = i.to_be_bytes();
        let val = (i * 7).to_be_bytes();
        tree.put(&key, &val).unwrap();
    }

    // Overwrite first 500 to create dead bytes
    for i in 0..500u64 {
        let key = i.to_be_bytes();
        let val = (i * 7 + 1).to_be_bytes();
        tree.put(&key, &val).unwrap();
    }

    // Spawn reader threads
    let mut handles = Vec::new();
    for _ in 0..3 {
        let tree = Arc::clone(&tree);
        handles.push(thread::spawn(move || {
            for iter in 0..1000u64 {
                let i = iter % 1000;
                let key = i.to_be_bytes();
                if let Some(val) = tree.get(&key) {
                    // Value must be one of the two valid values
                    let v = u64::from_be_bytes(val);
                    let expected_old = i * 7;
                    let expected_new = i * 7 + 1;
                    assert!(
                        v == expected_old || v == expected_new,
                        "unexpected value {} for key {}",
                        v,
                        i
                    );
                }
            }
        }));
    }

    // Compact in main thread while readers are running
    tree.compact().unwrap();

    for handle in handles {
        handle.join().expect("reader thread panicked");
    }

    // Verify final state: all 1000 keys with correct values
    for i in 0..1000u64 {
        let key = i.to_be_bytes();
        let expected = if i < 500 {
            (i * 7 + 1).to_be_bytes()
        } else {
            (i * 7).to_be_bytes()
        };
        let val = tree
            .get(&key)
            .unwrap_or_else(|| panic!("key {} not found after compaction", i));
        assert_eq!(val, expected, "value mismatch for key {}", i);
    }
}

#[test]
fn test_compaction_with_concurrent_writes() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.shard_count = 2;
    config.max_file_size = 4096;
    config.write_buffer_size = 4096;
    config.compaction_threshold = 0.1;

    let tree = Arc::new(ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap());

    // Write 1000 entries and overwrite 500
    for i in 0..1000u64 {
        let key = i.to_be_bytes();
        let val = i.to_be_bytes();
        tree.put(&key, &val).unwrap();
    }
    for i in 0..500u64 {
        let key = i.to_be_bytes();
        let val = (i + 1).to_be_bytes();
        tree.put(&key, &val).unwrap();
    }

    // Spawn writer thread that adds new keys 1000..2000
    let tree_w = Arc::clone(&tree);
    let writer = thread::spawn(move || {
        for i in 1000..2000u64 {
            let key = i.to_be_bytes();
            let val = (i * 3).to_be_bytes();
            tree_w.put(&key, &val).unwrap();
        }
    });

    // Compact in main thread while writer is running
    tree.compact().unwrap();

    writer.join().expect("writer thread panicked");

    // Verify all 2000 keys
    for i in 0..2000u64 {
        let key = i.to_be_bytes();
        let expected = if i < 500 {
            (i + 1).to_be_bytes()
        } else if i < 1000 {
            i.to_be_bytes()
        } else {
            (i * 3).to_be_bytes()
        };
        let val = tree
            .get(&key)
            .unwrap_or_else(|| panic!("key {} not found", i));
        assert_eq!(val, expected, "value mismatch for key {}", i);
    }
}

#[cfg(feature = "var-collections")]
#[test]
fn test_compaction_with_concurrent_reads_var() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.shard_count = 2;
    config.max_file_size = 4096;
    config.write_buffer_size = 4096;
    config.compaction_threshold = 0.1;

    let tree = Arc::new(VarTree::<[u8; 8]>::open(dir.path(), config).unwrap());

    for i in 0..1000u64 {
        let key = i.to_be_bytes();
        let val = format!("val_{i}");
        tree.put(&key, val.as_bytes()).unwrap();
    }

    // Overwrite first 500
    for i in 0..500u64 {
        let key = i.to_be_bytes();
        let val = format!("new_{i}");
        tree.put(&key, val.as_bytes()).unwrap();
    }

    // Reader threads
    let mut handles = Vec::new();
    for _ in 0..3 {
        let tree = Arc::clone(&tree);
        handles.push(thread::spawn(move || {
            for iter in 0..1000u64 {
                let i = iter % 1000;
                let key = i.to_be_bytes();
                if let Some(val) = tree.get(&key) {
                    let s = std::str::from_utf8(val.as_ref()).unwrap_or("???");
                    let old = format!("val_{i}");
                    let new = format!("new_{i}");
                    assert!(
                        s == old || s == new,
                        "unexpected value '{}' for key {}",
                        s,
                        i
                    );
                }
            }
        }));
    }

    tree.compact().unwrap();

    for handle in handles {
        handle.join().expect("reader thread panicked");
    }

    // Verify final state
    for i in 0..1000u64 {
        let key = i.to_be_bytes();
        let expected = if i < 500 {
            format!("new_{i}")
        } else {
            format!("val_{i}")
        };
        let val = tree
            .get(&key)
            .unwrap_or_else(|| panic!("key {} not found after compaction", i));
        assert_eq!(val.as_ref(), expected.as_bytes());
    }
}

#[test]
fn test_double_compaction() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.max_file_size = 4096;
    config.write_buffer_size = 4096;
    config.compaction_threshold = 0.1;

    let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();

    // Phase 1: write + overwrite + compact
    for i in 0..500u64 {
        let key = i.to_be_bytes();
        let val = i.to_be_bytes();
        tree.put(&key, &val).unwrap();
    }
    for i in 0..250u64 {
        let key = i.to_be_bytes();
        let val = (i + 1000).to_be_bytes();
        tree.put(&key, &val).unwrap();
    }

    tree.compact().unwrap();

    // Phase 2: more overwrites + compact again
    for i in 250..500u64 {
        let key = i.to_be_bytes();
        let val = (i + 2000).to_be_bytes();
        tree.put(&key, &val).unwrap();
    }

    tree.compact().unwrap();

    // Verify all 500 keys
    assert_eq!(tree.len(), 500);
    for i in 0..500u64 {
        let key = i.to_be_bytes();
        let expected = if i < 250 {
            (i + 1000).to_be_bytes()
        } else {
            (i + 2000).to_be_bytes()
        };
        assert_eq!(
            tree.get(&key).unwrap(),
            expected,
            "double compaction: value mismatch at key {}",
            i
        );
    }
}