diff --git a/CHANGELOG.md b/CHANGELOG.md index b09d95ddc..fba32b323 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 dispatched real DBSIZE commands — the two definitions disagreed inside a single reply). Known remaining parity gap, tracked separately: SCAN / KEYS / RANDOMKEY still enumerate the hot plane only. +- **Keyspace enumeration under disk-offload: SCAN/KEYS/RANDOMKEY now see + spilled keys (#364).** With disk-offload enabled, cold-only keys (spilled by + eviction, no in-RAM entry) were readable via GET/EXISTS but invisible to + enumeration — a 4-shard instance holding 400 logical keys returned only 116 + from `redis-cli --scan`, silently losing spilled keys for any + migration/backup consumer. SCAN, KEYS, and RANDOMKEY (both dispatch tracks) + now enumerate the union of the hot plane and the in-RAM cold index, + partitioned so a key present in both planes is returned exactly once and + TTL-expired cold entries are skipped — pure in-RAM, no disk reads and no + promotion. SCAN's `TYPE` filter judges cold keys from a new + `ColdLocation::value_type` cache (same cached-copy contract as `ttl_ms`: + populated at spill time, re-derived from the on-disk pages by + `ColdIndex::rebuild_from_manifest` after restart; fits existing struct + padding, so the cold index does not grow; no on-disk format change). - **Storage: recovery panic `double NeedsSplit after split_segment` in `DashTable::insert_or_update`.** The insert-or-update path split an overflowing segment exactly once and declared a second `NeedsSplit` unreachable — false under diff --git a/src/command/key.rs b/src/command/key.rs index c15a13549..07688c2b2 100644 --- a/src/command/key.rs +++ b/src/command/key.rs @@ -787,14 +787,24 @@ pub fn keys(db: &mut Database, args: &[Frame]) -> Frame { // Collect all keys first (need to release immutable borrow before calling db.get) let all_keys: Vec = db.keys().cloned().collect(); + let now_ms = db.now_ms(); let mut result = Vec::new(); for key in all_keys { - // Trigger lazy expiry by calling exists - if db.exists(key.as_bytes()) && glob_match(pattern, key.as_bytes()) { + // Trigger lazy expiry by calling exists; membership below is strict + // hot-aliveness so cold-only keys enter exactly once, via the cold + // loop (#364 plane partition — see Database::cold_only_keys). + let _ = db.exists(key.as_bytes()); + if db.get_if_alive(key.as_bytes(), now_ms).is_some() && glob_match(pattern, key.as_bytes()) + { result.push(Frame::BulkString(key.to_bytes())); } } + for key in db.cold_only_keys(now_ms) { + if glob_match(pattern, key.as_ref()) { + result.push(Frame::BulkString(key.clone())); + } + } Frame::Array(result.into()) } @@ -927,6 +937,19 @@ pub fn unlink(db: &mut Database, args: &[Frame]) -> Frame { Frame::Integer(count) } +/// SCAN's TYPE-filter judgment for a cold-only (spilled) key. +/// +/// Judged from the in-RAM `ColdLocation::value_type` cache — must never +/// read the cold value from disk or promote it into RAM, or SCAN over a +/// large offloaded keyspace turns into a disk crawl / memory-pressure +/// storm on the shard thread (#364). +fn cold_type_matches(db: &Database, key: &[u8], type_filter: &[u8]) -> bool { + db.cold_index + .as_ref() + .and_then(|ci| ci.lookup(key)) + .is_some_and(|loc| type_filter.eq_ignore_ascii_case(loc.value_type.type_name().as_bytes())) +} + /// SCAN cursor [MATCH pattern] [COUNT count] [TYPE type] /// /// Incrementally iterates the key space. Returns a cursor and a batch of keys. @@ -985,14 +1008,26 @@ pub fn scan(db: &mut Database, args: &[Frame]) -> Frame { i += 1; } - // Collect all non-expired keys sorted for deterministic iteration + // Collect all non-expired keys sorted for deterministic iteration. + // Two planes, partitioned with no overlap (#364): hot = live in-RAM + // entries (exists() keeps its lazy-expiry reclamation side effect); + // cold = spilled keys with no live hot shadow (`cold_only_keys`, pure + // in-RAM index probe — no disk I/O). The `is_cold` tag lets the TYPE + // filter below avoid a promoting/disk-reading lookup on cold keys. + let now_ms = db.now_ms(); let all_keys: Vec = db.keys().cloned().collect(); - let mut sorted_keys: Vec = Vec::new(); + let mut sorted_keys: Vec<(CompactKey, bool)> = Vec::new(); for key in all_keys { - if db.exists(key.as_bytes()) { - sorted_keys.push(key); + let _ = db.exists(key.as_bytes()); + if db.get_if_alive(key.as_bytes(), now_ms).is_some() { + sorted_keys.push((key, false)); } } + let cold_keys: Vec = db + .cold_only_keys(now_ms) + .map(|k| CompactKey::from(k.as_ref())) + .collect(); + sorted_keys.extend(cold_keys.into_iter().map(|k| (k, true))); sorted_keys.sort(); let total = sorted_keys.len(); @@ -1002,18 +1037,19 @@ pub fn scan(db: &mut Database, args: &[Frame]) -> Frame { // Iterate from cursor position, collect up to `count` matching keys let mut checked = 0; while pos < total && checked < count { - let key = &sorted_keys[pos]; + let (key, is_cold) = &sorted_keys[pos]; pos += 1; checked += 1; // TYPE filter if let Some(tf) = type_filter { - if let Some(entry) = db.get(key.as_bytes()) { - let tn = entry.value.type_name().as_bytes(); - if !tf.eq_ignore_ascii_case(tn) { - continue; - } + let matches = if *is_cold { + cold_type_matches(db, key.as_bytes(), tf) } else { + db.get_if_alive(key.as_bytes(), now_ms) + .is_some_and(|e| tf.eq_ignore_ascii_case(e.value.type_name().as_bytes())) + }; + if !matches { continue; } } @@ -1146,10 +1182,18 @@ pub fn keys_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { let mut result = Vec::new(); for key in db.keys() { - if db.exists_if_alive(key.as_bytes(), now_ms) && glob_match(pattern, key.as_bytes()) { + // Strict hot-aliveness: cold-visible keys enter exactly once, via + // the cold loop below (#364 plane partition). + if db.get_if_alive(key.as_bytes(), now_ms).is_some() && glob_match(pattern, key.as_bytes()) + { result.push(Frame::BulkString(key.to_bytes())); } } + for key in db.cold_only_keys(now_ms) { + if glob_match(pattern, key.as_ref()) { + result.push(Frame::BulkString(key.clone())); + } + } Frame::Array(result.into()) } @@ -1207,12 +1251,19 @@ pub fn scan_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { i += 1; } - // Collect all non-expired keys sorted for deterministic iteration - let mut sorted_keys: Vec = db + // Collect all non-expired keys sorted for deterministic iteration. + // Hot plane (live in-RAM entries) unioned with cold-only spilled keys — + // the two are partitioned by `cold_only_keys`, so no dedup pass (#364). + let mut sorted_keys: Vec<(CompactKey, bool)> = db .keys() - .filter(|k| db.exists_if_alive(k.as_bytes(), now_ms)) + .filter(|k| db.get_if_alive(k.as_bytes(), now_ms).is_some()) .cloned() + .map(|k| (k, false)) .collect(); + sorted_keys.extend( + db.cold_only_keys(now_ms) + .map(|k| (CompactKey::from(k.as_ref()), true)), + ); sorted_keys.sort(); let total = sorted_keys.len(); @@ -1221,18 +1272,19 @@ pub fn scan_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { let mut checked = 0; while pos < total && checked < count { - let key = &sorted_keys[pos]; + let (key, is_cold) = &sorted_keys[pos]; pos += 1; checked += 1; // TYPE filter if let Some(tf) = type_filter { - if let Some(entry) = db.get_if_alive(key.as_bytes(), now_ms) { - let tn = entry.value.type_name().as_bytes(); - if !tf.eq_ignore_ascii_case(tn) { - continue; - } + let matches = if *is_cold { + cold_type_matches(db, key.as_bytes(), tf) } else { + db.get_if_alive(key.as_bytes(), now_ms) + .is_some_and(|e| tf.eq_ignore_ascii_case(e.value.type_name().as_bytes())) + }; + if !matches { continue; } } @@ -1378,6 +1430,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, } } @@ -2302,4 +2355,288 @@ mod tests { let result = object(&mut db, &[bs(b"BOGUS")]); assert!(matches!(result, Frame::Error(_))); } + + // --- Cold-plane enumeration tests (issue #364) --- + // + // Under disk-offload, keys spilled by eviction live ONLY in + // `Database::cold_index` (no in-RAM Entry). SCAN/KEYS/RANDOMKEY must + // enumerate them from the in-RAM index alone — no disk I/O. + + mod cold_enumeration { + use super::*; + use crate::storage::tiered::cold_index::{ColdIndex, ColdLocation}; + + /// Db with `hot` resident string keys and `cold` cold-only keys. + fn db_with_planes(hot: &[&[u8]], cold: &[(&[u8], Option)]) -> Database { + let mut db = Database::new(); + for k in hot { + db.set( + Bytes::copy_from_slice(k), + Entry::new_string(Bytes::from_static(b"v")), + ); + } + let mut ci = ColdIndex::new(); + for (i, (k, ttl_ms)) in cold.iter().enumerate() { + ci.insert( + Bytes::copy_from_slice(k), + ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: i as u16, + ttl_ms: *ttl_ms, + value_type: crate::persistence::kv_page::ValueType::String, + }, + ); + } + db.cold_index = Some(ci); + db + } + + /// Drive SCAN (mutable track) to completion, collecting every key. + fn full_scan(db: &mut Database, extra: &[&[u8]]) -> Vec { + let mut cursor = Bytes::from_static(b"0"); + let mut keys = Vec::new(); + loop { + let mut args = vec![Frame::BulkString(cursor.clone())]; + for e in extra { + args.push(bs(e)); + } + let Frame::Array(parts) = scan(db, &args) else { + panic!("SCAN did not return an array"); + }; + let Frame::BulkString(next) = &parts[0] else { + panic!("SCAN cursor not a bulk string"); + }; + let Frame::Array(batch) = &parts[1] else { + panic!("SCAN batch not an array"); + }; + for f in batch.iter() { + if let Frame::BulkString(k) = f { + keys.push(k.clone()); + } + } + if next.as_ref() == b"0" { + return keys; + } + cursor = next.clone(); + } + } + + fn full_scan_readonly(db: &Database, now_ms: u64, extra: &[&[u8]]) -> Vec { + let mut cursor = Bytes::from_static(b"0"); + let mut keys = Vec::new(); + loop { + let mut args = vec![Frame::BulkString(cursor.clone())]; + for e in extra { + args.push(bs(e)); + } + let Frame::Array(parts) = scan_readonly(db, &args, now_ms) else { + panic!("SCAN did not return an array"); + }; + let Frame::BulkString(next) = &parts[0] else { + panic!("SCAN cursor not a bulk string"); + }; + let Frame::Array(batch) = &parts[1] else { + panic!("SCAN batch not an array"); + }; + for f in batch.iter() { + if let Frame::BulkString(k) = f { + keys.push(k.clone()); + } + } + if next.as_ref() == b"0" { + return keys; + } + cursor = next.clone(); + } + } + + #[test] + fn scan_includes_cold_only_keys() { + let mut db = db_with_planes(&[b"hot1", b"hot2"], &[(b"cold1", None), (b"cold2", None)]); + let mut keys = full_scan(&mut db, &[b"COUNT", b"100"]); + keys.sort(); + assert_eq!(keys, vec!["cold1", "cold2", "hot1", "hot2"]); + } + + #[test] + fn scan_readonly_includes_cold_only_keys() { + let db = db_with_planes(&[b"hot1"], &[(b"cold1", None)]); + let now_ms = current_time_ms(); + let mut keys = full_scan_readonly(&db, now_ms, &[b"COUNT", b"100"]); + keys.sort(); + assert_eq!(keys, vec!["cold1", "hot1"]); + } + + #[test] + fn scan_returns_both_planes_key_exactly_once() { + // A key present in BOTH planes (hot shadow over a stale cold + // entry, e.g. after AOF-replay) must be returned exactly once. + let mut db = db_with_planes(&[b"both"], &[(b"both", None), (b"coldonly", None)]); + let mut keys = full_scan(&mut db, &[b"COUNT", b"100"]); + keys.sort(); + assert_eq!(keys, vec!["both", "coldonly"]); + } + + #[test] + fn scan_skips_ttl_expired_cold_keys() { + let now_ms = current_time_ms(); + let mut db = db_with_planes( + &[], + &[ + (b"alive", Some(now_ms + 60_000)), + (b"dead", Some(now_ms - 60_000)), + ], + ); + let keys = full_scan(&mut db, &[b"COUNT", b"100"]); + assert_eq!(keys, vec!["alive"]); + } + + #[test] + fn scan_match_filter_applies_to_cold_keys() { + let mut db = db_with_planes(&[b"user:1"], &[(b"user:2", None), (b"other", None)]); + let mut keys = full_scan(&mut db, &[b"MATCH", b"user:*", b"COUNT", b"100"]); + keys.sort(); + assert_eq!(keys, vec!["user:1", "user:2"]); + } + + #[test] + fn scan_small_count_pages_through_cold_keys() { + let mut db = db_with_planes( + &[b"h1", b"h2", b"h3"], + &[(b"c1", None), (b"c2", None), (b"c3", None)], + ); + // COUNT 1 forces one key per page — exercises cursor continuity + // across the hot/cold boundary. + let mut keys = full_scan(&mut db, &[b"COUNT", b"1"]); + keys.sort(); + assert_eq!(keys, vec!["c1", "c2", "c3", "h1", "h2", "h3"]); + } + + #[test] + fn keys_includes_cold_only_keys() { + let mut db = db_with_planes(&[b"hot1"], &[(b"cold1", None), (b"both", None)]); + db.set( + Bytes::from_static(b"both"), + Entry::new_string(Bytes::from_static(b"v")), + ); + let Frame::Array(arr) = keys(&mut db, &[bs(b"*")]) else { + panic!("KEYS did not return an array"); + }; + let mut got: Vec = arr + .iter() + .filter_map(|f| match f { + Frame::BulkString(b) => Some(b.clone()), + _ => None, + }) + .collect(); + got.sort(); + assert_eq!(got, vec!["both", "cold1", "hot1"]); + } + + #[test] + fn keys_readonly_includes_cold_only_keys() { + let db = db_with_planes(&[b"hot1"], &[(b"cold1", None)]); + let now_ms = current_time_ms(); + let Frame::Array(arr) = keys_readonly(&db, &[bs(b"*")], now_ms) else { + panic!("KEYS did not return an array"); + }; + let mut got: Vec = arr + .iter() + .filter_map(|f| match f { + Frame::BulkString(b) => Some(b.clone()), + _ => None, + }) + .collect(); + got.sort(); + assert_eq!(got, vec!["cold1", "hot1"]); + } + + #[test] + fn scan_type_filter_judges_cold_keys_from_index() { + use crate::persistence::kv_page::ValueType; + // One cold hash + one cold string + one hot string. TYPE hash + // must surface ONLY the cold hash — judged from the in-RAM + // ColdLocation::value_type cache, no disk read available here + // (no spill file exists behind these locations). + let mut db = db_with_planes(&[b"hotstr"], &[]); + let mut ci = ColdIndex::new(); + ci.insert( + Bytes::from_static(b"coldhash"), + ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: 0, + ttl_ms: None, + value_type: ValueType::Hash, + }, + ); + ci.insert( + Bytes::from_static(b"coldstr"), + ColdLocation { + file_id: 1, + page_idx: 0, + slot_idx: 1, + ttl_ms: None, + value_type: ValueType::String, + }, + ); + db.cold_index = Some(ci); + + let hashes = full_scan(&mut db, &[b"TYPE", b"hash", b"COUNT", b"100"]); + assert_eq!(hashes, vec!["coldhash"]); + + let mut strings = full_scan(&mut db, &[b"TYPE", b"string", b"COUNT", b"100"]); + strings.sort(); + assert_eq!(strings, vec!["coldstr", "hotstr"]); + + // Read-only twin must agree. + let now_ms = current_time_ms(); + let ro_hashes = full_scan_readonly(&db, now_ms, &[b"TYPE", b"hash", b"COUNT", b"100"]); + assert_eq!(ro_hashes, vec!["coldhash"]); + } + + #[test] + fn randomkey_sees_all_cold_database() { + // Every key spilled: RANDOMKEY must not report an empty db. + let mut db = db_with_planes(&[], &[(b"cold1", None), (b"cold2", None)]); + match randomkey(&mut db, &[]) { + Frame::BulkString(k) => { + assert!(k.as_ref() == b"cold1" || k.as_ref() == b"cold2"); + } + other => panic!("expected a key, got {other:?}"), + } + } + + #[test] + fn randomkey_readonly_sees_all_cold_database() { + let db = db_with_planes(&[], &[(b"cold1", None)]); + match randomkey_readonly(&db, &[], current_time_ms()) { + Frame::BulkString(k) => assert_eq!(k.as_ref(), b"cold1"), + other => panic!("expected a key, got {other:?}"), + } + } + + #[test] + fn del_of_ttl_expired_cold_key_answers_zero_but_reclaims() { + // Redis parity: DEL of a logically-expired key deletes nothing + // (returns 0) — but the stale cold-index entry must still be + // reclaimed as a side effect, not left behind. + let now_ms = current_time_ms(); + let mut db = db_with_planes(&[], &[(b"dead", Some(now_ms - 60_000))]); + assert_eq!(del(&mut db, &[bs(b"dead")]), Frame::Integer(0)); + assert!( + db.cold_index + .as_ref() + .is_some_and(|ci| ci.lookup(b"dead").is_none()), + "stale cold entry must be reclaimed by the DEL attempt" + ); + } + + #[test] + fn unlink_of_alive_cold_key_still_counts() { + let mut db = db_with_planes(&[], &[(b"alive", None)]); + assert_eq!(unlink(&mut db, &[bs(b"alive")]), Frame::Integer(1)); + } + } } diff --git a/src/persistence/kv_page.rs b/src/persistence/kv_page.rs index 76af2983e..ceb43b678 100644 --- a/src/persistence/kv_page.rs +++ b/src/persistence/kv_page.rs @@ -65,6 +65,21 @@ impl ValueType { _ => None, } } + + /// Redis TYPE-command name for this value type. Must stay in sync with + /// `RedisValue::type_name` — SCAN's TYPE filter compares the two + /// case-insensitively when judging cold (spilled) keys (#364). + #[inline] + pub fn type_name(self) -> &'static str { + match self { + Self::String => "string", + Self::Hash => "hash", + Self::List => "list", + Self::Set => "set", + Self::ZSet => "zset", + Self::Stream => "stream", + } + } } // ── Entry flags (bitfield) ────────────────────────────── diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index 3d7b22500..d6acb6a93 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -622,6 +622,7 @@ fn apply_completion_vec( page_idx: entry.page_idx, slot_idx: entry.slot_idx, ttl_ms: entry.ttl_ms, + value_type: entry.value_type, }; crate::shard::slice::with_shard_db(entry.db_index, |db| { diff --git a/src/storage/db.rs b/src/storage/db.rs index a98fa8f24..4d5cab848 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -1264,12 +1264,21 @@ impl Database { /// Remove hot + cold copies; returns `true` when EITHER existed, so /// DEL/UNLINK count spilled keys as removed (Redis semantics: the key - /// logically exists). The removed hot entry, when present, is also - /// returned so UNLINK can size its async-drop decision. + /// logically exists). A cold entry that is already TTL-expired (judged + /// from the cached `ColdLocation::ttl_ms`, no disk read) is reclaimed + /// but NOT counted — DEL of a logically-expired key answers 0. The + /// removed hot entry, when present, is also returned so UNLINK can + /// size its async-drop decision. pub fn remove_counting_cold(&mut self, key: &[u8]) -> (bool, Option) { + let now_ms = self.cached_now_ms; + let cold_alive = self + .cold_index + .as_ref() + .and_then(|ci| ci.lookup(key)) + .is_some_and(|loc| loc.ttl_ms.is_none_or(|ttl| now_ms <= ttl)); let had_cold = self.remove_cold_only(key); let hot = self.remove_hot(key); - (hot.is_some() || had_cold, hot) + (hot.is_some() || (had_cold && cold_alive), hot) } #[inline] @@ -1406,29 +1415,79 @@ impl Database { } /// Iterator over all keys (caller does glob filtering). + /// + /// Hot plane only: under disk-offload, spilled keys have no in-RAM + /// `Entry` and are NOT yielded here — keyspace enumerators (SCAN / + /// KEYS / RANDOMKEY) must union this with [`Self::cold_only_keys`] + /// or spilled keys silently vanish from enumeration (#364). pub fn keys(&self) -> impl Iterator { self.data.keys() } + /// Keys visible ONLY via the cold plane at `now_ms`: present in the + /// in-RAM cold index, not TTL-expired (judged from the cached + /// [`crate::storage::tiered::cold_index::ColdLocation::ttl_ms`] — no + /// disk I/O), and NOT shadowed by a live hot-plane entry. + /// + /// Together with the hot-alive subset of [`Self::keys`] this + /// partitions the logical keyspace with no overlap, so keyspace + /// enumerators (SCAN / KEYS / RANDOMKEY, #364) can take the union of + /// the two planes without any dedup pass. A key present in BOTH + /// planes (hot shadow over a stale cold entry, e.g. after AOF replay) + /// 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 + '_ { + let base_ts = self.base_timestamp; + self.cold_index.as_ref().into_iter().flat_map(move |ci| { + ci.iter().filter_map(move |(key, loc)| { + let cold_alive = loc.ttl_ms.is_none_or(|ttl| now_ms <= ttl); + if !cold_alive { + return None; + } + let hot_alive = self + .data + .get(key.as_ref()) + .is_some_and(|e| !e.is_expired_at(base_ts, now_ms)); + if hot_alive { None } else { Some(key) } + }) + }) + } + /// Return a random non-expired key from the database, or None if empty. + /// + /// Samples the LOGICAL keyspace: hot-alive entries plus cold-only + /// spilled keys ([`Self::cold_only_keys`]) — an all-spilled database + /// must not answer "empty" (#364). + /// + /// Two passes — count, then walk to the selected position — so only + /// the winning key is ever cloned (the previous single-pass version + /// materialized a `Bytes` copy of EVERY live key per call). Stable + /// across the passes: `&self` is held throughout and each shard's + /// database is single-threaded, so neither plane can mutate between + /// the count and the walk. pub fn random_key(&self) -> Option { - if self.data.is_empty() { - return None; - } let now_ms = self.cached_now_ms; let base_ts = self.base_timestamp; - // Collect non-expired keys (iterator is already O(n)) - let live: Vec<_> = self + let hot_live = self .data .iter() .filter(|(_, e)| !e.is_expired_at(base_ts, now_ms)) - .map(|(k, _)| Bytes::copy_from_slice(k.as_ref())) - .collect(); - if live.is_empty() { + .count(); + let cold_live = self.cold_only_keys(now_ms).count(); + let total = hot_live + cold_live; + if total == 0 { return None; } - let idx = (current_time_ms() as usize) % live.len(); - Some(live.into_iter().nth(idx).unwrap_or_default()) + let idx = (current_time_ms() as usize) % total; + if idx < hot_live { + self.data + .iter() + .filter(|(_, e)| !e.is_expired_at(base_ts, now_ms)) + .nth(idx) + .map(|(k, _)| Bytes::copy_from_slice(k.as_ref())) + } else { + self.cold_only_keys(now_ms).nth(idx - hot_live).cloned() + } } /// Set or remove expiration on an existing key. @@ -2673,6 +2732,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); db.cold_index = Some(ci); @@ -2738,6 +2798,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); db.cold_index = Some(ci); diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index 243862f96..a1812373a 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -971,6 +971,7 @@ fn evict_batch_durable_no_aof( page_idx: entry.page_idx, slot_idx: entry.slot_idx, ttl_ms: entry.ttl_ms, + value_type: entry.value_type, }, ); } @@ -1177,6 +1178,7 @@ pub(crate) fn evict_one_with_spill( let mut spilled = false; let mut spilled_file_id = 0u64; let mut spilled_ttl_ms: Option = None; + let mut spilled_value_type = crate::persistence::kv_page::ValueType::String; if let Some(ctx) = spill { if let Some(entry) = db.data().get(key.as_bytes()) { let file_id = *ctx.next_file_id; @@ -1208,6 +1210,7 @@ pub(crate) fn evict_one_with_spill( spilled = true; spilled_file_id = file_id; spilled_ttl_ms = ttl_ms; + spilled_value_type = kv_spill::value_type_of(&entry.as_redis_value()); } } @@ -1230,6 +1233,7 @@ pub(crate) fn evict_one_with_spill( page_idx: 0, slot_idx: 0, ttl_ms: spilled_ttl_ms, + value_type: spilled_value_type, }, ); } diff --git a/src/storage/tiered/cold_index.rs b/src/storage/tiered/cold_index.rs index e762ce875..3fe297481 100644 --- a/src/storage/tiered/cold_index.rs +++ b/src/storage/tiered/cold_index.rs @@ -59,6 +59,19 @@ pub struct ColdLocation { /// index format to version and no old-format file that could ever fail to /// load because of it. pub ttl_ms: Option, + /// Redis value type of the on-disk entry, mirroring the on-disk + /// `KvEntry::value_type` this location points at. + /// + /// Same cached-copy contract as `ttl_ms` (#364): populated at insert + /// time (spill / recovery rebuild) purely so SCAN's TYPE filter can + /// judge a cold-only key from the in-RAM index alone — WITHOUT a pread + /// of the cold file and WITHOUT promoting the entry into hot RAM. The + /// on-disk `KvLeafPage` entry stays authoritative and + /// [`ColdIndex::rebuild_from_manifest`] re-derives this field from it + /// after a restart; no on-disk format changes. Fits the struct's + /// existing padding (after `slot_idx`), so the in-RAM index does not + /// grow. + pub value_type: crate::persistence::kv_page::ValueType, } /// In-memory index from key to cold disk location. @@ -620,6 +633,7 @@ impl ColdIndex { page_idx: page_idx as u32, slot_idx, ttl_ms: kv.ttl_ms, + value_type: kv.value_type, }, ); } @@ -644,6 +658,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }; idx.insert(Bytes::from_static(b"key1"), loc); assert_eq!(idx.len(), 1); @@ -670,6 +685,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }; idx.insert(Bytes::from_static(b"a_reasonably_long_key"), loc); let after_one = idx.resident_bytes(); @@ -693,12 +709,14 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }; let loc_b = ColdLocation { file_id: 2, page_idx: 0, slot_idx: 1, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }; idx.insert(Bytes::from_static(b"key1"), loc_a); let after_first = idx.resident_bytes(); @@ -717,6 +735,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }; idx.insert(Bytes::from_static(b"key1"), loc); idx.insert(Bytes::from_static(b"key2"), loc); @@ -763,6 +782,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); ci.insert( @@ -772,6 +792,7 @@ mod tests { page_idx: 0, slot_idx: 1, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -816,6 +837,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); ci.insert( @@ -825,6 +847,7 @@ mod tests { page_idx: 0, slot_idx: 1, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -862,6 +885,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); // Key re-spilled to a NEW file (re-eviction) -> file 10 orphaned. @@ -872,6 +896,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -910,6 +935,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: Some(1_000), // expires at t=1000ms + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -960,6 +986,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); ci.insert( @@ -969,6 +996,7 @@ mod tests { page_idx: 0, slot_idx: 1, ttl_ms: Some(5_000), + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -1002,6 +1030,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: Some(1_000), + value_type: crate::persistence::kv_page::ValueType::String, }, ); ci.insert( @@ -1011,6 +1040,7 @@ mod tests { page_idx: 0, slot_idx: 1, ttl_ms: Some(9_999_000), + value_type: crate::persistence::kv_page::ValueType::String, }, ); @@ -1059,6 +1089,7 @@ mod tests { page_idx: 0, slot_idx: i, ttl_ms: Some(1_000), + value_type: crate::persistence::kv_page::ValueType::String, }, ); } @@ -1106,6 +1137,7 @@ mod tests { page_idx: 0, slot_idx: 0, ttl_ms: None, + value_type: crate::persistence::kv_page::ValueType::String, }, ); diff --git a/src/storage/tiered/kv_spill.rs b/src/storage/tiered/kv_spill.rs index 33620ab63..e92ed1a89 100644 --- a/src/storage/tiered/kv_spill.rs +++ b/src/storage/tiered/kv_spill.rs @@ -105,6 +105,29 @@ pub fn write_kv_spill_pages( Ok((pages.total_pages as u64) * (PAGE_4K as u64)) } +/// On-disk [`ValueType`] tag for a hot value. +/// +/// Exhaustive over `RedisValueRef` by design: adding a new value variant +/// without a `ValueType` mapping must be a compile error, never a silent +/// mis-typed spill. Used by every spill entry point and by the sync +/// eviction path to populate `ColdLocation::value_type` (#364). +pub fn value_type_of(val: &RedisValueRef) -> ValueType { + match val { + RedisValueRef::String(_) => ValueType::String, + RedisValueRef::Hash(_) + | RedisValueRef::HashListpack(_) + | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, + RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, + RedisValueRef::Set(_) | RedisValueRef::SetListpack(_) | RedisValueRef::SetIntset(_) => { + ValueType::Set + } + RedisValueRef::SortedSet { .. } + | RedisValueRef::SortedSetBPTree { .. } + | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, + RedisValueRef::Stream(_) => ValueType::Stream, + } +} + /// Spill a single evicted KV entry to a DataFile on disk. /// /// Creates a single-page `.mpf` file at `{shard_dir}/data/heap-{file_id:06}.mpf`, @@ -133,22 +156,8 @@ pub fn spill_to_datafile( let (value_type, value_bytes): (ValueType, &[u8]) = match val_ref { RedisValueRef::String(s) => (ValueType::String, s), ref other => { - let vt = match other { - RedisValueRef::Hash(_) - | RedisValueRef::HashListpack(_) - | RedisValueRef::HashWithTtl { .. } => ValueType::Hash, - RedisValueRef::List(_) | RedisValueRef::ListListpack(_) => ValueType::List, - RedisValueRef::Set(_) - | RedisValueRef::SetListpack(_) - | RedisValueRef::SetIntset(_) => ValueType::Set, - RedisValueRef::SortedSet { .. } - | RedisValueRef::SortedSetBPTree { .. } - | RedisValueRef::SortedSetListpack(_) => ValueType::ZSet, - RedisValueRef::Stream(_) => ValueType::Stream, - RedisValueRef::String(_) => unreachable!(), - }; collection_buf = kv_serde::serialize_collection(other).unwrap_or_default(); - (vt, collection_buf.as_slice()) + (value_type_of(other), collection_buf.as_slice()) } }; @@ -201,6 +210,7 @@ pub fn spill_to_datafile( page_idx: 0, slot_idx: 0, ttl_ms, + value_type, }, ); } @@ -1049,6 +1059,7 @@ mod tests { page_idx, slot_idx, ttl_ms: None, + value_type: ValueType::String, }; let result = read_cold_entry_at(shard_dir, loc, 0); assert!( @@ -1101,6 +1112,7 @@ mod tests { page_idx, slot_idx, ttl_ms: None, + value_type: ValueType::String, }; let result = read_cold_entry_at(shard_dir, loc, 0); assert!( @@ -1265,6 +1277,7 @@ mod tests { page_idx, slot_idx, ttl_ms: None, + value_type: ValueType::String, }; let result = read_cold_entry_at(shard_dir, loc, 0); assert!( @@ -1326,6 +1339,7 @@ mod tests { page_idx, slot_idx, ttl_ms: None, + value_type: ValueType::String, }; let result = read_cold_entry_at(shard_dir, loc, 0); assert!( diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index aa2cd5beb..95252b49a 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -204,6 +204,10 @@ pub struct SpillCompletionEntry { /// `ColdLocation::ttl_ms` (R1, H-2: proactive TTL-expiry sweep) without /// re-reading the just-written file. pub ttl_ms: Option, + /// Value type, likewise carried through from the `SpillRequest` so the + /// event loop can populate `ColdLocation::value_type` (#364: SCAN TYPE + /// filter over cold keys) without re-reading the just-written file. + pub value_type: ValueType, } /// Completion sent from background thread back to event loop. @@ -327,6 +331,7 @@ pub(crate) fn flush_buffer(buffer: &mut Vec) -> Vec SpillCompletion { page_idx: 0, slot_idx: 0, ttl_ms: req.ttl_ms, + value_type: req.value_type, }], success: true, }, @@ -843,6 +849,7 @@ mod tests { page_idx: entry.page_idx, slot_idx: entry.slot_idx, ttl_ms: entry.ttl_ms, + value_type: ValueType::String, }; let result = read_cold_entry_at(tmp.path(), loc, 0); assert!(result.is_some(), "should read entry back"); @@ -1058,6 +1065,7 @@ mod tests { page_idx: entry.page_idx, slot_idx: entry.slot_idx, ttl_ms: entry.ttl_ms, + value_type: ValueType::String, }; let outcome = crate::storage::tiered::cold_read::read_cold_entry_at( tmp.path(), @@ -1163,6 +1171,7 @@ mod tests { page_idx: entry.page_idx, slot_idx: entry.slot_idx, ttl_ms: entry.ttl_ms, + value_type: ValueType::String, }; let result = crate::storage::tiered::cold_read::read_cold_entry_at(tmp.path(), loc, 0); diff --git a/tests/scan_offload_visibility.rs b/tests/scan_offload_visibility.rs new file mode 100644 index 000000000..ae85997d0 --- /dev/null +++ b/tests/scan_offload_visibility.rs @@ -0,0 +1,398 @@ +//! Issue #364: SCAN/KEYS/RANDOMKEY must enumerate the LOGICAL keyspace +//! under disk-offload, not just the hot plane. +//! +//! Found while fixing #355 (DBSIZE): with disk-offload enabled, spilled +//! keys were readable (GET/EXISTS) but invisible to enumeration — a +//! 4-shard instance holding 400 logical keys returned only 116 from +//! `redis-cli --scan` (hot residents only). Any SCAN consumer doing +//! migration/backup (`--scan | xargs MIGRATE`) silently lost spilled keys. +//! +//! This test drives the fix end-to-end: writes ~1.5x `--maxmemory` of +//! string keys plus a batch of hashes under disk-offload, confirms real +//! spill happened (ground truth: `heap-*.mpf` files), then asserts +//! 1. a full SCAN loop returns EVERY written key exactly (dedup'd), +//! 2. KEYS * agrees, +//! 3. RANDOMKEY answers non-nil, +//! 4. `SCAN ... TYPE hash` returns exactly the hash keys (cold keys are +//! judged from the in-RAM `ColdLocation::value_type` cache — no disk +//! reads, no promotion), +//! +//! then SIGKILLs the server, restarts on the same `--dir` (all spilled +//! keys recover as cold-only stubs, `ColdIndex::rebuild_from_manifest` +//! re-derives `value_type` from the on-disk pages), and asserts 1 and 4 +//! again post-restart. +//! +//! Run with (monoio default — matches CI): +//! cargo build --release +//! cargo test --release --test scan_offload_visibility -- --ignored --nocapture +//! +//! Requires: built release binary, `redis-cli` on PATH. + +#![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] + +mod common; + +use std::collections::HashSet; +use std::io::{BufRead, BufReader, Write}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +/// 8 MiB cap — small enough for a fast local test, large enough that the +/// filler below forces real eviction + disk spill. +const MAXMEMORY_BYTES: usize = 8 * 1024 * 1024; +const SHARDS: usize = 2; + +/// ~1.5x MAXMEMORY_BYTES of raw string payload — enough to force spill +/// without the long fill time of the 10x used_memory test. +const FILLER_COUNT: usize = 6_000; +const FILLER_VALUE_LEN: usize = 2_000; +/// Hash keys interleaved with the strings so the TYPE filter has both +/// planes and both types to discriminate. +const HASH_COUNT: usize = 50; + +/// Kill-on-drop child guard: a mid-test panic (failed assert) must never +/// orphan the spawned server. A leaked moon whose tmpdir is later cleaned +/// spins its persistence tick at ~100% CPU per shard thread with no +/// backoff (issue #366) — observed live at 667% CPU on 2026-07-17. +struct MoonGuard(Option); + +impl MoonGuard { + /// Kill + reap now (the deliberate mid-test SIGKILL leg). + fn kill_now(&mut self) { + if let Some(mut c) = self.0.take() { + common::sigkill(&mut c); + } + } +} + +impl Drop for MoonGuard { + fn drop(&mut self) { + if let Some(mut c) = self.0.take() { + let _ = c.kill(); + let _ = c.wait(); + } + } +} + +fn redis_cli_available() -> bool { + Command::new("redis-cli") + .arg("--version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|s| s.success()) + .unwrap_or(false) +} + +fn unique_dir(suffix: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + std::env::temp_dir().join(format!( + "moon-scan-offload-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +fn start_moon(port: u16, dir: &std::path::Path) -> Child { + let off_dir = dir.join("off"); + std::fs::create_dir_all(&off_dir).expect("create off dir"); + Command::new(common::find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--shards", + &SHARDS.to_string(), + "--admin-port", + "0", + "--maxmemory", + &MAXMEMORY_BYTES.to_string(), + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "enable", + "--disk-offload-dir", + off_dir.to_str().expect("off dir utf8"), + // Durability backstop required or disk-offload spill is inert + // (config::disk_offload_spill_inert) — eviction would plain-drop + // victims instead of spilling them. + "--appendonly", + "yes", + "--disk-free-min-pct", + "0", + "--dir", + ]) + .arg(dir) + // Captured to a log file, never Stdio::null(): a CI flake needs a + // real diagnostic, not silence. + .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 (run `cargo build --release` first)") +} + +const RESTART_ATTEMPTS: usize = 6; + +/// Start moon and return a child that is alive AND accepting PING, retrying +/// on a transient rebind EADDRINUSE self-termination (same pattern as +/// tests/used_memory_offload_truthful.rs). +fn start_moon_alive(port: u16, dir: &std::path::Path) -> Child { + for attempt in 1..=RESTART_ATTEMPTS { + let mut child = start_moon(port, dir); + let deadline = Instant::now() + Duration::from_secs(10); + let mut up = false; + while Instant::now() < deadline { + if let Ok(Some(_status)) = child.try_wait() { + break; // self-terminated — fall through to retry + } + if redis_cli(port, &["PING"]).as_deref() == Some("PONG") { + up = true; + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + if up { + return child; + } + let _ = child.kill(); + let _ = child.wait(); + if attempt < RESTART_ATTEMPTS { + std::thread::sleep(Duration::from_millis(300)); + } + } + panic!( + "moon failed to start+serve on port {} after {} attempts", + port, RESTART_ATTEMPTS + ); +} + +fn wait_for_ping(port: u16, deadline: Duration) { + let end = Instant::now() + deadline; + while Instant::now() < end { + if redis_cli(port, &["PING"]).as_deref() == Some("PONG") { + return; + } + std::thread::sleep(Duration::from_millis(100)); + } + panic!("moon did not respond to PING within {deadline:?} on port {port}"); +} + +fn redis_cli(port: u16, args: &[&str]) -> Option { + let output = Command::new("redis-cli") + .args(["-p", &port.to_string()]) + .args(args) + .output() + .ok()?; + let s = String::from_utf8_lossy(&output.stdout).trim().to_string(); + if s.is_empty() { None } else { Some(s) } +} + +/// Push `FILLER_COUNT` string keys via many small, paced `MSET` batches +/// (paced so the background SpillThread drains its bounded channel between +/// bursts — see tests/used_memory_offload_truthful.rs `write_filler` for +/// the full rationale), plus `HASH_COUNT` hashes via HSET. +fn write_dataset(port: u16) { + const FILLER_BATCH_SIZE: usize = 400; + let val = "F".repeat(FILLER_VALUE_LEN); + let mut written = 0usize; + while written < FILLER_COUNT { + let batch = FILLER_BATCH_SIZE.min(FILLER_COUNT - written); + let mut stream = std::net::TcpStream::connect(format!("127.0.0.1:{port}")) + .expect("connect for filler batch"); + stream.set_write_timeout(Some(Duration::from_secs(30))).ok(); + stream.set_read_timeout(Some(Duration::from_secs(30))).ok(); + + let total_args = 1 + 2 * batch; + let mut buf: Vec = Vec::with_capacity(batch * (FILLER_VALUE_LEN + 32)); + buf.extend_from_slice(format!("*{total_args}\r\n$4\r\nMSET\r\n").as_bytes()); + for i in written..written + batch { + let key = format!("filler:{i}"); + buf.extend_from_slice( + format!("${}\r\n{}\r\n${}\r\n{}\r\n", key.len(), key, val.len(), val).as_bytes(), + ); + } + stream.write_all(&buf).expect("filler MSET batch write"); + + let mut reply = String::new(); + let mut reader = BufReader::new(&stream); + reader + .read_line(&mut reply) + .expect("filler MSET batch reply"); + assert!( + reply.starts_with('+'), + "filler MSET batch (keys {written}..{}) must succeed, got: {reply}", + written + batch + ); + + written += batch; + std::thread::sleep(Duration::from_millis(30)); + } + + for i in 0..HASH_COUNT { + let key = format!("hobj:{i}"); + let reply = redis_cli(port, &["HSET", &key, "f", "v"]); + assert!( + reply.as_deref().is_some_and(|r| r.parse::().is_ok()), + "HSET {key} must succeed, got: {reply:?}" + ); + } +} + +fn count_heap_files(dir: &std::path::Path) -> usize { + let off = dir.join("off"); + fn walk(p: &std::path::Path, acc: &mut usize) { + if let Ok(rd) = std::fs::read_dir(p) { + for e in rd.flatten() { + let path = e.path(); + if path.is_dir() { + walk(&path, acc); + } else if path + .file_name() + .and_then(|n| n.to_str()) + .map(|n| n.starts_with("heap-") && n.ends_with(".mpf")) + .unwrap_or(false) + { + *acc += 1; + } + } + } + } + let mut acc = 0; + walk(&off, &mut acc); + acc +} + +/// Drive a full SCAN loop (with optional extra args like `TYPE hash`) to +/// cursor 0, returning the dedup'd key set. Bounded to catch a cursor that +/// never converges. +fn full_scan(port: u16, extra: &[&str]) -> HashSet { + let mut keys = HashSet::new(); + let mut cursor = String::from("0"); + for _round in 0..10_000 { + let mut args: Vec<&str> = vec!["SCAN", &cursor, "COUNT", "1000"]; + args.extend_from_slice(extra); + let out = redis_cli(port, &args).unwrap_or_default(); + let mut lines = out.lines(); + let next = lines + .next() + .unwrap_or_else(|| panic!("SCAN (extra={extra:?}) returned empty reply")) + .trim() + .to_string(); + for l in lines { + let l = l.trim(); + if !l.is_empty() { + keys.insert(l.to_string()); + } + } + if next == "0" { + return keys; + } + cursor = next; + } + panic!("SCAN cursor did not converge to 0 within 10000 rounds (extra={extra:?})"); +} + +fn expected_keys() -> HashSet { + let mut set: HashSet = (0..FILLER_COUNT).map(|i| format!("filler:{i}")).collect(); + set.extend((0..HASH_COUNT).map(|i| format!("hobj:{i}"))); + set +} + +fn expected_hash_keys() -> HashSet { + (0..HASH_COUNT).map(|i| format!("hobj:{i}")).collect() +} + +/// Assert `got` covers exactly `want`, printing a small sample of the +/// difference on failure (6000 raw keys in a panic message helps nobody). +fn assert_keyset(context: &str, got: &HashSet, want: &HashSet) { + let missing: Vec<_> = want.difference(got).take(10).collect(); + let extra: Vec<_> = got.difference(want).take(10).collect(); + assert!( + missing.is_empty() && extra.is_empty(), + "{context}: keyset mismatch — got {} keys, want {} keys; \ + first missing: {missing:?}, first unexpected: {extra:?}", + got.len(), + want.len() + ); +} + +#[test] +#[ignore] // Requires built release binary + redis-cli; run explicitly. +fn scan_keys_randomkey_enumerate_spilled_keys_and_survive_restart() { + if !redis_cli_available() { + eprintln!("skipping: redis-cli not in PATH"); + return; + } + + let dir = unique_dir("t364"); + std::fs::create_dir_all(&dir).expect("create test dir"); + + // -- Round 1: populate past the cap, force real disk spill ----------- + let (child, port) = common::spawn_listening(|p| start_moon(p, &dir)); + let mut child = MoonGuard(Some(child)); + wait_for_ping(port, Duration::from_secs(10)); + + write_dataset(port); + // Let the async spill thread + periodic eviction tick drain and commit + // manifests before enumerating. + std::thread::sleep(Duration::from_secs(6)); + + let heap_files = count_heap_files(&dir); + eprintln!("scan_offload_visibility: heap_files={heap_files}"); + assert!( + heap_files > 0, + "test setup: expected real disk spill (heap-*.mpf files), found none — \ + eviction never triggered, this test exercised nothing" + ); + + let want = expected_keys(); + let want_hashes = expected_hash_keys(); + + // 1. Full SCAN loop sees the whole logical keyspace. + let scanned = full_scan(port, &[]); + assert_keyset("SCAN (pre-restart)", &scanned, &want); + + // 2. KEYS * agrees. + let keys_out = redis_cli(port, &["KEYS", "*"]).unwrap_or_default(); + let keys_set: HashSet = keys_out + .lines() + .map(|l| l.trim().to_string()) + .filter(|l| !l.is_empty()) + .collect(); + assert_keyset("KEYS * (pre-restart)", &keys_set, &want); + + // 3. RANDOMKEY answers non-nil on a database that is mostly spilled. + let rk = redis_cli(port, &["RANDOMKEY"]); + assert!( + rk.as_deref().is_some_and(|k| want.contains(k)), + "RANDOMKEY must return a logical key, got: {rk:?}" + ); + + // 4. TYPE filter discriminates cold keys from the in-RAM index. + let scanned_hashes = full_scan(port, &["TYPE", "hash"]); + assert_keyset( + "SCAN TYPE hash (pre-restart)", + &scanned_hashes, + &want_hashes, + ); + + // -- SIGKILL + restart on the SAME dir/port --------------------------- + // All spilled keys recover as cold-only stubs; ColdIndex::rebuild_from_manifest + // re-derives ttl_ms AND value_type from the on-disk pages. + child.kill_now(); + common::wait_for_port_down(port); + + let mut child2 = MoonGuard(Some(start_moon_alive(port, &dir))); + + let scanned_after = full_scan(port, &[]); + let hashes_after = full_scan(port, &["TYPE", "hash"]); + child2.kill_now(); + + assert_keyset("SCAN (post-restart)", &scanned_after, &want); + assert_keyset("SCAN TYPE hash (post-restart)", &hashes_after, &want_hashes); + + let _ = std::fs::remove_dir_all(&dir); +}