fix: refresh peer metadata without blocking bucket mutations (#8486)

* fix: refresh peer bucket metadata after creation

* fix(s3): release bucket mutation slots before peer reloads

* test(ecstore): distinguish delete lock waits from disk latency

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

@@ -314,6 +314,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");
@@ -4856,6 +4883,117 @@ mod tests {
         );
     }
 
+    #[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/set_disk/ops/object.rs

@@ -22999,22 +22999,43 @@ 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");

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/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(