From 417a465097e55ec8b8def5aa2ecd4d74f91065d1 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 10 Aug 2026 17:35:52 +0700 Subject: [PATCH 1/3] fix(storage): in-flight spills are a storage plane, not a bookkeeping note (#459) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A key whose disk-offload spill was queued but not yet completed existed in no plane the database consults. `evict_one_async_spill` freed the hot entry the moment the `SpillRequest` was sent, and `cold_index` only learned about the key when the completion landed; `spill_inflight` held nothing but a request id and was read solely by the completion path. The eviction code documented the window and accepted it — "accept a brief read-miss ... AOF incr log is the durability backstop". The AOF does backstop durability. It does not backstop visibility, and nothing considered a write arriving inside the window. Measured on origin/main @4c9bd2c5, 400 x 4KiB keys against a 512KiB cap (the config of the failing test in #459): * DBSIZE answered 124 for 400 acked keys, then climbed to 400 on its own with no further writes — this is the `live 373 / recovered 400` that filed the issue; * GET and EXISTS denied live keys: `GET k0` -> nil, `EXISTS k0` -> 0, then 250ms later `GET k0` -> 4096 bytes, no intervening write; * 277 of 400 DELs answered :0 and were then reversed. The completion path inserted into `cold_index` unconditionally, so the resurrections reached the manifest and survived restart. A client that deleted data got it back. `spill_inflight` now carries the payload and is treated as the third plane it always was. The payload is the same refcounted `Bytes` the queued request already pins, so this is a refcount rather than a copy and adds no peak memory. * reads promote from it with no disk read, on all three dispatch paths — `promote_cold_if_present` (collections, Lua, MULTI), the monoio async GET pre-warm, and the inline GET fast path, which was framing `$-1` inline for keys EXISTS reported as present; * EXISTS / DEL count it; DBSIZE and `logical_len` count it; KEYS and 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, and it also stops a stale payload being published as a cold shadow behind a newer hot value; * unpublished completions are counted as `spill_completion_superseded`. Servers that are not spilling 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. That is within SCAN's contract (only keys present for the whole iteration are guaranteed); KEYS, whose contract is point-in-time, does include them. Tests: `tests/spill_inflight_visibility.rs` — one wire-level test per fault, each red on the pre-fix binary for the right reason (277 resurrections, retry-liars, DBSIZE 124->400) and green after. Retry-and-compare is the assertion shape throughout: a key that answers nil and then returns a value with no write in between cannot be explained by legitimate allkeys-lru eviction, which never brings a key back. Plus five unit tests on the plane's own semantics in `src/storage/db/mod.rs`. Verified: full suite 5110/5110 (monoio, the shipped runtime), clippy clean, fmt clean; the three independent Python repro harnesses that found the faults all report zero on the fixed binary; `tests/dbsize_offload_logical.rs` — the suite that filed #459 — passes. Fixes #459 author: Tin Dang --- CHANGELOG.md | 33 ++ src/server/conn/blocking.rs | 7 + src/server/conn/handler_monoio/mod.rs | 12 + src/shard/persistence_tick.rs | 22 +- src/storage/db/accessors.rs | 10 + src/storage/db/kv_ops.rs | 118 ++++++- src/storage/db/mod.rs | 268 ++++++++++++++-- src/storage/eviction.rs | 43 ++- src/storage/tiered/spill_thread.rs | 25 ++ tests/spill_inflight_visibility.rs | 431 ++++++++++++++++++++++++++ 10 files changed, 924 insertions(+), 45 deletions(-) create mode 100644 tests/spill_inflight_visibility.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6eeaa138..ede1719d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### 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`. + ### Added - **Client-compat harness: raw-RESP diff against a real `redis-server` (`scripts/test-client-compat.sh`).** Moon's existing Redis comparison diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs index b79762eb..57903586 100644 --- a/src/server/conn/blocking.rs +++ b/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) => { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index ed4d2ca9..56341771 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/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()) } diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index b76e8834..55f37182 100644 --- a/src/shard/persistence_tick.rs +++ b/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); }); } diff --git a/src/storage/db/accessors.rs b/src/storage/db/accessors.rs index 7ba0ccf8..0d99d129 100644 --- a/src/storage/db/accessors.rs +++ b/src/storage/db/accessors.rs @@ -442,6 +442,16 @@ impl Database { key: &[u8], now_ms: u64, ) -> Option { + // 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) = diff --git a/src/storage/db/kv_ops.rs b/src/storage/db/kv_ops.rs index e29f751a..2e741533 100644 --- a/src/storage/db/kv_ops.rs +++ b/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 { 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 + '_ { - 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, diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs index 39a4d742..7b1c5049 100644 --- a/src/storage/db/mod.rs +++ b/src/storage/db/mod.rs @@ -198,18 +198,52 @@ pub struct Database { pub cold_shard_dir: Option, /// Hot-key detection sketch, fed by sampled dispatch observations. hot_keys: crate::storage::hotkey::HotKeySketch, - /// Newest in-flight async-spill request id per key (deep-review P2 - /// stale-shadow guard). Marked at enqueue (`evict_one_async_spill`), - /// consumed when that request's completion applies. The failure - /// re-insert arm uses it to detect that a NEWER spill superseded the - /// failed one (key re-created and re-evicted while the failed pwrite - /// was in flight) — re-inserting the older payload would shadow the - /// newer cold value. Touched only on evict/completion paths, never on - /// command dispatch. A dropped completion (see - /// `spill_completion_dropped_total`) can strand an entry until the key - /// is next evicted — bounded and harmless (a stale id only ever - /// SUPPRESSES a re-insert of an equally stale payload). - spill_inflight: std::collections::HashMap, + /// Keys whose async spill is IN FLIGHT: enqueued to the spill thread, + /// hot entry already freed, not yet registered in `cold_index`. + /// + /// This is the THIRD storage plane, not a bookkeeping side table (#459). + /// It began as a write-only supersession guard holding just the request + /// id, which left the in-flight window visible to nobody: reads, deletes + /// and `logical_len` consult hot and cold only, so for the length of the + /// window an acked key was denied by `GET`/`EXISTS`, `DEL` answered `:0` + /// and then the completion resurrected it, and `DBSIZE` under-counted. + /// It therefore carries the PAYLOAD as well, so every plane-aware path + /// can answer from RAM with no disk read and no promotion. + /// + /// Costs nothing to hold: `value_bytes` is the very same refcounted + /// `Bytes` the queued `SpillRequest` already pins for the whole window, + /// so this is one refcount, not a copy. + /// + /// Empty for every server not actively spilling, which is what makes the + /// `is_empty()` fast bail on the hot paths below honest rather than a + /// hopeful guess. Size is bounded by the spill channel's capacity: + /// `evict_one_async_spill` marks only after a successful `try_send`, and + /// a full channel makes it bail. + /// + /// Retention note: a record is retired by its completion, by DEL, by an + /// overwriting SET, or by a promoting read. A LOST completion (see + /// `spill_completion_dropped_total` — the rare shutdown-with-full-channel + /// edge) therefore now strands a payload rather than the bare u64 it used + /// to, so the value stays charged to RAM until the key is next written or + /// deleted. Reads stay correct throughout; the cost is that the eviction + /// did not actually reclaim that key's memory. + spill_inflight: std::collections::HashMap, +} + +/// A spill that has left hot RAM but has not yet landed in `cold_index`. +/// +/// Holds enough to answer a read without touching disk — the window is short +/// but it is not free, and serving a stale-or-nil answer inside it was #459. +#[derive(Clone)] +pub struct PendingSpill { + /// The `SpillRequest.file_id` this record belongs to. A completion only + /// applies when it is still the newest (stale-shadow guard, deep-review + /// P2): a newer eviction of the same key supersedes it. + pub req_id: u64, + pub value_type: crate::persistence::kv_page::ValueType, + /// Refcounted handle to the queued request's payload — never a copy. + pub value_bytes: bytes::Bytes, + pub ttl_ms: Option, } impl Database { @@ -254,27 +288,114 @@ impl Database { } } - /// Record `req_id` as the newest in-flight async-spill request for - /// `key` (called at enqueue time, before the hot entry is removed). - pub fn spill_inflight_mark(&mut self, key: bytes::Bytes, req_id: u64) { - self.spill_inflight.insert(key, req_id); + /// Record `key` as in-flight, carrying the payload so reads answered + /// during the window are correct. + /// + /// Called AFTER the hot entry is removed (`evict_one_async_spill`). The + /// remove and this call are one synchronous run on the shard thread with + /// no `.await` between them, so no reader can observe the key absent from + /// every plane. + pub fn spill_inflight_mark(&mut self, key: bytes::Bytes, pending: PendingSpill) { + self.spill_inflight.insert(key, pending); } /// True when `req_id` is still the newest recorded spill request for - /// `key` — i.e. no later eviction re-enqueued the key while this - /// request was in flight. + /// `key` — i.e. no later eviction re-enqueued it, and no `DEL`/overwrite + /// retired the record while this request was in flight. pub fn spill_inflight_is_newest(&self, key: &[u8], req_id: u64) -> bool { - self.spill_inflight.get(key) == Some(&req_id) + self.spill_inflight.get(key).map(|p| p.req_id) == Some(req_id) } /// Consume the in-flight record for `key` if (and only if) it belongs /// to `req_id`; a newer request's record is left for its own completion. pub fn spill_inflight_clear(&mut self, key: &[u8], req_id: u64) { - if self.spill_inflight.get(key) == Some(&req_id) { + if self.spill_inflight.get(key).map(|p| p.req_id) == Some(req_id) { self.spill_inflight.remove(key); } } + /// Retire any in-flight record for `key`, whatever request it belongs to. + /// Returns `true` if one existed. + /// + /// This is what makes a `DEL` (or an overwriting `SET`) inside the window + /// FINAL: the record is the only thing that authorizes the completion to + /// insert into `cold_index`, so dropping it here is what stops the key + /// coming back when the spill lands. + #[inline] + pub fn spill_inflight_forget(&mut self, key: &[u8]) -> bool { + if self.spill_inflight.is_empty() { + return false; + } + self.spill_inflight.remove(key).is_some() + } + + /// Cheap (no disk I/O, no promotion) liveness probe for the in-flight + /// plane — the pending-spill sibling of `cold_contains_alive`. + #[inline] + pub fn spill_inflight_alive(&self, key: &[u8], now_ms: u64) -> bool { + if self.spill_inflight.is_empty() { + return false; + } + self.spill_inflight + .get(key) + .is_some_and(|p| p.ttl_ms.is_none_or(|ttl| now_ms <= ttl)) + } + + /// Rehydrate the pending payload for `key` into an `Entry`, if one is + /// in flight and not TTL-expired. Pure RAM — the payload is right here, + /// so this never reads disk. + /// + /// `None` when the payload cannot be rehydrated (corrupt encoding); the + /// caller then falls through to its normal path rather than serving a + /// wrong value. + pub fn spill_inflight_entry(&self, key: &[u8], now_ms: u64) -> Option { + if self.spill_inflight.is_empty() { + return None; + } + let p = self.spill_inflight.get(key)?; + if p.ttl_ms.is_some_and(|ttl| now_ms > ttl) { + return None; + } + crate::storage::eviction::rehydrate_spill_payload(p.value_type, &p.value_bytes, p.ttl_ms) + } + + /// Decode the pending payload for `key` into a `RedisValue` without + /// touching hot RAM or the cold index — the `&self` sibling of + /// [`Self::spill_inflight_entry`]. + /// + /// This is what the RwLock-shared-read dispatch path needs: it holds only + /// `&Database` and so cannot promote, exactly as it cannot promote a cold + /// hit. Unlike the cold case there is no disk read to pay for — the + /// payload is already in RAM. + pub fn spill_inflight_value(&self, key: &[u8], now_ms: u64) -> Option { + if self.spill_inflight.is_empty() { + return None; + } + let p = self.spill_inflight.get(key)?; + if p.ttl_ms.is_some_and(|ttl| now_ms > ttl) { + return None; + } + match p.value_type { + crate::persistence::kv_page::ValueType::String => { + Some(RedisValue::String(p.value_bytes.clone())) + } + vt => crate::storage::tiered::kv_serde::deserialize_collection(&p.value_bytes, vt), + } + } + + /// Keys currently in flight — used to count them as the logical keys they + /// are (`logical_len`) and to keep them enumerable. + pub fn spill_inflight_keys(&self) -> impl Iterator { + self.spill_inflight.keys() + } + + /// True when nothing is in flight — the fast bail every plane-aware + /// caller takes on a server that is not spilling. + #[inline] + pub fn spill_inflight_is_empty(&self) -> bool { + self.spill_inflight.is_empty() + } + /// Fast-path predicate for the active-expiry tick. Returns `false` only /// when the database is known to have zero entries with a TTL. Callers /// MUST treat `true` as "maybe has expiring keys" — the precise answer @@ -1194,11 +1315,11 @@ mod tests { fn spill_inflight_supersession_guard() { let mut db = Database::new(); let k = bytes::Bytes::from_static(b"k"); - db.spill_inflight_mark(k.clone(), 7); + db.spill_inflight_mark(k.clone(), pending(7, b"v7")); assert!(db.spill_inflight_is_newest(b"k", 7)); // Key re-evicted with a newer request: 7 is now superseded — its // failure arm must NOT re-insert its stale payload. - db.spill_inflight_mark(k, 9); + db.spill_inflight_mark(k, pending(9, b"v9")); assert!(!db.spill_inflight_is_newest(b"k", 7)); // The superseded completion's clear leaves the newer record intact. db.spill_inflight_clear(b"k", 7); @@ -1207,4 +1328,107 @@ mod tests { db.spill_inflight_clear(b"k", 9); assert!(!db.spill_inflight_is_newest(b"k", 9)); } + + fn pending(req_id: u64, value: &'static [u8]) -> PendingSpill { + PendingSpill { + req_id, + value_type: crate::persistence::kv_page::ValueType::String, + value_bytes: bytes::Bytes::from_static(value), + ttl_ms: None, + } + } + + /// #459: the in-flight plane must answer reads. A key mid-spill is in + /// neither hot nor cold, and before this it was invisible to everything + /// except the completion path. + #[test] + fn an_inflight_key_is_readable_and_counted_while_it_is_in_flight() { + let mut db = Database::new(); + let k = bytes::Bytes::from_static(b"k"); + db.spill_inflight_mark(k, pending(1, b"hello")); + + assert!( + db.spill_inflight_alive(b"k", 0), + "a queued spill still holds a live key" + ); + assert_eq!( + db.logical_len(), + 1, + "the key was acked to a client, so it must be counted" + ); + let entry = db + .spill_inflight_entry(b"k", 0) + .expect("payload rehydrates from RAM"); + assert_eq!(entry.value.as_bytes(), Some(&b"hello"[..])); + } + + /// The half that turned a transient miss into permanent data loss: DEL + /// must retire the record, because the record is the completion's + /// authorization to publish into `cold_index`. + #[test] + fn deleting_an_inflight_key_withdraws_the_completions_authorization() { + let mut db = Database::new(); + let k = bytes::Bytes::from_static(b"k"); + db.spill_inflight_mark(k, pending(1, b"hello")); + + let (removed, _) = db.remove_counting_cold(b"k"); + assert!(removed, "DEL of a mid-spill key must answer 1, not 0"); + assert!( + !db.spill_inflight_is_newest(b"k", 1), + "the completion must no longer be authorized to publish this key — \ + otherwise the spill lands in cold_index and the delete is undone" + ); + assert_eq!(db.logical_len(), 0, "the key is gone from every plane"); + } + + /// An overwrite makes the queued payload stale; publishing it would + /// leave the OLD value as a cold shadow behind the new hot one. + #[test] + fn overwriting_an_inflight_key_also_withdraws_authorization() { + let mut db = Database::new(); + let k = bytes::Bytes::from_static(b"k"); + db.spill_inflight_mark(k.clone(), pending(1, b"old")); + + db.set(k, Entry::new_string(bytes::Bytes::from_static(b"new"))); + + assert!( + !db.spill_inflight_is_newest(b"k", 1), + "the queued payload is stale after an overwrite" + ); + assert_eq!(db.logical_len(), 1, "one key, counted once, not twice"); + } + + /// A promoting read pulls the payload back from RAM and likewise cancels + /// the pending publish — the key is hot again, so a cold entry would + /// only shadow it. + #[test] + fn a_read_promotes_an_inflight_key_out_of_the_window() { + let mut db = Database::new(); + let k = bytes::Bytes::from_static(b"k"); + db.spill_inflight_mark(k, pending(1, b"hello")); + + assert!(db.promote_inflight_if_present(b"k", 0)); + assert!(db.is_hot(b"k"), "promotion puts the key back in hot RAM"); + assert!(!db.spill_inflight_is_newest(b"k", 1)); + assert_eq!(db.logical_len(), 1, "still exactly one key"); + } + + /// An in-flight record whose TTL has passed is not a live key. + #[test] + fn an_expired_inflight_payload_is_not_alive() { + let mut db = Database::new(); + let k = bytes::Bytes::from_static(b"k"); + db.spill_inflight_mark( + k, + PendingSpill { + req_id: 1, + value_type: crate::persistence::kv_page::ValueType::String, + value_bytes: bytes::Bytes::from_static(b"v"), + ttl_ms: Some(1_000), + }, + ); + assert!(db.spill_inflight_alive(b"k", 999), "not yet expired"); + assert!(!db.spill_inflight_alive(b"k", 1_001), "past its TTL"); + assert!(db.spill_inflight_entry(b"k", 1_001).is_none()); + } } diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index babe3e61..a5da9de3 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -907,6 +907,11 @@ fn evict_one_async_spill( let file_id = *next_file_id; *next_file_id += 1; + // Second handle on the SAME payload the request carries (refcount, + // not a copy) so the in-flight plane below can answer reads from RAM + // while the request is queued. + let pending_bytes = value_bytes.clone(); + let req = SpillRequest { key: Bytes::copy_from_slice(key.as_bytes()), db_index, @@ -925,23 +930,33 @@ fn evict_one_async_spill( if sender.try_send(req).is_err() { return false; } - // Deep-review P2 stale-shadow guard: record this request as the - // newest in-flight spill for the key BEFORE freeing RAM, so a later - // failed-pwrite re-insert of an OLDER superseded request for the - // same key can be detected and suppressed. - db.spill_inflight_mark(Bytes::copy_from_slice(key.as_bytes()), file_id); - - // Now safe to free RAM. The bg thread holds the SpillRequest and will - // produce a SpillCompletion that updates cold_index for this db_index. + // Free RAM, then hand the key to the in-flight plane. `db.remove` + // clears any in-flight record for the key (that is what makes DEL + // final), so the mark MUST follow it — marking first would erase + // itself. Both calls run synchronously on the shard thread with no + // `.await` between them, so no reader can catch the key between + // planes. db.remove(key.as_bytes()); - // NOTE: we do NOT insert a tentative cold_index entry here. + // NOTE: we still do NOT insert a tentative cold_index entry here. // Under batching the (page_idx, slot_idx) are unknown at evict time; - // slot 0 would return a *different* key's value in the pre-flush window. - // Accept a brief read-miss until the completion applies — the key is - // safe: it is in the SpillRequest and will be registered once the bg - // thread writes and the event loop processes the SpillCompletion. - // AOF incr log is the durability backstop for the pre-flush window. + // slot 0 would return a *different* key's value in the pre-flush + // window. Instead the key stays readable from the in-flight plane + // itself, which holds the payload in RAM until the completion lands + // (#459). Before that fix this window was documented as "accept a + // brief read-miss" with the AOF as backstop — but the AOF backstops + // DURABILITY, not VISIBILITY: measured on 400 keys, GET and EXISTS + // denied live keys, DEL answered :0 and was then undone by the + // completion, and DBSIZE under-counted by 276. + db.spill_inflight_mark( + Bytes::copy_from_slice(key.as_bytes()), + crate::storage::db::PendingSpill { + req_id: file_id, + value_type, + value_bytes: pending_bytes, + ttl_ms, + }, + ); } else { // Entry disappeared (race with expiry), just remove db.remove(key.as_bytes()); diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index 49564e6f..b1799485 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -171,6 +171,31 @@ pub fn record_spill_failed_reinserted() { SPILL_FAILED_REINSERTED.fetch_add(1, Ordering::Relaxed); } +/// Spill completions that landed on disk but were NOT published to the cold +/// index because their in-flight record had been retired — the key was +/// deleted, overwritten, or read-promoted while the spill was in flight +/// (#459). Publishing those would resurrect a deleted key. +/// +/// Not an error: it is the normal, correct outcome of a write racing a +/// spill. It is counted because the spilled bytes become orphaned in their +/// file, and a rate that tracks the eviction rate means the workload is +/// re-touching keys as fast as they are being spilled — the signal that +/// `maxmemory` is too small for the working set. Exposed as +/// `spill_completion_superseded` in INFO. +static SPILL_COMPLETION_SUPERSEDED: AtomicU64 = AtomicU64::new(0); + +/// Cumulative superseded (unpublished) spill completions. For INFO / metrics. +#[inline] +pub fn spill_completion_superseded_total() -> u64 { + SPILL_COMPLETION_SUPERSEDED.load(Ordering::Relaxed) +} + +/// Record one superseded spill completion. +#[inline] +pub fn record_spill_completion_superseded() { + SPILL_COMPLETION_SUPERSEDED.fetch_add(1, Ordering::Relaxed); +} + use bytes::Bytes; use tracing::warn; diff --git a/tests/spill_inflight_visibility.rs b/tests/spill_inflight_visibility.rs new file mode 100644 index 00000000..fc4d67a9 --- /dev/null +++ b/tests/spill_inflight_visibility.rs @@ -0,0 +1,431 @@ +//! Issue #459: a key whose async spill is IN FLIGHT must stay fully visible. +//! +//! `evict_one_async_spill` removes the hot entry as soon as the `SpillRequest` +//! is queued, and the key is only registered in `cold_index` when the +//! completion lands. In between, the key exists in NO plane the database +//! consults — `spill_inflight` is a supersession guard read solely by the +//! completion path, never by reads, deletes, or `logical_len`. +//! +//! The eviction code names the window and accepts it: +//! +//! > Accept a brief read-miss until the completion applies — the key is +//! > safe: it is in the SpillRequest and will be registered once the bg +//! > thread writes [...] AOF incr log is the durability backstop. +//! +//! AOF does backstop *durability*. It does not backstop *visibility*, and a +//! read-miss is not a latency artifact — it is a wrong answer to a client. +//! Measured on `origin/main` @4c9bd2c5 with the config below (400 × 4 KiB +//! against a 512 KiB cap), all three faults reproduce: +//! +//! * `DBSIZE` answered 124 immediately after the writes were acked, then +//! climbed to 400 over ~3s with no further writes (this is the +//! `live 373 / recovered 400` that filed #459); +//! * `GET k0` → nil and `EXISTS k0` → 0, then 250ms later `GET k0` → 4096 +//! bytes with no intervening write; +//! * 277 of 400 `DEL`s answered `:0`, and after the queue drained all 277 +//! keys read back — the deletes were acknowledged and then undone. The +//! completion path inserts into `cold_index` unconditionally, so those +//! resurrections reach the manifest and survive restart. +//! +//! Wire-level on purpose: the window is a property of the live +//! evict → spill-thread → completion pipeline and cannot be built in a unit +//! test. Scoped to `--disk-offload enable` (opt-in), 1 shard, so every key +//! races the one spill queue. +//! +//! Run with (monoio default — matches the shipped runtime): +//! cargo build --release +//! MOON_BIN=$PWD/target/release/moon cargo test --release \ +//! --test spill_inflight_visibility + +#![allow(clippy::unwrap_used)] + +mod common; + +use std::io::{BufReader, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use common::find_moon_binary; + +// --------------------------------------------------------------------------- +// Server (pattern: tests/dbsize_offload_logical.rs) +// --------------------------------------------------------------------------- + +fn test_tmpdir() -> tempfile::TempDir { + let base = + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("target/spill-459-test-tmp"); + std::fs::create_dir_all(&base).expect("create spill-459-test-tmp base dir"); + tempfile::Builder::new() + .prefix("spill-459-") + .tempdir_in(&base) + .expect("tempdir_in target/spill-459-test-tmp") +} + +struct ServerGuard(Child); + +impl Drop for ServerGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +const MAXMEMORY_BYTES: u64 = 512 * 1024; // 512 KiB — forces spill fast. +const N_KEYS: usize = 400; +const VAL_SIZE: usize = 4096; // 400 × 4KiB ≈ 1.6 MiB » 512 KiB cap → heavy spill. + +fn spawn_moon_offload(dir: &std::path::Path) -> (ServerGuard, u16) { + let (child, port) = common::spawn_listening(|port| { + Command::new(find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + // 1 shard: every key contends the same spill queue, so the + // in-flight window is wide and the race is not sharded away. + "--shards", + "1", + // Spill is INERT without a durability backstop — the + // async-spill path bails unless a ShardManifest exists, + // which needs --appendonly yes. + "--appendonly", + "yes", + "--disk-offload", + "enable", + "--maxmemory", + &MAXMEMORY_BYTES.to_string(), + "--maxmemory-policy", + "allkeys-lru", + "--maxmemory-samples", + "200", + "--disk-free-min-pct", + "0", + ]) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon") + }); + (ServerGuard(child), port) +} + +// --------------------------------------------------------------------------- +// Minimal RESP client (pattern: tests/dbsize_offload_logical.rs) +// --------------------------------------------------------------------------- + +#[derive(Debug, PartialEq, Clone)] +enum V { + Simple(String), + Err(String), + Int(i64), + Bulk(Vec), + Arr(Vec), + Null, +} + +struct Client { + reader: BufReader, + writer: TcpStream, +} + +impl Client { + fn try_connect(port: u16, window: Duration) -> Option { + let addr = ("127.0.0.1", port) + .to_socket_addrs() + .ok()? + .next() + .expect("resolve loopback"); + let start = Instant::now(); + let stream = loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => break s, + Err(_) if start.elapsed() < window => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(_) => return None, + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(30))) + .unwrap(); + let writer = stream.try_clone().unwrap(); + Some(Client { + reader: BufReader::new(stream), + writer, + }) + } + + fn encode(args: &[&[u8]]) -> Vec { + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n", a.len()).as_bytes()); + out.extend_from_slice(a); + out.extend_from_slice(b"\r\n"); + } + out + } + + fn read_line(&mut self) -> String { + let mut line = Vec::new(); + let mut b = [0u8; 1]; + loop { + self.reader.read_exact(&mut b).expect("read byte"); + if b[0] == b'\n' { + break; + } + if b[0] != b'\r' { + line.push(b[0]); + } + } + String::from_utf8_lossy(&line).into_owned() + } + + fn parse(&mut self) -> V { + let line = self.read_line(); + let (t, rest) = line.split_at(1); + match t { + "+" => V::Simple(rest.to_string()), + "-" => V::Err(rest.to_string()), + ":" => V::Int(rest.parse().expect("int")), + "$" => { + let n: i64 = rest.parse().expect("bulk len"); + if n < 0 { + return V::Null; + } + let mut buf = vec![0u8; n as usize + 2]; + self.reader.read_exact(&mut buf).expect("bulk body"); + buf.truncate(n as usize); + V::Bulk(buf) + } + "*" => { + let n: i64 = rest.parse().expect("arr len"); + if n < 0 { + return V::Null; + } + V::Arr((0..n).map(|_| self.parse()).collect()) + } + other => panic!("unexpected RESP type {other:?} (line {line:?})"), + } + } + + fn cmd(&mut self, args: &[&[u8]]) -> V { + self.writer.write_all(&Self::encode(args)).expect("send"); + self.parse() + } + + /// Send every command, THEN read every reply. Returning from this means + /// the server acknowledged all of them. + fn pipeline(&mut self, cmds: &[Vec>]) -> Vec { + let mut out = Vec::with_capacity(cmds.len()); + for c in cmds { + let refs: Vec<&[u8]> = c.iter().map(|a| a.as_slice()).collect(); + self.writer.write_all(&Self::encode(&refs)).expect("send"); + } + for _ in cmds { + out.push(self.parse()); + } + out + } + + fn try_ping(&mut self) -> std::io::Result { + self.writer.write_all(b"*1\r\n$4\r\nPING\r\n")?; + let mut buf = [0u8; 7]; + self.reader.read_exact(&mut buf)?; + Ok(&buf == b"+PONG\r\n") + } +} + +fn readiness_deadline() -> Duration { + if std::env::var_os("CI").is_some() { + Duration::from_secs(120) + } else { + Duration::from_secs(30) + } +} + +fn wait_ready(guard: &mut ServerGuard, dir: &std::path::Path, port: u16) -> Client { + let deadline = Instant::now() + readiness_deadline(); + loop { + if let Ok(Some(status)) = guard.0.try_wait() { + let tail = std::fs::read_to_string(dir.join("moon.stderr.log")) + .unwrap_or_else(|e| format!("")); + panic!("moon exited {status} before ready; stderr tail:\n{tail}"); + } + if let Some(mut c) = Client::try_connect(port, Duration::from_secs(2)) + && c.try_ping().unwrap_or(false) + { + return c; + } + assert!( + Instant::now() < deadline, + "moon never answered PING on port {port}" + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +fn key_name(i: usize) -> Vec { + format!("k{i}").into_bytes() +} + +/// Write every key and read every reply, so all writes are ACKED on return. +/// Whatever the server does with them afterwards, it already promised the +/// client these keys exist. +fn load_keys(c: &mut Client) { + let val = vec![b'x'; VAL_SIZE]; + let cmds: Vec>> = (0..N_KEYS) + .map(|i| vec![b"SET".to_vec(), key_name(i), val.clone()]) + .collect(); + for (i, r) in c.pipeline(&cmds).into_iter().enumerate() { + assert_eq!(r, V::Simple("OK".into()), "SET k{i} was not acknowledged"); + } +} + +/// Drain window for the spill queue + completion processing. Generous: these +/// tests must fail on the DEFECT, never on a slow machine. +fn drain() { + std::thread::sleep(Duration::from_secs(5)); +} + +// --------------------------------------------------------------------------- +// Fault 1 — a live key must never be denied +// --------------------------------------------------------------------------- + +/// A key that answers nil and then answers a value on retry, with no write +/// in between, was denied while it existed. Retry-and-compare is the +/// unambiguous form: it needs no knowledge of which keys spilled, and it +/// cannot be explained by legitimate `allkeys-lru` eviction — an evicted key +/// stays gone, it does not come back on its own. +#[test] +fn a_key_that_answers_nil_must_not_come_back_on_a_bare_retry() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + load_keys(&mut c); + + let mut liars = Vec::new(); + for i in 0..N_KEYS { + let k = key_name(i); + if c.cmd(&[b"GET", &k]) != V::Null { + continue; + } + let existed = c.cmd(&[b"EXISTS", &k]); + std::thread::sleep(Duration::from_millis(250)); + if let V::Bulk(v) = c.cmd(&[b"GET", &k]) { + liars.push((i, existed.clone(), v.len())); + } + } + + assert!( + liars.is_empty(), + "{} key(s) answered nil and then returned a value on a bare retry, \ + with no write in between — the server denied a key it was holding. \ + First offenders (key, EXISTS at the denial, bytes on retry): {:?}", + liars.len(), + &liars[..liars.len().min(5)] + ); +} + +// --------------------------------------------------------------------------- +// Fault 2 — an acknowledged DEL must be final +// --------------------------------------------------------------------------- + +/// The severe one. DEL every key while spills are still in flight, then let +/// the queue drain. A key that reads back afterwards was resurrected by the +/// completion path, which inserts into `cold_index` unconditionally — so the +/// resurrection is committed to the manifest, not transient. +/// +/// Asserted on readability rather than on the DEL return codes: a DEL that +/// answers `:0` for a live key is the same defect, but a client that deleted +/// data and got it back is the harm worth pinning. +#[test] +fn a_deleted_key_must_not_come_back_after_the_spill_queue_drains() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + load_keys(&mut c); + + // Immediately, while spills are still queued. + let dels: Vec>> = (0..N_KEYS) + .map(|i| vec![b"DEL".to_vec(), key_name(i)]) + .collect(); + let denied = c + .pipeline(&dels) + .into_iter() + .filter(|r| *r == V::Int(0)) + .count(); + + drain(); + + let gets: Vec>> = (0..N_KEYS) + .map(|i| vec![b"GET".to_vec(), key_name(i)]) + .collect(); + let resurrected: Vec = c + .pipeline(&gets) + .into_iter() + .enumerate() + .filter_map(|(i, r)| (r != V::Null).then_some(i)) + .collect(); + + let dbsize = c.cmd(&[b"DBSIZE"]); + + assert!( + resurrected.is_empty(), + "{} key(s) were readable again after DEL + drain (DBSIZE={:?}); \ + {denied} DEL(s) had answered :0. A key deleted by an acknowledged \ + command came back on its own. First: {:?}", + resurrected.len(), + dbsize, + &resurrected[..resurrected.len().min(8)] + ); + assert_eq!( + dbsize, + V::Int(0), + "every key was deleted, so DBSIZE must be 0" + ); +} + +// --------------------------------------------------------------------------- +// Fault 3 — the counter must not under-report acked writes +// --------------------------------------------------------------------------- + +/// The symptom that filed #459. Every SET was acknowledged before DBSIZE is +/// asked, so no key may be missing from the count: `logical_len` sums hot + +/// cold and an in-flight key is in neither. +/// +/// Read twice with a drain between: a count that RISES with no writes in +/// between is proof the first answer was wrong, and distinguishes this from +/// legitimate eviction (which can only lower a count, never raise it). +#[test] +fn dbsize_must_not_undercount_keys_whose_spill_is_in_flight() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + load_keys(&mut c); + + let immediate = c.cmd(&[b"DBSIZE"]); + drain(); + let settled = c.cmd(&[b"DBSIZE"]); + + let (V::Int(immediate), V::Int(settled)) = (immediate.clone(), settled.clone()) else { + panic!("DBSIZE must answer an integer, got {immediate:?} / {settled:?}"); + }; + + assert!( + immediate >= settled, + "DBSIZE rose from {immediate} to {settled} with no writes in between — \ + the first answer omitted {} key(s) whose spill was still in flight. \ + Eviction can only lower a key count; nothing legitimate raises one.", + settled - immediate + ); + assert_eq!( + immediate, N_KEYS as i64, + "all {N_KEYS} SETs were acknowledged before DBSIZE was asked, so every \ + key must be counted (settled count was {settled})" + ); +} From 3c19dd9b1b3c1d59136710e43f062f159df6a6af Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 10 Aug 2026 19:14:37 +0700 Subject: [PATCH 2/3] fix(storage): wire spill_completion_superseded into INFO; review fixes (#459) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the PR #465 review. - INFO: `spill_completion_superseded` is now actually emitted next to its sibling `spill_failed_reinserted`. The counter and getter existed but were never wired into the INFO persistence section, so the CHANGELOG claimed a field that did not exist. Verified on a live server. - CHANGELOG: `[Unreleased]` had gained a second `### Fixed` heading. Keep a Changelog allows one section per change type per release, and a parser that indexes by heading silently drops one block. Folded into the existing section. - tests: replace the fixed 5s drain with a poll for a STABLE DBSIZE, and fail a refused SET with an explicit "this run never reached the in-flight window" message. Both failure modes were environmental (slow disk backs the spill queue up, `try_send` fails, eviction surfaces OOM) and would have been read as the defect returning. Re-confirmed the amended tests still fail on the pre-fix binary with identical diagnostics — 277 resurrections, retry-liars, DBSIZE 124->400 — so they still discriminate. Also ~3x faster (12.9s -> 3.9s) since the poll returns as soon as the queue is quiet. - docs: record the memory-accounting gap the review surfaced. The pending payload is resident RAM `used_memory` does not count: `db.remove()` credits back the whole entry cost while the queued request still holds a full `Bytes::copy_from_slice` of the value. Verified PRE-EXISTING on main, which makes the same copy and the same credit-back; this PR adds only the key `Bytes` plus a small struct per in-flight key, and moves the end of the retention window from "channel drained" to "completion applied". Charging the bytes needs `evict_to_budget` taught to stop while pending bytes are large, or it evicts in a runaway loop chasing memory that cannot drop yet — a design change, tracked in #466 rather than smuggled into a visibility fix. Verified: fmt clean; clippy clean on default features AND on runtime-tokio,jemalloc; the three wire-level tests green on monoio and tokio, and still red on the pre-fix binary. author: Tin Dang --- CHANGELOG.md | 64 +++++++++++++++--------------- src/command/connection.rs | 2 + src/storage/db/mod.rs | 15 +++++++ tests/spill_inflight_visibility.rs | 49 ++++++++++++++++++++--- 4 files changed, 91 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ede1719d..87bb91e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,39 +6,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### 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`. - ### Added - **Client-compat harness: raw-RESP diff against a real `redis-server` (`scripts/test-client-compat.sh`).** Moon's existing Redis comparison @@ -77,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 diff --git a/src/command/connection.rs b/src/command/connection.rs index 2d73c6ec..d9522438 100644 --- a/src/command/connection.rs +++ b/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"); diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs index 7b1c5049..8a3b4683 100644 --- a/src/storage/db/mod.rs +++ b/src/storage/db/mod.rs @@ -227,6 +227,21 @@ pub struct Database { /// to, so the value stays charged to RAM until the key is next written or /// deleted. Reads stay correct throughout; the cost is that the eviction /// did not actually reclaim that key's memory. + /// + /// ACCOUNTING (pre-existing, tracked in #466): the payload is resident RAM that + /// `used_memory` does not count. `evict_one_async_spill` calls + /// `db.remove()`, whose `remove_hot` credits back the whole + /// `entry_overhead`, while the queued `SpillRequest` still holds a full + /// `Bytes::copy_from_slice` of the value — so for the length of the + /// window the eviction loop believes it reclaimed memory it has not. + /// That was equally true before this plane existed (the request has + /// always owned that copy); what is added here is the key `Bytes` plus + /// this struct per in-flight key, and a retention window that now ends + /// when the event loop APPLIES the completion rather than when the spill + /// thread drains the channel. Charging the pending bytes honestly would + /// require teaching `evict_to_budget` to stop when pending bytes are + /// large, or it would evict in a runaway loop chasing memory that cannot + /// drop yet — a design change, deliberately out of scope for #459. spill_inflight: std::collections::HashMap, } diff --git a/tests/spill_inflight_visibility.rs b/tests/spill_inflight_visibility.rs index fc4d67a9..083a214c 100644 --- a/tests/spill_inflight_visibility.rs +++ b/tests/spill_inflight_visibility.rs @@ -279,14 +279,51 @@ fn load_keys(c: &mut Client) { .map(|i| vec![b"SET".to_vec(), key_name(i), val.clone()]) .collect(); for (i, r) in c.pipeline(&cmds).into_iter().enumerate() { + // A refused SET is an ENVIRONMENT failure, not a #459 finding: on a + // slow disk the spill queue backs up, `try_send` fails in + // `evict_one_async_spill`, and `evict_to_budget` surfaces OOM. Say so + // explicitly so the run is not misread as the defect reappearing. + if let V::Err(ref e) = r { + panic!( + "SET k{i} was refused ({e}). The spill queue could not keep up with the \ + write burst, so this run cannot say anything about #459 — it never \ + reached the in-flight window. Re-run on a less loaded machine." + ); + } assert_eq!(r, V::Simple("OK".into()), "SET k{i} was not acknowledged"); } } -/// Drain window for the spill queue + completion processing. Generous: these -/// tests must fail on the DEFECT, never on a slow machine. -fn drain() { - std::thread::sleep(Duration::from_secs(5)); +/// Wait for the spill queue and completion processing to go quiet. +/// +/// Polls for a STABLE `DBSIZE` rather than sleeping a fixed span: a fixed +/// sleep that is too short on a slow disk fails the test for a reason that is +/// not the defect, and one long enough to be safe everywhere wastes the +/// common case. Returns once the count holds steady across consecutive +/// samples, or at the deadline (the assertions then report whatever the +/// server actually settled on, which is still a truthful observation). +fn drain(c: &mut Client) { + let deadline = Instant::now() + Duration::from_secs(30); + let mut last = i64::MIN; + let mut stable = 0; + while Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(250)); + let V::Int(n) = c.cmd(&[b"DBSIZE"]) else { + continue; + }; + if n == last { + stable += 1; + // Four consecutive identical samples ≈ 1s of quiet. The spill + // thread flushes on a sub-second cadence, so a queue still + // draining moves this number. + if stable >= 4 { + return; + } + } else { + stable = 0; + last = n; + } + } } // --------------------------------------------------------------------------- @@ -359,7 +396,7 @@ fn a_deleted_key_must_not_come_back_after_the_spill_queue_drains() { .filter(|r| *r == V::Int(0)) .count(); - drain(); + drain(&mut c); let gets: Vec>> = (0..N_KEYS) .map(|i| vec![b"GET".to_vec(), key_name(i)]) @@ -409,7 +446,7 @@ fn dbsize_must_not_undercount_keys_whose_spill_is_in_flight() { load_keys(&mut c); let immediate = c.cmd(&[b"DBSIZE"]); - drain(); + drain(&mut c); let settled = c.cmd(&[b"DBSIZE"]); let (V::Int(immediate), V::Int(settled)) = (immediate.clone(), settled.clone()) else { From 36c2977c25e8e9a666a149317832da6eac6c6eb3 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 10 Aug 2026 23:49:59 +0700 Subject: [PATCH 3/3] test(storage): drain on completion progress, not DBSIZE alone (#459) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit re-review of #465 caught a soundness bug in the test's own drain condition, and it fails in the direction that matters. `DBSIZE` alone cannot tell "the spill pipeline is quiet" from "the pipeline has not started applying completions yet". In the deletion-finality test that distinction is the whole point: after the DEL sweep every key is either deleted or in-flight-invisible, so DBSIZE reads 0 IMMEDIATELY while completions are still queued. A stability gate can therefore return before any completion is applied, and a completion that republishes a deleted key lands after the assertions have already run — a FALSE PASS hiding the exact P0 this file exists to catch. It caught the bug anyway only because completions land inside the poll's ~1s floor. That is luck, not a guarantee, and it would have decayed silently on a slower or busier machine. - `drain` now gates on server-reported spill PROGRESS as well as DBSIZE: `spill_batches_flushed` + `spill_completion_superseded` from INFO must ALSO hold still, for six consecutive samples. Either counter moves while anything is left in the pipeline. - the deletion test gains a vacuity guard. "Nothing came back" only means something if the completions actually ARRIVED and were refused, so the test now asserts `spill_completion_superseded > 0` — the count of completions the event loop applied and declined to publish because DEL had retired their in-flight record, which is the mechanism under test. Without it, a run where nothing spilled, or where the queue drained before the DELs, passes while proving nothing. `info_counter` returns None for an absent field rather than defaulting to 0, so a missing counter cannot satisfy the guard. Verified: still RED on the pre-fix binary with identical diagnostics — 277 resurrections, retry-liars, DBSIZE 124->400 — and the deletion test still fails on the resurrection assertion, not on the new guard, so the ordering reports the defect rather than the guard. Green on monoio and tokio. fmt and clippy clean. author: Tin Dang --- tests/spill_inflight_visibility.rs | 78 ++++++++++++++++++++++++------ 1 file changed, 64 insertions(+), 14 deletions(-) diff --git a/tests/spill_inflight_visibility.rs b/tests/spill_inflight_visibility.rs index 083a214c..8fa96d62 100644 --- a/tests/spill_inflight_visibility.rs +++ b/tests/spill_inflight_visibility.rs @@ -294,34 +294,69 @@ fn load_keys(c: &mut Client) { } } -/// Wait for the spill queue and completion processing to go quiet. +/// One `INFO persistence` counter, or `None` when the field is absent (the +/// pre-fix binary has no `spill_completion_superseded`, and saying "absent" +/// beats defaulting to 0 and silently satisfying a guard). +fn info_counter(c: &mut Client, field: &str) -> Option { + let V::Bulk(body) = c.cmd(&[b"INFO", b"persistence"]) else { + return None; + }; + let text = String::from_utf8_lossy(&body); + let prefix = format!("{field}:"); + text.lines() + .find_map(|l| l.strip_prefix(&prefix)) + .and_then(|v| v.trim().parse().ok()) +} + +/// Total spill work the server reports having COMPLETED. /// -/// Polls for a STABLE `DBSIZE` rather than sleeping a fixed span: a fixed -/// sleep that is too short on a slow disk fails the test for a reason that is -/// not the defect, and one long enough to be safe everywhere wastes the -/// common case. Returns once the count holds steady across consecutive -/// samples, or at the deadline (the assertions then report whatever the -/// server actually settled on, which is still a truthful observation). +/// `spill_batches_flushed` counts batches the background thread wrote; +/// `spill_completion_superseded` counts completions the event loop applied +/// and refused to publish. Together they move whenever the pipeline makes +/// progress, which `DBSIZE` alone does not. +fn spill_progress(c: &mut Client) -> u64 { + info_counter(c, "spill_batches_flushed").unwrap_or(0) + + info_counter(c, "spill_completion_superseded").unwrap_or(0) +} + +/// Wait for the spill pipeline — queue AND completion application — to go +/// quiet. +/// +/// Gating on a stable `DBSIZE` alone is NOT sufficient, and is unsound in the +/// direction that matters. After the deletion test's `DEL` sweep every key is +/// either deleted or in-flight-invisible, so `DBSIZE` reads 0 immediately +/// while completions are still queued; a stability gate can return before any +/// completion is applied, and a completion that republishes a deleted key +/// then lands AFTER the assertions have run. That is a FALSE PASS hiding the +/// exact P0 this file exists to catch. (It happened to catch it anyway, +/// because completions land inside the poll's ~1s floor — luck, not a +/// guarantee.) +/// +/// So gate on server-reported spill PROGRESS as well: hold until neither the +/// progress counters nor `DBSIZE` have moved for several consecutive samples. +/// A fixed sleep is avoided for the same reason as before — too short fails +/// on a slow disk for a reason that is not the defect. fn drain(c: &mut Client) { let deadline = Instant::now() + Duration::from_secs(30); - let mut last = i64::MIN; + let mut last = (i64::MIN, u64::MAX); let mut stable = 0; while Instant::now() < deadline { std::thread::sleep(Duration::from_millis(250)); let V::Int(n) = c.cmd(&[b"DBSIZE"]) else { continue; }; - if n == last { + let now = (n, spill_progress(c)); + if now == last { stable += 1; - // Four consecutive identical samples ≈ 1s of quiet. The spill - // thread flushes on a sub-second cadence, so a queue still - // draining moves this number. - if stable >= 4 { + // Six consecutive identical samples ≈ 1.5s with BOTH signals + // still. The spill thread flushes on a sub-second cadence, so a + // pipeline with anything left in it moves one of them. + if stable >= 6 { return; } } else { stable = 0; - last = n; + last = now; } } } @@ -410,6 +445,13 @@ fn a_deleted_key_must_not_come_back_after_the_spill_queue_drains() { let dbsize = c.cmd(&[b"DBSIZE"]); + // Vacuity guard. "Nothing came back" is only meaningful if the spill + // completions actually ARRIVED and were refused — otherwise a run where + // the queue never drained, or never spilled at all, passes while proving + // nothing. `spill_completion_superseded` counts completions the event + // loop applied and declined to publish because DEL had retired their + // in-flight record, which is precisely the mechanism under test. + let superseded = info_counter(&mut c, "spill_completion_superseded"); assert!( resurrected.is_empty(), "{} key(s) were readable again after DEL + drain (DBSIZE={:?}); \ @@ -424,6 +466,14 @@ fn a_deleted_key_must_not_come_back_after_the_spill_queue_drains() { V::Int(0), "every key was deleted, so DBSIZE must be 0" ); + assert!( + superseded.is_some_and(|n| n > 0), + "the run proved nothing: spill_completion_superseded={superseded:?}, so no \ + completion was observed arriving and being refused. Either the DELs never \ + raced an in-flight spill (nothing spilled, or the queue drained first), or \ + the drain returned early. A green result here would not mean deletes are \ + final — it would mean the window was never entered." + ); } // ---------------------------------------------------------------------------