perf(s3): resolve bucket versioning once per DeleteObjects request (#5340)

DeleteObjects rebuilt ObjectOptions inside its per-key phase-1 loop, and del_opts re-fetched the bucket versioning configuration on every call: a 1000-key request paid 1001 identical metadata-sys lookups, each taking the global bucket-metadata read lock and cloning a VersioningConfiguration. Split del_opts into the existing async entry point plus a synchronous del_opts_with_versioning that takes an already-resolved configuration, and reuse the request-level version_cfg the handler had already fetched. This also removes a snapshot inconsistency. The skip-stat decision (can_skip_delete_objects_pre_stat) and the post-delete accounting already derived from the request-level snapshot while del_opts re-read the config per key, so a PutBucketVersioning committing mid-request could give a key opts.versioned=true while the batch delete still ran with versioned=false: the object was deleted outright with replication state attached to a delete marker that was never created, and the advisory object-lock pre-check was skipped on a false premise. All three now derive from one snapshot. GET, HEAD and DeleteObject additionally establish bucket existence before building ObjectOptions, so a request naming a nonexistent bucket no longer performs bucket-metadata work first. GET keeps its cheap request-shape validations (key, range, partNumber) ahead of the existence check so InvalidArgument still wins over NoSuchBucket for malformed requests. Bucket validation moves to validate_bucket_exists, which takes an explicit store and shares the existing 5s-TTL cache. GET, PUT, HEAD and DeleteObject now resolve the store through the request-bound server context (backlog#1052 S6) instead of the process-global handle: get_validated_store resolves the first-published global AppContext, so in an embedded multi-instance process a request to the second server operated on the first server's store. Remaining global-handle call sites in bucket_usecase and select_object are left for a follow-up.
2026-07-28 17:18:58 +00:00 · 2026-07-28 06:33:17 +08:00
parent 6d3ce90c0f
commit 5a768d3a44
6 changed files with 319 additions and 47 deletions
@@ -2388,3 +2388,145 @@ async fn object_lock_handlers_schedule_replication() {
         legal hold never replicates and the peer copy stays deletable"
    );
 }
+
+/// Regression pin for the DeleteObjects hot path: the bucket's versioning
+/// configuration must be resolved exactly once per request — not once per
+/// key, which used to cost up to 1000 identical metadata lookups per call.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
+async fn delete_objects_resolves_bucket_versioning_once_per_request() {
+    use super::storage_api::test::{ReqInfo, VERSIONING_CONFIG_LOOKUPS};
+    use std::sync::atomic::Ordering;
+
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+
+    let bucket = format!("test-delobjs-vers-once-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+
+    let keys = ["batch/a.txt", "batch/b.txt", "batch/c.txt"];
+    for key in keys {
+        upload_test_object(&ecstore, bucket.as_str(), key, b"delete objects versioning payload").await;
+    }
+
+    let input = DeleteObjectsInput::builder()
+        .bucket(bucket.clone())
+        .delete(Delete {
+            objects: keys
+                .iter()
+                .map(|key| ObjectIdentifier {
+                    key: key.to_string(),
+                    version_id: None,
+                    ..Default::default()
+                })
+                .collect(),
+            quiet: None,
+        })
+        .build()
+        .expect("delete objects input should build");
+
+    let mut req = build_request(input, Method::POST);
+    req.extensions.insert(ReqInfo {
+        cred: Some(rustfs_credentials::Credentials::default()),
+        is_owner: true,
+        ..Default::default()
+    });
+
+    let lookups_before = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst);
+    let response = usecase
+        .execute_delete_objects(req)
+        .await
+        .expect("delete objects should succeed");
+
+    let output = response.output;
+    let errors = output.errors.unwrap_or_default();
+    assert!(errors.is_empty(), "DeleteObjects reported per-key errors: {errors:?}");
+    assert_eq!(
+        output.deleted.unwrap_or_default().len(),
+        keys.len(),
+        "every requested key should be deleted"
+    );
+
+    let lookups = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst) - lookups_before;
+    assert_eq!(
+        lookups,
+        1,
+        "DeleteObjects must resolve bucket versioning exactly once per request; got {lookups} lookups for {} keys",
+        keys.len()
+    );
+}
+
+/// Regression pin for the nonexistent-bucket fast path: GET/HEAD/DeleteObject
+/// must decide NoSuchBucket from the bucket-existence check, before any bucket
+/// versioning lookup runs — and cheap request-shape validation still wins over
+/// bucket existence for GET.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
+async fn nonexistent_bucket_fails_before_versioning_lookup_on_get_head_delete() {
+    use super::storage_api::test::VERSIONING_CONFIG_LOOKUPS;
+    use std::sync::atomic::Ordering;
+
+    let (_disk_paths, _ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+
+    // Never created.
+    let bucket = format!("test-missing-{}", &Uuid::new_v4().simple().to_string()[..8]);
+
+    // Request-shape validation keeps precedence over bucket existence: GET with
+    // both range and partNumber is InvalidArgument even on a missing bucket.
+    let get_invalid = GetObjectInput::builder()
+        .bucket(bucket.clone())
+        .key("k.txt".to_string())
+        .range(Some(Range::Int { first: 0, last: Some(1) }))
+        .part_number(Some(1))
+        .build()
+        .expect("get input should build");
+    let err = Box::pin(usecase.execute_get_object(build_request(get_invalid, Method::GET)))
+        .await
+        .expect_err("range+partNumber must be rejected");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
+
+    let lookups_before = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst);
+
+    let get = GetObjectInput::builder()
+        .bucket(bucket.clone())
+        .key("k.txt".to_string())
+        .build()
+        .expect("get input should build");
+    let err = Box::pin(usecase.execute_get_object(build_request(get, Method::GET)))
+        .await
+        .expect_err("GET on a missing bucket must fail");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);
+
+    // Bucket existence beats versionId-format validation (which lives in the
+    // opts builders) for HEAD and DeleteObject.
+    let head = HeadObjectInput::builder()
+        .bucket(bucket.clone())
+        .key("k.txt".to_string())
+        .version_id(Some("not-a-uuid".to_string()))
+        .build()
+        .expect("head input should build");
+    let err = Box::pin(usecase.execute_head_object(build_request(head, Method::HEAD)))
+        .await
+        .expect_err("HEAD on a missing bucket must fail");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);
+
+    let del = DeleteObjectInput::builder()
+        .bucket(bucket.clone())
+        .key("k.txt".to_string())
+        .version_id(Some("not-a-uuid".to_string()))
+        .build()
+        .expect("delete input should build");
+    let err = Box::pin(usecase.execute_delete_object(build_request(del, Method::DELETE)))
+        .await
+        .expect_err("DeleteObject on a missing bucket must fail");
+    assert_eq!(err.code(), &s3s::S3ErrorCode::NoSuchBucket);
+
+    let lookups = VERSIONING_CONFIG_LOOKUPS.load(Ordering::SeqCst) - lookups_before;
+    assert_eq!(
+        lookups, 0,
+        "requests naming a nonexistent bucket must fail before any bucket versioning lookup; got {lookups}"
+    );
+}
@@ -81,9 +81,9 @@ use super::storage_api::object_usecase::object_cache::lookup_get_object_body_cac
 use super::storage_api::object_usecase::object_cache::{GetObjectBodyCacheHookLookup, get_object_body_cache_plaintext_len};
 use super::storage_api::object_usecase::object_utils::to_s3s_etag;
 use super::storage_api::object_usecase::options::{
-    copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
-    filter_object_metadata, get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata,
-    normalize_content_encoding_for_storage, put_opts, validate_archive_content_encoding,
+    bucket_versioning_config, copy_dst_opts, copy_src_opts, del_opts, del_opts_with_versioning, extract_metadata,
+    extract_metadata_from_mime_with_object_name, filter_object_metadata, get_content_sha256_with_query, get_opts,
+    namespace_reserved_user_metadata, normalize_content_encoding_for_storage, put_opts, validate_archive_content_encoding,
 };
 use super::storage_api::object_usecase::request_context::{self, spawn_traced};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
@@ -99,9 +99,9 @@ use super::storage_api::object_usecase::storage_class as storageclass;
 use super::storage_api::object_usecase::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper};
 use super::storage_api::object_usecase::{ECStore, OldCurrentSize};
 use super::storage_api::object_usecase::{
-    RFC1123, check_preconditions, get_validated_store, has_replication_rules, parse_object_lock_legal_hold,
-    parse_object_lock_retention, parse_part_number_i32_to_usize, remove_object_lock_metadata_for_copy,
-    strip_managed_encryption_metadata, validate_bucket_object_lock_enabled, validate_object_key, validate_sse_headers_for_read,
+    RFC1123, check_preconditions, has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention,
+    parse_part_number_i32_to_usize, remove_object_lock_metadata_for_copy, strip_managed_encryption_metadata,
+    validate_bucket_exists, validate_bucket_object_lock_enabled, validate_object_key, validate_sse_headers_for_read,
    validate_sse_headers_for_write, validate_ssec_for_read, wrap_response_with_cors,
 };
 use crate::app::runtime_sources::{
@@ -609,6 +609,16 @@ struct GetObjectRequestContext {
    opts: ObjectOptions,
 }

+/// Request fields that passed the cheap GET validations, ready for the
+/// bucket-metadata work in [`DefaultObjectUsecase::prepare_get_object_request_context`].
+struct GetObjectValidatedRequest {
+    bucket: String,
+    key: String,
+    version_id: Option<String>,
+    part_number: Option<usize>,
+    rs: Option<HTTPRangeSpec>,
+}
+
 struct GetObjectReadSetup {
    info: ObjectInfo,
    final_stream: DynReader,
@@ -3552,7 +3562,9 @@ impl DefaultObjectUsecase {
        }
    }

-    async fn prepare_get_object_request_context(req: &S3Request<GetObjectInput>) -> S3Result<GetObjectRequestContext> {
+    /// Cheap request-shape validations, run before the bucket-existence store
+    /// lookup so invalid requests keep their InvalidArgument precedence.
+    fn validate_get_object_request(req: &S3Request<GetObjectInput>) -> S3Result<GetObjectValidatedRequest> {
        // Clone only the fields this path needs instead of the whole input.
        let bucket = req.input.bucket.clone();
        let key = req.input.key.clone();
@@ -3570,7 +3582,28 @@ impl DefaultObjectUsecase {
            return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
        }

-        let opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), part_number, &req.headers)
+        Ok(GetObjectValidatedRequest {
+            bucket,
+            key,
+            version_id,
+            part_number,
+            rs,
+        })
+    }
+
+    async fn prepare_get_object_request_context(
+        validated: GetObjectValidatedRequest,
+        headers: &HeaderMap,
+    ) -> S3Result<GetObjectRequestContext> {
+        let GetObjectValidatedRequest {
+            bucket,
+            key,
+            version_id,
+            part_number,
+            rs,
+        } = validated;
+
+        let opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), part_number, headers)
            .await
            .map_err(ApiError::from)?;

@@ -3588,6 +3621,7 @@ impl DefaultObjectUsecase {
        &self,
        req: &S3Request<GetObjectInput>,
        manager: &'static ConcurrencyManager,
+        store: Arc<ECStore>,
        wrapper: &RequestTimeoutWrapper,
        timeout_config: &GetObjectTimeoutPolicy,
        bucket: &str,
@@ -3596,17 +3630,6 @@ impl DefaultObjectUsecase {
        opts: &ObjectOptions,
        part_number: Option<usize>,
    ) -> S3Result<GetObjectPreparedRead> {
-        // SF05: Store lookup first (cached via SF01 moka cache).
-        let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
-        let store = get_validated_store(bucket).await?;
-        if let Some(store_lookup_start) = store_lookup_start {
-            rustfs_io_metrics::record_get_object_stage_duration(
-                "s3_handler",
-                "store_lookup",
-                store_lookup_start.elapsed().as_secs_f64(),
-            );
-        }
-
        let read_start = std::time::Instant::now();
        let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start);
        let cache_adapter = self.object_data_cache();
@@ -4730,7 +4753,13 @@ impl DefaultObjectUsecase {
            use_large_put_concurrency_tuning,
        );

-        let store = get_validated_store(&bucket).await?;
+        // Resolve the store through the request-bound server context
+        // (backlog#1052 S6), not the process-global handle, so an embedded
+        // second server never writes into the first server's store.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        validate_bucket_exists(&store, &bucket).await?;

        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
        debug!(
@@ -5513,8 +5542,40 @@ impl DefaultObjectUsecase {
        let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
        // mc get 3

+        // Cheap request-shape validations run first so invalid requests keep
+        // their InvalidArgument precedence over bucket existence.
+        let validated = match Self::validate_get_object_request(&req) {
+            Ok(validated) => validated,
+            Err(err) => {
+                lifecycle.finish_err();
+                return Err(err);
+            }
+        };
+
+        // SF05: Store lookup next (5s-TTL bucket-validation cache). Bucket
+        // existence is established before any bucket-metadata work, so requests
+        // naming nonexistent buckets fail before the versioning lookup in
+        // get_opts. The store comes from the request-bound server context
+        // (backlog#1052 S6), not the process-global handle.
+        let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
+        let Some(store) = self.object_store() else {
+            lifecycle.finish_err();
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await {
+            lifecycle.finish_err();
+            return Err(err);
+        }
+        if let Some(store_lookup_start) = store_lookup_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "store_lookup",
+                store_lookup_start.elapsed().as_secs_f64(),
+            );
+        }
+
        let request_context_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
-        let request_context = match Self::prepare_get_object_request_context(&req).await {
+        let request_context = match Self::prepare_get_object_request_context(validated, &req.headers).await {
            Ok(request_context) => request_context,
            Err(err) => {
                lifecycle.finish_err();
@@ -5540,7 +5601,18 @@ impl DefaultObjectUsecase {
        let manager = get_concurrency_manager();

        let prepared_read = match self
-            .prepare_get_object_read_execution(&req, manager, &wrapper, &timeout_config, &bucket, &key, rs, &opts, part_number)
+            .prepare_get_object_read_execution(
+                &req,
+                manager,
+                store,
+                &wrapper,
+                &timeout_config,
+                &bucket,
+                &key,
+                rs,
+                &opts,
+                part_number,
+            )
            .await
        {
            Ok(prepared_read) => prepared_read,
@@ -6474,7 +6546,7 @@ impl DefaultObjectUsecase {
            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
        };

-        let version_cfg = BucketVersioningSys::get(&bucket).await.unwrap_or_default();
+        let version_cfg = bucket_versioning_config(&bucket).await;
        let bypass_governance = has_bypass_governance_header(&req.headers);
        let bucket_lock_enabled = bucket_object_locking_enabled(&bucket).await;

@@ -6561,14 +6633,16 @@ impl DefaultObjectUsecase {
            };

            let metadata = extract_metadata(&req.headers);
-            let opts: ObjectOptions = del_opts(
+            // Reuse the request-level versioning config fetched above instead of
+            // re-resolving it per key (up to 1000 identical lookups per request).
+            let opts: ObjectOptions = del_opts_with_versioning(
                &bucket,
                &object.object_name,
                object.version_id.map(|f| f.to_string()),
                &req.headers,
                metadata,
+                &version_cfg,
            )
-            .await
            .map_err(ApiError::from)?;

            // backlog#929 (HP-8): the accounting branch after the store delete
@@ -6915,6 +6989,16 @@ impl DefaultObjectUsecase {
            authorize_request(&mut req, Action::S3Action(S3Action::ReplicateDeleteAction)).await?;
        }

+        // Establish bucket existence before any bucket-metadata work (matches
+        // PUT/GET): nonexistent buckets fail here instead of paying the
+        // versioning lookups in del_opts/get_opts first. Resolve the store
+        // through the request-bound server context (backlog#1052 S6), not the
+        // process-global handle.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        validate_bucket_exists(&store, &bucket).await?;
+
        let metadata = extract_metadata(&req.headers);
        // Clone version_id before it's moved
        let version_id_clone = version_id.clone();
@@ -6952,9 +7036,6 @@ impl DefaultObjectUsecase {
            ));
        }
        validate_undo_delete_version(expected_current_version_id.as_deref(), opts.version_id.as_deref())?;
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
        opts.expected_current_version_id = expected_current_version_id.clone();

        let replicate_force_delete = force_delete
@@ -7176,13 +7257,20 @@ impl DefaultObjectUsecase {
            return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
        }

+        // Establish bucket existence before any bucket-metadata work (matches
+        // PUT/GET): nonexistent buckets fail here instead of paying the
+        // versioning lookup in get_opts first. Resolve the store through the
+        // request-bound server context (backlog#1052 S6), not the
+        // process-global handle.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        validate_bucket_exists(&store, &bucket).await?;
+
        let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
            .await
            .map_err(ApiError::from)?;

-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
        // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors
        let info = match store.get_object_info(&bucket, &key, &opts).await {
            Ok(info) => info,
@@ -865,11 +865,13 @@ pub(crate) mod io {
 }

 pub(crate) mod options {
+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::options_consumer::VERSIONING_CONFIG_LOOKUPS;
    pub(crate) use crate::storage::storage_api::options_consumer::{
-        copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime,
-        extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
-        get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
-        parse_copy_source_range, put_opts, validate_archive_content_encoding,
+        bucket_versioning_config, copy_dst_opts, copy_src_opts, del_opts, del_opts_with_versioning, extract_metadata,
+        extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata,
+        get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata,
+        normalize_content_encoding_for_storage, parse_copy_source_range, put_opts, validate_archive_content_encoding,
    };
 }

@@ -993,8 +995,8 @@ pub(crate) mod object_usecase {
    pub(crate) use crate::storage::storage_api::{
        ECStore, GetObjectReader, OldCurrentSize, RFC1123, StorageDeletedObject, StorageObjectInfo,
        StorageObjectLockDeleteOptions, StorageObjectOptions, StorageObjectToDelete, StoragePutObjReader, check_preconditions,
-        get_validated_store, has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention,
-        parse_part_number_i32_to_usize, remove_object_lock_metadata_for_copy, strip_managed_encryption_metadata,
+        has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize,
+        remove_object_lock_metadata_for_copy, strip_managed_encryption_metadata, validate_bucket_exists,
        validate_bucket_object_lock_enabled, validate_object_key, validate_sse_headers_for_read, validate_sse_headers_for_write,
        validate_ssec_for_read, wrap_response_with_cors,
    };
@@ -1074,6 +1076,8 @@ pub(crate) mod test {
        }
    }

+    pub(crate) use super::access::ReqInfo;
+    pub(crate) use super::options::VERSIONING_CONFIG_LOOKUPS;
    pub(crate) use super::{bucket, data_usage, ecfs, object_utils, runtime};
    pub(crate) use crate::storage::storage_api::{
        ECStore, Endpoint, Endpoints, PoolEndpoints, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader,
@@ -843,11 +843,21 @@ pub(crate) async fn get_validated_store(bucket: &str) -> S3Result<Arc<super::ECS
        return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
    };

+    validate_bucket_exists(&store, bucket).await?;
+
+    Ok(store)
+}
+
+/// Validate that `bucket` exists on `store`, using the same adaptive 5s-TTL
+/// positive cache as [`get_validated_store`]. Handlers that resolve their
+/// store through the request-bound server context (backlog#1052 S6) use this
+/// to keep bucket validation on that store instead of the process-global one.
+pub(crate) async fn validate_bucket_exists(store: &super::ECStore, bucket: &str) -> S3Result<()> {
    // Check cache — TTL is checked manually.
    if let Some(inserted_at) = cache_get(bucket)
        && inserted_at.elapsed() < BUCKET_VALIDATION_TTL
    {
-        return Ok(store); // Cache hit, skip validation
+        return Ok(()); // Cache hit, skip validation
    }

    // Cache miss or expired, perform validation
@@ -859,7 +869,7 @@ pub(crate) async fn get_validated_store(bucket: &str) -> S3Result<Arc<super::ECS
    // Update cache
    cache_insert(bucket.to_string(), Instant::now());

-    Ok(store)
+    Ok(())
 }

 /// Quick check if CORS processing is needed (lightweight check for Origin header)
@@ -49,9 +49,21 @@ use crate::storage::storage_api::ecstore_bucket::versioning::VersioningApi as _;
 use crate::storage::storage_api::options_consumer::StorageObjectOptions as ObjectOptions;
 use s3s::dto::VersioningConfiguration;

+/// Test-only counter of versioning-config fetches, used to pin that batch
+/// handlers resolve the configuration once per request, not once per key
+/// (same counting pattern as MUST_REPLICATE_OBJECT_CALLS).
+///
+/// Blind spot: only fetches routed through [`bucket_versioning_config`] are
+/// counted — handler code that calls `BucketVersioningSys` directly bypasses
+/// this seam and its regression pins.
+#[cfg(test)]
+pub(crate) static VERSIONING_CONFIG_LOOKUPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
 /// Fetch the bucket's versioning configuration once so callers can derive
 /// enabled/suspended state without repeated metadata-sys lookups per request.
-async fn bucket_versioning_config(bucket: &str) -> VersioningConfiguration {
+pub(crate) async fn bucket_versioning_config(bucket: &str) -> VersioningConfiguration {
+    #[cfg(test)]
+    VERSIONING_CONFIG_LOOKUPS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    match BucketVersioningSys::get(bucket).await {
        Ok(cfg) => cfg,
        Err(err) => {
@@ -97,6 +109,20 @@ pub async fn del_opts(
    metadata: HashMap<String, String>,
 ) -> Result<ObjectOptions> {
    let versioning_cfg = bucket_versioning_config(bucket).await;
+    del_opts_with_versioning(bucket, object, vid, headers, metadata, &versioning_cfg)
+}
+
+/// Like [`del_opts`], but derives versioning state from an already-fetched
+/// configuration so batch callers (DeleteObjects) resolve the bucket's
+/// versioning once per request instead of once per key.
+pub fn del_opts_with_versioning(
+    bucket: &str,
+    object: &str,
+    vid: Option<String>,
+    headers: &HeaderMap<HeaderValue>,
+    metadata: HashMap<String, String>,
+    versioning_cfg: &VersioningConfiguration,
+) -> Result<ObjectOptions> {
    let versioned = versioning_cfg.prefix_enabled(object);
    let version_suspended = versioning_cfg.suspended();

@@ -126,7 +152,7 @@ pub async fn del_opts(
        None
    };

-    let mut opts = put_opts_from_headers(headers, metadata.clone()).map_err(|err| {
+    let mut opts = put_opts_from_headers(headers, metadata).map_err(|err| {
        error!("del_opts: invalid argument: {} error: {}", object, err);
        StorageError::InvalidArgument(bucket.to_owned(), object.to_owned(), err.to_string())
    })?;
@@ -91,8 +91,8 @@ pub(crate) use super::ecfs_extend::{
    RFC1123, apply_bucket_default_lock_retention, apply_cors_headers, check_preconditions, get_buffer_size_opt_in,
    get_validated_store, has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention,
    parse_part_number_i32_to_usize, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
-    remove_object_lock_metadata_for_copy, validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter,
-    validate_object_key, wrap_response_with_cors,
+    remove_object_lock_metadata_for_copy, validate_bucket_exists, validate_bucket_object_lock_enabled,
+    validate_list_object_unordered_with_delimiter, validate_object_key, wrap_response_with_cors,
 };
 pub(crate) use super::sse::{
    DecryptionRequest, EncryptionRequest, PrepareEncryptionRequest, extract_server_side_encryption_from_headers, sse_decryption,
@@ -180,11 +180,13 @@ pub(crate) mod helper_consumer {
 }

 pub(crate) mod options_consumer {
+    #[cfg(test)]
+    pub(crate) use super::super::options::VERSIONING_CONFIG_LOOKUPS;
    pub(crate) use super::super::options::{
-        copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime,
-        extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
-        get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
-        parse_copy_source_range, put_opts, validate_archive_content_encoding,
+        bucket_versioning_config, copy_dst_opts, copy_src_opts, del_opts, del_opts_with_versioning, extract_metadata,
+        extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata,
+        get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata,
+        normalize_content_encoding_for_storage, parse_copy_source_range, put_opts, validate_archive_content_encoding,
    };

    pub(crate) mod contract {