Make legacy storage migration crash recoverable

Kiwi could open unclassified Base-v1 and Vector-v1 layouts through live RocksDB creation paths, leaving missing Column Families, partial manifest writes, and interrupted directory switches ambiguous after a restart.

This decision adds a durable Root-manifest migration journal, strict source-profile classification, shadow-copy verification, per-instance atomic promotion, production-open confirmation, and a verified pre-admission rollback path. Resume validation now accepts only the one-step filesystem progress that can occur between a durable rename and its journal update.

Constraint: Rollback is available only before RollbackWindowClosed; after admission, startup validates the live v2 topology without requiring the intentionally stale backup to remain byte-identical.

Rejected: Upgrade a live RocksDB instance in place | A crash between Column Family creation, manifest persistence, and multi-instance switching cannot prove which copy is authoritative.

Confidence: high

Scope-risk: broad

Directive: Keep create_missing_column_families disabled outside fresh, manifest-authorized instance creation.

Tested: WSL 32-test migration and fault-injection matrix; WSL Storage and Server Clippy with -D warnings; Windows Storage checks with and without test-fault-injection; Windows Server check; cargo fmt --check; git diff --check.

Not-tested: Snapshot restore and mixed-version cluster behavior remain Task 3 and later; exact historical executable compatibility remains Task 7.

Related: #421

Co-authored-by: OmX <omx@oh-my-codex.dev>
Signed-off-by: Xin.Zh <alexstocks@foxmail.com>

Author: AlexStocks

src/server/src/main.rs

@@ -254,9 +254,36 @@ async fn initialize_storage(config: &Config) -> Result<GlobalStorage, DualRuntim
     let mut storage = Storage::new(config.db_instance_num, 0);
 
     info!("Opening storage at path: {:?}", data_dir);
-    let bg_task_receiver = storage
-        .open(storage_options, &data_dir)
-        .map_err(|e| DualRuntimeError::storage_runtime(format!("Failed to open storage: {}", e)))?;
+    let mut bg_task_receiver = match storage.open(Arc::clone(&storage_options), &data_dir) {
+        Ok(receiver) => receiver,
+        Err(error) => {
+            let rollback = storage::recover_or_rollback_before_admission(
+                &data_dir,
+                config.db_instance_num,
+                &storage_options,
+            );
+            let rollback_context = match rollback {
+                Ok(true) => "; verified legacy backup restored before admission".to_string(),
+                Ok(false) => String::new(),
+                Err(rollback_error) => format!("; rollback also failed: {rollback_error}"),
+            };
+            return Err(DualRuntimeError::storage_runtime(format!(
+                "Failed to open storage: {error}{rollback_context}"
+            )));
+        }
+    };
+    if storage::close_rollback_window(&data_dir).map_err(|error| {
+        DualRuntimeError::storage_runtime(format!("Failed to close rollback window: {error}"))
+    })? {
+        drop(bg_task_receiver);
+        bg_task_receiver = storage
+            .reopen(storage_options, &data_dir)
+            .map_err(|error| {
+                DualRuntimeError::storage_runtime(format!(
+                    "Failed to reopen storage after closing rollback window: {error}"
+                ))
+            })?;
+    }
     info!("Storage opened successfully");
 
     tokio::spawn(async move {

src/storage/src/lib.rs

@@ -22,6 +22,7 @@ mod format_member_data_key;
 pub mod format_vector;
 pub mod format_vector_member_key;
 mod storage_manifest;
+mod storage_migration;
 pub mod vector;
 pub mod vector_fault;
 mod vector_flat;
@@ -89,6 +90,8 @@ pub use format_base_key::BaseMetaKey;
 pub use format_base_value::*;
 pub use format_zset_score_key::{ScoreMember, ZsetScoreMember};
 pub use options::StorageOptions;
+#[cfg(any(test, feature = "test-fault-injection"))]
+pub use redis::fail_next_redis_open;
 pub use redis::{GenerationProvider, Redis, TypeCheckState};
 pub use redis_vectors::VectorDataSample;
 pub use statistics::KeyStatistics;
@@ -102,6 +105,13 @@ pub use storage_manifest::{
     ROOT_STORAGE_MANIFEST_VERSION, RootStorageManifestV2, SLOT_MAPPING_VERSION,
     STORAGE_MANIFEST_FILE, STORAGE_SCHEMA_VERSION_V2, slot_mapping_digest,
 };
+#[cfg(any(test, feature = "test-fault-injection"))]
+pub use storage_migration::fail_next_storage_migration;
+pub use storage_migration::{
+    MigrationFaultPoint, MigrationLayout, classify_storage_root, close_rollback_window,
+    finalize_migration_after_storage_open, prepare_or_resume_migration,
+    recover_or_rollback_before_admission,
+};
 pub use storage_schema::{
     CANONICAL_COLUMN_FAMILIES, CANONICAL_COLUMN_FAMILY_NAMES, ColumnFamilyIndex, ColumnFamilyRole,
     ColumnFamilySpec, ComparatorId, canonical_column_family_names,

src/storage/src/options.rs

@@ -90,7 +90,7 @@ impl Default for StorageOptions {
         let block_cache_size = 8 << 30; // 8GB
         let mut options = Options::default();
         options.create_if_missing(true);
-        options.create_missing_column_families(true);
+        options.create_missing_column_families(false);
         options.set_max_open_files(10000);
         options.set_write_buffer_size(64 << 20); // 64MB
         options.set_max_write_buffer_number(3);
@@ -124,7 +124,8 @@ impl StorageOptions {
 
     /// Build StorageOptions from a loaded [`conf::config::Config`].
     pub fn from_config(config: &conf::config::Config) -> Self {
-        let rocksdb_opts = config.get_rocksdb_options();
+        let mut rocksdb_opts = config.get_rocksdb_options();
+        rocksdb_opts.create_missing_column_families(false);
         // Build the shared block cache once when sharing is enabled and a
         // memory budget is configured. Every `Redis` instance receives the
         // same `Arc<StorageOptions>` and therefore reuses this single cache.

src/storage/src/redis.rs

@@ -16,8 +16,14 @@
 // limitations under the License.
 
 use std::collections::HashMap;
+#[cfg(any(test, feature = "test-fault-injection"))]
+use std::collections::HashSet;
 use std::ops::Deref;
 use std::path::Path;
+#[cfg(any(test, feature = "test-fault-injection"))]
+use std::path::PathBuf;
+#[cfg(any(test, feature = "test-fault-injection"))]
+use std::sync::LazyLock;
 use std::sync::Mutex;
 use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
@@ -65,6 +71,63 @@ use crate::storage_schema::{CANONICAL_COLUMN_FAMILIES, ColumnFamilySpec, Compara
 /// log index that created the key (wired up by the raft layer later).
 pub type GenerationProvider = Arc<dyn Fn() -> Result<u64> + Send + Sync>;
 
+#[cfg(any(test, feature = "test-fault-injection"))]
+static REDIS_OPEN_FAILURES: LazyLock<Mutex<HashSet<PathBuf>>> =
+    LazyLock::new(|| Mutex::new(HashSet::new()));
+
+#[cfg(any(test, feature = "test-fault-injection"))]
+#[doc(hidden)]
+pub struct RedisOpenFailureGuard {
+    db_path: PathBuf,
+}
+
+#[cfg(any(test, feature = "test-fault-injection"))]
+impl Drop for RedisOpenFailureGuard {
+    fn drop(&mut self) {
+        if let Ok(mut failures) = REDIS_OPEN_FAILURES.lock() {
+            failures.remove(&self.db_path);
+        }
+    }
+}
+
+#[cfg(any(test, feature = "test-fault-injection"))]
+#[doc(hidden)]
+#[must_use]
+pub fn fail_next_redis_open(db_path: &Path) -> RedisOpenFailureGuard {
+    let db_path = db_path.to_path_buf();
+    let inserted = REDIS_OPEN_FAILURES
+        .lock()
+        .map(|mut failures| failures.insert(db_path.clone()))
+        .unwrap_or(false);
+    assert!(
+        inserted,
+        "Redis open failure already registered for {}",
+        db_path.display()
+    );
+    RedisOpenFailureGuard { db_path }
+}
+
+fn maybe_fail_redis_open(db_path: &Path) -> Result<()> {
+    #[cfg(any(test, feature = "test-fault-injection"))]
+    {
+        let should_fail = REDIS_OPEN_FAILURES
+            .lock()
+            .map(|mut failures| failures.remove(db_path))
+            .unwrap_or(false);
+        if should_fail {
+            return Err(InvalidFormatSnafu {
+                message: format!(
+                    "injected Redis open failure after RocksDB open: {}",
+                    db_path.display()
+                ),
+            }
+            .build());
+        }
+    }
+    let _ = db_path;
+    Ok(())
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum TypeCheckState {
     Missing,
@@ -439,6 +502,8 @@ impl Redis {
         let db_directory = Path::new(db_path);
         let manifest_path = db_directory.join(crate::storage_manifest::STORAGE_MANIFEST_FILE);
         let current_path = db_directory.join("CURRENT");
+        db_opts.create_if_missing(allow_manifest_creation && !current_path.exists());
+        db_opts.create_missing_column_families(allow_manifest_creation && !current_path.exists());
         if !manifest_path.exists() && current_path.exists() {
             return Err(InvalidFormatSnafu {
                 message: format!(
@@ -502,6 +567,7 @@ impl Redis {
         let db = RocksDbOwner::new(
             DB::open_cf_descriptors(&db_opts, db_path, column_families).context(RocksSnafu)?,
         );
+        maybe_fail_redis_open(db_directory)?;
         let _ = db_once_cell.set(db.downgrade());
 
         let handles = CANONICAL_COLUMN_FAMILIES

src/storage/src/storage.rs

@@ -30,7 +30,13 @@ use crate::expiration_manager::ExpirationManager;
 use crate::format_base_value::DataType;
 use crate::options::OptionType;
 use crate::slot_indexer::SlotIndexer;
-use crate::storage_manifest::{load_or_create_root_manifest, validate_existing_instance_manifests};
+use crate::storage_manifest::{
+    MigrationPhase, RootStorageManifestV2, load_or_create_root_manifest,
+    validate_existing_instance_manifests,
+};
+use crate::storage_migration::{
+    finalize_migration_after_storage_open, prepare_or_resume_migration,
+};
 use crate::storage_scan::{SCAN_CURSOR_STATE_CAPACITY, ScanCursorState};
 use crate::{ColumnFamilyIndex, Redis, StorageOptions, data_type_to_tag};
 use conf::raft_type::{Binlog, OperateType};
@@ -197,16 +203,71 @@ impl Storage {
         db_path: impl AsRef<Path>,
     ) -> Result<mpsc::Receiver<BgTask>> {
         let db_path = db_path.as_ref();
-        let root_load = load_or_create_root_manifest(db_path, self.db_instance_num)?;
+        prepare_or_resume_migration(db_path, self.db_instance_num, &options)?;
+        let mut root_load = load_or_create_root_manifest(db_path, self.db_instance_num)?;
         validate_existing_instance_manifests(
             db_path,
             &root_load.manifest,
             root_load.created_this_call,
         )?;
-        let root_manifest = root_load.manifest;
+        let mut root_manifest = root_load.manifest;
 
         let (handler, receiver) = BgTaskHandler::new();
         let handler_arc = Arc::new(handler);
+        let mut opened_instances = self.open_instances(
+            Arc::clone(&options),
+            db_path,
+            &root_manifest,
+            root_load.created_this_call,
+            &handler_arc,
+        )?;
+
+        let requires_runtime_finalize = root_manifest.migration().is_some_and(|transaction| {
+            matches!(
+                transaction.phase,
+                MigrationPhase::ShadowPromoted | MigrationPhase::NewStorageOpened
+            )
+        });
+        if requires_runtime_finalize {
+            drop(opened_instances);
+            finalize_migration_after_storage_open(db_path, self.db_instance_num, &options)?;
+            root_load = load_or_create_root_manifest(db_path, self.db_instance_num)?;
+            validate_existing_instance_manifests(
+                db_path,
+                &root_load.manifest,
+                root_load.created_this_call,
+            )?;
+            root_manifest = root_load.manifest;
+            opened_instances = self.open_instances(
+                Arc::clone(&options),
+                db_path,
+                &root_manifest,
+                false,
+                &handler_arc,
+            )?;
+        }
+
+        let expiration_manager = Arc::new(ExpirationManager::new(Arc::clone(&handler_arc)));
+        let cleanup_task = expiration_manager.start_cleanup_task();
+
+        self.scan_cursor_states.clear();
+        self.insts = opened_instances;
+        self.bg_task_handler = Some(handler_arc);
+        self.expiration_manager = Some(expiration_manager);
+        self.expiration_cleanup_task = Some(cleanup_task);
+        self.is_opened.store(true, Ordering::SeqCst);
+
+        Ok(receiver)
+    }
+
+    fn open_instances(
+        &self,
+        options: Arc<StorageOptions>,
+        db_path: &Path,
+        root_manifest: &RootStorageManifestV2,
+        allow_manifest_creation: bool,
+        handler_arc: &Arc<BgTaskHandler>,
+    ) -> Result<Vec<Arc<Redis>>> {
         let mut opened_instances = Vec::with_capacity(self.db_instance_num);
         for i in 0..self.db_instance_num {
             let sub_path = db_path.join(i.to_string());
@@ -222,33 +283,22 @@ impl Storage {
             let mut inst = Redis::new(
                 options.clone(),
                 i as i32,
-                Arc::clone(&handler_arc),
+                Arc::clone(handler_arc),
                 Arc::clone(&self.lock_mgr),
             );
             if let Err(e) = inst.open_bound(
                 sub_path_str,
                 i as u32,
-                &root_manifest,
-                root_load.created_this_call,
+                root_manifest,
+                allow_manifest_creation,
             ) {
                 log::error!("open RocksDB{i} failed: {e:?}");
                 return Err(e);
             }
             log::info!("open RocksDB{i} success!");
             opened_instances.push(Arc::new(inst));
         }
-
-        let expiration_manager = Arc::new(ExpirationManager::new(Arc::clone(&handler_arc)));
-        let cleanup_task = expiration_manager.start_cleanup_task();
-
-        self.scan_cursor_states.clear();
-        self.insts = opened_instances;
-        self.bg_task_handler = Some(handler_arc);
-        self.expiration_manager = Some(expiration_manager);
-        self.expiration_cleanup_task = Some(cleanup_task);
-        self.is_opened.store(true, Ordering::SeqCst);
-
-        Ok(receiver)
+        Ok(opened_instances)
     }
 
     pub async fn shutdown(&mut self) {