fix(persistence): cold recovery re-attaches spilled keys to their own database (#139)

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

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,
         }

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,
         }

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,
         });

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 {

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
     }
 }