fix(storage): cold-tier correctness + reliability (DEL/FLUSH resurrection, expired-read leak, crash durability, orphan sweep, liveness)

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

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");
 

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.

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
+                );
+            }
         }
     }
 

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<Entry> {
+        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<Entry>) {
+        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<Entry> {
         if let Some(entry) = self.data.remove(key) {
             self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry));
             Some(entry)

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.