fix(migration): decrypt MinIO IAM & server config on drop-in migration (#4358)

* fix(migration): decrypt MinIO IAM & server config on drop-in migration

MinIO encrypts IAM identity/service-account files and the server config at
rest with a key derived from the root credentials. The drop-in migration
paths read those blobs from the legacy `.minio.sys` bucket and parsed them
as plaintext JSON, so any encrypted blob failed to parse and was silently
skipped with "incompatible format". This is why users migrating from MinIO
kept their buckets/objects/policies but lost users and access keys (#2212).

The IAM load path already knows how to decrypt these blobs (RustFS master
keys plus MinIO-compatible legacy keys derived from the root credentials),
but that logic lived behind a private method and was never used by the
migration paths. Expose it as `rustfs_iam::try_decrypt_iam_blob` and inject
it into both migration paths via a `LegacyBlobDecryptFn` callback (ecstore
cannot depend on the IAM crate, so the closure is wired in the binary crate).
When a blob cannot be decrypted the raw bytes are used as-is, preserving the
previous plaintext-only behavior with no regression.

Also improve object-layer migration observability without changing control
flow: `try_migrate_format` now distinguishes "no legacy format" (a normal
fresh install) from "legacy format present but incompatible", and the caller
logs a loud error before initializing a fresh format that would leave the
existing MinIO objects unreadable. Topology/version skip reasons are promoted
from debug to warn.

Fixes a pre-existing test isolation race by marking
`test_recovery_falls_back_to_default_config_when_blob_stays_corrupt` serial,
since it reads a process-wide env var toggled by a sibling test.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(migration): box FormatV3 in LegacyFormatOutcome to satisfy clippy

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): stabilize concurrent multipart resend lock timeout

concurrent_resend_same_part_commits_one_generation spawns 6 same-part
resends whose cross-disk commits serialize on the per-uploadId commit
lock. Under the full nextest suite the parallel disk load pushes those
serialized commits past the small default lock-acquire timeout (5s),
producing a spurious `Lock(Timeout ...)` unrelated to the property under
test (observed on CI at 5.775s vs ~0.5s in isolation).

Raise RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT to the production default (30s)
for the concurrent-commit section via temp_env, so the regression guard
reflects correctness (exactly one intact generation) rather than disk
latency under CI load. The meaningful assertions are unchanged, and
#[serial] keeps the process-wide env override isolated.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(lock): bound fast-lock notification wait to prevent lost-wakeup stall

The real cause of the concurrent_resend_same_part_commits_one_generation
failures was a lost wakeup in the fast-lock slow path, not disk latency:
raising the acquire timeout to 30s only delayed the failure (it then timed
out at 30s), proving a genuine stall rather than overload.

In acquire_lock_slow_path a waiter that reaches the notification phase did a
single `timeout(remaining, wait_for_write())` spanning the whole acquire
budget, and treated that wait's elapse as a hard `Timeout`. But the release
path only notifies when `writer_waiters > 0`, so if the holder releases in
the gap after the waiter's `try_acquire` fails and before it registers as a
waiter, no notification (and no stored permit, since the pooled `Notify` is
gated) is produced. The waiter then blocks until the deadline even though the
lock is free and stays free — a spurious lock-acquire timeout. The shared
process-wide notify pool makes it worse: a wakeup can be consumed by a waiter
of a different lock hashing to the same slot.

Bound each notification wait (NOTIFY_WAIT_CAP = 50ms) and, on elapse, loop
back and re-`try_acquire` instead of returning `Timeout`; the deadline check
at the top of the loop is the single source of truth for timing out. A
lost/stolen wakeup now degrades to bounded re-polling (acquire within ~50ms
of the lock becoming free) instead of stalling for the whole timeout.
Correctness (mutual exclusion) is unchanged — acquisition still only happens
via `try_acquire_*`.

Add a regression test that reproduces the stall (holder + late waiter across
many keys): it times out without the fix and passes in ~1s with it. Revert
the earlier acquire-timeout workaround in the multipart test now that the
underlying stall is fixed, so it runs under the default timeout again.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/api/mod.rs

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

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -4731,8 +4731,12 @@ mod tests {
             });
         }
 
-        // Every concurrent resend must succeed; the fix must not serialize the
-        // streaming phase into lock-acquire timeouts.
+        // Every concurrent resend must succeed. The per-uploadId commit lock must
+        // not starve a waiter into a lock-acquire timeout: the shared fast-lock
+        // notify pool could otherwise route this lock's wakeup to a waiter of an
+        // unrelated lock and strand this one until the deadline (fixed in
+        // fast_lock::shard by bounding each notification wait, so a lost/stolen
+        // wakeup degrades to bounded re-polling instead of a hard timeout).
         let mut results = Vec::new();
         while let Some(joined) = tasks.join_next().await {
             let outcome = joined.expect("put_object_part task should not panic");

crates/ecstore/src/bucket/migration.rs

@@ -52,6 +52,17 @@ type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
 type ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, Error>;
 type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
 
+/// 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
+/// with a key derived from the root credentials. The migration paths live in
+/// `ecstore`, which cannot depend on the IAM crate that owns the decryption keys,
+/// so the caller injects the decryption logic. Given raw bytes read from the
+/// legacy meta bucket, it returns the plaintext when a key succeeds, or `None`
+/// when the blob cannot be decrypted (in which case the raw bytes are used as-is,
+/// preserving the original plaintext-only behavior).
+pub type LegacyBlobDecryptFn = Arc<dyn Fn(&[u8]) -> Option<Vec<u8>> + Send + Sync>;
+
 #[derive(Debug, Serialize, Deserialize)]
 struct CompatIamFormat {
     #[serde(default)]
@@ -305,7 +316,7 @@ where
 /// 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>)
+pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>)
 where
     S: ListOperations<
             Error = crate::error::Error,
@@ -386,6 +397,13 @@ where
                     continue;
                 }
             };
+            // 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.
+            let data = match &decrypt_fn {
+                Some(decrypt) => decrypt(&data).unwrap_or(data),
+                None => data,
+            };
             let data = match normalize_iam_config_blob(path, &data) {
                 Ok(Some(normalized)) => normalized,
                 Ok(None) => {
@@ -445,6 +463,36 @@ mod tests {
         assert!(updated_at.contains('T'), "updatedAt should be RFC3339-like");
     }
 
+    #[test]
+    fn test_encrypted_identity_requires_decrypt_before_normalize() {
+        // Reproduces the MinIO drop-in migration gap: an IAM identity file that
+        // MinIO encrypted at rest is NOT valid JSON, so normalize fails outright.
+        // The migration's decrypt callback must run first to recover it.
+        let path = "config/iam/users/alice/identity.json";
+        let plaintext = br#"{"version":1,"credentials":{"accessKey":"alice","secretKey":"alicesecret"}}"#;
+
+        // Simulate MinIO's at-rest encryption of the identity blob.
+        let ciphertext = rustfs_crypto::encrypt_data(b"root-secret-key", plaintext).expect("encrypt identity blob");
+
+        // Without decryption (old behavior): normalize can't parse ciphertext -> Err -> skipped.
+        assert!(
+            normalize_iam_config_blob(path, &ciphertext).is_err(),
+            "ciphertext must not parse as a JSON identity"
+        );
+
+        // With the decrypt callback recovering plaintext first: normalize succeeds.
+        let decrypt_fn: super::LegacyBlobDecryptFn =
+            std::sync::Arc::new(|data: &[u8]| rustfs_crypto::decrypt_data(b"root-secret-key", data).ok());
+        let recovered = decrypt_fn(&ciphertext).expect("callback should decrypt the identity blob");
+        assert_eq!(recovered, plaintext);
+
+        let normalized = normalize_iam_config_blob(path, &recovered)
+            .expect("normalize should succeed on decrypted plaintext")
+            .expect("identity path should be supported");
+        let v: serde_json::Value = serde_json::from_slice(&normalized).expect("output should be valid JSON");
+        assert!(v.get("updatedAt").is_some(), "normalize should backfill updatedAt");
+    }
+
     #[test]
     fn test_normalize_bucket_meta_blob_resync_reencode() {
         let path = ".buckets/test/.replication/resync.bin";

crates/ecstore/src/config/com.rs

@@ -1089,7 +1089,7 @@ fn is_object_not_found(err: &Error) -> bool {
     *err == Error::FileNotFound || matches!(err, Error::ObjectNotFound(_, _) | Error::BucketNotFound(_))
 }
 
-pub async fn try_migrate_server_config<S>(api: Arc<S>)
+pub async fn try_migrate_server_config<S>(api: Arc<S>, decrypt_fn: Option<crate::bucket::migration::LegacyBlobDecryptFn>)
 where
     S: ObjectIO<
             Error = Error,
@@ -1162,6 +1162,15 @@ where
         }
     };
 
+    // MinIO encrypts the server config at rest with a key derived from the root
+    // credentials. Decrypt before decoding; fall back to the raw bytes when no key
+    // applies (plaintext configs) so existing behavior holds. The decrypted bytes
+    // also become the fidelity seed for `encode_server_config_blob` below.
+    let data = match &decrypt_fn {
+        Some(decrypt) => decrypt(&data).unwrap_or(data),
+        None => data,
+    };
+
     let cfg = match decode_server_config_blob(&data) {
         Ok(v) => v,
         Err(err) => {
@@ -1810,6 +1819,30 @@ mod tests {
         assert!(kvs.0[0].hidden_if_empty);
     }
 
+    #[test]
+    fn test_encrypted_server_config_requires_decrypt_before_decode() {
+        // Reproduces the MinIO drop-in migration gap for the server config: MinIO
+        // encrypts config.json at rest, so decode fails outright. The migration's
+        // decrypt callback must run first to recover it.
+        let plaintext = r#"{"storage_class":{"_":[{"key":"standard","value":"EC:2"}]}}"#.as_bytes();
+        let ciphertext = rustfs_crypto::encrypt_data(b"root-secret-key", plaintext).expect("encrypt config blob");
+
+        // Old behavior: decode cannot parse ciphertext -> Err -> migration skipped.
+        assert!(
+            decode_server_config_blob(&ciphertext).is_err(),
+            "ciphertext must not decode as a server config"
+        );
+
+        // With the decrypt callback recovering plaintext first, decode succeeds.
+        let decrypt_fn: crate::bucket::migration::LegacyBlobDecryptFn =
+            std::sync::Arc::new(|data: &[u8]| rustfs_crypto::decrypt_data(b"root-secret-key", data).ok());
+        let recovered = decrypt_fn(&ciphertext).expect("callback should decrypt the config blob");
+        assert_eq!(recovered, plaintext);
+        let cfg = decode_server_config_blob(&recovered).expect("decode should succeed on decrypted plaintext");
+        let kvs = cfg.get_value("storage_class", "_").expect("storage_class should exist");
+        assert_eq!(kvs.0[0].value, "EC:2");
+    }
+
     #[test]
     fn test_decode_server_config_accepts_missing_hidden_if_empty() {
         let input = r#"{"storage_class":{"_":[{"key":"standard","value":"EC:2"}]}}"#;
@@ -3023,10 +3056,13 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_recovery_falls_back_to_default_config_when_blob_stays_corrupt() {
         // Heal cannot repair the blob (all shards corrupt): boot with the
         // default config instead of crash-looping (issue #4156).
         // RUSTFS_CONFIG_RECOVER_ON_CORRUPTION is unset, so the default (on) applies.
+        // `#[serial]` keeps this off-by-default read from racing the sibling test
+        // that toggles ENV_CONFIG_RECOVER_ON_CORRUPTION via a process-wide env var.
         let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(corrupt_config_blob()), None));
 
         let cfg = read_config_without_migrate_with_recovery(store.clone())

crates/ecstore/src/config/mod.rs

@@ -85,8 +85,8 @@ pub async fn init_global_config_sys(api: Arc<ECStore>) -> Result<()> {
     GLOBAL_CONFIG_SYS.init(api).await
 }
 
-pub async fn try_migrate_server_config(api: Arc<ECStore>) {
-    com::try_migrate_server_config(api).await
+pub async fn try_migrate_server_config(api: Arc<ECStore>, decrypt_fn: Option<crate::bucket::migration::LegacyBlobDecryptFn>) {
+    com::try_migrate_server_config(api, decrypt_fn).await
 }
 
 pub fn init() {

crates/ecstore/src/store/init_format.rs

@@ -28,7 +28,7 @@ use crate::{
 use futures::future::join_all;
 use rustfs_config::server_config::KVS;
 use std::collections::{HashMap, hash_map::Entry};
-use tracing::{debug, info, warn};
+use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
 pub async fn init_disks(eps: &Endpoints, opt: &DiskOption) -> (Vec<Option<DiskStore>>, Vec<Option<DiskError>>) {
@@ -74,9 +74,27 @@ pub async fn connect_load_init_formats(
     if first_disk && should_init_erasure_disks(&errs) {
         // UnformattedDisk, try migrate from MinIO format first, else create new format
         info!("first_disk && should_init_erasure_disks");
-        if let Ok(fm) = try_migrate_format(disks, set_count, set_drive_count).await {
-            info!("Migrated format from MinIO config");
-            return Ok(fm);
+        match try_migrate_format(disks, set_count, set_drive_count).await {
+            Ok(LegacyFormatOutcome::Migrated(fm)) => {
+                info!("Migrated format from MinIO config");
+                return Ok(*fm);
+            }
+            Ok(LegacyFormatOutcome::Incompatible) => {
+                // A MinIO format.json was found on disk but could not be migrated
+                // (topology/version mismatch or parse failure). Falling through to
+                // create a FRESH RustFS format changes the object placement layout,
+                // so the pre-existing MinIO objects will not be readable. Surface
+                // this loudly instead of silently discarding the legacy data.
+                error!(
+                    "Detected MinIO format.json on disk but could NOT migrate it; initializing a fresh RustFS format instead. \
+                     Existing MinIO objects will not be readable under the new format. Ensure the RustFS pool / erasure-set \
+                     topology exactly matches the original MinIO deployment, and that no stale .rustfs.sys/format.json remains."
+                );
+            }
+            Ok(LegacyFormatOutcome::None) => {}
+            Err(e) => {
+                warn!("MinIO format migration attempt failed, will initialize a fresh format: {e}");
+            }
         }
         let fm = init_format_erasure(disks, set_count, set_drive_count, deployment_id).await?;
         return Ok(fm);
@@ -152,25 +170,54 @@ async fn init_format_erasure(
     get_format_erasure_in_quorum(&fms)
 }
 
-/// Tries to migrate format
-/// Returns Ok(FormatV3) if migration succeeds, Err otherwise.
-async fn try_migrate_format(disks: &[Option<DiskStore>], set_count: usize, set_drive_count: usize) -> Result<FormatV3> {
+/// Outcome of attempting to migrate an on-disk MinIO `format.json`.
+enum LegacyFormatOutcome {
+    /// A compatible MinIO format was found and migrated into RustFS format files.
+    /// Boxed to keep the enum small (`FormatV3` is large; the others are unit variants).
+    Migrated(Box<FormatV3>),
+    /// A MinIO `format.json` was present but could not be migrated (topology /
+    /// version mismatch, or a parse failure). The caller must decide how to
+    /// proceed; creating a fresh format would leave the legacy objects unreadable.
+    Incompatible,
+    /// No MinIO `format.json` was present on any disk (a normal fresh install).
+    None,
+}
+
+/// Tries to migrate an on-disk MinIO `format.json` into RustFS format files.
+///
+/// Returns [`LegacyFormatOutcome`] describing whether a legacy format was present
+/// and, if so, whether it was compatible. `Err` is only returned for genuine IO
+/// failures while persisting the migrated format.
+async fn try_migrate_format(
+    disks: &[Option<DiskStore>],
+    set_count: usize,
+    set_drive_count: usize,
+) -> Result<LegacyFormatOutcome> {
+    let mut legacy_seen = false;
+
     for disk in disks.iter().flatten() {
         let data = match disk.read_all(MIGRATING_META_BUCKET, FORMAT_CONFIG_FILE).await {
             Ok(d) if !d.is_empty() => d,
             _ => continue,
         };
+        // A non-empty MinIO format.json exists on at least one disk.
+        legacy_seen = true;
 
-        let fm = FormatV3::try_from(data.as_ref()).map_err(|e| Error::other(format!("parse MinIO format: {e}")))?;
+        let fm = match FormatV3::try_from(data.as_ref()) {
+            Ok(fm) => fm,
+            Err(e) => {
+                warn!("failed to parse MinIO format.json, skipping this disk: {e}");
+                continue;
+            }
+        };
 
-        let first_set = fm
-            .erasure
-            .sets
-            .first()
-            .ok_or_else(|| Error::other("MinIO format: erasure.sets is empty"))?;
+        let Some(first_set) = fm.erasure.sets.first() else {
+            warn!("MinIO format.json has empty erasure.sets, skipping this disk");
+            continue;
+        };
         if fm.erasure.sets.len() != set_count || first_set.len() != set_drive_count {
-            debug!(
-                "MinIO format set count mismatch: got {}x{}, expected {}x{}",
+            warn!(
+                "MinIO format topology mismatch: got {}x{}, expected {}x{}; skipping migration for this disk",
                 fm.erasure.sets.len(),
                 first_set.len(),
                 set_count,
@@ -180,7 +227,10 @@ async fn try_migrate_format(disks: &[Option<DiskStore>], set_count: usize, set_d
         }
 
         if fm.erasure.version != FormatErasureVersion::V3 {
-            debug!("MinIO format erasure version not V3: {:?}", fm.erasure.version);
+            warn!(
+                "MinIO format erasure version is not V3 ({:?}); skipping migration for this disk",
+                fm.erasure.version
+            );
             continue;
         }
 
@@ -200,10 +250,14 @@ async fn try_migrate_format(disks: &[Option<DiskStore>], set_count: usize, set_d
         }
 
         save_format_file_all(disks, &fms).await?;
-        return get_format_erasure_in_quorum(&fms);
+        return Ok(LegacyFormatOutcome::Migrated(Box::new(get_format_erasure_in_quorum(&fms)?)));
     }
 
-    Err(Error::other("no MinIO format to migrate"))
+    Ok(if legacy_seen {
+        LegacyFormatOutcome::Incompatible
+    } else {
+        LegacyFormatOutcome::None
+    })
 }
 
 pub fn get_format_erasure_in_quorum(formats: &[Option<FormatV3>]) -> Result<FormatV3> {