Merge origin/main into improve/base

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/upgrade_compatibility_test.rs

@@ -2345,7 +2345,7 @@ fn layout_reported_parts(attributes: &aws_sdk_s3::operation::get_object_attribut
     attributes.object_parts().map(|parts| parts.parts().len())
 }
 
-async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
+async fn assert_layout_full_body(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
     let label = case.label();
     let head = layout_head(client, case).await?;
     assert_eq!(
@@ -2362,7 +2362,12 @@ async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &st
     let full = layout_get(client, case, None, None).await?.body.collect().await?.into_bytes();
     assert_eq!(full.len(), case.body.len(), "{context}: {label}: full GET length");
     assert!(full == case.body, "{context}: {label}: full GET body must equal the rc.5 upload");
+    Ok(())
+}
 
+async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
+    assert_layout_full_body(client, case, context).await?;
+    let label = case.label();
     if case.is_multipart_layout() {
         let first = case.part_sizes[0];
         let range = format!("bytes={}-{}", first - 32, first + 31);
@@ -2641,10 +2646,11 @@ async fn direct_upgrade_from_rc5_preserves_multipart_layouts() -> TestResult {
             "rc.5 wrote a legacy layout"
         );
     }
-    // The rc.5 writer must itself still read what it wrote, so a later
-    // failure is attributable to the upgrade rather than to the fixture.
+    // Prove the rc.5 writer persisted every original byte before upgrading.
+    // Its compressed cross-part range reader has a legacy limitation; the
+    // upgraded reader must still satisfy every range and partNumber assertion.
     for case in &cases {
-        assert_layout_readable(&old_client, case, "rc.5").await?;
+        assert_layout_full_body(&old_client, case, "rc.5").await?;
     }
 
     // Upgrade in place.

crates/ecstore/src/api/mod.rs

@@ -199,10 +199,10 @@ pub mod bucket {
             get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config,
             get_on_demand_migration_config_in, get_public_access_block_config, get_quota_config, get_replication_config,
             get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config,
-            init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
-            update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
-            update_if_incarnation_at, update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock,
-            update_under_transaction_lock_at,
+            init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata,
+            remove_bucket_metadata_if_absent, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock,
+            update_config_with, update_if_incarnation, update_if_incarnation_at, update_quota_if_incarnation,
+            update_quota_if_incarnation_at, update_under_transaction_lock, update_under_transaction_lock_at,
         };
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::{ConfigWriteLockProbe, test_support};

crates/ecstore/src/bucket/metadata_sys.rs

@@ -321,6 +321,33 @@ pub async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
     Ok(lock.remove(bucket).await)
 }
 
+/// Remove a peer's cached metadata only when the bucket is absent under the
+/// namespace read fence. A delayed delete notification must not withdraw the
+/// metadata of a same-name bucket that has already been recreated.
+pub async fn remove_bucket_metadata_if_absent(api: Arc<ECStore>, bucket: &str) -> Result<bool> {
+    if is_meta_bucketname(bucket) {
+        return Ok(false);
+    }
+    // Lock order: bucket namespace read -> metadata system read (released)
+    // -> metadata publish, matching peer reloads.
+    let namespace_lock = api.new_ns_lock(bucket, bucket).await?;
+    let namespace_guard = namespace_lock
+        .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await?;
+    let sys = bucket_metadata_sys_of(&api.ctx)?;
+    let sys = sys.read().await.clone();
+    if sys
+        .bucket_exists(bucket, &namespace_guard, "peer bucket metadata removal existence check")
+        .await?
+    {
+        return Ok(false);
+    }
+    await_bucket_namespace_operation(Some(&namespace_guard), bucket, "peer bucket metadata removal", async {
+        Ok(sys.remove(bucket).await)
+    })
+    .await
+}
+
 fn start_refresh_buckets_metadata_loop(sys: Arc<RwLock<BucketMetadataSys>>) {
     let Some(cancel_token) = runtime_sources::background_services_cancel_token() else {
         warn!("bucket metadata refresh loop skipped because background cancellation token is not initialized");
@@ -4933,6 +4960,117 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn peer_reload_clears_negative_cache_after_remote_bucket_creation() {
+        let (dirs, node_a, node_b) = two_nodes_over_one_store().await;
+        let bucket = "cross-node-negative-cache-refresh";
+        let node_b_sys = node_b.read().await.clone();
+
+        assert_eq!(
+            node_b_sys.get_bucket_incarnation_id(bucket).await.unwrap_err(),
+            Error::BucketNotFound(bucket.to_string()),
+            "the probe must populate node B's missing-bucket cache"
+        );
+
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist before creation metadata");
+        }
+        let created = BucketMetadata::new(bucket);
+        let created_incarnation = created.bucket_incarnation_id;
+        node_a
+            .read()
+            .await
+            .persist_new_and_set(created)
+            .await
+            .expect("node A should publish the created bucket metadata");
+
+        node_b_sys
+            .reload_from_store(bucket)
+            .await
+            .expect("peer reload should clear the stale negative cache");
+        assert_eq!(
+            node_b_sys.get_bucket_incarnation_id(bucket).await.unwrap(),
+            created_incarnation,
+            "node B should resolve the newly created bucket immediately"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn peer_delete_clears_metadata_for_absent_bucket() {
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let sys = bucket_metadata_sys_of(&store.ctx).expect("metadata owner");
+        let bucket = "peer-delete-absent-bucket";
+
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("initial bucket creation");
+        let cached = sys.read().await.get(bucket).await.expect("initial cached metadata");
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("bucket deletion");
+
+        // Model a peer that still caches the metadata after the physical delete.
+        sys.read().await.set(bucket.to_string(), cached).await;
+        assert!(
+            remove_bucket_metadata_if_absent(store.clone(), bucket)
+                .await
+                .expect("conditional peer delete"),
+            "the peer must evict the deleted bucket's cached metadata"
+        );
+        let sys = sys.read().await;
+        assert!(matches!(sys.get(bucket).await, Err(Error::ConfigNotFound)));
+        assert_eq!(
+            sys.get_bucket_incarnation_id(bucket)
+                .await
+                .expect_err("deleted bucket lookup"),
+            Error::BucketNotFound(bucket.to_string()),
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn delayed_peer_delete_preserves_recreated_bucket_metadata() {
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let sys = bucket_metadata_sys_of(&store.ctx).expect("metadata owner");
+        let bucket = "delayed-peer-delete-recreated-bucket";
+
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("initial bucket creation");
+        let old_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("old incarnation");
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("bucket deletion");
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket recreation");
+        let new_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("new incarnation");
+        assert_ne!(old_incarnation, new_incarnation);
+
+        let removed = remove_bucket_metadata_if_absent(store.clone(), bucket)
+            .await
+            .expect("conditional peer delete");
+        assert!(!removed, "a delayed delete must not remove metadata for a live recreated bucket");
+        assert_eq!(
+            sys.read()
+                .await
+                .get(bucket)
+                .await
+                .expect("recreated metadata should remain cached")
+                .bucket_incarnation_id,
+            new_incarnation,
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]
     async fn stale_config_request_cannot_mutate_a_recreated_bucket() {

crates/ecstore/src/disk/local.rs

@@ -19169,6 +19169,326 @@ mod test {
         }]
     }
 
+    // This is a real process interruption of LocalDisk, not a server or
+    // power-loss test. The child stops at the rollback-backup unlink.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn quorum_delete_cleanup_process_crash_preserves_retry_evidence() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+        use std::os::unix::process::ExitStatusExt;
+
+        const CHILD_ROOT: &str = "RUSTFS_TEST_QUORUM_CLEANUP_ROOT";
+        const CHILD_TRANSACTION: &str = "RUSTFS_TEST_QUORUM_CLEANUP_TRANSACTION";
+        const CHILD_READY: &str = "RUSTFS_TEST_QUORUM_CLEANUP_READY";
+        if let Some(root) = std::env::var_os(CHILD_ROOT) {
+            let endpoint =
+                Endpoint::try_from(Path::new(&root).to_str().expect("UTF-8 child disk path")).expect("parse child endpoint");
+            let disk = LocalDisk::new(&endpoint, false).await.expect("open child disk");
+            let transaction =
+                Uuid::parse_str(&std::env::var(CHILD_TRANSACTION).expect("child transaction")).expect("parse child transaction");
+            let ready = std::env::var_os(CHILD_READY).expect("child readiness path");
+            let rollback_path = format!("object/{transaction}");
+            let backup = disk
+                .io_get_object_path("bucket", &rollback_path)
+                .expect("child transaction path")
+                .join(STORAGE_FORMAT_FILE_BACKUP);
+            let _hook = hooks::install_at(hooks::Stage::Remove, &backup, move || {
+                std::fs::write(ready, b"before-backup-unlink").expect("publish reached boundary");
+                loop {
+                    std::thread::park();
+                }
+            });
+            disk.delete(
+                "bucket",
+                &rollback_path,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    old_data_dir: Some(transaction),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("child cleanup should reach its boundary");
+            panic!("child unexpectedly returned without being killed");
+        }
+
+        let dir = tempfile::tempdir().expect("create test disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
+        ensure_test_volume(&disk, "bucket").await;
+        let residue = write_committed_delete_residue(&disk, "bucket", "object").await;
+        let original_backup = fs::read(residue.backup()).await.expect("read original backup");
+        let original_part = fs::read(residue.part()).await.expect("read original part");
+
+        for attempt in 0..2 {
+            let ready = dir.path().join(format!("cleanup-ready-{attempt}"));
+            let mut child = std::process::Command::new(std::env::current_exe().expect("current test binary"))
+                .args([
+                    "--exact",
+                    "disk::local::test::quorum_delete_cleanup_process_crash_preserves_retry_evidence",
+                    "--nocapture",
+                ])
+                .env(CHILD_ROOT, dir.path())
+                .env(CHILD_TRANSACTION, residue.transaction.to_string())
+                .env(CHILD_READY, &ready)
+                .spawn()
+                .expect("spawn this test's cleanup process");
+            let reached = tokio::time::timeout(Duration::from_secs(30), async {
+                while !ready.exists() {
+                    if child.try_wait().expect("check child status").is_some() {
+                        return false;
+                    }
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+                true
+            })
+            .await;
+            // Child is an owned process handle; no process-name matching or
+            // external PID is accepted by this test.
+            if child.try_wait().expect("check owned child").is_none() {
+                child.kill().expect("SIGKILL the owned cleanup child");
+            }
+            let status = child.wait().expect("reap cleanup child");
+            assert!(matches!(reached, Ok(true)), "child must reach the exact cleanup boundary");
+            assert_eq!(status.signal(), Some(9), "child must have been interrupted by SIGKILL");
+            println!(
+                "cleanup crash {attempt}: backup={} marker={} part={}",
+                residue.backup().exists(),
+                residue.marker().exists(),
+                residue.part().exists()
+            );
+            assert_eq!(fs::read(residue.backup()).await.expect("backup at paused unlink"), original_backup);
+            assert_eq!(fs::read(residue.part()).await.expect("payload survives each crash"), original_part);
+            assert!(residue.marker().is_file(), "transaction marker survives each crash");
+            assert!(
+                residue
+                    .object_dir
+                    .join(residue.transaction.to_string())
+                    .join(DELETE_QUORUM_COMMITTED_FILE)
+                    .is_file(),
+                "quorum receipt survives each crash"
+            );
+        }
+
+        let reopened = LocalDisk::new(&endpoint, false)
+            .await
+            .expect("reopen disk after process interruption");
+        reopened
+            .delete(
+                "bucket",
+                &format!("object/{}", residue.transaction),
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    old_data_dir: Some(residue.transaction),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("retry the already committed coordinator transaction");
+        assert_eq!(original_part, b"stale");
+        let bucket_result = reopened.delete_volume("bucket", false).await;
+        println!("non-force bucket deletion after retry: {bucket_result:?}");
+        assert!(
+            !residue.object_dir.exists(),
+            "repeated delete after two crashes must reclaim this known-successful transaction"
+        );
+        bucket_result.expect("non-force bucket deletion after retry");
+    }
+
+    // Normal post-quorum cleanup must preserve its data-dir marker until the
+    // rollback backup is gone, including across an interrupted retry.
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn quorum_delete_cleanup_removes_backup_before_data_and_can_retry() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+
+        let dir = tempfile::tempdir().expect("create test disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
+        let bucket = "bucket";
+        let object = "object";
+        ensure_test_volume(&disk, bucket).await;
+        let residue = write_committed_delete_residue(&disk, bucket, object).await;
+        let data_path = residue.part();
+        let rollback_path = format!("{object}/{}", residue.transaction);
+        let opts = DeleteOptions {
+            recursive: true,
+            immediate: true,
+            old_data_dir: Some(residue.transaction),
+            ..Default::default()
+        };
+        let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+        let (release, release_rx) = std::sync::mpsc::channel::<()>();
+        let _hook = hooks::install_at(hooks::Stage::Remove, &data_path, move || {
+            let _ = entered_tx.send(());
+            let _ = release_rx.recv();
+        });
+        let mut cleanup = Box::pin(disk.delete(bucket, &rollback_path, opts.clone()));
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                entered = entered_rx => entered.expect("cleanup must reach data removal"),
+                result = cleanup.as_mut() => panic!("cleanup ended before data removal: {result:?}"),
+            }
+        })
+        .await
+        .expect("cleanup reaches physical payload unlink");
+
+        let backup_gone = !residue.backup().exists();
+        let marker_present = residue.marker().exists();
+        let original_part = fs::read(residue.part()).await.expect("paused data must remain readable");
+        drop(cleanup);
+
+        // The in-flight unlink retains metadata ownership after cancellation.
+        // Retry must wait for that physical operation before removing proof.
+        let mut retry = Box::pin(disk.delete(bucket, &rollback_path, opts));
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), retry.as_mut())
+                .await
+                .is_err(),
+            "retry cannot bypass the parked cleanup's namespace ownership"
+        );
+        assert!(residue.marker().is_file());
+        assert!(
+            residue
+                .object_dir
+                .join(residue.transaction.to_string())
+                .join(DELETE_QUORUM_COMMITTED_FILE)
+                .is_file()
+        );
+        drop(release);
+        tokio::time::timeout(Duration::from_secs(10), retry)
+            .await
+            .expect("retry completes after the abandoned unlink releases ownership")
+            .expect("retry cleanup");
+
+        assert!(
+            backup_gone,
+            "post-quorum cleanup must remove the backup before touching its data-dir evidence"
+        );
+        assert!(marker_present, "data removal must still have its transaction marker");
+        assert_eq!(original_part, b"stale");
+        assert!(!residue.object_dir.exists(), "interrupted cleanup and retry must leave no object residue");
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn quorum_delete_cleanup_backup_error_preserves_data_and_marker() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+
+        let dir = tempfile::tempdir().expect("create test disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
+        let bucket = "bucket";
+        let object = "object";
+        ensure_test_volume(&disk, bucket).await;
+        let residue = write_committed_delete_residue(&disk, bucket, object).await;
+        let backup_bytes = fs::read(residue.backup()).await.expect("read rollback evidence");
+        let backup = residue.backup();
+        let held_backup = residue.object_dir.join("held-backup");
+        let blocked_backup = backup.clone();
+        let retained_backup = held_backup.clone();
+        // Replace only the classified backup inode with a directory so the
+        // real unlink fails, while retaining its exact bytes for retry.
+        let _hook = hooks::install_at(hooks::Stage::Remove, &backup, move || {
+            std::fs::rename(&blocked_backup, retained_backup).expect("retain rollback bytes");
+            std::fs::create_dir(blocked_backup).expect("block the actual backup unlink");
+        });
+        let rollback_path = format!("{object}/{}", residue.transaction);
+        let opts = DeleteOptions {
+            recursive: true,
+            immediate: true,
+            old_data_dir: Some(residue.transaction),
+            ..Default::default()
+        };
+        disk.delete(bucket, &rollback_path, opts.clone())
+            .await
+            .expect_err("backup removal failure must be returned");
+
+        assert!(backup.is_dir(), "the native backup unlink must have encountered the blocker");
+        assert_eq!(fs::read(&held_backup).await.expect("backup bytes must survive"), backup_bytes);
+        assert!(residue.marker().exists(), "backup failure must retain the data-dir transaction marker");
+        assert_eq!(fs::read(residue.part()).await.expect("payload must survive backup failure"), b"stale");
+
+        fs::remove_dir(&backup).await.expect("remove the one-shot unlink blocker");
+        fs::rename(&held_backup, &backup)
+            .await
+            .expect("restore the unchanged backup inode");
+        assert_eq!(fs::read(&backup).await.expect("canonical rollback backup"), backup_bytes);
+        disk.delete(bucket, &rollback_path, opts)
+            .await
+            .expect("retry cleanup after one-shot unlink failure");
+        assert!(!residue.object_dir.exists(), "successful retry must reclaim the transaction");
+    }
+
+    #[tokio::test]
+    async fn quorum_delete_cleanup_preserves_other_generations_transactions_and_children() {
+        let dir = tempfile::tempdir().expect("create test disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
+        let bucket = "bucket";
+        let object = "object";
+        ensure_test_volume(&disk, bucket).await;
+        let residue = write_committed_delete_residue(&disk, bucket, object).await;
+        let newer_data_dir = Uuid::new_v4();
+        let newer_data = residue.object_dir.join(newer_data_dir.to_string());
+        fs::create_dir_all(&newer_data).await.expect("create newer data directory");
+        fs::write(newer_data.join("part.1"), b"newer payload")
+            .await
+            .expect("write newer payload");
+        let metadata = test_meta(test_file_info(object, Uuid::new_v4(), Some(newer_data_dir), None));
+        fs::write(residue.object_dir.join(STORAGE_FORMAT_FILE), &metadata)
+            .await
+            .expect("publish newer metadata before old transaction cleanup");
+        let unrelated = Uuid::new_v4();
+        let unrelated_backup = residue
+            .object_dir
+            .join(unrelated.to_string())
+            .join(STORAGE_FORMAT_FILE_BACKUP);
+        fs::create_dir_all(unrelated_backup.parent().expect("backup parent"))
+            .await
+            .expect("create unrelated transaction");
+        fs::write(&unrelated_backup, b"corrupt unproven backup")
+            .await
+            .expect("write ambiguous backup");
+        let mismatched_marker = newer_data.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{unrelated}"));
+        fs::write(&mismatched_marker, []).await.expect("write unrelated marker");
+        let child = residue.object_dir.join("child").join(STORAGE_FORMAT_FILE);
+        fs::create_dir_all(child.parent().expect("child parent"))
+            .await
+            .expect("create descendant");
+        fs::write(&child, b"child metadata").await.expect("write descendant metadata");
+
+        disk.delete(
+            bucket,
+            &format!("{object}/{}", residue.transaction),
+            DeleteOptions {
+                recursive: true,
+                immediate: true,
+                old_data_dir: Some(residue.transaction),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("clean only the successful transaction");
+
+        assert!(!residue.backup().exists() && !residue.part().exists() && !residue.marker().exists());
+        assert_eq!(
+            fs::read(residue.object_dir.join(STORAGE_FORMAT_FILE))
+                .await
+                .expect("read newer metadata"),
+            metadata
+        );
+        assert_eq!(fs::read(newer_data.join("part.1")).await.expect("read newer payload"), b"newer payload");
+        assert_eq!(
+            fs::read(&unrelated_backup).await.expect("read unproven backup"),
+            b"corrupt unproven backup"
+        );
+        assert!(mismatched_marker.exists());
+        assert_eq!(fs::read(child).await.expect("read descendant"), b"child metadata");
+    }
+
     // Internal maintenance retains its legacy cleanup behavior while user
     // buckets require the separate set-wide authorization boundary.
     #[tokio::test]

crates/ecstore/src/set_disk/ops/object.rs

@@ -23931,7 +23931,7 @@ mod delete_objects_lock_gating_tests {
         }
         put_plain_object(&set_disks, bucket, "obj-a").await;
 
-        let _outer_guard = set_disks
+        let outer_guard = set_disks
             .new_ns_lock(bucket, "obj-a")
             .await
             .expect("namespace lock should be created")
@@ -23944,25 +23944,55 @@ mod delete_objects_lock_gating_tests {
             ..Default::default()
         }];
 
-        let (deleted, errs) = tokio::time::timeout(
-            Duration::from_secs(1),
-            set_disks.delete_objects(
-                bucket,
-                objects,
-                ObjectOptions {
-                    no_lock: true,
-                    object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
-                        ObjectLockConfigState::ConfirmedAbsent,
-                    ))),
-                    ..Default::default()
-                },
-            ),
-        )
-        .await
-        .expect("no_lock batch delete path must not wait for the outer lock");
+        let lock_manager = set_disks
+            .local_lock_manager
+            .as_fast_lock_manager()
+            .expect("isolated fixture should use the local lock manager");
+        let lock_metrics = lock_manager.get_metrics().shard_metrics;
+        assert_eq!(lock_manager.current_lock_counts(), (1, 0));
+        let mut delete = Box::pin(set_disks.delete_objects(
+            bucket,
+            objects,
+            ObjectOptions {
+                no_lock: true,
+                object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                    ObjectLockConfigState::ConfirmedAbsent,
+                ))),
+                ..Default::default()
+            },
+        ));
+        // Observe lock attempts on every poll: durable disk work can take more
+        // than a second on CI without waiting for the namespace lock at all.
+        let deletion = futures::future::poll_fn(|cx| {
+            let result = std::future::Future::poll(delete.as_mut(), cx);
+            assert_eq!(
+                lock_manager.current_lock_counts(),
+                (1, 0),
+                "no_lock batch delete must preserve the outer holder without adding lock waiters"
+            );
+            let current = lock_manager.get_metrics().shard_metrics;
+            assert_eq!(
+                (current.total_acquisitions(), current.timeouts),
+                (lock_metrics.total_acquisitions(), lock_metrics.timeouts),
+                "no_lock batch delete must not acquire or time out on another namespace lock"
+            );
+            result
+        });
+        let (deleted, errs) = tokio::time::timeout(Duration::from_secs(30), deletion)
+            .await
+            .expect("no_lock batch delete disk work should complete");
 
         assert!(errs[0].is_none(), "no_lock batch delete should not fail with a lock error: {:?}", errs[0]);
         assert!(deleted[0].found, "existing object must still be deleted");
+        for disk in &disk_stores {
+            assert_eq!(
+                disk.read_version("", bucket, "obj-a", "", &crate::disk::ReadOptions::default())
+                    .await
+                    .expect_err("every drive must confirm the deleted version is absent"),
+                DiskError::FileNotFound
+            );
+        }
+        drop(outer_guard);
     }
 }
 
@@ -24575,6 +24605,88 @@ mod single_delete_namespace_owner_tests {
         .expect("seed a complete real object version");
     }
 
+    #[cfg(not(windows))]
+    #[tokio::test]
+    async fn quorum_delete_cleanup_coordinator_binds_single_and_batch() {
+        use crate::disk::os::prepared_publication_test_hooks as hooks;
+
+        for batch in [false, true] {
+            let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+            let bucket = "quorum-cleanup-order";
+            let object = "object";
+            let version = Uuid::new_v4();
+            for disk in &disks {
+                disk.make_volume(bucket).await.expect("create fixture bucket");
+            }
+            seed_version(&set, bucket, object, version, &vec![0x51; 256 * 1024]).await;
+            let (observed, observations) = std::sync::mpsc::channel();
+            let mut hooks = Vec::new();
+            for disk in &disks {
+                let fi = disk
+                    .read_version("", bucket, object, &version.to_string(), &ReadOptions::default())
+                    .await
+                    .expect("read seeded version");
+                assert!(!fi.inline_data(), "fixture must reach physical data cleanup");
+                let crate::disk::Disk::Local(local) = disk.as_ref() else {
+                    panic!("local fixture required");
+                };
+                let object_dir = local.get_disk().get_object_path_for_io(bucket, object).expect("object path");
+                let data_dir = object_dir.join(fi.data_dir.expect("seeded data directory").to_string());
+                let observed = observed.clone();
+                hooks.push(hooks::install_at(hooks::Stage::Remove, &data_dir.join("part.1"), move || {
+                    let backup_present = std::fs::read_dir(&object_dir)
+                        .expect("object exists before payload unlink")
+                        .any(|entry| {
+                            entry
+                                .expect("read transaction entry")
+                                .path()
+                                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
+                                .exists()
+                        });
+                    observed.send(backup_present).expect("record actual cleanup order");
+                }));
+            }
+            if batch {
+                let (_, errors) = set
+                    .delete_objects(
+                        bucket,
+                        vec![ObjectToDelete {
+                            object_name: object.to_string(),
+                            version_id: Some(version),
+                            ..Default::default()
+                        }],
+                        ObjectOptions {
+                            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                                ObjectLockConfigState::ConfirmedAbsent,
+                            ))),
+                            ..Default::default()
+                        },
+                    )
+                    .await;
+                assert!(errors.iter().all(Option::is_none), "batch delete must commit: {errors:?}");
+            } else {
+                set.delete_object_version(
+                    bucket,
+                    object,
+                    &FileInfo {
+                        name: object.to_string(),
+                        version_id: Some(version),
+                        ..Default::default()
+                    },
+                    false,
+                )
+                .await
+                .expect("single delete must commit");
+            }
+            let observations = observations.try_iter().collect::<Vec<_>>();
+            assert_eq!(observations.len(), 4, "all four disks must reach cleanup: batch={batch}");
+            assert!(
+                observations.iter().all(|backup_present| !backup_present),
+                "the coordinator must bind post-quorum cleanup before any payload unlink: batch={batch}"
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn internal_metadata_mutations_do_not_reenter_pool_metadata() {

crates/ecstore/src/store/init.rs

@@ -9594,8 +9594,18 @@ mod tests {
     #[cfg(feature = "test-util")]
     async fn decommission_migrates_and_verifies_registered_durable_ilm_records_scenario() {
         let temp_dir = tempfile::tempdir().expect("create temp store dir");
-        let (ctx, store, _shutdown) =
-            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-decommission", &[4, 4])).await;
+        // The explicit recovery below must be the only journal consumer,
+        // including when receipt validation outlasts the background interval.
+        let mut instance_ctx = crate::runtime::instance::InstanceContext::new();
+        instance_ctx.suppress_tier_delete_journal_recovery_for_test();
+        let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store_with_layout(
+            temp_dir.path(),
+            "durable-ilm-decommission",
+            &[(1, 4), (1, 4)],
+            CancellationToken::new(),
+            Some(Arc::new(instance_ctx)),
+        ))
+        .await;
 
         let tier_name = "DECOMMISSION-ILM";
         let backend = register_transition_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
@@ -10067,6 +10077,16 @@ mod tests {
         .await
         .expect("manual task receipt should restore after the corruption check");
 
+        assert!(matches!(
+            com::read_config(store.pools[0].clone(), &tier_path).await,
+            Err(Error::ConfigNotFound)
+        ));
+        assert_eq!(
+            com::read_config(store.pools[1].clone(), &tier_path)
+                .await
+                .expect("manual recovery must still observe the exact migrated tier journal"),
+            tier_bytes
+        );
         let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
             .await
             .expect("tier journal recovery should consume the migrated record before completion");

crates/iam/src/oidc/provider.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 use super::{
     config::{OidcProviderConfig, OidcProviderValidationResult, SourcedOidcProviderConfig},
-    transport::{OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs},
+    transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics},
 };
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
 use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
@@ -25,7 +25,7 @@ use std::collections::HashMap;
 use std::sync::RwLock;
 use std::time::{Duration as StdDuration, Instant};
 use tokio::time::sleep;
-use tracing::warn;
+use tracing::{debug, warn};
 use url::Url;
 
 pub(super) const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
@@ -34,6 +34,55 @@ pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration
 pub(super) const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy";
 pub(super) const OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY: &str = "OIDC provider discovery blocked by outbound policy";
 
+/// Preserve recovery eligibility before discovery errors become public diagnostic strings.
+#[derive(Debug, thiserror::Error)]
+#[error("{message}")]
+pub(super) struct DiscoveryFailure {
+    message: String,
+    pub(super) retryable: bool,
+    pub(super) diagnostics: (&'static str, String),
+}
+
+impl From<String> for DiscoveryFailure {
+    fn from(message: String) -> Self {
+        Self {
+            message,
+            retryable: false,
+            diagnostics: ("validation", String::new()),
+        }
+    }
+}
+
+fn retryable_discovery_status(status: http::StatusCode) -> bool {
+    status.is_server_error() || matches!(status, http::StatusCode::REQUEST_TIMEOUT | http::StatusCode::TOO_MANY_REQUESTS)
+}
+
+fn retryable_discovery_transport(error: &OidcHttpError) -> bool {
+    matches!(error, OidcHttpError::Reqwest(error) if !error.is_builder())
+}
+
+fn discovery_failure(error: DiscoveryError<OidcHttpError>, context: &str, blocked_context: &str) -> DiscoveryFailure {
+    let retryable = match &error {
+        DiscoveryError::Request(error) => retryable_discovery_transport(error),
+        DiscoveryError::Response(status, _, _) => retryable_discovery_status(*status),
+        _ => false,
+    };
+    let diagnostics = match &error {
+        DiscoveryError::Request(error) => oidc_http_error_diagnostics(error),
+        DiscoveryError::Response(status, _, _) => ("http_status", status.as_u16().to_string()),
+        _ => ("validation", String::new()),
+    };
+    let message = match error {
+        DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => format!("{blocked_context}: {reason}"),
+        error => format!("{context}: {error}"),
+    };
+    DiscoveryFailure {
+        message,
+        retryable,
+        diagnostics,
+    }
+}
+
 // ---- Internal provider state ----
 
 /// Discovered OIDC provider metadata.
@@ -180,6 +229,7 @@ impl<'c> AsyncHttpClient<'c> for JwksAcceptClient<'_> {
                 http::HeaderValue::from_static("application/json, application/jwk-set+json"),
             );
         }
+        request.extensions_mut().insert(DiscoveryRequest);
         self.inner.call(request)
     }
 }
@@ -246,7 +296,17 @@ impl ProviderRuntime {
         provider_id: &str,
         config: &OidcProviderConfig,
     ) -> Result<ProviderState, String> {
-        let state = discover_provider(config, &self.http_client).await?;
+        self.recover_provider_state(provider_id, config)
+            .await
+            .map_err(|err| err.to_string())
+    }
+
+    pub(super) async fn recover_provider_state(
+        &self,
+        provider_id: &str,
+        config: &OidcProviderConfig,
+    ) -> Result<ProviderState, DiscoveryFailure> {
+        let state = discover_provider_for_recovery(config, &self.http_client).await?;
         let mut map = self.states.write().map_err(|e| {
             let msg = e.to_string();
             format!("provider state lock poisoned: {msg}")
@@ -301,6 +361,15 @@ pub(super) async fn discover_provider(
     config: &OidcProviderConfig,
     http_client: &ReqwestHttpClient,
 ) -> Result<ProviderState, String> {
+    discover_provider_for_recovery(config, http_client)
+        .await
+        .map_err(|err| err.to_string())
+}
+
+pub(super) async fn discover_provider_for_recovery(
+    config: &OidcProviderConfig,
+    http_client: &ReqwestHttpClient,
+) -> Result<ProviderState, DiscoveryFailure> {
     if let Some(issuer) = config.issuer.as_deref().filter(|issuer| !issuer.trim().is_empty()) {
         return discover_provider_from_config_url(config, issuer, http_client).await;
     }
@@ -310,6 +379,8 @@ pub(super) async fn discover_provider(
     let base_issuer = normalize_config_url(&config.config_url)?;
     let candidates = issuer_candidates(&base_issuer);
     let mut last_errors = Vec::new();
+    let mut retryable = false;
+    let mut diagnostics = ("validation", String::new());
 
     for candidate_issuer in candidates.iter() {
         let issuer_url = IssuerUrl::new(candidate_issuer.clone()).map_err(|e| format!("invalid issuer URL: {e}"))?;
@@ -332,26 +403,21 @@ pub(super) async fn discover_provider(
                         metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)),
                         discovered_at: Instant::now(),
                     })
-                    .map_err(|err| match err {
-                        DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => {
-                            format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")
-                        }
-                        err => format!("discovery failed: {err}"),
-                    })
+                    .map_err(|err| discovery_failure(err, "discovery failed", OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY))
             };
             match discovered {
                 Ok(state) => return Ok(state),
                 Err(error)
-                    if error.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY)
-                        || error.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) =>
+                    if error.message.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY)
+                        || error.message.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) =>
                 {
                     return Err(error);
                 }
                 Err(error) => {
-                    let is_transient_transport = error.contains("Request failed");
+                    let is_transient_transport = error.message.contains("Request failed");
                     let should_retry = is_transient_transport && attempt + 1 < OIDC_DISCOVERY_TRANSPORT_RETRIES;
                     if should_retry {
-                        warn!(
+                        debug!(
                             event = EVENT_OIDC_DIAGNOSTICS,
                             component = LOG_COMPONENT_IAM,
                             subsystem = LOG_SUBSYSTEM_OIDC,
@@ -368,8 +434,12 @@ pub(super) async fn discover_provider(
                         continue;
                     }
 
+                    if error.retryable || !retryable {
+                        diagnostics = error.diagnostics.clone();
+                    }
+                    retryable |= error.retryable;
                     last_errors.push(format!("issuer '{candidate_issuer}': {error}"));
-                    warn!(
+                    debug!(
                         event = EVENT_OIDC_DIAGNOSTICS,
                         component = LOG_COMPONENT_IAM,
                         subsystem = LOG_SUBSYSTEM_OIDC,
@@ -388,11 +458,11 @@ pub(super) async fn discover_provider(
         }
     }
 
-    Err(format!(
-        "discovery failed for all issuer variants {:?}: {}",
-        candidates,
-        last_errors.join("; ")
-    ))
+    Err(DiscoveryFailure {
+        message: format!("discovery failed for all issuer variants {:?}: {}", candidates, last_errors.join("; ")),
+        retryable,
+        diagnostics,
+    })
 }
 
 pub(super) async fn validate_oidc_provider_config(config: &OidcProviderConfig) -> Result<OidcProviderValidationResult, String> {
@@ -429,7 +499,7 @@ async fn discover_provider_from_config_url(
     config: &OidcProviderConfig,
     issuer: &str,
     http_client: &ReqwestHttpClient,
-) -> Result<ProviderState, String> {
+) -> Result<ProviderState, DiscoveryFailure> {
     let issuer_url = IssuerUrl::new(issuer.trim().to_string()).map_err(|e| format!("invalid issuer URL: {e}"))?;
     let explicit_issuer = config.issuer.as_deref().is_some_and(|issuer| !issuer.trim().is_empty());
     let discovery_url = if explicit_issuer {
@@ -440,6 +510,7 @@ async fn discover_provider_from_config_url(
             .map_err(|err| err.to_string())?
     };
     let request = http::Request::builder()
+        .extension(DiscoveryRequest)
         .uri(discovery_url.to_string())
         .method(http::Method::GET)
         .header(http::header::ACCEPT, "application/json")
@@ -449,12 +520,22 @@ async fn discover_provider_from_config_url(
     let response = match http_client.call(request).await {
         Ok(response) => response,
         Err(OidcHttpError::ForbiddenOutbound(reason)) => {
-            return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
+            return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}").into());
+        }
+        Err(err) => {
+            return Err(DiscoveryFailure {
+                retryable: retryable_discovery_transport(&err),
+                diagnostics: oidc_http_error_diagnostics(&err),
+                message: format!("discovery request failed: Request failed: {err}"),
+            });
         }
-        Err(err) => return Err(format!("discovery request failed: Request failed: {err}")),
     };
     if response.status() != http::StatusCode::OK {
-        return Err(format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url));
+        return Err(DiscoveryFailure {
+            retryable: retryable_discovery_status(response.status()),
+            diagnostics: ("http_status", response.status().as_u16().to_string()),
+            message: format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url),
+        });
     }
 
     if !explicit_issuer
@@ -466,7 +547,9 @@ async fn discover_provider_from_config_url(
                 .is_some_and(|essence| essence.eq_ignore_ascii_case("application/json"))
         })
     {
-        return Err("Unexpected response Content-Type: expected application/json".into());
+        return Err("Unexpected response Content-Type: expected application/json"
+            .to_string()
+            .into());
     }
 
     let provider_metadata = DiscoveredProviderMetadata::parse(response.body(), config.hide_from_ui)
@@ -476,7 +559,8 @@ async fn discover_provider_from_config_url(
             "unexpected issuer URI `{}` (expected `{}`)",
             provider_metadata.issuer().as_str(),
             issuer_url.as_str()
-        ));
+        )
+        .into());
     }
 
     let jwks_url = if explicit_issuer {
@@ -494,10 +578,7 @@ async fn discover_provider_from_config_url(
     .await
     {
         Ok(jwks) => jwks,
-        Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => {
-            return Err(format!("{OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
-        }
-        Err(err) => return Err(format!("failed to fetch JWKS: {err}")),
+        Err(err) => return Err(discovery_failure(err, "failed to fetch JWKS", OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY)),
     };
 
     Ok(ProviderState {

crates/iam/src/oidc/provider_tests.rs

@@ -299,6 +299,7 @@ async fn workload_discovery_verification_and_rotation() {
             configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
             provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])),
             state_store: OidcStateStore::new(),
+            pending_discovery: Default::default(),
         };
         assert!(sys.has_providers());
         assert!(sys.provider_configs().all(|provider| provider.hide_from_ui));

crates/iam/src/oidc/recovery_tests.rs

@@ -0,0 +1,489 @@
+// Copyright 2026 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.
+
+use super::*;
+use crate::oidc::test_support::{build_mocked_oidc_provider_config, oidc_es256_key_and_jwk, test_sourced_config};
+use jsonwebtoken::{Algorithm, EncodingKey, Header};
+use rustfs_utils::egress::OutboundPolicy;
+use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio_util::task::AbortOnDropHandle;
+
+struct RecoveryIdp {
+    base: String,
+    config: OidcProviderConfig,
+    key: EncodingKey,
+    discovery_status: Arc<AtomicU16>,
+    jwks_status: Arc<AtomicU16>,
+    requests: Arc<AtomicUsize>,
+    cancelled_requests: Arc<AtomicUsize>,
+    _server: AbortOnDropHandle<()>,
+}
+
+impl RecoveryIdp {
+    async fn start(id: &str, workload: bool, explicit_issuer: bool) -> Self {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind recovery IdP");
+        let base = format!("http://{}", listener.local_addr().expect("recovery IdP address"));
+        let mut config = build_mocked_oidc_provider_config(id, &base);
+        config.hide_from_ui = workload;
+        config.claim_name = "groups".into();
+        config.claim_prefix = "team/".into();
+        config.issuer = explicit_issuer.then(|| format!("https://issuer.example.com/{id}"));
+        let (key, jwk) = oidc_es256_key_and_jwk(id);
+        let mut document = serde_json::json!({
+            "issuer": config.issuer.as_deref().unwrap_or(&base),
+            "authorization_endpoint": format!("{base}/authorize"),
+            "token_endpoint": format!("{base}/token"),
+            "jwks_uri": format!("{base}/jwks"),
+            "response_types_supported": ["code"],
+            "subject_types_supported": ["public"],
+            "id_token_signing_alg_values_supported": ["ES256"],
+        });
+        if workload {
+            let fields = document.as_object_mut().expect("discovery document");
+            fields.remove("authorization_endpoint");
+            fields.remove("token_endpoint");
+        }
+        let document = document.to_string();
+        let jwks = serde_json::json!({"keys": [jwk]}).to_string();
+        let discovery_status = Arc::new(AtomicU16::new(503));
+        let jwks_status = Arc::new(AtomicU16::new(200));
+        let requests = Arc::new(AtomicUsize::new(0));
+        let cancelled_requests = Arc::new(AtomicUsize::new(0));
+        let server_discovery_status = discovery_status.clone();
+        let server_jwks_status = jwks_status.clone();
+        let server_requests = requests.clone();
+        let server_cancelled = cancelled_requests.clone();
+        let server = tokio::spawn(async move {
+            loop {
+                let (mut stream, _) = listener.accept().await.expect("accept discovery request");
+                let mut request = Vec::new();
+                while !request.ends_with(b"\r\n\r\n") {
+                    request.push(stream.read_u8().await.expect("read discovery request"));
+                    assert!(request.len() < 8192);
+                }
+                let headers = String::from_utf8(request).expect("HTTP request headers");
+                let path = headers
+                    .lines()
+                    .next()
+                    .expect("request line")
+                    .split_whitespace()
+                    .nth(1)
+                    .expect("path");
+                let (status, body) = if path.ends_with("/.well-known/openid-configuration") {
+                    (server_discovery_status.load(Ordering::SeqCst), document.as_str())
+                } else {
+                    assert_eq!(path, "/jwks");
+                    (server_jwks_status.load(Ordering::SeqCst), jwks.as_str())
+                };
+                server_requests.fetch_add(1, Ordering::SeqCst);
+                if status == 0 {
+                    continue; // Simulate a transport failure before a response is available.
+                }
+                if status == 1 {
+                    let mut byte = [0];
+                    assert_eq!(stream.read(&mut byte).await.expect("client closes cancelled discovery"), 0);
+                    server_cancelled.fetch_add(1, Ordering::SeqCst);
+                    continue;
+                }
+                let body = if status == 200 { body } else { "" };
+                let response = format!(
+                    "HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+                    body.len()
+                );
+                stream.write_all(response.as_bytes()).await.expect("write discovery response");
+            }
+        });
+        Self {
+            base,
+            config,
+            key,
+            discovery_status,
+            jwks_status,
+            requests,
+            cancelled_requests,
+            _server: AbortOnDropHandle::new(server),
+        }
+    }
+
+    fn token(&self, key: &EncodingKey, audience: &str) -> String {
+        let mut header = Header::new(Algorithm::ES256);
+        header.kid = Some(self.config.id.clone());
+        let now = time::OffsetDateTime::now_utc().unix_timestamp();
+        jsonwebtoken::encode(
+            &header,
+            &serde_json::json!({
+                "iss": self.config.issuer.as_deref().unwrap_or(&self.base),
+                "sub": "recovered-user", "aud": audience, "iat": now, "exp": now + 300,
+                "groups": ["readonly"],
+            }),
+            key,
+        )
+        .expect("sign recovery test token")
+    }
+}
+
+async fn initialize(idps: &[&RecoveryIdp]) -> Arc<OidcSys> {
+    let origins = idps.iter().map(|idp| idp.base.as_str()).collect::<Vec<_>>().join(",");
+    let client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&origins).expect("allow test IdPs"));
+    let snapshot = OidcConfigSnapshot::new(idps.iter().map(|idp| test_sourced_config(idp.config.clone())).collect());
+    Arc::new(
+        OidcSys::new_with_config_snapshot(client, snapshot)
+            .await
+            .expect("initialize OIDC with unavailable IdPs"),
+    )
+}
+
+async fn wait_until(mut condition: impl FnMut() -> bool) {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while !condition() {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("recovery condition should be reached");
+}
+
+#[tokio::test]
+async fn web_identity_recovers_transport_failure_and_preserves_verification() {
+    for explicit in [false, true] {
+        let idp = RecoveryIdp::start("workload", true, explicit).await;
+        idp.discovery_status.store(0, Ordering::SeqCst);
+        let sys = initialize(&[&idp]).await;
+        let adapter = StandardOidcAdapter::new(sys.clone());
+        let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+        assert!(sys.has_providers());
+        assert!(sys.list_visible_providers().is_empty());
+        let token = idp.token(&idp.key, &idp.config.client_id);
+        assert!(adapter.verify_identity(&token).await.is_err());
+
+        idp.discovery_status.store(200, Ordering::SeqCst);
+        tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
+            .await
+            .expect("recover workload discovery");
+        let identity = adapter
+            .verify_identity(&token)
+            .await
+            .expect("verify recovered workload token");
+        let authorization = mapper.map(identity);
+        assert_eq!(authorization.provider_id, "workload");
+        assert_eq!(authorization.claims.sub, "recovered-user");
+        assert_eq!(authorization.policies, vec!["team/readonly"]);
+        assert!(
+            sys.authorize_url("workload", "https://console.example.com/callback", None)
+                .await
+                .is_err()
+        );
+
+        let (wrong_key, _) = oidc_es256_key_and_jwk("wrong");
+        assert!(
+            adapter
+                .verify_identity(&idp.token(&wrong_key, &idp.config.client_id))
+                .await
+                .is_err()
+        );
+        assert!(adapter.verify_identity(&idp.token(&idp.key, "wrong-audience")).await.is_err());
+    }
+}
+
+#[tokio::test]
+async fn recovery_waits_for_jwks_and_claims_pending_providers_once() {
+    let idp = RecoveryIdp::start("console", false, true).await;
+    idp.discovery_status.store(200, Ordering::SeqCst);
+    idp.jwks_status.store(503, Ordering::SeqCst);
+    let sys = initialize(&[&idp]).await;
+    assert!(sys.provider_runtime.get_provider_state("console").is_err());
+    let requests = idp.requests.load(Ordering::SeqCst);
+    idp.jwks_status.store(200, Ordering::SeqCst);
+    tokio::time::timeout(
+        Duration::from_secs(5),
+        futures::future::join_all((0..16).map(|_| sys.recover_failed_providers())),
+    )
+    .await
+    .expect("concurrent recovery calls complete");
+    assert!(sys.provider_runtime.get_provider_state("console").is_ok());
+    assert_eq!(
+        idp.requests.load(Ordering::SeqCst),
+        requests + 2,
+        "only one discovery/JWKS round should run"
+    );
+}
+
+#[tokio::test]
+async fn recovery_backoff_is_independent_of_login_traffic() {
+    let idp = RecoveryIdp::start("console", false, true).await;
+    let sys = initialize(&[&idp]).await;
+    let requests = idp.requests.load(Ordering::SeqCst);
+    let recovery_sys = sys.clone();
+    let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
+    wait_until(|| idp.requests.load(Ordering::SeqCst) > requests).await;
+    assert_eq!(idp.requests.load(Ordering::SeqCst), requests + 1);
+    for _ in 0..32 {
+        assert!(
+            sys.authorize_url("console", "https://console.example.com/callback", None)
+                .await
+                .is_err()
+        );
+    }
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert_eq!(
+        idp.requests.load(Ordering::SeqCst),
+        requests + 1,
+        "login attempts must not bypass discovery backoff"
+    );
+    idp.discovery_status.store(200, Ordering::SeqCst);
+    tokio::time::timeout(Duration::from_secs(5), recovery)
+        .await
+        .expect("recovery deadline")
+        .expect("recovery task");
+    assert!(
+        sys.authorize_url("console", "https://console.example.com/callback", None)
+            .await
+            .is_ok()
+    );
+}
+
+#[tokio::test]
+async fn unavailable_provider_does_not_block_another_provider() {
+    let unavailable = RecoveryIdp::start("unavailable", false, true).await;
+    let available = RecoveryIdp::start("available", false, true).await;
+    let sys = initialize(&[&unavailable, &available]).await;
+    available.discovery_status.store(200, Ordering::SeqCst);
+    let recovery_sys = sys.clone();
+    let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
+    wait_until(|| sys.provider_runtime.get_provider_state("available").is_ok()).await;
+    assert!(sys.provider_runtime.get_provider_state("unavailable").is_err());
+    assert!(
+        sys.authorize_url("available", "https://console.example.com/callback", None)
+            .await
+            .is_ok()
+    );
+    drop(recovery);
+}
+
+#[tokio::test]
+async fn recovery_preserves_ready_provider_and_inflight_sessions() {
+    let ready = RecoveryIdp::start("ready", false, true).await;
+    let recovering = RecoveryIdp::start("recovering", false, true).await;
+    ready.discovery_status.store(200, Ordering::SeqCst);
+    let sys = initialize(&[&ready, &recovering]).await;
+    let auth_url = sys
+        .authorize_url("ready", "https://console.example.com/callback", Some("/buckets".into()))
+        .await
+        .expect("begin login with ready provider");
+    let auth: HashMap<String, String> = url::Url::parse(&auth_url)
+        .expect("authorization URL")
+        .query_pairs()
+        .into_owned()
+        .collect();
+    let logout = sys
+        .create_logout_token("ready", "verified-id-token")
+        .await
+        .expect("retain logout continuation");
+    let discovered_at = sys
+        .provider_runtime
+        .get_provider_state("ready")
+        .expect("ready state")
+        .discovered_at;
+    let ready_requests = ready.requests.load(Ordering::SeqCst);
+
+    recovering.discovery_status.store(200, Ordering::SeqCst);
+    tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
+        .await
+        .expect("recover second provider");
+    assert_eq!(ready.requests.load(Ordering::SeqCst), ready_requests);
+    assert_eq!(
+        sys.provider_runtime
+            .get_provider_state("ready")
+            .expect("ready state remains")
+            .discovered_at,
+        discovered_at
+    );
+    let session = sys
+        .state_store
+        .take(&auth["state"])
+        .await
+        .expect("in-flight login remains available");
+    assert_eq!(session.provider_id, "ready");
+    assert_eq!(session.nonce, auth["nonce"]);
+    assert_eq!(session.redirect_after.as_deref(), Some("/buckets"));
+    assert_eq!(
+        PkceCodeChallenge::from_code_verifier_sha256(&PkceCodeVerifier::new(session.pkce_verifier)).as_str(),
+        auth["code_challenge"]
+    );
+    assert_eq!(
+        sys.state_store
+            .take_logout(&logout)
+            .await
+            .expect("logout continuation remains available")
+            .provider_id,
+        "ready"
+    );
+}
+
+#[tokio::test]
+async fn startup_schedules_only_recoverable_http_errors() {
+    for explicit in [false, true] {
+        let idp = RecoveryIdp::start("status", false, explicit).await;
+        for (status, retryable) in [
+            (408, true),
+            (429, true),
+            (500, true),
+            (503, true),
+            (400, false),
+            (401, false),
+            (403, false),
+            (404, false),
+        ] {
+            idp.discovery_status.store(status, Ordering::SeqCst);
+            let sys = initialize(&[&idp]).await;
+            assert_eq!(
+                sys.pending_discovery.lock().await.len(),
+                usize::from(retryable),
+                "explicit={explicit}, status={status}"
+            );
+            assert!(sys.provider_runtime.get_provider_state("status").is_err());
+        }
+    }
+}
+
+#[tokio::test]
+async fn recovery_bounds_concurrency_and_cancels_inflight_requests() {
+    let mut idps = Vec::new();
+    for index in 0..6 {
+        idps.push(RecoveryIdp::start(&format!("provider-{index}"), false, true).await);
+    }
+    let refs: Vec<_> = idps.iter().collect();
+    let sys = initialize(&refs).await;
+    let initial_requests: usize = idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum();
+    for idp in &idps {
+        idp.discovery_status.store(1, Ordering::SeqCst);
+    }
+    let recovery_sys = sys.clone();
+    let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
+    let requests = || idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum::<usize>();
+    wait_until(|| requests() >= initial_requests + 4).await;
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert_eq!(requests(), initial_requests + 4, "at most four discovery rounds may run concurrently");
+    drop(recovery);
+    wait_until(|| {
+        idps.iter()
+            .map(|idp| idp.cancelled_requests.load(Ordering::SeqCst))
+            .sum::<usize>()
+            == 4
+    })
+    .await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert_eq!(requests(), initial_requests + 4, "cancelled recovery must not start queued work");
+    assert!(
+        idps.iter()
+            .all(|idp| sys.provider_runtime.get_provider_state(&idp.config.id).is_err())
+    );
+}
+
+#[tokio::test]
+async fn disabled_and_rejected_providers_do_not_schedule_recovery() {
+    let mut disabled = RecoveryIdp::start("disabled", false, true).await;
+    disabled.config.enabled = false;
+    let sys = initialize(&[&disabled]).await;
+    sys.recover_failed_providers().await;
+    assert!(!sys.has_providers());
+    assert_eq!(disabled.requests.load(Ordering::SeqCst), 0);
+
+    let blocked = RecoveryIdp::start("blocked", false, true).await;
+    let sys = OidcSys::new_with_config_snapshot(
+        ReqwestHttpClient::with_policy(OutboundPolicy::default()),
+        OidcConfigSnapshot::new(vec![test_sourced_config(blocked.config.clone())]),
+    )
+    .await
+    .expect("policy rejection remains non-fatal for startup");
+    assert!(sys.pending_discovery.lock().await.is_empty());
+    sys.recover_failed_providers().await;
+    assert_eq!(blocked.requests.load(Ordering::SeqCst), 0);
+    assert!(sys.provider_runtime.get_provider_state("blocked").is_err());
+}
+
+#[tokio::test]
+async fn recovery_stops_on_issuer_validation_failure() {
+    let mut idp = RecoveryIdp::start("console", false, true).await;
+    idp.config.issuer = Some("https://unexpected.example.com".into());
+    let sys = initialize(&[&idp]).await;
+    idp.discovery_status.store(200, Ordering::SeqCst);
+    tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
+        .await
+        .expect("validation failure stops recovery");
+    assert_eq!(
+        idp.requests.load(Ordering::SeqCst),
+        2,
+        "issuer rejection must not fetch JWKS or schedule another attempt"
+    );
+    assert!(sys.provider_runtime.get_provider_state("console").is_err());
+    assert!(
+        sys.authorize_url("console", "https://console.example.com/callback", None)
+            .await
+            .is_err()
+    );
+}
+
+#[tokio::test]
+async fn recovery_logs_failure_reasons_without_repeating_warnings() {
+    use std::io::{Read, Seek};
+
+    for (explicit, status, expected_kind) in [
+        (false, 503, "http_status"),
+        (true, 503, "http_status"),
+        (true, 0, "request"),
+        (true, 1, "timeout"),
+    ] {
+        let idp = RecoveryIdp::start("logging", false, explicit).await;
+        let sys = initialize(&[&idp]).await;
+        let requests = idp.requests.load(Ordering::SeqCst);
+        idp.discovery_status.store(status, Ordering::SeqCst);
+        let mut output = tempfile::tempfile().expect("create captured log file");
+        let subscriber = tracing_subscriber::fmt()
+            .without_time()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(output.try_clone().expect("clone captured log file"))
+            .finish();
+        let _subscriber = tracing::subscriber::set_default(subscriber);
+        tracing::callsite::rebuild_interest_cache();
+        tokio::time::timeout(Duration::from_secs(25), async {
+            tokio::join!(sys.recover_failed_providers(), async {
+                // Two failing rounds exercise suppression; a timeout is already bounded by the HTTP client.
+                while if status == 1 {
+                    idp.cancelled_requests.load(Ordering::SeqCst) == 0
+                } else {
+                    idp.requests.load(Ordering::SeqCst) < requests + if explicit { 2 } else { 4 }
+                } {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+                idp.discovery_status.store(200, Ordering::SeqCst);
+            });
+        })
+        .await
+        .expect("recover after recording failure diagnostics");
+        output.rewind().expect("rewind captured log file");
+        let mut logs = String::new();
+        output.read_to_string(&mut logs).expect("read captured log file");
+        assert_eq!(logs.lines().count(), 1, "explicit={explicit}, status={status}: {logs}");
+        assert!(logs.contains("provider_discovery_recovery_pending"), "{logs}");
+        assert!(logs.contains(&format!("error_kind=\"{expected_kind}\"")), "{logs}");
+        if status == 503 {
+            assert!(logs.contains("error_detail=503"), "{logs}");
+        }
+        assert!(logs.contains("provider_id=logging"), "{logs}");
+        assert!(logs.contains("suppressed=0"), "{logs}");
+    }
+}

crates/iam/src/oidc/runtime.rs

@@ -16,7 +16,7 @@ use super::{
         OidcConfigQuery, OidcConfigSnapshot, OidcProviderConfig, OidcProviderSummary, OidcSiteReplicationSnapshot,
         SourcedOidcProviderConfig, load_oidc_config_snapshot,
     },
-    provider::{ProviderRuntime, discover_provider},
+    provider::{ProviderRuntime, discover_provider_for_recovery},
     state::{OidcAuthSession, OidcLogoutSession, OidcStateStore},
     transport::{OidcExtraRootCaProvider, ReqwestHttpClient, inspect_token_response_body, oidc_http_error_diagnostics},
 };
@@ -37,11 +37,19 @@ use openidconnect::{
     PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
 };
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
+use rustfs_utils::logging::LogThrottle;
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::sync::Arc;
-use tracing::{debug, error, warn};
+use std::time::Duration;
+use tokio::sync::{Mutex, Semaphore};
+use tracing::{debug, error, info, warn};
+
+const OIDC_DISCOVERY_RECOVERY_CONCURRENCY: usize = 4;
+const OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS: u64 = 30_000;
+// A round can try two issuer variants three times, each fetching discovery and JWKS.
+const OIDC_DISCOVERY_RECOVERY_TIMEOUT: Duration = Duration::from_secs(125);
 
 /// Claims extracted from an OIDC ID token.
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -60,6 +68,7 @@ pub struct OidcSys {
     pub(super) configs: HashMap<String, SourcedOidcProviderConfig>,
     pub(super) provider_runtime: ProviderRuntime,
     pub(super) state_store: OidcStateStore,
+    pub(super) pending_discovery: Mutex<Vec<String>>,
 }
 
 pub(super) fn trusted_aud(other_audiences: &[String], audience: &Audience) -> bool {
@@ -81,8 +90,16 @@ impl OidcSys {
     async fn new_with_http_client(http_client: ReqwestHttpClient) -> Result<Self, String> {
         let server_config = crate::server_config::current_server_config();
         let parsed_configs = load_oidc_config_snapshot(server_config.as_ref());
+        Self::new_with_config_snapshot(http_client, parsed_configs).await
+    }
+
+    async fn new_with_config_snapshot(
+        http_client: ReqwestHttpClient,
+        parsed_configs: OidcConfigSnapshot,
+    ) -> Result<Self, String> {
         let mut configs = HashMap::new();
         let mut provider_states = HashMap::new();
+        let mut pending_discovery = Vec::new();
 
         for sourced_config in parsed_configs.into_providers() {
             let config = &sourced_config.config;
@@ -91,13 +108,15 @@ impl OidcSys {
                 continue;
             }
 
-            match discover_provider(config, &http_client).await {
+            match discover_provider_for_recovery(config, &http_client).await {
                 Ok(state) => {
                     debug!(provider = %config.id, "OIDC provider discovered");
                     provider_states.insert(config.id.clone(), state);
-                    configs.insert(config.id.clone(), sourced_config);
                 }
                 Err(e) => {
+                    if e.retryable {
+                        pending_discovery.push(config.id.clone());
+                    }
                     error!(
                         event = EVENT_OIDC_DIAGNOSTICS,
                         component = LOG_COMPONENT_IAM,
@@ -114,21 +133,95 @@ impl OidcSys {
                     );
                 }
             }
+            // Configuration also supplies redirect and authorization rules before recovery.
+            configs.insert(config.id.clone(), sourced_config);
         }
 
         Ok(Self {
             configs,
             provider_runtime: ProviderRuntime::new(http_client, provider_states),
             state_store: OidcStateStore::new(),
+            pending_discovery: Mutex::new(pending_discovery),
         })
     }
 
+    /// Recover providers that failed startup discovery due to a transient error.
+    /// The server owns and cancels this future; dropping it cancels all outstanding requests.
+    /// Pending providers are claimed once so concurrent callers cannot duplicate recovery.
+    pub async fn recover_failed_providers(&self) {
+        let pending = std::mem::take(&mut *self.pending_discovery.lock().await);
+        let permits = Semaphore::new(OIDC_DISCOVERY_RECOVERY_CONCURRENCY);
+        futures::future::join_all(pending.iter().filter_map(|id| self.get_provider_config(id)).map(|config| {
+            let permits = &permits;
+            async move {
+                let mut delay_ms = 1_000_u64;
+                let failure_log = LogThrottle::new(60_000);
+                loop {
+                    let sleep_ms = rand::random_range(delay_ms * 3 / 4..=delay_ms);
+                    tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
+                    let Ok(permit) = permits.acquire().await else {
+                        return;
+                    };
+                    let result = tokio::time::timeout(
+                        OIDC_DISCOVERY_RECOVERY_TIMEOUT,
+                        self.provider_runtime.recover_provider_state(&config.id, config),
+                    )
+                    .await;
+                    drop(permit);
+                    let (error_kind, error_detail) = match result {
+                        Ok(Ok(_)) => {
+                            info!(
+                                event = EVENT_OIDC_DIAGNOSTICS,
+                                component = LOG_COMPONENT_IAM,
+                                subsystem = LOG_SUBSYSTEM_OIDC,
+                                result = "provider_discovery_recovered",
+                                provider_id = %config.id,
+                                "oidc provider discovery recovered"
+                            );
+                            return;
+                        }
+                        Ok(Err(error)) if !error.retryable => {
+                            error!(
+                                event = EVENT_OIDC_DIAGNOSTICS,
+                                component = LOG_COMPONENT_IAM,
+                                subsystem = LOG_SUBSYSTEM_OIDC,
+                                result = "provider_discovery_recovery_stopped",
+                                provider_id = %config.id,
+                                error = %error,
+                                "oidc provider discovery requires configuration correction"
+                            );
+                            return;
+                        }
+                        Ok(Err(error)) => error.diagnostics,
+                        Err(_) => ("attempt_timeout", OIDC_DISCOVERY_RECOVERY_TIMEOUT.as_secs().to_string()),
+                    };
+                    if let Some(suppressed) = failure_log.claim() {
+                        warn!(
+                            event = EVENT_OIDC_DIAGNOSTICS,
+                            component = LOG_COMPONENT_IAM,
+                            subsystem = LOG_SUBSYSTEM_OIDC,
+                            result = "provider_discovery_recovery_pending",
+                            provider_id = %config.id,
+                            error_kind,
+                            error_detail = %error_detail,
+                            suppressed,
+                            "oidc provider discovery remains unavailable"
+                        );
+                    }
+                    delay_ms = delay_ms.saturating_mul(2).min(OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS);
+                }
+            }
+        }))
+        .await;
+    }
+
     /// Create an OidcSys with no providers (useful for when OIDC is not configured).
     pub fn empty() -> Result<Self, String> {
         Ok(Self {
             configs: HashMap::new(),
             provider_runtime: ProviderRuntime::new(ReqwestHttpClient::new()?, HashMap::new()),
             state_store: OidcStateStore::new(),
+            pending_discovery: Mutex::default(),
         })
     }
 
@@ -895,3 +988,7 @@ impl FederatedProviderQuery for StandardOidcAdapter {
 #[cfg(test)]
 #[path = "runtime_tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "recovery_tests.rs"]
+mod recovery_tests;

crates/iam/src/oidc/runtime_tests.rs

@@ -175,6 +175,7 @@ async fn web_identity_verification_refreshes_rotated_jwks() {
         configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
         provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])),
         state_store: OidcStateStore::new(),
+        pending_discovery: Default::default(),
     };
 
     let now = std::time::SystemTime::now()
@@ -208,6 +209,15 @@ async fn web_identity_verification_refreshes_rotated_jwks() {
 }
 #[tokio::test]
 async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
+    assert_complete_console_login(false).await;
+}
+
+#[tokio::test]
+async fn console_login_recovers_startup_discovery_with_authorization_rules() {
+    assert_complete_console_login(true).await;
+}
+
+async fn assert_complete_console_login(initially_unavailable: bool) {
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     for hidden in [false, true] {
@@ -221,6 +231,8 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
             };
             let mut config = build_mocked_oidc_provider_config("console", &format!("{base}/.well-known/openid-configuration"));
             config.hide_from_ui = hidden;
+            config.claim_name = "groups".into();
+            config.claim_prefix = "team/".into();
             config.issuer = explicit.then(|| issuer.clone());
             config.client_secret = Some(Nonce::new_random().secret().clone());
             let server_config = config.clone();
@@ -230,6 +242,23 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
             let (auth_tx, auth_rx) = tokio::sync::oneshot::channel::<HashMap<String, String>>();
             let server = tokio::spawn(async move {
                 let mut auth_rx = Some(auth_rx);
+                let failures = if initially_unavailable {
+                    if explicit { 1 } else { 2 }
+                } else {
+                    0
+                };
+                for _ in 0..failures {
+                    let (mut stream, _) = listener.accept().await.expect("startup discovery request");
+                    let mut request = Vec::new();
+                    while !request.ends_with(b"\r\n\r\n") {
+                        request.push(stream.read_u8().await.expect("discovery request headers"));
+                        assert!(request.len() < 8192);
+                    }
+                    stream
+                        .write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                        .await
+                        .expect("startup discovery failure response");
+                }
                 for expected_path in ["/.well-known/openid-configuration", "/jwks", "/token"] {
                     let (mut stream, _) = tokio::time::timeout(StdDuration::from_secs(10), listener.accept())
                         .await
@@ -309,12 +338,21 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
                 }
             });
             let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap());
-            let discovered = discover_provider(&config, &http_client).await.unwrap();
-            let sys = OidcSys {
-                configs: HashMap::from([(config.id.clone(), test_sourced_config(config))]),
-                provider_runtime: ProviderRuntime::new(http_client, HashMap::from([("console".into(), discovered)])),
-                state_store: OidcStateStore::new(),
-            };
+            let sys = Arc::new(
+                OidcSys::new_with_config_snapshot(http_client, OidcConfigSnapshot::new(vec![test_sourced_config(config)]))
+                    .await
+                    .expect("OIDC initialization remains available during an IdP outage"),
+            );
+            let adapter = StandardOidcAdapter::new(sys.clone());
+            let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
+            assert_eq!(sys.list_visible_providers().len(), usize::from(!hidden));
+            if initially_unavailable {
+                assert!(sys.provider_runtime.get_provider_state("console").is_err());
+                assert!(sys.authorize_url("console", redirect, None).await.is_err());
+                tokio::time::timeout(StdDuration::from_secs(10), sys.recover_failed_providers())
+                    .await
+                    .expect("discovery should recover after the IdP becomes available");
+            }
             let auth_url = sys.authorize_url("console", redirect, Some("/buckets".into())).await.unwrap();
             let auth_url = Url::parse(&auth_url).unwrap();
             assert_eq!(auth_url.as_str().split('?').next(), Some(format!("{base}/authorize").as_str()));
@@ -328,15 +366,18 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
             assert!(auth["scope"].split_whitespace().any(|scope| scope == "openid"));
             let state = auth["state"].clone();
             auth_tx.send(auth).unwrap();
-            let (claims, provider, session, _) = sys
-                .exchange_code(&state, "test-authorization-code", redirect)
+            let exchange = adapter
+                .exchange_identity(&state, "test-authorization-code", redirect)
                 .await
                 .unwrap_or_else(|err| panic!("hidden={hidden}, explicit={explicit}: {err}"));
-            assert_eq!(provider, "console");
-            assert_eq!(claims.sub, "existing-user");
-            assert_eq!(claims.email, "user@example.com");
-            assert_eq!(claims.groups, vec!["readwrite"]);
-            assert_eq!(session.redirect_after.as_deref(), Some("/buckets"));
+            let (identity, redirect_after, _) = exchange.into_parts();
+            let authorization = mapper.map(identity);
+            assert_eq!(authorization.provider_id, "console");
+            assert_eq!(authorization.claims.sub, "existing-user");
+            assert_eq!(authorization.claims.email, "user@example.com");
+            assert_eq!(authorization.claims.groups, vec!["readwrite"]);
+            assert_eq!(authorization.policies, vec!["team/readwrite"]);
+            assert_eq!(redirect_after.as_deref(), Some("/buckets"));
             assert!(matches!(sys.exchange_code(&state, "test-authorization-code", redirect).await,
                 Err(error) if error == "invalid or expired OIDC state"));
             server.await.unwrap();
@@ -470,6 +511,7 @@ async fn oidc_token_exchange_reports_forbidden_token_endpoint() {
             )]),
         ),
         state_store,
+        pending_discovery: Default::default(),
     };
 
     let error = match sys

crates/iam/src/oidc/test_support.rs

@@ -35,6 +35,7 @@ pub fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
             HashMap::new(),
         ),
         state_store: OidcStateStore::new(),
+        pending_discovery: Default::default(),
     }
 }
 

crates/iam/src/oidc/transport.rs

@@ -312,6 +312,10 @@ struct CachedOidcExtraRootCerts {
     certs: Vec<Certificate>,
 }
 
+/// Discovery callers report failures with provider context and recovery throttling.
+#[derive(Clone)]
+pub(super) struct DiscoveryRequest;
+
 /// HTTP client adapter bridging reqwest 0.13 to the `openidconnect` `AsyncHttpClient` trait.
 ///
 /// A fresh client is built for every request so the destination is re-validated and the
@@ -555,6 +559,7 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
         Box::pin(async move {
             let started_at = Instant::now();
             let (parts, body) = request.into_parts();
+            let discovery_request = parts.extensions.get::<DiscoveryRequest>().is_some();
             let method = parts.method.clone();
             let uri = parts.uri.to_string();
             if tracing::enabled!(tracing::Level::DEBUG) {
@@ -593,17 +598,19 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
 
             let response = response.map_err(|err| {
                 let error = oidc_http_error_from_reqwest(&url, err);
-                error!(
-                    event = EVENT_OIDC_HTTP,
-                    component = LOG_COMPONENT_IAM,
-                    subsystem = LOG_SUBSYSTEM_OIDC,
-                    result = "request_failed",
-                    method = %method,
-                    uri = %uri,
-                    elapsed_ms,
-                    error = %error,
-                    "oidc outbound http"
-                );
+                if !discovery_request {
+                    error!(
+                        event = EVENT_OIDC_HTTP,
+                        component = LOG_COMPONENT_IAM,
+                        subsystem = LOG_SUBSYSTEM_OIDC,
+                        result = "request_failed",
+                        method = %method,
+                        uri = %uri,
+                        elapsed_ms,
+                        error = %error,
+                        "oidc outbound http"
+                    );
+                }
                 error
             })?;
 
@@ -612,18 +619,20 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
             let body_bytes = read_bounded_response_body(response, MAX_OIDC_RESPONSE_SIZE)
                 .await
                 .map_err(|err| {
-                    error!(
-                        event = EVENT_OIDC_HTTP,
-                        component = LOG_COMPONENT_IAM,
-                        subsystem = LOG_SUBSYSTEM_OIDC,
-                        result = "response_body_failed",
-                        method = %method,
-                        uri = %uri,
-                        status = status.as_u16(),
-                        elapsed_ms,
-                        error = %err,
-                        "oidc outbound http"
-                    );
+                    if !discovery_request {
+                        error!(
+                            event = EVENT_OIDC_HTTP,
+                            component = LOG_COMPONENT_IAM,
+                            subsystem = LOG_SUBSYSTEM_OIDC,
+                            result = "response_body_failed",
+                            method = %method,
+                            uri = %uri,
+                            status = status.as_u16(),
+                            elapsed_ms,
+                            error = %err,
+                            "oidc outbound http"
+                        );
+                    }
                     err
                 })?;
             if tracing::enabled!(tracing::Level::DEBUG) {

docs/operations/bucket-operation-admission.md

@@ -1,10 +1,10 @@
 # Bucket creation and deletion admission
 
-`CreateBucket` and `DeleteBucket` share a process-wide execution budget of eight active transactions. Short bursts wait in a FIFO queue with space for 128 waiting requests. Each request can wait up to 30 seconds for an execution slot. These limits protect storage and namespace-lock resources while allowing parallel test workers to create isolated buckets.
+`CreateBucket` and `DeleteBucket` share a process-wide execution budget of eight active storage mutations. Short bursts wait in a FIFO queue with space for 128 waiting requests. Each request can wait up to 30 seconds for an execution slot. A separate total budget bounds all unfinished operations to 136, including queued requests, mutations, and post-commit peer reloads. These limits protect storage, namespace-lock, and notification resources while allowing parallel test workers to create isolated buckets.
 
-A request that finds the queue full or exceeds the queue wait returns HTTP 503 with the S3 error code `SlowDown` and `Retry-After: 1`. This is an overload response, not a bucket-name or filesystem error. The active and waiting limits are shared by creation and deletion and are currently fixed.
+A request that finds the queue full, exceeds the queue wait, or reaches the total in-flight limit returns HTTP 503 with the S3 error code `SlowDown` and `Retry-After: 1`. Admission rejection happens before storage mutation begins. This is an overload response, not a bucket-name or filesystem error. The limits are shared by creation and deletion and are currently fixed.
 
-Waiting requests do not start storage mutations. Disconnecting while waiting releases the queue slot. Once admitted, the transaction owns its execution slot until the mutation and post-commit hooks finish, including when its caller disconnects.
+Waiting requests do not start storage mutations. Disconnecting while waiting releases both the queue slot and its total-operation slot. Once admitted, the detached task continues even when its caller disconnects. Creation releases its execution slot after the storage mutation, site-replication hook, and associated locks finish, then waits for peer metadata reloads before returning. Its total-operation slot remains held until that work finishes, so slow reloads cannot accumulate unbounded background tasks or occupy every storage execution slot. Deletion holds its execution slot through its usecase; its existing asynchronous peer cache invalidation is unchanged.
 
 For automated tests and applications:
 

rustfs/src/app/bucket_usecase.rs

@@ -69,7 +69,8 @@ use crate::app::object::shared::mark_on_demand_migration_list_local_only;
 use crate::app::object_data_cache::invalidate_object_data_cache_bucket_after_delete;
 use crate::app::runtime_sources::{
     AppContext, current_app_context, current_encryption_service, current_notification_system,
-    current_notify_interface_for_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
+    current_notification_system_for_context, current_notify_interface_for_context, current_object_data_cache_for_context,
+    current_object_store_handle_for_context,
 };
 use crate::app::table_list_isolation;
 use crate::auth::get_condition_values_with_client_info;
@@ -151,6 +152,7 @@ static BUCKET_OPERATION_ADMISSION: LazyLock<Arc<BucketOperationAdmission>> = Laz
 struct BucketOperationAdmission {
     active: Arc<Semaphore>,
     waiting: Semaphore,
+    in_flight: Arc<Semaphore>,
 }
 
 impl BucketOperationAdmission {
@@ -158,12 +160,19 @@ impl BucketOperationAdmission {
         Self {
             active: Arc::new(Semaphore::new(concurrency)),
             waiting: Semaphore::new(queue_capacity),
+            in_flight: Arc::new(Semaphore::new(concurrency + queue_capacity)),
         }
     }
 
-    async fn acquire(&self, operation: &'static str) -> S3Result<OwnedSemaphorePermit> {
+    async fn acquire(&self, operation: &'static str) -> S3Result<(OwnedSemaphorePermit, OwnedSemaphorePermit)> {
+        // Reserve the entire operation before mutation, including post-commit
+        // work that can outlive both its execution slot and the request.
+        let in_flight = self.in_flight.clone().try_acquire_owned().map_err(|err| match err {
+            TryAcquireError::NoPermits => bucket_admission_slow_down(operation, "in-flight limit reached"),
+            TryAcquireError::Closed => bucket_admission_closed(operation),
+        })?;
         match self.active.clone().try_acquire_owned() {
-            Ok(permit) => return Ok(permit),
+            Ok(permit) => return Ok((permit, in_flight)),
             Err(TryAcquireError::Closed) => return Err(bucket_admission_closed(operation)),
             Err(TryAcquireError::NoPermits) => {}
         }
@@ -174,10 +183,11 @@ impl BucketOperationAdmission {
             TryAcquireError::NoPermits => bucket_admission_slow_down(operation, "queue is full"),
             TryAcquireError::Closed => bucket_admission_closed(operation),
         })?;
-        tokio::time::timeout(BUCKET_OPERATION_QUEUE_TIMEOUT, self.active.clone().acquire_owned())
+        let active = tokio::time::timeout(BUCKET_OPERATION_QUEUE_TIMEOUT, self.active.clone().acquire_owned())
             .await
             .map_err(|_| bucket_admission_slow_down(operation, "queue wait timed out"))?
-            .map_err(|_| bucket_admission_closed(operation))
+            .map_err(|_| bucket_admission_closed(operation))?;
+        Ok((active, in_flight))
     }
 }
 
@@ -615,12 +625,27 @@ async fn notify_bucket_metadata_reload(
     operation: &'static str,
     request_context: Option<request_context::RequestContext>,
     scanner_maintenance_change: bool,
+) {
+    notify_bucket_metadata_reload_with_context(bucket, operation, request_context, scanner_maintenance_change, None).await;
+}
+
+async fn notify_bucket_metadata_reload_with_context(
+    bucket: String,
+    operation: &'static str,
+    request_context: Option<request_context::RequestContext>,
+    scanner_maintenance_change: bool,
+    app_context: Option<Arc<AppContext>>,
 ) {
     record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change);
     // Keep reload detached across request cancellation, but wait before a healthy peer can serve the previous config.
     let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
     spawn_background_with_context(request_context, async move {
-        if let Some(notification_sys) = current_notification_system() {
+        let notification_sys = if let Some(context) = app_context.as_deref() {
+            current_notification_system_for_context(Some(context))
+        } else {
+            current_notification_system()
+        };
+        if let Some(notification_sys) = notification_sys {
             let result = if scanner_maintenance_change {
                 notification_sys.load_bucket_metadata_for_scanner_maintenance(&bucket).await
             } else {
@@ -1357,28 +1382,30 @@ pub struct DefaultBucketUsecase {
     context: Option<Arc<AppContext>>,
 }
 
-async fn await_bucket_usecase_on_fresh_task<T, F>(operation: &'static str, future: F) -> S3Result<T>
+async fn await_bucket_usecase_on_fresh_task<T, F, MakeFuture>(operation: &'static str, make_future: MakeFuture) -> S3Result<T>
 where
     T: Send + 'static,
     F: Future<Output = S3Result<T>> + Send + 'static,
+    MakeFuture: FnOnce(OwnedSemaphorePermit) -> F + Send + 'static,
 {
-    await_bucket_usecase_on_fresh_task_with_admission(operation, Arc::clone(&BUCKET_OPERATION_ADMISSION), future).await
+    await_bucket_usecase_on_fresh_task_with_admission(operation, Arc::clone(&BUCKET_OPERATION_ADMISSION), make_future).await
 }
 
-async fn await_bucket_usecase_on_fresh_task_with_admission<T, F>(
+async fn await_bucket_usecase_on_fresh_task_with_admission<T, F, MakeFuture>(
     operation: &'static str,
     admission: Arc<BucketOperationAdmission>,
-    future: F,
+    make_future: MakeFuture,
 ) -> S3Result<T>
 where
     T: Send + 'static,
     F: Future<Output = S3Result<T>> + Send + 'static,
+    MakeFuture: FnOnce(OwnedSemaphorePermit) -> F + Send + 'static,
 {
-    let permit = admission.acquire(operation).await?;
+    let (mutation_permit, in_flight) = admission.acquire(operation).await?;
     tokio::spawn(
         async move {
-            let _permit = permit;
-            future.await
+            let _in_flight = in_flight;
+            make_future(mutation_permit).await
         }
         .in_current_span(),
     )
@@ -1421,11 +1448,19 @@ impl DefaultBucketUsecase {
     )]
     pub async fn execute_create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
         let usecase = self.clone();
-        await_bucket_usecase_on_fresh_task("bucket creation", async move { usecase.execute_create_bucket_inner(req).await }).await
+        await_bucket_usecase_on_fresh_task("bucket creation", move |mutation_permit| async move {
+            usecase.execute_create_bucket_inner(req, mutation_permit).await
+        })
+        .await
     }
 
-    async fn execute_create_bucket_inner(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
+    async fn execute_create_bucket_inner(
+        &self,
+        req: S3Request<CreateBucketInput>,
+        mutation_permit: OwnedSemaphorePermit,
+    ) -> S3Result<S3Response<CreateBucketOutput>> {
         let helper = OperationHelper::new(&req, EventName::BucketCreated, S3Operation::CreateBucket);
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
         let requester_is_owner = match req_info_ref(&req) {
             Ok(r) => r.is_owner,
             Err(_) => {
@@ -1469,8 +1504,27 @@ impl DefaultBucketUsecase {
         })
         .await?;
 
+        // The storage/site mutation and its lock fences have finished. Keep
+        // the operation's in-flight slot while peers reload, without holding
+        // an execution slot needed by other creates and deletes.
+        drop(mutation_permit);
+
         match make_result {
-            Ok(()) => {}
+            Ok(()) => {
+                // A peer may have cached this bucket as missing before the
+                // create. Refresh peer metadata after the namespace and
+                // metadata commit fences are released so the peer reload can
+                // acquire its bucket namespace read lock without an ABBA
+                // interleaving with creation.
+                notify_bucket_metadata_reload_with_context(
+                    bucket.clone(),
+                    "create bucket",
+                    request_context,
+                    false,
+                    self.context.clone(),
+                )
+                .await;
+            }
             Err(StorageError::BucketExists(_)) => {
                 // Per S3 spec: bucket namespace is global. Owner recreating returns 200 OK;
                 // non-owner gets 409 BucketAlreadyExists.
@@ -1495,7 +1549,11 @@ impl DefaultBucketUsecase {
     #[instrument(level = "debug", skip(self, req))]
     pub async fn execute_delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
         let usecase = self.clone();
-        await_bucket_usecase_on_fresh_task("bucket deletion", async move { usecase.execute_delete_bucket_inner(req).await }).await
+        await_bucket_usecase_on_fresh_task("bucket deletion", move |mutation_permit| async move {
+            let _mutation_permit = mutation_permit;
+            usecase.execute_delete_bucket_inner(req).await
+        })
+        .await
     }
 
     async fn execute_delete_bucket_inner(
@@ -3308,7 +3366,8 @@ mod tests {
         let parent = tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
             "test bucket operation",
             admission.clone(),
-            async move {
+            move |mutation_permit| async move {
+                drop(mutation_permit);
                 committed_for_task.notify_one();
                 release_hook_for_task.notified().await;
                 hook_ran_for_task.store(true, Ordering::SeqCst);
@@ -3320,6 +3379,8 @@ mod tests {
 
         parent.abort();
         let _ = parent.await;
+        assert_eq!(admission.active.available_permits(), 1);
+        assert_eq!(admission.in_flight.available_permits(), 1, "cancelled post-commit work stays bounded");
         release_hook.notify_waiters();
         tokio::time::timeout(Duration::from_secs(1), finished_wait)
             .await
@@ -3330,6 +3391,137 @@ mod tests {
             "the fresh task must own both the storage mutation and its post-commit hooks"
         );
         assert_eq!(admission.active.available_permits(), 1);
+        assert_eq!(admission.in_flight.available_permits(), 2);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn bucket_usecase_admission_releases_execution_during_slow_post_commit_work() {
+        let admission = Arc::new(BucketOperationAdmission::new(
+            BUCKET_OPERATION_CONCURRENCY,
+            BUCKET_OPERATION_QUEUE_CAPACITY,
+        ));
+        let active_mutations = Arc::new(AtomicUsize::new(0));
+        let peak_mutations = Arc::new(AtomicUsize::new(0));
+        let (committed_tx, mut committed_rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut tasks = Vec::new();
+        for _ in 0..100 {
+            let active = active_mutations.clone();
+            let peak = peak_mutations.clone();
+            let committed = committed_tx.clone();
+            tasks.push(tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
+                "bucket creation",
+                admission.clone(),
+                move |mutation_permit| async move {
+                    let count = active.fetch_add(1, Ordering::SeqCst) + 1;
+                    peak.fetch_max(count, Ordering::SeqCst);
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                    active.fetch_sub(1, Ordering::SeqCst);
+                    drop(mutation_permit);
+                    committed.send(()).expect("record commit");
+                    tokio::time::sleep(Duration::from_secs(4)).await;
+                    Ok(())
+                },
+            )));
+        }
+        tokio::time::timeout(Duration::from_secs(1), async {
+            for _ in 0..100 {
+                committed_rx.recv().await.expect("every mutation commits");
+            }
+        })
+        .await
+        .expect("peer reloads must not serialize subsequent mutations");
+        assert_eq!(peak_mutations.load(Ordering::SeqCst), BUCKET_OPERATION_CONCURRENCY);
+        assert!(
+            tasks.iter().all(|task| !task.is_finished()),
+            "requests must still await their post-commit work"
+        );
+        assert_eq!(admission.active.available_permits(), BUCKET_OPERATION_CONCURRENCY);
+        assert_eq!(
+            admission.in_flight.available_permits(),
+            BUCKET_OPERATION_CONCURRENCY + BUCKET_OPERATION_QUEUE_CAPACITY - 100
+        );
+
+        await_bucket_usecase_on_fresh_task_with_admission("bucket deletion", admission.clone(), |mutation_permit| async move {
+            let _mutation_permit = mutation_permit;
+            Ok(())
+        })
+        .await
+        .expect("deletion can execute while creates await peer reloads");
+        for task in tasks {
+            task.await.expect("request joins").expect("queued creation succeeds");
+        }
+        assert_eq!(
+            admission.in_flight.available_permits(),
+            BUCKET_OPERATION_CONCURRENCY + BUCKET_OPERATION_QUEUE_CAPACITY
+        );
+    }
+
+    #[tokio::test]
+    async fn bucket_usecase_admission_bounds_post_commit_work_after_request_cancellation() {
+        let admission = Arc::new(BucketOperationAdmission::new(1, 1));
+        let release = Arc::new(Semaphore::new(0));
+        let (committed_tx, mut committed_rx) = tokio::sync::mpsc::unbounded_channel();
+        let (finished_tx, mut finished_rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut parents = Vec::new();
+        for _ in 0..2 {
+            let release = release.clone();
+            let committed = committed_tx.clone();
+            let finished = finished_tx.clone();
+            parents.push(tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
+                "bucket creation",
+                admission.clone(),
+                move |mutation_permit| async move {
+                    drop(mutation_permit);
+                    committed.send(()).expect("record commit");
+                    release.acquire().await.expect("release hook").forget();
+                    finished.send(()).expect("record completion");
+                    Ok(())
+                },
+            )));
+        }
+        tokio::time::timeout(Duration::from_secs(5), async {
+            for _ in 0..2 {
+                committed_rx.recv().await.expect("operation commits");
+            }
+        })
+        .await
+        .expect("both operations reach post-commit work");
+        for parent in parents {
+            parent.abort();
+            assert!(parent.await.expect_err("caller cancelled").is_cancelled());
+        }
+        assert_eq!(admission.active.available_permits(), 1);
+        assert_eq!(admission.in_flight.available_permits(), 0);
+        let started = Arc::new(AtomicBool::new(false));
+        let started_for_task = started.clone();
+        let error = await_bucket_usecase_on_fresh_task_with_admission(
+            "bucket creation",
+            admission.clone(),
+            move |mutation_permit| async move {
+                let _mutation_permit = mutation_permit;
+                started_for_task.store(true, Ordering::SeqCst);
+                Ok(())
+            },
+        )
+        .await
+        .expect_err("unfinished hooks consume the total operation budget");
+        assert_eq!(error.code(), &S3ErrorCode::SlowDown);
+        assert_eq!(error.message(), Some("bucket creation in-flight limit reached; retry later"));
+        assert_eq!(error.headers().expect("retry headers")[http::header::RETRY_AFTER], "1");
+        assert!(!started.load(Ordering::SeqCst));
+        release.add_permits(2);
+        tokio::time::timeout(Duration::from_secs(5), async {
+            for _ in 0..2 {
+                finished_rx.recv().await.expect("detached hook completes");
+            }
+            while admission.in_flight.available_permits() != 2 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("capacity recovers after hooks complete");
+        let permits = admission.acquire("bucket deletion").await.expect("new operation is admitted");
+        drop(permits);
     }
 
     #[tokio::test(start_paused = true)]
@@ -3338,16 +3530,22 @@ mod tests {
         let held = admission.active.clone().acquire_owned().await.expect("hold active slot");
         let started = Arc::new(AtomicBool::new(false));
         let started_for_task = started.clone();
-        let result = await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission.clone(), async move {
-            started_for_task.store(true, Ordering::SeqCst);
-            Ok(())
-        })
+        let result = await_bucket_usecase_on_fresh_task_with_admission(
+            "test bucket operation",
+            admission.clone(),
+            move |mutation_permit| async move {
+                let _mutation_permit = mutation_permit;
+                started_for_task.store(true, Ordering::SeqCst);
+                Ok(())
+            },
+        )
         .await;
         let error = result.expect_err("queued request must time out");
         assert_eq!(error.code(), &S3ErrorCode::SlowDown);
         assert_eq!(error.headers().expect("retry headers")[http::header::RETRY_AFTER], "1");
         assert!(!started.load(Ordering::SeqCst), "timed-out work must not start");
         assert_eq!(admission.waiting.available_permits(), 1);
+        assert_eq!(admission.in_flight.available_permits(), 2);
         drop(held);
         assert_eq!(admission.active.available_permits(), 1);
     }
@@ -3368,7 +3566,8 @@ mod tests {
             tasks.push(tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
                 "test bucket operation",
                 admission_for_task,
-                async move {
+                move |mutation_permit| async move {
+                    let _mutation_permit = mutation_permit;
                     started_for_task.fetch_add(1, Ordering::SeqCst);
                     release_for_task.acquire().await.expect("release gate").forget();
                     Ok(())
@@ -3403,7 +3602,8 @@ mod tests {
         let queued = tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
             "test bucket operation",
             admission.clone(),
-            async move {
+            move |mutation_permit| async move {
+                let _mutation_permit = mutation_permit;
                 started_for_task.store(true, Ordering::SeqCst);
                 Ok(())
             },
@@ -3430,35 +3630,58 @@ mod tests {
         queued.abort();
         assert!(queued.await.expect_err("queued request was cancelled").is_cancelled());
         assert_eq!(admission.waiting.available_permits(), 1);
+        assert_eq!(admission.in_flight.available_permits(), 2);
         drop(held);
         assert!(!started.load(Ordering::SeqCst), "cancelled waiter must not mutate storage");
-        await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission, async { Ok(()) })
-            .await
-            .expect("admission recovers after cancellation");
+        await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission, |mutation_permit| async move {
+            let _mutation_permit = mutation_permit;
+            Ok(())
+        })
+        .await
+        .expect("admission recovers after cancellation");
     }
 
     #[tokio::test]
     async fn bucket_usecase_admission_maps_closed_semaphore_to_internal_error() {
-        let admission = BucketOperationAdmission::new(1, 1);
-        admission.active.close();
-        let error = admission
-            .acquire("test bucket operation")
-            .await
-            .expect_err("closed admission must fail");
-        assert_eq!(error.code(), &S3ErrorCode::InternalError);
+        for close_in_flight in [false, true] {
+            let admission = BucketOperationAdmission::new(1, 1);
+            if close_in_flight {
+                admission.in_flight.close();
+            } else {
+                admission.active.close();
+            }
+            let error = admission
+                .acquire("test bucket operation")
+                .await
+                .expect_err("closed admission must fail");
+            assert_eq!(error.code(), &S3ErrorCode::InternalError);
+            assert_eq!(admission.in_flight.available_permits(), 2);
+        }
     }
 
     #[tokio::test]
     async fn bucket_usecase_task_maps_panics_to_internal_errors() {
-        async fn panicking_bucket_operation() -> S3Result<()> {
-            panic!("injected bucket task panic");
+        for post_commit in [false, true] {
+            let admission = Arc::new(BucketOperationAdmission::new(1, 1));
+            let result: S3Result<()> = await_bucket_usecase_on_fresh_task_with_admission(
+                "test bucket operation",
+                admission.clone(),
+                move |mutation_permit| async move {
+                    let mut mutation_permit = Some(mutation_permit);
+                    if post_commit {
+                        drop(mutation_permit.take());
+                    }
+                    panic!("injected bucket task panic");
+                },
+            )
+            .await;
+
+            let err = result.expect_err("a bucket task panic must be returned as an error");
+            assert_eq!(err.code(), &S3ErrorCode::InternalError);
+            assert!(err.message().is_some_and(|message| message.contains("task failed")));
+            assert_eq!(admission.active.available_permits(), 1);
+            assert_eq!(admission.in_flight.available_permits(), 2);
         }
-
-        let result = await_bucket_usecase_on_fresh_task("test bucket operation", panicking_bucket_operation()).await;
-
-        let err = result.expect_err("a bucket task panic must be returned as an error");
-        assert_eq!(err.code(), &S3ErrorCode::InternalError);
-        assert!(err.message().is_some_and(|message| message.contains("task failed")));
     }
 
     fn build_request<T>(input: T, method: Method) -> S3Request<T> {
@@ -3482,30 +3705,46 @@ mod tests {
     }
 
     fn usecase_method_source<'a>(source: &'a str, method: &str) -> &'a str {
-        let start_marker = format!("pub async fn {method}");
-        let start = source.find(&start_marker).expect("method should exist");
-        let rest = &source[start + start_marker.len()..];
-        let end = rest.find("\n    pub async fn ").unwrap_or(rest.len());
+        let public_start_marker = format!("pub async fn {method}");
+        let private_start_marker = format!("async fn {method}");
+        let (start, marker_len) = if let Some(start) = source.find(&public_start_marker) {
+            (start, public_start_marker.len())
+        } else {
+            let start = source.find(&private_start_marker).expect("method should exist");
+            (start, private_start_marker.len())
+        };
+        let rest = &source[start + marker_len..];
+        let end = [rest.find("\n    pub async fn "), rest.find("\n    async fn ")]
+            .into_iter()
+            .flatten()
+            .min()
+            .unwrap_or(rest.len());
         &rest[..end]
     }
 
     #[test]
     fn bucket_create_and_delete_use_admitted_fresh_tasks() {
         let source = include_str!("bucket_usecase.rs");
-        for (method, operation, inner_method) in [
-            ("execute_create_bucket", "bucket creation", "execute_create_bucket_inner"),
-            ("execute_delete_bucket", "bucket deletion", "execute_delete_bucket_inner"),
+        for (method, operation, inner_call) in [
+            (
+                "execute_create_bucket",
+                "bucket creation",
+                "execute_create_bucket_inner(req, mutation_permit).await",
+            ),
+            ("execute_delete_bucket", "bucket deletion", "execute_delete_bucket_inner(req).await"),
         ] {
             let body = usecase_method_source(source, method);
             assert!(
                 body.contains(&format!("await_bucket_usecase_on_fresh_task(\"{operation}\"")),
                 "{method} must use bounded detached-task admission"
             );
-            assert!(
-                body.contains(&format!("{inner_method}(req).await")),
-                "{method} must run its mutation inside the admitted task"
-            );
+            assert!(body.contains(inner_call), "{method} must run its mutation inside the admitted task");
         }
+        let create_body = usecase_method_source(source, "execute_create_bucket_inner");
+        assert!(
+            create_body.contains("notify_bucket_metadata_reload_with_context("),
+            "a successful bucket creation must refresh peer metadata caches"
+        );
     }
 
     #[test]

rustfs/src/app/runtime_sources.rs

@@ -19,7 +19,8 @@ use crate::runtime_sources as root_runtime_sources;
 pub(crate) use crate::runtime_sources::ServerContextSlot;
 pub(crate) use crate::runtime_sources::{
     AppContext, current_encryption_service, current_endpoints_handle, current_notification_system,
-    current_object_data_cache_handle_for_context, current_object_store_handle_for_context,
+    current_notification_system_for_context, current_object_data_cache_handle_for_context,
+    current_object_store_handle_for_context,
 };
 use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem};
 use s3s::dto::SelectObjectContentInput;

rustfs/src/startup_auth.rs

@@ -19,10 +19,12 @@ use rustfs_iam::{
 };
 use std::{
     collections::hash_map::DefaultHasher,
+    future::Future,
     hash::{Hash, Hasher},
     io::{Error, Result},
     sync::Arc,
 };
+use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle};
 use tracing::{error, info, warn};
 
 const LOG_COMPONENT_MAIN: &str = "main";
@@ -31,7 +33,7 @@ const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized";
 const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed";
 const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed";
 
-pub(crate) async fn init_auth_integrations() -> Result<()> {
+pub(crate) async fn init_auth_integrations(shutdown: CancellationToken) -> Result<Option<AbortOnDropHandle<()>>> {
     let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?;
     if keystone_config.enable {
         match crate::auth_keystone::init_keystone_auth(keystone_config).await {
@@ -55,8 +57,13 @@ pub(crate) async fn init_auth_integrations() -> Result<()> {
 
     match build_oidc_sys_with_extra_root_ca_provider(oidc_extra_root_ca_provider()).await {
         Ok(oidc) => {
-            let (service, oidc_config_query) = standard_oidc_runtime(oidc);
-            crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query);
+            let (service, oidc_config_query) = standard_oidc_runtime(oidc.clone());
+            if crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query) {
+                return Ok(Some(spawn_oidc_discovery_recovery(
+                    async move { oidc.recover_failed_providers().await },
+                    shutdown,
+                )));
+            }
         }
         Err(e) => {
             warn!(
@@ -69,7 +76,20 @@ pub(crate) async fn init_auth_integrations() -> Result<()> {
         }
     }
 
-    Ok(())
+    Ok(None)
+}
+
+fn spawn_oidc_discovery_recovery(
+    recovery: impl Future<Output = ()> + Send + 'static,
+    shutdown: CancellationToken,
+) -> AbortOnDropHandle<()> {
+    AbortOnDropHandle::new(tokio::spawn(async move {
+        tokio::select! {
+            biased;
+            () = shutdown.cancelled() => {}
+            () = recovery => {}
+        }
+    }))
 }
 
 fn standard_oidc_runtime(oidc: Arc<OidcSys>) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
@@ -133,4 +153,48 @@ mod tests {
         drop(oidc_config_query);
         assert_eq!(Arc::strong_count(&retained), 1, "dropping the runtime should release its OIDC Arc");
     }
+
+    #[tokio::test]
+    async fn oidc_recovery_shutdown_releases_pending_work() {
+        let shutdown = CancellationToken::new();
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>();
+        let recovery = spawn_oidc_discovery_recovery(
+            async move {
+                let _release = released_tx;
+                let _ = started_tx.send(());
+                std::future::pending::<()>().await;
+            },
+            shutdown.child_token(),
+        );
+        started_rx.await.expect("recovery started");
+        shutdown.cancel();
+        tokio::time::timeout(std::time::Duration::from_secs(5), recovery)
+            .await
+            .expect("recovery observes shutdown")
+            .expect("recovery joins cleanly");
+        assert!(released_rx.await.is_err(), "shutdown drops in-flight recovery work");
+    }
+
+    #[tokio::test]
+    async fn oidc_recovery_owner_drop_releases_pending_work() {
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>();
+        let recovery = spawn_oidc_discovery_recovery(
+            async move {
+                let _release = released_tx;
+                let _ = started_tx.send(());
+                std::future::pending::<()>().await;
+            },
+            CancellationToken::new(),
+        );
+        started_rx.await.expect("recovery started");
+        drop(recovery);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_secs(5), released_rx)
+                .await
+                .expect("owner drop aborts recovery")
+                .is_err()
+        );
+    }
 }

rustfs/src/startup_lifecycle.rs

@@ -130,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
         readiness,
     } = lifecycle;
     let StartupServiceRuntime {
+        oidc_recovery,
         optional_runtimes,
         heartbeat,
         inventory,
@@ -166,6 +167,18 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
         shutdown_token,
     )
     .await;
+    if let Some(recovery) = oidc_recovery
+        && let Err(error) = recovery.await
+    {
+        tracing::warn!(
+            event = EVENT_SERVER_SHUTDOWN_STATE,
+            component = LOG_COMPONENT_MAIN,
+            subsystem = LOG_SUBSYSTEM_STARTUP,
+            result = "oidc_discovery_recovery_join_failed",
+            error = %error,
+            "OIDC discovery recovery task failed to join"
+        );
+    }
     shutdown_connect_runtimes(heartbeat, inventory).await;
     if let Some(runtime) = local_trace_capture {
         runtime.shutdown().await;

rustfs/src/startup_services.rs

@@ -36,13 +36,14 @@ use crate::{
 };
 use rustfs_common::GlobalReadiness;
 use std::{collections::BTreeSet, io::Result, sync::Arc};
-use tokio_util::sync::CancellationToken;
+use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle};
 
 const EVENT_DRIVE_UNAVAILABLE: &str = "drive_unavailable";
 const LOG_COMPONENT_CONNECT: &str = "connect";
 const LOG_SUBSYSTEM_INVENTORY: &str = "inventory";
 
 pub(crate) struct StartupServiceRuntime {
+    pub(crate) oidc_recovery: Option<AbortOnDropHandle<()>>,
     pub(crate) optional_runtimes: OptionalRuntimeServices,
     pub(crate) heartbeat: Option<HeartbeatRuntime>,
     pub(crate) inventory: Option<InventoryRuntime>,
@@ -111,7 +112,7 @@ pub(crate) async fn init_startup_runtime_services(
     // self-pointing replication rules until the next restart. The task waits for IAM and
     // bucket metadata itself, and its first pass runs off this path.
     spawn_site_replication_reconcile_task(ctx.clone());
-    init_auth_integrations().await?;
+    let oidc_recovery = init_auth_integrations(ctx.child_token()).await?;
     init_notification_runtime(endpoint_pools, buckets).await?;
     let enable_scanner = init_background_service_runtime(store.clone()).await?;
     init_observability_runtime(store.clone(), ctx.clone()).await;
@@ -119,6 +120,7 @@ pub(crate) async fn init_startup_runtime_services(
     let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?;
 
     Ok(StartupServiceRuntime {
+        oidc_recovery,
         optional_runtimes,
         heartbeat,
         inventory,

rustfs/src/storage/rpc/node_service/bucket.rs

@@ -17,7 +17,8 @@ use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{
     BucketOptions, DeleteBucketOptions, MakeBucketOptions,
 };
 use crate::storage::storage_api::rpc_consumer::node_service::{
-    DiskError, StoragePeerS3ClientExt as _, decode_heal_bucket_rpc_options, reload_bucket_metadata, remove_bucket_metadata,
+    DiskError, StoragePeerS3ClientExt as _, decode_heal_bucket_rpc_options, reload_bucket_metadata,
+    remove_bucket_metadata_if_absent,
 };
 use rustfs_protos::proto_gen::node_service::*;
 use tonic::{Request, Response, Status};
@@ -38,9 +39,16 @@ impl NodeService {
             }));
         }
 
-        // Drop the bucket's cached metadata on this peer so it stops serving
-        // stale configuration after the bucket has been deleted.
-        match remove_bucket_metadata(&bucket).await {
+        // Drop the bucket's cached metadata only if the namespace is still
+        // absent. A delayed delete notification must not erase metadata for a
+        // same-name bucket that has already been recreated.
+        let Some(store) = self.resolve_object_store() else {
+            return Ok(Response::new(DeleteBucketMetadataResponse {
+                success: false,
+                error_info: Some("errServerNotInitialized".to_string()),
+            }));
+        };
+        match remove_bucket_metadata_if_absent(store, &bucket).await {
             Ok(_) => Ok(Response::new(DeleteBucketMetadataResponse {
                 success: true,
                 error_info: None,

rustfs/src/storage/storage_api.rs

@@ -275,7 +275,7 @@ pub(crate) mod rpc_consumer {
             SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt,
             StoragePeerS3ClientExt, TierDailyStatsWire, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics,
             find_local_disk_by_ref, get_global_transition_state, get_local_server_property, reload_bucket_metadata,
-            reload_transition_tier_config, remove_bucket_metadata, validate_batch_read_version_item_count,
+            reload_transition_tier_config, remove_bucket_metadata_if_absent, validate_batch_read_version_item_count,
         };
         pub(crate) type StorageResult<T> = super::super::Result<T>;
 
@@ -2010,8 +2010,8 @@ pub(crate) async fn reload_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> R
     ecstore_bucket::metadata_sys::reload_bucket_metadata(api, bucket).await
 }
 
-pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
-    ecstore_bucket::metadata_sys::remove_bucket_metadata(bucket).await
+pub(crate) async fn remove_bucket_metadata_if_absent(api: Arc<ECStore>, bucket: &str) -> Result<bool> {
+    ecstore_bucket::metadata_sys::remove_bucket_metadata_if_absent(api, bucket).await
 }
 
 pub(crate) async fn update_bucket_metadata_config(