fix(storage): in-flight spills are a storage plane, not a bookkeeping note (#459)

CHANGELOG.md

@@ -44,6 +44,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `--shards 1` and `--shards 4`).
 
 ### Fixed
+- **A key whose disk-offload spill was in flight was invisible to the whole
+  server — `DEL` on it was silently undone (#459).** `evict_one_async_spill`
+  freed the hot entry the moment the `SpillRequest` was queued, and the key
+  was only registered in `cold_index` when the completion landed. In between
+  it existed in no plane the database consults: `spill_inflight` held only a
+  request id and was read solely by the completion path. The eviction code
+  documented the window as an acceptable "brief read-miss" backstopped by the
+  AOF — but the AOF backstops *durability*, not *visibility*, and nothing
+  considered a write landing inside the window. Measured on 400 × 4 KiB keys
+  against a 512 KiB cap: `DBSIZE` answered 124 for 400 acked keys and then
+  climbed to 400 on its own; `GET`/`EXISTS` denied live keys that returned
+  unaided 250 ms later; and 277 of 400 `DEL`s answered `:0` and were then
+  reversed by the completion — which publishes into `cold_index`
+  unconditionally, so those resurrections reached the manifest and survived
+  restart. A client that deleted data got it back.
+
+  `spill_inflight` is now a real third storage plane carrying the payload
+  (the same refcounted `Bytes` the queued request already pins — a refcount,
+  not a copy, and no extra peak memory). Reads promote from it with no disk
+  read at all, across all three dispatch paths (`promote_cold_if_present` for
+  collections and Lua/MULTI, the monoio async GET pre-warm, and the inline
+  `GET` fast path). `EXISTS`/`DEL` count it, `DBSIZE`/`logical_len` count it,
+  and `KEYS`/`RANDOMKEY` enumerate it. `DEL`, an overwriting `SET`, and a
+  promoting read each retire the record, which withdraws the completion's
+  authorization to publish — that is what makes a delete inside the window
+  final. Unpublished completions are counted as `spill_completion_superseded`
+  in INFO. Non-spilling servers pay one `is_empty()` load on the affected
+  paths. Known remaining gap, documented at the call site: `SCAN`'s ordered
+  cursor does not merge the unordered in-flight plane, so it may skip a key
+  for the milliseconds its spill is queued — within SCAN's contract, unlike
+  `KEYS`.
 - **`GET` inside `MULTI` was executed instead of queued (monoio).** Third
   defect from the same ungated inline read path, and the one most visible to a
   working client: `MULTI; GET k; EXEC` answered `+OK`, `$1 v`, `*0` where Redis

src/command/connection.rs

@@ -307,6 +307,7 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
          spill_batches_flushed:{}\r\n\
          spill_completions_dropped:{}\r\n\
          spill_failed_reinserted:{}\r\n\
+         spill_completion_superseded:{}\r\n\
          spill_last_heartbeat_ms:{}\r\n",
         if crate::command::persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed)
         {
@@ -349,6 +350,7 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
         crate::storage::tiered::spill_thread::spill_batches_flushed_total(),
         crate::storage::tiered::spill_thread::spill_completion_dropped_total(),
         crate::storage::tiered::spill_thread::spill_failed_reinserted_total(),
+        crate::storage::tiered::spill_thread::spill_completion_superseded_total(),
         crate::storage::tiered::spill_thread::spill_last_heartbeat_ms(),
     ));
     sections.push_str("\r\n");

src/server/conn/blocking.rs

@@ -1755,6 +1755,13 @@ pub(crate) fn try_inline_dispatch(
             if db.hot_keys().tick() {
                 db.hot_keys().observe(key_bytes);
             }
+            // #459: a key mid-spill is in neither hot nor cold, so the miss
+            // arm below would frame `$-1` inline for a key that exists and
+            // that EXISTS reports as present. Pull it back first — RAM only,
+            // no disk read — so the hot lookup answers it. Costs one
+            // `is_empty()` load per inline GET on a server that is not
+            // spilling, which is every server without --disk-offload.
+            db.promote_inflight_if_present(key_bytes, now_ms);
             match db.get_if_alive(key_bytes, now_ms) {
                 Some(entry) => match entry.value.as_bytes() {
                     Some(val) => {

src/server/conn/handler_monoio/mod.rs

@@ -2525,10 +2525,22 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     // (`Database::promote_cold_if_present`), unchanged.
                     if cmd.eq_ignore_ascii_case(b"GET") {
                         if let Some(key) = cmd_args.first().and_then(extract_bytes) {
+                            let peek_now_ms = ctx.cached_clock.ms();
                             let cold_loc =
                                 crate::shard::slice::with_shard_db(conn.selected_db, |db| {
                                     if db.is_hot(key.as_ref()) {
                                         None
+                                    } else if db
+                                        .promote_inflight_if_present(key.as_ref(), peek_now_ms)
+                                    {
+                                        // #459: mid-spill, payload still in
+                                        // RAM. Now hot, so `dispatch_read`
+                                        // below answers it — and no disk read
+                                        // was needed. Without this the key is
+                                        // in no plane this peek consults and
+                                        // GET answers nil for a key EXISTS
+                                        // reports as present.
+                                        None
                                     } else {
                                         db.cold_lookup_location(key.as_ref())
                                     }

src/shard/persistence_tick.rs

@@ -653,8 +653,10 @@ fn apply_completion_vec(
                         tracing::warn!(
                             file_id = c.file_entry.file_id,
                             key_len = req.key.len(),
-                            "Spill pwrite failed for a SUPERSEDED request; skipping \
-                             re-insert (a newer spill of this key is in flight or landed)"
+                            "Spill pwrite failed for a request that no longer owns this \
+                             key; skipping re-insert (a newer spill is in flight or \
+                             landed, or the key was deleted / overwritten / read-promoted \
+                             while this write was in flight — #459)"
                         );
                         return;
                     }
@@ -718,12 +720,22 @@ fn apply_completion_vec(
             };
 
             crate::shard::slice::with_shard_db(entry.db_index, |db| {
+                // The in-flight record is this completion's AUTHORIZATION to
+                // publish, not just a stale-shadow guard (#459). It is gone
+                // when the key was deleted, overwritten, or read-promoted
+                // while the spill was in flight; publishing anyway resurrects
+                // a deleted key — and because this insert reaches the
+                // manifest, the resurrection survives restart. Measured
+                // pre-fix: 277 of 400 DELs undone this way.
+                if !db.spill_inflight_is_newest(&entry.key, entry.req_file_id) {
+                    crate::storage::tiered::spill_thread::record_spill_completion_superseded();
+                    return;
+                }
                 if let Some(ref mut ci) = db.cold_index {
                     ci.insert(entry.key.clone(), location);
                 }
-                // Retire this request's in-flight record (stale-shadow
-                // guard); a newer request's record is left for its own
-                // completion.
+                // Retire this request's record; a newer request's is left
+                // for its own completion.
                 db.spill_inflight_clear(&entry.key, entry.req_file_id);
             });
         }

src/storage/db/accessors.rs

@@ -442,6 +442,16 @@ impl Database {
         key: &[u8],
         now_ms: u64,
     ) -> Option<crate::storage::entry::RedisValue> {
+        // In-flight plane first (#459): a key mid-spill is in neither hot nor
+        // cold, so every `&self` reader that stops here would answer nil for
+        // a key `EXISTS` reports as present. No disk read — the payload is in
+        // RAM. This is the choke point for the RwLock-shared-read dispatch
+        // path (`*_readonly` handlers, which the tokio runtime's GET takes),
+        // so it must come before the `cold_shard_dir`/`cold_index` bails
+        // below.
+        if let Some(value) = self.spill_inflight_value(key, now_ms) {
+            return Some(value);
+        }
         let shard_dir = self.cold_shard_dir.as_ref()?;
         let ci = self.cold_index.as_ref()?;
         let (value, _ttl) =

src/storage/db/kv_ops.rs

@@ -97,6 +97,9 @@ impl Database {
         if self.data.contains_key(key) {
             return true;
         }
+        if self.promote_inflight_if_present(key, now_ms) {
+            return true;
+        }
         // Look up the location first (cheap, in-memory) so the outcome below
         // can be paired with the location it was read from -- required by
         // `promote_cold_outcome`'s revalidation (see its doc comment). This
@@ -112,6 +115,33 @@ impl Database {
         self.promote_cold_outcome(key, now_ms, location, outcome)
     }
 
+    /// Pull `key` back into hot RAM from the IN-FLIGHT spill plane, if it is
+    /// there and not TTL-expired. Returns `true` if the key is hot after this
+    /// call because of it.
+    ///
+    /// Cheaper than the cold path it precedes: the payload is already in RAM
+    /// (it is the queued `SpillRequest`'s own refcounted buffer), so this
+    /// costs a rehydrate and no disk read at all.
+    ///
+    /// Retiring the in-flight record is not merely tidy — it withdraws the
+    /// completion's authorization to publish into `cold_index`, which is
+    /// correct here: the key is hot again, and publishing would leave a
+    /// stale cold shadow behind it (#459).
+    ///
+    /// Every read path that can reach a cold key must call this first, or it
+    /// will answer nil for a key that was only ever mid-spill.
+    pub fn promote_inflight_if_present(&mut self, key: &[u8], now_ms: u64) -> bool {
+        if self.spill_inflight_is_empty() {
+            return false;
+        }
+        let Some(entry) = self.spill_inflight_entry(key, now_ms) else {
+            return false;
+        };
+        self.spill_inflight_forget(key);
+        self.set(Bytes::copy_from_slice(key), entry);
+        true
+    }
+
     /// Apply an already-computed [`cold_read::ColdReadOutcome`] to hot RAM +
     /// the cold index, without performing any disk I/O itself.
     ///
@@ -217,6 +247,13 @@ impl Database {
     /// promoting paths handle reclamation.
     #[inline]
     pub(super) fn cold_contains_alive(&self, key: &[u8], now_ms: u64) -> bool {
+        // A key whose spill is still in flight is in neither hot nor cold,
+        // but it exists — the client's write was acked and nothing deleted
+        // it. Answering `false` here is what made EXISTS deny live keys and
+        // DEL answer :0 inside the window (#459).
+        if self.spill_inflight_alive(key, now_ms) {
+            return true;
+        }
         let Some(ci) = self.cold_index.as_ref() else {
             return false;
         };
@@ -241,6 +278,9 @@ impl Database {
     /// mutate `self` to promote a cold hit into hot RAM, so the safe fix is
     /// "decode it from disk every time it's cold" rather than "silently
     /// report the key absent" (same P0 as [`Self::promote_cold_if_present`]).
+    ///
+    /// [`Self::get_cold_value`] consults the in-flight plane before the cold
+    /// one (#459), so these accessors inherit it and see a key mid-spill.
     #[inline]
     pub(super) fn cold_read_only(&self, key: &[u8], now_ms: u64) -> Option<RedisValue> {
         self.get_cold_value(key, now_ms)
@@ -275,6 +315,15 @@ impl Database {
     /// and miss paths. The old `get_mut` + `insert` pattern ran two probes on
     /// miss (PERF-08).
     pub fn set(&mut self, key: Bytes, entry: Entry) {
+        // An overwrite makes any in-flight spill payload for this key stale.
+        // Retiring the record here stops its completion publishing the OLD
+        // value into `cold_index`, where it would sit as a shadow behind the
+        // new hot value and become authoritative again after a restart
+        // demoted the hot copy (#459). One `is_empty()` load on the write
+        // hot path when nothing is spilling, which is the normal case.
+        if !self.spill_inflight_is_empty() {
+            self.spill_inflight_forget(&key);
+        }
         let new_cost = entry_overhead(&key, &entry);
         let has_expiry = entry.has_expiry();
         let mut old_cost: usize = 0;
@@ -450,13 +499,26 @@ impl Database {
             .as_ref()
             .and_then(|ci| ci.lookup(key))
             .is_some_and(|loc| loc.ttl_ms.is_none_or(|ttl| now_ms <= ttl));
+        // An in-flight key counts as removed for the same reason it counts as
+        // existing: the write was acked and nothing has deleted it yet.
+        let inflight_alive = self.spill_inflight_alive(key, now_ms);
         let had_cold = self.remove_cold_only(key);
         let hot = self.remove_hot(key);
-        (hot.is_some() || (had_cold && cold_alive), hot)
+        (
+            hot.is_some() || (had_cold && cold_alive) || inflight_alive,
+            hot,
+        )
     }
 
+    /// Drops the cold copy AND any in-flight spill record.
+    ///
+    /// Retiring the in-flight record is the load-bearing half (#459): it is
+    /// the completion's authorization to insert into `cold_index`, so
+    /// without this a DEL issued during the spill window was undone when the
+    /// spill landed — and committed to the manifest, so it survived restart.
     #[inline]
     fn remove_cold_only(&mut self, key: &[u8]) -> bool {
+        self.spill_inflight_forget(key);
         self.cold_index.as_mut().is_some_and(|ci| ci.remove(key))
     }
 
@@ -575,7 +637,7 @@ impl Database {
     /// (hot or cold) are counted.
     pub fn logical_len(&self) -> usize {
         let hot = self.data.len();
-        match &self.cold_index {
+        let hot_and_cold = match &self.cold_index {
             None => hot,
             Some(ci) => {
                 let overlap = ci
@@ -584,7 +646,27 @@ impl Database {
                     .count();
                 hot + ci.len() - overlap
             }
+        };
+        if self.spill_inflight_is_empty() {
+            return hot_and_cold;
         }
+        // Keys mid-spill are in neither plane above but are logically
+        // present (#459): counting only hot+cold made DBSIZE answer 124 for
+        // 400 acked keys and then climb to 400 on its own. Same
+        // count-each-key-once discipline as the cold overlap — a key can be
+        // in flight while a fresh SET has already re-created it hot, or
+        // while a superseded cold entry still exists.
+        let inflight_only = self
+            .spill_inflight_keys()
+            .filter(|k| {
+                !self.data.contains_key(k)
+                    && !self
+                        .cold_index
+                        .as_ref()
+                        .is_some_and(|ci| ci.lookup(k).is_some())
+            })
+            .count();
+        hot_and_cold + inflight_only
     }
 
     /// Iterator over all keys (caller does glob filtering).
@@ -664,11 +746,29 @@ impl Database {
     /// is classified hot; a hot entry that is TTL-expired above a live
     /// cold entry is classified cold.
     pub fn cold_only_keys(&self, now_ms: u64) -> impl Iterator<Item = &Bytes> + '_ {
-        self.cold_index.as_ref().into_iter().flat_map(move |ci| {
+        let cold = self.cold_index.as_ref().into_iter().flat_map(move |ci| {
             ci.iter()
                 .filter(move |(key, loc)| self.is_cold_only_alive(key, loc, now_ms))
                 .map(|(key, _)| key)
-        })
+        });
+        // Keys mid-spill are in neither plane but exist (#459) — KEYS and
+        // RANDOMKEY promise a point-in-time view of the keyspace, so
+        // omitting them would hide a live key for the length of the window.
+        // Same partition discipline as the cold half: skip anything a hot
+        // entry or a cold entry already accounts for.
+        let inflight = self
+            .spill_inflight
+            .iter()
+            .filter(move |(key, p)| {
+                p.ttl_ms.is_none_or(|ttl| now_ms <= ttl)
+                    && !self.data.contains_key(key)
+                    && !self
+                        .cold_index
+                        .as_ref()
+                        .is_some_and(|ci| ci.lookup(key).is_some())
+            })
+            .map(|(key, _)| key);
+        cold.chain(inflight)
     }
 
     /// The cold-only liveness predicate shared by [`Self::cold_only_keys`]
@@ -699,6 +799,16 @@ impl Database {
     /// [`Self::is_cold_only_alive`]. SCAN takes the first COUNT — since the
     /// order is ascending, those are exactly the smallest cold candidates —
     /// instead of filtering the entire index on every page.
+    ///
+    /// KNOWN GAP (#459): unlike [`Self::cold_only_keys`], this does NOT
+    /// include keys whose spill is still in flight. The in-flight map is
+    /// unordered, so merging it into an ascending `hash48` walk needs a
+    /// merge-sort against a plane that mutates under the cursor — real work,
+    /// deliberately not done here. A key can therefore be skipped by SCAN
+    /// for the milliseconds its spill is queued. That stays within SCAN's
+    /// contract (only keys present for the WHOLE iteration are guaranteed),
+    /// and `KEYS`/`RANDOMKEY`, whose contracts are point-in-time, do include
+    /// them.
     pub fn cold_only_keys_from(
         &self,
         from_h48: u64,