From 92390a733cf7bf180c099baaa8c5413bc2bb684e Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 4 Jul 2026 21:50:18 +0700 Subject: [PATCH] =?UTF-8?q?fix(storage):=20cold-tier=20correctness=20+=20r?= =?UTF-8?q?eliability=20=E2=80=94=20DEL/FLUSH=20resurrection,=20expired-re?= =?UTF-8?q?ad=20leak,=20crash=20durability,=20orphan=20sweep,=20liveness?= =?UTF-8?q?=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five fixes for the disk-offload (cold tier) path, from the offload architecture review (tmp/OFFLOAD-COMPRESSION-REVIEW.md), each red/green TDD-proven: D1 — DEL/UNLINK/FLUSH cold resurrection (P1 bug). Database::remove and clear() never touched the ColdIndex, so deleting a spilled key left its index entry alive and the next GET resurrected the deleted value from the .mpf heap file; DEL even returned 0 for cold-only keys. remove()/clear() now drop cold entries (queueing file unlinks), and DEL/UNLINK use a new remove_counting_cold() so cold-only keys count as removed. R1 — expired cold reads reclaim their index entry. cold_read returned a bare Option, so an expired-on-disk entry left its index entry + file refcount leaked forever. New ColdReadOutcome{Hit,Expired,Miss} lets the Database::get read-through remove the index entry on Expired only — transient I/O errors (Miss) never drop a key. D3 — directory fsync after spill publication. Spill writes fsynced the file but never data/, so a power loss could vanish the directory entry of a file the (dir-fsynced) manifest references. Both the batch (tmp+rename) and single-file spill paths now fsync the data directory. R3 — startup sweep of crash-orphaned heap files. A crash between spill write and manifest commit leaked unregistered heap-*.mpf/.tmp files forever (invisible to the cold index). Recovery now unlinks heap files not registered in the manifest — only after the manifest opened successfully, so a corrupt-manifest signal can never trigger deletion. R5 — spill-thread liveness metrics. A silently-dead spill thread was observable only as unbounded eviction backlog. The spill loop now stamps a heartbeat and counts flushed batches; INFO persistence exposes spill_batches_flushed, spill_completions_dropped, spill_last_heartbeat_ms. Tests: 5 new unit tests (3 in cold_read, 1 in kv_spill, 1 in spill_thread), all red before the fix; full lib suite 3658 pass; clippy clean on default + tokio,jemalloc feature sets. author: Tin Dang --- CHANGELOG.md | 31 +++++ src/command/connection.rs | 8 +- src/command/key.rs | 11 +- src/persistence/recovery.rs | 11 ++ src/storage/db.rs | 75 ++++++++--- src/storage/tiered/cold_index.rs | 20 ++- src/storage/tiered/cold_read.rs | 193 ++++++++++++++++++++++++++--- src/storage/tiered/kv_spill.rs | 101 +++++++++++++++ src/storage/tiered/spill_thread.rs | 67 ++++++++++ 9 files changed, 478 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5831ffaf..104ca6b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — cold-tier (disk offload) correctness & reliability (PR #TBD) + +- **DEL/UNLINK/FLUSHALL now reach the cold tier** — deleting a key whose value + had been offloaded to disk left its `ColdIndex` entry alive, so the next GET + resurrected the deleted value from the `.mpf` heap file (`DEL` even returned + 0 for cold-only keys). `Database::remove`/`clear` now drop the cold-index + entry (and queue file unlinks) alongside the hot entry, and `DEL`/`UNLINK` + count cold-only keys correctly. +- **Expired cold reads reclaim their index entry** — a cold read that found an + expired entry returned nothing but left the index entry + file refcount + behind forever (nothing else ever reclaims them). The read path now + distinguishes `Expired` from `Miss` and removes the index entry on expiry; + transient I/O errors still leave the entry alone. +- **Spill files survive a crash at the directory level** — spill writes fsynced + the file but never the directory, so after a power loss the (dir-fsynced) + manifest could reference a heap file whose directory entry vanished. Both the + batch (tmp+rename) and single-file spill paths now fsync `data/` after + publishing the file. +- **Crash-orphaned heap files are swept at startup** — a crash between the + spill write and the manifest commit left `heap-*.mpf`/`.tmp` files on disk + that no manifest references (invisible to the cold index, leaked disk + forever). Recovery now unlinks unregistered heap files once the manifest has + opened successfully. + +### Added + +- **Spill-thread liveness metrics in `INFO persistence`** — + `spill_batches_flushed`, `spill_completions_dropped`, and + `spill_last_heartbeat_ms` (0 = never ran) expose a silently-dead spill + thread, whose only prior symptom was an unbounded eviction backlog. + ## [0.5.1] — 2026-07-04 ### Fixed diff --git a/src/command/connection.rs b/src/command/connection.rs index 052ea657..79419353 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -203,7 +203,10 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { rdb_last_bgsave_status:{}\r\n\ aof_enabled:0\r\n\ aof_rewrite_in_progress:0\r\n\ - aof_backpressure_dropped:{}\r\n", + aof_backpressure_dropped:{}\r\n\ + spill_batches_flushed:{}\r\n\ + spill_completions_dropped:{}\r\n\ + spill_last_heartbeat_ms:{}\r\n", if crate::command::persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed) { 1 @@ -220,6 +223,9 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { }, crate::persistence::aof::AOF_BACKPRESSURE_DROPPED .load(std::sync::atomic::Ordering::Relaxed), + crate::storage::tiered::spill_thread::spill_batches_flushed_total(), + crate::storage::tiered::spill_thread::spill_completion_dropped_total(), + crate::storage::tiered::spill_thread::spill_last_heartbeat_ms(), )); sections.push_str("\r\n"); diff --git a/src/command/key.rs b/src/command/key.rs index 685da0ca..463f7510 100644 --- a/src/command/key.rs +++ b/src/command/key.rs @@ -35,7 +35,10 @@ pub fn del(db: &mut Database, args: &[Frame]) -> Frame { let mut count: i64 = 0; for arg in args { if let Some(key) = extract_key(arg) { - if db.remove(key).is_some() { + // Counting variant: a spilled (cold-only) key logically exists + // and must count as removed (D1). + let (removed, _hot) = db.remove_counting_cold(key); + if removed { count += 1; } } @@ -901,8 +904,12 @@ pub fn unlink(db: &mut Database, args: &[Frame]) -> Frame { let mut count: i64 = 0; for arg in args { if let Some(key) = extract_key(arg) { - if let Some(entry) = db.remove(key) { + // Counting variant: cold-only keys count as removed (D1). + let (removed, hot) = db.remove_counting_cold(key); + if removed { count += 1; + } + if let Some(entry) = hot { if should_async_drop(&entry) { // Async drop for large collections: spawn a blocking // task to avoid holding the event loop. diff --git a/src/persistence/recovery.rs b/src/persistence/recovery.rs index a57fc2a3..8b0c1a59 100644 --- a/src/persistence/recovery.rs +++ b/src/persistence/recovery.rs @@ -331,6 +331,17 @@ pub fn recover_shard_v3_pitr( ); result.cold_index = Some(cold_idx); } + // Crash-orphan sweep: heap files written but never registered in + // the manifest (crash between spill write and manifest commit) + // leak disk forever otherwise. Manifest opened OK — safe to sweep. + let swept = + crate::storage::tiered::kv_spill::sweep_orphan_heap_files(shard_dir, &manifest); + if swept > 0 { + info!( + "Shard {}: swept {} crash-orphaned heap file(s)", + shard_id, swept + ); + } } } diff --git a/src/storage/db.rs b/src/storage/db.rs index c5254732..95f81c30 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -707,31 +707,43 @@ impl Database { None } KeyState::Absent => { + use crate::storage::tiered::cold_read::ColdReadOutcome; // Cold fallback: read from disk DataFile via cold_read helper. // Extract owned result first to drop immutable borrows before mutation. let cold_result = self.cold_shard_dir.as_ref().and_then(|shard_dir| { - self.cold_index.as_ref().and_then(|ci| { - crate::storage::tiered::cold_read::cold_read_through( + self.cold_index.as_ref().map(|ci| { + crate::storage::tiered::cold_read::cold_read_through_outcome( ci, shard_dir, key, now_ms, ) }) }); - if let Some((redis_value, ttl_ms)) = cold_result { - let key_bytes = Bytes::copy_from_slice(key); - // Build an entry from the RedisValue (works for strings and collections) - let mut entry = Entry::new_string(Bytes::new()); // placeholder - entry.value = - crate::storage::compact_value::CompactValue::from_redis_value(redis_value); - if let Some(ttl) = ttl_ms { - entry.set_expires_at_ms(self.base_timestamp, ttl); + match cold_result { + Some(ColdReadOutcome::Hit(redis_value, ttl_ms)) => { + let key_bytes = Bytes::copy_from_slice(key); + // Build an entry from the RedisValue (works for strings and collections) + let mut entry = Entry::new_string(Bytes::new()); // placeholder + entry.value = crate::storage::compact_value::CompactValue::from_redis_value( + redis_value, + ); + if let Some(ttl) = ttl_ms { + entry.set_expires_at_ms(self.base_timestamp, ttl); + } + self.set(key_bytes, entry); + if let Some(ref mut ci) = self.cold_index { + ci.remove(key); + } + self.data.get(key) } - self.set(key_bytes, entry); - if let Some(ref mut ci) = self.cold_index { - ci.remove(key); + Some(ColdReadOutcome::Expired) => { + // Expired on disk: reclaim the index entry now, or it leaks + // (the orphan sweep only checks hot-shadowing, never TTL). + if let Some(ref mut ci) = self.cold_index { + ci.remove(key); + } + None } - return self.data.get(key); + Some(ColdReadOutcome::Miss) | None => None, } - None } } } @@ -822,6 +834,12 @@ impl Database { self.used_memory = 0; self.maybe_has_expiring_keys = false; self.hot_keys.clear(); + // D1: FLUSH must clear the cold tier too, or flushed keys stay + // readable via cold read-through. Files are queued for unlink and + // reclaimed by the orphan sweep (which holds the manifest handle). + if let Some(ci) = self.cold_index.as_mut() { + ci.clear_all(); + } } /// The hot-key sketch (sampling hooks, HOTKEYS command, coordinator @@ -880,7 +898,34 @@ impl Database { } /// Remove a key and return its entry. No expiry check needed (DEL removes regardless). + /// + /// Also removes any COLD (spilled) copy of the key — without this, a DEL + /// of an evicted key is a no-op and the next GET resurrects the value via + /// cold read-through (D1, tmp/OFFLOAD-COMPRESSION-REVIEW.md). A cold-only + /// key returns `None` (no in-RAM entry exists); callers that must COUNT + /// cold-only removals (DEL/UNLINK) use [`Self::remove_counting_cold`]. pub fn remove(&mut self, key: &[u8]) -> Option { + let _ = self.remove_cold_only(key); + self.remove_hot(key) + } + + /// Remove hot + cold copies; returns `true` when EITHER existed, so + /// DEL/UNLINK count spilled keys as removed (Redis semantics: the key + /// logically exists). The removed hot entry, when present, is also + /// returned so UNLINK can size its async-drop decision. + pub fn remove_counting_cold(&mut self, key: &[u8]) -> (bool, Option) { + let had_cold = self.remove_cold_only(key); + let hot = self.remove_hot(key); + (hot.is_some() || had_cold, hot) + } + + #[inline] + fn remove_cold_only(&mut self, key: &[u8]) -> bool { + self.cold_index.as_mut().is_some_and(|ci| ci.remove(key)) + } + + #[inline] + fn remove_hot(&mut self, key: &[u8]) -> Option { if let Some(entry) = self.data.remove(key) { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); Some(entry) diff --git a/src/storage/tiered/cold_index.rs b/src/storage/tiered/cold_index.rs index d4ad71e8..ae9f96b7 100644 --- a/src/storage/tiered/cold_index.rs +++ b/src/storage/tiered/cold_index.rs @@ -110,16 +110,32 @@ impl ColdIndex { } } - /// Remove a key from the cold index (e.g., when promoted back to RAM). + /// Remove a key from the cold index (promotion back to RAM, DEL/UNLINK, + /// expired-on-read reclaim). Returns `true` when the key was present. /// /// Decrements the backing file's live-ref count; if this removes the file's /// last referrer, the `file_id` is queued for unlink by the next sweep. - pub fn remove(&mut self, key: &[u8]) { + pub fn remove(&mut self, key: &[u8]) -> bool { if let Some(old) = self.map.remove(key) { if self.ref_dec(old.file_id) { self.pending_unlink.push(old.file_id); } + true + } else { + false + } + } + + /// Drop EVERY entry and queue every backing file for unlink (FLUSHDB / + /// FLUSHALL — D1: flushed keys must not stay readable from disk). The + /// files themselves are removed by the next orphan sweep, which holds + /// the manifest handle this method deliberately does not need. + pub fn clear_all(&mut self) { + self.map.clear(); + for (&file_id, _) in self.file_refs.iter() { + self.pending_unlink.push(file_id); } + self.file_refs.clear(); } /// Look up a key's cold location. diff --git a/src/storage/tiered/cold_read.rs b/src/storage/tiered/cold_read.rs index 3e12e259..94a33bee 100644 --- a/src/storage/tiered/cold_read.rs +++ b/src/storage/tiered/cold_read.rs @@ -13,18 +13,48 @@ use crate::persistence::kv_page::{ValueType, entry_flags, read_overflow_chain}; use crate::persistence::page::PAGE_4K; use crate::storage::entry::RedisValue; +/// Outcome of a cold read, distinguishing EXPIRED from plain miss so the +/// caller can reclaim the index entry (R1: expired cold entries used to leak +/// their index entry + file refcount forever — nothing else ever reclaims +/// them; the orphan sweep only checks hot-shadowing). +pub enum ColdReadOutcome { + /// Entry found and alive. + Hit(RedisValue, Option), + /// Entry found but its TTL has passed — caller must remove the index entry. + Expired, + /// Not found / file unreadable / corrupt. The index entry is left alone: + /// a transient I/O error must not permanently drop the key. + Miss, +} + /// Attempt to read a cold KV entry from disk. /// /// Returns `Some((RedisValue, ttl_ms))` on hit, `None` on miss/expired/error. /// The caller is responsible for promoting the entry back to the DashTable -/// and removing it from the cold index. +/// and removing it from the cold index. Callers that can reclaim expired +/// entries should prefer [`cold_read_through_outcome`]. pub fn cold_read_through( cold_index: &ColdIndex, shard_dir: &Path, key: &[u8], now_ms: u64, ) -> Option<(RedisValue, Option)> { - let location = cold_index.lookup(key)?; + match cold_read_through_outcome(cold_index, shard_dir, key, now_ms) { + ColdReadOutcome::Hit(v, ttl) => Some((v, ttl)), + ColdReadOutcome::Expired | ColdReadOutcome::Miss => None, + } +} + +/// Outcome-aware variant of [`cold_read_through`] (R1 reclaim path). +pub fn cold_read_through_outcome( + cold_index: &ColdIndex, + shard_dir: &Path, + key: &[u8], + now_ms: u64, +) -> ColdReadOutcome { + let Some(location) = cold_index.lookup(key) else { + return ColdReadOutcome::Miss; + }; read_cold_entry(shard_dir, location, now_ms) } @@ -37,32 +67,39 @@ pub fn read_cold_entry_at( location: ColdLocation, now_ms: u64, ) -> Option<(RedisValue, Option)> { - read_cold_entry(shard_dir, location, now_ms) + match read_cold_entry(shard_dir, location, now_ms) { + ColdReadOutcome::Hit(v, ttl) => Some((v, ttl)), + ColdReadOutcome::Expired | ColdReadOutcome::Miss => None, + } } -fn read_cold_entry( - shard_dir: &Path, - location: ColdLocation, - now_ms: u64, -) -> Option<(RedisValue, Option)> { +fn read_cold_entry(shard_dir: &Path, location: ColdLocation, now_ms: u64) -> ColdReadOutcome { let file_path = shard_dir .join("data") .join(format!("heap-{:06}.mpf", location.file_id)); - let file = std::fs::File::open(&file_path).ok()?; + let Ok(file) = std::fs::File::open(&file_path) else { + return ColdReadOutcome::Miss; + }; // Read only the specific 4KB page identified by page_idx (pread, no whole-file read). let page_offset = (location.page_idx as u64) * (PAGE_4K as u64); let mut leaf_buf = [0u8; PAGE_4K]; - crate::util::file_ext::read_exact_at(&file, &mut leaf_buf, page_offset).ok()?; + if crate::util::file_ext::read_exact_at(&file, &mut leaf_buf, page_offset).is_err() { + return ColdReadOutcome::Miss; + } - let page = crate::persistence::kv_page::KvLeafPage::from_bytes(leaf_buf)?; - let entry = page.get(location.slot_idx)?; + let Some(page) = crate::persistence::kv_page::KvLeafPage::from_bytes(leaf_buf) else { + return ColdReadOutcome::Miss; + }; + let Some(entry) = page.get(location.slot_idx) else { + return ColdReadOutcome::Miss; + }; // Check TTL expiry if let Some(ttl_ms) = entry.ttl_ms { if now_ms > ttl_ms { - return None; // Expired + return ColdReadOutcome::Expired; } } @@ -71,12 +108,20 @@ fn read_cold_entry( let value_bytes = if entry.flags & entry_flags::OVERFLOW != 0 { // Overflow pointer: start_page_idx as u32 LE if entry.value.len() < 4 { - return None; + return ColdReadOutcome::Miss; } - let start_page_idx = u32::from_le_bytes(entry.value[..4].try_into().ok()?) as usize; + let Ok(ptr_bytes) = <[u8; 4]>::try_from(&entry.value[..4]) else { + return ColdReadOutcome::Miss; + }; + let start_page_idx = u32::from_le_bytes(ptr_bytes) as usize; // Only read the full file when following an overflow chain. - let file_data = std::fs::read(&file_path).ok()?; - read_overflow_chain(&file_data, start_page_idx)? + let Ok(file_data) = std::fs::read(&file_path) else { + return ColdReadOutcome::Miss; + }; + match read_overflow_chain(&file_data, start_page_idx) { + Some(v) => v, + None => return ColdReadOutcome::Miss, + } } else { entry.value }; @@ -84,10 +129,13 @@ fn read_cold_entry( // Convert to RedisValue based on value_type let redis_value = match entry.value_type { ValueType::String => RedisValue::String(Bytes::from(value_bytes)), - _ => kv_serde::deserialize_collection(&value_bytes, entry.value_type)?, + _ => match kv_serde::deserialize_collection(&value_bytes, entry.value_type) { + Some(v) => v, + None => return ColdReadOutcome::Miss, + }, }; - Some((redis_value, entry.ttl_ms)) + ColdReadOutcome::Hit(redis_value, entry.ttl_ms) } #[cfg(test)] @@ -148,6 +196,113 @@ mod tests { } } + /// Build a Database with an active cold tier holding one spilled key. + fn db_with_spilled_key( + shard_dir: &std::path::Path, + key: &[u8], + value: &[u8], + ttl_ms: Option, + ) -> crate::storage::db::Database { + let manifest_path = shard_dir.join("shard.manifest"); + let mut manifest = ShardManifest::create(&manifest_path).unwrap(); + let mut cold_index = ColdIndex::new(); + + let mut entry = Entry::new_string(Bytes::copy_from_slice(value)); + if let Some(ttl) = ttl_ms { + entry.set_expires_at_ms(0, ttl); + } + spill_to_datafile( + shard_dir, + 40, + key, + &entry, + &mut manifest, + Some(&mut cold_index), + ) + .unwrap(); + + let mut db = crate::storage::db::Database::new(); + db.cold_shard_dir = Some(shard_dir.to_path_buf()); + db.cold_index = Some(cold_index); + db + } + + /// D1 (PR review of tmp/OFFLOAD-COMPRESSION-REVIEW.md): DEL of a spilled + /// key must actually delete it — count it, drop the index entry, and make + /// subsequent GETs return nil instead of resurrecting the cold value. + #[test] + fn test_del_removes_cold_entry_no_resurrection() { + let tmp = tempfile::tempdir().unwrap(); + let mut db = db_with_spilled_key(tmp.path(), b"doomed", b"value-on-disk", None); + + // Sanity: the key is reachable via cold read-through before DEL. + assert!( + db.cold_index.as_ref().unwrap().lookup(b"doomed").is_some(), + "precondition: key is cold-indexed" + ); + + let frame = crate::command::key::del( + &mut db, + &[crate::protocol::Frame::BulkString(Bytes::from_static( + b"doomed", + ))], + ); + assert_eq!( + frame, + crate::protocol::Frame::Integer(1), + "DEL of a cold-only key must count it as removed" + ); + assert!( + db.get(b"doomed").is_none(), + "GET after DEL must NOT resurrect the cold value" + ); + assert!( + db.cold_index.as_ref().unwrap().lookup(b"doomed").is_none(), + "cold index entry must be gone after DEL" + ); + assert!( + db.cold_index.as_ref().unwrap().has_pending_unlink(), + "last referrer removed: file must be queued for unlink" + ); + } + + /// D1: FLUSHDB/FLUSHALL (`Database::clear`) must clear the cold tier too — + /// flushed keys must not remain readable from disk. + #[test] + fn test_clear_flushes_cold_tier() { + let tmp = tempfile::tempdir().unwrap(); + let mut db = db_with_spilled_key(tmp.path(), b"flushed", b"value-on-disk", None); + + db.clear(); + + assert!( + db.get(b"flushed").is_none(), + "GET after FLUSH must NOT read the cold value back from disk" + ); + assert!( + db.cold_index.as_ref().unwrap().has_pending_unlink(), + "cold files must be queued for unlink after clear" + ); + } + + /// R1: a cold read that finds the entry EXPIRED must reclaim the index + /// entry (and thereby the file refcount) instead of leaking it forever. + #[test] + fn test_expired_cold_read_reclaims_index_entry() { + let tmp = tempfile::tempdir().unwrap(); + // TTL 1ms in the past relative to the read below. + let mut db = db_with_spilled_key(tmp.path(), b"stale", b"old", Some(1)); + + assert!( + db.get(b"stale").is_none(), + "expired cold entry reads as nil" + ); + assert!( + db.cold_index.as_ref().unwrap().lookup(b"stale").is_none(), + "expired cold entry must be reclaimed from the index on read" + ); + } + #[test] fn test_cold_read_overflow_entry() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/storage/tiered/kv_spill.rs b/src/storage/tiered/kv_spill.rs index e7093044..7dcdf464 100644 --- a/src/storage/tiered/kv_spill.rs +++ b/src/storage/tiered/kv_spill.rs @@ -98,6 +98,9 @@ pub fn write_kv_spill_pages( } else { write_datafile_mixed(&file_path, &pages.leaf, &pages.overflow)?; } + // Fsync the directory so the new file's directory entry survives a crash — + // the manifest (which IS dir-fsynced) must never reference a vanished file. + crate::persistence::fsync::fsync_directory(&data_dir)?; Ok((pages.total_pages as u64) * (PAGE_4K as u64)) } @@ -354,11 +357,68 @@ pub fn write_kv_spill_batch(shard_dir: &Path, file_id: u64, batch: &BatchPages) } std::fs::rename(&tmp_path, &final_path)?; + // Fsync the directory so the rename itself survives a crash — without it + // the manifest can point at a file whose directory entry was lost. + crate::persistence::fsync::fsync_directory(&data_dir)?; let total_pages = (batch.leaves.len() + batch.overflow.len()) as u64; Ok(total_pages * PAGE_4K as u64) } +/// Startup sweep of crash-orphaned heap files in `{shard_dir}/data`. +/// +/// Removes `heap-*.mpf` files whose `file_id` is not registered in the +/// manifest (spill wrote the file, crash before manifest commit — invisible +/// to the cold index, would otherwise leak disk forever) and all +/// `heap-*.tmp` leftovers from interrupted atomic-rename batch writes. +/// +/// Only call this AFTER the manifest has been opened successfully: with no +/// readable manifest every heap file would look orphaned, and deleting data +/// on a corrupt-manifest signal would be destructive. +/// +/// Returns the number of files removed. I/O errors are logged and skipped — +/// a sweep failure must never abort recovery. +pub fn sweep_orphan_heap_files(shard_dir: &Path, manifest: &ShardManifest) -> usize { + let data_dir = shard_dir.join("data"); + let Ok(read_dir) = std::fs::read_dir(&data_dir) else { + return 0; + }; + let registered: std::collections::HashSet = + manifest.files().iter().map(|e| e.file_id).collect(); + + let mut removed = 0usize; + for dir_entry in read_dir.flatten() { + let path = dir_entry.path(); + let Some(name) = path.file_name().and_then(|n| n.to_str()) else { + continue; + }; + let orphan = if let Some(id_str) = name + .strip_prefix("heap-") + .and_then(|rest| rest.strip_suffix(".mpf")) + { + match id_str.parse::() { + Ok(file_id) => !registered.contains(&file_id), + Err(_) => false, // not our naming scheme — leave it alone + } + } else { + // Interrupted batch write: tmp files are always safe to remove. + name.strip_prefix("heap-") + .and_then(|rest| rest.strip_suffix(".tmp")) + .is_some() + }; + if orphan { + match std::fs::remove_file(&path) { + Ok(()) => { + removed += 1; + tracing::info!("cold-tier sweep: removed crash-orphaned {}", name); + } + Err(e) => warn!("cold-tier sweep: failed to remove {}: {}", name, e), + } + } + } + removed +} + #[cfg(test)] mod tests { use super::*; @@ -399,6 +459,47 @@ mod tests { assert_eq!(manifest.files()[0].file_id, 1); } + #[test] + fn test_sweep_orphan_heap_files() { + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path(); + let manifest_path = shard_dir.join("shard.manifest"); + let mut manifest = ShardManifest::create(&manifest_path).unwrap(); + + // Registered spill: file 1 in manifest, must survive the sweep. + let entry = Entry::new_string(Bytes::from_static(b"registered")); + spill_to_datafile(shard_dir, 1, b"livekey", &entry, &mut manifest, None).unwrap(); + + // Crash orphan: heap file on disk but never registered in the manifest + // (spill wrote the file, crash before manifest commit). + let data_dir = shard_dir.join("data"); + std::fs::write(data_dir.join("heap-000099.mpf"), [0u8; PAGE_4K]).unwrap(); + // Crash leftover: interrupted atomic-rename batch write. + std::fs::write(data_dir.join("heap-000050.tmp"), b"partial").unwrap(); + // Unrelated file: must not be touched. + std::fs::write(data_dir.join("notes.txt"), b"keep me").unwrap(); + + let removed = sweep_orphan_heap_files(shard_dir, &manifest); + + assert_eq!(removed, 2, "orphan .mpf + stale .tmp must both be removed"); + assert!( + data_dir.join("heap-000001.mpf").exists(), + "registered file must survive" + ); + assert!( + !data_dir.join("heap-000099.mpf").exists(), + "orphan must be unlinked" + ); + assert!( + !data_dir.join("heap-000050.tmp").exists(), + "stale tmp must be unlinked" + ); + assert!( + data_dir.join("notes.txt").exists(), + "unrelated files must survive" + ); + } + #[test] fn test_spill_with_ttl() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index 73bed015..f4120280 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -47,6 +47,27 @@ pub fn spill_completion_dropped_total() -> u64 { SPILL_COMPLETION_DROPPED.load(Ordering::Relaxed) } +/// Liveness heartbeat: unix millis of the most recent spill-thread loop +/// iteration (across all shards). 0 = no spill thread has ever run. A stale +/// value under active eviction means the spill thread died silently — the +/// only other symptom is an unbounded eviction backlog. +static SPILL_LAST_HEARTBEAT_MS: AtomicU64 = AtomicU64::new(0); + +/// Cumulative number of spill batches flushed to disk across all shards. +static SPILL_BATCHES_FLUSHED: AtomicU64 = AtomicU64::new(0); + +/// Unix millis of the most recent spill-thread loop tick (0 = never ran). +#[inline] +pub fn spill_last_heartbeat_ms() -> u64 { + SPILL_LAST_HEARTBEAT_MS.load(Ordering::Relaxed) +} + +/// Cumulative spill batches flushed to disk. Exposed for INFO / metrics. +#[inline] +pub fn spill_batches_flushed_total() -> u64 { + SPILL_BATCHES_FLUSHED.load(Ordering::Relaxed) +} + use bytes::Bytes; use tracing::warn; @@ -151,6 +172,7 @@ fn flush_buffer(buffer: &mut Vec) -> Vec { if buffer.is_empty() { return Vec::new(); } + SPILL_BATCHES_FLUSHED.fetch_add(1, Ordering::Relaxed); let mut completions: Vec = Vec::new(); @@ -366,6 +388,12 @@ impl SpillThread { let mut buffer: Vec = Vec::with_capacity(FLUSH_ENTRY_CAP); loop { + // Liveness heartbeat (R5): stamped every loop tick so INFO can + // surface a silently-dead spill thread. Relaxed max-store is + // enough — monotonicity across shards is not required. + SPILL_LAST_HEARTBEAT_MS + .fetch_max(crate::storage::entry::current_time_ms(), Ordering::Relaxed); + // Check stop flag — drain any still-queued requests and flush them // before exiting, so spills queued at shutdown are not lost (their // keys may already be evicted from RAM). @@ -553,6 +581,45 @@ mod tests { let _ = st.shutdown(); } + /// R5 liveness: the spill thread must publish a heartbeat + batch counter + /// so a silently-dead spill thread is observable from INFO instead of + /// only via unbounded eviction backlog. Statics are process-global, so + /// assert relative progress, not absolute values. + #[test] + fn test_spill_thread_liveness_metrics() { + let tmp = tempfile::tempdir().unwrap(); + let batches_before = spill_batches_flushed_total(); + + let st = SpillThread::new(9); + let sender = st.sender(); + sender + .send(SpillRequest { + key: Bytes::from_static(b"liveness_key"), + db_index: 0, + value_bytes: Bytes::from_static(b"liveness_value"), + value_type: ValueType::String, + flags: 0, + ttl_ms: None, + file_id: 1, + shard_dir: tmp.path().to_path_buf(), + }) + .unwrap(); + drop(sender); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let completions = collect_entries(&st, 1, deadline); + assert!(!completions.is_empty(), "spill must complete"); + + assert!( + spill_batches_flushed_total() > batches_before, + "flushing a batch must increment spill_batches_flushed_total" + ); + assert!( + spill_last_heartbeat_ms() > 0, + "a running spill thread must publish a heartbeat timestamp" + ); + } + /// Single request produces a successful per-FILE completion with one entry. #[test] fn test_spill_request_roundtrip() {