test(ecstore): avoid lock cleanup refcount race

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

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

.config/nextest.toml

@@ -234,7 +234,7 @@ test-group = 'embedded-test-ports'
 # processes cannot exhaust those budgets. Keep deadlines and assertions intact.
 # The short HTTP fixtures are tracked in rustfs/backlog#2714.
 [[profile.default.overrides]]
-filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|oversized_response_is_rejected_without_buffering_it)$/)))'
+filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|cleanup_accepts_missing_object_and_still_deletes_the_bucket|cleanup_diagnostics_identify_each_disconnected_transport_stage|cleanup_diagnostics_identify_each_failed_http_stage_without_private_context|cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_deadline|oversized_response_is_rejected_without_buffering_it)$/)))'
 threads-required = "num-test-threads"
 
 # This cancellation probe hashes the serving executable before acquiring its
@@ -548,7 +548,7 @@ filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnost
 test-group = 'embedded-test-ports'
 
 [[profile.ci.overrides]]
-filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|oversized_response_is_rejected_without_buffering_it)$/)))'
+filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|cleanup_accepts_missing_object_and_still_deletes_the_bucket|cleanup_diagnostics_identify_each_disconnected_transport_stage|cleanup_diagnostics_identify_each_failed_http_stage_without_private_context|cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_deadline|oversized_response_is_rejected_without_buffering_it)$/)))'
 threads-required = "num-test-threads"
 
 # Preserve the cancellation probe's startup budget and assertions in CI too.
@@ -976,6 +976,12 @@ test-group = 'e2e-vault'
 filter = 'package(e2e_test) & (test(/^list_objects_v2_pagination_test::tests::(test_list_objects_v2_delimiter_small_page_traverses_all|test_list_objects_v2_max_keys_above_limit_returns_token|test_list_objects_v2_maxkeys_above_limit_with_delimiter)$/) | test(=reliant::tiering::test_hermetic_transition_restore_failure_expiry_and_retry))'
 threads-required = "num-test-threads"
 
+# The mc fixture uploads hundreds of objects concurrently before measuring listing.
+# Reserve the runner so other server fixtures do not starve its disk operations.
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(=mc_mirror_small_bucket_test::test_mc_mirror_small_bucket_completes_without_list_timeout)'
+threads-required = "num-test-threads"
+
 # Fixed, small contract for every executable server change. These existing
 # scenarios exercise write/read, anonymous denial and retained bytes on restart.
 [profile.core-contract]

crates/ecstore/src/set_disk/mod.rs

@@ -9038,6 +9038,13 @@ mod tests {
             "constructing an object lock must not clone each client Arc"
         );
 
+        // Check shared ownership before acquisition can spawn pending cleanup.
+        let read_lock = set
+            .new_ns_lock("bucket", "read-object")
+            .await
+            .expect("second namespace lock should be created");
+        assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2);
+
         let write_error = write_lock
             .get_write_lock(Duration::from_millis(500))
             .await
@@ -9052,11 +9059,6 @@ mod tests {
             ),
             "the shared client representation must preserve the exact write quorum result: {write_error}"
         );
-        let read_lock = set
-            .new_ns_lock("bucket", "read-object")
-            .await
-            .expect("second namespace lock should be created");
-        assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2);
         let read_guard = read_lock
             .get_read_lock(Duration::from_millis(500))
             .await

crates/ecstore/src/store/multipart.rs

@@ -19,6 +19,7 @@ use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::multipart::MultipartOperations as _;
 use crate::storage_api_contracts::object::ObjectOperations as _;
 use futures::{StreamExt, stream};
+use rustfs_utils::path::encode_dir_object_ref;
 use std::collections::HashSet;
 
 const MULTIPART_LIST_SET_CONCURRENCY: usize = 4;
@@ -1168,11 +1169,9 @@ impl ECStore {
         let oi = if require_all_pools {
             lookup_opts.metadata_cache_safe = false;
             lookup_opts.suppress_read_repair = true;
-            if let Err(err) = check_object_args(bucket, object) {
-                return Some(err);
-            }
-            let encoded_object = encode_dir_object(object);
-            self.get_latest_object_info_for_conditional_write(bucket, &encoded_object, &lookup_opts)
+            // Only completion uses this branch, after validating bucket and object.
+            let encoded_object = encode_dir_object_ref(object);
+            self.get_latest_object_info_for_conditional_write(bucket, encoded_object.as_ref(), &lookup_opts)
                 .await
                 .and_then(|(info, _)| {
                     lookup_opts.precondition_check(&info)?;