diff --git a/CHANGELOG.md b/CHANGELOG.md index 1949d4fcb..7f01581df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,62 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — cold-collection visibility: silent data loss on evicted Hash/List/Set/ZSet/Stream (task #41, P0) + +With disk-offload enabled (the default), the production eviction paths +(`evict_one_async_spill`, `evict_batch_durable_no_aof` in +`storage/eviction.rs`) spill Hash/List/Set/ZSet/Stream values to the cold +tier via `kv_serde::serialize_collection` — but the type-specific accessors +never consulted the cold index before this fix. Consequences: `HGET`/ +`HGETALL`/`LRANGE`/`SMEMBERS`/`ZRANGE`/`EXISTS` reported the key absent +after eviction even though it was durably spilled, and `HSET`/`LPUSH`/ +`SADD`/`ZADD` silently fabricated a **new empty container**, permanently +shadowing (destroying) the cold copy on the next flush — a genuine, +silent user-data-loss bug for any collection key subject to eviction. + +Fixed with a single promote-if-cold hook rather than rewriting every +accessor: + +- `Database::promote_cold_if_present` — on a hot miss, does one + `ColdIndex` lookup; on a hit it decodes the cold value, installs it hot, + and removes the cold-index entry (single owning copy, no dual + reference). All 8 `get_or_create_*`/`get_*` mutable accessors + (`get_or_create_hash[_listpack]`, `get_or_create_list[_listpack]`, + `get_or_create_set`/`get_or_create_intset`, `get_or_create_sorted_set`, + `get_or_create_stream`, `get_hash`, `get_list`, `get_set`, + `get_sorted_set`, `get_stream[_mut]`) now call this hook before falling + through to their existing "fabricate new" path — so HSET/LPUSH/SADD/ZADD/ + XADD merge into the promoted collection instead of shadowing it. +- The four `&self`-typed "`*_ref_if_alive`" shared-read accessors + (`get_hash_ref_if_alive`, `get_list_ref_if_alive`, `get_set_ref_if_alive`, + `get_sorted_set_ref_if_alive`, `get_stream_if_alive`) cannot promote + (they're also called from the `_readonly` dispatch path holding only a + shared `&Database`), so they instead do a **non-promoting** cold + read-through: new `Owned`/`Borrowed` variants on `HashRef`/`ListRef`/ + `SetRef`/`SortedSetRef` (and a new `StreamRef<'a>` `Deref` wrapper + replacing the old `&StreamData` return type) let a cold hit return a + freshly-decoded owned value without a backing hot entry, at the cost of + exactly one extra branch + one `ColdIndex` lookup on a hot miss. Hot-hit + cost is unchanged (single probe, same as before). +- `exists`/`exists_if_alive` now fall back to a cheap + `cold_contains_alive` check (ColdIndex presence + TTL only, no disk I/O, + no promotion) instead of unconditionally returning `false` on a hot + miss. +- `zrank_readonly`/`zrevrank_readonly` (`command/sorted_set/sorted_set_read.rs`) + updated to route the new `SortedSetRef::Owned` variant through the same + O(n) fallback as the existing `Listpack` variant (previously any + unmatched variant silently fell into a `Frame::Null` wildcard arm — + would have reported a promoted member as not-ranked). + +New tests: 13 unit tests in `storage::db::tests` (spill-then-promote for +all 5 collection types + `EXISTS`-without-promoting) plus a new black-box +suite `tests/cold_collection_visibility.rs` (real server, +`--disk-offload enable`, sampled-LRU-driven eviction, asserts `EXISTS`, +read-after-evict, and write-merges-with-promoted-data for Hash/List/Set/ +ZSet). Explicitly out of scope for this fix (unchanged): +`recovery.rs` rehydration, replication, and the eviction gates' Wave A +reporting sinks. + ### Fixed — Wave A adversarial-review fixes (task #34) Three defects found reviewing Wave A (plane replication) before merge, all diff --git a/src/command/sorted_set/sorted_set_read.rs b/src/command/sorted_set/sorted_set_read.rs index 6bb06301b..2eae21042 100644 --- a/src/command/sorted_set/sorted_set_read.rs +++ b/src/command/sorted_set/sorted_set_read.rs @@ -673,8 +673,14 @@ pub fn zrank_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { None => Frame::Null, } } - (SortedSetRef::Listpack(_), Some(score)) => { - // For listpack, compute rank from sorted entries + // Listpack has no O(log n) rank structure; Owned (P0 + // cold-collection-visibility fix: a value decoded fresh from + // the cold tier, see `SortedSetRef::Owned`) carries its own + // `BPTree` but `members_map`/`bptree` deliberately return + // `None` for it (same as Listpack) — both fall back to the + // same O(n) rank-from-sorted-entries computation. + (SortedSetRef::Listpack(_), Some(score)) + | (SortedSetRef::Owned { .. }, Some(score)) => { let entries = zref.entries_sorted(); let target_score = OrderedFloat(score); let target_member = Bytes::copy_from_slice(member); @@ -715,7 +721,11 @@ pub fn zrevrank_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { None => Frame::Null, } } - (SortedSetRef::Listpack(_), Some(score)) => { + // See the matching comment in `zrank_readonly`: Owned (P0 + // cold-collection-visibility fix) falls back to the same + // generic path as Listpack. + (SortedSetRef::Listpack(_), Some(score)) + | (SortedSetRef::Owned { .. }, Some(score)) => { let entries = zref.entries_sorted(); let target_score = OrderedFloat(score); let target_member = Bytes::copy_from_slice(member); diff --git a/src/storage/db.rs b/src/storage/db.rs index 6b03a7600..747450bbd 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -1,7 +1,7 @@ use bytes::Bytes; use std::collections::{HashMap, HashSet, VecDeque}; -pub use super::db_read::{HashRef, ListRef, SetRef, SortedSetRef}; +pub use super::db_read::{HashRef, ListRef, SetRef, SortedSetRef, StreamRef}; use super::bptree::BPTree; use super::compact_key::CompactKey; @@ -865,47 +865,127 @@ 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().map(|ci| { - crate::storage::tiered::cold_read::cold_read_through_outcome( - ci, shard_dir, key, now_ms, - ) - }) - }); - 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) - } - 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 - } - Some(ColdReadOutcome::Miss) | None => None, + // Cold fallback: promote from disk into hot RAM if spilled + // there by eviction, then re-probe. `promote_cold_if_present` + // also owns reclaiming a stale (TTL-expired) cold-index entry. + self.promote_cold_if_present(key, now_ms); + self.data.get(key) + } + } + } + + /// If `key` is missing from hot RAM but present in the cold tier + /// (spilled there by eviction), read it from disk and promote it into + /// hot RAM, removing the cold-index entry — this is the cold-fallback + /// branch [`Self::get`] has always used, factored out so every mutable + /// accessor can share it. + /// + /// Returns `true` if `key` is present in hot RAM after this call returns + /// (either because it already was, or because promotion just happened). + /// Returns `false` for a genuine miss (absent from both tiers), or when + /// the cold entry's TTL had already passed — the stale index entry is + /// reclaimed as a byproduct in that case, same as the expired-hot branch + /// above. + /// + /// Cheap on the common case callers actually care about (key already + /// hot): a single `contains_key` probe short-circuits before ever + /// touching the cold tier. Only a genuine miss pays for the `ColdIndex` + /// lookup + a blocking disk `pread`. + /// + /// P0 fix (`.planning/reviews/storage-audit-2026-07-12-kv.md`): + /// `get_or_create_hash`/`_list`/`_set`/`_sorted_set`/`_stream` (and their + /// compact-encoding siblings `get_or_create_hash_listpack`, + /// `get_or_create_list_listpack`, `get_or_create_intset`) used to skip + /// straight to fabricating a brand-new EMPTY container whenever + /// `self.data` missed the key — even when the real value was sitting + /// right there in the cold tier. The fabricated container then got + /// written back over the cold copy on the next `set`/mutation, + /// permanently destroying it. Calling this method first closes that gap. + pub fn promote_cold_if_present(&mut self, key: &[u8], now_ms: u64) -> bool { + if self.data.contains_key(key) { + return true; + } + use crate::storage::tiered::cold_read::ColdReadOutcome; + // Extract an owned result first to drop immutable borrows of + // `self.cold_index` / `self.cold_shard_dir` before the mutation below. + let cold_result = self.cold_shard_dir.as_ref().and_then(|shard_dir| { + self.cold_index.as_ref().map(|ci| { + crate::storage::tiered::cold_read::cold_read_through_outcome( + ci, shard_dir, key, now_ms, + ) + }) + }); + 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); + } + true + } + 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); + } + false } + Some(ColdReadOutcome::Miss) | None => false, } } + /// Cheap (no disk I/O, no promotion) check for whether `key` is present + /// in the cold tier and not yet TTL-expired. + /// + /// Used by `EXISTS`: a spilled key logically exists even though it has + /// no in-RAM `Entry`, but `EXISTS` doesn't need the *value* — paying for + /// a disk read (or, worse, promoting the key into RAM) just to answer a + /// boolean would be wasteful. A single `HashMap` probe against the + /// in-RAM `ColdIndex` (which caches `ttl_ms` at insert time, see + /// [`crate::storage::tiered::cold_index::ColdLocation::ttl_ms`]) is + /// enough. Does not reclaim an expired entry (that needs `&mut self` + /// and disk access to do safely) — the proactive sweep and the + /// promoting paths handle reclamation. + #[inline] + fn cold_contains_alive(&self, key: &[u8], now_ms: u64) -> bool { + let Some(ci) = self.cold_index.as_ref() else { + return false; + }; + match ci.lookup(key) { + Some(loc) => loc.ttl_ms.is_none_or(|ttl| now_ms <= ttl), + None => false, + } + } + + /// Non-promoting cold read-through for the `&self` "*_ref_if_alive" + /// accessors: decodes a spilled value from disk WITHOUT touching hot RAM + /// or the cold index. Thin wrapper over [`Self::get_cold_value`] using + /// this `Database`'s own cached clock semantics is left to the caller + /// (they already have `now_ms` in hand); kept as a private alias so the + /// call sites below read as "cold read" rather than repeating the + /// `cold_shard_dir`/`cold_index` plumbing. + /// + /// These accessors back BOTH the exclusive-dispatch path (`&mut + /// Database`, which downgrades to `&self` for the call) AND the + /// RwLock-shared-read dispatch path (`&Database` only — see + /// `dispatch_read` / `*_readonly` command handlers). The latter cannot + /// 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`]). + #[inline] + fn cold_read_only(&self, key: &[u8], now_ms: u64) -> Option { + self.get_cold_value(key, now_ms) + } + /// Get a mutable reference to an entry by key, performing lazy expiration and access tracking. /// /// Returns `None` if the key does not exist or has expired. @@ -1094,19 +1174,23 @@ impl Database { /// Check if a key exists, performing lazy expiration. /// Optimized: single lookup via get instead of check_expired + contains_key. + /// + /// A cold-only key (spilled by eviction, no in-RAM `Entry`) still counts + /// as existing — checked via the cheap in-RAM [`Self::cold_contains_alive`] + /// (no disk I/O, no promotion; P0 cold-collection-visibility fix). #[allow(clippy::unwrap_used)] // remove() after get() returned Some — key guaranteed present pub fn exists(&mut self, key: &[u8]) -> bool { let now_ms = self.cached_now_ms; let base_ts = self.base_timestamp; match self.data.get(key) { - None => false, + None => self.cold_contains_alive(key, now_ms), Some(entry) => { if entry.is_expired_at(base_ts, now_ms) { let removed = self.data.remove(key).unwrap(); self.used_memory = self .used_memory .saturating_sub(entry_overhead(key, &removed)); - false + self.cold_contains_alive(key, now_ms) } else { true } @@ -1253,10 +1337,17 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_hash(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: a hash may have been spilled to the cold tier by + // eviction. Promote it back to hot RAM before fabricating an + // empty container, or HSET on an evicted hash would silently + // destroy the cold copy (see `Self::promote_cold_if_present`). + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_hash(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = match self.data.get_mut(key) { Some(e) => e, @@ -1297,6 +1388,11 @@ impl Database { /// Get a hash entry (read-only). Returns None if key missing, Err if wrong type. /// Upgrades compact encoding to full HashMap if found. + /// + /// Promotes a cold-spilled hash back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`, so + /// unlike the enum-based `get_hash_ref_if_alive` it can promote directly + /// instead of decoding a throwaway copy on every call. #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade pub fn get_hash(&mut self, key: &[u8]) -> Result>, Frame> { let now_ms = self.cached_now_ms; @@ -1306,6 +1402,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } // Upgrade compact encoding if present if let Some(entry) = self.data.get_mut(key) { if let Some(RedisValue::HashListpack(lp)) = entry.value.as_redis_value_mut() { @@ -1339,10 +1438,15 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_list(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled list before fabricating an empty + // one (see `Self::promote_cold_if_present`). + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_list(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); // Upgrade compact listpack to full VecDeque if needed @@ -1358,6 +1462,9 @@ impl Database { /// Get a list entry (read-only). Returns None if key missing, Err if wrong type. /// Upgrades compact encoding to full VecDeque if found. + /// + /// Promotes a cold-spilled list back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`. #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade pub fn get_list(&mut self, key: &[u8]) -> Result>, Frame> { let now_ms = self.cached_now_ms; @@ -1367,6 +1474,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } // Upgrade compact encoding if present if let Some(entry) = self.data.get_mut(key) { if let Some(RedisValue::ListListpack(lp)) = entry.value.as_redis_value_mut() { @@ -1395,10 +1505,15 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_set(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled set before fabricating an empty + // one (see `Self::promote_cold_if_present`). + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_set(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); // Upgrade compact encodings to full HashSet @@ -1421,6 +1536,9 @@ impl Database { /// Get a set entry (read-only). Returns None if key missing, Err if wrong type. /// Upgrades compact encodings to full HashSet if found. + /// + /// Promotes a cold-spilled set back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`. #[allow(clippy::unwrap_used)] // as_redis_value_mut() on known compact type during upgrade pub fn get_set(&mut self, key: &[u8]) -> Result>, Frame> { let now_ms = self.cached_now_ms; @@ -1430,6 +1548,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } // Upgrade compact encodings if present if let Some(entry) = self.data.get_mut(key) { match entry.value.as_redis_value_mut() { @@ -1467,10 +1588,19 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_set_intset(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled set before fabricating an empty + // intset — a promoted value always decodes as `RedisValue::Set` + // (cold storage never persists the intset compact encoding), so + // it naturally falls into the `Ok(None)` "not an intset, caller + // should use get_or_create_set" arm below, which already routes + // callers (e.g. SADD) to `get_or_create_set` — no fabrication. + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_set_intset(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); match entry.value.as_redis_value_mut() { @@ -1516,10 +1646,19 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_hash_listpack(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled hash before fabricating an + // empty listpack — a promoted value always decodes as + // `RedisValue::Hash` (cold storage never persists the listpack + // compact encoding), so it naturally falls into the `Ok(None)` + // "not a listpack, fall through" arm below, which already routes + // callers (e.g. HSET) to `get_or_create_hash` — no fabrication. + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_hash_listpack(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); match entry.value.as_redis_value_mut() { @@ -1567,10 +1706,19 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_list_listpack(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled list before fabricating an + // empty listpack — a promoted value always decodes as + // `RedisValue::List` (cold storage never persists the listpack + // compact encoding), so it naturally falls into the `Ok(None)` + // "not a listpack, fall through" arm below, which already routes + // callers (e.g. LPUSH) to `get_or_create_list` — no fabrication. + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_list_listpack(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); match entry.value.as_redis_value_mut() { @@ -1616,10 +1764,15 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_sorted_set_bptree(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled sorted set before fabricating + // an empty one (see `Self::promote_cold_if_present`). + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_sorted_set_bptree(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } // Upgrade old SortedSet (BTreeMap) to SortedSetBPTree on access let entry = self.data.get_mut(key).unwrap(); @@ -1642,6 +1795,9 @@ impl Database { } /// Get a sorted set entry (read-only). Returns None if key missing, Err if wrong type. + /// + /// Promotes a cold-spilled sorted set back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`. #[allow(clippy::unwrap_used)] // get_mut() after confirmed existence; legacy BTreeMap upgrade is infallible pub fn get_sorted_set( &mut self, @@ -1654,6 +1810,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } // Upgrade old SortedSet (BTreeMap) to SortedSetBPTree on access if let Some(entry) = self.data.get(key) { if matches!( @@ -1798,12 +1957,17 @@ impl Database { } /// Read-only existence check: returns false if expired. + /// + /// Also counts a cold-only key (spilled by eviction) as existing — see + /// [`Self::cold_contains_alive`] (P0 cold-collection-visibility fix). + /// Used by the RwLock-shared-read `EXISTS` dispatch path, which cannot + /// mutate `self` to promote. pub fn exists_if_alive(&self, key: &[u8], now_ms: u64) -> bool { let base_ts = self.base_timestamp; - self.data - .get(key) - .map(|e| !e.is_expired_at(base_ts, now_ms)) - .unwrap_or(false) + match self.data.get(key) { + Some(e) if !e.is_expired_at(base_ts, now_ms) => true, + _ => self.cold_contains_alive(key, now_ms), + } } /// Read-only hash access. Returns None if key missing or expired, Err if wrong type. @@ -1894,16 +2058,25 @@ impl Database { /// `HashWithTtl` (HashMap with per-field TTL sidecar). For `HashWithTtl` /// returns a `HashRef::WithTtl` that filters expired fields on every /// field-level operation without mutating the database (lazy expiry). + /// + /// Takes `&self` because this backs BOTH the exclusive-dispatch path + /// (`&mut Database`, reborrowed) AND the RwLock-shared-read dispatch + /// path (`&Database` only — `hget_readonly`/`hgetall_readonly`/etc.). + /// The latter cannot promote a cold hit into hot RAM, so a hot miss + /// falls back to a non-promoting cold read-through returning + /// `HashRef::Owned`/`OwnedWithTtl` (P0 cold-collection-visibility fix) — + /// the fast path (key present hot) still costs exactly one probe. pub fn get_hash_ref_if_alive( &self, key: &[u8], now_ms: u64, ) -> Result>, Frame> { let base_ts = self.base_timestamp; - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(base_ts, now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(base_ts, now_ms) { + return Ok(None); + } + return match entry.value.as_redis_value() { RedisValueRef::Hash(map) => Ok(Some(HashRef::Map(map))), RedisValueRef::HashListpack(lp) => Ok(Some(HashRef::Listpack(lp))), RedisValueRef::HashWithTtl { @@ -1917,58 +2090,98 @@ impl Database { min_expiry_ms, })), _ => Err(Self::wrongtype_error()), - }, + }; + } + match self.cold_read_only(key, now_ms) { + Some(RedisValue::Hash(map)) => Ok(Some(HashRef::Owned(map))), + Some(RedisValue::HashWithTtl { + fields, + ttls, + min_expiry_ms, + }) => Ok(Some(HashRef::OwnedWithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + })), + Some(_) => Err(Self::wrongtype_error()), + None => Ok(None), } } /// Read-only list access via ListRef enum. Handles both VecDeque and Listpack. + /// + /// See [`Self::get_hash_ref_if_alive`] for why this consults the cold + /// tier without promoting on a hot miss (P0 cold-collection-visibility + /// fix). pub fn get_list_ref_if_alive( &self, key: &[u8], now_ms: u64, ) -> Result>, Frame> { let base_ts = self.base_timestamp; - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(base_ts, now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(base_ts, now_ms) { + return Ok(None); + } + return match entry.value.as_redis_value() { RedisValueRef::List(list) => Ok(Some(ListRef::Deque(list))), RedisValueRef::ListListpack(lp) => Ok(Some(ListRef::Listpack(lp))), _ => Err(Self::wrongtype_error()), - }, + }; + } + match self.cold_read_only(key, now_ms) { + Some(RedisValue::List(list)) => Ok(Some(ListRef::Owned(list))), + Some(_) => Err(Self::wrongtype_error()), + None => Ok(None), } } /// Read-only set access via SetRef enum. Handles HashSet, Listpack, and Intset. + /// + /// See [`Self::get_hash_ref_if_alive`] for why this consults the cold + /// tier without promoting on a hot miss (P0 cold-collection-visibility + /// fix). pub fn get_set_ref_if_alive( &self, key: &[u8], now_ms: u64, ) -> Result>, Frame> { let base_ts = self.base_timestamp; - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(base_ts, now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(base_ts, now_ms) { + return Ok(None); + } + return match entry.value.as_redis_value() { RedisValueRef::Set(set) => Ok(Some(SetRef::Hash(set))), RedisValueRef::SetListpack(lp) => Ok(Some(SetRef::Listpack(lp))), RedisValueRef::SetIntset(is) => Ok(Some(SetRef::Intset(is))), _ => Err(Self::wrongtype_error()), - }, + }; + } + match self.cold_read_only(key, now_ms) { + Some(RedisValue::Set(set)) => Ok(Some(SetRef::Owned(set))), + Some(_) => Err(Self::wrongtype_error()), + None => Ok(None), } } /// Read-only sorted set access via SortedSetRef enum. Handles BPTree, Listpack, and Legacy. + /// + /// See [`Self::get_hash_ref_if_alive`] for why this consults the cold + /// tier without promoting on a hot miss (P0 cold-collection-visibility + /// fix). pub fn get_sorted_set_ref_if_alive( &self, key: &[u8], now_ms: u64, ) -> Result>, Frame> { let base_ts = self.base_timestamp; - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(base_ts, now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(base_ts, now_ms) { + return Ok(None); + } + return match entry.value.as_redis_value() { RedisValueRef::SortedSetBPTree { tree, members } => { Ok(Some(SortedSetRef::BPTree { tree, members })) } @@ -1977,7 +2190,14 @@ impl Database { Ok(Some(SortedSetRef::Legacy { members, scores })) } _ => Err(Self::wrongtype_error()), - }, + }; + } + match self.cold_read_only(key, now_ms) { + Some(RedisValue::SortedSetBPTree { tree, members }) => { + Ok(Some(SortedSetRef::Owned { tree, members })) + } + Some(_) => Err(Self::wrongtype_error()), + None => Ok(None), } } @@ -2075,10 +2295,15 @@ impl Database { } } if !self.data.contains_key(key) { - let entry = Entry::new_stream(); - let k = CompactKey::from(key); - self.used_memory += entry_overhead(key, &entry); - self.data.insert(k, entry); + // P0 fix: promote a cold-spilled stream before fabricating an + // empty one (see `Self::promote_cold_if_present`). + self.promote_cold_if_present(key, now_ms); + if !self.data.contains_key(key) { + let entry = Entry::new_stream(); + let k = CompactKey::from(key); + self.used_memory += entry_overhead(key, &entry); + self.data.insert(k, entry); + } } let entry = self.data.get_mut(key).unwrap(); match entry.value.as_redis_value_mut() { @@ -2092,24 +2317,41 @@ impl Database { /// Checks expiry using `now_ms` but does NOT remove the expired key or /// touch LRU (mirrors `get_if_alive` semantics). Returns `Ok(None)` for /// missing or expired keys, `Err(WRONGTYPE)` for non-stream keys. + /// + /// Returns `StreamRef` rather than `&StreamData`: this backs both the + /// exclusive-dispatch path and the RwLock-shared-read dispatch path + /// (`&Database` only, cannot promote). A hot miss falls back to a + /// non-promoting cold read-through returning `StreamRef::Owned` (P0 + /// cold-collection-visibility fix) — `StreamRef` derefs to `&StreamData` + /// so existing call sites (`stream.length`, `stream.range(..)`, ...) are + /// unaffected. The fast path (key present hot) still costs one probe. pub fn get_stream_if_alive( &self, key: &[u8], now_ms: u64, - ) -> Result, Frame> { + ) -> Result>, Frame> { let base_ts = self.base_timestamp; - match self.data.get(key) { - None => Ok(None), - Some(entry) if entry.is_expired_at(base_ts, now_ms) => Ok(None), - Some(entry) => match entry.value.as_redis_value() { - RedisValueRef::Stream(s) => Ok(Some(s)), + if let Some(entry) = self.data.get(key) { + if entry.is_expired_at(base_ts, now_ms) { + return Ok(None); + } + return match entry.value.as_redis_value() { + RedisValueRef::Stream(s) => Ok(Some(StreamRef::Borrowed(s))), _ => Err(Self::wrongtype_error()), - }, + }; + } + match self.cold_read_only(key, now_ms) { + Some(RedisValue::Stream(s)) => Ok(Some(StreamRef::Owned(s))), + Some(_) => Err(Self::wrongtype_error()), + None => Ok(None), } } /// Get a read-only reference to a stream. Returns Ok(None) if key doesn't exist. /// Returns WRONGTYPE error if key holds another type. + /// + /// Promotes a cold-spilled stream back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`. pub fn get_stream(&mut self, key: &[u8]) -> Result, Frame> { let now_ms = self.cached_now_ms; let base_ts = self.base_timestamp; @@ -2118,6 +2360,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } match self.data.get(key) { None => Ok(None), Some(entry) => match entry.value.as_redis_value() { @@ -2128,6 +2373,9 @@ impl Database { } /// Get a mutable reference to an existing stream. Returns Ok(None) if key doesn't exist. + /// + /// Promotes a cold-spilled stream back to hot RAM on miss (P0 + /// cold-collection-visibility fix) — this accessor takes `&mut self`. pub fn get_stream_mut(&mut self, key: &[u8]) -> Result, Frame> { let now_ms = self.cached_now_ms; let base_ts = self.base_timestamp; @@ -2136,6 +2384,9 @@ impl Database { self.used_memory = self.used_memory.saturating_sub(entry_overhead(key, &entry)); } } + if !self.data.contains_key(key) { + self.promote_cold_if_present(key, now_ms); + } match self.data.get_mut(key) { None => Ok(None), Some(entry) => match entry.value.as_redis_value_mut() { @@ -2411,4 +2662,328 @@ mod tests { assert_eq!(db.now(), clock.secs()); assert_eq!(db.now_ms(), clock.ms()); } + + // ----------------------------------------------------------------- + // P0 cold-collection-visibility fix + // (.planning/reviews/storage-audit-2026-07-12-kv.md) + // + // Disk-offload spills Hash/List/Set/ZSet/Stream via + // `evict_one_async_spill` / `evict_batch_durable_no_aof` (production + // paths, `kv_serde::serialize_collection` handles every type), but the + // type-specific accessors used to consult only `self.data`, never the + // `ColdIndex` — so a spilled collection read as absent, and a write + // fabricated a brand-new EMPTY container that permanently shadowed + // (destroyed) the cold copy on the next `set`. + // + // These tests spill a collection directly via the same + // `spill_to_datafile` production helper the eviction paths use (mirrors + // the pattern in `tiered::cold_read`'s own tests), bypassing the + // eviction machinery itself (out of scope here) while exercising the + // exact on-disk format the accessors must decode. + // ----------------------------------------------------------------- + + use crate::persistence::manifest::ShardManifest; + use crate::storage::compact_value::CompactValue; + use crate::storage::entry::RedisValue as TestRedisValue; + use crate::storage::tiered::cold_index::ColdIndex; + use crate::storage::tiered::kv_spill::spill_to_datafile; + + /// Build a `Database` with an active cold tier holding one spilled + /// collection at `key`. Mirrors `cold_read::tests::db_with_spilled_key` + /// but accepts an arbitrary `RedisValue` (collections, not just strings). + fn db_with_spilled_value( + shard_dir: &std::path::Path, + key: &[u8], + value: TestRedisValue, + ) -> Database { + let manifest_path = shard_dir.join("shard.manifest"); + let mut manifest = ShardManifest::create(&manifest_path).unwrap(); + let mut cold_index = ColdIndex::new(); + + let mut entry = Entry::new_string(Bytes::new()); // placeholder, overwritten below + entry.value = CompactValue::from_redis_value(value); + + spill_to_datafile( + shard_dir, + 900, + key, + &entry, + &mut manifest, + Some(&mut cold_index), + ) + .unwrap(); + + let mut db = Database::new(); + db.cold_shard_dir = Some(shard_dir.to_path_buf()); + db.cold_index = Some(cold_index); + db + } + + #[test] + fn test_get_or_create_hash_promotes_cold_instead_of_fabricating() { + let tmp = tempfile::tempdir().unwrap(); + let mut fields = HashMap::new(); + fields.insert(Bytes::from_static(b"color"), Bytes::from_static(b"red")); + fields.insert(Bytes::from_static(b"size"), Bytes::from_static(b"large")); + let mut db = db_with_spilled_value(tmp.path(), b"myhash", TestRedisValue::Hash(fields)); + + // Precondition: nothing hot yet, key only exists cold. + assert!(!db.is_hot(b"myhash"), "precondition: key must be cold-only"); + + // Simulates HSET's get_or_create_hash call: must promote the real + // spilled fields, NOT fabricate an empty hash that would permanently + // shadow (destroy) the cold copy. + let map = db.get_or_create_hash(b"myhash").unwrap(); + assert_eq!( + map.len(), + 2, + "must promote existing fields, not fabricate empty hash" + ); + assert_eq!(map.get(b"color".as_slice()).unwrap(), b"red".as_slice()); + + // Promotion must also clear the cold index entry (no dual reference). + assert!(db.cold_index.as_ref().unwrap().lookup(b"myhash").is_none()); + assert!(db.is_hot(b"myhash")); + } + + #[test] + fn test_get_or_create_hash_merges_new_field_with_promoted_ones() { + let tmp = tempfile::tempdir().unwrap(); + let mut fields = HashMap::new(); + fields.insert( + Bytes::from_static(b"old_field"), + Bytes::from_static(b"old_value"), + ); + let mut db = db_with_spilled_value(tmp.path(), b"h", TestRedisValue::Hash(fields)); + + // HSET h new_field new_value + { + let map = db.get_or_create_hash(b"h").unwrap(); + map.insert( + Bytes::from_static(b"new_field"), + Bytes::from_static(b"new_value"), + ); + } + + let map = db.get_hash(b"h").unwrap().unwrap(); + assert_eq!( + map.len(), + 2, + "old spilled field + new field must both be present" + ); + assert_eq!( + map.get(b"old_field".as_slice()).unwrap(), + b"old_value".as_slice() + ); + assert_eq!( + map.get(b"new_field".as_slice()).unwrap(), + b"new_value".as_slice() + ); + } + + #[test] + fn test_get_hash_ref_if_alive_sees_cold_hash_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut fields = HashMap::new(); + fields.insert(Bytes::from_static(b"f"), Bytes::from_static(b"v")); + let db = db_with_spilled_value(tmp.path(), b"h", TestRedisValue::Hash(fields)); + + let href = db + .get_hash_ref_if_alive(b"h", 0) + .unwrap() + .expect("must see cold hash"); + assert_eq!(href.get_field(b"f").unwrap(), b"v".as_slice()); + assert_eq!(href.len(), 1); + + // Non-promoting: `&self` accessor must not have mutated hot RAM. + assert!( + !db.is_hot(b"h"), + "get_hash_ref_if_alive takes &self and must not promote" + ); + assert!( + db.cold_index.as_ref().unwrap().lookup(b"h").is_some(), + "cold index entry must remain untouched by a non-promoting read" + ); + } + + #[test] + fn test_get_or_create_list_promotes_cold_instead_of_fabricating() { + let tmp = tempfile::tempdir().unwrap(); + let mut list = VecDeque::new(); + list.push_back(Bytes::from_static(b"a")); + list.push_back(Bytes::from_static(b"b")); + let mut db = db_with_spilled_value(tmp.path(), b"mylist", TestRedisValue::List(list)); + + let l = db.get_or_create_list(b"mylist").unwrap(); + assert_eq!( + l.len(), + 2, + "must promote existing elements, not fabricate empty list" + ); + assert_eq!(l[0], Bytes::from_static(b"a")); + } + + #[test] + fn test_get_list_ref_if_alive_sees_cold_list_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut list = VecDeque::new(); + list.push_back(Bytes::from_static(b"x")); + let db = db_with_spilled_value(tmp.path(), b"l", TestRedisValue::List(list)); + + let lref = db + .get_list_ref_if_alive(b"l", 0) + .unwrap() + .expect("must see cold list"); + assert_eq!(lref.len(), 1); + assert!(!db.is_hot(b"l")); + } + + #[test] + fn test_get_or_create_set_promotes_cold_instead_of_fabricating() { + let tmp = tempfile::tempdir().unwrap(); + let mut set = HashSet::new(); + set.insert(Bytes::from_static(b"m1")); + set.insert(Bytes::from_static(b"m2")); + let mut db = db_with_spilled_value(tmp.path(), b"myset", TestRedisValue::Set(set)); + + let s = db.get_or_create_set(b"myset").unwrap(); + assert_eq!( + s.len(), + 2, + "must promote existing members, not fabricate empty set" + ); + assert!(s.contains(b"m1".as_slice())); + } + + #[test] + fn test_get_set_ref_if_alive_sees_cold_set_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut set = HashSet::new(); + set.insert(Bytes::from_static(b"m")); + let db = db_with_spilled_value(tmp.path(), b"s", TestRedisValue::Set(set)); + + let sref = db + .get_set_ref_if_alive(b"s", 0) + .unwrap() + .expect("must see cold set"); + assert!(sref.contains(b"m")); + assert!(!db.is_hot(b"s")); + } + + #[test] + fn test_get_or_create_sorted_set_promotes_cold_instead_of_fabricating() { + let tmp = tempfile::tempdir().unwrap(); + let mut tree = BPTree::new(); + let mut members = HashMap::new(); + tree.insert(OrderedFloat(1.5), Bytes::from_static(b"alice")); + members.insert(Bytes::from_static(b"alice"), 1.5); + let mut db = db_with_spilled_value( + tmp.path(), + b"myzset", + TestRedisValue::SortedSetBPTree { tree, members }, + ); + + let (members, _tree) = db.get_or_create_sorted_set(b"myzset").unwrap(); + assert_eq!( + members.len(), + 1, + "must promote existing members, not fabricate empty zset" + ); + assert_eq!(members.get(b"alice".as_slice()), Some(&1.5)); + } + + #[test] + fn test_get_sorted_set_ref_if_alive_sees_cold_zset_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut tree = BPTree::new(); + let mut members = HashMap::new(); + tree.insert(OrderedFloat(2.0), Bytes::from_static(b"bob")); + members.insert(Bytes::from_static(b"bob"), 2.0); + let db = db_with_spilled_value( + tmp.path(), + b"z", + TestRedisValue::SortedSetBPTree { tree, members }, + ); + + let zref = db + .get_sorted_set_ref_if_alive(b"z", 0) + .unwrap() + .expect("must see cold zset"); + assert_eq!(zref.score(b"bob"), Some(2.0)); + assert!(!db.is_hot(b"z")); + } + + #[test] + fn test_get_or_create_stream_promotes_cold_instead_of_fabricating() { + let tmp = tempfile::tempdir().unwrap(); + let mut stream = StreamData::new(); + let id = crate::storage::stream::StreamId { ms: 1000, seq: 0 }; + stream.last_id = id; + stream.length = 1; + stream.entries.insert( + id, + vec![(Bytes::from_static(b"field"), Bytes::from_static(b"value"))], + ); + let mut db = db_with_spilled_value( + tmp.path(), + b"mystream", + TestRedisValue::Stream(Box::new(stream)), + ); + + let s = db.get_or_create_stream(b"mystream").unwrap(); + assert_eq!( + s.length, 1, + "must promote existing entries, not fabricate empty stream" + ); + assert_eq!(s.entries.len(), 1); + } + + #[test] + fn test_get_stream_if_alive_sees_cold_stream_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut stream = StreamData::new(); + let id = crate::storage::stream::StreamId { ms: 2000, seq: 0 }; + stream.last_id = id; + stream.length = 1; + stream.entries.insert( + id, + vec![(Bytes::from_static(b"f"), Bytes::from_static(b"v"))], + ); + let db = db_with_spilled_value(tmp.path(), b"s", TestRedisValue::Stream(Box::new(stream))); + + let sref = db + .get_stream_if_alive(b"s", 0) + .unwrap() + .expect("must see cold stream"); + assert_eq!(sref.length, 1); + assert!(!db.is_hot(b"s")); + } + + #[test] + fn test_exists_counts_cold_only_hash_without_promoting() { + let tmp = tempfile::tempdir().unwrap(); + let mut fields = HashMap::new(); + fields.insert(Bytes::from_static(b"f"), Bytes::from_static(b"v")); + let mut db = db_with_spilled_value(tmp.path(), b"h", TestRedisValue::Hash(fields)); + + assert!(db.exists(b"h"), "cold-only key must count as existing"); + // Cheap presence check must not have promoted the key into hot RAM. + assert!( + !db.is_hot(b"h"), + "EXISTS must not promote — cheap check only" + ); + + let now_ms = db.now_ms(); + assert!(db.exists_if_alive(b"h", now_ms)); + assert!(!db.is_hot(b"h")); + } + + #[test] + fn test_exists_false_for_genuinely_missing_key() { + let mut db = Database::new(); + db.cold_index = Some(ColdIndex::new()); + assert!(!db.exists(b"nope")); + let now_ms = db.now_ms(); + assert!(!db.exists_if_alive(b"nope", now_ms)); + } } diff --git a/src/storage/db_read.rs b/src/storage/db_read.rs index af72d48f3..ee82b9ee1 100644 --- a/src/storage/db_read.rs +++ b/src/storage/db_read.rs @@ -5,6 +5,7 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use super::bptree::BPTree; use super::intset::Intset; use super::listpack::Listpack; +use super::stream::Stream as StreamData; // --------------------------------------------------------------------------- // Read-only Ref enums for immutable access to compact and full encodings @@ -29,6 +30,22 @@ pub enum HashRef<'a> { now_ms: u64, min_expiry_ms: u64, }, + /// A value read fresh from the cold tier with no backing hot `Entry` to + /// borrow from (P0 cold-collection-visibility fix, 2026-07-12). + /// + /// `get_hash_ref_if_alive` takes `&self` because some callers (the + /// RwLock-shared-read dispatch path) hold only a shared read guard and + /// cannot promote a cold hit into hot RAM. Rather than report the key + /// absent (silent data loss to the caller), the value is decoded fresh + /// from disk on every such access and carried here by value. + Owned(HashMap), + /// Owned counterpart of `WithTtl` for a cold `HashWithTtl` read. + OwnedWithTtl { + fields: HashMap, + ttls: HashMap, + now_ms: u64, + min_expiry_ms: u64, + }, } impl<'a> HashRef<'a> { @@ -65,6 +82,23 @@ impl<'a> HashRef<'a> { fields.get(field).cloned() } } + HashRef::Owned(map) => map.get(field).cloned(), + HashRef::OwnedWithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + } => { + if *now_ms < *min_expiry_ms { + return fields.get(field).cloned(); + } + let expired = ttls.get(field).is_some_and(|&t| t <= *now_ms); + if expired { + None + } else { + fields.get(field).cloned() + } + } } } @@ -92,6 +126,21 @@ impl<'a> HashRef<'a> { .filter(|f| ttls.get(*f).map_or(true, |&t| t > *now_ms)) .count() } + HashRef::Owned(map) => map.len(), + HashRef::OwnedWithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + } => { + if *now_ms < *min_expiry_ms { + return fields.len(); + } + fields + .keys() + .filter(|f| ttls.get(*f).map_or(true, |&t| t > *now_ms)) + .count() + } } } @@ -120,6 +169,22 @@ impl<'a> HashRef<'a> { .map(|(k, v)| (k.clone(), v.clone())) .collect() } + HashRef::Owned(map) => map.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + HashRef::OwnedWithTtl { + fields, + ttls, + now_ms, + min_expiry_ms, + } => { + if *now_ms < *min_expiry_ms { + return fields.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); + } + fields + .iter() + .filter(|(f, _)| ttls.get(*f).map_or(true, |&t| t > *now_ms)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect() + } } } } @@ -128,6 +193,9 @@ impl<'a> HashRef<'a> { pub enum ListRef<'a> { Deque(&'a VecDeque), Listpack(&'a Listpack), + /// Owned counterpart for a value decoded fresh from the cold tier — see + /// `HashRef::Owned` for the rationale (P0 cold-collection-visibility fix). + Owned(VecDeque), } impl<'a> ListRef<'a> { @@ -136,6 +204,7 @@ impl<'a> ListRef<'a> { match self { ListRef::Deque(d) => d.len(), ListRef::Listpack(lp) => lp.len(), + ListRef::Owned(d) => d.len(), } } @@ -144,6 +213,7 @@ impl<'a> ListRef<'a> { match self { ListRef::Deque(d) => d.get(index).cloned(), ListRef::Listpack(lp) => lp.get_at(index).map(|e| e.to_bytes()), + ListRef::Owned(d) => d.get(index).cloned(), } } @@ -154,6 +224,7 @@ impl<'a> ListRef<'a> { ListRef::Listpack(lp) => (start..=end) .filter_map(|i| lp.get_at(i).map(|e| e.to_bytes())) .collect(), + ListRef::Owned(d) => (start..=end).filter_map(|i| d.get(i).cloned()).collect(), } } @@ -162,6 +233,7 @@ impl<'a> ListRef<'a> { match self { ListRef::Deque(d) => d.iter().cloned().collect(), ListRef::Listpack(lp) => lp.iter().map(|e| e.to_bytes()).collect(), + ListRef::Owned(d) => d.iter().cloned().collect(), } } } @@ -171,6 +243,9 @@ pub enum SetRef<'a> { Hash(&'a HashSet), Listpack(&'a Listpack), Intset(&'a Intset), + /// Owned counterpart for a value decoded fresh from the cold tier — see + /// `HashRef::Owned` for the rationale (P0 cold-collection-visibility fix). + Owned(HashSet), } impl<'a> SetRef<'a> { @@ -180,6 +255,7 @@ impl<'a> SetRef<'a> { SetRef::Hash(s) => s.len(), SetRef::Listpack(lp) => lp.len(), SetRef::Intset(is) => is.len(), + SetRef::Owned(s) => s.len(), } } @@ -196,6 +272,7 @@ impl<'a> SetRef<'a> { } false } + SetRef::Owned(s) => s.contains(member), } } @@ -205,6 +282,7 @@ impl<'a> SetRef<'a> { SetRef::Hash(s) => s.iter().cloned().collect(), SetRef::Listpack(lp) => lp.iter().map(|e| e.to_bytes()).collect(), SetRef::Intset(is) => is.iter().map(|v| Bytes::from(v.to_string())).collect(), + SetRef::Owned(s) => s.iter().cloned().collect(), } } @@ -214,6 +292,7 @@ impl<'a> SetRef<'a> { SetRef::Hash(s) => (*s).clone(), SetRef::Listpack(lp) => lp.iter().map(|e| e.to_bytes()).collect(), SetRef::Intset(is) => is.iter().map(|v| Bytes::from(v.to_string())).collect(), + SetRef::Owned(s) => s.clone(), } } } @@ -230,6 +309,15 @@ pub enum SortedSetRef<'a> { members: &'a HashMap, scores: &'a BTreeMap<(OrderedFloat, Bytes), ()>, }, + /// Owned counterpart of `BPTree` for a value decoded fresh from the cold + /// tier — see `HashRef::Owned` for the rationale (P0 + /// cold-collection-visibility fix). `members_map`/`bptree` return `None` + /// for this variant (same as `Listpack`); callers already fall back to + /// the generic `score`/`entries_sorted` methods in that case. + Owned { + tree: BPTree, + members: HashMap, + }, } impl<'a> SortedSetRef<'a> { @@ -239,6 +327,7 @@ impl<'a> SortedSetRef<'a> { SortedSetRef::BPTree { members, .. } => members.len(), SortedSetRef::Listpack(lp) => lp.len() / 2, SortedSetRef::Legacy { members, .. } => members.len(), + SortedSetRef::Owned { members, .. } => members.len(), } } @@ -260,6 +349,7 @@ impl<'a> SortedSetRef<'a> { None } SortedSetRef::Legacy { members, .. } => members.get(member).copied(), + SortedSetRef::Owned { members, .. } => members.get(member).copied(), } } @@ -292,6 +382,10 @@ impl<'a> SortedSetRef<'a> { SortedSetRef::Legacy { scores, .. } => { scores.keys().map(|(s, m)| (m.clone(), s.0)).collect() } + SortedSetRef::Owned { tree, .. } => tree + .iter() + .map(|(score, member)| (member.clone(), score.0)) + .collect(), } } @@ -302,6 +396,7 @@ impl<'a> SortedSetRef<'a> { SortedSetRef::BPTree { members, .. } => Some(members), SortedSetRef::Legacy { members, .. } => Some(members), SortedSetRef::Listpack(_) => None, + SortedSetRef::Owned { .. } => None, } } @@ -313,3 +408,25 @@ impl<'a> SortedSetRef<'a> { } } } + +/// Read-only reference to a stream: either borrowed from a hot `Entry`, or +/// owned after a fresh cold-tier decode (P0 cold-collection-visibility fix, +/// 2026-07-12 — see `HashRef::Owned` for the general rationale). +/// +/// `Deref`s to `&StreamData` so existing call sites (`stream.length`, +/// `stream.entries.range(..)`, etc.) work unchanged for both variants. +pub enum StreamRef<'a> { + Borrowed(&'a StreamData), + Owned(Box), +} + +impl<'a> std::ops::Deref for StreamRef<'a> { + type Target = StreamData; + + fn deref(&self) -> &StreamData { + match self { + StreamRef::Borrowed(s) => s, + StreamRef::Owned(s) => s, + } + } +} diff --git a/tests/cold_collection_visibility.rs b/tests/cold_collection_visibility.rs new file mode 100644 index 000000000..8bbc17b8d --- /dev/null +++ b/tests/cold_collection_visibility.rs @@ -0,0 +1,887 @@ +//! P0 (task #41): with disk-offload enabled (the DEFAULT), the production +//! eviction paths (`evict_one_async_spill` / `evict_batch_durable_no_aof`, +//! `src/storage/eviction.rs`) spill Hash/List/Set/ZSet values to the cold +//! tier just like strings — but the type-specific command accessors never +//! consulted the cold index: +//! +//! * HGETALL/LRANGE/SMEMBERS/ZRANGE/EXISTS reported the key ABSENT. +//! * HSET/LPUSH/SADD/ZADD silently fabricated a brand-new EMPTY container, +//! which then got written back over the real cold copy on the next +//! flush — permanently destroying it. +//! +//! This is a wire-level black-box test on purpose: the bug lives in the +//! command-dispatch accessors (`Database::get_or_create_*`, +//! `get_*_ref_if_alive`), which unit tests could route around by calling the +//! storage layer directly. Driving it through the real RESP protocol proves +//! the fix is wired into the actual HSET/HGETALL/EXISTS command handlers, +//! not just the storage primitives (see `tests/oom_bypass_closure.rs` for the +//! same "wire-level on purpose" rationale against a sibling dispatch-wiring +//! bug). +//! +//! Design notes: +//! * `--disk-offload enable` alone does NOT spill collections — spill is +//! INERT without a durability backstop (`ServerConfig:: +//! disk_offload_spill_inert`): the async/durable-batch eviction paths +//! need a `ShardManifest`, which is only wired up when `--appendonly yes` +//! or `--save` is set. Without one, an evicting policy just DROPS +//! victims (no tiering) — which would not exercise this bug at all. So +//! this suite runs with `--appendonly yes`. +//! * **Two independent eviction paths race for the same victim, and only +//! one of them spills.** The interactive write-path gate (synchronous, +//! runs inline on a `SET`/etc that crosses `maxmemory`) spills every +//! type via `evict_one_async_spill`. The periodic background tick +//! (`shard::persistence_tick` / `src/shard/timers.rs`, no +//! `SpillContext`) plain-DROPS its victim instead — which is legal +//! `allkeys-lru` semantics (Wave A records a real reason-DEL for it), not +//! a bug. Which path claims a given sampled-oldest key is a genuine race +//! that differs by OS scheduler (observed: async-spill path usually wins +//! on macOS, the 100ms tick usually wins on Linux). A single probe key +//! therefore makes this suite nondeterministic about WHICH bug-relevant +//! path it exercises — a plain-dropped probe reads back as +//! `EXISTS == 0`, which is *correct* behavior, indistinguishable from the +//! pre-fix bug's *incorrect* `EXISTS == 0` on a still-cold key, purely +//! from the client's point of view. +//! * Fix: use `PROBE_COUNT` (32) probe collections per type instead of one, +//! half written before the filler wave and half interleaved DURING it +//! (`drive_eviction_interleaved`), so across the whole batch some probes +//! are very likely to be claimed by each path. After the wave, classify +//! every probe as `Complete` (present, full original content — the +//! promote-from-cold path ran), `Absent` (gone, empty content — a +//! legitimate plain-drop), or `Ghost` (anything else: `EXISTS` +//! disagreeing with content, or partial/corrupted content — the actual +//! P0 shape). The suite fails on any `Ghost`, and separately fails (with +//! an actionable diagnostic) if EVERY probe came back `Absent` — that +//! would mean this run never exercised the promote path at all, which +//! would make the rest of the assertions vacuous. +//! * The no-fabrication-on-write check then runs twice: once against a +//! `Complete` probe (the write MUST merge with the promoted content) and +//! once against an `Absent` probe if one exists (the write legitimately +//! creates a fresh container — there is no cold copy left to merge with, +//! the plain-drop path already deleted it). +//! * Eviction/spill is tick-driven or write-path-driven, not synchronous +//! with the specific filler write that crosses the threshold — a short +//! settle sleep follows the filler wave before probing. +//! * A classification's `EXISTS` and read-all round trips are not atomic: +//! `evict_one_async_spill` removes the hot entry immediately but hands +//! off to a background spill-thread channel, so a probe caught squarely +//! in that in-flight window can transiently answer `EXISTS == 0` and +//! then (a moment later, on the next round trip) read non-empty. This +//! is NOT the P0 shape — it resolves to a stable state within +//! milliseconds. `partition_probes` re-checks a ghost verdict a handful +//! of times (`GHOST_RECHECK_ATTEMPTS`/`GHOST_RECHECK_DELAY`) before +//! failing, so only a verdict that never stabilizes counts as a real +//! ghost (observed in practice: caught exactly this transient shape once +//! during hardening on the Linux VM, confirmed non-reproducing after the +//! retry was added). +//! +//! Run with (monoio default — matches CI): +//! cargo build --release +//! MOON_BIN=$PWD/target/release/moon cargo test --release --test cold_collection_visibility +//! +//! tokio runtime: +//! cargo build --release --no-default-features --features runtime-tokio,jemalloc +//! MOON_BIN=$PWD/target/release/moon cargo test --release --no-default-features \ +//! --features runtime-tokio,jemalloc --test cold_collection_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}; + +// --------------------------------------------------------------------------- +// Binary resolution + server spawn (pattern: tests/oom_bypass_closure.rs) +// --------------------------------------------------------------------------- + +fn find_moon_binary() -> std::path::PathBuf { + if let Ok(bin) = std::env::var("MOON_BIN") { + let p = std::path::PathBuf::from(bin); + if p.exists() { + return p; + } + } + std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")) +} + +/// Root the test's scratch dir under the repo's own volume, not `$TMPDIR` +/// (which can trip Moon's 5%-free diskfull write-pause guard — see +/// gotcha_vm_diskfull_shared_volume in project memory). Also pass +/// `--disk-free-min-pct 0` explicitly for the same reason. +fn test_tmpdir() -> tempfile::TempDir { + let base = + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("target/cold-vis-test-tmp"); + std::fs::create_dir_all(&base).expect("create cold-vis-test-tmp base dir"); + tempfile::Builder::new() + .prefix("cold-vis-") + .tempdir_in(&base) + .expect("tempdir_in target/cold-vis-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 — tiny, forces eviction fast. + +fn spawn_moon_cold_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(), + "--shards", + "1", + // Spill is inert without a durability backstop — see module + // doc. AOF also gives us a realistic production config. + "--appendonly", + "yes", + "--disk-offload", + "enable", + "--maxmemory", + &MAXMEMORY_BYTES.to_string(), + "--maxmemory-policy", + "allkeys-lru", + // Approximate (sampled) LRU, matching Redis semantics — the + // default sample size (5) makes it a coin flip per eviction + // round whether the single oldest probe key ever gets + // sampled out of hundreds of filler keys. A large sample + // size makes eviction of the globally-oldest key + // deterministic-in-practice within the filler wave below. + "--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 (binary-safe args, full-frame parser) +// --------------------------------------------------------------------------- + +#[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 = format!("127.0.0.1:{port}") + .to_socket_addrs() + .unwrap() + .next() + .unwrap(); + 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() + } + + 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 log_tail(dir: &std::path::Path, name: &str) -> String { + match std::fs::read_to_string(dir.join(name)) { + Ok(s) => { + let tail: Vec<&str> = s.lines().rev().take(20).collect(); + tail.into_iter().rev().collect::>().join("\n") + } + Err(e) => format!(""), + } +} + +fn wait_ready(guard: &mut ServerGuard, dir: &std::path::Path, port: u16) -> Client { + let deadline = readiness_deadline(); + let start = Instant::now(); + loop { + if let Ok(Some(status)) = guard.0.try_wait() { + panic!( + "moon exited ({status}) before accepting on port {port}\n\ + --- moon.stderr.log (tail) ---\n{}\n\ + --- moon.stdout.log (tail) ---\n{}", + log_tail(dir, "moon.stderr.log"), + log_tail(dir, "moon.stdout.log"), + ); + } + if let Some(mut c) = Client::try_connect(port, Duration::from_secs(1)) + && let Ok(true) = c.try_ping() + { + return c; + } + assert!( + start.elapsed() < deadline, + "server never answered PING on port {port} within {deadline:?}\n\ + --- moon.stderr.log (tail) ---\n{}", + log_tail(dir, "moon.stderr.log"), + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +fn blob(size: usize, fill: u8) -> Vec { + vec![fill; size] +} + +/// Number of probe collections per type. Deliberately > 1: a single probe +/// makes the suite nondeterministic about which of the two eviction paths +/// (interactive write-gate spill vs. periodic-tick plain-drop) claims it — +/// see the module doc. With many probes, both paths are very likely to claim +/// at least one each within a single run. +const PROBE_COUNT: usize = 32; + +/// Drive memory well past `MAXMEMORY_BYTES` with unique filler keys so +/// `allkeys-lru` evicts everything written before this point — namely the +/// probe keys the caller wrote up front. `extra_probes` additional probes are +/// written via `write_extra_probe(client, index)` INTERLEAVED across the +/// filler wave (evenly spaced), rather than all up front, so the +/// synchronous/interactive write-path eviction gate (which spills +/// collections) has as much chance of claiming a probe as the periodic +/// background tick (which does not — see module doc). Settles briefly +/// afterward: eviction/spill is not synchronous with the specific write that +/// crosses the threshold. +fn drive_eviction_interleaved( + c: &mut Client, + extra_probes: usize, + mut write_extra_probe: impl FnMut(&mut Client, usize), +) { + const FILLER_COUNT: usize = 400; + const FILLER_VALUE_LEN: usize = 4 * 1024; // 4 KiB — 400 * 4KiB ~= 1.6MB >> 512KB cap. + let value = blob(FILLER_VALUE_LEN, b'f'); + let interval = if extra_probes == 0 { + 0 + } else { + (FILLER_COUNT / extra_probes).max(1) + }; + let mut next_probe = 0usize; + for i in 0..FILLER_COUNT { + let key = format!("filler:{i}"); + let _ = c.cmd(&[b"SET", key.as_bytes(), &value]); + if next_probe < extra_probes && i % interval == 0 { + write_extra_probe(c, next_probe); + next_probe += 1; + } + } + // Rounding leftovers (shouldn't happen for PROBE_COUNT/FILLER_COUNT + // combinations in this file, but stay correct regardless). + while next_probe < extra_probes { + write_extra_probe(c, next_probe); + next_probe += 1; + } + std::thread::sleep(Duration::from_secs(3)); +} + +/// Classification of one probe's post-eviction state. +#[derive(Debug)] +enum ProbeState { + /// Present with the COMPLETE original content — the promote-from-cold + /// path ran (or the probe was never actually evicted). + Complete, + /// Gone entirely (`EXISTS == 0`, empty read) — a legitimate plain-drop + /// by the tick path (no `SpillContext`, no tiering, correct semantics). + Absent, + /// Anything else: `EXISTS` disagreeing with content, or partial/ + /// corrupted content. This is the actual P0 shape (a cold-spilled value + /// visible to neither read nor EXISTS, or a write that clobbered rather + /// than merged) and always fails the test. + Ghost(String), +} + +fn exists_bit(c: &mut Client, key: &str) -> Result { + match c.cmd(&[b"EXISTS", key.as_bytes()]) { + V::Int(1) => Ok(true), + V::Int(0) => Ok(false), + other => Err(format!( + "EXISTS returned unexpected value for {key}: {other:?}" + )), + } +} + +fn classify_hash(c: &mut Client, key: &str) -> ProbeState { + let exists = match exists_bit(c, key) { + Ok(b) => b, + Err(msg) => return ProbeState::Ghost(msg), + }; + let all = c.cmd(&[b"HGETALL", key.as_bytes()]); + let items = match &all { + V::Arr(items) => items, + other => { + return ProbeState::Ghost(format!("HGETALL returned non-array for {key}: {other:?}")); + } + }; + if items.is_empty() { + return if exists { + ProbeState::Ghost(format!("EXISTS=1 but HGETALL empty for {key}")) + } else { + ProbeState::Absent + }; + } + if !exists { + return ProbeState::Ghost(format!("EXISTS=0 but HGETALL non-empty for {key}: {all:?}")); + } + let mut map = std::collections::HashMap::new(); + for chunk in items.chunks(2) { + match (&chunk[0], chunk.get(1)) { + (V::Bulk(f), Some(V::Bulk(v))) => { + map.insert(f.clone(), v.clone()); + } + _ => return ProbeState::Ghost(format!("HGETALL malformed pair for {key}: {all:?}")), + } + } + let expected: std::collections::HashMap, Vec> = [ + (b"f1".to_vec(), b"v1".to_vec()), + (b"f2".to_vec(), b"v2".to_vec()), + ] + .into_iter() + .collect(); + if map == expected { + ProbeState::Complete + } else { + ProbeState::Ghost(format!("HGETALL partial/wrong content for {key}: {all:?}")) + } +} + +fn classify_list(c: &mut Client, key: &str) -> ProbeState { + let exists = match exists_bit(c, key) { + Ok(b) => b, + Err(msg) => return ProbeState::Ghost(msg), + }; + let range = c.cmd(&[b"LRANGE", key.as_bytes(), b"0", b"-1"]); + let items = match &range { + V::Arr(items) => items, + other => { + return ProbeState::Ghost(format!("LRANGE returned non-array for {key}: {other:?}")); + } + }; + if items.is_empty() { + return if exists { + ProbeState::Ghost(format!("EXISTS=1 but LRANGE empty for {key}")) + } else { + ProbeState::Absent + }; + } + if !exists { + return ProbeState::Ghost(format!( + "EXISTS=0 but LRANGE non-empty for {key}: {range:?}" + )); + } + let expected = vec![V::Bulk(b"a".to_vec()), V::Bulk(b"b".to_vec())]; + if *items == expected { + ProbeState::Complete + } else { + ProbeState::Ghost(format!("LRANGE partial/wrong content for {key}: {range:?}")) + } +} + +fn classify_set(c: &mut Client, key: &str) -> ProbeState { + let exists = match exists_bit(c, key) { + Ok(b) => b, + Err(msg) => return ProbeState::Ghost(msg), + }; + let members = c.cmd(&[b"SMEMBERS", key.as_bytes()]); + let items = match &members { + V::Arr(items) => items, + other => { + return ProbeState::Ghost(format!("SMEMBERS returned non-array for {key}: {other:?}")); + } + }; + if items.is_empty() { + return if exists { + ProbeState::Ghost(format!("EXISTS=1 but SMEMBERS empty for {key}")) + } else { + ProbeState::Absent + }; + } + if !exists { + return ProbeState::Ghost(format!( + "EXISTS=0 but SMEMBERS non-empty for {key}: {members:?}" + )); + } + let mut set = std::collections::HashSet::new(); + for it in items { + match it { + V::Bulk(b) => { + set.insert(b.clone()); + } + other => { + return ProbeState::Ghost(format!("SMEMBERS non-bulk item for {key}: {other:?}")); + } + } + } + let expected: std::collections::HashSet> = + [b"alpha".to_vec(), b"beta".to_vec()].into_iter().collect(); + if set == expected { + ProbeState::Complete + } else { + ProbeState::Ghost(format!( + "SMEMBERS partial/wrong content for {key}: {members:?}" + )) + } +} + +fn classify_zset(c: &mut Client, key: &str) -> ProbeState { + let exists = match exists_bit(c, key) { + Ok(b) => b, + Err(msg) => return ProbeState::Ghost(msg), + }; + let range = c.cmd(&[b"ZRANGE", key.as_bytes(), b"0", b"-1"]); + let items = match &range { + V::Arr(items) => items, + other => { + return ProbeState::Ghost(format!("ZRANGE returned non-array for {key}: {other:?}")); + } + }; + if items.is_empty() { + return if exists { + ProbeState::Ghost(format!("EXISTS=1 but ZRANGE empty for {key}")) + } else { + ProbeState::Absent + }; + } + if !exists { + return ProbeState::Ghost(format!( + "EXISTS=0 but ZRANGE non-empty for {key}: {range:?}" + )); + } + let expected = vec![V::Bulk(b"one".to_vec()), V::Bulk(b"two".to_vec())]; + if *items == expected { + ProbeState::Complete + } else { + ProbeState::Ghost(format!("ZRANGE partial/wrong content for {key}: {range:?}")) + } +} + +/// `classify_*` issues two separate round trips (`EXISTS`, then a +/// type-specific read-all) that are not atomic with respect to the server: +/// `evict_one_async_spill` removes the hot entry and hands the value to a +/// background spill-thread channel, which durably records it in the +/// `ColdIndex` slightly LATER. A probe caught squarely in that in-flight +/// window can legitimately answer `EXISTS == 0` on one round trip and then +/// `ZRANGE`-non-empty (post-completion) a moment later on the next — +/// transient, not the permanent P0 shape (a value durably cold-spilled but +/// invisible to every reader forever). Re-checking a handful of times lets a +/// transient window resolve to a stable `Complete`/`Absent` without masking a +/// real, persistent ghost, which never resolves. +const GHOST_RECHECK_ATTEMPTS: usize = 10; +const GHOST_RECHECK_DELAY: Duration = Duration::from_millis(150); + +/// Splits classified probe indices into (complete, absent), failing the test +/// immediately on any ghost and again if the promote-from-cold path was +/// never exercised at all (every probe plain-dropped). +fn partition_probes( + c: &mut Client, + kind: &str, + key_of: impl Fn(usize) -> String, + classify: impl Fn(&mut Client, &str) -> ProbeState, +) -> (Vec, Vec) { + let mut complete = Vec::new(); + let mut absent = Vec::new(); + let mut ghosts = Vec::new(); + for i in 0..PROBE_COUNT { + let key = key_of(i); + let mut state = classify(c, &key); + let mut attempts = 1; + while matches!(state, ProbeState::Ghost(_)) && attempts < GHOST_RECHECK_ATTEMPTS { + std::thread::sleep(GHOST_RECHECK_DELAY); + state = classify(c, &key); + attempts += 1; + } + match state { + ProbeState::Complete => complete.push(i), + ProbeState::Absent => absent.push(i), + ProbeState::Ghost(msg) => { + ghosts.push(format!("{key} (stable after {attempts} attempts): {msg}")); + } + } + } + assert!( + ghosts.is_empty(), + "P0 GHOST on {} of {PROBE_COUNT} {kind} probes — EXISTS/content disagreement or \ + partial content (promoted-but-invisible, or fabricated-over-cold-copy):\n{}", + ghosts.len(), + ghosts.join("\n") + ); + assert!( + !complete.is_empty(), + "all {PROBE_COUNT} {kind} probes were plain-dropped (Absent) by the background tick; \ + none were claimed by the interactive spill-and-promote write-path gate, so this run \ + never exercised the cold-collection promote path at all. Lower the filler wave's \ + FILLER_COUNT/FILLER_VALUE_LEN (drive_eviction_interleaved) or raise MAXMEMORY_BYTES so \ + eviction pressure ramps more gradually, giving the interactive gate more opportunities \ + to claim a probe before the tick does; or increase PROBE_COUNT for more samples." + ); + (complete, absent) +} + +fn arr_len(v: &V) -> usize { + match v { + V::Arr(items) => items.len(), + other => panic!("expected Arr, got {other:?}"), + } +} + +fn arr_as_bulk_set(v: &V) -> std::collections::HashSet> { + match v { + V::Arr(items) => items + .iter() + .map(|i| match i { + V::Bulk(b) => b.clone(), + other => panic!("expected Bulk in array, got {other:?}"), + }) + .collect(), + other => panic!("expected Arr, got {other:?}"), + } +} + +// --------------------------------------------------------------------------- +// Hash: HSET (listpack fast path) -> evict/spill (racy) -> ghost-free + +// no-fabrication-on-write checks +// --------------------------------------------------------------------------- + +#[test] +fn test_cold_hash_visible_after_eviction_no_fabrication_on_write() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_cold_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + let key_of = |i: usize| format!("probehash:{i}"); + let write_probe = |c: &mut Client, i: usize| { + // Small field/value sizes deliberately keep this on the listpack fast + // path (`get_or_create_hash_listpack`), the most common real-world + // HSET shape — proving the fix covers it, not just the full-HashMap + // path. + let _ = c.cmd(&[b"HSET", key_of(i).as_bytes(), b"f1", b"v1", b"f2", b"v2"]); + }; + + let upfront = PROBE_COUNT / 2; + for i in 0..upfront { + write_probe(&mut c, i); + } + drive_eviction_interleaved(&mut c, PROBE_COUNT - upfront, |c, j| { + write_probe(c, upfront + j); + }); + + let (complete, absent) = partition_probes(&mut c, "hash", key_of, classify_hash); + + // No-fabrication-on-write, PRESENT case: HSET on a promoted hash must + // MERGE with the promoted fields, never replace them. + let present_key = key_of(complete[0]); + assert_eq!( + c.cmd(&[b"HSET", present_key.as_bytes(), b"f3", b"v3"]), + V::Int(1), + "f3 must be counted as a newly-added field on {present_key}" + ); + let after = c.cmd(&[b"HGETALL", present_key.as_bytes()]); + assert_eq!( + arr_len(&after), + 6, + "P0: HSET must MERGE with the promoted cold fields on {present_key}, not fabricate an \ + empty hash and lose f1/f2: {after:?}" + ); + + // Fresh-container case: a genuinely absent (plain-dropped) probe's next + // HSET legitimately creates a brand-new hash — there is no cold copy to + // merge with, the tick path already deleted it. + if let Some(&idx) = absent.first() { + let absent_key = key_of(idx); + assert_eq!( + c.cmd(&[b"HSET", absent_key.as_bytes(), b"f3", b"v3"]), + V::Int(1) + ); + let fresh = c.cmd(&[b"HGETALL", absent_key.as_bytes()]); + assert_eq!( + arr_len(&fresh), + 2, + "fresh hash on genuinely-absent {absent_key} must contain only the new field: {fresh:?}" + ); + } +} + +// --------------------------------------------------------------------------- +// List: RPUSH (listpack fast path) -> evict/spill (racy) -> ghost-free + +// no-fabrication-on-write checks +// --------------------------------------------------------------------------- + +#[test] +fn test_cold_list_visible_after_eviction_no_fabrication_on_write() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_cold_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + let key_of = |i: usize| format!("probelist:{i}"); + let write_probe = |c: &mut Client, i: usize| { + let _ = c.cmd(&[b"RPUSH", key_of(i).as_bytes(), b"a", b"b"]); + }; + + let upfront = PROBE_COUNT / 2; + for i in 0..upfront { + write_probe(&mut c, i); + } + drive_eviction_interleaved(&mut c, PROBE_COUNT - upfront, |c, j| { + write_probe(c, upfront + j); + }); + + let (complete, absent) = partition_probes(&mut c, "list", key_of, classify_list); + + let present_key = key_of(complete[0]); + assert_eq!(c.cmd(&[b"RPUSH", present_key.as_bytes(), b"c"]), V::Int(3)); + let after = c.cmd(&[b"LRANGE", present_key.as_bytes(), b"0", b"-1"]); + match &after { + V::Arr(items) => assert_eq!( + items, + &[ + V::Bulk(b"a".to_vec()), + V::Bulk(b"b".to_vec()), + V::Bulk(b"c".to_vec()) + ], + "P0: RPUSH must MERGE with the promoted cold elements on {present_key}, not \ + fabricate an empty list and lose a/b: {after:?}" + ), + other => panic!("expected Arr from LRANGE, got {other:?}"), + } + + if let Some(&idx) = absent.first() { + let absent_key = key_of(idx); + assert_eq!(c.cmd(&[b"RPUSH", absent_key.as_bytes(), b"c"]), V::Int(1)); + let fresh = c.cmd(&[b"LRANGE", absent_key.as_bytes(), b"0", b"-1"]); + match &fresh { + V::Arr(items) => assert_eq!( + items, + &[V::Bulk(b"c".to_vec())], + "fresh list on genuinely-absent {absent_key} must contain only the new \ + element: {fresh:?}" + ), + other => panic!("expected Arr from LRANGE, got {other:?}"), + } + } +} + +// --------------------------------------------------------------------------- +// Set: SADD -> evict/spill (racy) -> ghost-free + no-fabrication-on-write +// checks +// --------------------------------------------------------------------------- + +#[test] +fn test_cold_set_visible_after_eviction_no_fabrication_on_write() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_cold_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + let key_of = |i: usize| format!("probeset:{i}"); + let write_probe = |c: &mut Client, i: usize| { + // Non-integer members keep this off the intset fast path, exercising + // `get_or_create_set` (the general HashSet path). + let _ = c.cmd(&[b"SADD", key_of(i).as_bytes(), b"alpha", b"beta"]); + }; + + let upfront = PROBE_COUNT / 2; + for i in 0..upfront { + write_probe(&mut c, i); + } + drive_eviction_interleaved(&mut c, PROBE_COUNT - upfront, |c, j| { + write_probe(c, upfront + j); + }); + + let (complete, absent) = partition_probes(&mut c, "set", key_of, classify_set); + + let present_key = key_of(complete[0]); + assert_eq!( + c.cmd(&[b"SADD", present_key.as_bytes(), b"gamma"]), + V::Int(1) + ); + let after = c.cmd(&[b"SMEMBERS", present_key.as_bytes()]); + assert_eq!( + arr_as_bulk_set(&after), + std::collections::HashSet::from([b"alpha".to_vec(), b"beta".to_vec(), b"gamma".to_vec()]), + "P0: SADD must MERGE with the promoted cold members on {present_key}, not fabricate an \ + empty set and lose alpha/beta: {after:?}" + ); + + if let Some(&idx) = absent.first() { + let absent_key = key_of(idx); + assert_eq!( + c.cmd(&[b"SADD", absent_key.as_bytes(), b"gamma"]), + V::Int(1) + ); + let fresh = c.cmd(&[b"SMEMBERS", absent_key.as_bytes()]); + assert_eq!( + arr_as_bulk_set(&fresh), + std::collections::HashSet::from([b"gamma".to_vec()]), + "fresh set on genuinely-absent {absent_key} must contain only the new member: {fresh:?}" + ); + } +} + +// --------------------------------------------------------------------------- +// ZSet: ZADD -> evict/spill (racy) -> ghost-free + no-fabrication-on-write +// checks +// --------------------------------------------------------------------------- + +#[test] +fn test_cold_zset_visible_after_eviction_no_fabrication_on_write() { + let dir = test_tmpdir(); + let (mut guard, port) = spawn_moon_cold_offload(dir.path()); + let mut c = wait_ready(&mut guard, dir.path(), port); + + let key_of = |i: usize| format!("probezset:{i}"); + let write_probe = |c: &mut Client, i: usize| { + let _ = c.cmd(&[b"ZADD", key_of(i).as_bytes(), b"1", b"one", b"2", b"two"]); + }; + + let upfront = PROBE_COUNT / 2; + for i in 0..upfront { + write_probe(&mut c, i); + } + drive_eviction_interleaved(&mut c, PROBE_COUNT - upfront, |c, j| { + write_probe(c, upfront + j); + }); + + let (complete, absent) = partition_probes(&mut c, "zset", key_of, classify_zset); + + let present_key = key_of(complete[0]); + assert_eq!( + c.cmd(&[b"ZADD", present_key.as_bytes(), b"3", b"three"]), + V::Int(1) + ); + let after = c.cmd(&[b"ZRANGE", present_key.as_bytes(), b"0", b"-1"]); + match &after { + V::Arr(items) => assert_eq!( + items, + &[ + V::Bulk(b"one".to_vec()), + V::Bulk(b"two".to_vec()), + V::Bulk(b"three".to_vec()) + ], + "P0: ZADD must MERGE with the promoted cold members on {present_key}, not \ + fabricate an empty zset and lose one/two: {after:?}" + ), + other => panic!("expected Arr from ZRANGE, got {other:?}"), + } + + if let Some(&idx) = absent.first() { + let absent_key = key_of(idx); + assert_eq!( + c.cmd(&[b"ZADD", absent_key.as_bytes(), b"3", b"three"]), + V::Int(1) + ); + let fresh = c.cmd(&[b"ZRANGE", absent_key.as_bytes(), b"0", b"-1"]); + match &fresh { + V::Arr(items) => assert_eq!( + items, + &[V::Bulk(b"three".to_vec())], + "fresh zset on genuinely-absent {absent_key} must contain only the new \ + member: {fresh:?}" + ), + other => panic!("expected Arr from ZRANGE, got {other:?}"), + } + } +}