use memdb::{Durability, MemDb, Page}; use serde::{Deserialize, Serialize}; use std::sync::Arc; use std::time::{Duration, Instant}; // ─── Test models ───────────────────────────────────────────────────────── #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] struct Record { id: String, name: String, amount: u64, tags: Vec, metadata: String, } impl Record { fn new(id: usize) -> Self { let tag_count = (id % 5) as usize + 1; Self { id: format!("rec_{:06}", id), name: format!("record_{}", id), amount: (id * 7 % 1_000_000) as u64 + 1, tags: (0..tag_count) .map(|t| format!("tag_{:02}", (id + t) % 20)) .collect(), metadata: format!("meta data blob for record {}", id), } } } #[derive(Debug, Clone, Serialize, Deserialize)] struct Counter { value: u64, updates: u64, } #[derive(Debug, Clone, Serialize, Deserialize)] struct SimpleVal { val: u64, } #[derive(Debug, Serialize, Deserialize)] struct Blob { id: usize, data: Vec, } // ─── Helpers ───────────────────────────────────────────────────────────── fn report(name: &str, count: u64, elapsed_ms: u64) { let ops_per_sec = if elapsed_ms > 0 { count * 1000 / elapsed_ms } else { count }; println!(" [{name}] {count} ops in {elapsed_ms}ms → {ops_per_sec} ops/sec"); } // ─── 1. Bulk insert performance ────────────────────────────────────────── #[test] fn stress_bulk_insert_performance() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 10_000u64; let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(col.count(), n as usize); report("bulk insert", n, elapsed); } // ─── 2. Bulk read performance ──────────────────────────────────────────── #[test] fn stress_bulk_read_performance() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 10_000u64; for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let start = Instant::now(); for i in 0..n { let _found: Option = col.get(&format!("{:06}", i)).unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("bulk read (by key)", n, elapsed); } // ─── 3. Bulk update performance ────────────────────────────────────────── #[test] fn stress_bulk_update_performance() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 5_000u64; for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let start = Instant::now(); for i in 0..n { let key = format!("{:06}", i); col.update(&key, |mut r: Record| { r.amount += 1; r }) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("bulk update", n, elapsed); let r: Record = col.get_required("000000").unwrap(); assert!(r.amount > 0); } // ─── 4. Prefix scan performance ────────────────────────────────────────── #[test] fn stress_prefix_scan_performance() { let db = MemDb::in_memory(); let col = db.collection("events"); for day in 1..=100 { for seq in 1..=200 { let id = format!("2026_{:03}_{:05}", day, seq); let rec = Record::new((day * 1000 + seq) as usize); col.insert(id, &rec).unwrap(); } } let start = Instant::now(); let results: Vec = col.scan_prefix("2026_050_").unwrap(); let elapsed = start.elapsed().as_millis() as u64; assert_eq!(results.len(), 200); report("prefix scan (200 of 20k)", 200, elapsed); let start = Instant::now(); let results: Vec = col.scan_prefix("9999_").unwrap(); let empty_ms = start.elapsed().as_millis() as u64; assert!(results.is_empty()); println!(" [prefix scan empty] in {empty_ms}ms"); } // ─── 5. Pagination performance ─────────────────────────────────────────── #[test] fn stress_pagination_performance() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 10_000u64; for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let start = Instant::now(); let page_size = 50u64; let total_pages = (n + page_size - 1) / page_size; let mut total_items = 0u64; for p in 1..=total_pages { let page = col .paginate::(&Page::new(p, page_size), false) .unwrap(); total_items += page.items.len() as u64; } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(total_items, n); report(&format!("paginate {} pages", total_pages), n, elapsed); } // ─── 6. Filter performance ─────────────────────────────────────────────── #[test] fn stress_filter_performance() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 10_000u64; for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let start = Instant::now(); let filtered: Vec = col .filter(|r: &Record| r.amount > 500_000 && r.tags.contains(&"tag_05".to_string())) .unwrap(); let elapsed = start.elapsed().as_millis() as u64; println!( " [filter] matched {}/{} in {}ms", filtered.len(), n, elapsed ); } // ─── 7. Concurrent writes ──────────────────────────────────────────────── #[tokio::test] async fn stress_concurrent_writes_no_data_loss() { let db = Arc::new(MemDb::in_memory()); let n_writers = 8u64; let n_per_writer = 1_250u64; let total = n_writers * n_per_writer; let start = Instant::now(); let mut handles = vec![]; for w in 0..n_writers { let db = db.clone(); handles.push(tokio::task::spawn_blocking(move || { for i in 0..n_per_writer { let id = format!("w{:02}_{:06}", w, i); let rec = Record::new((w * 10000 + i) as usize); db.collection("records").upsert(id, &rec).unwrap(); } })); } for h in handles { h.await.unwrap(); } let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("records").count() as u64; assert_eq!(count, total); report("concurrent writes (8×1250)", total, elapsed); } // ─── 8. Concurrent reads and writes ────────────────────────────────────── #[tokio::test] async fn stress_concurrent_read_write() { let db = Arc::new(MemDb::in_memory()); let n = 5_000u64; for i in 0..n { db.collection("records") .insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let start = Instant::now(); let db_w = db.clone(); let write_handle = tokio::task::spawn_blocking(move || { for i in n..n + 2_000 { let id = format!("{:06}", i); let rec = Record::new(i as usize); db_w.collection("records").upsert(id, &rec).unwrap(); } }); let db_r = db.clone(); let read_handle = tokio::task::spawn_blocking(move || { for i in 0..2_000u64 { let key = format!("{:06}", i % n); let _: Option = db_r.collection("records").get(&key).unwrap(); } }); let (wr, rr) = tokio::join!(write_handle, read_handle); wr.unwrap(); rr.unwrap(); let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("records").count() as u64; assert_eq!(count, n + 2_000); report("concurrent r/w (2k+2k)", 4_000, elapsed); } // ─── 9. Snapshot with concurrent writes (data integrity) ───────────────── #[tokio::test] async fn stress_snapshot_with_concurrent_writes() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let n_writes = 500u64; { let db = Arc::new(MemDb::open(&db_path).unwrap()); let db_w = db.clone(); let write_handle = tokio::task::spawn_blocking(move || { for i in 0..n_writes { let rec = Record::new(i as usize); db_w .collection("records") .upsert(format!("{:06}", i), &rec) .unwrap(); } }); let db_s = db.clone(); let snap_handle = tokio::spawn(async move { for _ in 0..10 { tokio::time::sleep(std::time::Duration::from_millis(5)).await; db_s.snapshot().unwrap(); } }); write_handle.await.unwrap(); snap_handle.await.unwrap(); db.snapshot().unwrap(); } // Recover and verify. { let db = MemDb::open(&db_path).unwrap(); let col = db.collection("records"); let count = col.count() as u64; println!(" [snapshot stress] wrote {n_writes}, recovered {count}"); assert_eq!( count, n_writes, "data loss: wrote {n_writes} but recovered {count}" ); for i in 0..n_writes { let key = format!("{:06}", i); assert!(col.exists(&key), "missing record: {key}"); } } } // ─── 10. Large dataset recovery ───────────────────────────────────────── #[test] fn stress_recover_large_dataset() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let n = 3_000u64; { let db = MemDb::open(&db_path).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("persistent insert", n, elapsed); db.snapshot().unwrap(); } { let start = Instant::now(); let db = MemDb::open(&db_path).unwrap(); let elapsed = start.elapsed().as_millis() as u64; println!(" [recovery] {n} records loaded in {elapsed}ms"); let col = db.collection("records"); assert_eq!(col.count() as u64, n); for i in 0..n { let key = format!("{:06}", i); assert!(col.exists(&key), "missing after recovery: {key}"); } } } // ─── 11. Transaction batch performance ─────────────────────────────────── #[test] fn stress_transaction_batch_performance() { let db = MemDb::in_memory(); let batch_size = 100usize; let batches = 100usize; let start = Instant::now(); for b in 0..batches { let mut txn = db.transaction(); for i in 0..batch_size { let id = format!("b{:03}_{:04}", b, i); let rec = Record::new(b * batch_size + i); txn = txn.upsert("records", &id, &rec).unwrap(); } txn.commit().unwrap(); } let elapsed = start.elapsed().as_millis() as u64; let total = (batches * batch_size) as u64; let count = db.collection("records").count() as u64; assert_eq!(count, total); report("txn batch (100×100)", total, elapsed); } // ─── 12. Cross-collection transactions ─────────────────────────────────── #[test] fn stress_cross_collection_transaction() { let db = MemDb::in_memory(); let n = 1_000u64; let start = Instant::now(); for i in 0..n { let rec = Record::new(i as usize); db.transaction() .upsert("records", format!("{:06}", i), &rec) .unwrap() .upsert("audit_log", format!("log_{:06}", i), &rec) .unwrap() .commit() .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(db.collection("records").count() as u64, n); assert_eq!(db.collection("audit_log").count() as u64, n); report("cross-collection txn", n, elapsed); } // ─── 13. Update-heavy workload ─────────────────────────────────────────── #[test] fn stress_update_heavy_workload() { let db = MemDb::in_memory(); let col = db.collection("counters"); let n = 1_000u64; for i in 0..n { col.insert( format!("counter_{:04}", i), &Counter { value: 0, updates: 0, }, ) .unwrap(); } let start = Instant::now(); let rounds = 10u64; for _ in 0..rounds { for i in 0..n { let key = format!("counter_{:04}", i); let _ = col .update::(&key, |mut c| { c.value += 1; c.updates += 1; c }) .unwrap(); } } let elapsed = start.elapsed().as_millis() as u64; let total_ops = rounds * n; let all: Vec = col.list_all().unwrap(); let total_value: u64 = all.iter().map(|c| c.value).sum(); let total_updates: u64 = all.iter().map(|c| c.updates).sum(); assert_eq!(total_value, total_updates); report("update-heavy (10×1000)", total_ops, elapsed); println!(" [update] total value={total_value}, total updates={total_updates}"); } // ─── 14. Mass delete ───────────────────────────────────────────────────── #[test] fn stress_mass_delete() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 5_000u64; for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } assert_eq!(col.count() as u64, n); let start = Instant::now(); for i in 0..n / 2 { let deleted = col.delete(format!("{:06}", i)).unwrap(); assert!(deleted); } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(col.count() as u64, n / 2); report("delete 2500 of 5000", n / 2, elapsed); } // ─── 15. Pagination edge cases ─────────────────────────────────────────── #[test] fn stress_pagination_edge_cases() { let db = MemDb::in_memory(); let col = db.collection("records"); let n = 100u64; for i in 0..n { col.insert(format!("{:06}", i), &SimpleVal { val: i }).unwrap(); } // Single-item page. let p1 = col .paginate::(&Page::new(1, 1), false) .unwrap(); assert_eq!(p1.items.len(), 1); assert_eq!(p1.total_pages, 100); // Out of range. let p2 = col .paginate::(&Page::new(999, 50), false) .unwrap(); assert!(p2.items.is_empty()); assert_eq!(p2.total, 100); // Out of range (descending). let p3 = col .paginate::(&Page::new(999, 50), true) .unwrap(); assert!(p3.items.is_empty()); // First page descending. let p4 = col .paginate::(&Page::new(1, 3), true) .unwrap(); assert_eq!(p4.items.len(), 3); // Last page ascending. let p5 = col .paginate::(&Page::new(34, 3), false) .unwrap(); assert_eq!(p5.items.len(), 1); assert_eq!(p5.total_pages, 34); } // ─── 16. WAL seq monotonicity under load ───────────────────────────────── #[test] fn stress_wal_seq_monotonic_under_load() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let n = 2_000; for i in 0..n { col.upsert(format!("{:06}", i), &Record::new(i)).unwrap(); } let wal_path = dir.path().join("wal.jsonl"); let entries = memdb::wal::read_after(&wal_path, 0).unwrap(); assert_eq!(entries.len(), n); let mut last = 0u64; for e in &entries { assert!(e.seq > last, "seq not monotonic: {} <= {}", e.seq, last); last = e.seq; } println!(" [wal seq] {n} entries, seq monotonic verified"); } // ─── 17. Snapshot truncation safety ───────────────────────────────────── #[test] fn stress_snapshot_truncate_safety() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); col.insert("a", &Record::new(1)).unwrap(); db.snapshot().unwrap(); // seq=1 snapshotted, WAL truncated col.insert("b", &Record::new(2)).unwrap(); // seq=2 must survive assert_eq!(col.count(), 2); drop(db); let db = MemDb::open(dir.path()).unwrap(); assert_eq!( db.collection("records").count(), 2, "data loss: record b missing after snapshot+truncate+recovery" ); println!(" [truncate safety] both records survived snapshot+truncate+recovery"); } // ─── 18. Multi-collection isolation ────────────────────────────────────── #[test] fn stress_multi_collection_isolation() { let db = MemDb::in_memory(); let n = 2_000; let collections = ["users", "orders", "products", "sessions", "audit_log"]; for (i, col_name) in collections.iter().enumerate() { let col = db.collection(col_name); let base_id = i * 10_000; for j in 0..n { let id = format!("{:06}", base_id + j); col.insert(&id, &Record::new(j)).unwrap(); } } for col_name in collections { assert_eq!( db.collection(col_name).count(), n, "collection {col_name} count mismatch" ); } println!(" [isolation] 5×2000 records, all counts verified"); } // ─── 19. Large value read/write ────────────────────────────────────────── #[test] fn stress_large_value_read_write() { let db = MemDb::in_memory(); let col = db.collection("blobs"); let blob_size = 64 * 1024; // 64 KiB let n = 100u64; let start = Instant::now(); for i in 0..n { let blob = Blob { id: i as usize, data: vec![(i % 256) as u8; blob_size], }; col.insert(format!("blob_{:04}", i), &blob).unwrap(); } let write_ms = start.elapsed().as_millis() as u64; report(&format!("large value write ({}×64KB)", n), n, write_ms); let start = Instant::now(); for i in 0..n { let blob: Blob = col.get_required(&format!("blob_{:04}", i)).unwrap(); assert_eq!(blob.id, i as usize); assert_eq!(blob.data.len(), blob_size); } let read_ms = start.elapsed().as_millis() as u64; report(&format!("large value read ({}×64KB)", n), n, read_ms); } // ─── 20. Realistic scenario: chat messages ─────────────────────────────── #[test] fn stress_realistic_chat_messages() { let db = MemDb::in_memory(); #[derive(Serialize, Deserialize)] struct Message { room_id: String, sender: String, text: String, ts: u64, } let users = ["alice", "bob", "charlie", "diana", "eve"]; let rooms = ["general", "random", "dev", "ops"]; let n_messages = 5_000u64; let start = Instant::now(); for i in 0..n_messages { let room = rooms[i as usize % rooms.len()]; let user = users[i as usize % users.len()]; let msg = Message { room_id: room.to_string(), sender: user.to_string(), text: format!("message number {} from {} in {}", i, user, room), ts: 1700000000 + i, }; // One collection per chat room. let col = db.collection(room); col.insert(format!("msg_{:06}", i), &msg).unwrap(); } let write_elapsed = start.elapsed().as_millis() as u64; report("chat insert (5k msgs)", n_messages, write_elapsed); // Each room should have messages. for room in rooms { let col = db.collection(room); assert!(col.count() > 0, "room {room} should have messages"); } // Prefix-scan messages in one room. let start = Instant::now(); let general_msgs: Vec = db.collection("general").scan_prefix("msg_000").unwrap(); let scan_elapsed = start.elapsed().as_millis() as u64; assert!(!general_msgs.is_empty()); report("chat prefix scan", general_msgs.len() as u64, scan_elapsed); } // ═══════════════════════════════════════════════════════════════════════════ // File-backed WAL stress tests // These exercise real disk I/O — fsync, recovery, WAL growth, snapshot // interleaving — at scale. Every test below uses MemDb::open(), not // in_memory(). // ═══════════════════════════════════════════════════════════════════════════ // ─── 21. Bulk persistent write (WAL + fsync cost) ──────────────────────── #[test] fn wal_bulk_insert_fsync_cost() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let n = 3_000u64; let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(col.count(), n as usize); report("WAL insert (fsync per op)", n, elapsed); // Verify WAL file exists and has content. let wal_path = dir.path().join("wal.jsonl"); let wal_size = std::fs::metadata(&wal_path).unwrap().len(); println!( " [wal file] {} ops → {} KiB ({:.1} bytes/op)", n, wal_size / 1024, wal_size as f64 / n as f64 ); // Recover and verify. drop(db); let db = MemDb::open(dir.path()).unwrap(); assert_eq!(db.collection("records").count(), n as usize); } // ─── 22. WAL recovery from large dataset (no snapshot) ─────────────────── #[test] fn wal_recover_pure_wal_no_snapshot() { let dir = tempfile::tempdir().unwrap(); let n = 5_000u64; // Write everything — never snapshot, so recovery must replay every entry. { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("WAL write (no snapshot)", n, elapsed); } // Recover from pure WAL replay. { let start = Instant::now(); let db = MemDb::open(dir.path()).unwrap(); let elapsed = start.elapsed().as_millis() as u64; report("WAL replay recovery", n, elapsed); let col = db.collection("records"); assert_eq!(col.count(), n as usize); // Spot-check random keys across the whole range. for i in (0..n).step_by(500) { let key = format!("{:06}", i); let rec: Record = col.get_required(&key).unwrap(); assert_eq!(rec.id, format!("rec_{:06}", i as usize)); } } } // ─── 23. Sustained write throughput over time ──────────────────────────── #[test] fn wal_sustained_write_throughput() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let rounds = 5u64; let per_round = 1_000u64; for r in 0..rounds { let base = r * per_round; let start = Instant::now(); for i in 0..per_round { let idx = base + i; col.upsert(format!("{:06}", idx), &Record::new(idx as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; let wal_size = std::fs::metadata(dir.path().join("wal.jsonl")) .unwrap() .len(); println!( " [round {}] {} ops in {}ms → {} ops/sec | WAL {} KiB", r + 1, per_round, elapsed, per_round * 1000 / elapsed.max(1), wal_size / 1024, ); } assert_eq!(col.count(), (rounds * per_round) as usize); } // ─── 24. Concurrent persistent writes ──────────────────────────────────── #[tokio::test] async fn wal_concurrent_persistent_writes() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let db = Arc::new(MemDb::open(&db_path).unwrap()); let n_writers = 8u64; let per_writer = 500u64; let total = n_writers * per_writer; let start = Instant::now(); let mut handles = vec![]; for w in 0..n_writers { let db = db.clone(); handles.push(tokio::task::spawn_blocking(move || { for i in 0..per_writer { let id = format!("w{:02}_{:06}", w, i); let rec = Record::new((w * 10000 + i) as usize); db.collection("records").upsert(id, &rec).unwrap(); } })); } for h in handles { h.await.unwrap(); } let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("records").count() as u64; assert_eq!(count, total); report("WAL concurrent writes (8×500)", total, elapsed); // Verify strict seq ordering in WAL under concurrent load. let wal_path = db_path.join("wal.jsonl"); let entries = memdb::wal::read_after(&wal_path, 0).unwrap(); assert_eq!(entries.len(), total as usize); let mut last = 0u64; for e in &entries { assert!(e.seq > last, "seq not monotonic under concurrency"); last = e.seq; } // Recover. drop(db); let db = MemDb::open(&db_path).unwrap(); assert_eq!(db.collection("records").count() as u64, total); } // ─── 25. Snapshot interleaved with sustained writes ────────────────────── #[tokio::test] async fn wal_snapshot_interleaved_heavy_writes() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let db = Arc::new(MemDb::open(&db_path).unwrap()); let n = 2_000u64; let start = Instant::now(); let db_w = db.clone(); let write_handle = tokio::task::spawn_blocking(move || { for i in 0..n { let rec = Record::new(i as usize); db_w .collection("records") .upsert(format!("{:06}", i), &rec) .unwrap(); } }); let db_s = db.clone(); let snap_handle = tokio::spawn(async move { for _ in 0..20 { tokio::time::sleep(std::time::Duration::from_millis(2)).await; if let Err(e) = db_s.snapshot() { eprintln!(" [snapshot] error: {e}"); } } }); write_handle.await.unwrap(); snap_handle.await.unwrap(); db.snapshot().unwrap(); let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("records").count() as u64; assert_eq!(count, n); report("WAL snapshot+write interleaved", n, elapsed); // Recover and verify ALL records survived every snapshot+truncate cycle. drop(db); let db = MemDb::open(&db_path).unwrap(); let col = db.collection("records"); assert_eq!(col.count() as u64, n, "data loss during snapshot interleaving"); for i in 0..n { assert!(col.exists(&format!("{:06}", i)), "missing record {i}"); } } // ─── 26. Crash recovery simulation: kill without snapshot ──────────────── #[test] fn wal_crash_recovery_no_snapshot() { let dir = tempfile::tempdir().unwrap(); let n = 2_000u64; // Simulate normal operation, then "crash" (just drop the db). { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } // No snapshot — crash! (db dropped without clean shutdown) } // Recover — all data must be intact from WAL alone. { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); assert_eq!( col.count() as u64, n, "crash recovery lost data (no snapshot)" ); // Verify a mix of early, middle, and late records. for &i in &[0, 1, n / 2, n - 2, n - 1] { let rec: Record = col.get_required(&format!("{:06}", i)).unwrap(); assert_eq!(rec.name, format!("record_{}", i as usize)); } } } // ─── 27. Mixed read/write with WAL ─────────────────────────────────────── #[tokio::test] async fn wal_mixed_read_write_workload() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); // Pre-populate. { let db = MemDb::open(&db_path).unwrap(); let col = db.collection("records"); for i in 0..2_000u64 { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } } let db = Arc::new(MemDb::open(&db_path).unwrap()); let start = Instant::now(); // Writer: append new records. let db_w = db.clone(); let write_handle = tokio::task::spawn_blocking(move || { for i in 2_000..4_000u64 { let rec = Record::new(i as usize); db_w .collection("records") .upsert(format!("{:06}", i), &rec) .unwrap(); } }); // Reader: random reads across existing range. let db_r = db.clone(); let read_handle = tokio::task::spawn_blocking(move || { for i in 0..5_000u64 { let key = format!("{:06}", i % 2500); let _: Option = db_r.collection("records").get(&key).unwrap(); } }); // Snapshotter: periodic snapshots during the workload. let db_s = db.clone(); let snap_handle = tokio::spawn(async move { for _ in 0..8 { tokio::time::sleep(std::time::Duration::from_millis(3)).await; let _ = db_s.snapshot(); } }); let (wr, rr, sr) = tokio::join!(write_handle, read_handle, snap_handle); wr.unwrap(); rr.unwrap(); sr.unwrap(); db.snapshot().unwrap(); let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("records").count() as u64; assert_eq!(count, 4_000); report("WAL mixed r/w/snapshot", 7_000, elapsed); // Final recovery check. drop(db); let db = MemDb::open(&db_path).unwrap(); assert_eq!(db.collection("records").count() as u64, 4_000); } // ─── 28. WAL behaviour: delete + insert same key ───────────────────────── #[test] fn wal_delete_insert_same_key_replay() { let dir = tempfile::tempdir().unwrap(); { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); col.insert("x", &Record::new(1)).unwrap(); // seq=1: insert col.delete("x").unwrap(); // seq=2: delete col.insert("x", &Record::new(3)).unwrap(); // seq=3: insert again // Final state: key "x" exists with record_3 data. let rec: Record = col.get_required("x").unwrap(); assert_eq!(rec.name, "record_3"); } // Recover and verify final state is preserved. { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); assert_eq!(col.count(), 1); let rec: Record = col.get_required("x").unwrap(); assert_eq!(rec.name, "record_3"); } } // ─── 29. WAL with cross-collection transactions ────────────────────────── #[test] fn wal_cross_collection_txn_replay() { let dir = tempfile::tempdir().unwrap(); let n = 500u64; { let db = MemDb::open(dir.path()).unwrap(); for i in 0..n { let rec = Record::new(i as usize); db.transaction() .upsert("alpha", format!("a_{:06}", i), &rec) .unwrap() .upsert("beta", format!("b_{:06}", i), &rec) .unwrap() .commit() .unwrap(); } assert_eq!(db.collection("alpha").count(), n as usize); assert_eq!(db.collection("beta").count(), n as usize); } // Recover and verify both collections. { let db = MemDb::open(dir.path()).unwrap(); assert_eq!(db.collection("alpha").count(), n as usize); assert_eq!(db.collection("beta").count(), n as usize); // Spot-check: each collection's records should match. let a: Record = db.collection("alpha").get_required("a_000123").unwrap(); let b: Record = db.collection("beta").get_required("b_000123").unwrap(); assert_eq!(a.name, "record_123"); assert_eq!(b.name, "record_123"); } } // ─── 30. Snapshot then immediate crash — verify no data loss ───────────── #[test] fn wal_snapshot_then_crash_recovery() { let dir = tempfile::tempdir().unwrap(); // Phase 1: write batch A, snapshot, write batch B, snapshot, crash. let snapshot_at: u64; { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); // Batch A. for i in 0..500u64 { col.insert(format!("a_{:06}", i), &Record::new(i as usize)) .unwrap(); } db.snapshot().unwrap(); // Batch B. for i in 0..500u64 { col.insert(format!("b_{:06}", i), &Record::new(500 + i as usize)) .unwrap(); } snapshot_at = col.count() as u64; // 1000 db.snapshot().unwrap(); // Batch C — no snapshot after this (simulates crash). for i in 0..500u64 { col.insert(format!("c_{:06}", i), &Record::new(1000 + i as usize)) .unwrap(); } } // crash // Phase 2: recover. Batches A+B must survive (snapshotted). // Batch C must also survive (in WAL). { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let count = col.count() as u64; println!( " [snapshot+cold crash] before crash={snapshot_at}, recovered={count}" ); assert_eq!(count, 1_500, "data loss across snapshot boundaries"); // Verify records from all three batches. assert!(col.exists("a_000000")); assert!(col.exists("a_000499")); assert!(col.exists("b_000000")); assert!(col.exists("b_000499")); assert!(col.exists("c_000000")); assert!(col.exists("c_000499")); } } // ─── 31. Large batch transaction with WAL ──────────────────────────────── #[test] fn wal_large_transaction_batch() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let n = 1_000u64; // Build one big transaction. let start = Instant::now(); let mut txn = db.transaction(); for i in 0..n { let rec = Record::new(i as usize); txn = txn .upsert("records", format!("{:06}", i), &rec) .unwrap(); } txn.commit().unwrap(); let elapsed = start.elapsed().as_millis() as u64; assert_eq!(db.collection("records").count(), n as usize); report(&format!("WAL large txn ({} ops)", n), n, elapsed); // The entire transaction should be a single WAL entry. let wal_path = dir.path().join("wal.jsonl"); let entries = memdb::wal::read_after(&wal_path, 0).unwrap(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].ops.len(), n as usize); // Recover and verify. drop(db); let db = MemDb::open(dir.path()).unwrap(); assert_eq!(db.collection("records").count(), n as usize); } // ─── 32. Update-heavy persistent workload ──────────────────────────────── #[test] fn wal_update_heavy_persistent() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("counters"); let n = 500u64; // Seed counters. for i in 0..n { col.insert( format!("c_{:04}", i), &Counter { value: 0, updates: 0, }, ) .unwrap(); } let rounds = 5u64; let start = Instant::now(); for _ in 0..rounds { for i in 0..n { col.update::(&format!("c_{:04}", i), |mut c| { c.value += 1; c.updates += 1; c }) .unwrap(); } } let elapsed = start.elapsed().as_millis() as u64; let total = rounds * n; report("WAL update-heavy", total, elapsed); // Verify in-memory state. let all: Vec = col.list_all().unwrap(); let sum_v: u64 = all.iter().map(|c| c.value).sum(); let sum_u: u64 = all.iter().map(|c| c.updates).sum(); assert_eq!(sum_v, sum_u); assert_eq!(sum_v, n * rounds); // Recover and verify again. drop(db); let db = MemDb::open(dir.path()).unwrap(); let all: Vec = db.collection("counters").list_all().unwrap(); assert_eq!(all.len(), n as usize); let sum_v: u64 = all.iter().map(|c| c.value).sum(); assert_eq!(sum_v, n * rounds); } // ─── 33. WAL seq gaps do not affect recovery ───────────────────────────── #[test] fn wal_seq_gaps_on_recovery() { let dir = tempfile::tempdir().unwrap(); { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); // Insert some records. for i in 1..=5u64 { col.insert(format!("k{}", i), &Record::new(i as usize)) .unwrap(); } // Snapshot captures seq=5. db.snapshot().unwrap(); // More writes after snapshot — seq continues 6, 7, 8... for i in 6..=10u64 { col.insert(format!("k{}", i), &Record::new(i as usize)) .unwrap(); } assert_eq!(col.count(), 10); } // Recover. Seq 6-10 must replay from WAL on top of snapshot (seq=5). { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); assert_eq!(col.count(), 10); for i in 1..=10u64 { let rec: Record = col.get_required(&format!("k{}", i)).unwrap(); assert_eq!(rec.name, format!("record_{}", i as usize)); } } } // ─── 34. Stress: many small snapshots during continuous writes ─────────── #[tokio::test] async fn wal_many_small_snapshots() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let db = Arc::new(MemDb::open(&db_path).unwrap()); let n = 1_000u64; let start = Instant::now(); let db_w = db.clone(); let write_handle = tokio::task::spawn_blocking(move || { for i in 0..n { let rec = Record::new(i as usize); db_w.collection("ticks") .upsert(format!("{:06}", i), &rec) .unwrap(); } }); let db_s = db.clone(); let snap_handle = tokio::spawn(async move { for _ in 0..50 { tokio::time::sleep(std::time::Duration::from_millis(1)).await; let _ = db_s.snapshot(); } }); let (wr, sr) = tokio::join!(write_handle, snap_handle); wr.unwrap(); sr.unwrap(); db.snapshot().unwrap(); let elapsed = start.elapsed().as_millis() as u64; let count = db.collection("ticks").count() as u64; assert_eq!(count, n); report("WAL 50 snapshots + 1k writes", n, elapsed); // Recover. drop(db); let db = MemDb::open(&db_path).unwrap(); assert_eq!(db.collection("ticks").count() as u64, n); } // ═══════════════════════════════════════════════════════════════════════════ // Durability mode comparison: Full vs Batch vs Off // ═══════════════════════════════════════════════════════════════════════════ // ─── 35. Batch mode bulk insert performance ────────────────────────────── #[test] fn durability_batch_bulk_insert_perf() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::batch(100)).unwrap(); let col = db.collection("records"); let n = 3_000u64; let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } // Flush the final partial batch. let _flushed = db.flush().unwrap(); let elapsed = start.elapsed().as_millis() as u64; let wal_size = std::fs::metadata(dir.path().join("wal.jsonl")) .unwrap() .len(); // 3000 ops / 100 batch = ~30 fsyncs (vs 3000 in Full mode) let expected_syncs = (n + 99) / 100 + 1; // +1 for final flush assert_eq!(col.count(), n as usize); report("Batch-100 insert", n, elapsed); println!( " WAL {:.0} KiB, ~{expected_syncs} fsyncs (vs {n} in Full mode)", wal_size as f64 / 1024.0, ); // Recover — flush ensures everything is on disk. drop(db); let db = MemDb::open(dir.path()).unwrap(); assert_eq!(db.collection("records").count(), n as usize); } // ─── 36. Batch mode: concurrent writes ─────────────────────────────────── #[tokio::test] async fn durability_batch_concurrent_writes() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let db = Arc::new(MemDb::open_with(&db_path, Durability::batch(50)).unwrap()); let n_writers = 4u64; let per_writer = 500u64; let total = n_writers * per_writer; let start = Instant::now(); let mut handles = vec![]; for w in 0..n_writers { let db = db.clone(); handles.push(tokio::task::spawn_blocking(move || { for i in 0..per_writer { let id = format!("w{:02}_{:06}", w, i); let rec = Record::new((w * 10000 + i) as usize); db.collection("records").upsert(id, &rec).unwrap(); } })); } for h in handles { h.await.unwrap(); } db.flush().unwrap(); let elapsed = start.elapsed().as_millis() as u64; assert_eq!(db.collection("records").count() as u64, total); report("Batch-50 concurrent (4×500)", total, elapsed); drop(db); let db = MemDb::open(&db_path).unwrap(); assert_eq!(db.collection("records").count() as u64, total); } // ─── 37. Batch mode: crash recovery of batched writes ──────────────────── #[test] fn durability_batch_crash_recovery() { let dir = tempfile::tempdir().unwrap(); { let db = MemDb::open_with(dir.path(), Durability::batch(20)).unwrap(); let col = db.collection("records"); // Write 95 records — triggers 4 batches of 20 + 15 buffered. for i in 0..95u64 { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } // Do NOT flush — last 15 are buffered, not yet on disk. } // "crash" // Recover: 80 flushed records (4 batches × 20) should survive. // The 15 buffered records are lost (expected behaviour). { let db = MemDb::open(dir.path()).unwrap(); let col = db.collection("records"); let count = col.count() as u64; println!(" [batch crash] wrote 95, no flush, recovered {count}"); assert!(count >= 80, "at least 4 batches should survive"); assert!(count < 95, "unflushed records should be lost on crash"); } } // ─── 38. Batch mode: flush worker ensures eventual durability ──────────── #[tokio::test] async fn durability_batch_flush_worker() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().to_path_buf(); let db = Arc::new(MemDb::open_with(&db_path, Durability::batch(200)).unwrap()); // Start a flush worker that fires every 50ms. let _flush_handle = db.start_flush_worker(Duration::from_millis(50)); let db_w = db.clone(); let handle = tokio::task::spawn_blocking(move || { for i in 0..500u64 { db_w.collection("records") .upsert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } }); // Meanwhile, the flush worker periodically commits buffered writes. handle.await.unwrap(); // Give the flush worker a moment to catch up. tokio::time::sleep(Duration::from_millis(100)).await; db.flush().unwrap(); assert_eq!(db.collection("records").count(), 500); drop(db); // All records should survive because the flush worker (and final flush) // pushed them to disk. let db = MemDb::open(&db_path).unwrap(); assert_eq!(db.collection("records").count(), 500); } // ─── 39. Durability::Off mode — maximum throughput, zero fsync ─────────── #[test] fn durability_off_bulk_insert_perf() { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::Off).unwrap(); let col = db.collection("records"); let n = 5_000u64; let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; assert_eq!(col.count(), n as usize); report("Off (no fsync) insert", n, elapsed); // Data is written but never synced. On crash, recovery may lose data. drop(db); let db = MemDb::open(dir.path()).unwrap(); // OS may have flushed some pages — count what survived. let recovered = db.collection("records").count(); println!(" [no fsync] wrote {n}, OS flushed {recovered} (may be 0 on crash)"); } // ─── 40. Full vs Batch vs Off side-by-side comparison ──────────────────── #[test] fn durability_full_vs_batch_vs_off() { let n = 1_000u64; // Full. { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::Full).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("Full (fsync every op)", n, elapsed); } // Batch 100. { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::batch(100)).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } db.flush().unwrap(); let elapsed = start.elapsed().as_millis() as u64; report("Batch-100 (1 fsync/100)", n, elapsed); } // Batch 10. { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::batch(10)).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } db.flush().unwrap(); let elapsed = start.elapsed().as_millis() as u64; report("Batch-10 (1 fsync/10)", n, elapsed); } // Off. { let dir = tempfile::tempdir().unwrap(); let db = MemDb::open_with(dir.path(), Durability::Off).unwrap(); let col = db.collection("records"); let start = Instant::now(); for i in 0..n { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } let elapsed = start.elapsed().as_millis() as u64; report("Off (never fsync)", n, elapsed); } } // ═══════════════════════════════════════════════════════════════════════════ // Concurrent read scalability — how well does the Mutex hold up under // pure read load at various thread counts? // ═══════════════════════════════════════════════════════════════════════════ #[tokio::test] async fn bench_concurrent_read_scaling() { use rand::Rng; let db = Arc::new(MemDb::in_memory()); let n_records = 50_000u64; let reads_per_thread = 20_000u64; // Pre-populate. { let col = db.collection("records"); for i in 0..n_records { col.insert(format!("{:06}", i), &Record::new(i as usize)) .unwrap(); } } println!( "\n dataset: {n_records} records, each thread does {reads_per_thread} random get() calls\n" ); // Baseline: single-threaded. let start = Instant::now(); let mut rng = rand::rng(); let col = db.collection("records"); for _ in 0..reads_per_thread { let key = format!("{:06}", rng.random_range(0..n_records)); let _: Option = col.get(&key).unwrap(); } let single_ms = start.elapsed().as_millis() as u64; let single_ops = reads_per_thread * 1000 / single_ms.max(1); println!(" [1 thread ] {reads_per_thread} reads in {single_ms}ms → {single_ops} ops/sec"); // Multi-threaded: 2, 4, 8, 16 threads. for &n_threads in &[2, 4, 8, 16] { let start = Instant::now(); let mut handles = vec![]; for _t in 0..n_threads { let db = db.clone(); handles.push(tokio::task::spawn_blocking(move || { let col = db.collection("records"); let mut rng = rand::rng(); let mut found = 0u64; for _ in 0..reads_per_thread { let key = format!("{:06}", rng.random_range(0..n_records)); let _: Option = col.get(&key).unwrap(); found += 1; } found })); } let mut total_found = 0u64; for h in handles { total_found += h.await.unwrap(); } let elapsed_ms = start.elapsed().as_millis() as u64; let total_reads = n_threads * reads_per_thread; let total_ops = total_reads * 1000 / elapsed_ms.max(1); let speedup = total_ops as f64 / single_ops as f64; println!( " [{n_threads:>2} threads] {total_reads} reads in {elapsed_ms}ms → {total_ops} ops/sec (×{speedup:.2})" ); assert_eq!(total_found, total_reads); } }