diff --git a/Cargo.lock b/Cargo.lock
index f01cc8e4a..09e2b3f94 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9054,6 +9054,7 @@ dependencies = [
  "rustfs-concurrency",
  "rustfs-config",
  "rustfs-credentials",
+ "rustfs-crypto",
  "rustfs-data-usage",
  "rustfs-erasure-codec",
  "rustfs-filemeta",
diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml
index cd18d3c61..f4c5cd35e 100644
--- a/crates/ecstore/Cargo.toml
+++ b/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"] }
diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs
index 758fae75a..8e99ac21f 100644
--- a/crates/ecstore/src/api/mod.rs
+++ b/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 {
diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
index 2badf75f9..f9d3658d7 100644
--- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+++ b/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");
diff --git a/crates/ecstore/src/bucket/migration.rs b/crates/ecstore/src/bucket/migration.rs
index db300fa42..a2c6692b1 100644
--- a/crates/ecstore/src/bucket/migration.rs
+++ b/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";
diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs
index 7babeee3b..78b5671a1 100644
--- a/crates/ecstore/src/config/com.rs
+++ b/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())
diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs
index f86f0a95e..7d54da384 100644
--- a/crates/ecstore/src/config/mod.rs
+++ b/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() {
diff --git a/crates/ecstore/src/store/init_format.rs b/crates/ecstore/src/store/init_format.rs
index fccd112d5..12d3179a5 100644
--- a/crates/ecstore/src/store/init_format.rs
+++ b/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> {
diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs
index f984255bb..2197f7dfb 100644
--- a/crates/iam/src/lib.rs
+++ b/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>,
 }
diff --git a/crates/iam/src/store/object.rs b/crates/iam/src/store/object.rs
index d41834554..f90917a2a 100644
--- a/crates/iam/src/store/object.rs
+++ b/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();
diff --git a/crates/lock/src/fast_lock/shard.rs b/crates/lock/src/fast_lock/shard.rs
index fb2e56c71..9de3d10a6 100644
--- a/crates/lock/src/fast_lock/shard.rs
+++ b/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;
         }
     }
diff --git a/crates/lock/src/fast_lock/tests.rs b/crates/lock/src/fast_lock/tests.rs
index ea05d9b9b..62c062927 100644
--- a/crates/lock/src/fast_lock/tests.rs
+++ b/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();
diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs
index 89430e120..03ed6d7af 100644
--- a/rustfs/src/storage/storage_api.rs
+++ b/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) {