fix(storage): prevent readiness after native migration failures (#7652)

* fix(storage): prevent readiness after native migration failures

* fix(storage): skip unsupported IAM records before reading

* fix(storage): use stable typed migration metadata errors

* fix(storage): include migration record in startup errors

* test(storage): cover native migration startup failures

* test(storage): use array chunks in migration fixture

---------

Co-authored-by: RJ Regenold <214054+rjregenold@users.noreply.github.com>
Co-authored-by: cxymds <cxymds@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -202,7 +202,9 @@ pub mod bucket {
     }
 
     pub mod migration {
-        pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
+        pub use crate::bucket::migration::{
+            LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_iam_config,
+        };
     }
 
     pub mod object_lock {

crates/ecstore/src/bucket/migration.rs

@@ -17,7 +17,7 @@
 use crate::bucket::metadata::BUCKET_METADATA_FILE;
 use crate::bucket::replication::ReplicationMigrationBridge;
 use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
-use crate::error::Error;
+use crate::error::{Error, Result, is_err_strict_not_found, is_err_strict_volume_not_found};
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
 use crate::storage_api_contracts::{
     bucket::{BucketOperations, BucketOptions},
@@ -33,7 +33,7 @@ use rustfs_utils::path::SLASH_SEPARATOR;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tracing::{debug, info, warn};
+use tracing::{debug, info};
 
 /// IAM config prefix under meta bucket (e.g. config/iam/).
 const IAM_CONFIG_PREFIX: &str = "config/iam";
@@ -53,6 +53,39 @@ type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
 type ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, Error>;
 type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
 
+#[derive(Clone, Debug, thiserror::Error)]
+enum MigrationMetadataError {
+    #[error("empty legacy metadata: {0}")]
+    Empty(String),
+    #[error("incompatible legacy metadata: {0}")]
+    Incompatible(String),
+}
+
+impl From<MigrationMetadataError> for Error {
+    fn from(error: MigrationMetadataError) -> Self {
+        let message = match &error {
+            MigrationMetadataError::Empty(_) => "empty legacy metadata",
+            MigrationMetadataError::Incompatible(_) => "incompatible legacy metadata",
+        };
+        // Keep the record path in the typed source, not in the quorum grouping key.
+        Self::other_with_context(message, error)
+    }
+}
+
+/// Converts a migration failure at the startup boundary, rendering the safe
+/// record path while leaving storage-layer error grouping stable.
+pub fn migration_startup_error(error: Error) -> std::io::Error {
+    if let Error::Io(io_error) = &error
+        && let Some(metadata_error) = io_error
+            .get_ref()
+            .and_then(|context| context.source())
+            .and_then(|source| source.downcast_ref::<MigrationMetadataError>())
+    {
+        return std::io::Error::other(metadata_error.clone());
+    }
+    std::io::Error::other(error)
+}
+
 /// Callback used to decrypt an at-rest config blob during MinIO -> RustFS migration.
 ///
 /// MinIO encrypts IAM identity/service-account files and the server config at rest
@@ -211,7 +244,7 @@ fn normalize_bucket_meta_blob(path: &str, data: &[u8]) -> std::result::Result<Op
 /// Uses list_bucket (from disk volumes) to get bucket names, since list_objects_v2 on the legacy
 /// meta bucket may not work (legacy format differs from object layer expectations).
 /// Skips buckets that already exist in RustFS (idempotent).
-pub async fn try_migrate_bucket_metadata<S>(store: Arc<S>)
+pub async fn try_migrate_bucket_metadata<S>(store: Arc<S>) -> Result<()>
 where
     S: BucketOperations<Error = crate::error::Error>
         + ObjectIO<
@@ -231,25 +264,18 @@ where
             DeletedObject = DeletedObject,
         >,
 {
-    let buckets_list = match store
+    let buckets_list = store
         .list_bucket(&BucketOptions {
             no_metadata: true,
             ..Default::default()
         })
-        .await
-    {
-        Ok(b) => b,
-        Err(e) => {
-            warn!("list buckets failed (skip migration): {e}");
-            return;
-        }
-    };
+        .await?;
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|b| b.name).collect();
 
     if buckets.is_empty() {
         debug!("No migrating bucket metadata found");
-        return;
+        return Ok(());
     }
 
     debug!("Found {} migrating bucket metadata, migrating...", buckets.len());
@@ -263,26 +289,40 @@ where
 
     for bucket in buckets {
         let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}");
-        migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await;
+        migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await?;
 
         let resync_path = format!(
             "{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{REPLICATION_META_DIR}{SLASH_SEPARATOR}{RESYNC_META_FILE}"
         );
-        migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await;
+        migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await?;
+    }
+    Ok(())
+}
+
+async fn migration_target_exists<S: EcstoreObjectOperations>(store: &S, path: &str) -> Result<bool> {
+    match store
+        .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
+        .await
+    {
+        Ok(_) => Ok(true),
+        Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => Ok(false),
+        Err(err) => Err(err),
     }
 }
 
-async fn migrate_one_if_missing<S>(store: Arc<S>, opts: &ObjectOptions, headers: &HeaderMap, path: &str, label: &str)
+async fn migrate_one_if_missing<S>(
+    store: Arc<S>,
+    opts: &ObjectOptions,
+    headers: &HeaderMap,
+    path: &str,
+    label: &str,
+) -> Result<()>
 where
     S: EcstoreObjectIO + EcstoreObjectOperations,
 {
-    if store
-        .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
-        .await
-        .is_ok()
-    {
+    if migration_target_exists(store.as_ref(), path).await? {
         debug!("{label} already exists in RustFS, skip");
-        return;
+        return Ok(());
     }
 
     let mut rd = match store
@@ -290,43 +330,31 @@ where
         .await
     {
         Ok(r) => r,
-        Err(e) => {
-            debug!("read migrating {label}: {e}");
-            return;
-        }
+        // Ordinary RustFS deployments have no legacy bucket, and optional
+        // legacy settings (such as replication resync) may not exist.
+        Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => return Ok(()),
+        Err(err) => return Err(err),
     };
 
-    let data = match rd.read_all().await {
-        Ok(d) if !d.is_empty() => d,
-        Ok(_) => return,
-        Err(e) => {
-            debug!("read migrating {label} body: {e}");
-            return;
-        }
-    };
-
-    let data = match normalize_bucket_meta_blob(path, &data) {
-        Ok(Some(normalized)) => normalized,
-        Ok(None) => data,
-        Err(e) => {
-            warn!("skip {label} migration due to incompatible format: {e}");
-            return;
-        }
-    };
+    let data = rd.read_all().await?;
+    if data.is_empty() {
+        return Err(MigrationMetadataError::Empty(path.to_owned()).into());
+    }
+    let data = normalize_bucket_meta_blob(path, &data)
+        .map_err(|_| MigrationMetadataError::Incompatible(path.to_owned()))?
+        .unwrap_or(data);
 
     let mut put_data = PutObjReader::from_vec(data);
-    if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await {
-        warn!("write {label}: {e}");
-    } else {
-        info!("Migrated {label}");
-    }
+    store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await?;
+    info!("Migrated {label}");
+    Ok(())
 }
 
 /// Migrates IAM config from legacy meta bucket `config/iam/` to RustFS meta bucket.
 /// Lists all objects under the IAM prefix in the source, copies each to the target if not present.
 /// Skips objects that already exist in RustFS (idempotent).
-/// If list_objects_v2 on the legacy bucket fails (e.g. format differs), migration is skipped.
-pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>)
+/// An absent legacy bucket is a no-op; migration errors prevent startup readiness.
+pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>) -> Result<()>
 where
     S: ListOperations<
             Error = crate::error::Error,
@@ -366,47 +394,36 @@ where
     loop {
         let list_result = match store
             .clone()
-            .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation, None, 500, false, None, false)
+            .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation.clone(), None, 500, false, None, false)
             .await
         {
             Ok(r) => r,
-            Err(e) => {
-                debug!("list IAM config from legacy bucket failed (skip migration): {e}");
-                return;
-            }
+            Err(err) if is_err_strict_volume_not_found(&err) => return Ok(()),
+            Err(err) => return Err(err),
         };
 
         for obj in list_result.objects {
             let path = &obj.name;
-            if path.is_empty() || path.ends_with('/') {
+            // Unsupported records must not trigger target lookups, reads, or decryption.
+            if path != IAM_FORMAT_FILE_PATH
+                && !is_identity_path(path)
+                && !is_group_path(path)
+                && !is_policy_doc_path(path)
+                && !is_policy_mapping_path(path)
+            {
                 continue;
             }
-            if store
-                .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
-                .await
-                .is_ok()
-            {
+            if migration_target_exists(store.as_ref(), path).await? {
                 debug!("IAM config already exists in RustFS, skip: {path}");
                 continue;
             }
-            let mut rd = match store
+            let mut rd = store
                 .get_object_reader(MIGRATING_META_BUCKET, path, None, h.clone(), &opts)
-                .await
-            {
-                Ok(r) => r,
-                Err(e) => {
-                    debug!("read migrating IAM config {path}: {e}");
-                    continue;
-                }
-            };
-            let data = match rd.read_all().await {
-                Ok(d) if !d.is_empty() => d,
-                Ok(_) => continue,
-                Err(e) => {
-                    debug!("read migrating IAM config {path} body: {e}");
-                    continue;
-                }
-            };
+                .await?;
+            let data = rd.read_all().await?;
+            if data.is_empty() {
+                return Err(MigrationMetadataError::Empty(path.to_owned()).into());
+            }
             // MinIO encrypts IAM identity/service-account files at rest. Decrypt
             // before normalizing; fall back to the raw bytes when no key applies
             // (plaintext blobs, or nothing to decrypt) so existing behavior holds.
@@ -420,22 +437,17 @@ where
                     debug!("skip unsupported IAM config path during migration: {path}");
                     continue;
                 }
-                Err(e) => {
-                    warn!("skip IAM config migration due to incompatible format, path: {path}, err: {e}");
-                    continue;
-                }
+                // Parser errors may contain credential data. Report only the path.
+                Err(_) => return Err(MigrationMetadataError::Incompatible(path.to_owned()).into()),
             };
             let mut put_data = PutObjReader::from_vec(data);
-            if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await {
-                warn!("write IAM config {path}: {e}");
-            } else {
-                info!("Migrated IAM config: {path}");
-                total_migrated += 1;
-            }
+            store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await?;
+            info!("Migrated IAM config: {path}");
+            total_migrated += 1;
         }
 
-        continuation = list_result.next_continuation_token.or(list_result.continuation_token);
-        if !list_result.is_truncated || continuation.is_none() {
+        continuation = next_iam_migration_page(list_result.is_truncated, continuation, list_result.next_continuation_token)?;
+        if continuation.is_none() {
             break;
         }
     }
@@ -443,10 +455,74 @@ where
     if total_migrated > 0 {
         info!("IAM migration complete: {} object(s) migrated", total_migrated);
     }
+    Ok(())
+}
+
+fn next_iam_migration_page(truncated: bool, previous: Option<String>, next: Option<String>) -> Result<Option<String>> {
+    if !truncated {
+        return Ok(None);
+    }
+    let next = next.filter(|token| !token.is_empty());
+    if next.is_none() || next == previous {
+        return Err(Error::other("legacy IAM migration listing did not advance"));
+    }
+    Ok(next)
 }
 
 #[cfg(test)]
 mod tests {
+    #[test]
+    fn migration_errors_group_by_cause_and_retain_typed_record_context() {
+        use super::{Error, MigrationMetadataError};
+
+        for (make_error, message) in [
+            (
+                MigrationMetadataError::Empty as fn(String) -> MigrationMetadataError,
+                "empty legacy metadata",
+            ),
+            (MigrationMetadataError::Incompatible, "incompatible legacy metadata"),
+        ] {
+            let first: Error = make_error("buckets/first/.metadata.bin".into()).into();
+            let second: Error = make_error("buckets/second/.metadata.bin".into()).into();
+            assert_eq!(first, second, "record paths must not fragment error grouping");
+            assert_eq!(first.clone(), second, "cloning must preserve error grouping");
+
+            let io_error = std::io::Error::from(first);
+            let detail = io_error
+                .get_ref()
+                .and_then(|context| context.source())
+                .expect("record context must remain in the error source");
+            assert!(detail.downcast_ref::<MigrationMetadataError>().is_some());
+            assert!(detail.to_string().contains("buckets/first/.metadata.bin"));
+
+            let startup_error = super::migration_startup_error(make_error("buckets/startup/.metadata.bin".into()).into());
+            assert!(
+                startup_error
+                    .get_ref()
+                    .is_some_and(|source| source.is::<MigrationMetadataError>())
+            );
+            assert_eq!(startup_error.to_string(), format!("{message}: buckets/startup/.metadata.bin"));
+        }
+        assert_ne!(
+            Error::from(MigrationMetadataError::Empty("record".into())),
+            Error::from(MigrationMetadataError::Incompatible("record".into())),
+            "different migration failures must remain distinguishable"
+        );
+    }
+
+    #[test]
+    fn truncated_iam_listing_cannot_report_completed_migration() {
+        use super::next_iam_migration_page;
+        assert_eq!(next_iam_migration_page(false, Some("old".into()), None).expect("final page"), None);
+        assert_eq!(
+            next_iam_migration_page(true, Some("old".into()), Some("next".into())).expect("advancing page"),
+            Some("next".into())
+        );
+        for next in [None, Some(String::new()), Some("old".into())] {
+            assert!(next_iam_migration_page(true, Some("old".into()), next).is_err());
+        }
+    }
+
     use super::{normalize_bucket_meta_blob, normalize_iam_config_blob};
     use crate::bucket::replication::{
         BucketReplicationResyncStatus, ReplicationMigrationBridge, ResyncStatusType, TargetReplicationResyncStatus,
@@ -659,6 +735,13 @@ mod tests {
             .collect();
         crate::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), existing).await;
 
+        super::try_migrate_bucket_metadata(ecstore.clone())
+            .await
+            .expect("fresh stores do not require a legacy metadata bucket");
+        super::try_migrate_iam_config(ecstore.clone(), None)
+            .await
+            .expect("fresh stores do not require a legacy IAM bucket");
+
         let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}interop{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}");
         let put_opts = ObjectOptions::default();
 
@@ -680,8 +763,31 @@ mod tests {
             .await
             .expect("seed .minio.sys bucket metadata");
 
-        // --- Run the real startup migration. ---
-        super::try_migrate_bucket_metadata(ecstore.clone()).await;
+        // A partial import must report failure, even if the main bucket
+        // metadata copied successfully before an incompatible resync record.
+        let resync_path = format!("{BUCKET_META_PREFIX}/interop/.replication/resync.bin");
+        ecstore
+            .put_object(
+                MIGRATING_META_BUCKET,
+                &resync_path,
+                &mut PutObjReader::from_vec(b"invalid resync metadata".to_vec()),
+                &put_opts,
+            )
+            .await
+            .expect("seed malformed legacy resync metadata");
+        assert!(
+            super::try_migrate_bucket_metadata(ecstore.clone()).await.is_err(),
+            "incompatible native metadata must not be reported as a completed migration"
+        );
+        ecstore
+            .delete_object(MIGRATING_META_BUCKET, &resync_path, ObjectOptions::default())
+            .await
+            .expect("remove invalid optional legacy resync record");
+
+        // Retry the real startup migration after repairing the source.
+        super::try_migrate_bucket_metadata(ecstore.clone())
+            .await
+            .expect("native bucket metadata migration completes");
 
         // --- The migrated `.rustfs.sys` blob must carry every MinIO config, ---
         // byte-identical to the source (typed XML/JSON parsing of these fields is

crates/iam/tests/minio_iam_migration_test.rs

@@ -100,6 +100,34 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
         .await;
     env.make_bucket(LEGACY_META_BUCKET, false).await;
 
+    for (path, body) in [
+        ("config/iam/empty.json", Vec::new()),
+        ("config/iam/users/ignored/extra.json", b"not JSON".to_vec()),
+    ] {
+        env.put_object_bytes(LEGACY_META_BUCKET, path, body).await;
+    }
+    try_migrate_iam_config(
+        env.ecstore.clone(),
+        Some(std::sync::Arc::new(|_| panic!("unsupported IAM records must not be decrypted"))),
+    )
+    .await
+    .expect("unsupported IAM records, including empty objects, must be skipped");
+
+    let format_path = "config/iam/format.json";
+    for body in [Vec::new(), b"invalid IAM format".to_vec()] {
+        env.put_object_bytes(LEGACY_META_BUCKET, format_path, body).await;
+        let error = try_migrate_iam_config(env.ecstore.clone(), None)
+            .await
+            .expect_err("empty or incompatible supported IAM metadata must prevent startup readiness");
+        let io_error = std::io::Error::from(error);
+        let detail = io_error
+            .get_ref()
+            .and_then(|context| context.source())
+            .expect("failure must retain the supported record in its source");
+        assert!(detail.to_string().contains(format_path), "failure must identify the supported record");
+    }
+    seed_legacy_iam_object(&env, format_path, &json!({"version": 1})).await;
+
     let regular_source = json!({
         "version": 1,
         "credentials": {
@@ -155,7 +183,12 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
     )
     .await;
 
-    try_migrate_iam_config(env.ecstore.clone(), None).await;
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("legacy IAM migration completes after source repair");
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("completed legacy IAM migration is idempotent");
 
     let store = ObjectStore::new(env.ecstore);
     assert_identity_survives(

rustfs/src/startup_bucket_metadata.rs

@@ -62,10 +62,10 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, c
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
-    try_migrate_bucket_metadata(store.clone()).await;
+    try_migrate_bucket_metadata(store.clone()).await?;
     init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
-    try_migrate_iam_config(store).await;
+    try_migrate_iam_config(store).await?;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false);
 
     Ok(buckets)
@@ -82,9 +82,9 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: Cance
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
-    try_migrate_bucket_metadata(store.clone()).await;
+    try_migrate_bucket_metadata(store.clone()).await?;
 
-    try_migrate_iam_config(store.clone()).await;
+    try_migrate_iam_config(store.clone()).await?;
     init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store, buckets.clone()).await;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true);

rustfs/src/storage/storage_api.rs

@@ -1185,17 +1185,21 @@ pub(crate) fn get_global_transition_state() -> Arc<TransitionState> {
     ecstore_bucket::lifecycle::bucket_lifecycle_ops::get_global_transition_state()
 }
 
-pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) {
-    ecstore_bucket::migration::try_migrate_bucket_metadata(store).await;
+pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) -> std::io::Result<()> {
+    ecstore_bucket::migration::try_migrate_bucket_metadata(store)
+        .await
+        .map_err(ecstore_bucket::migration::migration_startup_error)
 }
 
-pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) {
+pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) -> std::io::Result<()> {
     // MinIO encrypts IAM identity/service-account files at rest with a key derived
     // from the root credentials. Inject the IAM crate's decryption so those blobs
     // are decrypted before normalization instead of being skipped as "incompatible".
     let decrypt_fn: ecstore_bucket::migration::LegacyBlobDecryptFn =
         Arc::new(|data: &[u8]| rustfs_iam::try_decrypt_iam_blob(data));
-    ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn)).await;
+    ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn))
+        .await
+        .map_err(ecstore_bucket::migration::migration_startup_error)
 }
 
 pub(crate) fn init_ecstore_config() {

rustfs/tests/native_migration_startup_test.rs

@@ -0,0 +1,326 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![recursion_limit = "256"]
+
+use reqwest::StatusCode;
+use rustfs::embedded::{RustFSServerBuilder, find_available_port};
+use rustfs_ecstore::api::config::com::{delete_config, read_config};
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::time::Duration;
+use tokio::process::Command;
+
+mod common;
+
+const TEST_NAME: &str = "native_migration_failure_blocks_server_startup_and_repair_preserves_records";
+const STAGE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STAGE";
+const ROOT_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ROOT";
+const ADDRESS_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ADDRESS";
+const FAILURE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_FAILURE";
+const STOP_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STOP";
+const ACCESS_KEY: &str = "native-migration-root";
+const SECRET_KEY: &str = "native-migration-root-secret";
+const LEGACY_BUCKET: &str = ".minio.sys";
+const TARGET_BUCKET: &str = ".rustfs.sys";
+const BUCKET_METADATA: &str = "buckets/interop/.metadata.bin";
+const IAM_RECORD: &str = "config/iam/groups/migration-group/members.json";
+const IAM_FORMAT: &str = "config/iam/format.json";
+const EXISTING_FORMAT: &[u8] = br#"{"version":1}"#;
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(60);
+
+#[derive(Clone, Copy, Debug)]
+enum StartupMode {
+    Server,
+    Embedded,
+}
+
+fn volumes(root: &Path) -> Vec<PathBuf> {
+    (1..=4).map(|index| root.join(format!("disk{index}"))).collect()
+}
+
+fn minio_bucket_metadata() -> Vec<u8> {
+    let hex: String = include_str!("../../crates/ecstore/tests/fixtures/minio/bucket_metadata.blob.hex")
+        .chars()
+        .filter(|ch| !ch.is_whitespace())
+        .collect();
+    hex.as_bytes()
+        .as_chunks::<2>()
+        .0
+        .iter()
+        .map(|pair| {
+            u8::from_str_radix(std::str::from_utf8(pair).expect("fixture hex is UTF-8"), 16).expect("valid MinIO fixture hex")
+        })
+        .collect()
+}
+
+async fn prepare_or_verify_fixture(root: &Path, seed: bool) {
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .base_dir(root)
+        .disk_count(4)
+        .build()
+        .await;
+    if seed {
+        env.make_bucket("interop", false).await;
+        env.make_bucket(LEGACY_BUCKET, false).await;
+        env.put_object_bytes(LEGACY_BUCKET, BUCKET_METADATA, minio_bucket_metadata())
+            .await;
+        env.put_object_bytes(
+            LEGACY_BUCKET,
+            IAM_RECORD,
+            br#"{"version":1,"status":"enabled","members":[],"updatedAt":"2026-09-10T00:00:00Z"}"#.to_vec(),
+        )
+        .await;
+        env.put_object_bytes(TARGET_BUCKET, IAM_FORMAT, EXISTING_FORMAT.to_vec())
+            .await;
+        // A completed record must be skipped before reading even a broken old copy.
+        env.put_object_bytes(LEGACY_BUCKET, IAM_FORMAT, b"do not overwrite the existing target".to_vec())
+            .await;
+        delete_config(env.ecstore.clone(), BUCKET_METADATA)
+            .await
+            .expect("leave bucket metadata pending migration");
+    } else {
+        assert_eq!(
+            read_config(env.ecstore.clone(), BUCKET_METADATA)
+                .await
+                .expect("migrated bucket metadata"),
+            minio_bucket_metadata(),
+            "migration must preserve the MinIO bucket settings"
+        );
+        let group: serde_json::Value = serde_json::from_slice(
+            &read_config(env.ecstore.clone(), IAM_RECORD)
+                .await
+                .expect("migrated IAM group"),
+        )
+        .expect("valid migrated IAM JSON");
+        assert_eq!(group["status"], "enabled");
+        assert_eq!(group["members"], serde_json::json!([]));
+    }
+    assert_eq!(
+        read_config(env.ecstore.clone(), IAM_FORMAT)
+            .await
+            .expect("existing IAM format"),
+        EXISTING_FORMAT,
+        "retry must not overwrite records already migrated"
+    );
+}
+
+async fn run_embedded_child(root: &Path) {
+    let address = std::env::var(ADDRESS_ENV).expect("embedded child address");
+    let result = RustFSServerBuilder::new()
+        .address(address)
+        .access_key(ACCESS_KEY)
+        .secret_key(SECRET_KEY)
+        .volumes(volumes(root).iter().map(|path| path.to_string_lossy().into_owned()).collect())
+        .build()
+        .await;
+    match result {
+        Ok(server) => {
+            let stop = PathBuf::from(std::env::var_os(STOP_ENV).expect("embedded stop path"));
+            while !stop.exists() {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+            }
+            server.shutdown().await;
+        }
+        Err(error) => {
+            fs::write(std::env::var_os(FAILURE_ENV).expect("embedded failure path"), error.to_string())
+                .expect("record the actual embedded startup error");
+        }
+    }
+}
+
+fn child_command(root: &Path, stage: &str, log: &Path) -> Command {
+    let mut command = Command::new(std::env::current_exe().expect("integration test executable"));
+    command
+        .args(["--exact", TEST_NAME, "--nocapture"])
+        .env(STAGE_ENV, stage)
+        .env(ROOT_ENV, root);
+    configure_process(&mut command, log);
+    command
+}
+
+fn configure_process(command: &mut Command, log: &Path) {
+    let output = fs::File::create(log).expect("create isolated process log");
+    command
+        // These disposable erasure volumes intentionally share the test runner's disk.
+        .env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true")
+        .env("RUSTFS_CONSOLE_ENABLE", "false")
+        .env("NO_PROXY", "localhost,127.0.0.1,::1")
+        .env("no_proxy", "localhost,127.0.0.1,::1")
+        .env("RUST_LOG", "warn")
+        .stdin(Stdio::null())
+        .stdout(Stdio::from(output.try_clone().expect("clone process log")))
+        .stderr(Stdio::from(output))
+        .kill_on_drop(true);
+}
+
+async fn fixture_process(root: &Path, stage: &str) {
+    let log = root.join(format!("{stage}.log"));
+    let status = tokio::time::timeout(STARTUP_TIMEOUT, child_command(root, stage, &log).status())
+        .await
+        .expect("fixture process must finish")
+        .expect("run fixture process");
+    assert!(status.success(), "{stage} failed: {}", fs::read_to_string(log).expect("fixture log"));
+}
+
+async fn check_startup(root: &Path, mode: StartupMode, failure_record: Option<&str>, label: &str) {
+    let ready = failure_record.is_none();
+    let address = format!("127.0.0.1:{}", find_available_port().expect("free startup probe port"));
+    let log = root.join(format!("{label}.log"));
+    let failure = root.join(format!("{label}.failure"));
+    let stop = root.join(format!("{label}.stop"));
+    let mut command = match mode {
+        StartupMode::Server => {
+            let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs"));
+            command
+                .args(["--address", &address, "--access-key", ACCESS_KEY, "--secret-key", SECRET_KEY])
+                .args(volumes(root));
+            configure_process(&mut command, &log);
+            command
+        }
+        StartupMode::Embedded => {
+            let mut command = child_command(root, "embedded", &log);
+            command
+                .env(ADDRESS_ENV, &address)
+                .env(FAILURE_ENV, &failure)
+                .env(STOP_ENV, &stop);
+            command
+        }
+    };
+    let mut child = command.spawn().expect("start isolated server process");
+    let http = reqwest::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_millis(500))
+        .build()
+        .expect("local readiness client");
+    let result = tokio::time::timeout(STARTUP_TIMEOUT, async {
+        loop {
+            if let Ok(response) = http.get(format!("http://{address}/health/ready")).send().await
+                && response.status() == StatusCode::OK
+            {
+                assert!(ready, "{mode:?} published Ready after a migration I/O failure");
+                return;
+            }
+            if let Some(status) = child.try_wait().expect("poll server process") {
+                let details = fs::read_to_string(&log).expect("startup log");
+                assert!(!ready, "{mode:?} exited before Ready ({status}): {details}");
+                let record = failure_record.expect("failed startup has an obstructed record");
+                match mode {
+                    StartupMode::Server => {
+                        assert_eq!(status.code(), Some(1), "startup must fail: {details}");
+                        assert_migration_io_error(&details, record);
+                    }
+                    StartupMode::Embedded => {
+                        assert!(status.success(), "embedded test process failed unexpectedly: {details}");
+                        let error = fs::read_to_string(&failure).expect("embedded startup returned an error");
+                        assert_migration_io_error(&error, record);
+                    }
+                }
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(25)).await;
+        }
+    })
+    .await;
+    assert!(
+        result.is_ok(),
+        "{mode:?} did not reach the expected startup outcome: {}",
+        fs::read_to_string(&log).expect("startup diagnostics")
+    );
+    if ready {
+        match mode {
+            StartupMode::Embedded => {
+                fs::write(stop, b"stop").expect("request embedded shutdown");
+                assert!(
+                    tokio::time::timeout(STARTUP_TIMEOUT, child.wait())
+                        .await
+                        .expect("embedded shutdown completes")
+                        .expect("wait for embedded shutdown")
+                        .success()
+                );
+            }
+            StartupMode::Server => child.kill().await.expect("stop the isolated server"),
+        }
+    }
+}
+
+fn assert_migration_io_error(error: &str, record: &str) {
+    let lower = error.to_ascii_lowercase();
+    assert!(
+        (lower.contains("access denied")
+            || lower.contains("access is denied")
+            || lower.contains("not a directory")
+            || lower.contains("not regular"))
+            && error.contains(&format!("{TARGET_BUCKET}/{record}")),
+        "startup must fail because of the obstructed metadata record, not an unrelated initialization error: {error}"
+    );
+}
+
+async fn run_startup_cases(mode: StartupMode) {
+    let ordinary = tempfile::TempDir::with_prefix("rustfs-no-legacy-").expect("ordinary store");
+    for volume in volumes(ordinary.path()) {
+        fs::create_dir_all(volume).expect("ordinary volume");
+    }
+    check_startup(ordinary.path(), mode, None, "ordinary").await;
+
+    let control = tempfile::TempDir::with_prefix("rustfs-migration-control-").expect("control fixture");
+    fixture_process(control.path(), "seed").await;
+    check_startup(control.path(), mode, None, "control").await;
+    fixture_process(control.path(), "verify").await;
+
+    for record in [BUCKET_METADATA, IAM_RECORD] {
+        let target = tempfile::TempDir::with_prefix("rustfs-migration-failure-").expect("disposable migration target");
+        fixture_process(target.path(), "seed").await;
+        let blockers: Vec<_> = volumes(target.path())
+            .iter()
+            .map(|volume| volume.join(TARGET_BUCKET).join(record))
+            .collect();
+        for blocker in &blockers {
+            fs::create_dir_all(blocker.parent().expect("record parent")).expect("create target parent");
+            assert!(!blocker.exists(), "the record must still need migration");
+            // A non-directory target causes real filesystem I/O errors even when tests run as root.
+            fs::write(blocker, b"blocked migration target").expect("block only the destination record");
+        }
+        check_startup(target.path(), mode, Some(record), "blocked").await;
+        for blocker in blockers {
+            fs::remove_file(blocker).expect("repair the same partially migrated target");
+        }
+        check_startup(target.path(), mode, None, "repaired").await;
+        fixture_process(target.path(), "verify").await;
+    }
+}
+
+#[test]
+fn native_migration_failure_blocks_server_startup_and_repair_preserves_records() {
+    // Cold processes keep failed initialization and cached metadata out of subsequent restart attempts.
+    common::run_embedded_test(|| async {
+        match std::env::var(STAGE_ENV).ok().as_deref() {
+            Some("seed") => {
+                prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), true).await
+            }
+            Some("verify") => {
+                prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), false).await
+            }
+            Some("embedded") => run_embedded_child(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root"))).await,
+            None => run_startup_cases(StartupMode::Server).await,
+            Some(stage) => panic!("unknown native migration test stage: {stage}"),
+        }
+    });
+}
+
+#[test]
+fn native_migration_failure_blocks_embedded_startup_and_repair_preserves_records() {
+    common::run_embedded_test(|| run_startup_cases(StartupMode::Embedded));
+}