diff --git a/CHANGELOG.md b/CHANGELOG.md index 94095d2b..0fb7f56b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ephemeral; restart scans at 0). ### Fixed +- **Cold recovery re-attaches spilled keys to the logical database they + were evicted from (#139).** Under disk-offload with `SELECT N` (N>0), + the recovery rebuild used to merge every spilled key into db 0's cold + index: SELECT>0 keys answered nil after restart (their only durable + copy unreachable), and a same-named key could even serve the WRONG + database's value from db 0. Spill files are now attributed wholesale + via a `db_index` field in the manifest `FileEntry` (a byte-compatible + repurpose of the always-written-as-zero `min_key_hash` slot — old + manifests read as db 0, which matches their actual provenance), spill + flush chunks never cross a db boundary so each file is single-db, and + recovery attaches one rebuilt cold index per database before WAL/AOF + tail replay (so replayed DEL/FLUSH still tombstone the right plane). A + manifest referencing a db beyond the configured `--databases` count is + skipped with a loud warning instead of silently resurrecting keys into + a wrong database. - **SCAN now honors the Redis stable-key guarantee under churn (#368).** The cursor was a positional index into a keyspace snapshot re-collected and re-sorted on every page, so any insert/delete (or cold-plane diff --git a/src/persistence/manifest.rs b/src/persistence/manifest.rs index ec7f9ddb..7f3cfa8c 100644 --- a/src/persistence/manifest.rs +++ b/src/persistence/manifest.rs @@ -94,7 +94,7 @@ impl StorageTier { /// 12..16 4 page_count (u32 LE) /// 16..24 8 byte_size (u64 LE) /// 24..32 8 created_lsn (u64 LE) -/// 32..40 8 min_key_hash (u64 LE) +/// 32..40 8 db_index (u64 LE; formerly min_key_hash, always written 0 — see field doc) /// 40..48 8 max_key_hash (u64 LE) /// 48..56 8 last_modified_lsn (u64 LE) -- v2 only /// ``` @@ -108,7 +108,18 @@ pub struct FileEntry { pub page_count: u32, pub byte_size: u64, pub created_lsn: u64, - pub min_key_hash: u64, + /// Logical database index every entry in this file belongs to (#139: + /// cold recovery must re-attach spilled keys to the database they were + /// evicted from, not unconditionally db 0). + /// + /// Byte-compatible repurpose of the former `min_key_hash` field, which + /// every writer in the codebase's history serialized as 0 — so v1/v2 + /// manifests written before this field existed read back as db 0, + /// which is exactly correct for them (the spill buffer was db-blind + /// and recovery attached everything to db 0). The spill path + /// guarantees single-db files by cutting flush chunks at db + /// boundaries (`spill_thread::flush_buffer`). + pub db_index: u64, pub max_key_hash: u64, /// LSN of the last mutation to this file (v2). For files imported from /// a v1 manifest this is set equal to `created_lsn`. Used by PITR to @@ -144,7 +155,7 @@ impl FileEntry { buf[12..16].copy_from_slice(&self.page_count.to_le_bytes()); buf[16..24].copy_from_slice(&self.byte_size.to_le_bytes()); buf[24..32].copy_from_slice(&self.created_lsn.to_le_bytes()); - buf[32..40].copy_from_slice(&self.min_key_hash.to_le_bytes()); + buf[32..40].copy_from_slice(&self.db_index.to_le_bytes()); buf[40..48].copy_from_slice(&self.max_key_hash.to_le_bytes()); buf[48..56].copy_from_slice(&self.last_modified_lsn.to_le_bytes()); } @@ -187,7 +198,7 @@ impl FileEntry { let created_lsn = u64::from_le_bytes([ buf[24], buf[25], buf[26], buf[27], buf[28], buf[29], buf[30], buf[31], ]); - let min_key_hash = u64::from_le_bytes([ + let db_index = u64::from_le_bytes([ buf[32], buf[33], buf[34], buf[35], buf[36], buf[37], buf[38], buf[39], ]); let max_key_hash = u64::from_le_bytes([ @@ -203,7 +214,7 @@ impl FileEntry { page_count, byte_size, created_lsn, - min_key_hash, + db_index, max_key_hash, // v1 fallback — synthesize from created_lsn so existing manifests // remain readable and PITR target_lsn comparisons stay sane. @@ -1194,7 +1205,7 @@ mod tests { page_count: 1000, byte_size: 4_096_000, created_lsn: 42, - min_key_hash: 0x1111_2222_3333_4444, + db_index: 0x1111_2222_3333_4444, max_key_hash: 0xAAAA_BBBB_CCCC_DDDD, last_modified_lsn: 4242, }; @@ -1217,7 +1228,7 @@ mod tests { page_count: 500, byte_size: 32_768_000, created_lsn: 100, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: 200, }; @@ -1242,7 +1253,7 @@ mod tests { page_count: 10, byte_size: 40960, created_lsn: 1, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 99_999, }; @@ -1317,7 +1328,7 @@ mod tests { page_count: 100, byte_size: 409_600, created_lsn: 1, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 1, }; @@ -1359,7 +1370,7 @@ mod tests { page_count: 100, byte_size: 409_600, created_lsn: id, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: id, } diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs index 1d64c783..55b61f97 100644 --- a/src/persistence/manifest_sync.rs +++ b/src/persistence/manifest_sync.rs @@ -224,7 +224,7 @@ mod tests { page_count: 1, byte_size: 4096, created_lsn: file_id, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: file_id, } diff --git a/src/persistence/recovery.rs b/src/persistence/recovery.rs index facb17e8..9ccbc2e5 100644 --- a/src/persistence/recovery.rs +++ b/src/persistence/recovery.rs @@ -309,31 +309,52 @@ pub fn recover_shard_v3_pitr( // ONLY place a KvLeaf DataFile is read during recovery now. if manifest_path.exists() { if let Ok(manifest) = ShardManifest::open(&manifest_path) { - let cold_idx = crate::storage::tiered::cold_index::ColdIndex::rebuild_from_manifest( - shard_dir, &manifest, - ); + let per_db = + crate::storage::tiered::cold_index::ColdIndex::rebuild_from_manifest_per_db( + shard_dir, &manifest, + ); // Observability parity with the removed hot-reload loop's // counter: report how many keys the cold tier recovered, NOT // how many were loaded into RAM (zero, by design). - result.kv_heap_entries_loaded = cold_idx.len(); - if cold_idx.len() > 0 { + result.kv_heap_entries_loaded = per_db.iter().map(|(_, ci)| ci.len()).sum(); + // Attach each database's index to ITS database (#139 — spilled + // SELECT >0 keys used to recover into db0's index, unreachable + // from their own db) BEFORE Phase 4 replay. Replayed + // DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane: + // with `cold_index == None`, `remove_counting_cold()`/`clear()` + // are silent no-ops (`as_mut()` on None), so a key deleted in + // the WAL tail resurrects via cold read-through after restart + // whenever the crash lands inside the pre-orphan-sweep window + // (the manifest entry is still Active until the sweep). + for (db_index, cold_idx) in per_db { info!( - "Shard {}: rebuilt cold index with {} entries", + "Shard {}: rebuilt cold index for db {} with {} entries", shard_id, + db_index, cold_idx.len() ); - // Attach to the database BEFORE Phase 4 replay. Replayed - // DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane: - // with `cold_index == None`, `remove_counting_cold()`/`clear()` - // are silent no-ops (`as_mut()` on None), so a key deleted in - // the WAL tail resurrects via cold read-through after restart - // whenever the crash lands inside the pre-orphan-sweep window - // (the manifest entry is still Active until the sweep). - if let Some(db0) = databases.first_mut() { - db0.cold_shard_dir = Some(shard_dir.to_path_buf()); - match db0.cold_index.as_mut() { - Some(existing) => existing.merge(cold_idx), - None => db0.cold_index = Some(cold_idx), + match databases.get_mut(db_index) { + Some(db) => { + db.cold_shard_dir = Some(shard_dir.to_path_buf()); + match db.cold_index.as_mut() { + Some(existing) => existing.merge(cold_idx), + None => db.cold_index = Some(cold_idx), + } + } + None => { + // --databases was reduced below what this manifest + // was written under. The keys are unreachable either + // way (their db no longer exists); refuse to attach + // them to a WRONG db and say so loudly instead of + // silently resurrecting them elsewhere. + tracing::warn!( + shard_id, + db_index, + entries = cold_idx.len(), + "cold recovery: manifest references a logical db beyond the \ + configured --databases count; its spilled keys are NOT attached \ + (unreachable until the server is restarted with enough databases)" + ); } } } @@ -1103,7 +1124,7 @@ mod tests { page_count: 10, byte_size: 655360, created_lsn: 1, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: 1, }); @@ -1116,7 +1137,7 @@ mod tests { page_count: 5, byte_size: 20480, created_lsn: 2, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: 2, }); @@ -1160,7 +1181,7 @@ mod tests { page_count: 1, byte_size: 4096, created_lsn: 1, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: 1, }); diff --git a/src/server/conn/tests.rs b/src/server/conn/tests.rs index 6911e526..d9617b70 100644 --- a/src/server/conn/tests.rs +++ b/src/server/conn/tests.rs @@ -477,6 +477,7 @@ fn test_inline_get_declines_for_cold_key_instead_of_blocking() { 70, b"coldkey", &entry, + 0, &mut manifest, Some(&mut cold_index), ) diff --git a/src/shard/mod.rs b/src/shard/mod.rs index 6868bd4c..8940a65b 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -194,11 +194,12 @@ impl Shard { ); // Initialize cold_index + cold_shard_dir on all databases // so cold_read_through can find keys spilled to NVMe. - // The recovered index itself was already attached to - // databases[0] BEFORE Phase 4 replay (inside - // recover_shard_v3_pitr) so replayed DEL/FLUSH/EXPIRE - // tombstone the cold plane; here we only backfill empty - // indexes on the remaining databases. + // The recovered indexes were already attached to their + // OWN databases (per-file `FileEntry::db_index`, #139) + // BEFORE Phase 4 replay (inside recover_shard_v3_pitr) + // so replayed DEL/FLUSH/EXPIRE tombstone the cold + // plane; here we only backfill empty indexes on the + // databases that recovered nothing. { let cold_dir = shard_dir.clone(); for db in &mut self.databases { diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index b2ae3bfa..92f4161b 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -548,6 +548,8 @@ pub(crate) fn run_eviction_tick( shard_dir: &eviction_shard_dir, manifest, next_file_id, + // Placeholder — `run_eviction` restamps per database (#139). + db_index: 0, }); } super::timers::run_eviction( @@ -902,6 +904,9 @@ pub(crate) fn handle_memory_pressure( shard_dir: &shard_dir, manifest, next_file_id, + // #139: this fallback runs inside the + // per-db loop — attribute to db `i`. + db_index: i, }; // Durable spill (manifest reachable): a // STRING victim stays cold-readable, never a diff --git a/src/shard/timers.rs b/src/shard/timers.rs index b0bf36f6..bd82b5e7 100644 --- a/src/shard/timers.rs +++ b/src/shard/timers.rs @@ -159,6 +159,12 @@ pub(crate) fn run_eviction( for i in 0..db_count { crate::shard::slice::with_shard_db(i, |db| { let before = db.estimated_memory(); + // #139: the shared SpillContext must carry THIS iteration's + // db so every spill file's manifest entry attributes its + // victim to the database it actually came from. + if let Some(ctx) = spill.as_deref_mut() { + ctx.db_index = i; + } let _ = crate::storage::eviction::try_evict_if_needed_with_spill_and_total_budget_reporting( db, &rt, spill.as_deref_mut(), remaining, budget, &mut |key| { @@ -792,6 +798,7 @@ mod tests { shard_dir: &shard_dir, manifest: &mut manifest, next_file_id: &mut next_file_id, + db_index: 0, }; let mut h = NoOpReasonDelHandles::new(); diff --git a/src/storage/db.rs b/src/storage/db.rs index 1ad181e4..473bec6f 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -3203,6 +3203,7 @@ mod tests { 900, key, &entry, + 0, &mut manifest, Some(&mut cold_index), ) diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index a1812373..e40e2405 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -268,6 +268,13 @@ pub struct SpillContext<'a> { pub shard_dir: &'a Path, pub manifest: &'a mut ShardManifest, pub next_file_id: &'a mut u64, + /// Logical database the CURRENT victim belongs to — stamped into each + /// spill file's manifest `FileEntry` so per-db recovery attribution + /// (#139) survives the sync path too. Callers that loop over databases + /// with one shared context (`run_eviction`, the memory-pressure sync + /// fallback) MUST restamp this per iteration; the async batch path + /// carries the equivalent on `SpillRequest.db_index`. + pub db_index: usize, } /// Compute a shard's elastic memory budget from a snapshot of every shard's @@ -1192,6 +1199,7 @@ pub(crate) fn evict_one_with_spill( file_id, key.as_bytes(), entry, + ctx.db_index, ctx.manifest, None, ) { @@ -1816,6 +1824,7 @@ mod tests { shard_dir, manifest: &mut manifest, next_file_id: &mut next_file_id, + db_index: 0, }; let result = try_evict_if_needed_with_spill(&mut db, &config, Some(&mut ctx)); @@ -1837,6 +1846,51 @@ mod tests { assert_eq!(next_file_id, 2); } + /// RED for the #139 follow-up (PR #383 adversarial review, confirmed + /// defect): the SYNC spill path — `run_eviction`'s non-cascade branch → + /// `evict_one_with_spill` → `kv_spill::spill_to_datafile` — hardcoded + /// `db_index: 0` in the manifest `FileEntry`, so an ordinary maxmemory + /// eviction of a db>0 victim produced a spill file that recovery's + /// `rebuild_from_manifest_per_db` re-attached to db 0: exactly the #139 + /// wrong-database corruption, reintroduced through a path the async-spill + /// fix didn't cover. GREEN: `SpillContext.db_index` (restamped per + /// database by the `run_eviction` loop) is threaded through to the + /// manifest entry. + #[test] + fn sync_spill_stamps_victims_database_into_manifest() { + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path(); + let manifest_path = shard_dir.join("shard.manifest"); + let mut manifest = ShardManifest::create(&manifest_path).unwrap(); + let mut next_file_id = 1u64; + + let mut db = Database::new(); + db.set_string( + Bytes::from_static(b"db2_key"), + Bytes::from_static(b"db2_val"), + ); + + let config = make_config(1, "allkeys-lru"); + let mut ctx = SpillContext { + shard_dir, + manifest: &mut manifest, + next_file_id: &mut next_file_id, + db_index: 2, + }; + + let result = try_evict_if_needed_with_spill(&mut db, &config, Some(&mut ctx)); + assert!(result.is_ok()); + assert_eq!(db.len(), 0, "victim must have been evicted"); + + let files = manifest.files(); + assert_eq!(files.len(), 1, "exactly one spill file registered"); + assert_eq!( + files[0].db_index, 2, + "sync spill must attribute the file to the database the victim \ + was evicted from, not db 0 (#139 recovery corruption otherwise)" + ); + } + /// Fail-safe send-before-remove: when the async-spill request channel is /// FULL, eviction must NOT remove the victim from the hot tier (no data /// loss) and must surface OOM rather than pretend success. The live path @@ -2256,6 +2310,7 @@ mod tests { shard_dir, manifest: &mut manifest, next_file_id: &mut next_file_id, + db_index: 0, }; let result = try_evict_if_needed_with_spill(&mut db, &config, Some(&mut ctx)); @@ -2310,6 +2365,7 @@ mod tests { shard_dir, manifest: &mut manifest, next_file_id: &mut next_file_id, + db_index: 0, }; let mut reported: Vec> = Vec::new(); diff --git a/src/storage/tiered/cold_index.rs b/src/storage/tiered/cold_index.rs index 11a565b7..770dcc92 100644 --- a/src/storage/tiered/cold_index.rs +++ b/src/storage/tiered/cold_index.rs @@ -660,14 +660,42 @@ impl ColdIndex { shard_dir: &Path, manifest: &crate::persistence::manifest::ShardManifest, ) -> Self { + let mut merged = Self::new(); + for (_db, index) in Self::rebuild_from_manifest_per_db(shard_dir, manifest) { + merged.merge(index); + } + merged + } + + /// Rebuild one cold index PER LOGICAL DATABASE from the manifest's + /// KvLeaf files (#139). Each spill file is attributed wholesale via + /// `FileEntry::db_index` — the spill path guarantees single-db files + /// (flush chunks cut at db boundaries), and manifests from before the + /// field existed read as db 0, matching their actual (db-blind, + /// attach-to-db0) provenance. Returns `(db_index, index)` pairs for + /// every db that has at least one recovered entry. + pub fn rebuild_from_manifest_per_db( + shard_dir: &Path, + manifest: &crate::persistence::manifest::ShardManifest, + ) -> Vec<(usize, Self)> { use crate::persistence::manifest::FileStatus; use crate::persistence::page::{PAGE_4K, PageType}; - let mut index = Self::new(); + let mut per_db: Vec<(usize, Self)> = Vec::new(); let data_dir = shard_dir.join("data"); for entry in manifest.files() { if entry.status == FileStatus::Active && entry.file_type == PageType::KvLeaf as u8 { + let db = entry.db_index as usize; + let index = match per_db.iter_mut().find(|(d, _)| *d == db) { + Some((_, idx)) => idx, + None => { + per_db.push((db, Self::new())); + #[allow(clippy::unwrap_used)] // pushed on the previous line + let last = per_db.last_mut().unwrap(); + &mut last.1 + } + }; let heap_path = data_dir.join(format!("heap-{:06}.mpf", entry.file_id)); // Read raw bytes and iterate by absolute chunk index. // `read_datafile` skips overflow pages (returns only KvLeaf pages), @@ -699,7 +727,7 @@ impl ColdIndex { } } } - index + per_db } } diff --git a/src/storage/tiered/cold_read.rs b/src/storage/tiered/cold_read.rs index c9ebc9a7..03235264 100644 --- a/src/storage/tiered/cold_read.rs +++ b/src/storage/tiered/cold_read.rs @@ -265,6 +265,7 @@ mod tests { 20, b"myhash", &entry, + 0, &mut manifest, Some(&mut cold_index), ) @@ -314,6 +315,7 @@ mod tests { 21, b"cachekey", &entry, + 0, &mut manifest, Some(&mut cold_index), ) @@ -394,6 +396,7 @@ mod tests { 40, key, &entry, + 0, &mut manifest, Some(&mut cold_index), ) @@ -586,6 +589,7 @@ mod tests { 30, b"big_key", &entry, + 0, &mut manifest, Some(&mut cold_index), ) diff --git a/src/storage/tiered/cold_read_pool.rs b/src/storage/tiered/cold_read_pool.rs index 640a2a83..6b591243 100644 --- a/src/storage/tiered/cold_read_pool.rs +++ b/src/storage/tiered/cold_read_pool.rs @@ -187,6 +187,7 @@ mod tests { 50, key, &entry, + 0, &mut manifest, Some(&mut cold_index), ) diff --git a/src/storage/tiered/kv_spill.rs b/src/storage/tiered/kv_spill.rs index e92ed1a8..dcb6ed12 100644 --- a/src/storage/tiered/kv_spill.rs +++ b/src/storage/tiered/kv_spill.rs @@ -141,11 +141,17 @@ pub fn value_type_of(val: &RedisValueRef) -> ValueType { /// entries require overflow pages, also future work). /// /// Returns `Ok(())` on success, skip, or best-effort failure logging. +/// +/// `db_index` is the logical database the entry was evicted FROM — it is +/// stamped into the manifest `FileEntry` so `rebuild_from_manifest_per_db` +/// re-attaches the key to the right database after a crash (#139). Passing 0 +/// for a db>0 victim silently corrupts recovery attribution. pub fn spill_to_datafile( shard_dir: &Path, file_id: u64, key: &[u8], entry: &Entry, + db_index: usize, manifest: &mut ShardManifest, cold_index: Option<&mut super::cold_index::ColdIndex>, ) -> io::Result<()> { @@ -193,7 +199,7 @@ pub fn spill_to_datafile( page_count: pages.total_pages, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: db_index as u64, max_key_hash: 0, last_modified_lsn: 0, }); @@ -595,7 +601,7 @@ mod tests { let mut manifest = ShardManifest::create(&manifest_path).unwrap(); let entry = Entry::new_string(Bytes::from_static(b"hello world")); - spill_to_datafile(shard_dir, 1, b"mykey", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 1, b"mykey", &entry, 0, &mut manifest, None).unwrap(); // Verify file was created let file_path = shard_dir.join("data/heap-000001.mpf"); @@ -625,7 +631,7 @@ mod tests { // Registered spill: file 1 in manifest, must survive the sweep. let entry = Entry::new_string(Bytes::from_static(b"registered")); - spill_to_datafile(shard_dir, 1, b"livekey", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 1, b"livekey", &entry, 0, &mut manifest, None).unwrap(); // Crash orphan: heap file on disk but never registered in the manifest // (spill wrote the file, crash before manifest commit). @@ -674,7 +680,7 @@ mod tests { // Registered spill: must be classified as NOT orphaned. let entry = Entry::new_string(Bytes::from_static(b"registered")); - spill_to_datafile(shard_dir, 1, b"livekey", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 1, b"livekey", &entry, 0, &mut manifest, None).unwrap(); let data_dir = shard_dir.join("data"); // A larger batch of unregistered files, standing in for a @@ -740,7 +746,7 @@ mod tests { let future_ms = current_time_ms() + 60_000; entry.set_expires_at_ms(0, future_ms); - spill_to_datafile(shard_dir, 2, b"ttl_key", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 2, b"ttl_key", &entry, 0, &mut manifest, None).unwrap(); let file_path = shard_dir.join("data/heap-000002.mpf"); let pages = read_datafile(&file_path).unwrap(); @@ -774,7 +780,7 @@ mod tests { } let entry = Entry::new_string(Bytes::from(big_value)); - spill_to_datafile(shard_dir, 3, b"big_key", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 3, b"big_key", &entry, 0, &mut manifest, None).unwrap(); // File SHOULD now exist with overflow pages let file_path = shard_dir.join("data/heap-000003.mpf"); @@ -817,7 +823,7 @@ mod tests { let mut entry = Entry::new_string(Bytes::new()); entry.value = CompactValue::from_redis_value(RedisValue::Hash(map)); - spill_to_datafile(shard_dir, 10, b"hash_key", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 10, b"hash_key", &entry, 0, &mut manifest, None).unwrap(); let file_path = shard_dir.join("data/heap-000010.mpf"); assert!(file_path.exists(), "DataFile should exist for hash entry"); @@ -863,7 +869,7 @@ mod tests { let mut entry = Entry::new_string(Bytes::new()); entry.value = CompactValue::from_redis_value(RedisValue::List(list)); - spill_to_datafile(shard_dir, 11, b"list_key", &entry, &mut manifest, None).unwrap(); + spill_to_datafile(shard_dir, 11, b"list_key", &entry, 0, &mut manifest, None).unwrap(); let file_path = shard_dir.join("data/heap-000011.mpf"); assert!(file_path.exists(), "DataFile should exist for list entry"); @@ -913,6 +919,7 @@ mod tests { 50, b"overflow_key", &entry, + 0, &mut manifest, Some(&mut cold_index), ) @@ -1135,6 +1142,77 @@ mod tests { } } + /// #139 recovery attribution: two spill files tagged with different + /// `FileEntry::db_index` values must rebuild into SEPARATE per-db cold + /// indexes, each holding exactly its own file's keys — the db0-only + /// attach used to make every SELECT >0 spilled key unreachable after + /// restart. + #[test] + fn test_rebuild_per_db_attributes_files_to_their_databases() { + use crate::persistence::manifest::{FileEntry, FileStatus, ShardManifest, StorageTier}; + use crate::persistence::page::PageType; + use crate::storage::tiered::cold_index::ColdIndex; + + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path(); + let manifest_path = shard_dir.join("shard.manifest"); + let mut manifest = ShardManifest::create(&manifest_path).unwrap(); + + // One batch file per db, registered exactly as apply_spill_completions + // would (db attribution at file granularity). + for (db, file_id, prefix) in [(0u64, 201u64, "d0:"), (3u64, 202u64, "d3:")] { + let entries: Vec = (0..10) + .map(|i| SpillEntry { + key: Bytes::from(format!("{prefix}{i}")), + value_bytes: Bytes::from(format!("v{i}")), + value_type: ValueType::String, + flags: 0, + ttl_ms: None, + }) + .collect(); + let batch = build_kv_spill_batch(&entries, file_id).unwrap(); + let byte_size = write_kv_spill_batch(shard_dir, file_id, &batch).unwrap(); + manifest.add_file(FileEntry { + file_id, + file_type: PageType::KvLeaf as u8, + status: FileStatus::Active, + tier: StorageTier::Hot, + page_size_log2: 12, + page_count: batch.pages.len() as u32, + byte_size, + created_lsn: 0, + db_index: db, + max_key_hash: 0, + last_modified_lsn: 0, + }); + } + manifest.commit().unwrap(); + + let mut per_db = ColdIndex::rebuild_from_manifest_per_db(shard_dir, &manifest); + per_db.sort_by_key(|(db, _)| *db); + let dbs: Vec = per_db.iter().map(|(db, _)| *db).collect(); + assert_eq!(dbs, vec![0, 3], "one index per db present in the manifest"); + for (db, index) in &per_db { + assert_eq!(index.len(), 10, "db {db}: every entry recovered"); + let prefix = if *db == 0 { "d0:" } else { "d3:" }; + for i in 0..10 { + assert!( + index.lookup(format!("{prefix}{i}").as_bytes()).is_some(), + "db {db}: missing its own key {prefix}{i}" + ); + } + let other = if *db == 0 { "d3:0" } else { "d0:0" }; + assert!( + index.lookup(other.as_bytes()).is_none(), + "db {db}: must not contain the other db's keys" + ); + } + + // The merged wrapper still sees everything (single-db callers). + let merged = ColdIndex::rebuild_from_manifest(shard_dir, &manifest); + assert_eq!(merged.len(), 20); + } + /// RECOVERY-PATH test: prove `ColdIndex::rebuild_from_manifest` reconstructs /// the SAME (page_idx, slot_idx) mapping the builder produced. /// @@ -1181,7 +1259,7 @@ mod tests { page_count: batch.pages.len() as u32, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 0, }); @@ -1397,7 +1475,7 @@ mod tests { page_count: batch.pages.len() as u32, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 0, }); @@ -1476,7 +1554,7 @@ mod tests { page_count: batch.pages.len() as u32, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 0, }); diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index 95252b49..19a9469b 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -224,7 +224,7 @@ pub struct SpillCompletion { } /// Build a `FileEntry` skeleton for a spill file (fields not tracked by Moon are zero). -fn make_file_entry(file_id: u64, page_count: u32, byte_size: u64) -> FileEntry { +fn make_file_entry(file_id: u64, page_count: u32, byte_size: u64, db_index: usize) -> FileEntry { FileEntry { file_id, file_type: PageType::KvLeaf as u8, @@ -234,7 +234,7 @@ fn make_file_entry(file_id: u64, page_count: u32, byte_size: u64) -> FileEntry { page_count, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: db_index as u64, max_key_hash: 0, last_modified_lsn: 0, } @@ -290,9 +290,16 @@ pub(crate) fn flush_buffer(buffer: &mut Vec) -> Vec BATCH_BYTES_CAP { + if chunk_bytes + next_len > BATCH_BYTES_CAP || buffer[end].db_index != chunk_db { break; } chunk_bytes += next_len; @@ -335,7 +342,7 @@ pub(crate) fn flush_buffer(buffer: &mut Vec) -> Vec SpillCompletion { ) { Ok(pages) => match write_kv_spill_pages(&req.shard_dir, file_id, &pages) { Ok(byte_size) => SpillCompletion { - file_entry: make_file_entry(file_id, pages.total_pages, byte_size), + file_entry: make_file_entry(file_id, pages.total_pages, byte_size, req.db_index), entries: vec![SpillCompletionEntry { key: req.key.clone(), db_index: req.db_index, @@ -412,7 +419,7 @@ fn spill_single_entry(req: &SpillRequest, file_id: u64) -> SpillCompletion { "spill_thread: single-file write failed" ); SpillCompletion { - file_entry: make_file_entry(file_id, 0, 0), + file_entry: make_file_entry(file_id, 0, 0, req.db_index), entries: Vec::new(), success: false, } @@ -426,7 +433,7 @@ fn spill_single_entry(req: &SpillRequest, file_id: u64) -> SpillCompletion { "spill_thread: single-file build failed (key too large)" ); SpillCompletion { - file_entry: make_file_entry(file_id, 0, 0), + file_entry: make_file_entry(file_id, 0, 0, req.db_index), entries: Vec::new(), success: false, } @@ -873,7 +880,7 @@ mod tests { let (tx, rx) = flume::bounded::(1); let stop = Arc::new(AtomicBool::new(false)); let dummy = || SpillCompletion { - file_entry: make_file_entry(7, 1, PAGE_4K as u64), + file_entry: make_file_entry(7, 1, PAGE_4K as u64, 0), entries: Vec::new(), success: true, }; @@ -943,6 +950,50 @@ mod tests { ); } + /// Flush chunks must never cross a logical-DB boundary (#139): each + /// spill file is attributed wholesale to ONE `FileEntry::db_index`, so + /// a buffer interleaving dbs must produce one file per contiguous + /// same-db run, each completion carrying its db and only its keys. + #[test] + fn flush_buffer_cuts_chunks_at_db_boundaries() { + let tmp = tempfile::tempdir().unwrap(); + let mk = |key: &str, db: usize, file_id: u64| SpillRequest { + key: Bytes::from(key.to_owned()), + db_index: db, + value_bytes: Bytes::from_static(b"v"), + value_type: ValueType::String, + flags: 0, + ttl_ms: None, + file_id, + shard_dir: tmp.path().to_path_buf(), + }; + let mut buffer = vec![ + mk("a0", 0, 1), + mk("a1", 0, 2), + mk("b0", 1, 3), + mk("c0", 0, 4), + ]; + let completions = flush_buffer(&mut buffer); + assert!(completions.iter().all(|c| c.success)); + let summary: Vec<(u64, usize)> = completions + .iter() + .map(|c| (c.file_entry.db_index, c.entries.len())) + .collect(); + assert_eq!( + summary, + vec![(0, 2), (1, 1), (0, 1)], + "expected one file per contiguous same-db run: {summary:?}" + ); + for c in &completions { + for e in &c.entries { + assert_eq!( + e.db_index as u64, c.file_entry.db_index, + "entry db must match its file's manifest attribution" + ); + } + } + } + /// An entry that passes the `INLINE_MAX_VALUE_BYTES` pre-screen but does not /// fit a fresh inline leaf (large key + incompressible value) makes /// `build_kv_spill_batch` fail. That must NOT fail the whole inline flush — diff --git a/src/storage/tiered/warm_tier.rs b/src/storage/tiered/warm_tier.rs index bd7a655d..c9e6ec0c 100644 --- a/src/storage/tiered/warm_tier.rs +++ b/src/storage/tiered/warm_tier.rs @@ -94,7 +94,7 @@ pub fn transition_to_warm( page_count: codes_pages as u32, byte_size: codes_data.len() as u64, created_lsn: 0, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: 0, }; diff --git a/tests/cold_orphan_sweep.rs b/tests/cold_orphan_sweep.rs index 6599ed59..4631bd75 100644 --- a/tests/cold_orphan_sweep.rs +++ b/tests/cold_orphan_sweep.rs @@ -31,6 +31,7 @@ fn test_orphan_sweep_removes_hot_shadowed_key() { 1, b"k1", &make_string_entry("cold_value"), + 0, &mut manifest, Some(&mut cold_index), ) @@ -82,6 +83,7 @@ fn test_orphan_sweep_preserves_cold_only_key() { 2, b"k2", &make_string_entry("cold_only_value"), + 0, &mut manifest, Some(&mut cold_index), ) @@ -123,6 +125,7 @@ fn test_orphan_sweep_mixed_hot_and_cold_only() { 10, b"k1", &make_string_entry("v1_cold"), + 0, &mut manifest, Some(&mut cold_index), ) @@ -132,6 +135,7 @@ fn test_orphan_sweep_mixed_hot_and_cold_only() { 11, b"k2", &make_string_entry("v2_cold"), + 0, &mut manifest, Some(&mut cold_index), ) @@ -186,6 +190,7 @@ fn test_orphan_sweep_manifests_tombstone() { 20, b"kx", &make_string_entry("val"), + 0, &mut manifest, Some(&mut cold_index), ) diff --git a/tests/crash_matrix_cross_plane/harness.rs b/tests/crash_matrix_cross_plane/harness.rs index 122b1733..b6985e3a 100644 --- a/tests/crash_matrix_cross_plane/harness.rs +++ b/tests/crash_matrix_cross_plane/harness.rs @@ -165,14 +165,13 @@ impl Drop for ServerGuard { .args(["-fl"]) .arg(&self.dir_marker) .output() + && !out.stdout.is_empty() { - if !out.stdout.is_empty() { - eprintln!( - "[ServerGuard] pkill backstop for marker {:?} will hit:\n{}", - self.dir_marker, - String::from_utf8_lossy(&out.stdout).trim_end() - ); - } + eprintln!( + "[ServerGuard] pkill backstop for marker {:?} will hit:\n{}", + self.dir_marker, + String::from_utf8_lossy(&out.stdout).trim_end() + ); } let _ = Command::new("pkill") .args(["-9", "-f"]) diff --git a/tests/crash_recovery_cold_multidb.rs b/tests/crash_recovery_cold_multidb.rs new file mode 100644 index 00000000..3822780d --- /dev/null +++ b/tests/crash_recovery_cold_multidb.rs @@ -0,0 +1,315 @@ +//! #139: cold recovery must re-attach spilled keys to the logical database +//! they were evicted from — not unconditionally db 0. +//! +//! Before the fix, `recover_shard_v3`'s cold-index rebuild merged every +//! recovered entry into `databases[0]`: a key spilled from `SELECT 1` was +//! unreachable from db 1 after restart (cold-read miss), and — because this +//! test deliberately uses the SAME key names in db 0 and db 1 — db 0's +//! index would hold whichever file's entry was inserted last, i.e. db 0 +//! could serve db 1's VALUE. The fix attributes each spill file wholesale +//! via `FileEntry::db_index` (spill flush chunks never cross a db +//! boundary), and recovery attaches each rebuilt index to its own db. +//! +//! Setup mirrors `crash_recovery_disk_offload_no_aof`: offload WITHOUT AOF, +//! so the cold tier is the probes' ONLY durable copy and misattribution is +//! visible immediately (nothing replays the keys back into their dbs). +//! Probes land in db 0 AND db 1 under identical names with db-distinct +//! values; LRU filler in db 0 evicts them to the cold tier; SIGKILL; +//! restart; every readable probe must return ITS OWN db's value — any +//! cross-db value is an instant failure, and db 1 must recover at least +//! half its probes (async-tick durability slack, same convention as the +//! no-AOF suite). +//! +//! Run: +//! cargo build --release +//! cargo test --release --test crash_recovery_cold_multidb -- --ignored + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command}; +use std::time::Duration; + +const PROBE_COUNT: usize = 60; +const PROBE_VALUE_LEN: usize = 2048; +const FILLER_COUNT: usize = 3000; +const FILLER_VALUE_LEN: usize = 2048; +/// 4 MiB, 1 shard: probes (2 dbs x 60 x 2 KiB = 240 KiB) + filler (~6 MiB) +/// blow well past the 0.85 spill threshold. +const MAXMEMORY_BYTES: usize = 4 * 1024 * 1024; +/// Async spill/manifest ticks must commit the probes before the SIGKILL +/// (same rationale as crash_recovery_disk_offload_no_aof's settle). +const SETTLE_BEFORE_KILL: u64 = 8; + +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-cold-multidb-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +fn start_moon(dir: &std::path::Path, maxmemory: usize) -> (Child, u16) { + let off_dir = dir.join("off"); + std::fs::create_dir_all(&off_dir).expect("create off dir"); + common::spawn_listening(|port| { + Command::new(common::find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--maxmemory", + &maxmemory.to_string(), + "--maxmemory-policy", + "allkeys-lru", + "--disk-offload", + "enable", + "--disk-offload-dir", + off_dir.to_str().expect("off dir utf8"), + "--appendonly", + "no", + "--cold-orphan-sweep-interval-secs", + "60", + "--dir", + ]) + .arg(dir) + .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log")) + .spawn() + .expect("spawn moon (run `cargo build --release` first)") + }) +} + +// Minimal RESP2 client (self-contained per-file convention; copied from +// tests/crash_recovery_spill_batch_kill9.rs). +struct Conn { + s: TcpStream, + buf: Vec, + pos: usize, +} + +impl Conn { + fn open(port: u16) -> Self { + let s = TcpStream::connect(format!("127.0.0.1:{port}")).expect("connect"); + s.set_read_timeout(Some(Duration::from_secs(30))).ok(); + s.set_write_timeout(Some(Duration::from_secs(30))).ok(); + Conn { + s, + buf: Vec::with_capacity(32 * 1024), + pos: 0, + } + } + + fn cmd(&mut self, parts: &[&[u8]]) -> Option> { + let mut req = Vec::with_capacity(64 + parts.iter().map(|p| p.len() + 16).sum::()); + req.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + req.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + req.extend_from_slice(p); + req.extend_from_slice(b"\r\n"); + } + self.s.write_all(&req).expect("write cmd"); + self.frame() + } + + fn fill(&mut self) { + let mut chunk = [0u8; 32 * 1024]; + let n = self.s.read(&mut chunk).expect("read from server"); + assert!(n > 0, "connection closed mid-frame"); + self.buf.extend_from_slice(&chunk[..n]); + } + + fn line(&mut self) -> String { + loop { + if let Some(rel) = self.buf[self.pos..].windows(2).position(|w| w == b"\r\n") { + let line = + String::from_utf8_lossy(&self.buf[self.pos..self.pos + rel]).into_owned(); + self.pos += rel + 2; + return line; + } + self.fill(); + } + } + + fn exact(&mut self, n: usize) -> Vec { + while self.buf.len() - self.pos < n + 2 { + self.fill(); + } + let out = self.buf[self.pos..self.pos + n].to_vec(); + self.pos += n + 2; + out + } + + /// Returns the payload for bulk replies, Some(status bytes) for simple + /// strings / integers, None for nil. Panics on protocol errors. + fn frame(&mut self) -> Option> { + if self.pos > 0 && self.pos == self.buf.len() { + self.buf.clear(); + self.pos = 0; + } + let line = self.line(); + let (tag, rest) = line.split_at(1); + match tag { + "+" | ":" => Some(rest.as_bytes().to_vec()), + "-" => panic!("server error reply: {rest}"), + "$" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + None + } else { + Some(self.exact(n as usize)) + } + } + other => panic!("unexpected RESP tag {other:?} in line {line:?}"), + } + } +} + +/// RAII guard: SIGKILLs the server process when dropped (safe on an +/// already-dead child). Same shape as crash_recovery_spill_batch_kill9. +struct ServerGuard(Child); + +impl ServerGuard { + fn sigkill(&mut self) { + common::sigkill(&mut self.0); + } +} + +impl Drop for ServerGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn probe_key(i: usize) -> String { + format!("probe:{i:04}") +} + +fn probe_value(db: usize, i: usize) -> Vec { + let mut v = format!("db{db}:{i:04}:").into_bytes(); + v.resize(PROBE_VALUE_LEN, b'p'); + v +} + +fn count_spill_files(dir: &std::path::Path) -> usize { + let mut n = 0; + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + let Ok(rd) = std::fs::read_dir(&d) else { + continue; + }; + for e in rd.flatten() { + let p = e.path(); + if p.is_dir() { + stack.push(p); + } else if p + .file_name() + .and_then(|f| f.to_str()) + .is_some_and(|f| f.starts_with("heap-") && f.ends_with(".mpf")) + { + n += 1; + } + } + } + n +} + +#[test] +#[ignore = "real-server kill-9 test; needs a release moon binary (see file header)"] +fn cold_recovery_attributes_spilled_keys_to_their_database() { + let dir = unique_dir("attrib"); + std::fs::create_dir_all(&dir).expect("create test dir"); + + // Round 1: write probes into db0 AND db1 (same names, db-distinct + // values), then LRU-filler to evict them into the cold tier. + let (child, port) = start_moon(&dir, MAXMEMORY_BYTES); + let mut guard = ServerGuard(child); + { + let mut c = Conn::open(port); + for db in [0usize, 1usize] { + assert_eq!( + c.cmd(&[b"SELECT", db.to_string().as_bytes()]).as_deref(), + Some(b"OK".as_ref()) + ); + for i in 0..PROBE_COUNT { + let k = probe_key(i); + let v = probe_value(db, i); + assert_eq!( + c.cmd(&[b"SET", k.as_bytes(), &v]).as_deref(), + Some(b"OK".as_ref()), + "round-1 SET db{db} {k}" + ); + } + } + // Filler in db0 pushes the (older) probes out through the LRU + + // spill path. Values match probe size so victim selection is + // size-uniform. + assert_eq!(c.cmd(&[b"SELECT", b"0"]).as_deref(), Some(b"OK".as_ref())); + let filler_v = vec![b'f'; FILLER_VALUE_LEN]; + for i in 0..FILLER_COUNT { + let k = format!("filler:{i:06}"); + // Under allkeys-lru SET never OOMs; it evicts. + let _ = c.cmd(&[b"SET", k.as_bytes(), &filler_v]); + } + } + std::thread::sleep(Duration::from_secs(SETTLE_BEFORE_KILL)); + + let spill_files = count_spill_files(&dir); + assert!( + spill_files > 0, + "test premise broken: no spill files were written — eviction/offload \ + never engaged (check maxmemory/policy flags)" + ); + guard.sigkill(); + + // Round 2: generous maxmemory so recovery/promotion doesn't re-evict + // mid-assertion (same convention as crash_recovery_spill_batch_kill9). + let (child2, port2) = start_moon(&dir, 64 * 1024 * 1024); + let mut guard2 = ServerGuard(child2); + let mut c = Conn::open(port2); + + let mut db1_recovered = 0usize; + for db in [0usize, 1usize] { + assert_eq!( + c.cmd(&[b"SELECT", db.to_string().as_bytes()]).as_deref(), + Some(b"OK".as_ref()) + ); + for i in 0..PROBE_COUNT { + let k = probe_key(i); + match c.cmd(&[b"GET", k.as_bytes()]) { + None => {} // not durable by kill time — tolerated (bounded below) + Some(v) => { + let expect = probe_value(db, i); + // THE #139 assertion: a readable probe must carry ITS + // OWN db's value. Pre-fix, db0 could serve db1's value + // (same key name, misattributed cold entry) and db1 + // answered nil for everything. + assert_eq!( + v, expect, + "db{db} {k}: recovered value belongs to the wrong database \ + (cold-plane db attribution broken)" + ); + if db == 1 { + db1_recovered += 1; + } + } + } + } + } + assert!( + db1_recovered >= PROBE_COUNT / 2, + "db1 recovered only {db1_recovered}/{PROBE_COUNT} spilled probes — \ + SELECT>0 cold entries are not being re-attached (#139)" + ); + + guard2.sigkill(); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/tests/crash_recovery_spill_batch_kill9.rs b/tests/crash_recovery_spill_batch_kill9.rs index 104d3370..f35c5fd2 100644 --- a/tests/crash_recovery_spill_batch_kill9.rs +++ b/tests/crash_recovery_spill_batch_kill9.rs @@ -488,7 +488,7 @@ fn spill_batch_shared_file_survives_cold_read_after_kill9() { page_count: batch.pages.len() as u32, byte_size, created_lsn: 0, - min_key_hash: 0, + db_index: 0, max_key_hash: 0, last_modified_lsn: 0, }); diff --git a/tests/crash_recovery_vacuum.rs b/tests/crash_recovery_vacuum.rs index 66d34d7b..77da5926 100644 --- a/tests/crash_recovery_vacuum.rs +++ b/tests/crash_recovery_vacuum.rs @@ -70,7 +70,7 @@ fn crash_manifest_gc_before_commit_recovers_pre_staging_root() { page_count: 100, byte_size: 409_600, created_lsn: id, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: id, }; diff --git a/tests/manifest_gc_tombstones.rs b/tests/manifest_gc_tombstones.rs index 2fc9e7a8..2e65c82b 100644 --- a/tests/manifest_gc_tombstones.rs +++ b/tests/manifest_gc_tombstones.rs @@ -25,7 +25,7 @@ fn make_entry(id: u64) -> FileEntry { page_count: 100, byte_size: 409_600, created_lsn: id, - min_key_hash: 0, + db_index: 0, max_key_hash: u64::MAX, last_modified_lsn: id, }