use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Barrier};
use std::thread;
use tempfile::tempdir;
use holt::{Durability, Tree, TreeConfig, DB};
fn wal_path(dir: &Path) -> PathBuf {
dir.join("journal.wal")
}
fn manual_checkpoint_cfg(dir: &std::path::Path) -> TreeConfig {
let mut cfg = TreeConfig::new(dir);
cfg.checkpoint.enabled = false;
cfg
}
fn durable_cfg(dir: &std::path::Path) -> TreeConfig {
let mut cfg = manual_checkpoint_cfg(dir);
cfg.durability = Durability::Wal { sync: true };
cfg
}
#[test]
fn db_named_trees_replay_from_one_wal() {
let dir = tempdir().unwrap();
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.create_tree("objects").unwrap();
let inodes = db.create_tree("inodes").unwrap();
objects.put(b"same/key", b"object").unwrap();
inodes.put(b"same/key", b"inode").unwrap();
assert!(db
.atomic(|batch| {
batch.put("objects", b"bucket/a.jpg", b"etag-a");
batch.put("inodes", b"42", b"mode=0644");
})
.unwrap());
}
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.open_tree("objects").unwrap();
let inodes = db.open_tree("inodes").unwrap();
assert_eq!(db.list_trees().unwrap(), vec!["inodes", "objects"]);
assert_eq!(
objects.get(b"same/key").unwrap().as_deref(),
Some(&b"object"[..])
);
assert_eq!(
inodes.get(b"same/key").unwrap().as_deref(),
Some(&b"inode"[..])
);
assert_eq!(
objects.get(b"bucket/a.jpg").unwrap().as_deref(),
Some(&b"etag-a"[..])
);
assert_eq!(
inodes.get(b"42").unwrap().as_deref(),
Some(&b"mode=0644"[..])
);
}
}
#[test]
fn db_checkpoint_flushes_replayed_multi_tree_without_tree_handles() {
let dir = tempdir().unwrap();
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let _objects = db.create_tree("objects").unwrap();
let _inodes = db.create_tree("inodes").unwrap();
assert!(db
.atomic(|batch| {
batch.put("objects", b"bucket/a.jpg", b"etag-a");
batch.put("inodes", b"42", b"mode=0644");
})
.unwrap());
}
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
db.checkpoint().unwrap();
}
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.open_tree("objects").unwrap();
let inodes = db.open_tree("inodes").unwrap();
assert_eq!(
objects.get(b"bucket/a.jpg").unwrap().as_deref(),
Some(&b"etag-a"[..])
);
assert_eq!(
inodes.get(b"42").unwrap().as_deref(),
Some(&b"mode=0644"[..])
);
}
}
#[test]
fn db_view_flushes_deferred_tree_writes_before_snapshot() {
let dir = tempdir().unwrap();
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.create_tree("objects").unwrap();
objects.put(b"bucket/live", b"etag").unwrap();
db.view(&[("objects", b"bucket/")], |view| {
let objects = view.tree("objects").unwrap();
assert_eq!(objects.get(b"bucket/live")?.as_deref(), Some(&b"etag"[..]));
Ok::<(), holt::Error>(())
})
.unwrap();
}
#[test]
fn db_drop_tree_survives_checkpoint_and_reopen() {
let dir = tempdir().unwrap();
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.create_tree("objects").unwrap();
objects.put(b"bucket/a.jpg", b"etag-a").unwrap();
db.checkpoint().unwrap();
db.drop_tree("objects").unwrap();
drop(objects);
db.checkpoint().unwrap();
}
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
assert!(db.list_trees().unwrap().is_empty());
assert!(matches!(
db.open_tree("objects"),
Err(holt::Error::TreeNotFound { .. })
));
let recreated = db.create_tree("objects").unwrap();
assert!(recreated.get(b"bucket/a.jpg").unwrap().is_none());
recreated.put(b"bucket/b.jpg", b"etag-b").unwrap();
db.checkpoint().unwrap();
}
{
let db = DB::open(durable_cfg(dir.path())).unwrap();
let objects = db.open_tree("objects").unwrap();
assert!(objects.get(b"bucket/a.jpg").unwrap().is_none());
assert_eq!(
objects.get(b"bucket/b.jpg").unwrap().as_deref(),
Some(&b"etag-b"[..])
);
}
}
#[test]
fn view_snapshots_uncheckpointed_persistent_bytes() {
let dir = tempdir().unwrap();
let tree = Tree::open(durable_cfg(dir.path())).unwrap();
tree.put(b"tenant-a/file", b"old").unwrap();
tree.view(b"tenant-a/", |view| {
tree.put(b"tenant-a/file", b"new").unwrap();
tree.put(b"tenant-a/after-view", b"new").unwrap();
assert_eq!(view.get(b"tenant-a/file")?.as_deref(), Some(&b"old"[..]));
assert!(view.get(b"tenant-a/after-view")?.is_none());
Ok(())
})
.unwrap();
}
#[test]
fn durable_writers_share_group_commit_syncs() {
let dir = tempdir().unwrap();
let tree = Arc::new(Tree::open(durable_cfg(dir.path())).unwrap());
const WRITERS: usize = 12;
let barrier = Arc::new(Barrier::new(WRITERS));
let handles: Vec<_> = (0..WRITERS)
.map(|i| {
let tree = Arc::clone(&tree);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
let key = format!("gc/key-{i:02}");
let value = format!("value-{i:02}");
barrier.wait();
tree.put(key.as_bytes(), value.as_bytes()).unwrap();
})
})
.collect();
for h in handles {
h.join().unwrap();
}
let stats = tree.stats().unwrap();
let journal = stats.journal.expect("persistent tree has journal stats");
assert_eq!(journal.appends, WRITERS as u64);
assert!(
journal.syncs <= journal.appends,
"durable writers should not issue duplicate fsyncs; appends={}, syncs={}",
journal.appends,
journal.syncs,
);
drop(tree);
let reopened = Tree::open(durable_cfg(dir.path())).unwrap();
for i in 0..WRITERS {
let key = format!("gc/key-{i:02}");
let value = format!("value-{i:02}");
assert_eq!(
reopened.get(key.as_bytes()).unwrap().as_deref(),
Some(value.as_bytes()),
);
}
}
#[test]
fn clean_checkpoint_skips_empty_wal_flush() {
let dir = tempdir().unwrap();
let tree = Tree::open(TreeConfig::new(dir.path())).unwrap();
let before = tree.stats().unwrap().journal.unwrap();
tree.checkpoint().unwrap();
let after = tree.stats().unwrap().journal.unwrap();
assert_eq!(
after.syncs, before.syncs,
"checkpoint on a fresh tree must not fsync an empty WAL",
);
tree.put(b"wal-clean/k", b"v").unwrap();
tree.checkpoint().unwrap();
assert_eq!(fs::metadata(wal_path(dir.path())).unwrap().len(), 32);
let after_truncate = tree.stats().unwrap().journal.unwrap();
tree.checkpoint().unwrap();
let clean_again = tree.stats().unwrap().journal.unwrap();
assert_eq!(
clean_again.syncs, after_truncate.syncs,
"checkpoint after WAL truncate must stay a no-op",
);
}
#[test]
fn checkpoint_reuses_durable_group_commit_wal_sync() {
let dir = tempdir().unwrap();
let tree = Tree::open(durable_cfg(dir.path())).unwrap();
tree.put(b"durable-checkpoint/k", b"v").unwrap();
let after_put = tree.stats().unwrap().journal.unwrap();
assert!(after_put.syncs > 0);
tree.checkpoint().unwrap();
let after_checkpoint = tree.stats().unwrap().journal.unwrap();
assert_eq!(
after_checkpoint.syncs, after_put.syncs,
"checkpoint must not fsync WAL records already made durable by group commit",
);
assert_eq!(fs::metadata(wal_path(dir.path())).unwrap().len(), 32);
}
#[test]
fn persistent_put_then_reopen_via_wal_replay() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..50u32 {
let k = format!("k{i:03}");
let v = format!("v-{i}");
tree.put(k.as_bytes(), v.as_bytes()).unwrap();
}
}
let wal_size_after_drop = fs::metadata(wal_path(dir.path())).unwrap().len();
assert!(wal_size_after_drop > 32, "WAL should hold records");
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..50u32 {
let k = format!("k{i:03}");
let v = format!("v-{i}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(v.as_bytes()),
"WAL replay should have restored key {k}",
);
}
}
}
#[test]
fn deferred_put_is_visible_to_point_and_scan_before_checkpoint() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
let tree = Tree::open(cfg).unwrap();
tree.put(b"bucket/a/001", b"one").unwrap();
assert_eq!(
tree.get(b"bucket/a/001").unwrap().as_deref(),
Some(&b"one"[..]),
"point lookup must consult deferred writes before base ART",
);
let listed: Vec<_> = tree
.scan(b"bucket/a/")
.into_iter()
.map(|entry| match entry.unwrap() {
holt::RangeEntry::Key { key, value, .. } => (key, value),
holt::RangeEntry::CommonPrefix(_) => unreachable!("no delimiter"),
_ => unreachable!("unknown range entry"),
})
.collect();
assert_eq!(listed, vec![(b"bucket/a/001".to_vec(), b"one".to_vec())]);
}
#[test]
fn checkpoint_merges_deferred_put_before_truncating_wal() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"checkpoint/deferred", b"value").unwrap();
tree.checkpoint().unwrap();
assert_eq!(fs::metadata(wal_path(dir.path())).unwrap().len(), 32);
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(
tree.get(b"checkpoint/deferred").unwrap().as_deref(),
Some(&b"value"[..])
);
}
#[test]
fn deferred_delete_survives_checkpoint_and_reopen() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"delete/me", b"value").unwrap();
tree.checkpoint().unwrap();
assert!(tree.delete(b"delete/me").unwrap());
assert_eq!(tree.get(b"delete/me").unwrap(), None);
tree.checkpoint().unwrap();
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(tree.get(b"delete/me").unwrap(), None);
}
#[test]
fn replay_then_checkpoint_then_reopen_preserves_data() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..50u32 {
tree.put(format!("k{i:03}").as_bytes(), format!("v-{i}").as_bytes())
.unwrap();
}
}
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..50u32 {
assert_eq!(
tree.get(format!("k{i:03}").as_bytes()).unwrap().as_deref(),
Some(format!("v-{i}").as_bytes()),
);
}
tree.checkpoint().unwrap();
let wal_size_after = fs::metadata(wal_path(dir.path())).unwrap().len();
assert_eq!(wal_size_after, 32, "WAL truncated to header-only");
}
{
let tree = Tree::open(cfg).unwrap();
for i in 0..50u32 {
let k = format!("k{i:03}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(format!("v-{i}").as_bytes()),
"key {k} lost — replay→checkpoint didn't persist replayed state",
);
}
}
}
#[test]
fn checkpoint_truncates_wal_and_keys_survive_reopen() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..20u32 {
tree.put(format!("k{i:02}").as_bytes(), format!("v{i}").as_bytes())
.unwrap();
}
let wal_size_before = fs::metadata(wal_path(dir.path())).unwrap().len();
assert!(wal_size_before > 32);
tree.checkpoint().unwrap();
let wal_size_after = fs::metadata(wal_path(dir.path())).unwrap().len();
assert_eq!(
wal_size_after, 32,
"checkpoint should truncate WAL to header-only",
);
}
{
let tree = Tree::open(cfg).unwrap();
for i in 0..20u32 {
let k = format!("k{i:02}");
let v = format!("v{i}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(v.as_bytes()),
);
}
}
}
#[test]
fn delete_through_wal_replays_correctly() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..10u32 {
tree.put(format!("k{i}").as_bytes(), format!("v{i}").as_bytes())
.unwrap();
}
for i in 0..10u32 {
if i % 2 == 0 {
let removed = tree.delete(format!("k{i}").as_bytes()).unwrap();
assert!(removed);
}
}
}
let tree = Tree::open(cfg).unwrap();
for i in 0..10u32 {
let got = tree.get(format!("k{i}").as_bytes()).unwrap();
if i % 2 == 0 {
assert_eq!(got, None, "k{i} should have been deleted");
} else {
assert_eq!(got.as_deref(), Some(format!("v{i}").as_bytes()));
}
}
}
#[test]
fn rename_through_wal_replays_correctly() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"a", b"v-a").unwrap();
tree.put(b"b", b"v-b").unwrap();
tree.rename(b"a", b"a2", false).unwrap();
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(tree.get(b"a").unwrap(), None);
assert_eq!(tree.get(b"a2").unwrap().as_deref(), Some(&b"v-a"[..]));
assert_eq!(tree.get(b"b").unwrap().as_deref(), Some(&b"v-b"[..]));
}
#[test]
fn conditional_writes_replay_through_wal() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
assert!(tree.put_if_absent(b"cas/k", b"v1").unwrap());
assert!(!tree.put_if_absent(b"cas/k", b"blocked").unwrap());
let v1 = tree.get_version(b"cas/k").unwrap().unwrap();
assert!(tree.compare_and_put(b"cas/k", v1, b"v2").unwrap());
assert!(!tree.compare_and_put(b"cas/k", v1, b"stale").unwrap());
let v2 = tree.get_version(b"cas/k").unwrap().unwrap();
assert!(!tree.delete_if_version(b"cas/k", v1).unwrap());
assert!(tree.delete_if_version(b"cas/k", v2).unwrap());
assert!(tree.put_if_absent(b"cas/resurrected", b"live").unwrap());
}
let tree = Tree::open(cfg).unwrap();
assert!(tree.get(b"cas/k").unwrap().is_none());
assert!(tree.get_version(b"cas/k").unwrap().is_none());
assert_eq!(
tree.get(b"cas/resurrected").unwrap().as_deref(),
Some(&b"live"[..]),
);
}
#[test]
fn enqueue_mode_loses_writes_without_checkpoint_or_fsync() {
let dir = tempdir().unwrap();
let cfg = manual_checkpoint_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..50u32 {
tree.put(
format!("transient{i}").as_bytes(),
format!("v{i}").as_bytes(),
)
.unwrap();
}
}
let wal_size = fs::metadata(wal_path(dir.path())).unwrap().len();
assert_eq!(wal_size, 32);
let tree = Tree::open(cfg).unwrap();
for i in 0..50u32 {
assert_eq!(
tree.get(format!("transient{i}").as_bytes()).unwrap(),
None,
"transient{i} should have been lost",
);
}
}
#[test]
fn batched_mode_loses_writes_without_checkpoint() {
let dir = tempdir().unwrap();
let mut cfg = manual_checkpoint_cfg(dir.path());
cfg.memory_flush_on_write = false;
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..10u32 {
tree.put(
format!("transient{i}").as_bytes(),
format!("v{i}").as_bytes(),
)
.unwrap();
}
}
let tree = Tree::open(cfg).unwrap();
for i in 0..10u32 {
assert_eq!(
tree.get(format!("transient{i}").as_bytes()).unwrap(),
None,
"transient{i} should have been lost",
);
}
}
#[test]
fn batched_mode_with_checkpoint_persists_everything() {
let dir = tempdir().unwrap();
let mut cfg = TreeConfig::new(dir.path());
cfg.memory_flush_on_write = false;
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..30u32 {
tree.put(
format!("batch{i:02}").as_bytes(),
format!("v{i}").as_bytes(),
)
.unwrap();
}
tree.checkpoint().unwrap();
}
let tree = Tree::open(cfg).unwrap();
for i in 0..30u32 {
let v = tree
.get(format!("batch{i:02}").as_bytes())
.unwrap()
.expect("batch key survives via blob image");
assert_eq!(v, format!("v{i}").into_bytes());
}
}
#[test]
fn next_seq_resumes_past_replayed_records() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..5u32 {
tree.put(format!("k{i}").as_bytes(), b"v").unwrap();
}
}
{
let tree = Tree::open(cfg).unwrap();
tree.put(b"after-replay", b"v").unwrap();
assert_eq!(
tree.get(b"after-replay").unwrap().as_deref(),
Some(&b"v"[..])
);
for i in 0..5u32 {
assert_eq!(
tree.get(format!("k{i}").as_bytes()).unwrap().as_deref(),
Some(&b"v"[..]),
);
}
}
}
#[test]
fn open_with_blob_store_attaches_no_wal() {
use holt::{BlobStore, MemoryBlobStore, TreeBuilder};
use std::sync::Arc;
let dir = tempdir().unwrap();
let store: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
{
let tree = TreeBuilder::new(dir.path())
.open_with_blob_store(store.clone())
.unwrap();
tree.put(b"k", b"v").unwrap();
}
assert!(!wal_path(dir.path()).exists());
}
#[test]
fn many_round_trips_through_checkpoint_boundaries() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..20u32 {
tree.put(format!("a{i:02}").as_bytes(), b"A").unwrap();
}
tree.checkpoint().unwrap();
for i in 0..20u32 {
tree.put(format!("b{i:02}").as_bytes(), b"B").unwrap();
}
tree.checkpoint().unwrap();
for i in 0..20u32 {
tree.put(format!("c{i:02}").as_bytes(), b"C").unwrap();
}
}
let tree = Tree::open(cfg).unwrap();
for i in 0..20u32 {
assert_eq!(
tree.get(format!("a{i:02}").as_bytes()).unwrap().as_deref(),
Some(&b"A"[..]),
);
assert_eq!(
tree.get(format!("b{i:02}").as_bytes()).unwrap().as_deref(),
Some(&b"B"[..]),
);
assert_eq!(
tree.get(format!("c{i:02}").as_bytes()).unwrap().as_deref(),
Some(&b"C"[..]),
);
}
}
#[test]
fn batch_persists_through_crash_and_replay() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"seed", b"S").unwrap();
tree.atomic(|b| {
b.put(b"batch-a", b"A");
b.put(b"batch-b", b"B");
b.delete(b"seed");
b.rename(b"batch-a", b"batch-aa", false);
})
.unwrap();
}
let tree = Tree::open(cfg).unwrap();
assert!(tree.get(b"seed").unwrap().is_none());
assert!(tree.get(b"batch-a").unwrap().is_none());
assert_eq!(tree.get(b"batch-aa").unwrap().as_deref(), Some(&b"A"[..]));
assert_eq!(tree.get(b"batch-b").unwrap().as_deref(), Some(&b"B"[..]));
}
#[test]
fn compact_insert_run_batch_persists_through_replay() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
let mut versions = Vec::new();
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.atomic(|b| {
for i in 0..16u32 {
let key = format!("bulk/object-{i:04}");
let value = format!("metadata-{i:04}");
b.put(key.as_bytes(), value.as_bytes());
}
})
.unwrap();
for i in 0..16u32 {
let key = format!("bulk/object-{i:04}");
versions.push(tree.get_record(key.as_bytes()).unwrap().unwrap().version);
}
}
let tree = Tree::open(cfg).unwrap();
for i in 0..16u32 {
let key = format!("bulk/object-{i:04}");
let value = format!("metadata-{i:04}");
let record = tree.get_record(key.as_bytes()).unwrap().unwrap();
assert_eq!(record.value, value.as_bytes());
assert_eq!(
record.version, versions[i as usize],
"compact insert-run replay must preserve per-inner record versions",
);
}
}
#[test]
fn batch_conditional_ops_replay_with_stable_versions() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
let seed_version_after_atomic;
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"seed", b"v1").unwrap();
let seed_v1 = tree.get_record(b"seed").unwrap().unwrap().version;
assert!(tree
.atomic(|b| {
b.delete(b"missing");
b.compare_and_put(b"seed", seed_v1, b"v2");
b.put_if_absent(b"created", b"new");
})
.unwrap());
seed_version_after_atomic = tree.get_record(b"seed").unwrap().unwrap().version;
}
let tree = Tree::open(cfg).unwrap();
let seed = tree.get_record(b"seed").unwrap().unwrap();
assert_eq!(seed.value, b"v2");
assert_eq!(
seed.version, seed_version_after_atomic,
"Batch replay must preserve per-inner seq even when an earlier inner op is a no-op",
);
assert_eq!(tree.get(b"created").unwrap().as_deref(), Some(&b"new"[..]));
}
#[test]
fn batch_prefix_assertions_do_not_shift_replay_versions() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
let k1_version;
let k2_version;
{
let tree = Tree::open(cfg.clone()).unwrap();
assert!(tree
.atomic(|b| {
b.assert_prefix_empty(b"guard/");
b.put(b"batch/k1", b"v1");
b.assert_prefix_empty(b"other/");
b.put(b"batch/k2", b"v2");
})
.unwrap());
k1_version = tree.get_record(b"batch/k1").unwrap().unwrap().version;
k2_version = tree.get_record(b"batch/k2").unwrap().unwrap().version;
}
let tree = Tree::open(cfg).unwrap();
let k1 = tree.get_record(b"batch/k1").unwrap().unwrap();
let k2 = tree.get_record(b"batch/k2").unwrap().unwrap();
assert_eq!(k1.value, b"v1");
assert_eq!(k2.value, b"v2");
assert_eq!(
k1.version, k1_version,
"prefix assertions must not consume Batch WAL inner sequence numbers",
);
assert_eq!(
k2.version, k2_version,
"prefix assertions must not shift later replay versions",
);
}
#[test]
fn batch_version_assertions_do_not_shift_replay_versions() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
let seed_version;
let copied_version;
let later_version;
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"seed", b"payload").unwrap();
let seed = tree.get_record(b"seed").unwrap().unwrap();
assert!(tree
.atomic(|b| {
b.assert_version(b"seed", seed.version);
b.put(b"copied", &seed.value);
b.assert_version(b"seed", seed.version);
b.put(b"later", b"v2");
})
.unwrap());
seed_version = tree.get_record(b"seed").unwrap().unwrap().version;
copied_version = tree.get_record(b"copied").unwrap().unwrap().version;
later_version = tree.get_record(b"later").unwrap().unwrap().version;
}
let tree = Tree::open(cfg).unwrap();
let seed = tree.get_record(b"seed").unwrap().unwrap();
let copied = tree.get_record(b"copied").unwrap().unwrap();
let later = tree.get_record(b"later").unwrap().unwrap();
assert_eq!(seed.value, b"payload");
assert_eq!(copied.value, b"payload");
assert_eq!(later.value, b"v2");
assert_eq!(
seed.version, seed_version,
"assert_version must not rewrite the guarded source",
);
assert_eq!(
copied.version, copied_version,
"assert_version must not consume Batch WAL inner sequence numbers",
);
assert_eq!(
later.version, later_version,
"assert_version must not shift later replay versions",
);
}
#[test]
fn atomic_assert_only_batch_does_not_append_wal_or_consume_seq() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"seed", b"payload").unwrap();
let seed = tree.get_record(b"seed").unwrap().unwrap();
let wal_len = fs::metadata(wal_path(dir.path())).unwrap().len();
assert!(tree
.atomic(|b| {
b.assert_version(b"seed", seed.version);
b.assert_prefix_empty(b"empty/");
})
.unwrap());
assert_eq!(
fs::metadata(wal_path(dir.path())).unwrap().len(),
wal_len,
"assert-only atomic batches must not emit WAL records",
);
assert_eq!(
tree.get_record(b"seed").unwrap().unwrap().version,
seed.version,
"assert-only atomic batches must not consume record versions",
);
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(tree.get(b"seed").unwrap().as_deref(), Some(&b"payload"[..]));
}
#[test]
fn failed_atomic_guard_does_not_append_wal_or_publish() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"guarded", b"v1").unwrap();
let stale = tree.get_version(b"guarded").unwrap().unwrap();
tree.put(b"guarded", b"v2").unwrap();
let wal_len = fs::metadata(wal_path(dir.path())).unwrap().len();
let committed = tree
.atomic(|b| {
b.assert_version(b"guarded", stale);
b.put(b"side", b"should-not-publish");
})
.unwrap();
assert!(!committed);
assert_eq!(
fs::metadata(wal_path(dir.path())).unwrap().len(),
wal_len,
"failed preflight must not append a Batch WAL record",
);
assert!(tree.get(b"side").unwrap().is_none());
assert_eq!(tree.get(b"guarded").unwrap().as_deref(), Some(&b"v2"[..]));
}
let tree = Tree::open(cfg).unwrap();
assert!(tree.get(b"side").unwrap().is_none());
assert_eq!(tree.get(b"guarded").unwrap().as_deref(), Some(&b"v2"[..]));
}
#[test]
fn batch_crash_before_flush_loses_whole_batch() {
let dir = tempdir().unwrap();
let cfg = manual_checkpoint_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"durable", b"D").unwrap();
tree.checkpoint().unwrap();
tree.atomic(|b| {
b.put(b"vanish-a", b"VA");
b.put(b"vanish-b", b"VB");
})
.unwrap();
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(tree.get(b"durable").unwrap().as_deref(), Some(&b"D"[..]));
assert!(tree.get(b"vanish-a").unwrap().is_none());
assert!(tree.get(b"vanish-b").unwrap().is_none());
}
#[test]
fn background_checkpointer_truncates_wal_and_keeps_data_durable() {
use holt::{CheckpointConfig, TreeBuilder};
use std::thread;
use std::time::{Duration, Instant};
let dir = tempdir().unwrap();
{
let tree = TreeBuilder::new(dir.path())
.checkpoint(CheckpointConfig {
enabled: true,
idle_interval: Duration::from_millis(25),
dirty_blob_threshold: 1,
auto_merge: true,
..CheckpointConfig::default()
})
.open()
.unwrap();
for i in 0..500u32 {
tree.put(format!("bg/{i:04}").as_bytes(), format!("v-{i}").as_bytes())
.unwrap();
}
let header_size_after_truncate = 32u64; let deadline = Instant::now() + Duration::from_secs(15);
loop {
let wal_len = fs::metadata(wal_path(dir.path())).unwrap().len();
if wal_len <= header_size_after_truncate + 128 {
break;
}
assert!(
Instant::now() < deadline,
"background checkpointer never truncated WAL (size={wal_len})",
);
thread::sleep(Duration::from_millis(20));
}
}
let tree = Tree::open(TreeConfig::new(dir.path())).unwrap();
for i in 0..500u32 {
let k = format!("bg/{i:04}");
let want = format!("v-{i}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(want.as_bytes()),
"key {k} lost after bg-checkpoint-and-reopen",
);
}
}
#[test]
fn spillover_new_blobs_deferred_to_store_until_checkpoint() {
use holt::{BlobStore, MemoryBlobStore};
use std::sync::Arc;
let inner: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
let mut cfg = TreeConfig::memory();
cfg.memory_flush_on_write = false;
let tree = Tree::open_with_blob_store(cfg, Arc::clone(&inner)).unwrap();
let initial = inner.list_blobs().unwrap();
assert_eq!(initial.len(), 1, "open seeds only the root blob");
let payload = vec![b'x'; 1024];
for i in 0..1000u32 {
tree.put(format!("k{i:05}").as_bytes(), &payload).unwrap();
}
let stats = tree.stats().unwrap();
assert!(
stats.blob_count > 1,
"spillover should have created at least one child blob (got {})",
stats.blob_count,
);
assert!(
stats.bm_dirty_count >= 1,
"every spillover'd blob + root must be tracked dirty",
);
let mid = inner.list_blobs().unwrap();
assert_eq!(
mid.len(),
1,
"inner store must NOT see spillover'd children until checkpoint (got {} blobs)",
mid.len(),
);
tree.checkpoint().unwrap();
let stats_after = tree.stats().unwrap();
assert_eq!(
stats_after.bm_dirty_count, 0,
"checkpoint must drain every dirty entry",
);
let final_blobs = inner.list_blobs().unwrap();
assert_eq!(
final_blobs.len() as u32,
stats.blob_count,
"after checkpoint, inner store has every reachable blob",
);
for i in 0..1000u32 {
let k = format!("k{i:05}");
let v = tree.get(k.as_bytes()).unwrap().expect("key present");
assert_eq!(v.as_slice(), payload.as_slice(), "value drift for {k}");
}
}
#[test]
fn k8s_shaped_hot_prefix_updates_do_not_expose_spillover_gap() {
let dir = tempdir().unwrap();
let mut cfg = TreeConfig::new(dir.path());
cfg.durability = Durability::Wal { sync: false };
let tree = Tree::open(cfg).unwrap();
let prefixes = [
"/registry/p0",
"/registry/p1",
"/registry/p2",
"/registry/p3",
"/registry/p4",
"/registry/p5",
"/registry/p6",
"/registry/p7",
];
let mut value = vec![0u8; 24 * 1024];
for i in 0..768u64 {
let prefix = prefixes[(i as usize) % prefixes.len()];
let obj = i / prefixes.len() as u64;
value.fill((i & 0xff) as u8);
put_k8s_update(&tree, prefix, obj, i + 1, &value);
}
tree.checkpoint().unwrap();
for round in 0..64u64 {
assert!(tree
.atomic(|batch| {
for slot in 0..16u64 {
let prefix = prefixes[((round + slot) as usize) % prefixes.len()];
let obj = (round * 16 + slot) % 128;
let rev = round * 16 + slot + 1024;
value.fill((rev & 0xff) as u8);
let object_key = format!("{prefix}/objects/{obj:08}");
let head_key = format!("{prefix}/heads/{obj:08}");
let event_key = format!("{prefix}/watch/{rev:012}");
let head_value = format!("rev={rev};object={object_key}");
batch.put(object_key.as_bytes(), &value);
batch.put(head_key.as_bytes(), head_value.as_bytes());
batch.put(event_key.as_bytes(), object_key.as_bytes());
}
})
.unwrap());
}
for prefix in prefixes {
for obj in 0..8u64 {
let head_key = format!("{prefix}/heads/{obj:08}");
assert!(tree.get(head_key.as_bytes()).unwrap().is_some());
}
}
}
fn put_k8s_update(tree: &Tree, prefix: &str, obj: u64, rev: u64, value: &[u8]) {
assert!(tree
.atomic(|batch| {
let object_key = format!("{prefix}/objects/{obj:08}");
let head_key = format!("{prefix}/heads/{obj:08}");
let event_key = format!("{prefix}/watch/{rev:012}");
let head_value = format!("rev={rev};object={object_key}");
batch.put(object_key.as_bytes(), value);
batch.put(head_key.as_bytes(), head_value.as_bytes());
batch.put(event_key.as_bytes(), object_key.as_bytes());
})
.unwrap());
}
#[test]
#[ignore = "heavy issue #23 reproducer; run explicitly in release mode"]
fn issue_23_k8s_shaped_spillover_reproducer() {
let dir = tempdir().unwrap();
let mut cfg = TreeConfig::new(dir.path());
cfg.durability = Durability::Wal { sync: false };
let tree = Arc::new(Tree::open(cfg).unwrap());
let prefixes = [
"/registry/p0",
"/registry/p1",
"/registry/p2",
"/registry/p3",
"/registry/p4",
"/registry/p5",
"/registry/p6",
"/registry/p7",
];
for i in 0..20_000u64 {
let prefix = prefixes[(i as usize) % prefixes.len()];
let obj = i / prefixes.len() as u64;
let mut value = vec![0u8; issue_23_value_len(i)];
value.fill((i & 0xff) as u8);
put_k8s_update(&tree, prefix, obj, i + 1, &value);
}
tree.checkpoint().unwrap();
let barrier = Arc::new(Barrier::new(5));
let mut handles = Vec::new();
for worker in 0..4u64 {
let tree = Arc::clone(&tree);
let barrier = Arc::clone(&barrier);
handles.push(thread::spawn(move || {
barrier.wait();
for round in 0..128u64 {
let base = 20_000 + worker * 128 * 16 + round * 16;
assert!(tree
.atomic(|batch| {
for slot in 0..16u64 {
let rev = base + slot + 1;
let prefix = prefixes[((rev + worker) as usize) % prefixes.len()];
let obj = rev % 2_500;
let mut value = vec![0u8; issue_23_value_len(rev)];
value.fill((rev & 0xff) as u8);
let object_key = format!("{prefix}/objects/{obj:08}");
let head_key = format!("{prefix}/heads/{obj:08}");
let event_key = format!("{prefix}/watch/{rev:012}");
let head_value = format!("rev={rev};object={object_key}");
batch.put(object_key.as_bytes(), &value);
batch.put(head_key.as_bytes(), head_value.as_bytes());
batch.put(event_key.as_bytes(), object_key.as_bytes());
}
})
.unwrap());
}
}));
}
barrier.wait();
for handle in handles {
handle.join().unwrap();
}
tree.checkpoint().unwrap();
let stats = tree.stats().unwrap();
assert!(stats.blob_count > 1);
}
fn issue_23_value_len(seed: u64) -> usize {
20 * 1024 + ((seed as usize * 7919) % (20 * 1024 + 1))
}
#[test]
fn compact_does_not_leak_pre_wal_state_to_store() {
use holt::{BlobStore, MemoryBlobStore};
use std::sync::Arc;
let inner: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
let mut cfg = TreeConfig::memory();
cfg.memory_flush_on_write = false; let tree = Tree::open_with_blob_store(cfg, Arc::clone(&inner)).unwrap();
let initial = inner.list_blobs().unwrap();
assert_eq!(initial.len(), 1, "open seeds only the root blob");
tree.put(b"key", b"value").unwrap();
assert!(
tree.stats().unwrap().bm_dirty_count >= 1,
"put must leave root dirty",
);
tree.compact().unwrap();
let after_compact = inner.list_blobs().unwrap();
assert_eq!(
after_compact.len(),
1,
"compact must not push cache state to store",
);
assert!(
tree.stats().unwrap().bm_dirty_count >= 1,
"compact must leave dirty entries (not auto-flush)",
);
tree.checkpoint().unwrap();
assert_eq!(tree.stats().unwrap().bm_dirty_count, 0);
assert_eq!(tree.get(b"key").unwrap().as_deref(), Some(&b"value"[..]),);
}
#[test]
fn multi_blob_compact_does_not_leak_pre_wal_state_to_store() {
use holt::{BlobStore, MemoryBlobStore};
use std::sync::Arc;
let inner: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
let mut cfg = TreeConfig::memory();
cfg.memory_flush_on_write = false;
let tree = Tree::open_with_blob_store(cfg, Arc::clone(&inner)).unwrap();
assert_eq!(inner.list_blobs().unwrap().len(), 1);
let payload = vec![b'q'; 1024];
for i in 0..1500u32 {
tree.put(format!("k{i:05}").as_bytes(), &payload).unwrap();
}
let stats = tree.stats().unwrap();
assert!(
stats.blob_count > 1,
"multi-blob compact precondition: spillover must trigger (got {} blobs)",
stats.blob_count,
);
tree.compact().unwrap();
let after_compact = inner.list_blobs().unwrap();
assert_eq!(
after_compact.len(),
1,
"multi-blob compact must not push cache state to store (got {} blobs)",
after_compact.len(),
);
assert!(
tree.stats().unwrap().bm_dirty_count >= 1,
"compact must leave dirty entries waiting for the next checkpoint",
);
tree.checkpoint().unwrap();
assert_eq!(tree.stats().unwrap().bm_dirty_count, 0);
for i in 0..1500u32 {
let k = format!("k{i:05}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(payload.as_slice()),
"key {k} lost after multi-blob compact + checkpoint",
);
}
}
#[test]
fn tree_stats_does_not_perturb_cache_counters_or_lru() {
let tree = Tree::open(TreeConfig::memory()).unwrap();
let payload = vec![b'q'; 1024];
for i in 0..800u32 {
tree.put(format!("k{i:05}").as_bytes(), &payload).unwrap();
}
let baseline = tree.stats().unwrap();
let baseline_hits = baseline.bm_cache_hits;
let baseline_misses = baseline.bm_cache_misses;
assert!(
baseline.blob_count > 1,
"test premise: multi-blob tree (got {})",
baseline.blob_count,
);
for _ in 0..50 {
let s = tree.stats().unwrap();
assert_eq!(
s.bm_cache_hits, baseline_hits,
"Tree::stats() must not bump cache_hits",
);
assert_eq!(
s.bm_cache_misses, baseline_misses,
"Tree::stats() must not bump cache_misses",
);
}
}
#[test]
fn indexed_read_touches_one_child_blob_after_reopen() {
let dir = tempdir().unwrap();
let mut cfg = manual_checkpoint_cfg(dir.path());
cfg.buffer_pool_size = 1;
let target_key = b"cold/bucket/table/part-0999/object";
let target_value = vec![b'm'; 96];
{
let tree = Tree::open(cfg.clone()).unwrap();
let payload = vec![b'x'; 1024];
for i in 0..1000u32 {
let key = format!("cold/bucket/table/part-{i:04}/object");
let value = if key.as_bytes() == target_key {
&target_value
} else {
&payload
};
tree.put(key.as_bytes(), value).unwrap();
}
assert!(
tree.stats().unwrap().blob_count > 1,
"test must force at least one child blob"
);
tree.checkpoint().unwrap();
for _ in 0..8 {
tree.compact().unwrap();
tree.checkpoint().unwrap();
}
assert!(
tree.stats().unwrap().total_compactions > 0,
"test must route at least one cold child blob",
);
}
let tree = Tree::open(cfg).unwrap();
let before = tree.stats().unwrap();
assert_eq!(
tree.get(target_key).unwrap().as_deref(),
Some(target_value.as_slice())
);
let after = tree.stats().unwrap();
assert!(
after.bm_point_full_blob_reads <= before.bm_point_full_blob_reads + 1,
"point get must touch one child blob, not all (before={}, after={})",
before.bm_point_full_blob_reads,
after.bm_point_full_blob_reads,
);
let before_miss = tree.stats().unwrap();
assert!(tree
.get(b"cold/bucket/table/part-0999/missing")
.unwrap()
.is_none());
let after_miss = tree.stats().unwrap();
assert!(
after_miss.bm_point_full_blob_reads <= before_miss.bm_point_full_blob_reads + 1,
"cold miss fallback must pin at most one child blob (before={}, after={})",
before_miss.bm_point_full_blob_reads,
after_miss.bm_point_full_blob_reads,
);
}
#[test]
fn indexed_large_value_read_uses_value_segment_after_reopen() {
let dir = tempdir().unwrap();
let mut cfg = manual_checkpoint_cfg(dir.path());
cfg.buffer_pool_size = 1;
let target_key = b"cold/value/bucket/part-0999/object";
let target_value = vec![0xA7; 4096];
{
let tree = Tree::open(cfg.clone()).unwrap();
let filler = vec![0x11; 4096];
for i in 0..1000u32 {
let key = format!("cold/value/bucket/part-{i:04}/object");
let value = if key.as_bytes() == target_key {
target_value.as_slice()
} else {
filler.as_slice()
};
tree.put(key.as_bytes(), value).unwrap();
}
tree.checkpoint().unwrap();
for _ in 0..8 {
tree.compact().unwrap();
tree.checkpoint().unwrap();
}
assert!(
tree.stats().unwrap().total_compactions > 0,
"test must route at least one cold child blob",
);
}
let tree = Tree::open(cfg).unwrap();
assert_eq!(
tree.get(target_key).unwrap().as_deref(),
Some(target_value.as_slice())
);
}
#[test]
fn db_indexed_route_neighbor_get_falls_back_after_local_route_answer() {
let dir = tempdir().unwrap();
let mut cfg = durable_cfg(dir.path());
cfg.buffer_pool_size = 64;
let key0 = b"db-crash/d15ea5ed500a0001/0000000000000000";
let key1 = b"db-crash/d15ea5ed500a0001/0000000000000001";
let object0 = b"crash-value:10812959806829638721";
let inode0 = b"crash-value:14416686724497167465";
let object1 = b"crash-value:3592466528075977345";
let inode1 = b"crash-value:2198126064033428708";
{
let db = DB::open(cfg.clone()).unwrap();
db.open_or_create_tree("objects").unwrap();
db.open_or_create_tree("inodes").unwrap();
let filler = vec![b'x'; 1024];
for i in 0..1000u32 {
let key = format!("db-crash/d15ea5ed500a0001/{i:016x}");
let object = if key.as_bytes() == key0 {
object0.as_slice()
} else if key.as_bytes() == key1 {
object1.as_slice()
} else {
filler.as_slice()
};
let inode = if key.as_bytes() == key0 {
inode0.as_slice()
} else if key.as_bytes() == key1 {
inode1.as_slice()
} else {
filler.as_slice()
};
db.atomic(|batch| {
batch.put("objects", key.as_bytes(), object);
batch.put("inodes", key.as_bytes(), inode);
})
.unwrap();
}
db.checkpoint().unwrap();
for _ in 0..8 {
db.compact().unwrap();
db.checkpoint().unwrap();
}
}
cfg.buffer_pool_size = 1;
let db = DB::open(cfg).unwrap();
let objects = db.open_tree("objects").unwrap();
let inodes = db.open_tree("inodes").unwrap();
assert_eq!(
objects.get(key0).unwrap().as_deref(),
Some(object0.as_slice())
);
assert_eq!(
objects.get(key1).unwrap().as_deref(),
Some(object1.as_slice())
);
assert_eq!(
inodes.get(key0).unwrap().as_deref(),
Some(inode0.as_slice())
);
assert_eq!(
inodes.get(key1).unwrap().as_deref(),
Some(inode1.as_slice())
);
}
#[test]
fn batch_replay_then_checkpoint_then_reopen_preserves_data() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.atomic(|b| {
b.put(b"a", b"1");
b.put(b"b", b"2");
b.put(b"c", b"3");
b.delete(b"a");
b.rename(b"b", b"B", false);
})
.unwrap();
}
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.checkpoint().unwrap();
}
{
let tree = Tree::open(cfg).unwrap();
assert_eq!(tree.get(b"a").unwrap(), None);
assert_eq!(tree.get(b"B").unwrap().as_deref(), Some(&b"2"[..]));
assert_eq!(tree.get(b"c").unwrap().as_deref(), Some(&b"3"[..]));
}
}
#[test]
fn rename_replay_is_idempotent_across_two_drops() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
tree.put(b"src", b"v1").unwrap();
tree.rename(b"src", b"dst", false).unwrap();
}
{
let tree = Tree::open(cfg.clone()).unwrap();
assert!(tree.get(b"src").unwrap().is_none());
assert_eq!(tree.get(b"dst").unwrap().as_deref(), Some(&b"v1"[..]));
}
{
let tree = Tree::open(cfg.clone()).unwrap();
assert!(tree.get(b"src").unwrap().is_none());
assert_eq!(tree.get(b"dst").unwrap().as_deref(), Some(&b"v1"[..]));
tree.checkpoint().unwrap();
}
{
let tree = Tree::open(cfg).unwrap();
assert!(tree.get(b"src").unwrap().is_none());
assert_eq!(tree.get(b"dst").unwrap().as_deref(), Some(&b"v1"[..]));
}
}
#[test]
fn subtree_gone_replay_reconstructs_correctly() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
let payload = vec![b'z'; 1024];
for i in 0..1000u32 {
tree.put(format!("k{i:05}").as_bytes(), &payload).unwrap();
}
for i in 0..500u32 {
tree.delete(format!("k{i:05}").as_bytes()).unwrap();
}
}
{
let tree = Tree::open(cfg.clone()).unwrap();
for i in 0..500u32 {
assert!(
tree.get(format!("k{i:05}").as_bytes()).unwrap().is_none(),
"deleted key k{i:05} resurrected on replay",
);
}
for i in 500..1000u32 {
assert!(
tree.get(format!("k{i:05}").as_bytes()).unwrap().is_some(),
"surviving key k{i:05} lost on replay",
);
}
tree.checkpoint().unwrap();
}
{
let tree = Tree::open(cfg).unwrap();
for i in 0..500u32 {
assert!(tree.get(format!("k{i:05}").as_bytes()).unwrap().is_none());
}
for i in 500..1000u32 {
assert!(tree.get(format!("k{i:05}").as_bytes()).unwrap().is_some());
}
}
}
#[test]
fn cross_blob_writes_replay_correctly_through_wal_without_checkpoint() {
let dir = tempdir().unwrap();
let cfg = durable_cfg(dir.path());
{
let tree = Tree::open(cfg.clone()).unwrap();
let payload = vec![b'y'; 1024];
for i in 0..1000u32 {
tree.put(format!("k{i:05}").as_bytes(), &payload).unwrap();
}
let stats = tree.stats().unwrap();
assert!(
stats.blob_count > 1,
"test premise: spillover must trigger (got blob_count={})",
stats.blob_count,
);
}
{
let tree = Tree::open(cfg).unwrap();
let payload = vec![b'y'; 1024];
for i in 0..1000u32 {
let k = format!("k{i:05}");
assert_eq!(
tree.get(k.as_bytes()).unwrap().as_deref(),
Some(payload.as_slice()),
"cross-blob key {k} lost after replay",
);
}
}
}
#[test]
fn concurrent_writers_and_bg_checkpoint_preserve_acked_ops() {
use holt::{CheckpointConfig, TreeBuilder};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
const WRITERS: usize = 4;
const OPS_PER_WRITER: u32 = 250;
const PAYLOAD_LEN: usize = 256;
let dir = tempdir().unwrap();
let payload = vec![b'y'; PAYLOAD_LEN];
{
let tree = Arc::new(
TreeBuilder::new(dir.path())
.durability(holt::Durability::Wal { sync: true }) .checkpoint(CheckpointConfig {
enabled: true,
idle_interval: Duration::from_millis(5),
dirty_blob_threshold: 1,
auto_merge: false,
..CheckpointConfig::default()
})
.open()
.unwrap(),
);
let handles: Vec<_> = (0..WRITERS)
.map(|writer_id| {
let tree = Arc::clone(&tree);
let payload = payload.clone();
thread::spawn(move || {
for i in 0..OPS_PER_WRITER {
let key = format!("w{writer_id:02}/k{i:05}");
tree.put(key.as_bytes(), &payload).unwrap();
}
})
})
.collect();
for h in handles {
h.join().unwrap();
}
thread::sleep(Duration::from_millis(100));
}
{
let tree = Tree::open(TreeConfig::new(dir.path())).unwrap();
for writer_id in 0..WRITERS {
for i in 0..OPS_PER_WRITER {
let key = format!("w{writer_id:02}/k{i:05}");
assert_eq!(
tree.get(key.as_bytes()).unwrap().as_deref(),
Some(payload.as_slice()),
"key {key} lost after concurrent put + bg checkpoint + reopen",
);
}
}
}
}
#[test]
fn concurrent_writers_and_manual_checkpoints_preserve_acked_ops() {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
const WRITERS: usize = 4;
const OPS_PER_WRITER: u32 = 200;
const PAYLOAD_LEN: usize = 256;
let dir = tempdir().unwrap();
let payload = vec![b'y'; PAYLOAD_LEN];
{
let cfg = durable_cfg(dir.path()); let tree = Arc::new(Tree::open(cfg).unwrap());
let done = Arc::new(AtomicBool::new(false));
let ck_tree = Arc::clone(&tree);
let ck_done = Arc::clone(&done);
let ck_handle = thread::spawn(move || {
while !ck_done.load(Ordering::Relaxed) {
let _ = ck_tree.checkpoint();
thread::sleep(Duration::from_millis(5));
}
});
let writer_handles: Vec<_> = (0..WRITERS)
.map(|writer_id| {
let tree = Arc::clone(&tree);
let payload = payload.clone();
thread::spawn(move || {
for i in 0..OPS_PER_WRITER {
let key = format!("w{writer_id:02}/k{i:05}");
tree.put(key.as_bytes(), &payload).unwrap();
}
})
})
.collect();
for h in writer_handles {
h.join().unwrap();
}
done.store(true, Ordering::Relaxed);
ck_handle.join().unwrap();
tree.checkpoint().unwrap();
}
{
let tree = Tree::open(durable_cfg(dir.path())).unwrap();
for writer_id in 0..WRITERS {
for i in 0..OPS_PER_WRITER {
let key = format!("w{writer_id:02}/k{i:05}");
assert_eq!(
tree.get(key.as_bytes()).unwrap().as_deref(),
Some(payload.as_slice()),
"key {key} lost after concurrent put + manual checkpoint + reopen",
);
}
}
}
}
#[test]
fn concurrent_writers_with_deletes_and_bg_checkpoint() {
use holt::{CheckpointConfig, TreeBuilder};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
const PUT_WRITERS: usize = 3;
const OPS: u32 = 200;
const PAYLOAD_LEN: usize = 512;
let dir = tempdir().unwrap();
let payload = vec![b'z'; PAYLOAD_LEN];
{
let tree = Arc::new(
TreeBuilder::new(dir.path())
.durability(holt::Durability::Wal { sync: true })
.checkpoint(CheckpointConfig {
enabled: true,
idle_interval: Duration::from_millis(5),
dirty_blob_threshold: 1,
auto_merge: true,
..CheckpointConfig::default()
})
.open()
.unwrap(),
);
for writer_id in 0..PUT_WRITERS {
for i in 0..OPS {
let k = format!("p{writer_id:02}/k{i:05}");
tree.put(k.as_bytes(), &payload).unwrap();
}
}
let putters: Vec<_> = (0..PUT_WRITERS)
.map(|writer_id| {
let tree = Arc::clone(&tree);
let payload = payload.clone();
thread::spawn(move || {
for i in 0..OPS {
let k = format!("q{writer_id:02}/k{i:05}");
tree.put(k.as_bytes(), &payload).unwrap();
}
})
})
.collect();
let deleter_tree = Arc::clone(&tree);
let deleter = thread::spawn(move || {
for writer_id in 0..PUT_WRITERS {
for i in 0..OPS {
let k = format!("p{writer_id:02}/k{i:05}");
let _ = deleter_tree.delete(k.as_bytes()).unwrap();
}
}
});
for h in putters {
h.join().unwrap();
}
deleter.join().unwrap();
thread::sleep(Duration::from_millis(150));
}
{
let tree = Tree::open(TreeConfig::new(dir.path())).unwrap();
for writer_id in 0..PUT_WRITERS {
for i in 0..OPS {
let kq = format!("q{writer_id:02}/k{i:05}");
assert_eq!(
tree.get(kq.as_bytes()).unwrap().as_deref(),
Some(payload.as_slice()),
"{kq} (put) lost",
);
let kp = format!("p{writer_id:02}/k{i:05}");
assert!(
tree.get(kp.as_bytes()).unwrap().is_none(),
"{kp} (deleted) resurrected",
);
}
}
}
}