diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs index ef7803a4c..79ae713ad 100644 --- a/seaweed-volume/src/storage/needle_map.rs +++ b/seaweed-volume/src/storage/needle_map.rs @@ -15,8 +15,10 @@ use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; mod compact_map; pub mod file_pool; +mod idx_metric; pub mod sorted_file; use compact_map::CompactMap; +use idx_metric::metrics_from_idx; use sorted_file::SortedFileNeedleMap; use redb::{Database, Durability, ReadableDatabase, ReadableTable, TableDefinition}; @@ -182,12 +184,19 @@ impl NeedleMapKind { /// Trait for appending to an index file. pub trait IdxFileWriter: Write + Send + Sync { fn sync_all(&self) -> io::Result<()>; + /// Truncate the file to `len` bytes. Used to remove an orphan .idx row + /// left by a failed redb commit so `idx_file_offset` stays a contiguous + /// replay watermark. + fn truncate_to(&mut self, len: u64) -> io::Result<()>; } impl IdxFileWriter for std::fs::File { fn sync_all(&self) -> io::Result<()> { std::fs::File::sync_all(self) } + fn truncate_to(&mut self, len: u64) -> io::Result<()> { + self.set_len(len) + } } // ============================================================================ @@ -410,6 +419,14 @@ const NEEDLE_TABLE: TableDefinition = TableDefinition::new("needles" const META_TABLE: TableDefinition<&str, u64> = TableDefinition::new("meta"); const META_IDX_SIZE: &str = "idx_size"; +/// Writes between two durable redb checkpoints. Every non-durable commit +/// leaves an entry in redb's transaction tracker and pins the pages that +/// were on disk at the last durable commit; only a durable commit clears +/// both. Without a cadence they grow for the life of the process (#11179). +/// The map only counts; the volume takes the checkpoint (see +/// `RedbNeedleMap::checkpoint_due`) because the .dat must be flushed first. +const REDB_CHECKPOINT_INTERVAL: u32 = 1000; + /// Disk-backed needle map using redb. /// Low memory usage — data lives on disk behind a small, bounded redb page /// cache sized by `NeedleMapKind::redb_cache_bytes`. @@ -418,12 +435,15 @@ pub struct RedbNeedleMap { metric: NeedleMapMetric, idx_file: Option>, idx_file_offset: u64, + /// Puts/deletes since the last durable checkpoint. + writes_since_checkpoint: u32, } impl RedbNeedleMap { /// Begin a write transaction with `Durability::None` (no fsync). - /// The .idx file is the source of truth for crash recovery, so redb - /// is always rebuilt from .idx on startup — fsync is unnecessary. + /// The .idx file is the source of truth for crash recovery: a crash + /// loses at most the writes since the last checkpoint from redb, and + /// the next load replays them from .idx. fn begin_write_no_fsync(db: &Database) -> io::Result { let mut txn = db.begin_write().map_err(|e| { io::Error::new(io::ErrorKind::Other, format!("redb begin_write: {}", e)) @@ -432,6 +452,52 @@ impl RedbNeedleMap { Ok(txn) } + /// True once `REDB_CHECKPOINT_INTERVAL` puts/deletes have been committed + /// non-durably. The caller (the volume) must then flush the .dat and + /// call [`checkpoint`](Self::checkpoint): a checkpoint makes the index + /// durable, so the bytes it points at have to be on disk before it. + pub fn checkpoint_due(&self) -> bool { + self.writes_since_checkpoint >= REDB_CHECKPOINT_INTERVAL + } + + /// Make the table durable and record how much of the .idx it reflects. + /// Precondition: the .dat the index points into has been flushed. + /// + /// When `sync_idx` is true the .idx file is fsynced first — the recorded + /// size must never exceed what is on disk, or the reload would have to + /// rebuild from scratch. A caller that has already fsynced the .idx (e.g. + /// the volume's `flush_idx` on the fsync=true write path) may pass false + /// to avoid a redundant fsync. + pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> { + let txn = self.begin_checkpoint(sync_idx)?; + txn.commit() + .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)))?; + self.writes_since_checkpoint = 0; + Ok(()) + } + + /// Begin a durable transaction (fsync on commit) that also records how + /// much of the .idx the table reflects, so a reload replays only the + /// tail appended after it. When `sync_idx` is true the .idx is fsynced + /// first (see [`checkpoint`](Self::checkpoint)). + fn begin_checkpoint(&self, sync_idx: bool) -> io::Result { + if sync_idx { + self.sync()?; + } + let txn = self.db.begin_write().map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("redb begin_write: {}", e)) + })?; + if self.idx_file.is_some() { + let mut meta = txn.open_table(META_TABLE).map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("redb open meta: {}", e)) + })?; + meta.insert(META_IDX_SIZE, self.idx_file_offset).map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("redb insert meta: {}", e)) + })?; + } + Ok(txn) + } + /// Create a new redb-backed needle map at the given path. /// The database file will be created if it does not exist. /// `cache_bytes` bounds redb's page cache for this one database. @@ -461,6 +527,7 @@ impl RedbNeedleMap { metric: NeedleMapMetric::default(), idx_file: None, idx_file_offset: 0, + writes_since_checkpoint: 0, }) } @@ -501,49 +568,6 @@ impl RedbNeedleMap { } } - /// Rebuild metrics by scanning all entries in the redb table. - /// Called when reusing an existing .rdb without a full rebuild. - fn rebuild_metrics_from_db(&self, version: Version) -> io::Result<()> { - let txn = self - .db - .begin_read() - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb begin_read: {}", e)))?; - let table = txn - .open_table(NEEDLE_TABLE) - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb open_table: {}", e)))?; - let iter = table - .iter() - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb iter: {}", e)))?; - for entry in iter { - let (key_guard, val_guard) = entry.map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("redb iter next: {}", e)) - })?; - let key = NeedleId(key_guard.value()); - let bytes: &[u8] = val_guard.value(); - if bytes.len() == PACKED_NEEDLE_VALUE_SIZE { - let mut arr = [0u8; PACKED_NEEDLE_VALUE_SIZE]; - arr.copy_from_slice(bytes); - let nv = unpack_needle_value(&arr); - self.metric.maybe_set_max_file_key(key); - self.metric - .maybe_set_max_needle_end(nv.offset, nv.size, version); - if nv.size.is_valid() { - self.metric.file_count.fetch_add(1, Ordering::Relaxed); - self.metric - .file_byte_count - .fetch_add(nv.size.0 as u64, Ordering::Relaxed); - } else { - // Deleted entry (negative size) - self.metric.deletion_count.fetch_add(1, Ordering::Relaxed); - self.metric - .deletion_byte_count - .fetch_add((-nv.size.0) as u64, Ordering::Relaxed); - } - } - } - Ok(()) - } - /// Load from an .idx file, reusing an existing .rdb if it is consistent. /// /// Strategy: @@ -586,11 +610,12 @@ impl RedbNeedleMap { .open(db_path) .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb open: {}", e)))?; - let nm = RedbNeedleMap { + let mut nm = RedbNeedleMap { db, metric: NeedleMapMetric::default(), idx_file: None, idx_file_offset: 0, + writes_since_checkpoint: 0, }; let stored_idx_size = nm @@ -605,8 +630,10 @@ impl RedbNeedleMap { )); } - // Rebuild metrics from existing data - nm.rebuild_metrics_from_db(version)?; + // Counters come from the whole .idx history, never from the table, + // so the replay below is free to re-apply rows the table already + // holds (redb flushes on drop even without a checkpoint). + nm.metric = metrics_from_idx(reader, version)?; if stored_idx_size < idx_size { // .idx grew — replay new entries incrementally @@ -617,14 +644,12 @@ impl RedbNeedleMap { io::Error::new(io::ErrorKind::Other, format!("redb open_table: {}", e)) })?; idx::walk_index_file(reader, start_entry, |key, offset, size| { - nm.metric.maybe_set_max_needle_end(offset, size, version); let key_u64: u64 = key.into(); if offset.is_zero() || size.is_deleted() { - // Delete: look up old value for metric update, then - // store tombstone (negative size with original offset) + // Delete: store a tombstone (negative size, original + // offset) over a live value; already deleted is a no-op. if let Ok(Some(old)) = nm.get_via_table(&table, key_u64) { if old.size.is_valid() { - nm.metric.on_delete(&old); let deleted_nv = NeedleValue { offset: old.offset, size: Size(-(old.size.0)), @@ -639,14 +664,10 @@ impl RedbNeedleMap { } } } else { - // Put: look up old value for metric update - let old = nm.get_via_table(&table, key_u64).ok().flatten(); - let nv = NeedleValue { offset, size }; - let packed = pack_needle_value(&nv); + let packed = pack_needle_value(&NeedleValue { offset, size }); table.insert(key_u64, packed.as_slice()).map_err(|e| { io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)) })?; - nm.metric.on_put(key, old.as_ref(), size); } Ok(()) })?; @@ -695,12 +716,11 @@ impl RedbNeedleMap { cache_bytes: usize, ) -> io::Result { let _ = std::fs::remove_file(db_path); - let nm = RedbNeedleMap::new(db_path, cache_bytes)?; + let mut nm = RedbNeedleMap::new(db_path, cache_bytes)?; // Collect entries from idx file, resolving duplicates/deletions let mut entries: HashMap> = HashMap::new(); idx::walk_index_file(reader, 0, |key, offset, size| { - nm.metric.maybe_set_max_needle_end(offset, size, version); if offset.is_zero() || size.is_deleted() { entries.insert(key, None); } else { @@ -723,7 +743,6 @@ impl RedbNeedleMap { table.insert(key_u64, packed.as_slice()).map_err(|e| { io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)) })?; - nm.metric.on_put(*key, None, nv.size); } else { // Entry was deleted — remove from redb if present table.remove(key_u64).map_err(|e| { @@ -736,6 +755,7 @@ impl RedbNeedleMap { .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)))?; nm.save_idx_size_meta(idx_size)?; + nm.metric = metrics_from_idx(reader, version)?; Ok(nm) } @@ -755,10 +775,13 @@ impl RedbNeedleMap { /// Insert or update an entry. Writes to idx file first, then redb. pub fn put(&mut self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> { - // Persist to idx file BEFORE mutating redb state for crash consistency + // Persist to idx file BEFORE mutating redb state for crash consistency. + // The offset is advanced only after the redb commit succeeds: a failed + // commit leaves an orphan row in .idx that redb doesn't reflect, and + // advancing the offset here would let a later checkpoint record it as + // reflected, making the reload skip it permanently. if let Some(ref mut idx_file) = self.idx_file { idx::write_index_entry(idx_file, key, offset, size)?; - self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64; } let key_u64: u64 = key.into(); @@ -777,8 +800,14 @@ impl RedbNeedleMap { .insert(key_u64, packed.as_slice()) .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)))?; } - txn.commit() - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)))?; + if let Err(e) = txn.commit() { + self.truncate_idx_to_offset(); + return Err(io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e))); + } + if self.idx_file.is_some() { + self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64; + } + self.writes_since_checkpoint = self.writes_since_checkpoint.saturating_add(1); self.metric.on_put(key, old.as_ref(), size); Ok(()) @@ -826,13 +855,13 @@ impl RedbNeedleMap { if let Some(old) = self.get_internal(key_u64)? { if old.size.is_valid() { - // Persist tombstone to idx file BEFORE mutating redb + // Persist tombstone to idx file BEFORE mutating redb. The + // offset is advanced only after the redb commit succeeds + // (see put). if let Some(ref mut idx_file) = self.idx_file { idx::write_index_entry(idx_file, key, offset, TOMBSTONE_FILE_SIZE)?; - self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64; } - self.metric.on_delete(&old); let deleted_size = Size(-(old.size.0)); // Keep original offset so readDeleted can find original data (matching Go behavior) let deleted_nv = NeedleValue { @@ -850,10 +879,17 @@ impl RedbNeedleMap { io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)) })?; } - txn.commit().map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)) - })?; + if let Err(e) = txn.commit() { + self.truncate_idx_to_offset(); + return Err(io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e))); + } + if self.idx_file.is_some() { + self.idx_file_offset += NEEDLE_MAP_ENTRY_SIZE as u64; + } + self.writes_since_checkpoint = self.writes_since_checkpoint.saturating_add(1); + // Only now is the tombstone in the table the metrics describe. + self.metric.on_delete(&old); return Ok(Some(old.size)); } } @@ -900,9 +936,38 @@ impl RedbNeedleMap { Ok(()) } - /// Close index file. + /// Remove any .idx bytes past `idx_file_offset` — the orphan row left by + /// a failed redb commit. Without this the next successful write appends + /// after the orphan, `idx_file_offset` advances past it, and a later + /// checkpoint records an offset that makes the reload skip the orphan. + fn truncate_idx_to_offset(&mut self) { + if let Some(ref mut idx_file) = self.idx_file { + if let Err(e) = idx_file.truncate_to(self.idx_file_offset) { + tracing::warn!("failed to truncate orphan .idx row: {}", e); + } + } + } + + /// Close the index file, checkpointing first so a reload starts from + /// the recorded .idx size instead of replaying entries the table + /// already holds. pub fn close(&mut self) { - let _ = self.sync(); + if let Err(e) = self.checkpoint(true) { + tracing::warn!("redb checkpoint on close failed: {}", e); + } + self.idx_file = None; + } + + /// Sync the .idx and drop the writer without taking a durable checkpoint. + /// Used when the .dat flush failed: a checkpoint would make the index + /// durable with entries that may point past the unflushed .dat tail, so + /// the reload's max_needle_end check would mark the volume read-only. + /// Without the checkpoint, META_IDX_SIZE stays at the last successful one + /// and the reload replays the uncheckpointed tail (redb flushes on drop). + pub fn close_without_checkpoint(&mut self) { + if let Err(e) = self.sync() { + tracing::warn!("redb idx sync on close failed: {}", e); + } self.idx_file = None; } @@ -1068,6 +1133,27 @@ impl NeedleMap { } } + /// Whether the backend wants a durable checkpoint. Only the redb map + /// commits non-durably between checkpoints; see + /// `RedbNeedleMap::checkpoint_due`. + pub fn checkpoint_due(&self) -> bool { + match self { + NeedleMap::Redb(nm) => nm.checkpoint_due(), + NeedleMap::InMemory(_) | NeedleMap::SortedFile(_) => false, + } + } + + /// Take the checkpoint `checkpoint_due` asked for. The caller must have + /// flushed the .dat first. Pass `sync_idx = false` when the .idx has + /// already been fsynced (e.g. by `flush_idx` on the fsync=true path) to + /// avoid a redundant fsync. + pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> { + match self { + NeedleMap::Redb(nm) => nm.checkpoint(sync_idx), + NeedleMap::InMemory(_) | NeedleMap::SortedFile(_) => Ok(()), + } + } + /// Content byte count. pub fn content_size(&self) -> u64 { match self { @@ -1151,6 +1237,16 @@ impl NeedleMap { } } + /// Close without checkpointing — sync the .idx and drop the writer only. + /// See [`RedbNeedleMap::close_without_checkpoint`]. + pub fn close_without_checkpoint(&mut self) { + match self { + NeedleMap::InMemory(nm) => nm.close(), + NeedleMap::Redb(nm) => nm.close_without_checkpoint(), + NeedleMap::SortedFile(nm) => nm.close(), + } + } + /// Save to an index file. pub fn save_to_idx(&self, path: &str) -> io::Result<()> { match self { @@ -1198,6 +1294,29 @@ impl NeedleMap { // Tests // ============================================================================ +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + + /// The `.idx` size recorded in the durable state of the `.rdb` at + /// `rdb_path`, read from a copy taken while the map may still be open: + /// exactly what a crash would leave behind. `None` when nothing durable + /// has been recorded yet. + pub(crate) fn durable_idx_size(rdb_path: &Path) -> Option { + let copy = rdb_path.with_extension("crash-copy.rdb"); + std::fs::copy(rdb_path, ©).unwrap(); + let db = Database::open(©).unwrap(); + let txn = db.begin_read().unwrap(); + let meta = txn.open_table(META_TABLE).ok()?; + let size = meta.get(META_IDX_SIZE).unwrap().map(|g| g.value()); + drop(meta); + drop(txn); + drop(db); + let _ = std::fs::remove_file(©); + size + } +} + #[cfg(test)] mod tests { use super::*; @@ -1338,6 +1457,36 @@ mod tests { NeedleMapKind::Redb.redb_cache_bytes() } + /// Open a redb map on an empty .idx with an append writer attached, the + /// way `Volume::load_index_redb` opens a writable volume. + fn open_writable_redb( + dir: &std::path::Path, + ) -> (RedbNeedleMap, std::path::PathBuf, std::path::PathBuf) { + let db_path = dir.join("v.rdb"); + let idx_path = dir.join("v.idx"); + let idx_file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(&idx_path) + .unwrap(); + let idx_size = idx_file.metadata().unwrap().len(); + let mut reader = std::io::BufReader::new(&idx_file); + let mut nm = RedbNeedleMap::load_from_idx( + db_path.to_str().unwrap(), + &mut reader, + Version::current(), + redb_test_cache(), + ) + .unwrap(); + let writer = std::fs::OpenOptions::new() + .append(true) + .open(&idx_path) + .unwrap(); + nm.set_idx_file(Box::new(writer), idx_size); + (nm, db_path, idx_path) + } + #[test] fn test_redb_needle_map_put_get() { let dir = tempfile::tempdir().unwrap(); @@ -1462,7 +1611,11 @@ mod tests { assert!(nm.get(NeedleId(1)).unwrap().is_some()); assert!(nm.get(NeedleId(2)).unwrap().is_none()); // deleted and removed assert!(nm.get(NeedleId(3)).unwrap().is_some()); - assert_eq!(nm.file_count(), 2); + // Same history as test_needle_map_load_from_idx: the counters must + // match what the in-memory map (and a live volume) accumulates. + assert_eq!(nm.file_count(), 3); + assert_eq!(nm.deleted_count(), 1); + assert_eq!(nm.deleted_size(), 200); } #[test] @@ -1628,4 +1781,161 @@ mod tests { } assert_eq!(nm.file_count(), n as i64); } + + #[test] + fn test_redb_checkpoint_is_explicit_and_due_every_interval() { + use test_support::durable_idx_size; + + // Every non-durable redb commit leaves bookkeeping behind until a + // durable one clears it, so a writable map asks for a checkpoint on + // a fixed cadence. The map never takes it by itself: the volume has + // to flush the .dat first, then call checkpoint(). + const EXPECTED_INTERVAL: u64 = 1000; + let dir = tempfile::tempdir().unwrap(); + let (mut nm, db_path, _idx_path) = open_writable_redb(dir.path()); + for i in 1..EXPECTED_INTERVAL { + nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1)) + .unwrap(); + assert!(!nm.checkpoint_due(), "due after only {i} writes"); + } + nm.put( + NeedleId(EXPECTED_INTERVAL), + Offset::from_actual_offset((EXPECTED_INTERVAL * 8) as i64), + Size(1), + ) + .unwrap(); + assert!(nm.checkpoint_due()); + assert_eq!(durable_idx_size(&db_path), None, "put() must not commit durably"); + + nm.checkpoint(true).unwrap(); + assert!(!nm.checkpoint_due()); + assert_eq!( + durable_idx_size(&db_path), + Some(EXPECTED_INTERVAL * NEEDLE_MAP_ENTRY_SIZE as u64), + "checkpoint records how much of the .idx the table reflects" + ); + + // Snapshot the .rdb while the map is still open: what a crash leaves. + let crash_copy = dir.path().join("crash.rdb"); + std::fs::copy(&db_path, &crash_copy).unwrap(); + drop(nm); + let db = Database::open(&crash_copy).unwrap(); + let txn = db.begin_read().unwrap(); + let table = txn.open_table(NEEDLE_TABLE).unwrap(); + assert!( + table.get(EXPECTED_INTERVAL).unwrap().is_some(), + "entries up to the checkpoint are durable" + ); + } + + #[test] + fn test_redb_close_records_idx_size_so_reload_does_not_double_count() { + let dir = tempfile::tempdir().unwrap(); + let (mut nm, db_path, idx_path) = open_writable_redb(dir.path()); + for i in 1..=5u64 { + nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1)) + .unwrap(); + } + nm.close(); + drop(nm); + + // A clean close leaves the table durable; the recorded .idx size + // must match it, or the reload replays the same 5 entries on top. + let mut idx = std::fs::File::open(&idx_path).unwrap(); + let reloaded = RedbNeedleMap::load_from_idx( + db_path.to_str().unwrap(), + &mut idx, + Version::current(), + redb_test_cache(), + ) + .unwrap(); + assert_eq!(reloaded.file_count(), 5); + assert_eq!(reloaded.deleted_count(), 0); + } + + #[test] + fn test_redb_reload_with_stale_idx_size_does_not_double_count() { + let dir = tempfile::tempdir().unwrap(); + let (mut nm, db_path, idx_path) = open_writable_redb(dir.path()); + for i in 1..=5u64 { + nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1)) + .unwrap(); + } + // Drop without close(): redb makes the table durable on drop, but the + // recorded .idx size stays at its load-time value (0), so the reload + // replays all 5 entries over rows the table already holds. + drop(nm); + + let mut idx = std::fs::File::open(&idx_path).unwrap(); + let reloaded = RedbNeedleMap::load_from_idx( + db_path.to_str().unwrap(), + &mut idx, + Version::current(), + redb_test_cache(), + ) + .unwrap(); + assert_eq!(reloaded.file_count(), 5); + assert_eq!(reloaded.deleted_count(), 0); + assert_eq!( + reloaded.get(NeedleId(5)).unwrap().unwrap().offset, + Offset::from_actual_offset(40) + ); + } + + #[test] + fn test_redb_reload_metrics_keep_overwrite_and_delete_history() { + // garbage_level() is deleted_size / content_size. Every load path + // must rebuild both from the whole .idx history, the way the live + // counters accumulate, not from the table's final state, or the + // bytes of overwritten and deleted needles stop counting as garbage + // after a restart. + for (close_first, rebuild) in [(true, false), (false, false), (true, true)] { + let dir = tempfile::tempdir().unwrap(); + let (mut nm, db_path, idx_path) = open_writable_redb(dir.path()); + nm.put(NeedleId(1), Offset::from_actual_offset(8), Size(100)) + .unwrap(); + // Overwrite: the first 100 bytes become garbage. + nm.put(NeedleId(1), Offset::from_actual_offset(200), Size(200)) + .unwrap(); + nm.put(NeedleId(2), Offset::from_actual_offset(500), Size(50)) + .unwrap(); + nm.delete(NeedleId(2), Offset::from_actual_offset(600)) + .unwrap(); + let live = ( + nm.file_count(), + nm.content_size(), + nm.deleted_count(), + nm.deleted_size(), + ); + assert_eq!(live, (3, 350, 2, 150)); + if close_first { + nm.close(); + } + drop(nm); + if rebuild { + std::fs::remove_file(&db_path).unwrap(); + } + + let mut idx = std::fs::File::open(&idx_path).unwrap(); + let reloaded = RedbNeedleMap::load_from_idx( + db_path.to_str().unwrap(), + &mut idx, + Version::current(), + redb_test_cache(), + ) + .unwrap(); + let after = ( + reloaded.file_count(), + reloaded.content_size(), + reloaded.deleted_count(), + reloaded.deleted_size(), + ); + assert_eq!( + after, live, + "close_first={close_first} rebuild={rebuild}" + ); + assert_eq!(reloaded.get(NeedleId(1)).unwrap().unwrap().size, Size(200)); + assert!(reloaded.get(NeedleId(2)).unwrap().map_or(true, |v| v.size.is_deleted())); + } + } } diff --git a/seaweed-volume/src/storage/needle_map/idx_metric.rs b/seaweed-volume/src/storage/needle_map/idx_metric.rs new file mode 100644 index 000000000..67cd9c368 --- /dev/null +++ b/seaweed-volume/src/storage/needle_map/idx_metric.rs @@ -0,0 +1,161 @@ +//! Rebuild a needle map's counters from the whole `.idx` history. +//! +//! The volume's garbage ratio is `deleted_size / content_size`, and both are +//! additive over the life of the volume: an overwritten needle keeps its +//! bytes in `content_size` and adds them to `deleted_size`. A backend whose +//! table only holds the final value per key (redb) cannot recover that from +//! the table, so on load the counters come from the `.idx` file instead. +//! +//! This mirrors Go's `needleMapMetricFromIndexFile`: walk the index newest +//! entry first with a bloom filter of the keys already seen, so the memory +//! cost is a few bits per entry instead of a map of every key. The counting +//! rule reproduces what the live `on_put`/`on_delete` path accumulates: +//! +//! - every live entry is one put: `file_count`, `file_byte_count`; +//! - a live entry with a newer entry for the same key was overwritten or +//! deleted later, so it is also one deletion: `deletion_count`, +//! `deletion_byte_count`; +//! - a tombstone only marks its key as seen. +//! +//! A bloom false positive (0.1%) can only add a spurious deletion, which +//! over-reports garbage slightly; it never hides any. + +use std::io::{self, Read, Seek, SeekFrom}; +use std::sync::atomic::Ordering; + +use xxhash_rust::xxh64::xxh64; + +use super::NeedleMapMetric; +use crate::storage::types::*; + +/// Entries read per batch while walking backwards (64 KiB of index). +const BATCH_ENTRIES: usize = 4096; +/// Same target false-positive rate as the Go server's filter. +const FALSE_POSITIVE_RATE: f64 = 0.001; + +/// Minimal bloom filter over needle ids, double hashing with xxh64. +struct SeenKeys { + bits: Vec, + bit_count: u64, + hashes: u64, +} + +impl SeenKeys { + fn new(expected: u64, false_positive_rate: f64) -> Self { + let n = expected.max(1) as f64; + let ln2 = std::f64::consts::LN_2; + let bit_count = (-(n * false_positive_rate.ln()) / (ln2 * ln2)) + .ceil() + .max(64.0) as u64; + let hashes = ((bit_count as f64 / n) * ln2).round().clamp(1.0, 16.0) as u64; + SeenKeys { + bits: vec![0u64; bit_count.div_ceil(64) as usize], + bit_count, + hashes, + } + } + + /// Whether `key` was (probably) seen before; marks it seen either way. + fn test_and_add(&mut self, key: u64) -> bool { + let bytes = key.to_le_bytes(); + let h1 = xxh64(&bytes, 0); + let h2 = xxh64(&bytes, 0x9E37_79B9_7F4A_7C15) | 1; + let mut seen = true; + for i in 0..self.hashes { + let bit = h1.wrapping_add(i.wrapping_mul(h2)) % self.bit_count; + let word = (bit / 64) as usize; + let mask = 1u64 << (bit % 64); + if self.bits[word] & mask == 0 { + seen = false; + self.bits[word] |= mask; + } + } + seen + } +} + +/// Walk `reader` (an `.idx` file) newest entry first and return the counters +/// a live volume would hold after applying the same history. A torn partial +/// entry at the tail is ignored, as `walk_index_file` does. The reader is +/// left positioned at the start of the file. +pub(super) fn metrics_from_idx( + reader: &mut R, + version: Version, +) -> io::Result { + let metric = NeedleMapMetric::default(); + let file_size = reader.seek(SeekFrom::End(0))?; + let entry_count = file_size / NEEDLE_MAP_ENTRY_SIZE as u64; + let mut seen = SeenKeys::new(entry_count, FALSE_POSITIVE_RATE); + let mut buf = vec![0u8; NEEDLE_MAP_ENTRY_SIZE * BATCH_ENTRIES]; + + let mut remaining = entry_count; + while remaining > 0 { + let batch = remaining.min(BATCH_ENTRIES as u64) as usize; + let first_entry = remaining - batch as u64; + let len = batch * NEEDLE_MAP_ENTRY_SIZE; + reader.seek(SeekFrom::Start(first_entry * NEEDLE_MAP_ENTRY_SIZE as u64))?; + reader.read_exact(&mut buf[..len])?; + for i in (0..batch).rev() { + let entry = &buf[i * NEEDLE_MAP_ENTRY_SIZE..(i + 1) * NEEDLE_MAP_ENTRY_SIZE]; + let (key, offset, size) = idx_entry_from_bytes(entry); + metric.maybe_set_max_needle_end(offset, size, version); + let superseded = seen.test_and_add(key.into()); + if offset.is_zero() || size.is_deleted() { + // Tombstone: reserves no bytes, only marks the key as seen. + continue; + } + metric.maybe_set_max_file_key(key); + metric.file_count.fetch_add(1, Ordering::Relaxed); + metric + .file_byte_count + .fetch_add(size.0 as u64, Ordering::Relaxed); + if superseded && size.0 > 0 { + metric.deletion_count.fetch_add(1, Ordering::Relaxed); + metric + .deletion_byte_count + .fetch_add(size.0 as u64, Ordering::Relaxed); + } + } + remaining = first_entry; + } + + reader.seek(SeekFrom::Start(0))?; + Ok(metric) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_seen_keys_reports_repeats_and_not_fresh_keys() { + let mut seen = SeenKeys::new(10_000, FALSE_POSITIVE_RATE); + // Fresh keys may occasionally collide (that is the false-positive + // rate), but only rarely. + let fresh_reported_seen = (0..10_000u64) + .filter(|&key| seen.test_and_add(key)) + .count(); + assert!( + fresh_reported_seen < 50, + "fresh keys reported seen: {fresh_reported_seen}" + ); + // A repeated key is never reported fresh: no false negatives. + for key in 0..10_000u64 { + assert!(seen.test_and_add(key), "repeated key {key} reported fresh"); + } + // Over 100k never-inserted keys the false-positive rate stays near + // the 0.1% target; allow a generous margin. + let false_positives = (1_000_000..1_100_000u64) + .filter(|k| { + let bytes = k.to_le_bytes(); + let h1 = xxh64(&bytes, 0); + let h2 = xxh64(&bytes, 0x9E37_79B9_7F4A_7C15) | 1; + (0..seen.hashes).all(|i| { + let bit = h1.wrapping_add(i.wrapping_mul(h2)) % seen.bit_count; + seen.bits[(bit / 64) as usize] & (1u64 << (bit % 64)) != 0 + }) + }) + .count(); + assert!(false_positives < 500, "false positives: {false_positives}"); + } +} diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index fc6800adf..3963d373f 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -1874,10 +1874,45 @@ impl Volume { self.last_modified_ts_seconds = n.last_modified; } + self.maybe_checkpoint_index(fsync); + // Return Size(n.DataSize) as the logical size, matching Go's doWriteRequest Ok((offset, Size(n.data_size as i32), false)) } + /// Take the index checkpoint the needle map asked for, data first: the + /// checkpoint makes the index durable, and an index row that outlives + /// the bytes it points at loads read-only (see load()). A failed .dat + /// flush therefore skips the checkpoint; the map keeps asking, so it is + /// retried on the next write. + /// + /// When `idx_already_synced` is true the .idx has already been fsynced by + /// `flush_idx` on the fsync=true write path, so the checkpoint skips its + /// own .idx fsync to avoid a redundant one. + fn maybe_checkpoint_index(&mut self, idx_already_synced: bool) { + let due = self.nm.as_ref().is_some_and(|nm| nm.checkpoint_due()); + if !due { + return; + } + if let Err(e) = self.flush_dat() { + self.check_read_write_error(Some(&e)); + tracing::warn!( + "volume {}: skipping index checkpoint, .dat flush failed: {}", + self.id.0, + e + ); + return; + } + let checkpointed = match self.nm.as_mut() { + Some(nm) => nm.checkpoint(!idx_already_synced), + None => Ok(()), + }; + if let Err(e) = checkpointed { + self.check_read_write_error(Some(&e)); + tracing::warn!("volume {}: index checkpoint failed: {}", self.id.0, e); + } + } + fn read_needle_header_unlocked(&self, n: &mut Needle, offset: i64) -> Result<(), VolumeError> { let mut header = [0u8; NEEDLE_HEADER_SIZE]; self.read_exact_at_backend(&mut header, offset as u64)?; @@ -2014,6 +2049,7 @@ impl Volume { if let Some(nm) = &mut self.nm { nm.delete(n.id, Offset::from_actual_offset(offset as i64))?; } + self.maybe_checkpoint_index(false); Ok(size) } @@ -2726,6 +2762,7 @@ impl Volume { if let Some(ref mut nm) = self.nm { nm.put(key, offset, size).map_err(VolumeError::Io)?; } + self.maybe_checkpoint_index(false); Ok(()) } @@ -3360,6 +3397,7 @@ impl Volume { if let Some(ref mut nm) = self.nm { nm.put(needle_id, offset, size)?; } + self.maybe_checkpoint_index(false); Ok(()) } @@ -3967,13 +4005,29 @@ impl Volume { } pub fn close(&mut self) { - if let Some(ref dat_file) = self.dat_file { - let _ = dat_file.sync_all(); - } + let dat_synced = if let Some(ref dat_file) = self.dat_file { + dat_file.sync_all().is_ok() + } else { + true + }; self.dat_file = None; self.remote_dat_file = None; - if let Some(ref nm) = self.nm { - let _ = nm.sync(); + // When the .dat flushed, checkpoint the index so the next load starts + // from the recorded .idx size. When it did not, skip the checkpoint: + // a durable index pointing past an unflushed .dat tail would make the + // reload's max_needle_end check mark the volume read-only. Without the + // checkpoint, META_IDX_SIZE stays at the last successful one and the + // reload replays the uncheckpointed tail (redb flushes on drop). + if let Some(ref mut nm) = self.nm { + if dat_synced { + nm.close(); + } else { + tracing::warn!( + "volume {}: .dat sync failed on close, skipping index checkpoint", + self.id.0 + ); + nm.close_without_checkpoint(); + } } self.nm = None; } @@ -5507,6 +5561,106 @@ mod tests { assert_eq!(std::str::from_utf8(&n.data).unwrap(), "data 2"); } + #[test] + fn test_redb_volume_close_then_reload_keeps_counters_exact() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let open = || { + Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::Redb, + None, + None, + 0, + Version::current(), + ) + .unwrap() + }; + + { + let mut v = open(); + for i in 1..=3 { + let data = format!("data {}", i); + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: data.as_bytes().to_vec(), + data_size: data.len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + } + // Volume::close is the shutdown path (store -> disk location -> + // volume). It must checkpoint the redb index, or the reload + // replays rows the table already holds and inflates the counters. + v.close(); + } + + let v = open(); + assert_eq!(v.file_count(), 3); + assert_eq!(v.deleted_count(), 0); + let mut n = Needle { + id: NeedleId(2), + ..Needle::default() + }; + v.read_needle(&mut n).unwrap(); + assert_eq!(std::str::from_utf8(&n.data).unwrap(), "data 2"); + } + + #[test] + fn test_redb_volume_checkpoint_flushes_dat_before_index() { + use crate::storage::needle_map::test_support::durable_idx_size; + + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::Redb, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + let rdb_path = std::path::PathBuf::from(v.file_name(".rdb")); + let write = |v: &mut Volume, i: u64| { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(1), + data: b"x".to_vec(), + data_size: 1, + ..Needle::default() + }; + // fsync=false: the .dat append is not flushed by the write itself. + v.write_needle(&mut n, true, false).unwrap(); + }; + for i in 1..1000 { + write(&mut v, i); + } + assert_eq!(durable_idx_size(&rdb_path), None); + + // The 1000th write makes an index checkpoint due. A checkpoint makes + // the index durable, so the volume has to flush the .dat first, and + // when that flush fails nothing may be checkpointed: a durable index + // row pointing past the end of an unflushed .dat loads read-only. + v.fail_next_fsync_for_test(true); + write(&mut v, 1000); + assert_eq!(durable_idx_size(&rdb_path), None); + + v.fail_next_fsync_for_test(false); + write(&mut v, 1001); + assert_eq!( + durable_idx_size(&rdb_path), + Some(1001 * NEEDLE_MAP_ENTRY_SIZE as u64) + ); + } + #[test] fn test_relocate_index_to_moves_index_and_serves_reads() { let root = TempDir::new().unwrap();