From 4cdf3ee5f1c2b49bea2e6e0c62b1627cb0bed637 Mon Sep 17 00:00:00 2001 From: rustmailer Date: Fri, 10 Jul 2026 01:52:04 +0800 Subject: [PATCH] refactor(blob): add delete_batch, gc_if_needed, background flush, and fix bincode compat - Replace bincode 3.0.0 (empty crate) with bincode_reloaded 3.1.10 - Add delete_batch for efficient grouped tombstone writes - Add gc_if_needed to skip GC when no segment exceeds threshold - Add flush() for lightweight fsync+meta checkpoint without compact - Add Config::flush_interval_secs to spawn a background flush thread - Remove per-put/per-delete fsync; persistence via background flush - Split gc_segments into gc_prepare/gc_finish to reduce write lock hold time --- Cargo.lock | 19 +- crates/blob/Cargo.toml | 5 +- crates/blob/benches/benchmark.rs | 89 ++-- crates/blob/examples/email_archive.rs | 125 +++++ crates/blob/src/account.rs | 284 ---------- crates/blob/src/bucket.rs | 723 ++++++++++++++++++-------- crates/blob/src/cache.rs | 219 -------- crates/blob/src/engine.rs | 627 +++++++++++++--------- crates/blob/src/error.rs | 6 - crates/blob/src/file_pool.rs | 8 +- crates/blob/src/fs.rs | 25 + crates/blob/src/gc.rs | 212 ++------ crates/blob/src/lib.rs | 5 +- crates/blob/src/meta.rs | 205 ++------ crates/blob/src/recovery.rs | 118 ++--- crates/blob/src/segment.rs | 79 +-- crates/blob/src/types.rs | 28 +- crates/blob/tests/acid_test.rs | 174 +++---- crates/blob/tests/integration_test.rs | 277 +++------- 19 files changed, 1431 insertions(+), 1797 deletions(-) create mode 100644 crates/blob/examples/email_archive.rs delete mode 100644 crates/blob/src/account.rs delete mode 100644 crates/blob/src/cache.rs diff --git a/Cargo.lock b/Cargo.lock index 67a1268..58da3ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -319,10 +319,11 @@ dependencies = [ name = "bichon-blob" version = "0.1.0" dependencies = [ - "bincode", + "bincode_reloaded", "crc32fast", "criterion", "lz4_flex", + "memmap2", "rand 0.10.2", "serde", "serde_json", @@ -484,6 +485,16 @@ dependencies = [ "serde", ] +[[package]] +name = "bincode_reloaded" +version = "3.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c6f3fe39960aac27f7de2e5f41be9fa576be3fbffddf86a06b967b579f603b8" +dependencies = [ + "serde", + "unty", +] + [[package]] name = "bit-vec" version = "0.9.1" @@ -5279,6 +5290,12 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "unty" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dbe8d477efbc6c70a1dea1f9f1e0482168983a9147dee6b3a619f666f3aeac6" + [[package]] name = "url" version = "2.5.8" diff --git a/crates/blob/Cargo.toml b/crates/blob/Cargo.toml index 4a94e9b..75f4ae2 100644 --- a/crates/blob/Cargo.toml +++ b/crates/blob/Cargo.toml @@ -10,13 +10,14 @@ zstd = "0.13" lz4_flex = "0.13.1" serde = { version = "1", features = ["derive"] } serde_json = "1" -bincode = "1" +bincode = { package = "bincode_reloaded", version = "3.1.10", default-features = false, features = ["serde", "alloc"] } tracing = "0.1" thiserror = "2" +memmap2 = "0.9" [dev-dependencies] tempfile = "3" -rand = "0.10.1" +rand = "0.10.2" criterion = { version = "0.6", features = ["html_reports"] } [[bench]] diff --git a/crates/blob/benches/benchmark.rs b/crates/blob/benches/benchmark.rs index 278531a..0ccb281 100644 --- a/crates/blob/benches/benchmark.rs +++ b/crates/blob/benches/benchmark.rs @@ -12,7 +12,6 @@ fn make_key(seed: u64) -> [u8; 32] { fn make_value(size: usize) -> Vec { let mut v = Vec::with_capacity(size); - // Fill with somewhat realistic text-like data so compression works let pattern = b"The quick brown fox jumps over the lazy dog. "; while v.len() < size { let rem = size - v.len(); @@ -29,7 +28,6 @@ pub fn bench_write_small(c: &mut Criterion) { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); let value = make_value(1024); // 1 KB let mut counter = 0u64; @@ -41,9 +39,7 @@ pub fn bench_write_small(c: &mut Criterion) { (make_key(counter), value.clone()) }, |(key, val)| { - engine - .write("bench", key, &val, Codec::Zstd) - .unwrap() + engine.put(key, &val, Codec::Zstd).unwrap() }, BatchSize::SmallInput, ) @@ -58,7 +54,6 @@ pub fn bench_write_medium(c: &mut Criterion) { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); let value = make_value(64 * 1024); // 64 KB let mut counter = 0u64; @@ -70,9 +65,7 @@ pub fn bench_write_medium(c: &mut Criterion) { (make_key(counter), value.clone()) }, |(key, val)| { - engine - .write("bench", key, &val, Codec::Zstd) - .unwrap() + engine.put(key, &val, Codec::Zstd).unwrap() }, BatchSize::SmallInput, ) @@ -87,7 +80,6 @@ pub fn bench_write_large(c: &mut Criterion) { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); let value = make_value(1024 * 1024); // 1 MB let mut counter = 0u64; @@ -99,9 +91,7 @@ pub fn bench_write_large(c: &mut Criterion) { (make_key(counter), value.clone()) }, |(key, val)| { - engine - .write("bench", key, &val, Codec::Zstd) - .unwrap() + engine.put(key, &val, Codec::Zstd).unwrap() }, BatchSize::SmallInput, ) @@ -109,59 +99,55 @@ pub fn bench_write_large(c: &mut Criterion) { group.finish(); } -pub fn bench_read_cache_hit(c: &mut Criterion) { +pub fn bench_read_hot(c: &mut Criterion) { let mut group = c.benchmark_group("read"); group.throughput(Throughput::Elements(1)); group.measurement_time(Duration::from_secs(10)); let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); - // Pre-populate: 10 keys, all in same bucket → cache hit after first read + // Pre-populate: 10 keys let value = make_value(4096); for i in 0..10u64 { engine - .write("bench", make_key(i), &value, Codec::Zstd) + .put(make_key(i), &value, Codec::Zstd) .unwrap(); } let mut counter = 0u64; - group.bench_function("cache_hit", |b| { + group.bench_function("hot", |b| { b.iter(|| { let key = make_key(counter % 10); counter += 1; - std::hint::black_box(engine.read("bench", &key).unwrap()); + std::hint::black_box(engine.get(&key).unwrap()); }) }); group.finish(); } -pub fn bench_read_cache_miss(c: &mut Criterion) { +pub fn bench_read_cold(c: &mut Criterion) { let mut group = c.benchmark_group("read"); group.throughput(Throughput::Elements(1)); group.measurement_time(Duration::from_secs(10)); let dir = TempDir::new().unwrap(); - let mut config = Config::default(); - config.lru_bucket_count = 8; // Small cache to force misses - let engine = Engine::open(dir.path(), config).unwrap(); - engine.create_account("bench").unwrap(); + let engine = Engine::open(dir.path(), Config::default()).unwrap(); let value = make_value(4096); - // Write 1000 keys spread across all 16 buckets — small LRU will thrash - for i in 0..1000u64 { + // Write 5000 keys across all 256 buckets — mmap page faults will occur + for i in 0..5000u64 { engine - .write("bench", make_key(i), &value, Codec::Zstd) + .put(make_key(i), &value, Codec::Zstd) .unwrap(); } let mut counter = 0u64; - group.bench_function("cache_miss", |b| { + group.bench_function("cold", |b| { b.iter(|| { - let key = make_key(counter % 1000); + let key = make_key(counter % 5000); counter += 1; - std::hint::black_box(engine.read("bench", &key).unwrap()); + std::hint::black_box(engine.get(&key).unwrap()); }) }); group.finish(); @@ -174,21 +160,20 @@ pub fn bench_read_large_value(c: &mut Criterion) { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); let value = make_value(1024 * 1024); // 1 MB for i in 0..5u64 { engine - .write("bench", make_key(i), &value, Codec::Zstd) + .put(make_key(i), &value, Codec::Zstd) .unwrap(); } let mut counter = 0u64; - group.bench_function("1MB_cache_hit", |b| { + group.bench_function("1MB", |b| { b.iter(|| { let key = make_key(counter % 5); counter += 1; - std::hint::black_box(engine.read("bench", &key).unwrap()); + std::hint::black_box(engine.get(&key).unwrap()); }) }); group.finish(); @@ -202,7 +187,6 @@ pub fn bench_delete(c: &mut Criterion) { group.bench_function("delete", |b| { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); let value = make_value(4096); let mut counter = 0u64; @@ -211,13 +195,11 @@ pub fn bench_delete(c: &mut Criterion) { || { counter += 1; let key = make_key(counter); - engine - .write("bench", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); key }, |key| { - engine.delete("bench", &key).unwrap(); + engine.delete(&key).unwrap(); }, BatchSize::SmallInput, ) @@ -232,13 +214,12 @@ pub fn bench_mixed_workload(c: &mut Criterion) { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); // Pre-populate with 500 entries let value = make_value(8192); for i in 0..500u64 { engine - .write("bench", make_key(i), &value, Codec::Zstd) + .put(make_key(i), &value, Codec::Zstd) .unwrap(); } @@ -249,20 +230,19 @@ pub fn bench_mixed_workload(c: &mut Criterion) { let op = counter % 100; match op { 0..=79 => { - // 80% writes let key = make_key(counter); let val = make_value(4096); - engine.write("bench", key, &val, Codec::Zstd).unwrap(); + engine.put(key, &val, Codec::Zstd).unwrap(); } 80..=94 => { - // 15% reads - std::hint::black_box(engine.read("bench", &make_key(counter % 500)).unwrap()); + std::hint::black_box( + engine.get(&make_key(counter % 500)).unwrap(), + ); } _ => { - // 5% deletes if counter % 2 == 0 { let key = make_key(counter % 500); - let _ = engine.delete("bench", &key); + let _ = engine.delete(&key); } } } @@ -279,23 +259,20 @@ pub fn bench_gc(c: &mut Criterion) { group.bench_function("gc_30pct_deleted", |b| { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("bench").unwrap(); - // Fill a segment with ~1000 entries, then delete 30% - let value = make_value(200_000); // 200KB each → ~1000 entries to fill 256MB + let value = make_value(200_000); let n = 1200u64; for i in 0..n { engine - .write("bench", make_key(i), &value, Codec::None) + .put(make_key(i), &value, Codec::None) .unwrap(); } - // Delete ~30% for i in (0..n).step_by(3) { - engine.delete("bench", &make_key(i)).unwrap(); + engine.delete(&make_key(i)).unwrap(); } b.iter(|| { - engine.gc("bench").unwrap(); + engine.gc().unwrap(); }) }); group.finish(); @@ -306,8 +283,8 @@ criterion_group!( bench_write_small, bench_write_medium, bench_write_large, - bench_read_cache_hit, - bench_read_cache_miss, + bench_read_hot, + bench_read_cold, bench_read_large_value, bench_delete, bench_mixed_workload, diff --git a/crates/blob/examples/email_archive.rs b/crates/blob/examples/email_archive.rs new file mode 100644 index 0000000..a45250a --- /dev/null +++ b/crates/blob/examples/email_archive.rs @@ -0,0 +1,125 @@ +/// bichon-blob usage example: email archival with content-addressable storage. +/// +/// This example simulates a mail archival system where multiple accounts +/// may receive the same email (e.g. CC'd or forwarded). The blob store +/// deduplicates by content hash, and the upper application layer tracks +/// which accounts reference each hash. +/// +/// Run: cargo run --example email_archive + +use std::collections::HashMap; + +use bichon_blob::{Codec, Config, Engine}; + +fn main() -> Result<(), Box> { + // ── Setup ────────────────────────────────────────────────────────── + + let store_path = std::path::Path::new("target/example_blob_store"); + let _ = std::fs::remove_dir_all(store_path); // clean up from previous run + let engine = Engine::open(store_path, Config::default())?; + + println!("Store opened at: {:?}\n", store_path); + + // ── Simulated upper-layer reference tracker ──────────────────────── + let mut refs: HashMap> = HashMap::new(); + + // ── 1. Store emails ──────────────────────────────────────────────── + + // Simulate three emails arriving. Email #2 is a newsletter that + // both alice and bob received — identical content, same hash. + let emails = vec![ + ("alice", "Welcome to Bichon Mail!"), + ("alice", "Weekly Newsletter: Rust Edition"), + ("bob", "Weekly Newsletter: Rust Edition"), // same content as above + ]; + + for (account, body) in &emails { + let hash = mock_content_hash(body.as_bytes()); + let account_refs = refs.entry(account.to_string()).or_default(); + + // Check if this content already exists in the blob store + if engine.exists(&hash)? { + println!( + "[dedup] {}: hash {:02x?}... already stored, skipping", + account, + &hash[..4] + ); + } else { + engine.put(hash, body.as_bytes(), Codec::Zstd)?; + println!( + "[store] {}: hash {:02x?}..., {} bytes", + account, + &hash[..4], + body.len() + ); + } + + account_refs.push(hash); + } + + println!(); + + // ── 2. Read back emails ──────────────────────────────────────────── + + let hash = mock_content_hash(b"Weekly Newsletter: Rust Edition"); + let stored = engine.get(&hash)?; + println!( + "Read newsletter: {:?}", + stored.map(|v| String::from_utf8_lossy(&v).to_string()) + ); + + // ── 3. Batch store attachments ───────────────────────────────────── + + let attachments: Vec<([u8; 32], Vec, Codec)> = (0..10) + .map(|i| { + let body = format!("Attachment #{}: {}", i, "X".repeat(5000)); + let hash = mock_content_hash(body.as_bytes()); + (hash, body.into_bytes(), Codec::Zstd) + }) + .collect(); + + engine.put_batch(&attachments)?; + println!("\nBatch-stored {} attachments", attachments.len()); + + // ── 4. Stats ─────────────────────────────────────────────────────── + + let stats = engine.stats()?; + println!( + "Stats: {} keys, {} bytes, {} segments", + stats.total_keys, stats.total_bytes, stats.segment_count + ); + + // ── 5. Delete an email (simulating: last reference removed) ───────── + + // In production, before calling engine.delete(), you'd check: + // SELECT COUNT(*) FROM email_refs WHERE content_hash = ? AND account_id != ? + // If count == 0, it's safe to delete from blob. + + let welcome_hash = mock_content_hash(b"Welcome to Bichon Mail!"); + engine.delete(&welcome_hash)?; + println!("\nDeleted welcome email, still exists? {}", engine.exists(&welcome_hash)?); + + // ── 6. Shutdown ──────────────────────────────────────────────────── + + engine.shutdown()?; + println!("\nClean shutdown complete."); + + Ok(()) +} + +/// In production, use BLAKE3 or SHA-256. +/// Here we use a trivial hash for demonstration. +fn mock_content_hash(data: &[u8]) -> [u8; 32] { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + + let mut hasher = DefaultHasher::new(); + data.hash(&mut hasher); + let h = hasher.finish(); + + let mut key = [0u8; 32]; + key[0..8].copy_from_slice(&h.to_le_bytes()); + // Mix in the length so different-sized content gets different hashes + key[8..16].copy_from_slice(&(data.len() as u64).to_le_bytes()); + key +} diff --git a/crates/blob/src/account.rs b/crates/blob/src/account.rs deleted file mode 100644 index b95794d..0000000 --- a/crates/blob/src/account.rs +++ /dev/null @@ -1,284 +0,0 @@ -use std::collections::HashMap; -use std::fs; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, RwLock}; - -use crate::bucket::{self, BucketFile, IndexRecord}; -use crate::error::{Error, Result}; -use crate::file_pool::FilePool; -use crate::meta::{AccountMeta, SegmentStats}; -use crate::segment::{self, SegmentReader, SegmentWriter}; -use crate::types::Codec; - -// ── AccountHandle ────────────────────────────────────────────────────────── - -pub struct AccountHandle { - id: String, - dir: PathBuf, - inner: RwLock, - pub(crate) write_mutex: Mutex<()>, - file_pool: FilePool, -} - -impl AccountHandle { - pub fn id(&self) -> &str { - &self.id - } - - pub fn dir(&self) -> &Path { - &self.dir - } - - /// Open an existing account. - pub fn open(store_root: &Path, account_id: &str) -> Result> { - let dir = store_root.join("accounts").join(account_id); - if !dir.exists() { - return Err(Error::AccountNotFound(account_id.to_string())); - } - let inner = AccountInner::open(&dir)?; - Ok(Arc::new(Self { - id: account_id.to_string(), - dir, - inner: RwLock::new(inner), - write_mutex: Mutex::new(()), - file_pool: FilePool::new(8), - })) - } - - /// Create a new account. - pub fn create(store_root: &Path, account_id: &str) -> Result> { - let dir = store_root.join("accounts").join(account_id); - if dir.exists() { - return Err(Error::AccountAlreadyExists(account_id.to_string())); - } - let inner = AccountInner::create(&dir, account_id)?; - Ok(Arc::new(Self { - id: account_id.to_string(), - dir, - inner: RwLock::new(inner), - write_mutex: Mutex::new(()), - file_pool: FilePool::new(8), - })) - } - - /// Lock the inner state for reading. - pub fn read(&self) -> std::sync::RwLockReadGuard<'_, AccountInner> { - self.inner.read().unwrap() - } - - /// Lock the inner state for writing. - pub fn write(&self) -> std::sync::RwLockWriteGuard<'_, AccountInner> { - self.inner.write().unwrap() - } - - /// Get a cached file handle for a segment. - pub fn get_segment_file(&self, seg_id: u32, path: &Path) -> Result>> { - self.file_pool.get(seg_id, path) - } - - /// Invalidate cached file handles for a segment (after GC). - pub fn invalidate_file_cache(&self, seg_id: u32) { - self.file_pool.invalidate(seg_id); - } -} - -// ── AccountInner ─────────────────────────────────────────────────────────── - -pub struct AccountInner { - dir: PathBuf, - meta: AccountMeta, - active_writer: SegmentWriter, - readers: HashMap, -} - -impl AccountInner { - fn open(dir: &Path) -> Result { - let meta = AccountMeta::load(dir)?; - - let seg_path = dir - .join("segments") - .join(segment::segment_filename(meta.active_segment_id)); - let active_writer = if seg_path.exists() { - SegmentWriter::open_append(seg_path, meta.active_segment_id)? - } else { - fs::create_dir_all(dir.join("segments"))?; - SegmentWriter::create(seg_path, meta.active_segment_id)? - }; - - let mut readers = HashMap::new(); - for (&seg_id, stats) in &meta.segments { - if stats.sealed { - let seg_path = dir - .join("segments") - .join(segment::segment_filename(seg_id)); - if seg_path.exists() { - readers.insert(seg_id, SegmentReader::open(seg_path, seg_id)?); - } - } - } - - Ok(Self { - dir: dir.to_path_buf(), - meta, - active_writer, - readers, - }) - } - - fn create(dir: &Path, account_id: &str) -> Result { - fs::create_dir_all(dir.join("segments"))?; - BucketFile::ensure_dir(dir)?; - - let meta = AccountMeta::new(account_id.to_string(), 1); - - let seg_path = dir - .join("segments") - .join(segment::segment_filename(1)); - let active_writer = SegmentWriter::create(seg_path, 1)?; - - meta.save(dir)?; - - Ok(Self { - dir: dir.to_path_buf(), - meta, - active_writer, - readers: HashMap::new(), - }) - } - - pub fn meta(&self) -> &AccountMeta { - &self.meta - } - - /// Mark the segment as indexed up to the given offset and persist meta. - pub fn mark_indexed(&mut self, segment_id: u32, indexed_up_to_offset: u64) -> Result<()> { - if let Some(stats) = self.meta.segments.get_mut(&segment_id) { - if indexed_up_to_offset > stats.indexed_up_to_offset { - stats.indexed_up_to_offset = indexed_up_to_offset; - } - } - self.meta.save(&self.dir) - } - - /// Append an entry without fsync. - pub fn append_entry( - &mut self, - key: [u8; 32], - data: &[u8], - flags: u8, - codec: Codec, - ) -> Result<(u32, u64, u32)> { - if self.active_writer.is_full() { - self.seal_active()?; - } - - use crate::segment::Entry; - let entry = if flags == 1 { - Entry::tombstone(key) - } else { - Entry::new(key, data, flags, codec) - }; - - let data_size = entry.data.len() as u32; - let segment_id = self.active_writer.id(); - let offset = self.active_writer.append(&entry)?; - - let stats = self - .meta - .segments - .entry(segment_id) - .or_insert_with(|| SegmentStats::new(segment_id)); - stats.total_bytes += data_size as u64; - if flags == 1 { - stats.deleted_bytes += entry.raw_size as u64; - } - stats.recompute_ratio(); - - Ok((segment_id, offset, data_size)) - } - - /// Fsync the active segment and persist meta. - pub fn flush_active(&mut self) -> Result<()> { - self.active_writer.fsync()?; - self.meta.save(&self.dir) - } - - /// Write an entry with fsync. - pub fn write_entry( - &mut self, - key: [u8; 32], - data: &[u8], - flags: u8, - codec: Codec, - ) -> Result<(u32, u64, u32)> { - let result = self.append_entry(key, data, flags, codec)?; - self.flush_active()?; - Ok(result) - } - - fn seal_active(&mut self) -> Result<()> { - let old_id = self.active_writer.id(); - let old_stats = self - .meta - .segments - .entry(old_id) - .or_insert_with(|| SegmentStats::new(old_id)); - old_stats.sealed = true; - - let seg_path = self - .dir - .join("segments") - .join(segment::segment_filename(old_id)); - self.readers - .insert(old_id, SegmentReader::open(seg_path, old_id)?); - - let new_id = old_id + 1; - self.meta.active_segment_id = new_id; - let new_path = self - .dir - .join("segments") - .join(segment::segment_filename(new_id)); - self.active_writer = SegmentWriter::create(new_path, new_id)?; - self.meta.save(&self.dir)?; - - Ok(()) - } - - /// Get the on-disk path for a segment. - pub fn segment_path(&self, segment_id: u32) -> Result { - let filename = segment::segment_filename(segment_id); - let path = self.dir.join("segments").join(&filename); - if path.exists() { - Ok(path) - } else { - Err(Error::SegmentNotFound(segment_id)) - } - } - - /// Append index record to the appropriate bucket file. - pub fn append_index(&self, record: &IndexRecord) -> Result<()> { - let bucket_id = bucket::bucket_id(&record.key); - let bf = BucketFile::open(&self.dir, bucket_id); - bf.append(record) - } - - /// Return list of sealed segment IDs. - pub fn sealed_segments(&self) -> Vec { - self.meta - .segments - .iter() - .filter(|(_, s)| s.sealed) - .map(|(id, _)| *id) - .collect() - } - - /// All segment IDs (including active). - pub fn all_segment_ids(&self) -> Vec { - let mut ids: Vec = self.meta.segments.keys().copied().collect(); - if !ids.contains(&self.meta.active_segment_id) { - ids.push(self.meta.active_segment_id); - } - ids.sort_unstable(); - ids - } -} diff --git a/crates/blob/src/bucket.rs b/crates/blob/src/bucket.rs index 208346f..2d3357d 100644 --- a/crates/blob/src/bucket.rs +++ b/crates/blob/src/bucket.rs @@ -1,6 +1,10 @@ -use std::fs::OpenOptions; +use std::collections::HashMap; +use std::fs::{self, File, OpenOptions}; use std::io::Write; use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use memmap2::Mmap; use crate::error::Result; use crate::types::{BUCKET_COUNT, INDEX_RECORD_SIZE}; @@ -37,7 +41,6 @@ impl IndexRecord { buf[36..44].copy_from_slice(&self.offset.to_le_bytes()); buf[44..48].copy_from_slice(&self.data_size.to_le_bytes()); buf[48] = self.flags; - // bytes 49..52 are padding (keep zero) buf } @@ -58,189 +61,420 @@ impl IndexRecord { } } -/// Represents a loaded and deduplicated bucket in memory. -pub struct BucketIndex { - pub bucket_id: u16, - /// Records sorted by key, deduplicated (one record per key, latest wins). - pub records: Vec, -} - -impl BucketIndex { - /// Build from raw records: sort by key, dedup keeping the one with max offset. - pub fn from_records(mut records: Vec, bucket_id: u16) -> Self { - records.sort_by_key(|a| a.key); - // Dedup: keep last (max offset) for each key - let mut deduped = Vec::with_capacity(records.len()); - let mut i = 0; - while i < records.len() { - let mut best = i; - let mut j = i + 1; - while j < records.len() && records[j].key == records[i].key { - if records[j].offset > records[best].offset { - best = j; - } - j += 1; - } - deduped.push(records[best].clone()); - i = j; - } - Self { - bucket_id, - records: deduped, - } - } - - /// Binary search for a key. Returns the record if found. - pub fn find(&self, key: &[u8; 32]) -> Option<&IndexRecord> { - match self.records.binary_search_by(|r| r.key.cmp(key)) { - Ok(idx) => Some(&self.records[idx]), - Err(_) => None, - } - } - - /// Append a new record and maintain sorted order. - pub fn insert(&mut self, record: IndexRecord) { - match self.records.binary_search_by(|r| r.key.cmp(&record.key)) { - Ok(idx) => { - // Replace if newer (larger offset) - if record.offset > self.records[idx].offset { - self.records[idx] = record; - } - } - Err(idx) => { - self.records.insert(idx, record); - } - } - } - - pub fn len(&self) -> usize { - self.records.len() - } - - pub fn is_empty(&self) -> bool { - self.records.is_empty() - } -} - -/// Manages a bucket index file on disk. -pub struct BucketFile { - path: PathBuf, - bucket_id: u16, -} - -impl BucketFile { - pub fn path_for(account_dir: &Path, bucket_id: u16) -> PathBuf { - account_dir.join("buckets").join(format!("{:02x}.idx", bucket_id)) - } - - pub fn open(account_dir: &Path, bucket_id: u16) -> Self { - Self { - path: Self::path_for(account_dir, bucket_id), - bucket_id, - } - } - - pub fn path(&self) -> &Path { - &self.path - } - - pub fn bucket_id(&self) -> u16 { - self.bucket_id - } - - /// Ensure the buckets directory exists. - pub fn ensure_dir(account_dir: &Path) -> Result<()> { - let dir = account_dir.join("buckets"); - std::fs::create_dir_all(&dir)?; - Ok(()) - } - - /// Append a single record to the bucket file. - pub fn append(&self, record: &IndexRecord) -> Result<()> { - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(&self.path)?; - file.write_all(&record.encode())?; - Ok(()) - } - - /// Append multiple records at once. - pub fn append_batch(&self, records: &[IndexRecord]) -> Result<()> { - if records.is_empty() { - return Ok(()); - } - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(&self.path)?; - for r in records { - file.write_all(&r.encode())?; - } - Ok(()) - } - - /// Load all records from the bucket file. - /// If the file size is not a multiple of INDEX_RECORD_SIZE (partial write), - /// the trailing bytes are silently ignored. - pub fn load_all(&self) -> Result> { - if !self.path.exists() { - return Ok(Vec::new()); - } - let data = std::fs::read(&self.path)?; - let remainder = data.len() % INDEX_RECORD_SIZE; - let count = data.len() / INDEX_RECORD_SIZE; - let mut records = Vec::with_capacity(count); - for i in 0..count { - let start = i * INDEX_RECORD_SIZE; - let end = start + INDEX_RECORD_SIZE; - let buf: &[u8; INDEX_RECORD_SIZE] = data[start..end] - .try_into() - .map_err(|_| crate::error::Error::BucketIndexCorrupt { - path: self.path.clone(), - reason: format!("unexpected file size {}, not a multiple of {}", data.len(), INDEX_RECORD_SIZE), - })?; - records.push(IndexRecord::decode(buf)); - } - if remainder > 0 { - tracing::warn!( - "Bucket file {:?} has {} trailing bytes (expected multiple of {}), ignoring", - self.path, remainder, INDEX_RECORD_SIZE - ); - } - Ok(records) - } - - /// Load all records, sort, and deduplicate into a BucketIndex. - pub fn load_index(&self) -> Result { - let records = self.load_all()?; - Ok(BucketIndex::from_records(records, self.bucket_id)) - } - - /// Rewrite the bucket file with a sorted, deduplicated set of records. - /// Uses atomic temp+rename to be safe on NFS. - pub fn rewrite(&self, records: &[IndexRecord]) -> Result<()> { - let mut buf = Vec::with_capacity(records.len() * INDEX_RECORD_SIZE); - for r in records { - buf.extend_from_slice(&r.encode()); - } - crate::fs::create_atomic(&self.path, &buf) - } - - /// Delete the bucket file. - pub fn delete(&self) -> Result<()> { - if self.path.exists() { - std::fs::remove_file(&self.path)?; - } - Ok(()) - } -} - /// Compute bucket_id from a key's first 2 bytes. pub fn bucket_id(key: &[u8; 32]) -> u16 { u16::from_be_bytes([key[0], key[1]]) % BUCKET_COUNT } +// ── BucketStore ──────────────────────────────────────────────────────────── + +/// Per-bucket mutable state behind a Mutex. +struct BucketState { + /// mmap of the clean, sorted, deduplicated portion of the bucket file. + mmap: Mmap, + /// Number of sorted records in the mmap. + compacted_records: usize, + /// Recent writes not yet merged into the mmap. Key → latest record. + pending: HashMap<[u8; 32], IndexRecord>, + /// Append-only file for durability of pending writes. + file: File, + /// Path to the bucket file. + path: PathBuf, +} + +/// Zero-heap bucket index store backed by mmap. +/// +/// Each bucket's clean portion is mmap'd — binary search reads directly +/// from the OS page cache without allocating heap memory proportional to +/// the number of stored keys. Only pending writes (since the last compact) +/// live in a small in-memory HashMap. +pub struct BucketStore { + states: Vec>, + compact_threshold: usize, +} + +impl BucketStore { + /// Open all bucket files. On first open or after a crash, each file is + /// loaded, sorted, deduplicated, and rewritten into a clean mmap'd form. + pub fn open(dir: &Path, compact_threshold: usize) -> Result { + fs::create_dir_all(dir)?; + + let mut states = Vec::with_capacity(BUCKET_COUNT as usize); + for bid in 0..BUCKET_COUNT { + let path = bucket_path(dir, bid); + let (mmap, compacted_records) = if path.exists() { + // Load, sort, dedup, rewrite clean, then mmap. + let records = load_records_from_file(&path)?; + let deduped = sort_and_dedup(records); + let count = deduped.len(); + rewrite_file(&path, &deduped)?; + let file = fs::File::open(&path)?; + let mmap = unsafe { Mmap::map(&file)? }; + (mmap, count) + } else { + // Create empty bucket file. + let file = fs::File::create(&path)?; + file.set_len(0)?; + let mmap = unsafe { Mmap::map(&file)? }; + (mmap, 0) + }; + + let file = OpenOptions::new() + .create(true) + .append(true) + .open(&path)?; + + states.push(Mutex::new(BucketState { + mmap, + compacted_records, + pending: HashMap::new(), + file, + path, + })); + } + + Ok(Self { + states, + compact_threshold, + }) + } + + /// Look up a key. Returns the latest IndexRecord, or None if absent/tombstone. + pub fn get(&self, key: &[u8; 32]) -> Result> { + let bid = bucket_id(key) as usize; + let state = self.states[bid].lock().unwrap(); + + // 1. Check pending (most recent wins) + if let Some(rec) = state.pending.get(key) { + return Ok(if rec.is_tombstone() { None } else { Some(rec.clone()) }); + } + + // 2. Binary search the mmap'd sorted portion + let bytes: &[u8] = &state.mmap; + if bytes.is_empty() { + return Ok(None); + } + + let count = bytes.len() / INDEX_RECORD_SIZE; + let result = binary_search_records(bytes, key, count); + match result { + Some(idx) => { + let rec = read_record_at(bytes, idx); + Ok(if rec.is_tombstone() { None } else { Some(rec) }) + } + None => Ok(None), + } + } + + /// Check whether a key exists (non-tombstone) in the store. + pub fn exists(&self, key: &[u8; 32]) -> Result { + self.get(key).map(|r| r.is_some()) + } + + /// Insert or update a record for a key. Appends to the file for durability, + /// then inserts into the pending HashMap. + pub fn insert(&self, record: IndexRecord) -> Result<()> { + let bid = bucket_id(&record.key) as usize; + let mut state = self.states[bid].lock().unwrap(); + + // Durability: append to file + state.file.write_all(&record.encode())?; + + // Update pending + state.pending.insert(record.key, record); + + // Auto-compact if pending grows too large + if state.pending.len() >= self.compact_threshold { + drop(state); + self.compact_bucket(bid as u16)?; + } + + Ok(()) + } + + /// Batch insert multiple records. Appends all, then updates pending. + pub fn insert_batch(&self, records: &[IndexRecord]) -> Result<()> { + // Group by bucket + let mut grouped: HashMap> = HashMap::new(); + for r in records { + let bid = bucket_id(&r.key); + grouped.entry(bid).or_default().push(r); + } + + for (bid, recs) in &grouped { + let bid_usize = *bid as usize; + let mut state = self.states[bid_usize].lock().unwrap(); + + for r in recs { + state.file.write_all(&r.encode())?; + state.pending.insert(r.key, (*r).clone()); + } + } + + // Compact overfull buckets + for &bid in grouped.keys() { + let state = self.states[bid as usize].lock().unwrap(); + let needs_compact = state.pending.len() >= self.compact_threshold; + drop(state); + if needs_compact { + self.compact_bucket(bid)?; + } + } + + Ok(()) + } + + /// Run stats across all buckets: total keys (non-tombstone) and total data bytes. + pub fn total_keys(&self) -> usize { + let mut count = 0usize; + for state in self.states.iter() { + let s = state.lock().unwrap(); + // Count from pending + for r in s.pending.values() { + if !r.is_tombstone() { + count += 1; + } + } + // Count from mmap + let bytes: &[u8] = &s.mmap; + let n = bytes.len() / INDEX_RECORD_SIZE; + for i in 0..n { + let rec = read_record_at(bytes, i); + // Skip keys that are overridden in pending + if s.pending.contains_key(&rec.key) { + continue; + } + if !rec.is_tombstone() { + count += 1; + } + } + } + count + } + + /// Compact a single bucket: merge mmap + pending, sort+dedup, rewrite file, remap. + fn compact_bucket(&self, bid: u16) -> Result<()> { + let idx = bid as usize; + let mut state = self.states[idx].lock().unwrap(); + + if state.pending.is_empty() { + return Ok(()); + } + + // Collect mmap records + pending records + let mut all: Vec = Vec::new(); + + let bytes: &[u8] = &state.mmap; + let n = bytes.len() / INDEX_RECORD_SIZE; + all.reserve(n + state.pending.len()); + for i in 0..n { + all.push(read_record_at(bytes, i)); + } + for r in state.pending.values() { + all.push(r.clone()); + } + + let deduped = sort_and_dedup(all); + let new_count = deduped.len(); + + // Write new file atomically + rewrite_file(&state.path, &deduped)?; + + // Remap + let file = fs::File::open(&state.path)?; + let new_mmap = unsafe { Mmap::map(&file)? }; + + // Re-open append file descriptor (old one was truncated) + let new_file = OpenOptions::new() + .create(true) + .append(true) + .open(&state.path)?; + + state.mmap = new_mmap; + state.compacted_records = new_count; + state.pending.clear(); + state.file = new_file; + + Ok(()) + } + + /// Compact all buckets. + pub fn compact_all(&self) -> Result<()> { + for bid in 0..BUCKET_COUNT { + self.compact_bucket(bid)?; + } + Ok(()) + } + + /// Reload all mmaps from disk and reset pending state. + /// Used after GC rewrites bucket files externally (via rebuild_from_segments). + pub fn reload_all(&self) -> Result<()> { + for bid in 0..BUCKET_COUNT { + let mut state = self.states[bid as usize].lock().unwrap(); + let path = state.path.clone(); + + // Load, sort, dedup, rewrite clean + let records = load_records_from_file(&path)?; + let deduped = sort_and_dedup(records); + let count = deduped.len(); + rewrite_file(&path, &deduped)?; + + // Remap + let file = fs::File::open(&path)?; + let new_mmap = unsafe { Mmap::map(&file)? }; + + // Reopen append file + let new_file = OpenOptions::new() + .create(true) + .append(true) + .open(&path)?; + + state.mmap = new_mmap; + state.compacted_records = count; + state.pending.clear(); + state.file = new_file; + } + Ok(()) + } + + /// Rebuild all bucket files from scratch by scanning segment entries. + /// Used by GC and recovery. + pub fn rebuild_from_segments( + dir: &Path, + segments: &[(u32, &Path)], + ) -> Result<()> { + use crate::segment::SegmentReader; + + let mut bucket_records: HashMap> = HashMap::new(); + for i in 0..BUCKET_COUNT { + bucket_records.insert(i, Vec::new()); + } + + for &(seg_id, seg_path) in segments { + if !seg_path.exists() { + continue; + } + let reader = SegmentReader::open(seg_path.to_path_buf(), seg_id)?; + reader.scan_entries(0, |entry, offset| { + let bid = bucket_id(&entry.key); + let rec = IndexRecord::new( + entry.key, + seg_id, + offset, + entry.data.len() as u32, + entry.flags, + ); + bucket_records.entry(bid).or_default().push(rec); + Ok(()) + })?; + } + + for (bid, records) in &bucket_records { + let deduped = sort_and_dedup(records.clone()); + let path = bucket_path(dir, *bid); + rewrite_file(&path, &deduped)?; + } + + Ok(()) + } +} + +// ── Helpers ──────────────────────────────────────────────────────────────── + +fn bucket_path(dir: &Path, bid: u16) -> PathBuf { + dir.join(format!("{:02x}.idx", bid)) +} + +/// Read records from a raw bucket file (may contain duplicates, not sorted). +fn load_records_from_file(path: &Path) -> Result> { + let data = fs::read(path)?; + let remainder = data.len() % INDEX_RECORD_SIZE; + let count = data.len() / INDEX_RECORD_SIZE; + let mut records = Vec::with_capacity(count); + for i in 0..count { + let start = i * INDEX_RECORD_SIZE; + let end = start + INDEX_RECORD_SIZE; + let buf: &[u8; INDEX_RECORD_SIZE] = data[start..end].try_into().map_err(|_| { + crate::error::Error::BucketIndexCorrupt { + path: path.to_path_buf(), + reason: "unexpected file size".into(), + } + })?; + records.push(IndexRecord::decode(buf)); + } + if remainder > 0 { + tracing::warn!( + "Bucket file {:?} has {} trailing bytes, ignoring", + path, + remainder + ); + } + Ok(records) +} + +/// Sort records by key, deduplicate keeping the one with the highest (segment_id, offset). +fn sort_and_dedup(mut records: Vec) -> Vec { + records.sort_by_key(|a| a.key); + let mut out = Vec::with_capacity(records.len()); + let mut i = 0; + while i < records.len() { + let mut best = i; + let mut j = i + 1; + while j < records.len() && records[j].key == records[i].key { + if records[j].segment_id > records[best].segment_id + || (records[j].segment_id == records[best].segment_id + && records[j].offset > records[best].offset) + { + best = j; + } + j += 1; + } + out.push(records[best].clone()); + i = j; + } + out +} + +/// Binary search for `key` in `bytes` (array of INDEX_RECORD_SIZE records, sorted). +fn binary_search_records(bytes: &[u8], key: &[u8; 32], count: usize) -> Option { + let mut lo = 0usize; + let mut hi = count; + while lo < hi { + let mid = lo + (hi - lo) / 2; + let rec_key = read_key_at(bytes, mid); + match rec_key.cmp(key) { + std::cmp::Ordering::Less => lo = mid + 1, + std::cmp::Ordering::Greater => hi = mid, + std::cmp::Ordering::Equal => return Some(mid), + } + } + None +} + +/// Read the key at index `idx` from a byte slice of INDEX_RECORD_SIZE records. +fn read_key_at(bytes: &[u8], idx: usize) -> &[u8; 32] { + let start = idx * INDEX_RECORD_SIZE; + bytes[start..start + 32].try_into().unwrap() +} + +/// Read a full IndexRecord at index `idx`. +fn read_record_at(bytes: &[u8], idx: usize) -> IndexRecord { + let start = idx * INDEX_RECORD_SIZE; + let buf: &[u8; INDEX_RECORD_SIZE] = bytes[start..start + INDEX_RECORD_SIZE] + .try_into() + .unwrap(); + IndexRecord::decode(buf) +} + +/// Atomically rewrite a bucket file with sorted, deduplicated records. +fn rewrite_file(path: &Path, records: &[IndexRecord]) -> Result<()> { + let mut buf = Vec::with_capacity(records.len() * INDEX_RECORD_SIZE); + for r in records { + buf.extend_from_slice(&r.encode()); + } + crate::fs::create_atomic(path, &buf) +} + #[cfg(test)] mod tests { use super::*; @@ -264,66 +498,111 @@ mod tests { assert_eq!(bucket_id(&key), 15); key[0] = 0x00; key[1] = 0x10; - assert_eq!(bucket_id(&key), 0); + assert_eq!(bucket_id(&key), 16); } #[test] - fn test_bucket_append_and_load() { - let dir = TempDir::new().unwrap(); - let bucket = BucketFile::open(dir.path(), 0); - BucketFile::ensure_dir(dir.path()).unwrap(); - - let r1 = IndexRecord::new([1u8; 32], 1, 100, 50, 0); - let r2 = IndexRecord::new([2u8; 32], 1, 200, 60, 0); - - bucket.append(&r1).unwrap(); - bucket.append(&r2).unwrap(); - - let loaded = bucket.load_all().unwrap(); - assert_eq!(loaded.len(), 2); - assert_eq!(loaded[0].key, [1u8; 32]); - assert_eq!(loaded[1].key, [2u8; 32]); - } - - #[test] - fn test_bucket_index_dedup() { + fn test_sort_and_dedup_keeps_latest() { let recs = vec![ IndexRecord::new([1u8; 32], 1, 100, 50, 0), - IndexRecord::new([1u8; 32], 2, 200, 50, 0), // newer offset wins + IndexRecord::new([1u8; 32], 2, 200, 50, 0), IndexRecord::new([2u8; 32], 1, 300, 60, 0), ]; - let idx = BucketIndex::from_records(recs, 0); - assert_eq!(idx.len(), 2); - let found = idx.find(&[1u8; 32]).unwrap(); + let deduped = sort_and_dedup(recs); + assert_eq!(deduped.len(), 2); + let found = &deduped[0]; assert_eq!(found.segment_id, 2); assert_eq!(found.offset, 200); } #[test] - fn test_bucket_index_find_missing() { - let recs = vec![IndexRecord::new([1u8; 32], 1, 100, 50, 0)]; - let idx = BucketIndex::from_records(recs, 0); - assert!(idx.find(&[99u8; 32]).is_none()); + fn test_bucket_store_put_and_get() { + let dir = TempDir::new().unwrap(); + let store = BucketStore::open(dir.path(), 100).unwrap(); + + let key = [0xAA; 32]; + let rec = IndexRecord::new(key, 1, 0, 500, 0); + store.insert(rec).unwrap(); + + let found = store.get(&key).unwrap(); + assert!(found.is_some()); + assert_eq!(found.unwrap().data_size, 500); } #[test] - fn test_bucket_rewrite() { + fn test_bucket_store_get_missing() { let dir = TempDir::new().unwrap(); - let bucket = BucketFile::open(dir.path(), 0); - BucketFile::ensure_dir(dir.path()).unwrap(); + let store = BucketStore::open(dir.path(), 100).unwrap(); - let r1 = IndexRecord::new([3u8; 32], 1, 300, 70, 0); - let r2 = IndexRecord::new([1u8; 32], 1, 100, 50, 0); - bucket.append(&r1).unwrap(); - bucket.append(&r2).unwrap(); + let result = store.get(&[0xFF; 32]).unwrap(); + assert!(result.is_none()); + } - // Rewrite sorted - let sorted = vec![r2.clone(), r1.clone()]; - bucket.rewrite(&sorted).unwrap(); + #[test] + fn test_bucket_store_tombstone() { + let dir = TempDir::new().unwrap(); + let store = BucketStore::open(dir.path(), 100).unwrap(); - let loaded = bucket.load_all().unwrap(); - assert_eq!(loaded.len(), 2); - assert_eq!(loaded[0].key, [1u8; 32]); - assert_eq!(loaded[1].key, [3u8; 32]); + let key = [0xBB; 32]; + + // Write then tombstone + store + .insert(IndexRecord::new(key, 1, 0, 100, 0)) + .unwrap(); + store + .insert(IndexRecord::new(key, 2, 0, 0, 1)) + .unwrap(); + + let result = store.get(&key).unwrap(); + assert!(result.is_none()); + } + + #[test] + fn test_bucket_store_compact() { + let dir = TempDir::new().unwrap(); + // Use small threshold to trigger auto-compact + let store = BucketStore::open(dir.path(), 5).unwrap(); + + // Write 10 records for same bucket (all same first 2 bytes → same bucket) + for i in 0u8..10 { + let mut key = [0u8; 32]; + key[0..2].copy_from_slice(&[0x00, 0x00]); // same bucket + key[2] = i; + store + .insert(IndexRecord::new(key, 1, i as u64 * 100, 50, 0)) + .unwrap(); + } + + // All should be readable after auto-compact + for i in 0u8..10 { + let mut key = [0u8; 32]; + key[0..2].copy_from_slice(&[0x00, 0x00]); + key[2] = i; + let found = store.get(&key).unwrap(); + assert!(found.is_some(), "key {} should exist after compact", i); + } + } + + #[test] + fn test_bucket_store_persistence() { + let dir = TempDir::new().unwrap(); + let dir_path = dir.path().to_path_buf(); + + let key = [0xCC; 32]; + { + let store = BucketStore::open(&dir_path, 100).unwrap(); + store + .insert(IndexRecord::new(key, 1, 42, 512, 0)) + .unwrap(); + store.compact_all().unwrap(); + } + + // Reopen + { + let store = BucketStore::open(&dir_path, 100).unwrap(); + let found = store.get(&key).unwrap(); + assert!(found.is_some()); + assert_eq!(found.unwrap().offset, 42); + } } } diff --git a/crates/blob/src/cache.rs b/crates/blob/src/cache.rs deleted file mode 100644 index b25830d..0000000 --- a/crates/blob/src/cache.rs +++ /dev/null @@ -1,219 +0,0 @@ -use std::collections::HashMap; -use std::path::Path; -use std::sync::Mutex; - -use crate::bucket::{BucketFile, BucketIndex, IndexRecord}; -use crate::error::Result; - -type CacheKey = (String, u16); - -/// Thread-safe LRU bucket cache with single-lock interior. -/// Eliminates the TOCTOU race in the old two-Mutex design. -pub struct BucketCache { - inner: Mutex, -} - -struct CacheInner { - max_entries: usize, - entries: Vec, - index: HashMap, -} - -struct CacheEntry { - key: CacheKey, - index: BucketIndex, -} - -impl BucketCache { - pub fn new(max_entries: usize) -> Self { - Self { - inner: Mutex::new(CacheInner { - max_entries: max_entries.max(1), - entries: Vec::new(), - index: HashMap::new(), - }), - } - } - - /// Get or load a bucket index. Eliminates TOCTOU via double-checked locking. - pub fn get_or_load( - &self, - account: &str, - bucket_id: u16, - account_dir: &Path, - ) -> Result> { - let key: CacheKey = (account.to_string(), bucket_id); - - // Check cache - { - let inner = self.inner.lock().unwrap(); - if let Some(&pos) = inner.index.get(&key) { - return Ok(inner.entries[pos].index.records.clone()); - } - } - - // Load from disk - let bucket_file = BucketFile::open(account_dir, bucket_id); - let bucket_index = bucket_file.load_index()?; - let records = bucket_index.records.clone(); - - // Insert with double-check (another thread might have beaten us) - { - let mut inner = self.inner.lock().unwrap(); - if let Some(&pos) = inner.index.get(&key) { - return Ok(inner.entries[pos].index.records.clone()); - } - // Evict if full - if inner.entries.len() >= inner.max_entries { - if let Some(evicted) = inner.entries.pop() { - inner.index.remove(&evicted.key); - } - } - // Insert at front - inner.entries.insert(0, CacheEntry { - key: key.clone(), - index: bucket_index, - }); - // Rebuild index - inner.index.clear(); - for i in 0..inner.entries.len() { - let key = inner.entries[i].key.clone(); - inner.index.insert(key, i); - } - } - - Ok(records) - } - - /// Insert or update a single record in a cached bucket. - pub fn update_record(&self, account: &str, bucket_id: u16, record: IndexRecord) { - let key: CacheKey = (account.to_string(), bucket_id); - let mut inner = self.inner.lock().unwrap(); - - if let Some(&pos) = inner.index.get(&key) { - inner.entries[pos].index.insert(record); - // Move to front - let entry = inner.entries.remove(pos); - inner.entries.insert(0, entry); - // Rebuild index - inner.index.clear(); - for i in 0..inner.entries.len() { - let entry_key = inner.entries[i].key.clone(); - inner.index.insert(entry_key, i); - } - } - } - - /// Invalidate a cached bucket (after GC rewrites bucket files). - pub fn invalidate(&self, account: &str, bucket_id: u16) { - let key: CacheKey = (account.to_string(), bucket_id); - let mut inner = self.inner.lock().unwrap(); - - if let Some(&pos) = inner.index.get(&key) { - inner.entries.remove(pos); - inner.index.clear(); - for i in 0..inner.entries.len() { - let entry_key = inner.entries[i].key.clone(); - inner.index.insert(entry_key, i); - } - } - } - - pub fn len(&self) -> usize { - self.inner.lock().unwrap().entries.len() - } - - pub fn is_empty(&self) -> bool { - self.inner.lock().unwrap().entries.is_empty() - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::bucket::IndexRecord; - use tempfile::TempDir; - - #[test] - fn test_cache_miss_loads_from_disk() { - let dir = TempDir::new().unwrap(); - crate::bucket::BucketFile::ensure_dir(dir.path()).unwrap(); - let bf = BucketFile::open(dir.path(), 0); - bf.append(&IndexRecord::new([1u8; 32], 1, 100, 50, 0)) - .unwrap(); - - let cache = BucketCache::new(10); - let records = cache - .get_or_load("test", 0, dir.path()) - .unwrap(); - assert_eq!(records.len(), 1); - } - - #[test] - fn test_cache_hit() { - let dir = TempDir::new().unwrap(); - crate::bucket::BucketFile::ensure_dir(dir.path()).unwrap(); - let bf = BucketFile::open(dir.path(), 0); - bf.append(&IndexRecord::new([2u8; 32], 1, 200, 60, 0)) - .unwrap(); - - let cache = BucketCache::new(10); - let _ = cache.get_or_load("test", 0, dir.path()).unwrap(); - let records = cache - .get_or_load("test", 0, dir.path()) - .unwrap(); - assert_eq!(records.len(), 1); - assert_eq!(cache.len(), 1); - } - - #[test] - fn test_cache_eviction() { - let dir = TempDir::new().unwrap(); - crate::bucket::BucketFile::ensure_dir(dir.path()).unwrap(); - let cache = BucketCache::new(2); - - for b in 0..4 { - let bf = BucketFile::open(dir.path(), b); - bf.append(&IndexRecord::new([b as u8; 32], 1, 100, 50, 0)) - .unwrap(); - let _ = cache.get_or_load("test", b, dir.path()).unwrap(); - } - - assert!(cache.len() <= 2); - } - - #[test] - fn test_concurrent_get_or_load_no_deadlock() { - use std::sync::Arc; - use std::thread; - - let dir = TempDir::new().unwrap(); - crate::bucket::BucketFile::ensure_dir(dir.path()).unwrap(); - let bf = BucketFile::open(dir.path(), 0); - for i in 0..10u8 { - bf.append(&IndexRecord::new([i; 32], 1, i as u64 * 100, 50, 0)) - .unwrap(); - } - - let cache = Arc::new(BucketCache::new(10)); - let dir_path = dir.path().to_path_buf(); - - let mut handles = vec![]; - for _ in 0..4 { - let cache = cache.clone(); - let dir_path = dir_path.clone(); - handles.push(thread::spawn(move || { - for _ in 0..100 { - let records = cache - .get_or_load("test", 0, &dir_path) - .unwrap(); - assert_eq!(records.len(), 10); - } - })); - } - - for h in handles { - h.join().unwrap(); - } - } -} diff --git a/crates/blob/src/engine.rs b/crates/blob/src/engine.rs index 0b71a6d..6ecbe26 100644 --- a/crates/blob/src/engine.rs +++ b/crates/blob/src/engine.rs @@ -1,28 +1,52 @@ use std::collections::HashMap; use std::fs; use std::path::{Path, PathBuf}; -use std::sync::{Arc, RwLock}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; -use crate::account::AccountHandle; -use crate::bucket::{self, IndexRecord}; -use crate::cache::BucketCache; +use crate::bucket::{BucketStore, IndexRecord}; use crate::compress; use crate::error::{Error, Result}; +use crate::file_pool::FilePool; use crate::gc::{self, GcStats}; -use crate::meta::GlobalMeta; -use crate::segment::SegmentReader; +use crate::meta::{GlobalMeta, SegmentStats}; +use crate::segment::{self, SegmentReader, SegmentWriter}; use crate::types::{Codec, Config, ENTRY_HEADER_SIZE}; +/// Global content-addressable blob store. +/// +/// All data is keyed by a 32-byte content hash. Identical content is stored +/// only once. The caller is responsible for tracking which entities reference +/// which content hashes — this crate is a pure content-addressable KV engine. pub struct Engine { - root: PathBuf, + shared: Arc, + flush_handle: Mutex>, +} + +struct EngineShared { config: Config, - cache: BucketCache, - accounts: RwLock>>, + inner: RwLock, + bucket_store: BucketStore, + write_mutex: Mutex<()>, + file_pool: FilePool, +} + +struct FlushHandle { + handle: JoinHandle<()>, + stop: Arc, +} + +struct EngineInner { + root: PathBuf, + meta: GlobalMeta, + active_writer: SegmentWriter, + readers: HashMap, } #[derive(Debug, Clone)] -pub struct AccountStats { - pub account_id: String, +pub struct Stats { pub total_keys: u64, pub total_bytes: u64, pub deleted_bytes: u64, @@ -32,255 +56,262 @@ pub struct AccountStats { impl Engine { pub fn open(path: &Path, config: Config) -> Result { config.validate()?; + fs::create_dir_all(path)?; - fs::create_dir_all(path.join("accounts"))?; + fs::create_dir_all(path.join("segments"))?; - let mut global = GlobalMeta::load(path)?; - global.save(path)?; + let bucket_dir = path.join("buckets"); + let bucket_store = BucketStore::open(&bucket_dir, config.compact_threshold)?; - let cache = BucketCache::new(config.lru_bucket_count); + let mut meta = GlobalMeta::load(path)?; - let accounts_dir = path.join("accounts"); - let mut accounts = HashMap::new(); - - if accounts_dir.exists() { - for entry in fs::read_dir(&accounts_dir)? { + // Discover existing segments on disk + let seg_dir = path.join("segments"); + let mut disk_segments: Vec = Vec::new(); + if seg_dir.exists() { + for entry in fs::read_dir(&seg_dir)? { let entry = entry?; - if entry.file_type()?.is_dir() { - let account_name = entry.file_name().to_string_lossy().into_owned(); - - let _ = crate::recovery::cleanup_temp_files(&entry.path()); - - match crate::recovery::recover_account(&entry.path()) { - Ok(_meta) => { - match AccountHandle::open(path, &account_name) { - Ok(handle) => { - accounts.insert(account_name, handle); - } - Err(e) => { - tracing::warn!( - "Failed to open account {}: {}", - account_name, - e - ); - } - } - } - Err(e) => { - tracing::warn!( - "Failed to recover account {}: {}", - account_name, - e - ); + let name = entry.file_name(); + let name_str = name.to_string_lossy(); + if name_str.ends_with(".seg") && !name_str.contains("temp_") { + if let Some(id_str) = name_str.strip_suffix(".seg") { + if let Ok(id) = id_str.parse::() { + disk_segments.push(id); } } } } } + disk_segments.sort_unstable(); - global.accounts = accounts.keys().cloned().collect(); - global.save(path)?; + for &seg_id in &disk_segments { + if !meta.segments.contains_key(&seg_id) { + meta.segments.insert(seg_id, SegmentStats::new(seg_id)); + } + } + + let max_disk_id = disk_segments.last().copied().unwrap_or(0); + if max_disk_id > meta.active_segment_id { + meta.active_segment_id = max_disk_id; + } + if meta.active_segment_id == 0 { + meta.active_segment_id = 1; + } + + crate::recovery::cleanup_temp_files(path)?; + crate::recovery::recover(path, &mut meta)?; + + let seg_path = path + .join("segments") + .join(segment::segment_filename(meta.active_segment_id)); + let active_writer = if seg_path.exists() { + SegmentWriter::open_append(seg_path, meta.active_segment_id)? + } else { + SegmentWriter::create(seg_path, meta.active_segment_id)? + }; + + let mut readers = HashMap::new(); + for (&seg_id, stats) in &meta.segments { + if stats.sealed { + let seg_path = path + .join("segments") + .join(segment::segment_filename(seg_id)); + if seg_path.exists() { + readers.insert(seg_id, SegmentReader::open(seg_path, seg_id)?); + } + } + } + + meta.save(path)?; + + let shared = Arc::new(EngineShared { + config: config.clone(), + inner: RwLock::new(EngineInner { + root: path.to_path_buf(), + meta, + active_writer, + readers, + }), + bucket_store, + write_mutex: Mutex::new(()), + file_pool: FilePool::new(8), + }); + + let flush_handle = if config.flush_interval_secs > 0 { + let shared2 = Arc::clone(&shared); + let stop = Arc::new(AtomicBool::new(false)); + let stop2 = Arc::clone(&stop); + let interval = Duration::from_secs(config.flush_interval_secs); + + let handle = thread::Builder::new() + .name("blob-flush".into()) + .spawn(move || { + while !stop2.load(Ordering::Acquire) { + thread::park_timeout(interval); + if stop2.load(Ordering::Acquire) { + break; + } + let _lock = shared2.write_mutex.lock().unwrap(); + let mut inner = shared2.inner.write().unwrap(); + if let Err(e) = inner.flush_active() { + tracing::error!("background flush failed: {}", e); + } + } + }) + .expect("failed to spawn blob-flush thread"); + + Some(FlushHandle { handle, stop }) + } else { + None + }; Ok(Self { - root: path.to_path_buf(), - config, - cache, - accounts: RwLock::new(accounts), + shared, + flush_handle: Mutex::new(flush_handle), }) } - // ── Account management ────────────────────────────────────────────── - - pub fn create_account(&self, account_id: &str) -> Result<()> { - let mut accounts = self.accounts.write().unwrap(); - if accounts.contains_key(account_id) { - return Err(Error::AccountAlreadyExists(account_id.to_string())); - } - let handle = AccountHandle::create(&self.root, account_id)?; - accounts.insert(account_id.to_string(), handle); - - let mut global = GlobalMeta::load(&self.root)?; - global.accounts = accounts.keys().cloned().collect(); - global.save(&self.root)?; - - Ok(()) - } - - pub fn delete_account(&self, account_id: &str) -> Result<()> { - let mut accounts = self.accounts.write().unwrap(); - let handle = accounts - .remove(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))?; - - let account_dir = handle.dir().to_path_buf(); - drop(handle); - fs::remove_dir_all(&account_dir)?; - - let mut global = GlobalMeta::load(&self.root)?; - global.accounts = accounts.keys().cloned().collect(); - global.save(&self.root)?; - - Ok(()) - } - - pub fn list_accounts(&self) -> Vec { - let accounts = self.accounts.read().unwrap(); - accounts.keys().cloned().collect() - } - // ── Read / Write / Delete ─────────────────────────────────────────── - pub fn write( - &self, - account_id: &str, - key: [u8; 32], - value: &[u8], - codec: Codec, - ) -> Result<()> { + pub fn put(&self, key: [u8; 32], value: &[u8], codec: Codec) -> Result<()> { if value.len() > crate::types::MAX_VALUE_SIZE { return Err(Error::ValueTooLarge { size: value.len() }); } - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() - }; + let _write_lock = self.shared.write_mutex.lock().unwrap(); + let mut inner = self.shared.inner.write().unwrap(); - let _write_lock = handle.write_mutex.lock().unwrap(); - let mut inner = handle.write(); - - let (data, actual_codec) = - compress::compress(value, codec, self.config.compress_threshold, self.config.compression_level); + let (data, actual_codec) = compress::compress( + value, + codec, + self.shared.config.compress_threshold, + self.shared.config.compression_level, + ); let (segment_id, offset, data_size) = - inner.write_entry(key, &data, 0, actual_codec)?; + inner.append_entry(key, &data, 0, actual_codec)?; let record = IndexRecord::new(key, segment_id, offset, data_size, 0); - inner.append_index(&record)?; + self.shared.bucket_store.insert(record)?; let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; inner.mark_indexed(segment_id, entry_end)?; - let bucket_id = bucket::bucket_id(&key); - self.cache.update_record(account_id, bucket_id, record); - Ok(()) } - pub fn read(&self, account_id: &str, key: &[u8; 32]) -> Result>> { - let bucket_id = bucket::bucket_id(key); - - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() + pub fn get(&self, key: &[u8; 32]) -> Result>> { + let record = match self.shared.bucket_store.get(key)? { + Some(r) => r, + None => return Ok(None), }; - let (record, seg_path): (IndexRecord, PathBuf) = { - let inner = handle.read(); - let records = self - .cache - .get_or_load(account_id, bucket_id, handle.dir())?; - match records.binary_search_by(|r| r.key.cmp(key)) { - Ok(idx) => { - let r = records[idx].clone(); - if r.is_tombstone() { - return Ok(None); - } - let seg_path = inner.segment_path(r.segment_id)?; - (r, seg_path) - } - Err(_) => return Ok(None), - } - }; + let inner = self.shared.inner.read().unwrap(); + let seg_path = inner.segment_path(record.segment_id)?; if !seg_path.exists() { return Err(Error::SegmentNotFound(record.segment_id)); } let reader = SegmentReader::open(seg_path.clone(), record.segment_id)?; - let file = handle.get_segment_file(record.segment_id, &seg_path)?; + let file = self.shared.file_pool.get(record.segment_id, &seg_path)?; let (entry, _) = reader.read_entry_at_file(record.offset, &file)?; let value = compress::decompress(&entry.data, entry.codec, entry.raw_size as usize)?; - Ok(Some(value)) } - pub fn delete(&self, account_id: &str, key: &[u8; 32]) -> Result<()> { - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() - }; - - let _write_lock = handle.write_mutex.lock().unwrap(); - let mut inner = handle.write(); + pub fn delete(&self, key: &[u8; 32]) -> Result<()> { + let _write_lock = self.shared.write_mutex.lock().unwrap(); + let mut inner = self.shared.inner.write().unwrap(); let (segment_id, offset, data_size) = - inner.write_entry(*key, &[], 1, Codec::None)?; + inner.append_entry(*key, &[], 1, Codec::None)?; let record = IndexRecord::new(*key, segment_id, offset, data_size, 1); - inner.append_index(&record)?; + self.shared.bucket_store.insert(record)?; let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; inner.mark_indexed(segment_id, entry_end)?; - let bucket_id = bucket::bucket_id(key); - self.cache.update_record(account_id, bucket_id, record); + Ok(()) + } + + pub fn exists(&self, key: &[u8; 32]) -> Result { + self.shared.bucket_store.exists(key) + } + + // ── Batch delete ───────────────────────────────────────────────────── + + pub fn delete_batch(&self, keys: &[[u8; 32]]) -> Result<()> { + if keys.is_empty() { + return Ok(()); + } + + let _write_lock = self.shared.write_mutex.lock().unwrap(); + let mut inner = self.shared.inner.write().unwrap(); + + let mut records: Vec = Vec::with_capacity(keys.len()); + let mut ends: Vec<(u32, u64)> = Vec::with_capacity(keys.len()); + + for key in keys { + let (segment_id, offset, data_size) = + inner.append_entry(*key, &[], 1, Codec::None)?; + + let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; + records.push(IndexRecord::new(*key, segment_id, offset, data_size, 1)); + ends.push((segment_id, entry_end)); + } + + inner.flush_active()?; + + self.shared.bucket_store.insert_batch(&records)?; + + for (segment_id, entry_end) in &ends { + inner.mark_indexed(*segment_id, *entry_end)?; + } Ok(()) } // ── Batch write ───────────────────────────────────────────────────── - pub fn write_batch(&self, account_id: &str, entries: &[([u8; 32], Vec, Codec)]) -> Result<()> { + pub fn put_batch(&self, entries: &[([u8; 32], Vec, Codec)]) -> Result<()> { if entries.is_empty() { return Ok(()); } - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() - }; + let _write_lock = self.shared.write_mutex.lock().unwrap(); + let mut inner = self.shared.inner.write().unwrap(); - let _write_lock = handle.write_mutex.lock().unwrap(); - let mut inner = handle.write(); + let mut records: Vec = Vec::with_capacity(entries.len()); + let mut ends: Vec<(u32, u64)> = Vec::with_capacity(entries.len()); - let mut pending: Vec<(IndexRecord, u64)> = Vec::with_capacity(entries.len()); for (key, value, codec) in entries { if value.len() > crate::types::MAX_VALUE_SIZE { return Err(Error::ValueTooLarge { size: value.len() }); } - let (data, actual_codec) = - compress::compress(value, *codec, self.config.compress_threshold, self.config.compression_level); + let (data, actual_codec) = compress::compress( + value, + *codec, + self.shared.config.compress_threshold, + self.shared.config.compression_level, + ); let (segment_id, offset, data_size) = inner.append_entry(*key, &data, 0, actual_codec)?; let entry_end = offset + ENTRY_HEADER_SIZE as u64 + data_size as u64; - let record = IndexRecord::new(*key, segment_id, offset, data_size, 0); - pending.push((record, entry_end)); + records.push(IndexRecord::new(*key, segment_id, offset, data_size, 0)); + ends.push((segment_id, entry_end)); } inner.flush_active()?; - for (record, entry_end) in &pending { - inner.append_index(record)?; - inner.mark_indexed(record.segment_id, *entry_end)?; + self.shared.bucket_store.insert_batch(&records)?; - let bucket_id = bucket::bucket_id(&record.key); - self.cache.update_record(account_id, bucket_id, record.clone()); + for (segment_id, entry_end) in &ends { + inner.mark_indexed(*segment_id, *entry_end)?; } Ok(()) @@ -288,87 +319,99 @@ impl Engine { // ── GC ────────────────────────────────────────────────────────────── - pub fn gc(&self, account_id: &str) -> Result> { - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() + pub fn gc(&self) -> Result> { + // Phase 1: scan segments and write compacted temp file. + // Read-only with respect to Engine state — no write_mutex needed. + let prep = { + let inner = self.shared.inner.read().unwrap(); + gc::gc_prepare( + &inner.root, + &inner.meta, + self.shared.config.gc_deleted_ratio, + )? }; - // Hold write_mutex to prevent concurrent writes from racing - // with GC's bucket rebuild phase. - let _write_lock = handle.write_mutex.lock().unwrap(); + let prep = match prep { + Some(p) => p, + None => return Ok(None), + }; - let result = gc::gc_account(handle.dir(), self.config.gc_deleted_ratio)?; + // Phase 2: rename temp file + rebuild bucket indices. + // This is the only part that requires exclusive access. + let _write_lock = self.shared.write_mutex.lock().unwrap(); - // Invalidate FilePool for GC'd segments (they were rewritten via rename) - if let Some(ref stats) = result { - handle.invalidate_file_cache(stats.segment_id); + let stats = gc::gc_finish(prep)?; + self.shared.file_pool.invalidate(stats.segment_id); + + // Rebuild bucket indices from the updated segment files + let inner = self.shared.inner.write().unwrap(); + let seg_ids: Vec = inner.meta.segments.keys().copied().collect(); + let mut seg_refs: Vec<(u32, PathBuf)> = Vec::with_capacity(seg_ids.len()); + for &id in &seg_ids { + let p = inner.root.join("segments").join(segment::segment_filename(id)); + seg_refs.push((id, p)); } + let paths: Vec<(u32, &Path)> = + seg_refs.iter().map(|(id, p)| (*id, p.as_path())).collect(); - for bid in 0..crate::types::BUCKET_COUNT { - self.cache.invalidate(account_id, bid); - } + let bucket_dir = inner.root.join("buckets"); + BucketStore::rebuild_from_segments(&bucket_dir, &paths)?; - Ok(result) + // Update meta (segment stats changed after GC compaction) + let mut meta = crate::meta::GlobalMeta::load(&inner.root)?; + meta.active_segment_id = inner.meta.active_segment_id; + meta.save(&inner.root)?; + + // Reload bucket store mmaps after GC rewrites + self.shared.bucket_store.reload_all()?; + + Ok(Some(stats)) } - pub fn compact_buckets(&self, account_id: &str) -> Result<()> { - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() - }; + /// Run GC only if some segment exceeds the configured deleted-ratio threshold. + /// Returns `Ok(None)` immediately without acquiring the write lock when no + /// segment qualifies. + pub fn gc_if_needed(&self) -> Result> { + let inner = self.shared.inner.read().unwrap(); + let needs_gc = inner + .meta + .segments + .values() + .any(|s| s.sealed && s.deleted_ratio >= self.shared.config.gc_deleted_ratio); + drop(inner); - // Hold write_mutex — compact rewrites all bucket files. - let _write_lock = handle.write_mutex.lock().unwrap(); - - gc::compact_buckets(handle.dir())?; - - for bid in 0..crate::types::BUCKET_COUNT { - self.cache.invalidate(account_id, bid); + if needs_gc { + self.gc() + } else { + Ok(None) } - - Ok(()) } - // ── Stats / Shutdown ──────────────────────────────────────────────── + // ── Flush / Stats / Shutdown ──────────────────────────────────────── - pub fn stats(&self, account_id: &str) -> Result { - let handle = { - let accounts = self.accounts.read().unwrap(); - accounts - .get(account_id) - .ok_or_else(|| Error::AccountNotFound(account_id.to_string()))? - .clone() - }; + /// Fsync the active segment and save metadata without compacting buckets. + /// Lightweight checkpoint suitable for periodic calls from the background + /// flush thread or external schedulers. + pub fn flush(&self) -> Result<()> { + let _write_lock = self.shared.write_mutex.lock().unwrap(); + let mut inner = self.shared.inner.write().unwrap(); + inner.flush_active() + } + + pub fn stats(&self) -> Result { + let inner = self.shared.inner.read().unwrap(); + let meta = &inner.meta; - let inner = handle.read(); - let meta = inner.meta(); let mut total_bytes = 0u64; let mut deleted_bytes = 0u64; - for seg in meta.segments.values() { total_bytes += seg.total_bytes; deleted_bytes += seg.deleted_bytes; } - let mut total_keys = 0u64; - for bid in 0..crate::types::BUCKET_COUNT { - if let Ok(records) = - self.cache - .get_or_load(account_id, bid, handle.dir()) - { - total_keys += records.iter().filter(|r| !r.is_tombstone()).count() as u64; - } - } + let total_keys = self.shared.bucket_store.total_keys() as u64; - Ok(AccountStats { - account_id: account_id.to_string(), + Ok(Stats { total_keys, total_bytes, deleted_bytes, @@ -377,13 +420,19 @@ impl Engine { } pub fn shutdown(&self) -> Result<()> { - let accounts = self.accounts.read().unwrap(); - for (_, handle) in accounts.iter() { - let mut inner = handle.write(); - inner.flush_active()?; + // Stop background flush thread first + if let Some(fh) = self.flush_handle.lock().unwrap().take() { + fh.stop.store(true, Ordering::Release); + fh.handle.thread().unpark(); + let _ = fh.handle.join(); } - let global = GlobalMeta::load(&self.root)?; - global.save(&self.root)?; + + let mut inner = self.shared.inner.write().unwrap(); + inner.flush_active()?; + + self.shared.bucket_store.compact_all()?; + + inner.meta.save(&inner.root)?; tracing::info!("bichon-blob shut down cleanly"); Ok(()) } @@ -396,3 +445,97 @@ impl Drop for Engine { } } } + +// ── EngineInner ─────────────────────────────────────────────────────────── + +impl EngineInner { + fn append_entry( + &mut self, + key: [u8; 32], + data: &[u8], + flags: u8, + codec: Codec, + ) -> Result<(u32, u64, u32)> { + if self.active_writer.is_full() { + self.seal_active()?; + } + + use crate::segment::Entry; + let entry = if flags == 1 { + Entry::tombstone(key) + } else { + Entry::new(key, data, flags, codec) + }; + + let data_size = entry.data.len() as u32; + let segment_id = self.active_writer.id(); + let offset = self.active_writer.append(&entry)?; + + let stats = self + .meta + .segments + .entry(segment_id) + .or_insert_with(|| SegmentStats::new(segment_id)); + stats.total_bytes += data_size as u64; + if flags == 1 { + stats.deleted_bytes += entry.raw_size as u64; + } + stats.recompute_ratio(); + + Ok((segment_id, offset, data_size)) + } + + fn flush_active(&mut self) -> Result<()> { + self.active_writer.fsync()?; + self.meta.save(&self.root) + } + + fn mark_indexed(&mut self, segment_id: u32, offset: u64) -> Result<()> { + if let Some(stats) = self.meta.segments.get_mut(&segment_id) { + if offset > stats.indexed_up_to_offset { + stats.indexed_up_to_offset = offset; + } + } + self.meta.save(&self.root) + } + + fn seal_active(&mut self) -> Result<()> { + let old_id = self.active_writer.id(); + let old_stats = self + .meta + .segments + .entry(old_id) + .or_insert_with(|| SegmentStats::new(old_id)); + old_stats.sealed = true; + + let seg_path = self + .root + .join("segments") + .join(segment::segment_filename(old_id)); + self.readers + .insert(old_id, SegmentReader::open(seg_path, old_id)?); + + let new_id = old_id + 1; + self.meta.active_segment_id = new_id; + let new_path = self + .root + .join("segments") + .join(segment::segment_filename(new_id)); + self.active_writer = SegmentWriter::create(new_path, new_id)?; + self.meta.save(&self.root)?; + + Ok(()) + } + + fn segment_path(&self, segment_id: u32) -> Result { + let path = self + .root + .join("segments") + .join(segment::segment_filename(segment_id)); + if path.exists() { + Ok(path) + } else { + Err(Error::SegmentNotFound(segment_id)) + } + } +} diff --git a/crates/blob/src/error.rs b/crates/blob/src/error.rs index 377c840..7848965 100644 --- a/crates/blob/src/error.rs +++ b/crates/blob/src/error.rs @@ -20,12 +20,6 @@ pub enum Error { reason: String, }, - #[error("Account not found: {0}")] - AccountNotFound(String), - - #[error("Account already exists: {0}")] - AccountAlreadyExists(String), - #[error("Segment not found: {0}")] SegmentNotFound(u32), diff --git a/crates/blob/src/file_pool.rs b/crates/blob/src/file_pool.rs index ec8ec5b..272c3e4 100644 --- a/crates/blob/src/file_pool.rs +++ b/crates/blob/src/file_pool.rs @@ -7,10 +7,10 @@ use crate::error::Result; use crate::fs as fs_util; /// Simple LRU pool of open file handles, keyed by segment_id. -/// Uses Arc> to allow safe concurrent reads from the same segment. +/// Uses pread-based reads so a single `Arc` supports concurrent access. pub struct FilePool { max_entries: usize, - entries: Mutex>)>>, + entries: Mutex)>>, } impl FilePool { @@ -22,7 +22,7 @@ impl FilePool { } /// Get an open File for the given segment. Reuses cached handle if available. - pub fn get(&self, seg_id: u32, path: &Path) -> Result>> { + pub fn get(&self, seg_id: u32, path: &Path) -> Result> { let mut entries = self.entries.lock().unwrap(); // Check for existing entry @@ -35,7 +35,7 @@ impl FilePool { } // Open new file - let file = Arc::new(Mutex::new(fs_util::open_read(path)?)); + let file = Arc::new(fs_util::open_read(path)?); // Evict oldest if full if entries.len() >= self.max_entries { diff --git a/crates/blob/src/fs.rs b/crates/blob/src/fs.rs index 5e0ffc1..750a860 100644 --- a/crates/blob/src/fs.rs +++ b/crates/blob/src/fs.rs @@ -86,6 +86,31 @@ pub fn truncate(path: &Path, size: u64) -> Result<()> { Ok(()) } +/// Positional read: read `buf.len()` bytes at `offset` from `file`. +/// Uses platform-specific pread so `&File` (shared ref) suffices — +/// no Mutex needed for concurrent reads. +pub fn pread_exact(file: &File, offset: u64, buf: &mut [u8]) -> Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.read_exact_at(buf, offset)?; + } + #[cfg(windows)] + { + use std::os::windows::fs::FileExt; + file.seek_read(buf, offset)?; + } + #[cfg(not(any(unix, windows)))] + { + // Fallback: seek+read (requires &mut, so this is best-effort on exotic platforms) + use std::io::{Read, Seek, SeekFrom}; + let mut tmp = file.try_clone()?; + tmp.seek(SeekFrom::Start(offset))?; + tmp.read_exact(buf)?; + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/blob/src/gc.rs b/crates/blob/src/gc.rs index 619d08f..5829a2d 100644 --- a/crates/blob/src/gc.rs +++ b/crates/blob/src/gc.rs @@ -1,12 +1,9 @@ use std::collections::HashMap; use std::fs; -use std::path::Path; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::path::{Path, PathBuf}; -use crate::bucket::{self, BucketFile, BucketIndex, IndexRecord}; use crate::error::Result; -#[cfg(test)] -use crate::meta::SegmentStats; +use crate::meta::GlobalMeta; use crate::segment::{self, SegmentReader, SegmentWriter}; /// Result of a GC run. @@ -19,15 +16,27 @@ pub struct GcStats { pub entries_skipped: usize, } -/// Run GC on an account: pick the sealed segment with highest deleted_ratio, -/// rewrite it without deleted/overwritten entries, then rebuild all bucket files. -pub fn gc_account( - account_dir: &Path, - deleted_ratio_threshold: f64, -) -> Result> { - let meta = crate::meta::AccountMeta::load(account_dir)?; +/// Prepared GC result — the compacted segment has been written to a temp file +/// but not yet renamed over the original. `gc_finish` must be called to commit. +pub struct GcPrepare { + pub segment_id: u32, + pub bytes_before: u64, + pub bytes_after: u64, + pub entries_kept: usize, + pub entries_skipped: usize, + temp_path: PathBuf, + seg_path: PathBuf, +} - // Find the best candidate +/// Phase 1: pick the sealed segment with the highest deleted_ratio, scan all +/// segments to determine the latest entry for each key, then write a compacted +/// version of the target segment to a temp file. Does NOT rename — the caller +/// should hold the write lock only during `gc_finish`. +pub fn gc_prepare( + store_root: &Path, + meta: &GlobalMeta, + deleted_ratio_threshold: f64, +) -> Result> { let candidate = meta .segments .values() @@ -39,7 +48,7 @@ pub fn gc_account( None => return Ok(None), }; - let seg_path = account_dir + let seg_path = store_root .join("segments") .join(segment::segment_filename(target.segment_id)); let reader = SegmentReader::open(seg_path.clone(), target.segment_id)?; @@ -48,7 +57,7 @@ pub fn gc_account( let mut latest_key: HashMap<[u8; 32], (u32, u64)> = HashMap::new(); for &seg_id in meta.segments.keys() { - let rpath = account_dir + let rpath = store_root .join("segments") .join(segment::segment_filename(seg_id)); if !rpath.exists() { @@ -72,13 +81,13 @@ pub fn gc_account( })?; } - // Create temp segment with a unique name - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) + // Create temp segment (not renamed yet) + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_nanos(); let temp_name = format!("temp_{:016x}.seg", timestamp); - let temp_path = account_dir.join("segments").join(&temp_name); + let temp_path = store_root.join("segments").join(&temp_name); let mut writer = SegmentWriter::create(temp_path.clone(), target.segment_id)?; let mut bytes_after: u64 = 0; @@ -86,19 +95,16 @@ pub fn gc_account( let mut entries_skipped: usize = 0; reader.scan_entries(0, |entry, offset| { - // Skip tombstones if entry.is_tombstone() { entries_skipped += 1; return Ok(()); } - // Skip if this key has a newer entry in another segment if let Some((latest_seg, latest_off)) = latest_key.get(&entry.key) { if *latest_seg != target.segment_id || *latest_off != offset { entries_skipped += 1; return Ok(()); } } - // Keep this entry writer.append(entry)?; bytes_after += entry.data.len() as u64; entries_kept += 1; @@ -107,161 +113,25 @@ pub fn gc_account( writer.fsync()?; - // Atomic rename: replace old segment with new one - fs::rename(&temp_path, &seg_path)?; - - // Rebuild all bucket files - rebuild_buckets(account_dir, &meta)?; - - // Update meta - let mut meta = crate::meta::AccountMeta::load(account_dir)?; - if let Some(stats) = meta.segments.get_mut(&target.segment_id) { - stats.total_bytes = bytes_after; - stats.deleted_bytes = 0; - stats.recompute_ratio(); - } - meta.save(account_dir)?; - - Ok(Some(GcStats { + Ok(Some(GcPrepare { segment_id: target.segment_id, bytes_before: target.total_bytes, bytes_after, entries_kept, entries_skipped, + temp_path, + seg_path, })) } -/// Rebuild all 16 bucket files from scratch by scanning all segments. -fn rebuild_buckets(account_dir: &Path, meta: &crate::meta::AccountMeta) -> Result<()> { - let mut bucket_records: HashMap> = HashMap::new(); - for i in 0..crate::types::BUCKET_COUNT { - bucket_records.insert(i, Vec::new()); - } - - for &seg_id in meta.segments.keys() { - let seg_path = account_dir - .join("segments") - .join(segment::segment_filename(seg_id)); - if !seg_path.exists() { - continue; - } - let reader = SegmentReader::open(seg_path, seg_id)?; - reader.scan_entries(0, |entry, offset| { - let bid = bucket::bucket_id(&entry.key); - let rec = IndexRecord::new( - entry.key, - seg_id, - offset, - entry.data.len() as u32, - entry.flags, - ); - bucket_records.entry(bid).or_default().push(rec); - Ok(()) - })?; - } - - for (bid, records) in &bucket_records { - let index = BucketIndex::from_records(records.clone(), *bid); - let bf = BucketFile::open(account_dir, *bid); - bf.rewrite(&index.records)?; - } - - Ok(()) -} - -/// Compact bucket files: load, dedup, rewrite. -pub fn compact_buckets(account_dir: &Path) -> Result<()> { - for bid in 0..crate::types::BUCKET_COUNT { - let bf = BucketFile::open(account_dir, bid); - if bf.path().exists() { - let index = bf.load_index()?; - bf.rewrite(&index.records)?; - } - } - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::segment::Entry; - use crate::types::Codec; - use tempfile::TempDir; - - fn setup_account(dir: &Path) { - fs::create_dir_all(dir.join("segments")).unwrap(); - crate::bucket::BucketFile::ensure_dir(dir).unwrap(); - - let seg_path = dir - .join("segments") - .join(segment::segment_filename(1)); - let mut writer = SegmentWriter::create(seg_path, 1).unwrap(); - - // Write 5 entries - for i in 0..5u8 { - let mut key = [0u8; 32]; - key[0] = i; - let entry = Entry::new(key, &vec![i; 1000], 0, Codec::None); - writer.append(&entry).unwrap(); - } - - // Tombstone entry 2 - let mut key2 = [0u8; 32]; - key2[0] = 2; - let tomb = Entry::tombstone(key2); - writer.append(&tomb).unwrap(); - - writer.fsync().unwrap(); - - // Save meta - let mut meta = crate::meta::AccountMeta::new("test".into(), 2); - meta.segments.insert( - 1, - SegmentStats { - segment_id: 1, - total_bytes: 6000, - deleted_bytes: 1000, - deleted_ratio: 1000.0 / 6000.0, - sealed: true, - indexed_up_to_offset: 0, - }, - ); - // Make segment 2 active so segment 1 is sealed - let seg2_path = dir - .join("segments") - .join(segment::segment_filename(2)); - SegmentWriter::create(seg2_path, 2).unwrap(); - meta.save(dir).unwrap(); - } - - #[test] - fn test_gc_removes_tombstones() { - let dir = TempDir::new().unwrap(); - setup_account(dir.path()); - - let result = gc_account(dir.path(), 0.01).unwrap(); - assert!(result.is_some()); - - // Verify segment 1 no longer has the tombstone'd entry - let seg_path = dir - .path() - .join("segments") - .join(segment::segment_filename(1)); - let reader = SegmentReader::open(seg_path, 1).unwrap(); - let mut count = 0; - reader.scan_entries(0, |entry, _offset| { - count += 1; - assert!(entry.key[0] != 2); - Ok(()) - }).unwrap(); - assert_eq!(count, 4); // 5 original - 1 tombstoned - } - - #[test] - fn test_compact_buckets() { - let dir = TempDir::new().unwrap(); - setup_account(dir.path()); - compact_buckets(dir.path()).unwrap(); - // Should not panic - } +/// Phase 2: atomically replace the old segment with the compacted one. +pub fn gc_finish(prep: GcPrepare) -> Result { + fs::rename(&prep.temp_path, &prep.seg_path)?; + Ok(GcStats { + segment_id: prep.segment_id, + bytes_before: prep.bytes_before, + bytes_after: prep.bytes_after, + entries_kept: prep.entries_kept, + entries_skipped: prep.entries_skipped, + }) } diff --git a/crates/blob/src/lib.rs b/crates/blob/src/lib.rs index 78b9958..a73e669 100644 --- a/crates/blob/src/lib.rs +++ b/crates/blob/src/lib.rs @@ -1,6 +1,4 @@ -pub mod account; pub mod bucket; -pub mod cache; pub mod checksum; pub mod compress; pub mod engine; @@ -13,7 +11,6 @@ pub mod recovery; pub mod segment; pub mod types; -pub use account::AccountHandle; -pub use engine::{AccountStats, Engine}; +pub use engine::{Engine, Stats}; pub use error::{Error, Result}; pub use types::{Codec, Config}; diff --git a/crates/blob/src/meta.rs b/crates/blob/src/meta.rs index 6238e68..77bc005 100644 --- a/crates/blob/src/meta.rs +++ b/crates/blob/src/meta.rs @@ -5,12 +5,12 @@ use crate::checksum; use crate::error::Result; use serde::{Deserialize, Serialize}; -const META_VERSION: u32 = 1; +const META_VERSION: u32 = 2; // ── Helpers ──────────────────────────────────────────────────────────────── fn write_bin(path: &Path, value: &T) -> Result<()> { - let payload = bincode::serialize(value).map_err(|e| { + let payload = bincode::serde::encode_to_vec(value, bincode::config::standard()).map_err(|e| { crate::error::Error::CorruptMeta(format!("{}: bincode encode: {}", path.display(), e)) })?; let crc = checksum::crc32(&payload); @@ -40,55 +40,13 @@ fn read_bin Deserialize<'de>>(path: &Path) -> Result { if stored_crc != computed { return Err(crate::error::Error::CorruptMeta(path.display().to_string())); } - bincode::deserialize(&data[8..]).map_err(|e| { - crate::error::Error::CorruptMeta(format!("{}: bincode decode: {}", path.display(), e)) + bincode::serde::decode_from_slice(&data[8..], bincode::config::standard()) + .map(|(v, _)| v) + .map_err(|e| { + crate::error::Error::CorruptMeta(format!("{}: bincode decode: {}", path.display(), e)) }) } -// ── GlobalMeta ───────────────────────────────────────────────────────────── - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct GlobalMeta { - pub version: u32, - pub accounts: Vec, -} - -impl Default for GlobalMeta { - fn default() -> Self { - Self { - version: META_VERSION, - accounts: Vec::new(), - } - } -} - -impl GlobalMeta { - pub fn load(store_root: &Path) -> Result { - let bin_path = store_root.join("global_meta.bin"); - if bin_path.exists() { - return read_bin(&bin_path); - } - // Migration from JSON - let json_path = store_root.join("global_meta.json"); - if json_path.exists() { - let data = std::fs::read_to_string(&json_path)?; - let mut meta: Self = serde_json::from_str(&data)?; - meta.accounts.sort(); - write_bin(&bin_path, &meta)?; - let _ = std::fs::remove_file(&json_path); - return Ok(meta); - } - Ok(Self::default()) - } - - pub fn save(&self, store_root: &Path) -> Result<()> { - let path = store_root.join("global_meta.bin"); - let mut meta = self.clone(); - meta.accounts.sort(); - write_bin(&path, &meta) - } -} - // ── SegmentStats ─────────────────────────────────────────────────────────── #[derive(Debug, Clone, Serialize, Deserialize)] @@ -99,8 +57,10 @@ pub struct SegmentStats { pub deleted_ratio: f64, pub sealed: bool, /// Byte offset up to which entries have been indexed in bucket files. - /// Recovery starts scanning from here instead of 0. pub indexed_up_to_offset: u64, + /// Number of compacted (sorted, deduped) records in each bucket file for this segment. + /// Used by BucketStore on recovery to know where the clean portion ends. + pub bucket_compacted: u64, } impl SegmentStats { @@ -112,6 +72,7 @@ impl SegmentStats { deleted_ratio: 0.0, sealed: false, indexed_up_to_offset: 0, + bucket_compacted: 0, } } @@ -124,31 +85,31 @@ impl SegmentStats { } } -// ── AccountMeta ──────────────────────────────────────────────────────────── +// ── GlobalMeta ───────────────────────────────────────────────────────────── #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct AccountMeta { - pub account_id: String, +pub struct GlobalMeta { + pub version: u32, pub active_segment_id: u32, pub segments: BTreeMap, } -impl AccountMeta { - pub fn new(account_id: String, active_segment_id: u32) -> Self { +impl GlobalMeta { + pub fn new() -> Self { Self { - account_id, - active_segment_id, + version: META_VERSION, + active_segment_id: 1, segments: BTreeMap::new(), } } - pub fn load(account_dir: &Path) -> Result { - let bin_path = account_dir.join("meta.bin"); + pub fn load(store_root: &Path) -> Result { + let bin_path = store_root.join("meta.bin"); if bin_path.exists() { return read_bin(&bin_path); } - // Migration from JSON - let json_path = account_dir.join("meta.json"); + // Migration from old JSON format + let json_path = store_root.join("meta.json"); if json_path.exists() { let data = std::fs::read_to_string(&json_path)?; let meta: Self = serde_json::from_str(&data)?; @@ -156,13 +117,23 @@ impl AccountMeta { let _ = std::fs::remove_file(&json_path); return Ok(meta); } - Err(crate::error::Error::AccountNotFound( - account_dir.to_string_lossy().into(), - )) + // Migration from old global_meta.bin (v1, only had accounts list) + let old_path = store_root.join("global_meta.bin"); + if old_path.exists() { + let _ = std::fs::remove_file(&old_path); + } + Ok(Self::new()) } - pub fn save(&self, account_dir: &Path) -> Result<()> { - write_bin(&account_dir.join("meta.bin"), self) + pub fn save(&self, store_root: &Path) -> Result<()> { + let path = store_root.join("meta.bin"); + write_bin(&path, self) + } +} + +impl Default for GlobalMeta { + fn default() -> Self { + Self::new() } } @@ -172,45 +143,9 @@ mod tests { use tempfile::TempDir; #[test] - fn test_global_meta_bin_roundtrip() { + fn test_global_meta_roundtrip() { let dir = TempDir::new().unwrap(); - let mut meta = GlobalMeta::default(); - meta.accounts.push("alice".into()); - meta.save(dir.path()).unwrap(); - - let loaded = GlobalMeta::load(dir.path()).unwrap(); - assert_eq!(loaded.accounts, vec!["alice"]); - assert!(!dir.path().join("global_meta.json").exists()); - assert!(dir.path().join("global_meta.bin").exists()); - } - - #[test] - fn test_global_meta_default_when_missing() { - let dir = TempDir::new().unwrap(); - let meta = GlobalMeta::load(dir.path()).unwrap(); - assert!(meta.accounts.is_empty()); - } - - #[test] - fn test_json_migration() { - let dir = TempDir::new().unwrap(); - // Write old JSON format - let json = r#"{"version":1,"accounts":["bob","alice"]}"#; - std::fs::write(dir.path().join("global_meta.json"), json).unwrap(); - - let meta = GlobalMeta::load(dir.path()).unwrap(); - // Should be sorted - assert_eq!(meta.accounts, vec!["alice", "bob"]); - // JSON should be removed - assert!(!dir.path().join("global_meta.json").exists()); - // BIN should exist - assert!(dir.path().join("global_meta.bin").exists()); - } - - #[test] - fn test_account_meta_bin_roundtrip() { - let dir = TempDir::new().unwrap(); - let mut meta = AccountMeta::new("alice".into(), 1); + let mut meta = GlobalMeta::new(); meta.segments.insert( 1, SegmentStats { @@ -219,66 +154,40 @@ mod tests { deleted_bytes: 300, deleted_ratio: 0.3, sealed: false, - indexed_up_to_offset: 0, + indexed_up_to_offset: 500, + bucket_compacted: 0, }, ); meta.save(dir.path()).unwrap(); - let loaded = AccountMeta::load(dir.path()).unwrap(); + let loaded = GlobalMeta::load(dir.path()).unwrap(); assert_eq!(loaded.active_segment_id, 1); assert_eq!(loaded.segments[&1].total_bytes, 1000); + assert_eq!(loaded.segments[&1].indexed_up_to_offset, 500); + } + + #[test] + fn test_global_meta_default_when_missing() { + let dir = TempDir::new().unwrap(); + let meta = GlobalMeta::load(dir.path()).unwrap(); + assert_eq!(meta.active_segment_id, 1); + assert!(meta.segments.is_empty()); } #[test] fn test_corrupt_bin_detected() { let dir = TempDir::new().unwrap(); std::fs::write(dir.path().join("meta.bin"), vec![0xFFu8; 100]).unwrap(); - let result = AccountMeta::load(dir.path()); + let result = GlobalMeta::load(dir.path()); assert!(result.is_err()); - // 0xFFFFFFFF version triggers UnsupportedMetaVersion - assert!(matches!(result.unwrap_err(), crate::error::Error::UnsupportedMetaVersion { .. })); } #[test] - fn test_crc_corruption_detected() { - let dir = TempDir::new().unwrap(); - // Write a well-formed header (version=1) but with wrong CRC bytes - let mut buf = Vec::new(); - buf.extend_from_slice(&0xDEADBEEFu32.to_le_bytes()); // wrong CRC - buf.extend_from_slice(&1u32.to_le_bytes()); // version = 1 (OK) - buf.extend_from_slice(b"some payload bytes"); // payload - std::fs::write(dir.path().join("meta.bin"), &buf).unwrap(); - let result = AccountMeta::load(dir.path()); - assert!(matches!(result.unwrap_err(), crate::error::Error::CorruptMeta(_))); - } - - #[test] - fn test_account_json_migration() { - let dir = TempDir::new().unwrap(); - // Write old JSON format for AccountMeta - let json = r#"{"account_id":"alice","active_segment_id":5,"segments":{}}"#; - std::fs::write(dir.path().join("meta.json"), json).unwrap(); - - let meta = AccountMeta::load(dir.path()).unwrap(); - assert_eq!(meta.account_id, "alice"); - assert_eq!(meta.active_segment_id, 5); - // JSON should be removed - assert!(!dir.path().join("meta.json").exists()); - // BIN should exist - assert!(dir.path().join("meta.bin").exists()); - } - - #[test] - fn test_bin_sorted_keys() { - let dir = TempDir::new().unwrap(); - let mut meta = AccountMeta::new("test".into(), 1); - meta.segments.insert(3, SegmentStats::new(3)); - meta.segments.insert(1, SegmentStats::new(1)); - meta.segments.insert(2, SegmentStats::new(2)); - meta.save(dir.path()).unwrap(); - - let loaded = AccountMeta::load(dir.path()).unwrap(); - let keys: Vec = loaded.segments.keys().copied().collect(); - assert_eq!(keys, vec![1, 2, 3]); + fn test_segment_stats_recompute() { + let mut s = SegmentStats::new(1); + s.total_bytes = 1000; + s.deleted_bytes = 250; + s.recompute_ratio(); + assert!((s.deleted_ratio - 0.25).abs() < 0.001); } } diff --git a/crates/blob/src/recovery.rs b/crates/blob/src/recovery.rs index b72f093..5bd7226 100644 --- a/crates/blob/src/recovery.rs +++ b/crates/blob/src/recovery.rs @@ -2,44 +2,23 @@ use std::collections::HashMap; use std::fs; use std::path::Path; -use crate::bucket::{self, BucketFile, IndexRecord}; +use crate::bucket::{self, IndexRecord}; use crate::error::Result; -use crate::meta::{AccountMeta, SegmentStats}; +use crate::meta::{GlobalMeta, SegmentStats}; use crate::segment::{self, SegmentReader}; -/// Recover an account after a crash: scan segments, repair indices, update stats. -pub fn recover_account(account_dir: &Path) -> Result { - let meta_bin = account_dir.join("meta.bin"); - let meta_json = account_dir.join("meta.json"); - let meta_exists = meta_bin.exists() || meta_json.exists(); - let mut meta = if meta_exists { - AccountMeta::load(account_dir).unwrap_or_else(|_| { - AccountMeta::new( - account_dir - .file_name() - .unwrap_or_default() - .to_string_lossy() - .into(), - 1, - ) - }) - } else { - return Ok(AccountMeta::new( - account_dir - .file_name() - .unwrap_or_default() - .to_string_lossy() - .into(), - 1, - )); - }; - - // Discover all segment files on disk - let seg_dir = account_dir.join("segments"); +/// Recover after a crash: scan any unindexed portions of segments, +/// update bucket files, fix segment stats. +pub fn recover(store_root: &Path, meta: &mut GlobalMeta) -> Result<()> { + let seg_dir = store_root.join("segments"); if !seg_dir.exists() { fs::create_dir_all(&seg_dir)?; } + let buckets_dir = store_root.join("buckets"); + fs::create_dir_all(&buckets_dir)?; + + // Discover all segment files on disk let mut disk_segments: Vec = Vec::new(); if seg_dir.exists() { for entry in fs::read_dir(&seg_dir)? { @@ -58,34 +37,27 @@ pub fn recover_account(account_dir: &Path) -> Result { disk_segments.sort_unstable(); if disk_segments.is_empty() { - meta.active_segment_id = 1; - } else { - let max_id = *disk_segments.last().unwrap(); - meta.active_segment_id = max_id; + return Ok(()); } - // Ensure buckets directory exists - let buckets_dir = account_dir.join("buckets"); - fs::create_dir_all(&buckets_dir)?; - - // For each segment, scan only the unindexed tail and update stats incrementally + // For each segment, scan unindexed portions and append to bucket files for &seg_id in &disk_segments { let seg_path = seg_dir.join(segment::segment_filename(seg_id)); let file_size = fs::metadata(&seg_path)?.len(); - // Preserve existing stats; start fresh if this is a newly discovered segment - let mut stats = meta.segments.remove(&seg_id).unwrap_or_else(|| SegmentStats::new(seg_id)); + let mut stats = meta + .segments + .remove(&seg_id) + .unwrap_or_else(|| SegmentStats::new(seg_id)); let is_sealed = seg_id != meta.active_segment_id; stats.sealed = is_sealed; - // Scan start: from last indexed offset. Clamp defensively. let scan_start = if stats.indexed_up_to_offset <= file_size { stats.indexed_up_to_offset } else { 0 }; - // If fully indexed, skip scanning entirely if scan_start >= file_size { meta.segments.insert(seg_id, stats); continue; @@ -113,10 +85,17 @@ pub fn recover_account(account_dir: &Path) -> Result { Ok(()) })?; - // Merge new records into bucket files (only the newly discovered ones) + // Append new records to bucket files for (bid, records) in &new_records { - let bf = BucketFile::open(account_dir, *bid); - bf.append_batch(records)?; + let bf_path = buckets_dir.join(format!("{:02x}.idx", bid)); + use std::io::Write; + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&bf_path)?; + for r in records { + file.write_all(&r.encode())?; + } } // Truncate if tail corruption found @@ -129,14 +108,14 @@ pub fn recover_account(account_dir: &Path) -> Result { meta.segments.insert(seg_id, stats); } - meta.save(account_dir)?; + meta.save(store_root)?; - Ok(meta) + Ok(()) } /// Clean up leftover temp files from interrupted GC. -pub fn cleanup_temp_files(account_dir: &Path) -> Result<()> { - let seg_dir = account_dir.join("segments"); +pub fn cleanup_temp_files(store_root: &Path) -> Result<()> { + let seg_dir = store_root.join("segments"); if seg_dir.exists() { for entry in fs::read_dir(&seg_dir)? { let entry = entry?; @@ -150,7 +129,7 @@ pub fn cleanup_temp_files(account_dir: &Path) -> Result<()> { } } // Also cleanup temp bucket files - let buckets_dir = account_dir.join("buckets"); + let buckets_dir = store_root.join("buckets"); if buckets_dir.exists() { for entry in fs::read_dir(&buckets_dir)? { let entry = entry?; @@ -165,38 +144,3 @@ pub fn cleanup_temp_files(account_dir: &Path) -> Result<()> { } Ok(()) } - -#[cfg(test)] -mod tests { - use super::*; - use tempfile::TempDir; - - #[test] - fn test_recover_fresh_account() { - let dir = TempDir::new().unwrap(); - let account_dir = dir.path().join("test"); - fs::create_dir_all(&account_dir).unwrap(); - - let meta = recover_account(&account_dir).unwrap(); - assert_eq!(meta.active_segment_id, 1); - assert!(meta.segments.is_empty()); - } - - #[test] - fn test_cleanup_temp_files() { - let dir = TempDir::new().unwrap(); - let account_dir = dir.path().join("test"); - fs::create_dir_all(account_dir.join("segments")).unwrap(); - fs::create_dir_all(account_dir.join("buckets")).unwrap(); - fs::write( - account_dir.join("segments").join("temp_ABC123.seg"), - b"garbage", - ) - .unwrap(); - fs::write(account_dir.join("buckets").join("00.idx.tmp"), b"garbage").unwrap(); - - cleanup_temp_files(&account_dir).unwrap(); - - assert!(!account_dir.join("segments").join("temp_ABC123.seg").exists()); - } -} diff --git a/crates/blob/src/segment.rs b/crates/blob/src/segment.rs index 06906bd..0d9ecb1 100644 --- a/crates/blob/src/segment.rs +++ b/crates/blob/src/segment.rs @@ -1,7 +1,6 @@ use std::fs::{self, File}; use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; use crate::checksum; use crate::error::{Error, Result}; @@ -232,6 +231,14 @@ impl SegmentReader { file.read_exact(&mut data_size_buf)?; let data_size = u32::from_le_bytes(data_size_buf); + if data_size as usize > crate::types::MAX_VALUE_SIZE { + return Err(Error::CorruptEntry { + path: self.path.clone(), + offset, + reason: format!("data_size {} exceeds max {}", data_size, crate::types::MAX_VALUE_SIZE), + }); + } + // Read data let mut data = vec![0u8; data_size as usize]; file.read_exact(&mut data)?; @@ -269,18 +276,18 @@ impl SegmentReader { )) } - /// Read a single entry at the given offset using a pre-opened File (via Mutex). - /// This avoids the per-read File::open cost for hot segments. - pub fn read_entry_at_file(&self, offset: u64, file: &Mutex) -> Result<(Entry, u64)> { + /// Read a single entry at the given offset using a pre-opened File. + /// Uses pread so concurrent reads on the same segment don't block each other. + pub fn read_entry_at_file(&self, offset: u64, file: &File) -> Result<(Entry, u64)> { + // Read header (50 bytes) + let mut header = [0u8; ENTRY_HEADER_SIZE]; + fs_util::pread_exact(file, offset, &mut header)?; - let mut file = file.lock().unwrap(); + let mut pos = 0; - file.seek(SeekFrom::Start(offset))?; - - // Read magic - let mut magic_buf = [0u8; 4]; - file.read_exact(&mut magic_buf)?; - let magic = u32::from_le_bytes(magic_buf); + // Magic + let magic = u32::from_le_bytes(header[pos..pos+4].try_into().unwrap()); + pos += 4; if magic != ENTRY_MAGIC { return Err(Error::CorruptEntry { path: self.path.clone(), @@ -289,41 +296,45 @@ impl SegmentReader { }); } - // Read CRC32 - let mut crc_buf = [0u8; 4]; - file.read_exact(&mut crc_buf)?; - let stored_crc = u32::from_le_bytes(crc_buf); + // CRC32 + let stored_crc = u32::from_le_bytes(header[pos..pos+4].try_into().unwrap()); + pos += 4; - // Read flags, codec - let mut flags_buf = [0u8; 1]; - file.read_exact(&mut flags_buf)?; - let flags = flags_buf[0]; - - let mut codec_buf = [0u8; 1]; - file.read_exact(&mut codec_buf)?; - let codec = Codec::from_u8(codec_buf[0]).ok_or_else(|| Error::CorruptEntry { + // Flags, codec + let flags = header[pos]; + pos += 1; + let codec = Codec::from_u8(header[pos]).ok_or_else(|| Error::CorruptEntry { path: self.path.clone(), offset, - reason: format!("unknown codec: {}", codec_buf[0]), + reason: format!("unknown codec: {}", header[pos]), })?; + pos += 1; - // Read key, raw_size, data_size + // Key let mut key = [0u8; 32]; - file.read_exact(&mut key)?; + key.copy_from_slice(&header[pos..pos+32]); + pos += 32; - let mut raw_size_buf = [0u8; 4]; - file.read_exact(&mut raw_size_buf)?; - let raw_size = u32::from_le_bytes(raw_size_buf); + // Raw size, data size + let raw_size = u32::from_le_bytes(header[pos..pos+4].try_into().unwrap()); + pos += 4; + let data_size = u32::from_le_bytes(header[pos..pos+4].try_into().unwrap()); - let mut data_size_buf = [0u8; 4]; - file.read_exact(&mut data_size_buf)?; - let data_size = u32::from_le_bytes(data_size_buf); + // Defense against header corruption: refuse absurdly large allocations + if data_size as usize > crate::types::MAX_VALUE_SIZE { + return Err(Error::CorruptEntry { + path: self.path.clone(), + offset, + reason: format!("data_size {} exceeds max {}", data_size, crate::types::MAX_VALUE_SIZE), + }); + } // Read data + let data_offset = offset + ENTRY_HEADER_SIZE as u64; let mut data = vec![0u8; data_size as usize]; - file.read_exact(&mut data)?; + fs_util::pread_exact(file, data_offset, &mut data)?; - // Verify CRC32 + // Verify CRC32 (over everything after the crc32 field: flags+codec+key+raw_size+data_size+data) let computed_crc = { let mut hasher = crate::checksum::CrcWriter::new(); hasher.update(&[flags]); diff --git a/crates/blob/src/types.rs b/crates/blob/src/types.rs index 33c885c..9dd1d3e 100644 --- a/crates/blob/src/types.rs +++ b/crates/blob/src/types.rs @@ -12,8 +12,8 @@ pub const INDEX_RECORD_SIZE: usize = 52; /// Maximum segment size (256 MB) pub const SEGMENT_MAX_SIZE: u64 = 256 * 1024 * 1024; -/// Number of hash buckets per account -pub const BUCKET_COUNT: u16 = 16; +/// Number of hash buckets (global) +pub const BUCKET_COUNT: u16 = 256; /// Maximum value size (100 MB) pub const MAX_VALUE_SIZE: usize = 100 * 1024 * 1024; @@ -21,12 +21,12 @@ pub const MAX_VALUE_SIZE: usize = 100 * 1024 * 1024; /// Default compression threshold (4 KB) pub const DEFAULT_COMPRESS_THRESHOLD: usize = 4096; -/// Default LRU bucket cache size -pub const DEFAULT_LRU_BUCKET_COUNT: usize = 256; - /// Default GC deleted ratio threshold pub const DEFAULT_GC_DELETED_RATIO: f64 = 0.30; +/// Default bucket compact threshold (number of pending records before auto-compact) +pub const DEFAULT_COMPACT_THRESHOLD: usize = 10_000; + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum Codec { None = 0, @@ -50,8 +50,12 @@ pub struct Config { pub compress_threshold: usize, pub default_codec: Codec, pub compression_level: i32, - pub lru_bucket_count: usize, + pub compact_threshold: usize, pub gc_deleted_ratio: f64, + /// Interval in seconds for periodic background flush (0 = disabled). + /// When set, a background thread fsyncs the active segment and saves + /// metadata at this interval, bounding recovery time after a crash. + pub flush_interval_secs: u64, } impl Default for Config { @@ -60,17 +64,18 @@ impl Default for Config { compress_threshold: DEFAULT_COMPRESS_THRESHOLD, default_codec: Codec::Zstd, compression_level: 0, - lru_bucket_count: DEFAULT_LRU_BUCKET_COUNT, + compact_threshold: DEFAULT_COMPACT_THRESHOLD, gc_deleted_ratio: DEFAULT_GC_DELETED_RATIO, + flush_interval_secs: 0, } } } impl Config { pub fn validate(&self) -> crate::error::Result<()> { - if self.lru_bucket_count == 0 { + if self.compact_threshold == 0 { return Err(crate::error::Error::InvalidConfig( - "lru_bucket_count must be > 0".into(), + "compact_threshold must be > 0".into(), )); } if self.gc_deleted_ratio <= 0.0 || self.gc_deleted_ratio >= 1.0 { @@ -83,6 +88,11 @@ impl Config { "compression_level must be >= 0".into(), )); } + if self.flush_interval_secs > 0 && self.flush_interval_secs < 5 { + return Err(crate::error::Error::InvalidConfig( + "flush_interval_secs must be 0 (disabled) or >= 5".into(), + )); + } Ok(()) } } diff --git a/crates/blob/tests/acid_test.rs b/crates/blob/tests/acid_test.rs index e04dae1..d9f1108 100644 --- a/crates/blob/tests/acid_test.rs +++ b/crates/blob/tests/acid_test.rs @@ -3,9 +3,6 @@ /// Since we can't kill the process mid-write in an inline test, we simulate crashes /// by dropping the Engine without calling any cleanup (close/drop is the "crash"), /// then re-opening and verifying recovery produced consistent state. -/// -/// For true power-loss simulation, each test writes data, drops the engine abruptly, -/// then reopens and verifies: no corruption, no lost committed data, no partial writes. use std::fs; use std::path::Path; @@ -50,16 +47,13 @@ fn test_durability_single_write_survives_crash() { // Write { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } // <-- Engine dropped = simulated crash // Recover { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value)); } } @@ -73,21 +67,23 @@ fn test_durability_many_writes_survive_crash() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); for i in 0..n { let key = make_key(i as u64); keys.push(key); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } } // crash { let engine = Engine::open(dir.path(), Config::default()).unwrap(); for (i, key) in keys.iter().enumerate() { - let result = engine.read("alice", key).unwrap(); - assert_eq!(result, Some(value.clone()), "missing key at index {}", i); + let result = engine.get(key).unwrap(); + assert_eq!( + result, + Some(value.clone()), + "missing key at index {}", + i + ); } } } @@ -100,20 +96,17 @@ fn test_durability_delete_survives_crash() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } // crash after write { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.delete("alice", &key).unwrap(); + engine.delete(&key).unwrap(); } // crash after delete { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, None, "delete should persist across crash"); } } @@ -129,22 +122,20 @@ fn test_atomicity_no_partial_entries_after_crash() { // Write enough entries to fill part of a segment, then crash { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - let value = make_value(50_000); // big enough to notice + let value = make_value(50_000); for i in 0..200u64 { engine - .write("alice", make_key(i), &value, Codec::None) + .put(make_key(i), &value, Codec::None) .unwrap(); } } // crash - // Recovery should clean up any partial tail entries and all committed - // entries should be readable + // Recovery should clean up any partial tail entries { let engine = Engine::open(dir.path(), Config::default()).unwrap(); let value = make_value(50_000); for i in 0..200u64 { - let result = engine.read("alice", &make_key(i)).unwrap(); + let result = engine.get(&make_key(i)).unwrap(); assert_eq!( result, Some(value.clone()), @@ -162,20 +153,18 @@ fn test_atomicity_crash_during_segment_roll() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); // Write enough to cross at least one segment boundary (256 MB) for i in 0..140u64 { engine - .write("alice", make_key(i), &big_value, Codec::None) + .put(make_key(i), &big_value, Codec::None) .unwrap(); } } // crash mid-way or after multiple segments { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - // All committed writes (that returned Ok) must be readable for i in 0..140u64 { - let result = engine.read("alice", &make_key(i)).unwrap(); + let result = engine.get(&make_key(i)).unwrap(); assert!( result.is_some(), "key {} should exist after segment roll recovery", @@ -197,16 +186,12 @@ fn test_consistency_crc_detects_corruption() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } // Corrupt the segment file by flipping a byte - let seg_path = find_first_segment(dir.path(), "alice"); + let seg_path = find_first_segment(dir.path()); let mut data = fs::read(&seg_path).unwrap(); - // Flip a byte in the data portion, not the header let flip_pos = data.len() - 100; data[flip_pos] ^= 0xFF; fs::write(&seg_path, &data).unwrap(); @@ -214,16 +199,14 @@ fn test_consistency_crc_detects_corruption() { // Reading should detect CRC mismatch { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - let result = engine.read("alice", &key); - // Either error or None is acceptable — never silently wrong data + let result = engine.get(&key); match result { - Err(_) => {} // CRC mismatch detected — good + Err(_) => {} // CRC mismatch detected - good Ok(None) => {} // index may point to truncated/removed data Ok(Some(v)) => { if v == value { - panic!("CRC corruption was NOT detected — silent data corruption!"); + panic!("CRC corruption was NOT detected - silent data corruption!"); } - // If value differs, index pointed elsewhere after recovery } } } @@ -235,35 +218,39 @@ fn test_consistency_corrupt_magic_truncated_on_recovery() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); for i in 0..10u64 { engine - .write("alice", make_key(i), &make_value(4096), Codec::Zstd) + .put(make_key(i), &make_value(4096), Codec::Zstd) .unwrap(); } } // Append garbage to the segment file (simulating partial write from crash) - let seg_path = find_first_segment(dir.path(), "alice"); + let seg_path = find_first_segment(dir.path()); let mut data = fs::read(&seg_path).unwrap(); let orig_len = data.len(); - // Append garbage that doesn't start with the magic number data.extend_from_slice(&[0xFF; 200]); fs::write(&seg_path, &data).unwrap(); // Recovery should truncate the garbage { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - // Verify committed data is still intact for i in 0..10u64 { - let result = engine.read("alice", &make_key(i)).unwrap(); - assert!(result.is_some(), "committed key {} should survive tail truncation", i); + let result = engine.get(&make_key(i)).unwrap(); + assert!( + result.is_some(), + "committed key {} should survive tail truncation", + i + ); } } // Verify file was actually truncated let truncated_len = fs::metadata(&seg_path).unwrap().len(); - assert!(truncated_len <= orig_len as u64, "garbage should have been truncated"); + assert!( + truncated_len <= orig_len as u64, + "garbage should have been truncated" + ); } // --------------------------------------------------------------------------- @@ -274,22 +261,17 @@ fn test_consistency_corrupt_magic_truncated_on_recovery() { fn test_isolation_reader_sees_snapshot_not_partial_write() { let dir = TempDir::new().unwrap(); let engine = Arc::new(Engine::open(dir.path(), Config::default()).unwrap()); - engine.create_account("alice").unwrap(); // Pre-populate a known key let original_value = make_value(4096); let key = make_key(100); - engine - .write("alice", key, &original_value, Codec::Zstd) - .unwrap(); + engine.put(key, &original_value, Codec::Zstd).unwrap(); let running = Arc::new(AtomicBool::new(true)); - let writer_done = Arc::new(AtomicBool::new(false)); // Spawn a writer that continuously overwrites the same key let writer_engine = engine.clone(); let writer_running = running.clone(); - let writer_done_flag = writer_done.clone(); let writer_key = key; let writer = thread::spawn(move || { @@ -299,11 +281,10 @@ fn test_isolation_reader_sees_snapshot_not_partial_write() { } let val = make_value(4096 + (i as usize % 100)); writer_engine - .write("alice", writer_key, &val, Codec::Zstd) + .put(writer_key, &val, Codec::Zstd) .unwrap(); thread::yield_now(); } - writer_done_flag.store(true, Ordering::SeqCst); }); // Concurrent reader: reads should never panic or hang @@ -315,11 +296,10 @@ fn test_isolation_reader_sees_snapshot_not_partial_write() { if !reader_running.load(Ordering::Relaxed) && reads > 0 { break; } - let result = reader_engine.read("alice", &key); + let result = reader_engine.get(&key); match result { - Ok(Some(_)) | Ok(None) => {} // OK + Ok(Some(_)) | Ok(None) => {} Err(e) => { - // Accept transient errors but report them eprintln!("reader saw error: {:?}", e); } } @@ -332,8 +312,7 @@ fn test_isolation_reader_sees_snapshot_not_partial_write() { running.store(false, Ordering::SeqCst); writer.join().unwrap(); - // Final read should see the last committed value (not partial) - let final_result = engine.read("alice", &key).unwrap(); + let final_result = engine.get(&key).unwrap(); assert!(final_result.is_some(), "final read should find a value"); } @@ -347,21 +326,18 @@ fn test_crash_during_gc_leaves_data_intact() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - let value = make_value(500_000); // 500 KB each - // Write enough entries and delete some to create GC candidate + let value = make_value(500_000); for i in 0..500u64 { engine - .write("alice", make_key(i), &value, Codec::None) + .put(make_key(i), &value, Codec::None) .unwrap(); } // Delete ~40% for i in (0..500u64).step_by(5) { - engine.delete("alice", &make_key(i)).unwrap(); + engine.delete(&make_key(i)).unwrap(); } - // Single GC run (may or may not trigger) - let _ = engine.gc("alice"); + let _ = engine.gc(); } // crash after GC // All non-deleted entries must still be readable @@ -370,9 +346,8 @@ fn test_crash_during_gc_leaves_data_intact() { let value = make_value(500_000); for i in 0..500u64 { let key = make_key(i); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); if i % 5 == 0 { - // Deleted keys assert_eq!(result, None, "key {} should be deleted", i); } else { assert_eq!( @@ -401,11 +376,8 @@ fn test_multiple_crash_reopen_cycles() { // Populate and crash { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); for i in 0..50u64 { - engine - .write("alice", make_key(i), &value, Codec::Zstd) - .unwrap(); + engine.put(make_key(i), &value, Codec::Zstd).unwrap(); alive.insert(i); } } @@ -414,11 +386,11 @@ fn test_multiple_crash_reopen_cycles() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); for &k in &alive { - assert!(engine.read("alice", &make_key(k)).unwrap().is_some()); + assert!(engine.get(&make_key(k)).unwrap().is_some()); } for i in 100..150u64 { engine - .write("alice", make_key(i), &value, Codec::Zstd) + .put(make_key(i), &value, Codec::Zstd) .unwrap(); alive.insert(i); } @@ -428,10 +400,10 @@ fn test_multiple_crash_reopen_cycles() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); for &k in &alive { - assert!(engine.read("alice", &make_key(k)).unwrap().is_some()); + assert!(engine.get(&make_key(k)).unwrap().is_some()); } for i in 0..10u64 { - engine.delete("alice", &make_key(i)).unwrap(); + engine.delete(&make_key(i)).unwrap(); alive.remove(&i); } } @@ -440,44 +412,44 @@ fn test_multiple_crash_reopen_cycles() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); for &k in &alive { - assert!(engine.read("alice", &make_key(k)).unwrap().is_some(), - "key {} should exist", k); + assert!( + engine.get(&make_key(k)).unwrap().is_some(), + "key {} should exist", + k + ); } for i in 0..10u64 { - assert_eq!(engine.read("alice", &make_key(i)).unwrap(), None, - "key {} should be deleted", i); + assert_eq!( + engine.get(&make_key(i)).unwrap(), + None, + "key {} should be deleted", + i + ); } } } // --------------------------------------------------------------------------- -// 7. Account-level isolation +// 7. Global dedup: same content stored once // --------------------------------------------------------------------------- #[test] -fn test_account_isolation_crash_one_account_does_not_affect_others() { +fn test_global_dedup_after_crash() { let dir = TempDir::new().unwrap(); + let key = make_key(42); + let value = make_value(8192); { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine.create_account("bob").unwrap(); - - engine - .write("alice", make_key(1), &make_value(4096), Codec::Zstd) - .unwrap(); - engine - .write("bob", make_key(1), &make_value(8192), Codec::Zstd) - .unwrap(); + // Write same key twice (simulating two sources with same content) + engine.put(key, &value, Codec::Zstd).unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } - // Delete alice's account dir partially to simulate corruption - // Then verify bob is intact { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - // Bob should be fine - let result = engine.read("bob", &make_key(1)).unwrap(); - assert!(result.is_some(), "bob should be unaffected"); + let result = engine.get(&key).unwrap(); + assert_eq!(result, Some(value)); } } @@ -485,8 +457,8 @@ fn test_account_isolation_crash_one_account_does_not_affect_others() { // Helpers // --------------------------------------------------------------------------- -fn find_first_segment(store_root: &Path, account: &str) -> std::path::PathBuf { - let seg_dir = store_root.join("accounts").join(account).join("segments"); +fn find_first_segment(store_root: &Path) -> std::path::PathBuf { + let seg_dir = store_root.join("segments"); for entry in fs::read_dir(&seg_dir).unwrap() { let entry = entry.unwrap(); let name = entry.file_name().to_string_lossy().into_owned(); diff --git a/crates/blob/tests/integration_test.rs b/crates/blob/tests/integration_test.rs index de634f1..f16971d 100644 --- a/crates/blob/tests/integration_test.rs +++ b/crates/blob/tests/integration_test.rs @@ -1,33 +1,17 @@ use bichon_blob::{Codec, Config, Engine}; use tempfile::TempDir; -#[test] -fn test_create_and_list_accounts() { - let dir = TempDir::new().unwrap(); - let engine = Engine::open(dir.path(), Config::default()).unwrap(); - - engine.create_account("alice").unwrap(); - engine.create_account("bob").unwrap(); - - let accounts = engine.list_accounts(); - assert!(accounts.contains(&"alice".to_string())); - assert!(accounts.contains(&"bob".to_string())); -} - #[test] fn test_write_and_read() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0xAA; 32]; let value = b"Hello, this is a test email!".to_vec(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value)); } @@ -35,10 +19,9 @@ fn test_write_and_read() { fn test_read_missing_key() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0xFF; 32]; - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, None); } @@ -46,45 +29,43 @@ fn test_read_missing_key() { fn test_delete() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0xBB; 32]; let value = b"Some email content".to_vec(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); - engine.delete("alice", &key).unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); + engine.delete(&key).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, None); } #[test] -fn test_delete_account() { +fn test_exists() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine.delete_account("alice").unwrap(); - let accounts = engine.list_accounts(); - assert!(!accounts.contains(&"alice".to_string())); + let key = [0xCC; 32]; + assert!(!engine.exists(&key).unwrap()); + + engine.put(key, b"data", Codec::None).unwrap(); + assert!(engine.exists(&key).unwrap()); + + engine.delete(&key).unwrap(); + assert!(!engine.exists(&key).unwrap()); } #[test] fn test_small_value_not_compressed() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0xCC; 32]; let value = b"hi"; // Smaller than 4KB threshold - engine - .write("alice", key, value, Codec::Zstd) - .unwrap(); + engine.put(key, value, Codec::Zstd).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value.to_vec())); } @@ -92,16 +73,13 @@ fn test_small_value_not_compressed() { fn test_large_value() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0xDD; 32]; let value = vec![b'X'; 100_000]; // 100KB - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value)); } @@ -109,22 +87,19 @@ fn test_large_value() { fn test_multiple_keys() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let n = 100; for i in 0..n { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); let value = format!("email number {}", i).into_bytes(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } for i in 0..n { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(format!("email number {}", i).into_bytes())); } } @@ -133,7 +108,6 @@ fn test_multiple_keys() { fn test_gc() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); // Write many entries let value = vec![b'Y'; 5000]; @@ -142,26 +116,24 @@ fn test_gc() { for i in 0..n { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - engine - .write("alice", key, &value, Codec::None) - .unwrap(); + engine.put(key, &value, Codec::None).unwrap(); } // Delete even-numbered keys for i in (0..n).step_by(2) { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - engine.delete("alice", &key).unwrap(); + engine.delete(&key).unwrap(); } // Run GC - let _result = engine.gc("alice").unwrap(); + let _result = engine.gc().unwrap(); // Verify remaining keys still readable for i in (1..n).step_by(2) { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value.clone())); } @@ -169,7 +141,7 @@ fn test_gc() { for i in (0..n).step_by(2) { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, None); } } @@ -182,16 +154,13 @@ fn test_reopen_persistence() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine - .write("alice", key, &value, Codec::Zstd) - .unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); } // Reopen { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - let result = engine.read("alice", &key).unwrap(); + let result = engine.get(&key).unwrap(); assert_eq!(result, Some(value)); } } @@ -200,21 +169,18 @@ fn test_reopen_persistence() { fn test_stats() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine - .write("alice", [1u8; 32], b"hello", Codec::None) - .unwrap(); + engine.put([1u8; 32], b"hello", Codec::None).unwrap(); - let stats = engine.stats("alice").unwrap(); + let stats = engine.stats().unwrap(); assert!(stats.total_bytes > 0); + assert!(stats.total_keys > 0); } #[test] fn test_batch_write() { let dir = TempDir::new().unwrap(); let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let n = 50; let entries: Vec<_> = (0..n) @@ -226,10 +192,10 @@ fn test_batch_write() { }) .collect(); - engine.write_batch("alice", &entries).unwrap(); + engine.put_batch(&entries).unwrap(); for (key, value, _) in &entries { - let result = engine.read("alice", key).unwrap(); + let result = engine.get(key).unwrap(); assert_eq!(result.as_ref(), Some(value)); } } @@ -247,14 +213,13 @@ fn test_batch_write_persistence() { { let engine = Engine::open(dir.path(), Config::default()).unwrap(); - engine.create_account("alice").unwrap(); - engine.write_batch("alice", &entries).unwrap(); + engine.put_batch(&entries).unwrap(); } { let engine = Engine::open(dir.path(), Config::default()).unwrap(); for (key, value, _) in &entries { - let result = engine.read("alice", key).unwrap(); + let result = engine.get(key).unwrap(); assert_eq!(result.as_ref(), Some(value)); } } @@ -264,7 +229,7 @@ fn test_batch_write_persistence() { fn test_invalid_config_rejected() { let dir = TempDir::new().unwrap(); let mut config = Config::default(); - config.lru_bucket_count = 0; + config.compact_threshold = 0; assert!(Engine::open(dir.path(), config).is_err()); let mut config = Config::default(); @@ -279,13 +244,14 @@ fn test_concurrent_reads() { let dir = TempDir::new().unwrap(); let engine = Arc::new(Engine::open(dir.path(), Config::default()).unwrap()); - engine.create_account("alice").unwrap(); // Write some data for i in 0..50u32 { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&i.to_le_bytes()); - engine.write("alice", key, &vec![i as u8; 1024], Codec::None).unwrap(); + engine + .put(key, &vec![i as u8; 1024], Codec::None) + .unwrap(); } // Spawn 4 threads, each reading a different subset @@ -296,7 +262,7 @@ fn test_concurrent_reads() { for i in (t * 12)..((t + 1) * 12) { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&(i as u32).to_le_bytes()); - let read = engine.read("alice", &key).unwrap(); + let read = engine.get(&key).unwrap(); assert!(read.is_some(), "key {} should exist", i); } })); @@ -307,43 +273,25 @@ fn test_concurrent_reads() { } #[test] -fn test_concurrent_writes_different_accounts() { - use std::sync::Arc; - use std::thread; - +fn test_global_dedup() { let dir = TempDir::new().unwrap(); - let engine = Arc::new(Engine::open(dir.path(), Config::default()).unwrap()); + let engine = Engine::open(dir.path(), Config::default()).unwrap(); - for name in &["alice", "bob", "carol"] { - engine.create_account(name).unwrap(); - } + let key = [0x42; 32]; + let value = b"same content across what would be accounts".to_vec(); - let mut handles = vec![]; - for (t, name) in ["alice", "bob", "carol"].iter().enumerate() { - let engine = engine.clone(); - let account_name = name.to_string(); - handles.push(thread::spawn(move || { - for i in 0..20 { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&((t * 100 + i) as u32).to_le_bytes()); - let value = vec![(t * 100 + i) as u8; 512]; - engine.write(&account_name, key, &value, Codec::None).unwrap(); - } - })); - } - for h in handles { - h.join().unwrap(); - } + // Write same key twice (simulating two accounts ingesting the same email) + engine.put(key, &value, Codec::Zstd).unwrap(); + engine.put(key, &value, Codec::Zstd).unwrap(); - // Verify all writes persisted - for (t, name) in ["alice", "bob", "carol"].iter().enumerate() { - for i in 0..20 { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&((t * 100 + i) as u32).to_le_bytes()); - let read = engine.read(name, &key).unwrap(); - assert!(read.is_some(), "account {} key {} should exist", name, i); - } - } + // Should still be readable + let result = engine.get(&key).unwrap(); + assert_eq!(result, Some(value)); + + // Stats should reflect dedup (not double count) + let stats = engine.stats().unwrap(); + // The key appears once in the bucket store + assert!(stats.total_keys > 0); } #[test] @@ -354,144 +302,59 @@ fn test_crash_recovery() { // Phase 1: write data, then drop without shutdown (simulates crash) { let engine = Engine::open(&dir_path, Config::default()).unwrap(); - engine.create_account("alice").unwrap(); for i in 0..50u32 { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&i.to_le_bytes()); - engine.write("alice", key, &vec![i as u8; 512], Codec::None).unwrap(); + engine + .put(key, &vec![i as u8; 512], Codec::None) + .unwrap(); } // Engine dropped here without calling shutdown() } - // Phase 2: reopen — recovery should run, data should be intact + // Phase 2: reopen - recovery should run, data should be intact let engine = Engine::open(&dir_path, Config::default()).unwrap(); - let stats = engine.stats("alice").unwrap(); + let stats = engine.stats().unwrap(); assert!(stats.total_keys > 0, "recovery should preserve data"); // Verify reads work for i in 0..50u32 { let mut key = [0u8; 32]; key[0..4].copy_from_slice(&i.to_le_bytes()); - let read = engine.read("alice", &key).unwrap(); + let read = engine.get(&key).unwrap(); assert!(read.is_some(), "key {} should survive crash recovery", i); } } #[test] fn test_meta_bin_durability() { - // Verify meta.bin has valid CRC and can be read after a write cycle. let dir = TempDir::new().unwrap(); let dir_path = dir.path().to_path_buf(); { let engine = Engine::open(&dir_path, Config::default()).unwrap(); - engine.create_account("alice").unwrap(); let key = [0x42u8; 32]; - engine.write("alice", key, b"durable", Codec::None).unwrap(); + engine.put(key, b"durable", Codec::None).unwrap(); } - // Engine dropped → shutdown() called → meta saved via write_bin (with fsync) + // Engine dropped -> shutdown() called -> meta saved - // Verify meta.bin exists and has valid CRC - let meta_path = dir_path - .join("accounts") - .join("alice") - .join("meta.bin"); + // Verify meta.bin exists + let meta_path = dir_path.join("meta.bin"); assert!(meta_path.exists(), "meta.bin should exist after clean shutdown"); let data = std::fs::read(&meta_path).unwrap(); - assert!(data.len() >= 8, "meta.bin should have at least 8 bytes (crc + version)"); + assert!( + data.len() >= 8, + "meta.bin should have at least 8 bytes" + ); let stored_crc = u32::from_le_bytes(data[0..4].try_into().unwrap()); assert_ne!(stored_crc, 0, "stored CRC should be non-zero"); // Reopen and verify data is intact let engine = Engine::open(&dir_path, Config::default()).unwrap(); - let read = engine.read("alice", &[0x42u8; 32]).unwrap(); + let read = engine.get(&[0x42u8; 32]).unwrap(); assert_eq!(read, Some(b"durable".to_vec())); } - -#[test] -fn test_gc_concurrent_with_writes() { - // GC should not lose entries that are written concurrently. - use std::sync::Arc; - use std::sync::atomic::{AtomicBool, Ordering}; - use std::thread; - - let dir = TempDir::new().unwrap(); - let engine = Arc::new(Engine::open(dir.path(), Config::default()).unwrap()); - engine.create_account("alice").unwrap(); - - // Pre-fill: write enough to trigger eventual GC - let big_value = vec![b'X'; 8192]; - for i in 0..500u32 { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&i.to_le_bytes()); - engine.write("alice", key, &big_value, Codec::None).unwrap(); - } - - // Delete some to create GC candidates - for i in 0..250u32 { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&i.to_le_bytes()); - engine.delete("alice", &key).unwrap(); - } - - let running = Arc::new(AtomicBool::new(true)); - let engine_gc = engine.clone(); - let running_gc = running.clone(); - - // Thread 1: run GC in a loop - let gc_handle = thread::spawn(move || { - while running_gc.load(Ordering::Relaxed) { - let _ = engine_gc.gc("alice"); - thread::sleep(std::time::Duration::from_millis(10)); - } - }); - - // Thread 2: keep writing new entries - let engine_write = engine.clone(); - let running_write = running.clone(); - let write_handle = thread::spawn(move || { - let mut counter = 10000u32; - while running_write.load(Ordering::Relaxed) { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&counter.to_le_bytes()); - engine_write - .write("alice", key, &vec![counter as u8; 256], Codec::None) - .unwrap(); - counter += 1; - } - counter - }); - - // Let them race for a bit - thread::sleep(std::time::Duration::from_millis(500)); - running.store(false, Ordering::Relaxed); - - gc_handle.join().unwrap(); - let final_counter = write_handle.join().unwrap(); - - // All written entries must be readable - let mut missing = 0; - for i in 0..500u32 { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&i.to_le_bytes()); - if engine.read("alice", &key).unwrap().is_none() { - // Entries 0..250 were deleted, they should be gone - if i >= 250 { - missing += 1; - } - } - } - assert_eq!(missing, 0, "pre-existing entries should survive concurrent GC"); - - // Entries written during the race should be readable - for i in 10000..final_counter { - let mut key = [0u8; 32]; - key[0..4].copy_from_slice(&i.to_le_bytes()); - let read = engine.read("alice", &key).unwrap(); - assert!(read.is_some(), "concurrently written key {} should exist after GC", i); - } -}