diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index fc26177a9..ee36d0b76 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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}; diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index e8553f433..1dc3e43e1 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -314,6 +314,33 @@ pub async fn remove_bucket_metadata(bucket: &str) -> Result { 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, bucket: &str) -> Result { + 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>) { 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() { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c4e25a5da..fba95e548 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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"); diff --git a/docs/operations/bucket-operation-admission.md b/docs/operations/bucket-operation-admission.md index bab36f986..3e46f1bb7 100644 --- a/docs/operations/bucket-operation-admission.md +++ b/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: diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 074ec0c66..445978a09 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/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> = Laz struct BucketOperationAdmission { active: Arc, waiting: Semaphore, + in_flight: Arc, } 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 { + 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, 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, + scanner_maintenance_change: bool, + app_context: Option>, ) { 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>, } -async fn await_bucket_usecase_on_fresh_task(operation: &'static str, future: F) -> S3Result +async fn await_bucket_usecase_on_fresh_task(operation: &'static str, make_future: MakeFuture) -> S3Result where T: Send + 'static, F: Future> + 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( +async fn await_bucket_usecase_on_fresh_task_with_admission( operation: &'static str, admission: Arc, - future: F, + make_future: MakeFuture, ) -> S3Result where T: Send + 'static, F: Future> + 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) -> S3Result> { 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) -> S3Result> { + async fn execute_create_bucket_inner( + &self, + req: S3Request, + mutation_permit: OwnedSemaphorePermit, + ) -> S3Result> { let helper = OperationHelper::new(&req, EventName::BucketCreated, S3Operation::CreateBucket); + let request_context = req.extensions.get::().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) -> S3Result> { 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(input: T, method: Method) -> S3Request { @@ -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] diff --git a/rustfs/src/app/runtime_sources.rs b/rustfs/src/app/runtime_sources.rs index ad7c7f949..389d97517 100644 --- a/rustfs/src/app/runtime_sources.rs +++ b/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; diff --git a/rustfs/src/storage/rpc/node_service/bucket.rs b/rustfs/src/storage/rpc/node_service/bucket.rs index e7f8f82ae..40317d898 100644 --- a/rustfs/src/storage/rpc/node_service/bucket.rs +++ b/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, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 67c0e586a..c0a9608d5 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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 = super::super::Result; @@ -2010,8 +2010,8 @@ pub(crate) async fn reload_bucket_metadata(api: Arc, bucket: &str) -> R ecstore_bucket::metadata_sys::reload_bucket_metadata(api, bucket).await } -pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result { - ecstore_bucket::metadata_sys::remove_bucket_metadata(bucket).await +pub(crate) async fn remove_bucket_metadata_if_absent(api: Arc, bucket: &str) -> Result { + ecstore_bucket::metadata_sys::remove_bucket_metadata_if_absent(api, bucket).await } pub(crate) async fn update_bucket_metadata_config(