From 1a2780c83e8567c591c4768575dc225cc4f6bb31 Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Wed, 29 Jul 2026 16:37:03 +0900 Subject: [PATCH 1/6] feat: add sorted fresh-store bulk loading --- DESIGN.md | 10 + README.md | 20 ++ crates/fluent31/src/compaction.rs | 26 ++- crates/fluent31/src/db.rs | 229 +++++++++++++++++++++-- crates/fluent31/src/journal.rs | 72 ++++--- crates/fluent31/tests/bulk_load.rs | 195 +++++++++++++++++++ crates/fluent31/tests/journal_rebuild.rs | 34 ++++ 7 files changed, 546 insertions(+), 40 deletions(-) create mode 100644 crates/fluent31/tests/bulk_load.rs diff --git a/DESIGN.md b/DESIGN.md index 390281a..e0054bb 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -72,6 +72,16 @@ directory entry is fsynced before any write to it is acknowledged. Writers stall (100ms poll on the progress signal) when frozen memtables pile past `max_immutable_memtables` or L0 exceeds `l0_stall_trigger`. +**Fresh-store bulk bootstrap.** `Db::create_from_sorted` is the one path that +does not use the write pipeline above. Before background threads or a public +handle exist, it validates a strictly increasing stream of unique user keys, +applies the same inline/vlog placement, and writes fragmented tables directly +into one bottom-level run. Vlog payloads and tables are synced before a single +manifest flip publishes the base and its flush watermark. A failure before +that flip leaves the previously published empty store, never a partial base. +The API refuses any destination that already contains a database or unrelated +files; it cannot bypass MVCC, triggers, or replication on a live store. + ## 3. Tables (sorted-run fragments) `[data block]* [filter block] [index block] [stats block] [footer 48B]`, diff --git a/README.md b/README.md index 3e53d69..84cd4da 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,26 @@ let a = db.fork_at("replica-a", s)?; let b = db.fork_at("replica-b", s)?; // same cut as replica-a ``` +## Bulk bootstrap + +When the complete initial keyspace is already available in strictly sorted, +unique-key order, `Db::create_from_sorted` streams it directly into a fresh +store's bottom-level run: + +```rust +let entries = vec![(b"acct/1", b"100"), (b"acct/2", b"250")]; +let db = Db::create_from_sorted("./import", Options::default(), entries)?; +``` + +The builder preserves the configured compression, Bloom filters, value +separation, vlog rotation, and SST fragment sizing. It makes the whole base +visible with one manifest update and does not write it through the WAL, +memtable, L0, or compaction. This is a creation primitive, not a live-ingest +path: the destination must be absent or empty, keys must be valid user keys, +and duplicates or out-of-order input are rejected. Fallible source iterators +can use `Db::create_from_sorted_fallible`; an input error never publishes a +partial base. + ## What MVCC is (and isn't) for MVCC is how the engine gives you consistency — it is **not** an diff --git a/crates/fluent31/src/compaction.rs b/crates/fluent31/src/compaction.rs index c65656c..6d44427 100644 --- a/crates/fluent31/src/compaction.rs +++ b/crates/fluent31/src/compaction.rs @@ -18,6 +18,7 @@ use std::sync::atomic::Ordering; use std::sync::Arc; use crate::batch::BatchOp; +use crate::config::Options; use crate::db::{DbInner, RetiredVlog}; use crate::error::Result; use crate::iter::{InternalIterator, MergeIterator}; @@ -65,14 +66,29 @@ const MAX_DYNAMIC_LEVELS: usize = 16; /// Per-level byte budget for deepening decisions: the volume of one /// L0->L1 merge is the unit; each level down multiplies by `tier_width`. -fn level_target_bytes(db: &DbInner, level: usize) -> u64 { - let unit = (db.opts.memtable_size as u64) - .saturating_mul(db.opts.l0_compaction_trigger as u64) +fn level_target_bytes(opts: &Options, level: usize) -> u64 { + let unit = (opts.memtable_size as u64) + .saturating_mul(opts.l0_compaction_trigger as u64) .max(1); - let width = db.opts.tier_width.max(2) as u64; + let width = opts.tier_width.max(2) as u64; (0..level).fold(unit, |acc, _| acc.saturating_mul(width)) } +/// Return a level count whose bottom budget can hold `bottom_bytes` without +/// immediately deepening the tree. Fresh-store bulk loading uses this to +/// install its only run directly at a stable bottom level. +pub(crate) fn level_count_for_bottom_bytes( + opts: &Options, + current: usize, + bottom_bytes: u64, +) -> usize { + let mut levels = current.max(1); + while levels < MAX_DYNAMIC_LEVELS && bottom_bytes > level_target_bytes(opts, levels - 1) { + levels += 1; + } + levels +} + /// One pass of the maintenance loop; returns whether any work happened. pub(crate) fn maintenance_pass(db: &Arc) -> Result { let mut did = false; @@ -136,7 +152,7 @@ fn pick(db: &Arc, force: bool) -> Option { if !force && v.levels.len() < MAX_DYNAMIC_LEVELS && !v.levels[last].is_empty() - && bottom_bytes > level_target_bytes(db, last) + && bottom_bytes > level_target_bytes(&db.opts, last) { return Some(Job { level: last, diff --git a/crates/fluent31/src/db.rs b/crates/fluent31/src/db.rs index 2f258ff..dd9dd7c 100644 --- a/crates/fluent31/src/db.rs +++ b/crates/fluent31/src/db.rs @@ -415,18 +415,7 @@ impl DbInner { BatchOp::Put { key, value } => (key, value.len()), BatchOp::Delete { key } => (key, 0), }; - validate_user_key(key)?; - if key.len() > self.opts.max_key_size { - return Err(Error::InvalidArgument(format!( - "key of {} bytes exceeds max_key_size", - key.len() - ))); - } - if vlen > self.opts.max_value_size { - return Err(Error::InvalidArgument(format!( - "value of {vlen} bytes exceeds max_value_size" - ))); - } + validate_user_entry(&self.opts, key, vlen)?; } Ok(()) } @@ -1161,6 +1150,139 @@ impl DbInner { )) } + /// Build and atomically install the initial state of a fresh database. + /// No background thread or public handle exists while this runs. + fn load_sorted(&self, entries: I) -> Result<()> + where + I: IntoIterator>, + K: AsRef<[u8]>, + V: AsRef<[u8]>, + { + let _ws = self.write_mu.lock(); + { + let state = self.state.read(); + let occupied = self.visible_seqno.load(Ordering::Acquire) != 0 + || !state.mem.is_empty() + || !state.imms.is_empty() + || state.version.levels.iter().any(|level| !level.is_empty()); + if occupied { + return Err(Error::InvalidArgument( + "sorted bulk loading requires a fresh database".into(), + )); + } + } + + let mut tables = Vec::new(); + let mut builder: Option<(u64, TableBuilder)> = None; + let mut previous_key = Vec::new(); + let mut count = 0u64; + + for entry in entries { + let (key, value) = entry?; + let key = key.as_ref(); + let value = value.as_ref(); + validate_user_entry(&self.opts, key, value.len())?; + if !previous_key.is_empty() && key <= previous_key.as_slice() { + return Err(Error::InvalidArgument( + "bulk-load keys must be strictly increasing".into(), + )); + } + let seq = count + .checked_add(1) + .filter(|seq| *seq < MAX_SEQNO) + .ok_or_else(|| Error::InvalidArgument("seqno space exhausted".into()))?; + + if builder + .as_ref() + .is_some_and(|(_, b)| b.estimated_size() >= self.opts.target_file_size) + { + let (id, completed) = builder.take().unwrap(); + tables.push(self.finish_table(id, completed)?); + } + if builder.is_none() { + let id = self.alloc_file_id(); + let file = self.io.create_new(&self.paths.table(id))?; + builder = Some(( + id, + TableBuilder::new( + file, + self.opts.block_size, + self.opts.bloom_bits_per_key, + self.opts.compression, + ), + )); + } + + let repr = if value.len() >= self.opts.value_threshold { + let ptr = self.vlog.append(key, value)?; + let (_, written, _) = self.vlog.head_state(); + if written >= self.opts.vlog_file_size { + self.rotate_vlog_locked()?; + } + encode_ptr(ptr) + } else { + encode_inline(value) + }; + let ikey = make_ikey(key, seq, ValueKind::Put); + builder.as_mut().unwrap().1.add(&ikey, &repr)?; + previous_key.clear(); + previous_key.extend_from_slice(key); + count = seq; + } + + if let Some((id, completed)) = builder.take() { + tables.push(self.finish_table(id, completed)?); + } + if tables.is_empty() { + return Ok(()); + } + + // Every pointer payload and table must be durable before the + // manifest can make the base visible. TableBuilder::finish syncs + // each table; sync the current vlog head and their directory entries. + self.vlog.sync_head()?; + io::sync_dir(&self.paths.dir)?; + + let run = Run { + id: self.alloc_file_id(), + tables, + }; + let run_meta = RunMeta { + id: run.id, + table_ids: run.tables.iter().map(|table| table.id).collect(), + }; + let level_count = crate::compaction::level_count_for_bottom_bytes( + &self.opts, + self.state.read().version.levels.len(), + run.size(), + ); + let bottom = level_count - 1; + + let mut manifest = self.manifest.lock(); + let mut data = manifest.data.clone(); + data.levels.resize(level_count, Vec::new()); + data.levels[bottom].push(run_meta); + data.last_flushed_seqno = count; + { + let state = self.state.read(); + data.vlog_live = state.version.vlogs.keys().copied().collect(); + data.vlog_head = state.version.vlog_head_id; + } + data.next_file_id = self.next_file_id.load(Ordering::SeqCst); + let gen = manifest.gen + 1; + manifest::save(&self.paths, gen, &data)?; + manifest.gen = gen; + manifest.data = data; + + let mut state = self.state.write(); + let mut version = state.version.clone_shape(); + version.levels.resize(level_count, Vec::new()); + version.levels[bottom].push(run); + state.version = Arc::new(version); + self.visible_seqno.store(count, Ordering::Release); + Ok(()) + } + pub(crate) fn finish_table(&self, id: u64, b: TableBuilder) -> Result> { let (_stats, size) = b.finish()?; let path = self.paths.table(id); @@ -1170,6 +1292,22 @@ impl DbInner { } } +pub(crate) fn validate_user_entry(opts: &Options, key: &[u8], value_len: usize) -> Result<()> { + validate_user_key(key)?; + if key.len() > opts.max_key_size { + return Err(Error::InvalidArgument(format!( + "key of {} bytes exceeds max_key_size", + key.len() + ))); + } + if value_len > opts.max_value_size { + return Err(Error::InvalidArgument(format!( + "value of {value_len} bytes exceeds max_value_size" + ))); + } + Ok(()) +} + fn parse_file_id(name: &str, prefix: &str, suffix: &str) -> Option { name.strip_prefix(prefix)? .strip_suffix(suffix)? @@ -1235,6 +1373,43 @@ impl Db { Self::spawn_from_inner(open_inner(dir.as_ref(), opts)?) } + /// Create a fresh database from strictly increasing, unique user keys. + /// + /// The input is streamed directly into final-level table files while + /// preserving the configured compression, Bloom filters, value + /// separation, and file-size targets. The completed base becomes visible + /// through one manifest update; it never enters the WAL, memtable, or L0. + /// The destination must be absent or an empty directory. + pub fn create_from_sorted( + dir: impl AsRef, + opts: Options, + entries: I, + ) -> Result + where + I: IntoIterator, + K: AsRef<[u8]>, + V: AsRef<[u8]>, + { + Self::create_from_sorted_fallible(dir, opts, entries.into_iter().map(Ok)) + } + + /// Fallible-input form of [`Db::create_from_sorted`]. An input error is + /// returned without publishing any partial base. + pub fn create_from_sorted_fallible( + dir: impl AsRef, + opts: Options, + entries: I, + ) -> Result + where + I: IntoIterator>, + K: AsRef<[u8]>, + V: AsRef<[u8]>, + { + let inner = open_fresh_inner(dir.as_ref(), opts)?; + inner.load_sorted(entries)?; + Self::spawn_from_inner(inner) + } + /// Open with a caller-supplied IO backend, bypassing `opts.io_backend`. /// The **fault-injection test seam** (feature `fault-injection`, off by /// default): a custom `Io` can fail/short/corrupt `append`, `read_at`, and @@ -1924,7 +2099,11 @@ fn compact_thread(db: Arc) { // --------------------------------------------------------------------------- fn open_inner(dir: &Path, opts: Options) -> Result> { - open_inner_with(dir, opts, None) + open_inner_with_mode(dir, opts, None, false) +} + +fn open_fresh_inner(dir: &Path, opts: Options) -> Result> { + open_inner_with_mode(dir, opts, None, true) } /// `open_inner` with an optional pre-built IO backend. `None` resolves the @@ -1934,6 +2113,15 @@ fn open_inner_with( dir: &Path, opts: Options, io_override: Option<(Arc, &'static str)>, +) -> Result> { + open_inner_with_mode(dir, opts, io_override, false) +} + +fn open_inner_with_mode( + dir: &Path, + opts: Options, + io_override: Option<(Arc, &'static str)>, + require_fresh: bool, ) -> Result> { let paths = DbPaths::new(dir); if !dir.exists() { @@ -1963,6 +2151,21 @@ fn open_inner_with( ))); } + if require_fresh { + let mut entries = std::fs::read_dir(dir)?; + let has_existing_data = entries.any(|entry| { + entry + .map(|entry| entry.file_name() != std::ffi::OsStr::new("LOCK")) + .unwrap_or(true) + }); + if has_existing_data { + return Err(Error::InvalidArgument(format!( + "{} is not an empty destination", + dir.display() + ))); + } + } + let (io_backend, backend_name) = match io_override { Some(pair) => pair, None => io::backend(opts.io_backend)?, diff --git a/crates/fluent31/src/journal.rs b/crates/fluent31/src/journal.rs index 7506e4a..e7d33ba 100644 --- a/crates/fluent31/src/journal.rs +++ b/crates/fluent31/src/journal.rs @@ -648,30 +648,18 @@ pub fn rebuild(journal_dir: impl AsRef, dest: impl AsRef, opts: Opti // after the base-end is replayable deltas. let anchor = find_last_complete_checkpoint(&records)?; - let db = Db::open(dest, opts)?; - let mut base_keys = 0u64; + let base_span = &records[anchor.base_start..anchor.base_end]; + let base_keys = validate_base_span(base_span, &opts)?; + let base_entries = base_span + .iter() + .filter_map(|rec| match decode_base_record(rec) { + Ok(Some(entry)) => Some(Ok(entry)), + Ok(None) => None, + Err(err) => Some(Err(err)), + }); + let db = Db::create_from_sorted_fallible(dest, opts, base_entries)?; let mut deltas_applied = 0u64; let mut last_seqno = 0u64; - - // apply the base - for rec in &records[anchor.base_start..anchor.base_end] { - let mut r = Reader::new(rec); - let tag = r.u8()?; - // journals written before the mid-base rotation fix (#29) can carry a - // base span that straddles a file boundary, with the next file's - // header interleaved; its provenance was already verified per file in - // read_all_records, so skip it rather than strand the journal - if tag == TAG_HEADER { - continue; - } - if tag != TAG_BASE { - return Err(corrupt("expected base record in base span")); - } - let key = r.len_prefixed()?.to_vec(); - let value = r.len_prefixed()?.to_vec(); - db.put(key, value)?; - base_keys += 1; - } last_seqno = last_seqno.max(anchor.base_seqno); // replay deltas after the base-end, in file (== seqno) order @@ -719,6 +707,46 @@ pub fn rebuild(journal_dir: impl AsRef, dest: impl AsRef, opts: Opti }) } +/// Decode one record in a base span. Historical journals can contain a +/// provenance header at a segment boundary inside the span; provenance was +/// already verified by `read_all_records`, so it carries no entry here. +fn decode_base_record(rec: &[u8]) -> Result> { + let mut reader = Reader::new(rec); + match reader.u8()? { + TAG_HEADER => Ok(None), + TAG_BASE => { + let key = reader.len_prefixed()?; + let value = reader.len_prefixed()?; + if !reader.is_empty() { + return Err(corrupt("base record has trailing bytes")); + } + Ok(Some((key, value))) + } + _ => Err(corrupt("expected base record in base span")), + } +} + +/// Validate the complete base before creating the destination. Rebuilds can +/// be long-running, so malformed or unsorted input must fail before any SST +/// or value-log output is produced. +fn validate_base_span(records: &[Vec], opts: &Options) -> Result { + let mut previous_key = Vec::new(); + let mut count = 0u64; + for rec in records { + let Some((key, value)) = decode_base_record(rec)? else { + continue; + }; + crate::db::validate_user_entry(opts, key, value.len())?; + if !previous_key.is_empty() && key <= previous_key.as_slice() { + return Err(corrupt("journal base keys are not strictly increasing")); + } + previous_key.clear(); + previous_key.extend_from_slice(key); + count += 1; + } + Ok(count) +} + struct Anchor { base_seqno: SeqNo, /// index of the first BASE record (checkpoint index + 1) diff --git a/crates/fluent31/tests/bulk_load.rs b/crates/fluent31/tests/bulk_load.rs new file mode 100644 index 0000000..dac9a24 --- /dev/null +++ b/crates/fluent31/tests/bulk_load.rs @@ -0,0 +1,195 @@ +use fluent31::{Compression, Db, Error, Options, SyncMode}; + +fn options() -> Options { + Options { + sync: SyncMode::Always, + io_backend: fluent31::IoBackend::Std, + block_size: 256, + compression: Compression::Lz4, + target_file_size: 2 << 10, + value_threshold: 48, + vlog_file_size: 2 << 10, + ..Options::default() + } +} + +fn entries(count: u32) -> Vec<(Vec, Vec)> { + (0..count) + .map(|i| { + let key = format!("key/{i:06}").into_bytes(); + let len = if i % 2 == 0 { 32 } else { 96 }; + let value = vec![(i % 251) as u8; len]; + (key, value) + }) + .collect() +} + +#[test] +fn creates_final_level_base_with_configured_storage_features() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("store"); + let expected = entries(400); + + let db = Db::create_from_sorted(&path, options(), expected.clone()).unwrap(); + assert_eq!(db.seqno(), expected.len() as u64); + assert_eq!( + db.iter(None, None, false) + .unwrap() + .collect::, _>>() + .unwrap(), + expected + ); + + let stats = db.stats(); + assert_eq!(stats.memtable_bytes, 0); + assert_eq!(stats.immutable_memtables, 0); + assert!(stats.levels[..stats.levels.len() - 1] + .iter() + .all(|(runs, fragments, _)| *runs == 0 && *fragments == 0)); + let (runs, fragments, bytes) = stats.levels.last().copied().unwrap(); + assert_eq!(runs, 1); + assert!(fragments > 1); + assert!(bytes > 0); + assert!(stats.vlog_files > 1); + + let before_compact = stats.levels.clone(); + db.compact_all().unwrap(); + assert_eq!(db.stats().levels, before_compact); + drop(db); + + let db = Db::open(&path, options()).unwrap(); + assert_eq!( + db.iter(None, None, false) + .unwrap() + .collect::, _>>() + .unwrap(), + expected + ); + db.put(b"key/000001".to_vec(), b"updated".to_vec()).unwrap(); + db.delete(b"key/000002".to_vec()).unwrap(); + db.put(b"key/999999".to_vec(), b"new".to_vec()).unwrap(); + db.flush().unwrap(); + db.compact_all().unwrap(); + drop(db); + + let db = Db::open(&path, options()).unwrap(); + assert_eq!(db.get(b"key/000001").unwrap().unwrap(), b"updated"); + assert!(db.get(b"key/000002").unwrap().is_none()); + assert_eq!(db.get(b"key/999999").unwrap().unwrap(), b"new"); +} + +#[test] +fn rejects_non_increasing_keys_without_publishing_a_partial_base() { + let mut after_completed_fragments = entries(200); + after_completed_fragments.push((b"key/000100".to_vec(), b"duplicate".to_vec())); + for input in [ + vec![ + (b"b".to_vec(), b"1".to_vec()), + (b"a".to_vec(), b"2".to_vec()), + ], + vec![ + (b"a".to_vec(), b"1".to_vec()), + (b"a".to_vec(), b"2".to_vec()), + ], + after_completed_fragments, + ] { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("store"); + let err = Db::create_from_sorted(&path, options(), input) + .err() + .expect("unsorted input must fail"); + assert!( + matches!(err, Error::InvalidArgument(message) if message.contains("strictly increasing")) + ); + + let db = Db::open(&path, options()).unwrap(); + assert_eq!(db.seqno(), 0); + assert_eq!(db.iter(None, None, false).unwrap().count(), 0); + } +} + +#[test] +fn fallible_input_error_does_not_publish_completed_fragments() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("store"); + let input = + entries(200) + .into_iter() + .map(Ok) + .chain(std::iter::once(Err(Error::InvalidArgument( + "source failed".into(), + )))); + + let err = Db::create_from_sorted_fallible(&path, options(), input) + .err() + .expect("the source error must be returned"); + assert!(matches!(err, Error::InvalidArgument(message) if message == "source failed")); + + let db = Db::open(&path, options()).unwrap(); + assert_eq!(db.seqno(), 0); + assert_eq!(db.iter(None, None, false).unwrap().count(), 0); +} + +#[test] +fn validates_keys_and_values_like_the_normal_write_path() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("reserved"); + let err = Db::create_from_sorted(&path, options(), vec![(vec![0, b'k'], b"value".to_vec())]) + .err() + .expect("reserved keys must fail"); + assert!(matches!(err, Error::InvalidArgument(_))); + + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("oversized"); + let mut opts = options(); + opts.max_value_size = 3; + let err = Db::create_from_sorted(&path, opts, vec![(b"key".to_vec(), b"value".to_vec())]) + .err() + .expect("oversized values must fail"); + assert!(matches!(err, Error::InvalidArgument(message) if message.contains("max_value_size"))); +} + +#[test] +fn chooses_a_bottom_level_large_enough_for_the_base() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("store"); + let mut opts = options(); + opts.max_levels = 1; + opts.memtable_size = 128; + opts.l0_compaction_trigger = 1; + opts.tier_width = 2; + + let db = Db::create_from_sorted(&path, opts, entries(400)).unwrap(); + let stats = db.stats(); + assert!(stats.levels.len() > 1); + assert!(stats.levels[..stats.levels.len() - 1] + .iter() + .all(|(runs, fragments, _)| *runs == 0 && *fragments == 0)); + assert_eq!(stats.levels.last().unwrap().0, 1); +} + +#[test] +fn refuses_an_existing_store_or_nonempty_directory() { + let existing = tempfile::tempdir().unwrap(); + drop(Db::open(existing.path(), options()).unwrap()); + let err = Db::create_from_sorted( + existing.path(), + options(), + std::iter::empty::<(Vec, Vec)>(), + ) + .err() + .expect("an existing store must be refused"); + assert!( + matches!(err, Error::InvalidArgument(message) if message.contains("not an empty destination")) + ); + + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("store"); + std::fs::create_dir(&path).unwrap(); + std::fs::write(path.join("unrelated"), b"keep").unwrap(); + assert!( + Db::create_from_sorted(&path, options(), std::iter::empty::<(Vec, Vec)>(),) + .is_err() + ); + assert_eq!(std::fs::read(path.join("unrelated")).unwrap(), b"keep"); +} diff --git a/crates/fluent31/tests/journal_rebuild.rs b/crates/fluent31/tests/journal_rebuild.rs index 9f07357..8796e24 100644 --- a/crates/fluent31/tests/journal_rebuild.rs +++ b/crates/fluent31/tests/journal_rebuild.rs @@ -151,6 +151,40 @@ fn journal_captures_state_that_predates_attach() { assert_eq!(dump(&rebuilt), expected); } +#[test] +fn rebuild_installs_the_base_directly_at_the_bottom_level() { + let db_dir = tempfile::tempdir().unwrap(); + let jrn_dir = tempfile::tempdir().unwrap(); + + let expected = { + let db = Arc::new(Db::open(db_dir.path(), opts()).unwrap()); + for i in 0..500u32 { + db.put(k(i), v(i, "base")).unwrap(); + } + db.flush().unwrap(); + let expected = dump(&db); + let journal = Journal::attach(db, jrn_dir.path()).unwrap(); + drop(journal); + expected + }; + + let rebuilt_root = tempfile::tempdir().unwrap(); + let rebuilt_path = rebuilt_root.path().join("store"); + let report = journal::rebuild(jrn_dir.path(), &rebuilt_path, opts()).unwrap(); + assert_eq!(report.base_keys, expected.len() as u64); + assert_eq!(report.deltas_applied, 0); + + let rebuilt = Db::open(&rebuilt_path, opts()).unwrap(); + assert_eq!(dump(&rebuilt), expected); + let stats = rebuilt.stats(); + assert_eq!(stats.memtable_bytes, 0); + assert_eq!(stats.immutable_memtables, 0); + assert!(stats.levels[..stats.levels.len() - 1] + .iter() + .all(|(runs, fragments, _)| *runs == 0 && *fragments == 0)); + assert_eq!(stats.levels.last().unwrap().0, 1); +} + // --------------------------------------------------------------------------- // The journal keeps up under sustained write pressure (heals lag if it can't) // --------------------------------------------------------------------------- From 87e1e1957d3b4cdbbba6f940ad3e262252fed30f Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Fri, 7 Aug 2026 00:34:34 +0900 Subject: [PATCH 2/6] fix(manifest): prune stale generations at save, not only at open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Older MANIFEST files were swept only in open(); a long-running process rotates a manifest every few minutes and never reopens, so stale generations accumulate without bound — measured 441 files / 441 MB in one production store after 31 hours of uptime (~14 GB across its 32 stores, ~10% of the volume). save() now sweeps generations below the one it just flipped CURRENT to. Best-effort by design: a crash mid-sweep leaves work for the next save, and newer generations (pre-flip crash artifacts) are deliberately left for the open-time sweep, which already handles them. Co-Authored-By: Claude Fable 5 --- crates/fluent31/src/manifest.rs | 36 +++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/crates/fluent31/src/manifest.rs b/crates/fluent31/src/manifest.rs index 5d78f03..45ac208 100644 --- a/crates/fluent31/src/manifest.rs +++ b/crates/fluent31/src/manifest.rs @@ -295,6 +295,23 @@ pub(crate) fn save(paths: &DbPaths, gen: u64, data: &ManifestData) -> Result<()> } sync_dir(&paths.dir)?; atomic_write(&paths.current(), format!("MANIFEST-{gen:06}\n").as_bytes())?; + // Older generations are otherwise swept only at open; a long-running process + // rotates a manifest every few minutes and never reopens, so they accumulate + // without bound. Best-effort: a crash mid-sweep just leaves work for the next + // save, and the open-time sweep still covers pre-flip crashed newer gens. + if let Ok(rd) = std::fs::read_dir(&paths.dir) { + for entry in rd.flatten() { + let name = entry.file_name(); + let stale = name + .to_str() + .and_then(|n| n.strip_prefix("MANIFEST-")) + .and_then(|s| s.parse::().ok()) + .is_some_and(|g| g < gen); + if stale { + let _ = std::fs::remove_file(entry.path()); + } + } + } Ok(()) } @@ -451,4 +468,23 @@ mod tests { assert_eq!(gen, 2); assert_eq!(got, d2); } + + #[test] + fn save_prunes_older_generations() { + let dir = tempfile::tempdir().unwrap(); + let paths = DbPaths::new(dir.path()); + let d = sample(); + for gen in 1..=3 { + save(&paths, gen, &d).unwrap(); + } + assert!(!paths.manifest(1).exists()); + assert!(!paths.manifest(2).exists()); + assert!(paths.manifest(3).exists()); + // A newer gen (pre-flip crash artifact) must survive a save of an older one. + save(&paths, 5, &d).unwrap(); + std::fs::rename(paths.manifest(5), paths.manifest(7)).unwrap(); + save(&paths, 6, &d).unwrap(); + assert!(paths.manifest(7).exists()); + assert!(paths.manifest(6).exists()); + } } From 02126c889cdd925eb209b7ca6e645e4d0542d875 Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Fri, 7 Aug 2026 01:05:09 +0900 Subject: [PATCH 3/6] =?UTF-8?q?perf(table):=20flat-arena=20block=20index?= =?UTF-8?q?=20=E2=80=94=20half=20the=20pinned=20heap,=20zero=20per-entry?= =?UTF-8?q?=20allocs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pinned index held one heap Vec per block entry; against ~40-byte keys the Vec header + malloc header + size-class rounding roughly doubled resident bytes, and a 148 GB production index pays that across ~30 M entries (~3 GB observed). One concatenated keys arena + u32 ends + a BlockRef vector keeps the same binary search over the same keys with no per-entry allocations. Existing multi-block tests (256-byte blocks, 500 keys, every key looked up, forward/reverse iteration and seeks) cover the lookup equivalence. Co-Authored-By: Claude Fable 5 --- crates/fluent31/src/table/reader.rs | 91 +++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 25 deletions(-) diff --git a/crates/fluent31/src/table/reader.rs b/crates/fluent31/src/table/reader.rs index 3979630..53196b4 100644 --- a/crates/fluent31/src/table/reader.rs +++ b/crates/fluent31/src/table/reader.rs @@ -11,20 +11,55 @@ use crate::coding::Reader; use crate::error::{corrupt, Result}; use crate::io::DbFile; use crate::iter::InternalIterator; -use crate::types::{ - cmp_ikey, ikey_kind, ikey_seqno, ikey_ukey, make_seek_ikey, SeqNo, ValueKind, -}; +use crate::types::{cmp_ikey, ikey_kind, ikey_seqno, ikey_ukey, make_seek_ikey, SeqNo, ValueKind}; + +/// The pinned block index, flat: one keys arena + one offsets vector instead of a +/// heap `Vec` per entry. The index lives for the table's whole life and there is +/// one entry per data block, so per-entry allocator overhead (Vec header + malloc +/// header + size-class rounding, ~60-80 bytes against ~40-byte keys) roughly doubled +/// its residency. Same lookups, half the heap, no per-entry allocations to churn. +struct TableIndex { + /// Concatenated last-ikeys, in index order. + keys: Vec, + /// `ends[i]` = exclusive end of entry i's key in `keys` (entry i starts at `ends[i-1]`). + ends: Vec, + blocks: Vec, +} + +impl TableIndex { + fn len(&self) -> usize { + self.blocks.len() + } + + fn key(&self, i: usize) -> &[u8] { + let start = if i == 0 { 0 } else { self.ends[i - 1] as usize }; + &self.keys[start..self.ends[i] as usize] + } + + fn block(&self, i: usize) -> BlockRef { + self.blocks[i] + } -struct IndexEntry { - last_ikey: Vec, - block: BlockRef, + /// First entry whose key is `>= target` (the partition point of `< target`). + fn lower_bound(&self, target: &[u8]) -> usize { + let (mut lo, mut hi) = (0, self.len()); + while lo < hi { + let mid = lo + (hi - lo) / 2; + if cmp_ikey(self.key(mid), target) == std::cmp::Ordering::Less { + lo = mid + 1; + } else { + hi = mid; + } + } + lo + } } pub(crate) struct Table { pub id: u64, file: Arc, cache: Arc, - index: Vec, + index: TableIndex, filter: Vec, pub stats: TableStats, } @@ -43,26 +78,32 @@ impl Table { let stats = TableStats::decode(&read_block_verified(file.as_ref(), footer.stats)?)?; let index_payload = read_block_verified(file.as_ref(), footer.index)?; - let mut index = Vec::new(); + let mut index = TableIndex { + keys: Vec::new(), + ends: Vec::new(), + blocks: Vec::new(), + }; let mut r = Reader::new(&index_payload); while !r.is_empty() { - let last_ikey = r.len_prefixed()?.to_vec(); + let last_ikey = r.len_prefixed()?; if last_ikey.len() < crate::types::TRAILER_LEN { return Err(corrupt("index key shorter than trailer")); } let off = r.uvarint()?; let len = r.uvarint()?; - index.push(IndexEntry { - last_ikey, - block: BlockRef { - off, - len: len as u32, - }, + index.keys.extend_from_slice(last_ikey); + let end = u32::try_from(index.keys.len()) + .map_err(|_| corrupt("index keys exceed u32 arena"))?; + index.ends.push(end); + index.blocks.push(BlockRef { + off, + len: len as u32, }); } - if index.is_empty() { + if index.blocks.is_empty() { return Err(corrupt("table has no data blocks")); } + index.keys.shrink_to_fit(); Ok(Table { id, file, @@ -83,7 +124,9 @@ impl Table { pub fn read_chunk(&self, off: u64, len: usize) -> Result> { let flen = self.file.len()?; if off >= flen { - return Err(corrupt(format!("chunk offset {off} beyond table end {flen}"))); + return Err(corrupt(format!( + "chunk offset {off} beyond table end {flen}" + ))); } let n = (len as u64).min(flen - off) as usize; let mut buf = vec![0u8; n]; @@ -92,7 +135,7 @@ impl Table { } fn load_block(&self, idx: usize) -> Result> { - let r = self.index[idx].block; + let r = self.index.block(idx); let payload = match self.cache.get(self.id, r.off) { Some(p) => p, None => { @@ -107,8 +150,7 @@ impl Table { /// First block whose last key is `>= target` — the only block that can /// contain the lower bound for `target`. fn index_lower_bound(&self, target: &[u8]) -> usize { - self.index - .partition_point(|e| cmp_ikey(&e.last_ikey, target) == std::cmp::Ordering::Less) + self.index.lower_bound(target) } pub fn may_contain_ukey(&self, ukey: &[u8]) -> bool { @@ -401,14 +443,14 @@ mod tests { // seek beyond the end / before the start it.seek(&make_seek_ikey(b"zzz", MAX_SEQNO)).unwrap(); assert!(!it.valid()); - it.seek_for_prev(&make_seek_ikey(b"aaa", MAX_SEQNO)).unwrap(); + it.seek_for_prev(&make_seek_ikey(b"aaa", MAX_SEQNO)) + .unwrap(); assert!(!it.valid()); } #[test] fn bloom_filters_absent_keys() { - let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = - vec![(b"only", 1, ValueKind::Put, b"v")]; + let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = vec![(b"only", 1, ValueKind::Put, b"v")]; let (_dir, t) = build_table(&refs, 4096); assert!(t.may_contain_ukey(b"only")); assert!(!t.may_contain_ukey(b"absent")); // outside key range @@ -456,8 +498,7 @@ mod tests { /// stays format 1 — readable by binaries that predate compression. #[test] fn incompressible_lz4_table_stays_format_1() { - let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = - vec![(b"only", 1, ValueKind::Put, b"v")]; + let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = vec![(b"only", 1, ValueKind::Put, b"v")]; let (_dir, t, _) = build_table_sized(&refs, 4096, Compression::Lz4); assert_eq!(footer_format(&t), FORMAT); let got = t.get(b"only", MAX_SEQNO).unwrap().unwrap(); From 71a1e0d5324f9823825d76b8ef3fed8551624c5b Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Fri, 7 Aug 2026 12:02:32 +0900 Subject: [PATCH 4/6] perf(table): bloom filters ride the block cache instead of pinning the heap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every open table pinned its whole bloom filter for life. At 10 bits/key that is the single largest pinned population on a big store — measured 2.76 GB across one production deployment's 2,481 tables — and it grows with data size, not load, so no cache budget ever governed it. The filter now loads through the shared block cache under its natural (file_id, filter.off) key: hot filters stay resident by being used, cold tables cost nothing, and block_cache_size finally bounds every block the reader touches. may_contain_ukey becomes fallible (a cache miss re-reads the block); the two non-reader callers — the run-level check and the compaction tombstone-drop predicate — propagate the error instead of swallowing it. Corruption still surfaces at open via a verify-and-drop read. Quick bench probe (single run, examples/bench): cold gets -6.5% (the added cache probe per lookup), everything else within noise. Co-Authored-By: Claude Fable 5 --- crates/fluent31/src/compaction.rs | 13 +++-- crates/fluent31/src/table/reader.rs | 73 +++++++++++++++++++++++++---- crates/fluent31/src/version.rs | 11 +++-- 3 files changed, 81 insertions(+), 16 deletions(-) diff --git a/crates/fluent31/src/compaction.rs b/crates/fluent31/src/compaction.rs index 6d44427..cda001f 100644 --- a/crates/fluent31/src/compaction.rs +++ b/crates/fluent31/src/compaction.rs @@ -262,9 +262,16 @@ fn run_job(db: &Arc, job: Job) -> Result<()> { kept_le_w = true; // the newest version at-or-below the watermark: keep, unless // it is a tombstone provably shadowing nothing older - if kind == ValueKind::Delete - && !job.older.iter().any(|r| r.may_contain_ukey(&cur_ukey)) - { + let mut shadows_older = false; + if kind == ValueKind::Delete { + for r in job.older.iter() { + if r.may_contain_ukey(&cur_ukey)? { + shadows_older = true; + break; + } + } + } + if kind == ValueKind::Delete && !shadows_older { (false, false) } else { (true, false) diff --git a/crates/fluent31/src/table/reader.rs b/crates/fluent31/src/table/reader.rs index 53196b4..5a00941 100644 --- a/crates/fluent31/src/table/reader.rs +++ b/crates/fluent31/src/table/reader.rs @@ -60,7 +60,13 @@ pub(crate) struct Table { file: Arc, cache: Arc, index: TableIndex, - filter: Vec, + /// Where the bloom filter lives in the file. The filter itself rides the shared + /// block cache under `(id, filter.off)` like any data block: at ~10 bits/key it + /// is the single largest pinned population on a big store (measured 2.76 GB + /// across one production deployment's tables), it grows with data rather than + /// load, and a cold table's filter is pure dead weight. Hot filters stay + /// resident by being used; cold ones fall out with the LRU. + filter: BlockRef, pub stats: TableStats, } @@ -74,7 +80,10 @@ impl Table { file.read_exact_at(flen - FOOTER_LEN as u64, &mut fbuf)?; let footer = Footer::decode(&fbuf)?; - let filter = read_block_verified(file.as_ref(), footer.filter)?; + // Verify the filter block is readable now (corruption should surface at open, + // as it always has), but do not keep the bytes — queries reload it through the + // block cache on demand. + read_block_verified(file.as_ref(), footer.filter)?; let stats = TableStats::decode(&read_block_verified(file.as_ref(), footer.stats)?)?; let index_payload = read_block_verified(file.as_ref(), footer.index)?; @@ -109,7 +118,7 @@ impl Table { file, cache, index, - filter, + filter: footer.filter, stats, }) } @@ -153,16 +162,32 @@ impl Table { self.index.lower_bound(target) } - pub fn may_contain_ukey(&self, ukey: &[u8]) -> bool { + fn load_filter(&self) -> Result>> { + match self.cache.get(self.id, self.filter.off) { + Some(p) => Ok(p), + None => { + let p = Arc::new(read_block_verified(self.file.as_ref(), self.filter)?); + self.cache.insert(self.id, self.filter.off, p.clone()); + Ok(p) + } + } + } + + /// Fallible since the filter rides the block cache: a miss re-reads it from + /// the file, and that read can fail. The key-range check stays free. + pub fn may_contain_ukey(&self, ukey: &[u8]) -> Result { if ukey < self.stats.min_ukey() || ukey > self.stats.max_ukey() { - return false; + return Ok(false); } - bloom::may_contain(&self.filter, bloom::hash64(ukey)) + Ok(bloom::may_contain( + &self.load_filter()?, + bloom::hash64(ukey), + )) } /// Newest version of `ukey` with `seqno <= seq` in this table. pub fn get(&self, ukey: &[u8], seq: SeqNo) -> Result)>> { - if !self.may_contain_ukey(ukey) { + if !self.may_contain_ukey(ukey)? { return Ok(None); } let target = make_seek_ikey(ukey, seq); @@ -452,8 +477,38 @@ mod tests { fn bloom_filters_absent_keys() { let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = vec![(b"only", 1, ValueKind::Put, b"v")]; let (_dir, t) = build_table(&refs, 4096); - assert!(t.may_contain_ukey(b"only")); - assert!(!t.may_contain_ukey(b"absent")); // outside key range + assert!(t.may_contain_ukey(b"only").unwrap()); + assert!(!t.may_contain_ukey(b"absent").unwrap()); // outside key range + } + + /// The filter rides the block cache: it must land there after use, and a query + /// must still answer correctly after the cache forgets it (reload path). + #[test] + fn bloom_filter_rides_the_block_cache() { + let data = many(); + let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = data + .iter() + .map(|(k, s, kind, v)| (k.as_slice(), *s, *kind, v.as_slice())) + .collect(); + let (_dir, t) = build_table(&refs, 256); + assert!(t.cache.get(t.id, t.filter.off).is_none(), "not pre-warmed"); + assert!(t.may_contain_ukey(b"key00042").unwrap()); + assert!( + t.cache.get(t.id, t.filter.off).is_some(), + "filter must be charged to the cache after use" + ); + // Flood the 1 MiB test cache with far more junk than its capacity — without + // touching the filter, so its LRU slot ages out in whichever shard holds it — + // then query again: the filter must reload from the file and answer identically. + for i in 0..2048u64 { + t.cache.insert(u64::MAX, i, Arc::new(vec![0u8; 8 << 10])); + } + assert!( + t.cache.get(t.id, t.filter.off).is_none(), + "flood must evict the filter" + ); + assert!(t.may_contain_ukey(b"key00042").unwrap()); + assert!(!t.may_contain_ukey(b"nope-not-here").unwrap()); } /// Lz4 tables round-trip every read path, shrink the file, and carry the diff --git a/crates/fluent31/src/version.rs b/crates/fluent31/src/version.rs index 082533d..c17d9e6 100644 --- a/crates/fluent31/src/version.rs +++ b/crates/fluent31/src/version.rs @@ -80,10 +80,13 @@ impl Run { } /// Can this run possibly contain `ukey`? Bloom-backed; false means - /// provably absent (used by the tombstone-drop predicate). - pub fn may_contain_ukey(&self, ukey: &[u8]) -> bool { - self.fragment_for(ukey) - .is_some_and(|t| t.table.may_contain_ukey(ukey)) + /// provably absent (used by the tombstone-drop predicate). Fallible because + /// the filter rides the block cache and a miss re-reads it from the file. + pub fn may_contain_ukey(&self, ukey: &[u8]) -> Result { + match self.fragment_for(ukey) { + Some(t) => t.table.may_contain_ukey(ukey), + None => Ok(false), + } } pub fn iter(&self) -> RunIter { From f5ae8c398f86df62f846ea0208160074f548d610 Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Fri, 7 Aug 2026 15:49:44 +0900 Subject: [PATCH 5/6] feat(table): zstd codec + bottom-level compression placement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compression grows a Zstd variant (codec byte 2, level fixed at 3 — measured on real store data, level 8 buys ~0.5% for several times the CPU) and Options grows bottom_compression: the codec for compaction outputs landing in the deepest level, where ~90% of a store's bytes live and rewrites are rarest. None means "same as compression", so nothing changes without opting in. Measured on two production tables (154 MB raw): zstd-3 stores 33% of raw at 32 KiB blocks vs lz4's 41% — about a quarter less disk — at 2.3 GB/s single-thread decompress, which is noise on the read path. Tables carrying a zstd block bump to format 3, so pre-zstd readers reject them at open ("unsupported table format") instead of failing mid-read on a codec byte. Per-block store-raw-if-not-smaller behavior matches lz4, and the zstd payload is size-prefixed like lz4's. The placement flag is wired at all three Job constructions and proven through the production compaction path: with a 2-level tree every output is the bottom, so the test asserts format-3 tables appear with bottom_compression set and never without it. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 47 +++++++++++++++++++ crates/fluent31/Cargo.toml | 1 + crates/fluent31/src/compaction.rs | 12 ++++- crates/fluent31/src/config.rs | 13 ++++++ crates/fluent31/src/db.rs | 67 ++++++++++++++++++++++++++++ crates/fluent31/src/table/builder.rs | 27 +++++++++-- crates/fluent31/src/table/mod.rs | 25 ++++++++++- crates/fluent31/src/table/reader.rs | 38 ++++++++++++++++ 8 files changed, 225 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cbfe330..06fe506 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -293,6 +293,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -822,6 +824,7 @@ dependencies = [ "sha2", "tempfile", "wasmtime", + "zstd", ] [[package]] @@ -1154,6 +1157,16 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "leb128fmt" version = "0.1.0" @@ -1407,6 +1420,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "polling" version = "3.11.0" @@ -2510,3 +2529,31 @@ name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/crates/fluent31/Cargo.toml b/crates/fluent31/Cargo.toml index 5bcb249..5ab9691 100644 --- a/crates/fluent31/Cargo.toml +++ b/crates/fluent31/Cargo.toml @@ -11,6 +11,7 @@ crc32fast = "1" # deterministic store identity (identity.rs): truncated SHA-256 lineage hash sha2 = "0.10" lz4_flex = { version = "0.11", default-features = false, features = ["std", "safe-encode", "safe-decode"] } +zstd = { version = "0.13", default-features = false } wasmtime = { version = "46.0.1", default-features = false, features = ["runtime", "cranelift", "wat"], optional = true } [target.'cfg(target_os = "linux")'.dependencies] diff --git a/crates/fluent31/src/compaction.rs b/crates/fluent31/src/compaction.rs index cda001f..53c7af8 100644 --- a/crates/fluent31/src/compaction.rs +++ b/crates/fluent31/src/compaction.rs @@ -39,6 +39,9 @@ pub(crate) struct Job { /// may only be dropped if none of these can contain its key. older: Vec, kind: JobKind, + /// Output lands in (or creates) the deepest level, where + /// `Options::bottom_compression` applies. + bottom: bool, } enum JobKind { @@ -141,6 +144,7 @@ fn pick(db: &Arc, force: bool) -> Option { inputs: v.levels[i].clone(), older, kind: JobKind::Tier, + bottom: i + 1 == v.levels.len() - 1, }); } } @@ -160,6 +164,7 @@ fn pick(db: &Arc, force: bool) -> Option { inputs: v.levels[last].clone(), older: Vec::new(), kind: JobKind::Tier, + bottom: true, }); } if v.levels[last].len() >= 2 { @@ -219,6 +224,7 @@ fn pick(db: &Arc, force: bool) -> Option { keep_right, new_run_id: db.alloc_file_id(), }, + bottom: true, }); } None @@ -306,7 +312,11 @@ fn run_job(db: &Arc, job: Job) -> Result<()> { file, db.opts.block_size, db.opts.bloom_bits_per_key, - db.opts.compression, + if job.bottom { + db.opts.bottom_compression.unwrap_or(db.opts.compression) + } else { + db.opts.compression + }, ), )); } diff --git a/crates/fluent31/src/config.rs b/crates/fluent31/src/config.rs index 441204f..c981247 100644 --- a/crates/fluent31/src/config.rs +++ b/crates/fluent31/src/config.rs @@ -27,6 +27,12 @@ pub enum Compression { /// shrink are stored raw, and a table's format version is bumped only /// when it actually contains a compressed block. Lz4, + /// Zstd (level 3) block compression. Denser than LZ4 (~20-25% smaller on + /// real stores) at a decompress speed that is still far above read-path + /// needs; the right pick for data that is written once and rarely read, + /// i.e. the bottom level. Tables carrying zstd blocks bump to format 3, + /// so pre-zstd readers reject them at open instead of failing mid-read. + Zstd, } /// IO backend selection. @@ -77,6 +83,12 @@ pub struct Options { pub block_size: usize, /// Per-block SST compression codec (applies to newly written tables). pub compression: Compression, + /// Codec for compaction outputs landing in the BOTTOM level, where ~90% + /// of a store's bytes live and rewrites are rarest. `None` (the default) + /// means "same as `compression`". The classic pairing is fast-codec upper + /// levels + `Zstd` bottom: hot rewrites stay cheap, the cold bulk gets + /// the dense codec. + pub bottom_compression: Option, /// Bloom filter budget per key, in bits. pub bloom_bits_per_key: usize, /// Shared block cache capacity in bytes. @@ -156,6 +168,7 @@ impl Default for Options { max_immutable_memtables: 2, block_size: 8 << 10, compression: Compression::None, + bottom_compression: None, bloom_bits_per_key: 10, block_cache_size: 64 << 20, l0_compaction_trigger: 4, diff --git a/crates/fluent31/src/db.rs b/crates/fluent31/src/db.rs index dd9dd7c..0a43fe8 100644 --- a/crates/fluent31/src/db.rs +++ b/crates/fluent31/src/db.rs @@ -3265,3 +3265,70 @@ mod group_commit_tests { ); } } + +#[cfg(test)] +mod bottom_compression_tests { + use super::*; + use crate::config::Compression; + + /// Footer layout: filter(12) + index(12) + stats(12) + format(4) + magic(8). + fn table_formats(dir: &std::path::Path) -> Vec { + let mut out = Vec::new(); + for e in std::fs::read_dir(dir).unwrap().flatten() { + let name = e.file_name().to_string_lossy().into_owned(); + if name.starts_with("sst-") && name.ends_with(".tbl") { + let b = std::fs::read(e.path()).unwrap(); + out.push(u32::from_le_bytes( + b[b.len() - 12..b.len() - 8].try_into().unwrap(), + )); + } + } + out + } + + fn formats_after_compaction(bottom: Option) -> Vec { + let dir = tempfile::tempdir().unwrap(); + let db = crate::Db::open( + dir.path(), + Options { + sync: SyncMode::Never, + wasm_enabled: false, + memtable_size: 4 << 10, + block_size: 256, + max_levels: 2, + compression: Compression::Lz4, + bottom_compression: bottom, + ..Options::default() + }, + ) + .unwrap(); + for i in 0..2000u32 { + db.put( + format!("key{i:06}").into_bytes(), + format!("value-{i}-{}", "x".repeat(64)).into_bytes(), + ) + .unwrap(); + } + db.flush().unwrap(); + db.compact_all().unwrap(); + table_formats(dir.path()) + } + + /// Wiring, not unit: with a 2-level tree every compaction output IS the + /// bottom, so `bottom_compression: Zstd` must yield format-3 tables from + /// the production compaction path — and the same workload without it + /// must not, proving the flag (not something else) selects the codec. + #[test] + fn bottom_compression_reaches_the_bottom_level() { + let with = formats_after_compaction(Some(Compression::Zstd)); + assert!( + with.iter().any(|f| *f == 3), + "no zstd table written: {with:?}" + ); + let without = formats_after_compaction(None); + assert!( + without.iter().all(|f| *f <= 2), + "unexpected format-3 table: {without:?}" + ); + } +} diff --git a/crates/fluent31/src/table/builder.rs b/crates/fluent31/src/table/builder.rs index 4ae0ca4..e2d0b89 100644 --- a/crates/fluent31/src/table/builder.rs +++ b/crates/fluent31/src/table/builder.rs @@ -2,12 +2,15 @@ use std::sync::Arc; -use super::{BlockRef, Footer, TableStats, CODEC_LZ4, CODEC_NONE, FORMAT, FORMAT_COMPRESSED}; +use super::{ + BlockRef, Footer, TableStats, CODEC_LZ4, CODEC_NONE, CODEC_ZSTD, FORMAT, FORMAT_COMPRESSED, + FORMAT_ZSTD, ZSTD_LEVEL, +}; use crate::block::BlockBuilder; use crate::bloom; use crate::coding::{crc32, put_len_prefixed, put_uvarint}; use crate::config::Compression; -use crate::error::Result; +use crate::error::{Error, Result}; use crate::io::DbFile; use crate::types::{ikey_kind, ikey_seqno, ikey_ukey, ValueKind}; @@ -24,6 +27,7 @@ pub(crate) struct TableBuilder { /// Whether any block was actually stored compressed — gates the footer /// format bump (old readers keep opening tables that stayed all-raw). wrote_compressed: bool, + wrote_zstd: bool, key_hashes: Vec, last_hashed_ukey: Vec, @@ -48,6 +52,7 @@ impl TableBuilder { index: Vec::new(), offset: 0, wrote_compressed: false, + wrote_zstd: false, key_hashes: Vec::new(), last_hashed_ukey: Vec::new(), stats: TableStats { @@ -119,9 +124,23 @@ impl TableBuilder { (CODEC_NONE, payload) } } + Compression::Zstd => { + let mut framed = Vec::with_capacity(payload.len() / 2 + 8); + framed.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + match zstd::bulk::compress(&payload, ZSTD_LEVEL) { + Ok(compressed) => framed.extend_from_slice(&compressed), + Err(e) => return Err(Error::Io(e)), + } + if framed.len() < payload.len() { + (CODEC_ZSTD, framed) + } else { + (CODEC_NONE, payload) + } + } Compression::None => (CODEC_NONE, payload), }; self.wrote_compressed |= codec != CODEC_NONE; + self.wrote_zstd |= codec == CODEC_ZSTD; buf.push(codec); let crc = crc32(&buf); buf.extend_from_slice(&crc.to_le_bytes()); @@ -172,7 +191,9 @@ impl TableBuilder { filter: filter_ref, index: index_ref, stats: stats_ref, - format: if self.wrote_compressed { + format: if self.wrote_zstd { + FORMAT_ZSTD + } else if self.wrote_compressed { FORMAT_COMPRESSED } else { FORMAT diff --git a/crates/fluent31/src/table/mod.rs b/crates/fluent31/src/table/mod.rs index c6fea8f..b381a6a 100644 --- a/crates/fluent31/src/table/mod.rs +++ b/crates/fluent31/src/table/mod.rs @@ -25,12 +25,22 @@ pub(crate) const FORMAT: u32 = 1; /// table actually contains a compressed block, so stores that never enable /// compression stay readable by format-1 binaries. pub(crate) const FORMAT_COMPRESSED: u32 = 2; +/// At least one block is zstd-compressed (codec byte 2). Pre-zstd binaries +/// reject the table at open ("unsupported table format") instead of failing +/// on the first zstd block mid-read. +pub(crate) const FORMAT_ZSTD: u32 = 3; pub(crate) const FOOTER_LEN: usize = 48; pub(crate) const BLOCK_TRAILER_LEN: usize = 5; /// Block codec bytes (the trailer's `compression u8`). pub(crate) const CODEC_NONE: u8 = 0; pub(crate) const CODEC_LZ4: u8 = 1; +pub(crate) const CODEC_ZSTD: u8 = 2; +/// Zstd compression level for `Compression::Zstd`. Fixed rather than +/// configurable: measured on real store data, level 8 buys ~0.5% over +/// level 3 for several times the compress CPU. Not part of the wire +/// format (only the codec byte is), so it can change without a rebuild. +pub(crate) const ZSTD_LEVEL: i32 = 3; #[derive(Debug, Clone, Copy)] pub(crate) struct BlockRef { @@ -81,7 +91,7 @@ impl Footer { if magic != MAGIC { return Err(corrupt("bad table magic")); } - if !(FORMAT..=FORMAT_COMPRESSED).contains(&format) { + if !(FORMAT..=FORMAT_ZSTD).contains(&format) { return Err(corrupt(format!("unsupported table format {format}"))); } Ok(Footer { @@ -113,6 +123,19 @@ pub(crate) fn read_block_verified(file: &dyn DbFile, r: BlockRef) -> Result lz4_flex::block::decompress_size_prepended(&buf[..payload_end - 1]) .map_err(|e| corrupt(format!("lz4 block decode: {e}"))), + CODEC_ZSTD => { + let body = &buf[..payload_end - 1]; + if body.len() < 4 { + return Err(corrupt("zstd block shorter than its size prefix")); + } + let raw_len = u32::from_le_bytes(body[..4].try_into().unwrap()) as usize; + let raw = zstd::bulk::decompress(&body[4..], raw_len) + .map_err(|e| corrupt(format!("zstd block decode: {e}")))?; + if raw.len() != raw_len { + return Err(corrupt("zstd block size prefix mismatch")); + } + Ok(raw) + } codec => Err(corrupt(format!("unsupported compression {codec}"))), } } diff --git a/crates/fluent31/src/table/reader.rs b/crates/fluent31/src/table/reader.rs index 5a00941..8528879 100644 --- a/crates/fluent31/src/table/reader.rs +++ b/crates/fluent31/src/table/reader.rs @@ -549,6 +549,44 @@ mod tests { assert_eq!(n, 500); } + /// Zstd tables round-trip every read path, land at or below the lz4 size, + /// and carry format 3 so pre-zstd readers reject them at open. + #[test] + fn zstd_round_trip_shrinks_and_bumps_format() { + let data = many(); + let refs: Vec<(&[u8], u64, ValueKind, &[u8])> = data + .iter() + .map(|(k, s, kind, v)| (k.as_slice(), *s, *kind, v.as_slice())) + .collect(); + let (_d1, lz4, lz4_size) = build_table_sized(&refs, 256, Compression::Lz4); + let (_d2, z, z_size) = build_table_sized(&refs, 256, Compression::Zstd); + assert!( + z_size <= lz4_size, + "zstd table ({z_size}) larger than lz4 ({lz4_size})" + ); + assert_eq!(footer_format(&lz4), FORMAT_COMPRESSED); + assert_eq!(footer_format(&z), super::super::FORMAT_ZSTD); + + for (k, s, kind, v) in &data { + let got = z.get(k, MAX_SEQNO).unwrap().unwrap(); + assert_eq!(got.0, *kind); + assert_eq!(got.1, *s); + if *kind == ValueKind::Put { + assert_eq!(got.2, encode_inline(v)); + } + } + assert!(z.get(b"key99999x", MAX_SEQNO).unwrap().is_none()); + + let mut it = z.iter(); + it.seek_to_last().unwrap(); + let mut n = 0; + while it.valid() { + n += 1; + it.prev().unwrap(); + } + assert_eq!(n, 500); + } + /// A table written with compression enabled but where no block shrinks /// stays format 1 — readable by binaries that predate compression. #[test] From a2c071f123b2cb8bbedba424cefa1f42f71e8ce1 Mon Sep 17 00:00:00 2001 From: byeongsu-hong Date: Sun, 9 Aug 2026 18:29:47 +0900 Subject: [PATCH 6/6] Prioritize L0 compaction with bounded slices --- DESIGN.md | 19 +- crates/fluent31/src/compaction.rs | 659 +++++++++++++++++++++++------- crates/fluent31/src/config.rs | 6 + crates/fluent31/src/db.rs | 10 +- 4 files changed, 538 insertions(+), 156 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index e0054bb..87f2dea 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -114,8 +114,19 @@ vlog-file set; it is published under the state lock and pinned by `Arc`. Installation removes exactly the pinned inputs, so flushes prepending to L0 mid-merge are safe. Full-tier merges preserve the newest-first recency invariant. -- **Bottom level**: whenever it holds ≥ 2 runs, everything merges into one - (leveling at the bottom). +- **Bottom level**: whenever it holds ≥ 2 runs, the run adjacent to the + leveled base merges into only the base fragments overlapping its key + range. Untouched fragments survive by identity. A bottom past its byte + budget deepens by moving its runs into a new manifest level without + rewriting table files; repeated moves find a level whose budget fits. +- **Priority scheduler**: a rewrite consumes at most + `compaction_slice_bytes` of input before returning to the picker (the + current user key always finishes). A newly eligible upper level suspends + the deeper job; after the upper job installs, the immutable deep inputs + resume. Jobs still install atomically, and exact-input removal preserves + every run installed while they were suspended. `compact_all` uses this + same scheduler with force thresholds instead of owning an exclusive + full-store merge path. - **Point lookup**: memtable → frozen → runs newest-first; each run binary searches its single candidate fragment after a bloom + range check; the first version with `seqno <= snap` wins; a run whose versions are all @@ -459,8 +470,8 @@ Lock order (strict): `write_mu → manifest → state → snapshots`, with `gc_mu`/`compaction_mu` outermost within their flows. Never hold the state guard while taking the manifest lock (a stats() violation deadlocked exactly as predicted and is fixed). `compaction_mu` serializes the -maintenance thread and user `compact_all` — two concurrent pickers would -merge the same inputs. +maintenance thread and user `compact_all`; the one scheduler may suspend a +deep job between bounded slices, but never executes two jobs concurrently. ## 12. Testing diff --git a/crates/fluent31/src/compaction.rs b/crates/fluent31/src/compaction.rs index 53c7af8..d3d66f2 100644 --- a/crates/fluent31/src/compaction.rs +++ b/crates/fluent31/src/compaction.rs @@ -6,12 +6,15 @@ //! run placed at the FRONT (newest position) of the next level. The last //! level is leveled, maintained INCREMENTALLY: one newer run at a time //! merges into the base run, touching only the base tables its key range -//! overlaps — untouched fragments are spliced through by identity, so job -//! cost is bounded by the newer run, never the whole bottom. When the -//! bottom outgrows its byte budget (`level_target_bytes`) the tree deepens: -//! a new level is created below and the old bottom tier-merges into it. -//! Inputs are pinned at pick time and installation removes exactly the -//! pinned runs — flush can concurrently prepend to L0. +//! overlaps — untouched fragments are spliced through by identity. When the +//! bottom outgrows its byte budget (`level_target_bytes`) the tree deepens by +//! moving its runs to a new level without rewriting their tables. +//! +//! One compaction executor owns the picker. Rewrite jobs run in bounded input +//! slices; after every slice the scheduler may suspend the job and service any +//! newly eligible higher level. Installation remains atomic at job completion, +//! and removes exactly the pinned inputs, so flushes and higher-priority output +//! installed while a job is suspended are preserved. use std::collections::HashMap; use std::sync::atomic::Ordering; @@ -46,9 +49,11 @@ pub(crate) struct Job { enum JobKind { /// Tiered merge: output run lands at the FRONT (newest) of the target - /// level. Also used for deepening (target == current level count: a new - /// bottom level is created on install). + /// level. Tier, + /// Move the current bottom runs into a new, deeper level by manifest edit. + /// Table encoding is self-describing, so every file remains byte-for-byte. + Deepen, /// Incremental bottom merge: `inputs` are the newest bottom run plus /// ONLY the base run's tables overlapping its key range; the output is /// spliced between the base run's untouched fragments to form the new @@ -63,6 +68,26 @@ enum JobKind { }, } +struct RunningJob { + job: Job, + watermark: SeqNo, + merge: Option, + run_id: u64, + tables: Vec>, + builder: Option<(u64, TableBuilder)>, + discard: HashMap, + cur_ukey: Vec, + have_key: bool, + kept_le_w: bool, +} + +struct Scheduler { + active: Option, + suspended: Vec, + force: bool, + did_work: bool, +} + /// Levels can grow (deepening) but never past this: 16 tiers at any sane /// `tier_width` is more data than a single node stores. const MAX_DYNAMIC_LEVELS: usize = 16; @@ -94,17 +119,10 @@ pub(crate) fn level_count_for_bottom_bytes( /// One pass of the maintenance loop; returns whether any work happened. pub(crate) fn maintenance_pass(db: &Arc) -> Result { - let mut did = false; - { + let mut did = { let _guard = db.compaction_mu.lock(); - while let Some(job) = pick(db, false) { - run_job(db, job)?; - did = true; - if db.shutdown.load(Ordering::Acquire) { - return Ok(did); - } - } - } + Scheduler::new(false).run(db)? + }; did |= process_retired(db)?; did |= auto_gc(db)?; Ok(did) @@ -115,12 +133,95 @@ pub(crate) fn maintenance_pass(db: &Arc) -> Result { pub(crate) fn compact_until_quiet(db: &Arc) -> Result<()> { db.check_bg_error()?; let _guard = db.compaction_mu.lock(); - while let Some(job) = pick(db, true) { - run_job(db, job)?; - } + Scheduler::new(true).run(db)?; Ok(()) } +impl Scheduler { + fn new(force: bool) -> Self { + Self { + active: None, + suspended: Vec::new(), + force, + did_work: false, + } + } + + fn run(mut self, db: &Arc) -> Result { + while self.step(db)? { + if db.shutdown.load(Ordering::Acquire) { + break; + } + } + Ok(self.did_work) + } + + /// Run one bounded slice. Newly eligible upper levels preempt a deeper + /// rewrite; suspended jobs retain immutable inputs and install later by + /// exact run id, so outputs created above them are preserved. + fn step(&mut self, db: &Arc) -> Result { + if let Some(level) = self.active.as_ref().map(|job| job.job.level) { + if let Some(job) = pick_before(db, level) { + self.suspended.push(self.active.take().unwrap()); + self.active = Some(RunningJob::new(db, job)?); + } + } + + if self.active.is_none() { + if let Some(level) = self.suspended.last().map(|job| job.job.level) { + if let Some(job) = pick_before(db, level) { + self.active = Some(RunningJob::new(db, job)?); + } else { + self.active = self.suspended.pop(); + } + } else if let Some(job) = pick(db, self.force) { + self.active = Some(RunningJob::new(db, job)?); + } else { + return Ok(false); + } + } + + self.did_work = true; + if self.active.as_mut().unwrap().run_slice(db)? { + self.active = None; + } + Ok(true) + } +} + +fn pick_tier(v: &crate::version::Version, level: usize) -> Job { + let mut older = v.levels[level + 1].clone(); + for deeper in &v.levels[level + 2..] { + older.extend(deeper.iter().cloned()); + } + Job { + level, + target: level + 1, + inputs: v.levels[level].clone(), + older, + kind: JobKind::Tier, + bottom: level + 1 == v.levels.len() - 1, + } +} + +/// Normal-trigger work strictly above `before`, in picker priority order. +fn pick_before(db: &Arc, before: usize) -> Option { + let s = db.state.read(); + let v = &s.version; + let last = v.levels.len() - 1; + for level in 0..last.min(before) { + let trigger = if level == 0 { + db.opts.l0_compaction_trigger + } else { + db.opts.tier_width + }; + if v.levels[level].len() >= trigger { + return Some(pick_tier(v, level)); + } + } + None +} + fn pick(db: &Arc, force: bool) -> Option { let s = db.state.read(); let v = &s.version; @@ -134,36 +235,24 @@ fn pick(db: &Arc, force: bool) -> Option { db.opts.tier_width }; if v.levels[i].len() >= trigger { - let mut older: Vec = v.levels[i + 1].clone(); - for deeper in &v.levels[i + 2..] { - older.extend(deeper.iter().cloned()); - } - return Some(Job { - level: i, - target: i + 1, - inputs: v.levels[i].clone(), - older, - kind: JobKind::Tier, - bottom: i + 1 == v.levels.len() - 1, - }); + return Some(pick_tier(v, i)); } } - // Deepen before merging in place: a bottom level past its byte budget - // gets a NEW level below it — its runs tier-merge down, and the old - // budget wall stops being rewritten wholesale forever. Not under - // `force` (compact_until_quiet wants convergence, not growth). + // Deepen before merging in place. A one-level tree first grows a real L1 + // so future L0 work can preempt bottom maintenance. Otherwise only normal + // maintenance deepens, when the current bottom crosses its byte budget. let bottom_bytes: u64 = v.levels[last].iter().map(|r| r.size()).sum(); - if !force - && v.levels.len() < MAX_DYNAMIC_LEVELS + if v.levels.len() < MAX_DYNAMIC_LEVELS && !v.levels[last].is_empty() - && bottom_bytes > level_target_bytes(&db.opts, last) + && ((last == 0 && v.levels[last].len() >= 2) + || (!force && bottom_bytes > level_target_bytes(&db.opts, last))) { return Some(Job { level: last, target: last + 1, inputs: v.levels[last].clone(), older: Vec::new(), - kind: JobKind::Tier, + kind: JobKind::Deepen, bottom: true, }); } @@ -230,120 +319,200 @@ fn pick(db: &Arc, force: bool) -> Option { None } -fn run_job(db: &Arc, job: Job) -> Result<()> { - let watermark = db.watermark(); +impl RunningJob { + fn new(db: &Arc, job: Job) -> Result { + let deepen = matches!(job.kind, JobKind::Deepen); + let merge = if deepen { + None + } else { + let children: Vec> = job + .inputs + .iter() + .map(|r| Box::new(r.iter()) as Box) + .collect(); + let mut merge = MergeIterator::new(children, false); + merge.seek_to_first()?; + Some(merge) + }; + Ok(Self { + job, + watermark: db.watermark(), + merge, + run_id: if deepen { 0 } else { db.alloc_file_id() }, + tables: Vec::new(), + builder: None, + discard: HashMap::new(), + cur_ukey: Vec::new(), + have_key: false, + kept_le_w: false, + }) + } - let children: Vec> = job - .inputs - .iter() - .map(|r| Box::new(r.iter()) as Box) - .collect(); - let mut merge = MergeIterator::new(children, false); - merge.seek_to_first()?; - - let run_id = db.alloc_file_id(); - let mut tables = Vec::new(); - let mut builder: Option<(u64, TableBuilder)> = None; - let mut discard: HashMap = HashMap::new(); - - let mut cur_ukey: Vec = Vec::new(); - let mut have_key = false; - let mut kept_le_w = false; - - while merge.valid() { - let (keep, is_ptr_drop) = { - let ik = merge.ikey(); - let uk = ikey_ukey(ik); - if !have_key || uk != cur_ukey.as_slice() { - cur_ukey = uk.to_vec(); - have_key = true; - kept_le_w = false; + /// Process at most one scheduling quantum, finishing the current user + /// key so its versions never straddle priority boundaries. Returns true + /// once the job is durably installed. + fn run_slice(&mut self, db: &Arc) -> Result { + if matches!(self.job.kind, JobKind::Deepen) { + install_deepen(db, &self.job)?; + db.progress_signal.notify(); + return Ok(true); + } + + let budget = db.opts.compaction_slice_bytes.max(1); + let mut processed = 0u64; + while self.merge.as_ref().unwrap().valid() { + let new_user_key = { + let merge = self.merge.as_ref().unwrap(); + !self.have_key || ikey_ukey(merge.ikey()) != self.cur_ukey.as_slice() + }; + if processed >= budget && new_user_key { + return Ok(false); } - let seq = ikey_seqno(ik); - let kind = ikey_kind(ik)?; - if seq > watermark { - // still visible to some possible snapshot: keep verbatim - (true, false) - } else if !kept_le_w { - kept_le_w = true; - // the newest version at-or-below the watermark: keep, unless - // it is a tombstone provably shadowing nothing older - let mut shadows_older = false; - if kind == ValueKind::Delete { - for r in job.older.iter() { - if r.may_contain_ukey(&cur_ukey)? { - shadows_older = true; - break; + + let entry_bytes = { + let merge = self.merge.as_ref().unwrap(); + (merge.ikey().len() as u64).saturating_add(merge.value().len() as u64) + }; + let (keep, is_ptr_drop) = { + let merge = self.merge.as_ref().unwrap(); + let ik = merge.ikey(); + let uk = ikey_ukey(ik); + if !self.have_key || uk != self.cur_ukey.as_slice() { + self.cur_ukey = uk.to_vec(); + self.have_key = true; + self.kept_le_w = false; + } + let seq = ikey_seqno(ik); + let kind = ikey_kind(ik)?; + if seq > self.watermark { + // still visible to some possible snapshot: keep verbatim + (true, false) + } else if !self.kept_le_w { + self.kept_le_w = true; + // the newest version at-or-below the watermark: keep, unless + // it is a tombstone provably shadowing nothing older + let mut shadows_older = false; + if kind == ValueKind::Delete { + for r in &self.job.older { + if r.may_contain_ukey(&self.cur_ukey)? { + shadows_older = true; + break; + } } } - } - if kind == ValueKind::Delete && !shadows_older { - (false, false) + if kind == ValueKind::Delete && !shadows_older { + (false, false) + } else { + (true, false) + } } else { - (true, false) + // shadowed by a kept newer version for every live snapshot + (false, kind == ValueKind::Put) } - } else { - // shadowed by a kept newer version for every live snapshot - (false, kind == ValueKind::Put) - } - }; + }; - if keep { - let ukey_changed_boundary = { - // fragments split only between user keys - match &builder { - Some((_, b)) => { - b.estimated_size() >= db.opts.target_file_size - && ikey_ukey(merge.ikey()) != b.last_ukey() + if keep { + let ukey_changed_boundary = { + // fragments split only between user keys + match &self.builder { + Some((_, b)) => { + b.estimated_size() >= db.opts.target_file_size + && ikey_ukey(self.merge.as_ref().unwrap().ikey()) != b.last_ukey() + } + None => false, } - None => false, + }; + if ukey_changed_boundary { + let (id, b) = self.builder.take().unwrap(); + self.tables.push(db.finish_table(id, b)?); + } + if self.builder.is_none() { + let id = db.alloc_file_id(); + let file = db.io.create_new(&db.paths.table(id))?; + self.builder = Some(( + id, + TableBuilder::new( + file, + db.opts.block_size, + db.opts.bloom_bits_per_key, + if self.job.bottom { + db.opts.bottom_compression.unwrap_or(db.opts.compression) + } else { + db.opts.compression + }, + ), + )); + } + let merge = self.merge.as_ref().unwrap(); + self.builder + .as_mut() + .unwrap() + .1 + .add(merge.ikey(), merge.value())?; + } else if is_ptr_drop { + if let ReprRef::Ptr(p) = decode_repr(self.merge.as_ref().unwrap().value())? { + *self.discard.entry(p.file).or_insert(0) += u64::from(p.len); } - }; - if ukey_changed_boundary { - let (id, b) = builder.take().unwrap(); - tables.push(db.finish_table(id, b)?); - } - if builder.is_none() { - let id = db.alloc_file_id(); - let file = db.io.create_new(&db.paths.table(id))?; - builder = Some(( - id, - TableBuilder::new( - file, - db.opts.block_size, - db.opts.bloom_bits_per_key, - if job.bottom { - db.opts.bottom_compression.unwrap_or(db.opts.compression) - } else { - db.opts.compression - }, - ), - )); - } - builder.as_mut().unwrap().1.add(merge.ikey(), merge.value())?; - } else if is_ptr_drop { - if let ReprRef::Ptr(p) = decode_repr(merge.value())? { - *discard.entry(p.file).or_insert(0) += u64::from(p.len); } + self.merge.as_mut().unwrap().next()?; + processed = processed.saturating_add(entry_bytes); } - merge.next()?; - } - if let Some((id, b)) = builder.take() { - tables.push(db.finish_table(id, b)?); + + if let Some((id, b)) = self.builder.take() { + self.tables.push(db.finish_table(id, b)?); + } + crate::io::sync_dir(&db.paths.dir)?; + + let output = if self.tables.is_empty() { + None + } else { + Some(Run { + id: self.run_id, + tables: std::mem::take(&mut self.tables), + }) + }; + install(db, &self.job, output, std::mem::take(&mut self.discard))?; + db.progress_signal.notify(); + Ok(true) } - crate::io::sync_dir(&db.paths.dir)?; +} - let output = if tables.is_empty() { - None - } else { - Some(Run { - id: run_id, - tables, - }) - }; +#[cfg(test)] +fn run_job(db: &Arc, job: Job) -> Result<()> { + let mut running = RunningJob::new(db, job)?; + while !running.run_slice(db)? {} + Ok(()) +} + +fn install_deepen(db: &Arc, job: &Job) -> Result<()> { + debug_assert!(matches!(job.kind, JobKind::Deepen)); + let input_ids: Vec = job.inputs.iter().map(|r| r.id).collect(); + + let mut manifest = db.manifest.lock(); + let mut data = manifest.data.clone(); + debug_assert_eq!(job.target, data.levels.len()); + data.levels[job.level].retain(|run| !input_ids.contains(&run.id)); + data.levels.push( + job.inputs + .iter() + .map(|run| RunMeta { + id: run.id, + table_ids: run.tables.iter().map(|table| table.id).collect(), + }) + .collect(), + ); + data.next_file_id = db.next_file_id.load(Ordering::SeqCst); + let gen = manifest.gen + 1; + manifest::save(&db.paths, gen, &data)?; + manifest.gen = gen; + manifest.data = data; - install(db, &job, output, discard)?; - db.progress_signal.notify(); + let mut state = db.state.write(); + let mut version = state.version.clone_shape(); + debug_assert_eq!(job.target, version.levels.len()); + version.levels[job.level].retain(|run| !input_ids.contains(&run.id)); + version.levels.push(job.inputs.clone()); + state.version = Arc::new(version); Ok(()) } @@ -360,10 +529,6 @@ fn install( data.levels[job.level].retain(|r| !input_ids.contains(&r.id)); match &job.kind { JobKind::Tier => { - // deepening: the target level may not exist yet - if job.target == data.levels.len() { - data.levels.push(Vec::new()); - } if let Some(run) = &output { data.levels[job.target].insert( 0, @@ -374,6 +539,7 @@ fn install( ); } } + JobKind::Deepen => unreachable!("deepening has no rewrite output"), JobKind::BottomSplice { base_id, keep_left, @@ -425,13 +591,11 @@ fn install( v.levels[job.level].retain(|r| !input_ids.contains(&r.id)); match job.kind { JobKind::Tier => { - if job.target == v.levels.len() { - v.levels.push(Vec::new()); - } if let Some(run) = output { v.levels[job.target].insert(0, run); } } + JobKind::Deepen => unreachable!("deepening has no rewrite output"), JobKind::BottomSplice { base_id, ref keep_left, @@ -750,11 +914,210 @@ mod shape_tests { l0_compaction_trigger: 2, tier_width: 2, max_levels: 2, + compaction_slice_bytes: 1, value_threshold: 4096, ..Options::default() } } + fn flush_run(db: &crate::Db, prefix: &str, count: u32) { + for i in 0..count { + db.put(format!("{prefix}/{i:06}"), vec![i as u8; 32]) + .unwrap(); + } + db.flush().unwrap(); + } + + #[test] + fn scheduler_preempts_a_sliced_deep_job_for_l0() { + let dir = tempfile::tempdir().unwrap(); + let mut opts = tiny_opts(); + opts.max_levels = 3; + opts.memtable_size = 16 << 20; + opts.l0_stall_trigger = 2; + opts.compaction_slice_bytes = 1; + let reopen_opts = opts.clone(); + let db = Arc::new(crate::Db::open(dir.path(), opts).unwrap()); + let inner = db.inner.clone(); + let hold = inner.compaction_mu.lock(); + + // Build two L1 runs, then begin (but do not finish) their L1->L2 job. + for generation in 0..2 { + flush_run(&db, &format!("old/{generation}/a"), 200); + flush_run(&db, &format!("old/{generation}/b"), 200); + let job = pick(&inner, false).expect("L0 tier job"); + assert_eq!(job.level, 0); + run_job(&inner, job).unwrap(); + } + let deep = pick(&inner, false).expect("L1 tier job"); + assert_eq!(deep.level, 1); + let mut scheduler = Scheduler::new(false); + scheduler.active = Some(RunningJob::new(&inner, deep).unwrap()); + assert!(scheduler.step(&inner).unwrap()); + assert_eq!(scheduler.active.as_ref().unwrap().job.level, 1); + + // Fill L0 to the write-stall threshold while the deep job is suspended. + flush_run(&db, "new/a", 40); + flush_run(&db, "new/b", 40); + assert_eq!(inner.state.read().version.levels[0].len(), 2); + + let (sent, received) = std::sync::mpsc::channel(); + let writer_db = db.clone(); + let writer = std::thread::spawn(move || { + sent.send(writer_db.put("writer/after-stall", "ok")) + .unwrap(); + }); + assert!(received + .recv_timeout(std::time::Duration::from_millis(50)) + .is_err()); + + // The next slice is L0, not another slice of the active deep job. + assert!(scheduler.step(&inner).unwrap()); + assert_eq!(scheduler.active.as_ref().unwrap().job.level, 0); + assert_eq!(scheduler.suspended.len(), 1); + for _ in 0..10_000 { + if inner.state.read().version.levels[0].is_empty() { + break; + } + assert!(scheduler.step(&inner).unwrap()); + } + received + .recv_timeout(std::time::Duration::from_secs(2)) + .expect("writer stayed stalled behind the deep job") + .unwrap(); + writer.join().unwrap(); + + while scheduler.step(&inner).unwrap() {} + assert_eq!( + db.get(b"old/1/b/000199").unwrap().as_deref(), + Some(&[199; 32][..]) + ); + assert_eq!( + db.get(b"writer/after-stall").unwrap().as_deref(), + Some(&b"ok"[..]) + ); + + drop(hold); + drop(inner); + drop(db); + let db = crate::Db::open(dir.path(), reopen_opts).unwrap(); + assert_eq!( + db.get(b"writer/after-stall").unwrap().as_deref(), + Some(&b"ok"[..]) + ); + } + + #[test] + fn scheduler_preempts_bottom_splice_without_losing_order() { + let dir = tempfile::tempdir().unwrap(); + let mut opts = tiny_opts(); + opts.max_levels = 2; + opts.memtable_size = 16 << 20; + opts.l0_stall_trigger = 2; + opts.compaction_slice_bytes = 1; + let reopen_opts = opts.clone(); + let db = Arc::new(crate::Db::open(dir.path(), opts).unwrap()); + let inner = db.inner.clone(); + let hold = inner.compaction_mu.lock(); + + db.put("victim", "alive").unwrap(); + flush_run(&db, "base/a", 200); + flush_run(&db, "base/b", 200); + let base = pick(&inner, false).expect("base L0 tier job"); + assert_eq!(base.level, 0); + run_job(&inner, base).unwrap(); + + db.delete("victim").unwrap(); + flush_run(&db, "upper/a", 200); + flush_run(&db, "upper/b", 200); + let upper = pick(&inner, false).expect("upper L0 tier job"); + assert_eq!(upper.level, 0); + run_job(&inner, upper).unwrap(); + + let splice = pick(&inner, false).expect("bottom splice job"); + assert!(matches!(splice.kind, JobKind::BottomSplice { .. })); + let mut scheduler = Scheduler::new(false); + scheduler.active = Some(RunningJob::new(&inner, splice).unwrap()); + assert!(scheduler.step(&inner).unwrap()); + + flush_run(&db, "newer/a", 40); + flush_run(&db, "newer/b", 40); + let (sent, received) = std::sync::mpsc::channel(); + let writer_db = db.clone(); + let writer = std::thread::spawn(move || { + sent.send(writer_db.put("writer/through-bottom", "ok")) + .unwrap(); + }); + assert!(received + .recv_timeout(std::time::Duration::from_millis(50)) + .is_err()); + + assert!(scheduler.step(&inner).unwrap()); + assert_eq!(scheduler.active.as_ref().unwrap().job.level, 0); + for _ in 0..10_000 { + if inner.state.read().version.levels[0].is_empty() { + break; + } + assert!(scheduler.step(&inner).unwrap()); + } + received + .recv_timeout(std::time::Duration::from_secs(2)) + .expect("writer stayed stalled behind the bottom splice") + .unwrap(); + writer.join().unwrap(); + while scheduler.step(&inner).unwrap() {} + + assert_eq!(db.get(b"victim").unwrap(), None); + assert_eq!( + db.get(b"newer/a/000000").unwrap().as_deref(), + Some(&[0; 32][..]) + ); + assert_eq!( + db.get(b"writer/through-bottom").unwrap().as_deref(), + Some(&b"ok"[..]) + ); + + drop(hold); + db.compact_all().unwrap(); + drop(inner); + drop(db); + let db = crate::Db::open(dir.path(), reopen_opts).unwrap(); + assert_eq!(db.get(b"victim").unwrap(), None); + assert_eq!( + db.get(b"writer/through-bottom").unwrap().as_deref(), + Some(&b"ok"[..]) + ); + } + + #[test] + fn deepening_moves_existing_tables_by_identity() { + let dir = tempfile::tempdir().unwrap(); + let mut opts = tiny_opts(); + opts.max_levels = 1; + opts.memtable_size = 16 << 20; + let db = crate::Db::open(dir.path(), opts).unwrap(); + let inner = db.inner.clone(); + let _hold = inner.compaction_mu.lock(); + + flush_run(&db, "first", 40); + flush_run(&db, "second", 40); + let before: Vec = inner.state.read().version.levels[0] + .iter() + .flat_map(|run| run.tables.iter().map(|table| table.id)) + .collect(); + let job = pick(&inner, false).expect("one-level tree must deepen"); + assert!(matches!(job.kind, JobKind::Deepen)); + run_job(&inner, job).unwrap(); + + let state = inner.state.read(); + assert!(state.version.levels[0].is_empty()); + let after: Vec = state.version.levels[1] + .iter() + .flat_map(|run| run.tables.iter().map(|table| table.id)) + .collect(); + assert_eq!(after, before, "deepening rewrote an existing table"); + } + /// Regression (review finding): a 3+-run bottom must read the NEWEST /// value after splicing. The broken version merged the FRONT run into /// the base, leaving middle runs positioned "newer" than newer data. diff --git a/crates/fluent31/src/config.rs b/crates/fluent31/src/config.rs index c981247..5a36f44 100644 --- a/crates/fluent31/src/config.rs +++ b/crates/fluent31/src/config.rs @@ -106,6 +106,11 @@ pub struct Options { /// Compaction output runs split into fragments of roughly this size, /// bounding per-file blooms/indexes and transient merge space. pub target_file_size: u64, + /// Input bytes a compaction may process before the scheduler rechecks + /// higher-priority levels. The current user key is always finished, so + /// one key with many versions may exceed this soft bound. Defaults to + /// 1 MiB. + pub compaction_slice_bytes: u64, /// Values >= this many bytes go to the value log; smaller stay inline in /// the LSM tree. 0 separates everything; usize::MAX disables separation. @@ -176,6 +181,7 @@ impl Default for Options { max_levels: 7, l0_stall_trigger: 12, target_file_size: 64 << 20, + compaction_slice_bytes: 1 << 20, value_threshold: 4096, vlog_file_size: 128 << 20, vlog_gc_ratio: 0.5, diff --git a/crates/fluent31/src/db.rs b/crates/fluent31/src/db.rs index 0a43fe8..5174dba 100644 --- a/crates/fluent31/src/db.rs +++ b/crates/fluent31/src/db.rs @@ -201,9 +201,9 @@ pub(crate) struct DbInner { /// writes happened — by seqno count, OR by the head having rotated /// (large-value workloads accrue garbage fast while seqnos crawl). pub gc_sampled_at: Mutex>, - /// Serializes compaction jobs: the maintenance thread and user-invoked - /// `compact_all` must never pick/merge concurrently (both would grab the - /// same input runs). + /// Serializes the compaction scheduler: background maintenance and + /// user-invoked `compact_all` share one priority picker. Rewrite jobs may + /// be suspended between bounded slices, but never run concurrently. pub compaction_mu: Mutex<()>, /// Replication stream subscribers (see the replication-surface section @@ -1581,7 +1581,9 @@ impl Db { self.inner.wait_flushed() } - /// Run compaction until no trigger fires (test/CLI helper). + /// Run compaction until no trigger fires (test/CLI helper). Manual work + /// uses the same bounded priority scheduler as automatic maintenance, so + /// newly eligible upper levels are serviced between deep-job slices. pub fn compact_all(&self) -> Result<()> { crate::compaction::compact_until_quiet(&self.inner) }