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>

Cargo.lock

@@ -9054,6 +9054,7 @@ dependencies = [
  "rustfs-concurrency",
  "rustfs-config",
  "rustfs-credentials",
+ "rustfs-crypto",
  "rustfs-data-usage",
  "rustfs-erasure-codec",
  "rustfs-filemeta",

crates/ecstore/Cargo.toml

@@ -148,6 +148,7 @@ opentelemetry_sdk = { workspace = true }
 proptest = "1"
 rcgen.workspace = true
 insta = { workspace = true }
+rustfs-crypto = { workspace = true }
 
 [build-dependencies]
 shadow-rs = { workspace = true, features = ["build", "metadata"] }

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

crates/iam/src/lib.rs

@@ -49,6 +49,12 @@ pub fn is_root_access_key(access_key: &str) -> bool {
     root_credentials::is_root_access_key(access_key)
 }
 
+/// Decrypts an at-rest IAM config blob using the same key sources as the IAM load
+/// path (RustFS master keys and MinIO-compatible legacy keys derived from the root
+/// credentials). Used by the MinIO -> RustFS migration path. See
+/// [`store::object::try_decrypt_iam_blob`].
+pub use store::object::try_decrypt_iam_blob;
+
 pub(crate) struct IamNotificationPeerErr {
     pub(crate) err: Option<IamEcstoreError>,
 }

crates/iam/src/store/object.rs

@@ -147,6 +147,22 @@ fn split_path(s: &str, last_index: bool) -> (&str, &str) {
     }
 }
 
+/// Attempts to decrypt an at-rest IAM config blob using the same key sources as
+/// the IAM load path: the RustFS master keys (`RUSTFS_IAM_MASTER_KEY[_OLD_KEYS]`)
+/// and the MinIO-compatible legacy keys derived from the root credentials
+/// (`secret_key` and `access_key:secret_key`).
+///
+/// Returns the plaintext when a key succeeds. When `data` is already plaintext
+/// JSON it is returned unchanged. Returns `None` if no configured key can decrypt
+/// it, so the caller can fall back to treating `data` as-is.
+///
+/// This is used by the MinIO -> RustFS migration path to decrypt IAM
+/// identity/service-account files that MinIO encrypts at rest before normalizing
+/// and re-persisting them under the RustFS system bucket.
+pub fn try_decrypt_iam_blob(data: &[u8]) -> Option<Vec<u8>> {
+    ObjectStore::decrypt_data_with_source(data).ok().map(|outcome| outcome.plain)
+}
+
 #[derive(Clone)]
 pub struct ObjectStore {
     object_api: Arc<IamStore>,
@@ -1275,6 +1291,33 @@ mod tests {
         assert_eq!(out.plain, plain);
     }
 
+    // Public helper used by the MinIO -> RustFS migration path (see
+    // ecstore::bucket::migration::try_migrate_iam_config).
+    #[test]
+    fn test_try_decrypt_iam_blob_passes_plaintext_through() {
+        let raw = br#"{"Version":1,"policy":"readonly"}"#;
+        let out = super::try_decrypt_iam_blob(raw).expect("plaintext json should pass through");
+        assert_eq!(out, raw);
+    }
+
+    #[test]
+    fn test_try_decrypt_iam_blob_decrypts_legacy_secret_encryption() {
+        // Mirrors a MinIO IAM identity file encrypted at rest with the root secret key.
+        let cred = test_cred();
+        let plain = br#"{"accessKey":"ak","secretKey":"sk"}"#;
+        let encrypted = rustfs_crypto::encrypt_data(cred.secret_key.as_bytes(), plain).expect("encrypt with rustfs secret");
+        let out = super::try_decrypt_iam_blob(&encrypted).expect("legacy-encrypted blob should decrypt");
+        assert_eq!(out, plain);
+    }
+
+    #[test]
+    fn test_try_decrypt_iam_blob_returns_none_on_undecryptable() {
+        // A blob that is neither plaintext JSON nor decryptable with any known key
+        // must return None so the migration falls back to the raw bytes (skip).
+        let garbage = [0x00u8, 0x01, 0x02, 0x03, 0x04, 0x05];
+        assert!(super::try_decrypt_iam_blob(&garbage).is_none());
+    }
+
     #[test]
     fn test_decrypt_data_corrupt_stream_io_fails() {
         let cred = test_cred();

crates/lock/src/fast_lock/shard.rs

@@ -169,6 +169,15 @@ impl LockShard {
 
         let mut retry_count = 0u32;
         const MAX_RETRIES: u32 = 10;
+        // Upper bound for a single notification wait. The notification pool is
+        // shared process-wide (a fixed set of `Notify` slots hashed by lock), so a
+        // wakeup meant for this lock can be consumed by a waiter of a different
+        // lock that hashes to the same slot, and this lock's waiter would then
+        // sleep until the full deadline. Capping each wait turns that lost-wakeup
+        // into bounded re-polling: on cap elapse we loop and re-`try_acquire`,
+        // only returning `Timeout` once the real deadline passes. The notification
+        // still delivers prompt wakeups in the common (no-collision) case.
+        const NOTIFY_WAIT_CAP: Duration = Duration::from_millis(50);
 
         loop {
             // Get or create object state
@@ -217,23 +226,25 @@ impl LockShard {
                 }
             }
 
-            // If we've exhausted quick retries or have little time left, use notification wait
+            // If we've exhausted quick retries or have little time left, use a
+            // notification wait, but bounded by NOTIFY_WAIT_CAP so a lost/stolen
+            // wakeup cannot strand this waiter until the deadline.
+            let wait = remaining.min(NOTIFY_WAIT_CAP);
             let wait_result = match request.mode {
                 LockMode::Shared => {
                     let _waiter_guard = WaiterCounterGuard::new(state.clone(), LockMode::Shared);
-                    timeout(remaining, state.optimized_notify.wait_for_read()).await
+                    timeout(wait, state.optimized_notify.wait_for_read()).await
                 }
                 LockMode::Exclusive => {
                     let _waiter_guard = WaiterCounterGuard::new(state.clone(), LockMode::Exclusive);
-                    timeout(remaining, state.optimized_notify.wait_for_write()).await
+                    timeout(wait, state.optimized_notify.wait_for_write()).await
                 }
             };
 
-            if wait_result.is_err() {
-                self.metrics.record_timeout();
-                return Err(LockResult::Timeout);
-            }
-
+            // A capped-wait elapse is not a real timeout: loop back and re-try the
+            // acquisition. The deadline check at the top of the loop is the single
+            // source of truth for returning `Timeout`.
+            let _ = wait_result;
             retry_count += 1;
         }
     }

crates/lock/src/fast_lock/tests.rs

@@ -410,6 +410,66 @@ mod fast_lock_tests {
         }
     }
 
+    // Regression for the fast-lock lost-wakeup (backlog#853 follow-up).
+    //
+    // A waiter that enters the notification wait can miss its wakeup: the release
+    // path only calls `notify_one` when `writer_waiters > 0`, so if the holder
+    // releases in the narrow gap after the waiter's `try_acquire` fails but before
+    // it registers as a waiter, no notification (and no stored permit) is produced.
+    // The waiter then blocks until the acquire deadline even though the lock is free
+    // and stays free — surfacing as a spurious `LockResult::Timeout`.
+    //
+    // Each key here has one long holder plus one waiter that starts slightly later,
+    // so the waiter is pushed past the early backoff phase into the notification
+    // wait and there is no re-contention after the single release. With the fix
+    // (bounded notification wait + re-poll) the waiter acquires within ~50ms of the
+    // release; without it, the missed wakeup strands the waiter until timeout.
+    // Many independent keys make hitting the narrow race reliable.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn write_lock_waiter_is_not_stranded_by_missed_wakeup() {
+        let manager = Arc::new(create_test_manager());
+        const KEYS: usize = 64;
+        // Long enough to push the waiter past the ~850ms backoff phase into the
+        // notification wait, where the missed-wakeup bug lives.
+        const HOLDER_HOLD: Duration = Duration::from_millis(950);
+
+        let mut handles = Vec::new();
+        for k in 0..KEYS {
+            let key = ObjectKey::new("bucket", format!("object-{k}"));
+
+            // Holder: grabs the lock immediately and holds it across the waiter's
+            // backoff-to-notification transition, then releases exactly once.
+            let holder_mgr = manager.clone();
+            let holder_key = key.clone();
+            handles.push(tokio::spawn(async move {
+                let mut guard = holder_mgr
+                    .acquire_write_lock(holder_key, "holder")
+                    .await
+                    .expect("holder should acquire immediately");
+                sleep(HOLDER_HOLD).await;
+                assert!(guard.release());
+            }));
+
+            // Waiter: starts a touch later so the holder wins the lock first, then
+            // must survive the whole hold and acquire promptly after the release.
+            let waiter_mgr = manager.clone();
+            let waiter_key = key.clone();
+            handles.push(tokio::spawn(async move {
+                sleep(Duration::from_millis(10)).await;
+                let request = ObjectLockRequest::new_write(waiter_key, "waiter").with_acquire_timeout(Duration::from_secs(5));
+                let mut guard = waiter_mgr
+                    .acquire_lock(request)
+                    .await
+                    .expect("waiter must acquire after the holder releases, not time out on a missed wakeup");
+                assert!(guard.release());
+            }));
+        }
+
+        for handle in handles {
+            handle.await.expect("lock task should not panic");
+        }
+    }
+
     #[tokio::test]
     async fn test_lock_timeout() {
         let manager = create_test_manager();

rustfs/src/storage/storage_api.rs

@@ -732,7 +732,12 @@ pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) {
 }
 
 pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) {
-    ecstore_bucket::migration::try_migrate_iam_config(store).await;
+    // 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;
 }
 
 pub(crate) fn init_ecstore_config() {
@@ -799,7 +804,13 @@ pub(crate) fn set_global_rustfs_port(value: u16) {
 }
 
 pub(crate) async fn try_migrate_server_config(store: Arc<ECStore>) {
-    ecstore_config::try_migrate_server_config(store).await;
+    // MinIO encrypts the server config at rest with a key derived from the root
+    // credentials. Inject the IAM crate's decryption (same key sources as IAM
+    // config) so an encrypted legacy config is decrypted before decoding 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_config::try_migrate_server_config(store, Some(decrypt_fn)).await;
 }
 
 pub(crate) async fn update_erasure_type(setup_type: SetupType) {