test(table-catalog): fold the admin object backend into the shared fake

Second half of PR2: the admin handler tests' TestTableCatalogObjectBackend (struct, inherent impl, trait impl, lock alias — 201 lines) is deleted and its instrumentation folded into the shared TestCatalogObjectBackend, which the admin tests now take through a type alias so no call site is renamed.

Two behavioral differences between the two fakes were found by the test suites rather than assumed away, and both are preserved:

- Lock observability: the admin fake implemented only acquire_write_lock, so the trait's default acquire_read_lock -> acquire_write_lock delegation made read acquisitions visible in lock_attempts. The shared fake implements both independently, so five fence/lock tests timed out until the read path also records attempts.

- Etag generation: the admin fake used content-addressed sha256 etags (its tests observe an etag and expect rewriting identical bytes to reproduce it) while the store fake uses an incrementing counter. Instead of silently picking one, the union carries a content_addressed_etags flag; the 80 admin construction sites go through TestCatalogObjectBackend::content_addressed() and the store tests keep counter semantics.

The six one-shot path-keyed injection knobs (fail/corrupt put, missing/fail read, put barrier) run before the store fake's attempt-indexed injection maps, matching each fake's original ordering.

Verification: cargo test -p rustfs --lib table_catalog 481 passed; --lib admin::handlers::table_catalog 167 passed; clippy --lib --tests -D warnings clean; make pre-commit green.

Ref rustfs/backlog#1837 (PR2, part 2).

crates/ecstore/src/config/storageclass.rs

@@ -101,7 +101,6 @@ const DEFAULT_RRS_STORAGE_CLASS: &str = "EC:1";
 const ZERO_SET_DRIVE_COUNT_ERROR: &str = "set drive count must be greater than zero";
 
 pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
-const DEFAULT_INLINE_OBJECT_BUDGET: usize = 2 * DEFAULT_INLINE_BLOCK;
 
 pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
     let kvs = vec![
@@ -151,8 +150,6 @@ pub struct Config {
     optimize: Option<String>,
     inline_block: usize,
     initialized: bool,
-    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
-    inline_block_explicit: bool,
     #[serde(skip)]
     standard_parities: Vec<PoolParity>,
     #[serde(skip)]
@@ -236,19 +233,17 @@ impl Config {
             .map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
     }
 
-    pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool {
-        if shard_size < 0 || data_shards == 0 {
+    pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
+        if shard_size < 0 {
             return false;
         }
 
         let shard_size = shard_size as usize;
-        // Keep the historical two-data-shard object budget while preventing
-        // wider EC layouts from multiplying the maximum inline object size.
-        let inline_block = if self.initialized && self.inline_block_explicit {
-            self.inline_block
-        } else {
-            (DEFAULT_INLINE_OBJECT_BUDGET / data_shards).min(DEFAULT_INLINE_BLOCK)
-        };
+
+        let mut inline_block = DEFAULT_INLINE_BLOCK;
+        if self.initialized {
+            inline_block = self.inline_block;
+        }
 
         if versioned {
             shard_size <= inline_block / 8
@@ -397,7 +392,6 @@ fn lookup_config_for_pools_with_env(
     }
 
     let optimize = overrides.optimize;
-    let inline_block_explicit = overrides.inline_block.is_some();
     let inline_block = if let Some(value) = overrides.inline_block {
         let block = value
             .parse::<bytesize::ByteSize>()
@@ -430,7 +424,6 @@ fn lookup_config_for_pools_with_env(
         optimize,
         inline_block,
         initialized: true,
-        inline_block_explicit,
         standard_parities,
         rrs_parities,
     })
@@ -548,26 +541,22 @@ mod tests {
     }
 
     #[test]
-    fn should_inline_scales_default_threshold_by_data_shards() {
-        let config = lookup_config_for_pools_with_env(&KVS::new(), &[3, 12], no_env_overrides())
-            .expect("default inline policy should resolve for EC2+1 and EC8+4");
+    fn should_inline_preserves_exact_default_shard_boundaries() {
+        let config = Config::default();
 
-        for (case, shard_size, data_shards, versioned, expected) in [
-            ("EC2+1 unversioned exact", 128 * 1024, 2, false, true),
-            ("EC2+1 unversioned above", 128 * 1024 + 1, 2, false, false),
-            ("EC2+1 versioned exact", 16 * 1024, 2, true, true),
-            ("EC2+1 versioned above", 16 * 1024 + 1, 2, true, false),
-            ("EC8+4 unversioned exact", 32 * 1024, 8, false, true),
-            ("EC8+4 unversioned above", 32 * 1024 + 1, 8, false, false),
-            ("EC8+4 versioned exact", 4 * 1024, 8, true, true),
-            ("EC8+4 versioned above", 4 * 1024 + 1, 8, true, false),
-            ("negative", -1, 2, false, false),
-            ("zero data shards", 0, 0, false, false),
+        for (case, shard_size, versioned, expected) in [
+            ("unversioned below", 128 * 1024 - 1, false, true),
+            ("unversioned exact", 128 * 1024, false, true),
+            ("unversioned above", 128 * 1024 + 1, false, false),
+            ("versioned below", 16 * 1024 - 1, true, true),
+            ("versioned exact", 16 * 1024, true, true),
+            ("versioned above", 16 * 1024 + 1, true, false),
+            ("negative", -1, false, false),
         ] {
             assert_eq!(
-                config.should_inline(shard_size, data_shards, versioned),
+                config.should_inline(shard_size, versioned),
                 expected,
-                "{case}: shard_size={shard_size}, data_shards={data_shards}, versioned={versioned}"
+                "{case}: shard_size={shard_size}, versioned={versioned}"
             );
         }
     }
@@ -588,28 +577,13 @@ mod tests {
             let shard_size = erasure.shard_file_size(object_size);
             assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}");
             assert_eq!(
-                config.should_inline(shard_size, erasure.data_shards, versioned),
+                config.should_inline(shard_size, versioned),
                 expected,
                 "{case}: object_size={object_size}, shard_size={shard_size}, versioned={versioned}"
             );
         }
     }
 
-    #[test]
-    fn explicit_inline_block_preserves_fixed_per_shard_rollback() {
-        let overrides = StorageClassEnvOverrides {
-            inline_block: Some("128KiB".to_string()),
-            ..Default::default()
-        };
-        let config = lookup_config_for_pools_with_env(&KVS::new(), &[12], overrides)
-            .expect("explicit inline block should resolve for EC8+4");
-
-        assert!(config.should_inline(128 * 1024, 8, false));
-        assert!(!config.should_inline(128 * 1024 + 1, 8, false));
-        assert!(config.should_inline(16 * 1024, 8, true));
-        assert!(!config.should_inline(16 * 1024 + 1, 8, true));
-    }
-
     #[test]
     fn write_capability_contract_only_accepts_implemented_layouts() {
         assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]);
@@ -803,7 +777,6 @@ mod tests {
         let encoded = serde_json::to_string(&cfg).expect("config should serialize");
         assert!(!encoded.contains("standard_parities"));
         assert!(!encoded.contains("rrs_parities"));
-        assert!(!encoded.contains("inline_block_explicit"));
 
         let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize");
         assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2));
@@ -813,25 +786,6 @@ mod tests {
         assert!(validate_parity(0, 0).is_err());
     }
 
-    #[test]
-    fn explicit_inline_block_survives_config_round_trip() {
-        let cfg = lookup_config_for_pools_with_env(
-            &KVS::new(),
-            &[12],
-            StorageClassEnvOverrides {
-                inline_block: Some("128KiB".to_string()),
-                ..Default::default()
-            },
-        )
-        .expect("explicit inline block should resolve");
-        assert!(cfg.should_inline(100 * 1024, 8, false));
-
-        let encoded = serde_json::to_string(&cfg).expect("config should serialize");
-        assert!(encoded.contains("\"inline_block_explicit\":true"));
-        let decoded: Config = serde_json::from_str(&encoded).expect("explicit inline config should deserialize");
-        assert!(decoded.should_inline(100 * 1024, 8, false));
-    }
-
     #[test]
     fn lookup_config_reads_rrs_from_class_rrs_key() {
         // Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of

crates/ecstore/src/runtime/sources.rs

@@ -236,6 +236,10 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize>
     get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default())
 }
 
+pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool {
+    get_global_storage_class_snapshot().should_inline(shard_size, versioned)
+}
+
 pub(crate) fn deployment_upload_id(upload_id: &str) -> String {
     base64_simd::URL_SAFE_NO_PAD
         .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())

crates/ecstore/src/set_disk/mod.rs

@@ -880,44 +880,12 @@ mod prepared_get_object_metadata_tests {
     use super::*;
     use crate::ecstore_validation_blackbox::make_local_set_disks;
     use crate::object_api::{BLOCK_SIZE_V2, PutObjReader};
-    use crate::set_disk::core::io_primitives::{bounded_metadata_fanout_order, disk_call_counters, rename_fanout_barrier};
+    use crate::set_disk::core::io_primitives::disk_call_counters;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
-    use crate::test_metrics::CapturingRecorder;
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
-    const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
-
-    fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
-        (0..1000)
-            .map(|index| format!("{prefix}-{index}.bin"))
-            .find(|name| {
-                let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
-                let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
-                let mut seen = [false; 2];
-                for disk_index in order.into_iter().take(3) {
-                    if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
-                        seen[block_index - 1] = true;
-                    }
-                }
-                seen.into_iter().all(|seen| seen)
-            })
-            .expect("test should find an object whose initial fanout covers both data shards")
-    }
-
-    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
-        *bounded_metadata_fanout_order(bucket, object, 4, 2)
-            .get(3)
-            .expect("4-disk test geometry should leave one bounded spare disk")
-    }
-
-    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
-        *bounded_metadata_fanout_order(bucket, object, 4, 2)
-            .get(2)
-            .expect("4-disk test geometry should include a third initial metadata disk")
-    }
-
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1034,307 +1002,6 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
-    #[test]
-    #[serial_test::serial(body_cache_hook)]
-    fn inline_data_read_early_stop_reader_returns_exact_body() {
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("current-thread runtime should build");
-        let bucket = "inline-data-read-early-stop-reader";
-        let object = object_with_initial_data_shards(bucket, "inline-data-read-early-stop-reader-object");
-        let payload = b"inline early-stop reader payload".repeat(256);
-        let recorder = CapturingRecorder::default();
-        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
-        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-
-        let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
-            runtime.block_on(async {
-                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
-                let opts = ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                };
-
-                set_disks
-                    .make_bucket(bucket, &MakeBucketOptions::default())
-                    .await
-                    .expect("bucket should be created");
-                let mut put_reader = PutObjReader::from_vec(payload.clone());
-                set_disks
-                    .put_object(bucket, &object, &mut put_reader, &opts)
-                    .await
-                    .expect("inline object should be written");
-
-                temp_env::async_with_vars(
-                    [
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
-                    ],
-                    async {
-                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
-                        let barrier =
-                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
-                        let calls = disk_call_counters::observe(&object);
-                        let set_disks_for_read = Arc::clone(&set_disks);
-                        let opts_for_read = opts.clone();
-                        let object_for_read = object.clone();
-                        let mut open_reader = tokio::spawn(async move {
-                            set_disks_for_read
-                                .get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts_for_read)
-                                .await
-                        });
-
-                        tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
-                            .await
-                            .expect("bounded inline GET should pause a slow initial metadata read");
-                        let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
-                            .await
-                            .expect("production inline GET should return before the paused metadata response")
-                            .expect("inline GET reader task should not panic")
-                            .expect("inline GET reader should open");
-                        let object_size = reader.object_info.size;
-                        let mut restored = Vec::new();
-                        reader
-                            .stream
-                            .read_to_end(&mut restored)
-                            .await
-                            .expect("inline GET body should stream");
-
-                        (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
-                    },
-                )
-                .await
-            })
-        });
-        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
-
-        assert_eq!(object_size, payload.len() as i64);
-        assert_eq!(restored, payload);
-        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_scheduled",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![4.0],
-            "bounded production GET should record all scheduled metadata tasks"
-        );
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_completed",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![3.0],
-            "bounded production GET should record only observed metadata responses as completed"
-        );
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_cancelled",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![1.0],
-            "bounded production GET should record the aborted slow metadata task"
-        );
-    }
-
-    #[test]
-    #[serial_test::serial(body_cache_hook)]
-    fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("current-thread runtime should build");
-        let bucket = "prepared-metadata-early-stop-enabled";
-        let object = object_with_initial_data_shards(bucket, "prepared-metadata-early-stop-enabled-object");
-        let payload = b"prepared metadata early-stop enabled payload".repeat(16);
-        let recorder = CapturingRecorder::default();
-        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
-        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-
-        let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
-            runtime.block_on(async {
-                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
-                let opts = ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                };
-
-                set_disks
-                    .make_bucket(bucket, &MakeBucketOptions::default())
-                    .await
-                    .expect("bucket should be created");
-                let mut put_reader = PutObjReader::from_vec(payload.clone());
-                set_disks
-                    .put_object(bucket, &object, &mut put_reader, &opts)
-                    .await
-                    .expect("object should be written");
-
-                temp_env::async_with_vars(
-                    [
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
-                    ],
-                    async {
-                        let calls = disk_call_counters::observe(&object);
-                        let metadata = set_disks
-                            .prepare_get_object_metadata(bucket, &object, &opts)
-                            .await
-                            .expect("prepared metadata should resolve");
-                        let calls_total = calls.total(disk_call_counters::KIND_READ_VERSION);
-
-                        let mut reader = set_disks
-                            .get_object_reader_with_prepared_metadata(bucket, &object, None, HeaderMap::new(), &opts, metadata)
-                            .await
-                            .expect("prepared body reader should open");
-                        let mut restored = Vec::new();
-                        reader
-                            .stream
-                            .read_to_end(&mut restored)
-                            .await
-                            .expect("prepared body should stream");
-                        (restored, calls_total)
-                    },
-                )
-                .await
-            })
-        });
-        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
-
-        assert_eq!(restored, payload);
-        assert_eq!(
-            calls_total, 4,
-            "prepared metadata must opt out of data-read early-stop until the read shape is known"
-        );
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_scheduled",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![4.0],
-            "prepared metadata should schedule the full metadata fanout"
-        );
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_completed",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![4.0],
-            "prepared metadata must wait for every scheduled metadata response"
-        );
-        assert_eq!(
-            recorder.histogram_values(
-                "rustfs_io_get_object_metadata_fanout_cancelled",
-                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-            ),
-            vec![0.0],
-            "prepared metadata must not cancel metadata responses"
-        );
-    }
-
-    #[test]
-    #[serial_test::serial(body_cache_hook)]
-    fn data_read_early_stop_request_shapes_full_wait_in_production_reader() {
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("current-thread runtime should build");
-        let bucket = "data-read-early-stop-shape-reader";
-        let payload = b"shape-gated inline reader payload".repeat(256);
-
-        for (object_prefix, range, configure_opts, expected_body) in [
-            (
-                "data-read-early-stop-range-reader-object",
-                Some(HTTPRangeSpec {
-                    start: 0,
-                    end: 3,
-                    is_suffix_length: false,
-                }),
-                None,
-                payload[..4].to_vec(),
-            ),
-            ("data-read-early-stop-part-reader-object", None, Some(1), payload.clone()),
-        ] {
-            let recorder = CapturingRecorder::default();
-            let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
-            rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-            let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
-                runtime.block_on(async {
-                    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
-                    let object = object_with_initial_data_shards(bucket, object_prefix);
-                    let mut opts = ObjectOptions {
-                        no_lock: true,
-                        ..Default::default()
-                    };
-                    opts.part_number = configure_opts;
-
-                    set_disks
-                        .make_bucket(bucket, &MakeBucketOptions::default())
-                        .await
-                        .expect("bucket should be created");
-                    let mut put_reader = PutObjReader::from_vec(payload.clone());
-                    set_disks
-                        .put_object(bucket, &object, &mut put_reader, &opts)
-                        .await
-                        .expect("inline object should be written");
-
-                    temp_env::async_with_vars(
-                        [
-                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
-                        ],
-                        async {
-                            let calls = disk_call_counters::observe(&object);
-                            let mut reader = set_disks
-                                .get_object_reader(bucket, &object, range, HeaderMap::new(), &opts)
-                                .await
-                                .expect("shape-gated GET reader should open");
-                            let mut restored = Vec::new();
-                            reader
-                                .stream
-                                .read_to_end(&mut restored)
-                                .await
-                                .expect("shape-gated GET body should stream");
-                            (restored, calls.total(disk_call_counters::KIND_READ_VERSION))
-                        },
-                    )
-                    .await
-                })
-            });
-            rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
-
-            assert_eq!(restored, expected_body);
-            assert_eq!(calls_total, 4, "shape-gated production GET should keep full metadata fanout");
-            assert_eq!(
-                recorder.histogram_values(
-                    "rustfs_io_get_object_metadata_fanout_scheduled",
-                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-                ),
-                vec![4.0],
-                "shape-gated production GET should schedule the full metadata fanout"
-            );
-            assert_eq!(
-                recorder.histogram_values(
-                    "rustfs_io_get_object_metadata_fanout_completed",
-                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-                ),
-                vec![4.0],
-                "shape-gated production GET must wait for every scheduled metadata response"
-            );
-            assert_eq!(
-                recorder.histogram_values(
-                    "rustfs_io_get_object_metadata_fanout_cancelled",
-                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
-                ),
-                vec![0.0],
-                "shape-gated production GET must not cancel metadata responses"
-            );
-        }
-    }
-
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_rebuilds_object_info_when_precomputed_value_is_absent() {
@@ -1438,7 +1105,7 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
         let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
             snapshot,
@@ -2741,12 +2408,12 @@ mod write_layout_tests {
 
         let held_layout = resolve_write_layout(&held, 0, 4, 2, None, false).expect("held snapshot should remain valid");
         assert_eq!(held_layout.parity_drives, 2);
-        assert!(held.should_inline(512, held_layout.data_drives, false));
+        assert!(held.should_inline(512, false));
 
         let current = published.load_full();
         let current_layout = resolve_write_layout(&current, 0, 4, 2, None, false).expect("new snapshot should resolve");
         assert_eq!(current_layout.parity_drives, 1);
-        assert!(!current.should_inline(512, current_layout.data_drives, false));
+        assert!(!current.should_inline(512, false));
     }
 }
 
@@ -3744,15 +3411,9 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
-        if file_info.erasure.index != block_index {
-            continue;
-        }
         if !file_info.has_valid_erasure_geometry() {
             continue;
         }
-        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
-            continue;
-        }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
             continue;
         }
@@ -9560,9 +9221,6 @@ mod tests {
             HashAlgorithm::HighwayHash256S
         };
         let shards = erasure.encode_data(payload).expect("payload should encode");
-        let version_id = Some(Uuid::new_v4());
-        let data_dir = Some(Uuid::new_v4());
-        let mod_time = Some(OffsetDateTime::now_utc());
         let mut files = Vec::with_capacity(shards.len());
 
         for shard in shards {
@@ -9575,16 +9233,6 @@ mod tests {
             writer.shutdown().await.expect("inline writer should shutdown");
             let data = writer.into_inline_data().expect("inline data should be retained");
             let mut file = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-            file.volume = "bucket".to_string();
-            file.name = "object".to_string();
-            file.size = i64::try_from(payload.len()).expect("test payload should fit i64");
-            file.is_latest = true;
-            file.version_id = version_id;
-            file.data_dir = data_dir;
-            file.mod_time = mod_time;
-            file.metadata.insert("etag".to_string(), "etag-inline".to_string());
-            file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None);
-            file.set_inline_data();
             file.erasure.index = files.len() + 1;
             file.data = Some(Bytes::from(data));
             files.push(file);
@@ -9597,40 +9245,16 @@ mod tests {
         inline_bitrot_files_for_payload_with_mode(payload, false).await
     }
 
-    fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec<FileInfo> {
-        let distribution = &files
-            .first()
-            .expect("inline data shard fixture should include metadata")
-            .erasure
-            .distribution;
-        distribution
-            .iter()
-            .map(|block_index| {
-                files
-                    .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
-                    .expect("inline data shard fixture should include every distributed shard")
-                    .clone()
-            })
-            .collect()
-    }
-
     fn inline_data_shard_fileinfo(
+        name: &str,
         data_blocks: usize,
         parity_blocks: usize,
         erasure_index: usize,
         distribution: &[usize],
         data: Option<&'static [u8]>,
     ) -> FileInfo {
-        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
-        fi.name = "object".to_string();
-        fi.volume = "bucket".to_string();
-        fi.size = 4;
-        fi.is_latest = true;
-        fi.data_dir = Some(Uuid::nil());
-        fi.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
-        fi.metadata.insert("etag".to_string(), "etag-inline".to_string());
-        fi.add_object_part(1, "part-etag-inline".to_string(), 4, fi.mod_time, 4, None, None);
-        fi.set_inline_data();
+        let mut fi = FileInfo::new(name, data_blocks, parity_blocks);
+        fi.name = name.to_string();
         fi.erasure.index = erasure_index;
         fi.erasure.distribution = distribution.to_vec();
         fi.data = data.map(Bytes::from_static);
@@ -9640,41 +9264,36 @@ mod tests {
     #[test]
     fn collect_inline_data_shards_by_index_uses_distribution_order() {
         let distribution = vec![3, 1, 5, 2, 4, 6];
-        let mut fi = inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"x"));
-        fi.erasure.index = 1;
+        let mut fi = FileInfo::new("object", 4, 2);
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo(4, 2, 3, &distribution, Some(b"c")),
-            inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo(4, 2, 5, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo(4, 2, 2, &distribution, Some(b"b")),
-            inline_data_shard_fileinfo(4, 2, 4, &distribution, Some(b"d")),
-            inline_data_shard_fileinfo(4, 2, 6, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo("block-3", 4, 2, 3, &distribution, Some(b"c")),
+            inline_data_shard_fileinfo("block-1", 4, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo("parity-5", 4, 2, 5, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo("block-2", 4, 2, 2, &distribution, Some(b"b")),
+            inline_data_shard_fileinfo("block-4", 4, 2, 4, &distribution, Some(b"d")),
+            inline_data_shard_fileinfo("parity-6", 4, 2, 6, &distribution, Some(b"q")),
         ];
 
         let data_files =
             collect_inline_data_shard_fileinfos_by_index(&files, &fi, 4, |_| true).expect("all data shards should be collected");
 
         assert_eq!(
-            data_files
-                .iter()
-                .map(|file| file.data.as_deref().expect("fixture carries inline bytes"))
-                .collect::<Vec<_>>(),
-            [b"a".as_slice(), b"b".as_slice(), b"c".as_slice(), b"d".as_slice()]
+            data_files.iter().map(|file| file.name.as_str()).collect::<Vec<_>>(),
+            ["block-1", "block-2", "block-3", "block-4"]
         );
     }
 
     #[test]
     fn collect_inline_data_shards_by_index_rejects_missing_data_shard() {
         let distribution = vec![1, 2, 3, 4];
-        let mut fi = inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"x"));
-        fi.erasure.index = 1;
+        let mut fi = FileInfo::new("object", 2, 2);
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo(2, 2, 2, &distribution, None),
-            inline_data_shard_fileinfo(2, 2, 3, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo(2, 2, 4, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo("block-1", 2, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo("block-2", 2, 2, 2, &distribution, None),
+            inline_data_shard_fileinfo("parity-3", 2, 2, 3, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo("parity-4", 2, 2, 4, &distribution, Some(b"q")),
         ];
 
         assert!(collect_inline_data_shard_fileinfos_by_index(&files, &fi, 2, |_| true).is_none());
@@ -9807,8 +9426,10 @@ mod tests {
 
         let payload = vec![b'i'; 192 * 1024];
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let fi = files[0].clone();
-        let disk_files = disk_ordered_fileinfos(&files);
+        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+        fi.size = payload.len() as i64;
+        fi.data = files[0].data.clone();
+        fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
 
         let disks = vec![Some(disk); erasure.total_shard_count()];
         let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
@@ -9817,7 +9438,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &disk_files,
+            &files,
             &disks,
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9848,8 +9469,10 @@ mod tests {
         let payload = vec![b'v'; 64 * 1024];
         let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let fi = files[0].clone();
-        let disk_files = disk_ordered_fileinfos(&files);
+        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+        fi.size = payload_size;
+        fi.data = files[0].data.clone();
+        fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
 
         let mut object_info = ObjectInfo {
             size: payload_size,
@@ -9880,7 +9503,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &disk_files,
+            &files,
             &vec![Some(disk); erasure.total_shard_count()],
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,

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

@@ -2296,157 +2296,140 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
-        let commit_set = self.clone();
-        let commit_bucket = bucket.to_owned();
-        let commit_object = object.to_owned();
-        let commit_upload_id = upload_id.to_owned();
-        let commit_upload_id_path = upload_id_path.clone();
-        let commit_version_suspended = opts.version_suspended;
-        let commit_is_versioned = opts.versioned || opts.version_suspended;
-        let commit_capacity_scope_token = opts.capacity_scope_token;
-        let commit_object_lock_guard = object_lock_guard.take();
-        let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some();
-        let commit = async move {
-            let _object_lock_guard = commit_object_lock_guard;
+        let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
+
+        // Crash-consistency injection: hard power loss after the upload is fully
+        // staged and locked but before the authoritative rename_data commit. No
+        // disk has moved the staged data, so a crash here must leave any prior
+        // committed version byte-for-byte intact (rustfs/backlog#864) and the
+        // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
+        if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, object) {
+            return Err(StorageError::Unexpected);
+        }
+
+        // The trailing `_` drops the rename_data old-size backfill
+        // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
+        // `get_object_info` lookup, so the backfill has no consumer here yet.
+        let (online_disks, convergence, op_old_dir, cleanup_disks, _) = Self::rename_data(
+            &shuffle_disks,
+            RUSTFS_META_MULTIPART_BUCKET,
+            &upload_id_path,
+            &parts_metadatas,
+            bucket,
+            object,
+            write_quorum,
+        )
+        .await?;
+
+        // Detach admission before any post-commit await: client cancellation
+        // must not couple durable convergence repair to cleanup work.
+        if convergence.needs_heal() {
+            let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                bucket.to_string(),
+                Some(object.to_string()),
+                false,
+                Some(HealChannelPriority::Normal),
+                Some(self.pool_index),
+                Some(self.set_index),
+            );
+            request.object_version_id = fi
+                .version_id
+                .or_else(|| opts.version_suspended.then(Uuid::nil))
+                .map(|version_id| version_id.to_string());
+            tokio::spawn(async move {
+                let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+            });
+        }
+
+        // Crash-consistency injection: hard power loss after the authoritative
+        // rename_data commit succeeded but before the stale part.N.meta cleanup.
+        // The new version is durably committed and visible, so a crash here must
+        // leave the object readable as the new version; the un-reclaimed staging
+        // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
+        // Compiles to a no-op outside `#[cfg(test)]`.
+        if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object) {
+            return Err(StorageError::Unexpected);
+        }
+
+        // backlog#946: reclaim the stale per-part metadata (and any superfluous
+        // part.N data files no longer in the completed set) only *after* the
+        // authoritative rename_data commit above has succeeded. If rename_data
+        // fails write quorum and returns via `?`, the upload directory must keep
+        // its part.N.meta so a retried CompleteMultipartUpload can still read the
+        // parts; deleting them before the commit would strand the upload
+        // permanently. This mirrors the "clean up only after commit" pattern
+        // already used for the old data-dir GC and the upload-dir delete_all below.
+        self.cleanup_multipart_path(&parts).await;
+
+        if let Some(old_dir) = op_old_dir {
+            let committed_dir = fi.data_dir.unwrap_or_default().to_string();
+            // backlog#898: best-effort reclaim of the dereferenced old data dir.
+            // Returns a receipt (never `Err`); a failed GC must not turn an
+            // already-committed multipart completion into a 503.
+            let cleanup = self
+                .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
+                .await;
+            self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
+                .await;
+        }
+
+        if let Some(stage_start) = complete_tail_stage_start {
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "multipart_complete_tail",
+                stage_start.elapsed().as_secs_f64() * 1000.0,
+            );
+        }
+
+        #[cfg(test)]
+        pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
+
+        let cleanup_store = self.clone();
+        let cleanup_upload_id_path = upload_id_path.clone();
+        let cleanup_bucket = bucket.to_owned();
+        let cleanup_object = object.to_owned();
+        let cleanup_upload_id = upload_id.to_owned();
+        let cleanup_handle = tokio::spawn(async move {
             let _upload_guard = upload_guard;
-            let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
-
-            // Crash-consistency injection: hard power loss after the upload is fully
-            // staged and locked but before the authoritative rename_data commit. No
-            // disk has moved the staged data, so a crash here must leave any prior
-            // committed version byte-for-byte intact (rustfs/backlog#864) and the
-            // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
-            if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, &commit_object) {
-                return Err(StorageError::Unexpected);
-            }
-
-            // The trailing `_` drops the rename_data old-size backfill
-            // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
-            // `get_object_info` lookup, so the backfill has no consumer here yet.
-            let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
-                &shuffle_disks,
-                RUSTFS_META_MULTIPART_BUCKET,
-                &commit_upload_id_path,
-                &parts_metadatas,
-                &commit_bucket,
-                &commit_object,
-                write_quorum,
-            )
-            .await?;
-
-            // Detach admission before any post-commit await: client cancellation
-            // must not couple durable convergence repair to cleanup work.
-            if convergence.needs_heal() {
-                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                    commit_bucket.clone(),
-                    Some(commit_object.clone()),
-                    false,
-                    Some(HealChannelPriority::Normal),
-                    Some(commit_set.pool_index),
-                    Some(commit_set.set_index),
-                );
-                request.object_version_id = fi
-                    .version_id
-                    .or_else(|| commit_version_suspended.then(Uuid::nil))
-                    .map(|version_id| version_id.to_string());
-                tokio::spawn(async move {
-                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-                });
-            }
-
-            // Crash-consistency injection: hard power loss after the authoritative
-            // rename_data commit succeeded but before the stale part.N.meta cleanup.
-            // The new version is durably committed and visible, so a crash here must
-            // leave the object readable as the new version; the un-reclaimed staging
-            // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
-            // Compiles to a no-op outside `#[cfg(test)]`.
-            if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) {
-                return Err(StorageError::Unexpected);
-            }
-
-            // backlog#946: reclaim the stale per-part metadata (and any superfluous
-            // part.N data files no longer in the completed set) only *after* the
-            // authoritative rename_data commit above has succeeded. If rename_data
-            // fails write quorum and returns via `?`, the upload directory must keep
-            // its part.N.meta so a retried CompleteMultipartUpload can still read the
-            // parts; deleting them before the commit would strand the upload
-            // permanently. This mirrors the "clean up only after commit" pattern
-            // already used for the old data-dir GC and the upload-dir delete_all below.
-            commit_set.cleanup_multipart_path(&parts).await;
-
-            if let Some(old_dir) = op_old_dir {
-                let committed_dir = fi.data_dir.unwrap_or_default().to_string();
-                // backlog#898: best-effort reclaim of the dereferenced old data dir.
-                // Returns a receipt (never `Err`); a failed GC must not turn an
-                // already-committed multipart completion into a 503.
-                let cleanup = commit_set
-                    .commit_rename_data_dir(
-                        &cleanup_disks,
-                        &commit_bucket,
-                        &commit_object,
-                        &old_dir.to_string(),
-                        &committed_dir,
-                        write_quorum,
-                    )
-                    .await;
-                commit_set
-                    .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
-                    .await;
-            }
-
-            if let Some(stage_start) = complete_tail_stage_start {
-                rustfs_io_metrics::record_put_object_stage_duration(
-                    "multipart_complete_tail",
-                    stage_start.elapsed().as_secs_f64() * 1000.0,
-                );
-            }
-
-            #[cfg(test)]
-            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
-
-            if let Err(err) = commit_set
-                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
+            if let Err(err) = cleanup_store
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
                 .await
             {
                 warn!(
-                    bucket = %commit_bucket,
-                    object = %commit_object,
-                    upload_id = %commit_upload_id,
+                    bucket = %cleanup_bucket,
+                    object = %cleanup_object,
+                    upload_id = %cleanup_upload_id,
                     error = ?err,
                     "completed multipart upload staging cleanup did not reach write quorum"
                 );
             }
-
-            for (i, op_disk) in online_disks.iter().enumerate() {
-                if let Some(disk) = op_disk
-                    && disk.is_online().await
-                {
-                    fi = parts_metadatas[i].clone();
-                    break;
-                }
-            }
-
-            commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
-
-            fi.is_latest = true;
-
-            commit_set
-                .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
-                .await;
-
-            drop(_object_lock_guard); // drop object lock guard to release the lock
-            drop(_upload_guard);
-
-            Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
-        };
-
-        if detach_commit_owner {
-            tokio::spawn(commit)
-                .await
-                .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
-        } else {
-            commit.await
+        });
+        if let Err(err) = cleanup_handle.await {
+            warn!(
+                bucket = %bucket,
+                object = %object,
+                upload_id = %upload_id,
+                error = ?err,
+                "completed multipart upload staging cleanup task failed"
+            );
         }
+        drop(object_lock_guard); // drop object lock guard to release the lock
+
+        for (i, op_disk) in online_disks.iter().enumerate() {
+            if let Some(disk) = op_disk
+                && disk.is_online().await
+            {
+                fi = parts_metadatas[i].clone();
+                break;
+            }
+        }
+
+        self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
+
+        fi.is_latest = true;
+
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+
+        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
 }
 
@@ -4900,74 +4883,6 @@ mod tests {
         .await;
     }
 
-    #[tokio::test(flavor = "multi_thread")]
-    #[serial]
-    async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {
-        temp_env::async_with_vars(
-            [
-                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
-                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
-            ],
-            async {
-                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
-                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
-                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
-                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
-                let bucket = "multipart-cancelled-tail-lock-bucket";
-                let object = "object";
-                make_bucket_on_all(&disk_stores, bucket).await;
-                let (upload_id, parts) =
-                    stage_upload_with_create_opts(&set_disks, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
-                let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
-                signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
-                signaling.clear_observed();
-                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
-                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
-
-                let complete_store = set_disks.clone();
-                let complete_upload_id = upload_id.clone();
-                let complete = tokio::spawn(async move {
-                    complete_store
-                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
-                        .await
-                });
-                barrier.wait_until_paused().await;
-
-                let abort_store = set_disks.clone();
-                let abort_upload_id = upload_id.clone();
-                let abort = tokio::spawn(async move {
-                    abort_store
-                        .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
-                        .await
-                });
-                signaling.wait_for_attempts(2).await;
-                tokio::task::yield_now().await;
-                assert!(!abort.is_finished(), "abort must wait while completion tail owns the upload lock");
-
-                complete.abort();
-                assert!(
-                    complete
-                        .await
-                        .expect_err("the completion request should be cancellable while the tail is paused")
-                        .is_cancelled()
-                );
-                tokio::task::yield_now().await;
-                assert!(!abort.is_finished(), "cancelling the completion waiter must not release the upload lock");
-
-                barrier.release();
-                let abort_err = abort
-                    .await
-                    .expect("abort task should not panic")
-                    .expect_err("the committed upload should no longer exist when abort acquires the lock");
-                assert!(
-                    matches!(abort_err, StorageError::InvalidUploadID(..)),
-                    "abort should return InvalidUploadID after the detached completion tail, got {abort_err:?}"
-                );
-            },
-        )
-        .await;
-    }
-
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_validates_parts_after_an_inflight_upload_part_commit() {

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

@@ -188,74 +188,6 @@ async fn get_object_reader_with_context(
     GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
 }
 
-fn data_read_metadata_early_stop_request_shape_allowed(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions) -> bool {
-    range.is_none()
-        && opts.part_number.is_none()
-        && opts.version_id.is_none()
-        && !opts.incl_free_versions
-        && !opts.skip_free_version
-        && !opts.raw_data_movement_read
-        && !opts.data_movement
-        && !crate::object_api::restore_request_active(opts)
-}
-
-#[cfg(test)]
-mod data_read_metadata_early_stop_request_shape_tests {
-    use super::*;
-
-    #[test]
-    fn data_read_metadata_early_stop_only_allows_whole_latest_plain_get_shape() {
-        assert!(data_read_metadata_early_stop_request_shape_allowed(&None, &ObjectOptions::default()));
-
-        let range = Some(HTTPRangeSpec {
-            is_suffix_length: false,
-            start: 0,
-            end: 0,
-        });
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&range, &ObjectOptions::default()));
-
-        let part_opts = ObjectOptions {
-            part_number: Some(1),
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &part_opts));
-
-        let version_opts = ObjectOptions {
-            version_id: Some(Uuid::new_v4().to_string()),
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &version_opts));
-
-        let incl_free_opts = ObjectOptions {
-            incl_free_versions: true,
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &incl_free_opts));
-
-        let skip_free_opts = ObjectOptions {
-            skip_free_version: true,
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &skip_free_opts));
-
-        let data_movement_opts = ObjectOptions {
-            data_movement: true,
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &data_movement_opts));
-
-        let raw_data_movement_opts = ObjectOptions {
-            raw_data_movement_read: true,
-            ..Default::default()
-        };
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &raw_data_movement_opts));
-
-        let mut restore_opts = ObjectOptions::default();
-        restore_opts.transition.restore_request.days = Some(1);
-        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &restore_opts));
-    }
-}
-
 /// Length of the full plaintext body when — and only when — this read's output
 /// is exactly the object's complete plaintext, so the app-layer body cache may
 /// serve it in place of the erasure read.
@@ -499,16 +431,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
             (prepared.snapshot, prepared.object_info)
         } else {
-            match self
-                .get_object_fileinfo(
-                    bucket,
-                    object,
-                    opts,
-                    true,
-                    data_read_metadata_early_stop_request_shape_allowed(&range, opts),
-                )
-                .await
-            {
+            match self.get_object_fileinfo(bucket, object, opts, true, true).await {
                 Ok(snapshot) => (snapshot, None),
                 Err(err) => {
                     rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64());
@@ -1183,13 +1106,11 @@ impl SetDisks {
 
         let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
 
-        let mut tmp_cleanup_owned = false;
         let result: Result<(ObjectInfo, Option<OldCurrentSize>)> = async {
             let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
 
             let put_object_size = known_put_object_storage_size(data.size());
-            let is_inline_buffer =
-                storage_class_config.should_inline(erasure.shard_file_size(put_object_size), erasure.data_shards, opts.versioned);
+            let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned);
 
             let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
             let shard_file_size = erasure.shard_file_size(put_object_size);
@@ -1676,236 +1597,169 @@ impl SetDisks {
                 });
             }
 
-            let commit_set = self.clone();
-            let commit_bucket = bucket.to_owned();
-            let commit_object = object.to_owned();
-            let commit_tmp_dir = tmp_dir.clone();
-            let commit_object_lock_guard = object_lock_guard.take();
-            let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take();
-            let detach_commit_owner = commit_object_lock_guard.is_some() || commit_bucket_lifecycle_guard.is_some();
-            let commit_write_path_label = write_path.metric_label();
-            let commit_is_versioned = opts.versioned || opts.version_suspended;
-            let commit_capacity_scope_token = opts.capacity_scope_token;
-            let commit_replication_state = replication_state_to_filemeta(&opts.put_replication_state());
-            tmp_cleanup_owned = true;
+            let rename_stage_start = Instant::now();
+            let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
+                &shuffle_disks,
+                RUSTFS_META_TMP_BUCKET,
+                tmp_dir.as_str(),
+                &parts_metadatas,
+                bucket,
+                object,
+                write_quorum,
+            )
+            .await?;
+            // Do this before any post-commit await so request cancellation cannot
+            // bypass best-effort admission. A process crash before admission
+            // remains subject to the existing scanner reconciliation path.
+            if convergence.needs_heal() {
+                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                    bucket.to_string(),
+                    Some(object.to_string()),
+                    false,
+                    Some(HealChannelPriority::Normal),
+                    Some(self.pool_index),
+                    Some(self.set_index),
+                );
+                request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
+                tokio::spawn(async move {
+                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                });
+            }
 
-            let commit = async move {
-                let _object_lock_guard = commit_object_lock_guard;
-                let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
-                let rename_stage_start = Instant::now();
-                let rename_result = SetDisks::rename_data(
-                    &shuffle_disks,
-                    RUSTFS_META_TMP_BUCKET,
-                    commit_tmp_dir.as_str(),
-                    &parts_metadatas,
-                    &commit_bucket,
-                    &commit_object,
+            let rename_stage_elapsed = rename_stage_start.elapsed();
+            let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
+
+            self.invalidate_get_object_metadata_cache(bucket, object).await;
+
+            // `rename_data` has completed the authoritative quorum commit. The
+            // exact old-data-dir reclamation below is best-effort space cleanup;
+            // it must not serialize the next operation on this object.
+            drop(object_lock_guard);
+
+            rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
+            if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                warn!(
+                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    stage = "rename_data",
+                    bucket = %bucket,
+                    object = %object,
+                    tmp_dir = %tmp_dir,
+                    duration_ms = { rename_stage_ms },
                     write_quorum,
-                )
-                .await;
-                let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = match rename_result {
-                    Ok(commit) => commit,
-                    Err(err) => {
-                        if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
-                            warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
-                        } else if issue3031_diag_enabled() {
-                            warn!(
-                                target: "rustfs_ecstore::set_disk",
-                                bucket = %commit_bucket,
-                                object = %commit_object,
-                                tmp_dir = %commit_tmp_dir,
-                                "issue3031_put_object_tmp_cleanup_done"
-                            );
-                        }
-                        return Err(err.into());
-                    }
-                };
-                // Do this before any post-commit await so request cancellation cannot
-                // bypass best-effort admission. A process crash before admission
-                // remains subject to the existing scanner reconciliation path.
-                if convergence.needs_heal() {
-                    let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                        commit_bucket.clone(),
-                        Some(commit_object.clone()),
-                        false,
-                        Some(HealChannelPriority::Normal),
-                        Some(commit_set.pool_index),
-                        Some(commit_set.set_index),
-                    );
-                    request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
-                    tokio::spawn(async move {
-                        let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-                    });
-                }
+                    state = "slow",
+                    "SetDisk commit tail stage is slow"
+                );
+            }
 
-                let rename_stage_elapsed = rename_stage_start.elapsed();
-                let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
-
-                commit_set
-                    .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
+            let mut cleanup_stage_ms: Option<u64> = None;
+            if let Some(old_dir) = op_old_dir {
+                let committed_dir = committed_data_dir.unwrap_or_default().to_string();
+                let cleanup_stage_start = Instant::now();
+                // backlog#898: reclaiming the dereferenced old data dir is
+                // best-effort and returns a receipt (never `Err`). A failed GC
+                // here must not negate an already-committed, durable write, so we
+                // deliberately do NOT `?`-propagate it into a 503. On residue the
+                // report path emits the leak metric and enqueues a heal.
+                let cleanup = self
+                    .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
                     .await;
-
-                // `rename_data` has completed the authoritative quorum commit. The
-                // exact old-data-dir reclamation below is best-effort space cleanup;
-                // it must not serialize the next operation on this object.
-                drop(_object_lock_guard);
-                drop(_bucket_lifecycle_guard);
-
-                rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
-                if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                let cleanup_elapsed = cleanup_stage_start.elapsed();
+                let cleanup_ms = cleanup_elapsed.as_millis() as u64;
+                cleanup_stage_ms = Some(cleanup_ms);
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "set_disk_old_data_cleanup",
+                    duration_millis_f64(cleanup_elapsed),
+                );
+                self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
+                    .await;
+                if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
                     warn!(
                         event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
                         component = LOG_COMPONENT_ECSTORE,
                         subsystem = LOG_SUBSYSTEM_SET_DISK,
-                        stage = "rename_data",
-                        bucket = %commit_bucket,
-                        object = %commit_object,
-                        tmp_dir = %commit_tmp_dir,
-                        duration_ms = { rename_stage_ms },
+                        stage = "commit_rename_data_dir",
+                        bucket = %bucket,
+                        object = %object,
+                        tmp_dir = %tmp_dir,
+                        old_dir = %old_dir,
+                        duration_ms = cleanup_ms,
                         write_quorum,
                         state = "slow",
                         "SetDisk commit tail stage is slow"
                     );
                 }
-
-                let mut cleanup_stage_ms: Option<u64> = None;
-                if let Some(old_dir) = op_old_dir {
-                    let committed_dir = committed_data_dir.unwrap_or_default().to_string();
-                    let cleanup_stage_start = Instant::now();
-                    // backlog#898: reclaiming the dereferenced old data dir is
-                    // best-effort and returns a receipt (never `Err`). A failed GC
-                    // here must not negate an already-committed, durable write, so we
-                    // deliberately do NOT `?`-propagate it into a 503. On residue the
-                    // report path emits the leak metric and enqueues a heal.
-                    let cleanup = commit_set
-                        .commit_rename_data_dir(
-                            &cleanup_disks,
-                            &commit_bucket,
-                            &commit_object,
-                            &old_dir.to_string(),
-                            &committed_dir,
-                            write_quorum,
-                        )
-                        .await;
-                    let cleanup_elapsed = cleanup_stage_start.elapsed();
-                    let cleanup_ms = cleanup_elapsed.as_millis() as u64;
-                    cleanup_stage_ms = Some(cleanup_ms);
-                    rustfs_io_metrics::record_put_object_stage_duration(
-                        "set_disk_old_data_cleanup",
-                        duration_millis_f64(cleanup_elapsed),
-                    );
-                    commit_set
-                        .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
-                        .await;
-                    if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
-                        warn!(
-                            event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_SET_DISK,
-                            stage = "commit_rename_data_dir",
-                            bucket = %commit_bucket,
-                            object = %commit_object,
-                            tmp_dir = %commit_tmp_dir,
-                            old_dir = %old_dir,
-                            duration_ms = cleanup_ms,
-                            write_quorum,
-                            state = "slow",
-                            "SetDisk commit tail stage is slow"
-                        );
-                    }
-                }
-
-                let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
-                let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
-
-                if is_compressed {
-                    record_compression_total_memory(actual_size as u64, w_size as u64).await;
-                }
-                commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
-
-                fi.replication_state_internal = Some(commit_replication_state);
-
-                fi.is_latest = true;
-
-                if issue3031_diag_enabled() {
-                    let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
-                    warn!(
-                        target: "rustfs_ecstore::set_disk",
-                        bucket = %commit_bucket,
-                        object = %commit_object,
-                        tmp_dir = %commit_tmp_dir,
-                        data_dir = ?fi.data_dir,
-                        write_quorum,
-                        online_success_count,
-                        op_old_dir = ?op_old_dir,
-                        "issue3031_put_object_commit_succeeded"
-                    );
-                }
-
-                let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
-                if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
-                    warn!(
-                        event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_SET_DISK,
-                        stage = "put_object_commit_tail",
-                        bucket = %commit_bucket,
-                        object = %commit_object,
-                        tmp_dir = %commit_tmp_dir,
-                        duration_ms = total_commit_tail_ms as u64,
-                        write_quorum,
-                        state = "slow",
-                        "SetDisk commit tail is slow"
-                    );
-                }
-
-                if issue3031_diag_enabled() {
-                    warn!(
-                        event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_SET_DISK,
-                        bucket = %commit_bucket,
-                        object = %commit_object,
-                        write_quorum,
-                        write_path = commit_write_path_label,
-                        writer_setup_ms,
-                        encode_ms,
-                        rename_ms = rename_stage_ms,
-                        cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
-                        cleanup_present = cleanup_stage_ms.is_some(),
-                        commit_tail_ms = total_commit_tail_ms as u64,
-                        result = "success",
-                        "SetDisk put_object stage summary"
-                    );
-                }
-
-                let cleanup_set = commit_set.clone();
-                let cleanup_tmp_dir = commit_tmp_dir.clone();
-                tokio::spawn(async move {
-                    if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
-                        warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
-                    } else if issue3031_diag_enabled() {
-                        warn!(
-                            target: "rustfs_ecstore::set_disk",
-                            tmp_dir = %cleanup_tmp_dir,
-                            "issue3031_put_object_tmp_cleanup_done"
-                        );
-                    }
-                });
-
-                Ok((
-                    ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned),
-                    old_current_size,
-                ))
-            };
-
-            if detach_commit_owner {
-                tokio::spawn(commit)
-                    .await
-                    .map_err(|err| Error::other(format!("put_object commit task failed: {err}")))?
-            } else {
-                commit.await
             }
+
+            let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
+            let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
+
+            if is_compressed {
+                record_compression_total_memory(actual_size as u64, w_size as u64).await;
+            }
+            self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
+
+            fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state()));
+
+            fi.is_latest = true;
+
+            if issue3031_diag_enabled() {
+                let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
+                warn!(
+                    target: "rustfs_ecstore::set_disk",
+                    bucket = %bucket,
+                    object = %object,
+                    tmp_dir = %tmp_dir,
+                    data_dir = ?fi.data_dir,
+                    write_quorum,
+                    online_success_count,
+                    op_old_dir = ?op_old_dir,
+                    "issue3031_put_object_commit_succeeded"
+                );
+            }
+
+            let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
+            if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                warn!(
+                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    stage = "put_object_commit_tail",
+                    bucket = %bucket,
+                    object = %object,
+                    tmp_dir = %tmp_dir,
+                    duration_ms = total_commit_tail_ms as u64,
+                    write_quorum,
+                    state = "slow",
+                    "SetDisk commit tail is slow"
+                );
+            }
+
+            if issue3031_diag_enabled() {
+                warn!(
+                    event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket = %bucket,
+                    object = %object,
+                    write_quorum,
+                    write_path = write_path.metric_label(),
+                    writer_setup_ms,
+                    encode_ms,
+                    rename_ms = rename_stage_ms,
+                    cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
+                    cleanup_present = cleanup_stage_ms.is_some(),
+                    commit_tail_ms = total_commit_tail_ms as u64,
+                    result = "success",
+                    "SetDisk put_object stage summary"
+                );
+            }
+
+            Ok((
+                ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended),
+                old_current_size,
+            ))
         }
         .await;
 
@@ -1941,8 +1795,7 @@ impl SetDisks {
             );
         }
 
-        if tmp_cleanup_owned && result.is_ok() {
-        } else if result.is_ok() {
+        if result.is_ok() {
             // Success path: `rename_data` has already moved the data dir out of
             // the tmp workspace and removed the (empty) tmp dir where it could,
             // so this delete_all is a speculative safety net that normally hits
@@ -5922,10 +5775,8 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
 mod inline_put_commit_path_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
     use super::*;
-    use crate::config::storageclass::lookup_config_for_pools_without_env;
     use crate::disk::{DiskAPI as _, ReadOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
-    use rustfs_config::server_config::KVS;
     use tokio::io::AsyncReadExt;
 
     async fn make_bucket(disks: &[DiskStore], bucket: &str) {
@@ -5989,91 +5840,6 @@ mod inline_put_commit_path_tests {
         assert_eq!(restored, payload);
     }
 
-    #[tokio::test]
-    async fn ec_8_4_default_budget_keeps_large_inline_candidate_out_of_xl_meta() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await;
-        set_disks.set_test_storage_class_config(
-            lookup_config_for_pools_without_env(&KVS::new(), &[12]).expect("EC8+4 storage class should resolve"),
-        );
-        let bucket = "ec-8-4-inline-budget";
-        let object = "object.bin";
-        let payload = vec![0x5c; 300 * 1024];
-        make_bucket(&disk_stores, bucket).await;
-
-        let mut reader = PutObjReader::from_vec(payload.clone());
-        set_disks
-            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("EC8+4 PUT should commit through the non-inline path");
-
-        for (disk_index, disk) in disk_stores.iter().enumerate() {
-            let file_info = disk
-                .read_version("", bucket, object, "", &ReadOptions::default())
-                .await
-                .unwrap_or_else(|err| panic!("disk {disk_index} should persist EC8+4 metadata: {err}"));
-            assert_eq!(file_info.erasure.data_blocks, 8);
-            assert_eq!(file_info.erasure.parity_blocks, 4);
-            assert!(!file_info.inline_data(), "disk {disk_index} must keep the shard outside xl.meta");
-        }
-
-        let mut object_reader = set_disks
-            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
-            .await
-            .expect("non-inline EC8+4 object should remain readable");
-        let mut restored = Vec::new();
-        object_reader
-            .stream
-            .read_to_end(&mut restored)
-            .await
-            .expect("non-inline EC8+4 object should stream");
-        assert_eq!(restored, payload);
-    }
-
-    #[tokio::test]
-    async fn ec_8_4_versioned_budget_reaches_put_placement_decision() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await;
-        set_disks.set_test_storage_class_config(
-            lookup_config_for_pools_without_env(&KVS::new(), &[12]).expect("EC8+4 storage class should resolve"),
-        );
-        let bucket = "ec-8-4-versioned-inline-budget";
-        let object = "object.bin";
-        let payload = vec![0x73; 64 * 1024];
-        make_bucket(&disk_stores, bucket).await;
-
-        let options = ObjectOptions {
-            versioned: true,
-            ..Default::default()
-        };
-        let mut reader = PutObjReader::from_vec(payload.clone());
-        set_disks
-            .put_object(bucket, object, &mut reader, &options)
-            .await
-            .expect("versioned EC8+4 PUT should use the reduced inline budget");
-
-        for (disk_index, disk) in disk_stores.iter().enumerate() {
-            let file_info = disk
-                .read_version("", bucket, object, "", &ReadOptions::default())
-                .await
-                .unwrap_or_else(|err| panic!("disk {disk_index} should persist versioned EC8+4 metadata: {err}"));
-            assert!(
-                !file_info.inline_data(),
-                "disk {disk_index} must keep the versioned shard outside xl.meta"
-            );
-        }
-
-        let mut object_reader = set_disks
-            .get_object_reader(bucket, object, None, HeaderMap::new(), &options)
-            .await
-            .expect("versioned non-inline EC8+4 object should remain readable");
-        let mut restored = Vec::new();
-        object_reader
-            .stream
-            .read_to_end(&mut restored)
-            .await
-            .expect("versioned non-inline EC8+4 object should stream");
-        assert_eq!(restored, payload);
-    }
-
     #[tokio::test]
     async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -6267,10 +6033,7 @@ mod inline_put_commit_path_tests {
 mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;
     use super::*;
-    use crate::diagnostics::get::{
-        GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
-        GetObjectFailureReason,
-    };
+    use crate::diagnostics::get::{GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason};
     use crate::disk::RUSTFS_META_BUCKET;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -6389,22 +6152,7 @@ mod get_object_downstream_close_accounting_tests {
         let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
         rustfs_io_metrics::set_get_stage_metrics_enabled(true);
 
-        let (
-            internal_missing,
-            legacy_unknown,
-            internal_fanout,
-            legacy_fanout,
-            internal_scheduled,
-            legacy_scheduled,
-            internal_completed,
-            legacy_completed,
-            internal_cancelled,
-            legacy_cancelled,
-            internal_unsafe_miss,
-            legacy_unsafe_miss,
-            internal_saved,
-            legacy_saved,
-        ) = metrics::with_local_recorder(&recorder, || {
+        let (internal_missing, legacy_unknown, internal_fanout, legacy_fanout) = metrics::with_local_recorder(&recorder, || {
             runtime.block_on(async {
                 let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
                 let options = ObjectOptions {
@@ -6448,54 +6196,6 @@ mod get_object_downstream_close_accounting_tests {
                         "rustfs_io_get_object_metadata_fanout_error_responses",
                         &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
                     ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_scheduled",
-                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_scheduled",
-                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_completed",
-                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_completed",
-                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_cancelled",
-                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_fanout_cancelled",
-                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
-                    ),
-                    recorder.counter_value(
-                        "rustfs_io_get_object_metadata_early_stop_total",
-                        &[
-                            ("path", GET_OBJECT_PATH_INTERNAL_META),
-                            ("decision", "miss"),
-                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
-                        ],
-                    ),
-                    recorder.counter_value(
-                        "rustfs_io_get_object_metadata_early_stop_total",
-                        &[
-                            ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
-                            ("decision", "miss"),
-                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
-                        ],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
-                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
-                    ),
-                    recorder.histogram_values(
-                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
-                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
-                    ),
                 )
             })
         });
@@ -6508,50 +6208,6 @@ mod get_object_downstream_close_accounting_tests {
         );
         assert_eq!(internal_fanout, vec![4.0], "internal metadata fanout must retain its path label");
         assert!(legacy_fanout.is_empty(), "internal metadata fanout must not leak into legacy_duplex");
-        assert_eq!(
-            internal_scheduled,
-            vec![4.0],
-            "internal metadata lifecycle scheduled count must retain its path label"
-        );
-        assert!(
-            legacy_scheduled.is_empty(),
-            "internal metadata lifecycle scheduled count must not leak into legacy_duplex"
-        );
-        assert_eq!(
-            internal_completed,
-            vec![4.0],
-            "internal metadata lifecycle completed count must retain its path label"
-        );
-        assert!(
-            legacy_completed.is_empty(),
-            "internal metadata lifecycle completed count must not leak into legacy_duplex"
-        );
-        assert_eq!(
-            internal_cancelled,
-            vec![0.0],
-            "internal metadata full-wait lifecycle must record zero cancellations"
-        );
-        assert!(
-            legacy_cancelled.is_empty(),
-            "internal metadata lifecycle cancelled count must not leak into legacy_duplex"
-        );
-        assert_eq!(
-            internal_unsafe_miss, 1,
-            "internal metadata unsafe early-stop miss must retain its path label"
-        );
-        assert_eq!(
-            legacy_unsafe_miss, 0,
-            "internal metadata unsafe early-stop miss must not leak into legacy_duplex"
-        );
-        assert_eq!(
-            internal_saved,
-            vec![0.0],
-            "internal metadata unsafe miss must record zero saved responses on internal_meta"
-        );
-        assert!(
-            legacy_saved.is_empty(),
-            "internal metadata unsafe miss saved responses must not leak into legacy_duplex"
-        );
     }
 }
 
@@ -10106,74 +9762,6 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
     }
 
-    #[tokio::test]
-    async fn cancelled_rename_keeps_namespace_lock_until_publication() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "put-commit-lock-cancelled-rename";
-        let object = "commit-lock-cancelled-rename-object";
-        for disk in &disk_stores {
-            disk.make_volume(bucket).await.expect("bucket volume should be created");
-        }
-
-        let rename_tasks = rename_fanout_barrier::observe_tasks(object);
-        let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
-        let first_store = Arc::clone(&set_disks);
-        let first = tokio::spawn(async move {
-            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
-            first_store
-                .put_object(bucket, object, &mut reader, &ObjectOptions::default())
-                .await
-        });
-        tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
-            .await
-            .expect("first PUT should pause during the authoritative rename");
-
-        let second_namespace_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
-        let second_store = Arc::clone(&set_disks);
-        let second = tokio::spawn(async move {
-            let mut reader = PutObjReader::from_vec(vec![b'2'; TEST_OBJECT_SIZE]);
-            second_store
-                .put_object(bucket, object, &mut reader, &ObjectOptions::default())
-                .await
-        });
-        second_namespace_barrier.release_and_wait_until_namespace_pending().await;
-
-        first.abort();
-        assert!(
-            first
-                .await
-                .expect_err("the first request should be cancelled while rename is parked")
-                .is_cancelled()
-        );
-        tokio::task::yield_now().await;
-        assert!(
-            !second.is_finished(),
-            "the second writer must remain blocked by the cancelled commit owner"
-        );
-
-        rename_barrier.release();
-        drop(rename_barrier);
-        tokio::time::timeout(Duration::from_secs(30), async {
-            while rename_tasks.running() != 0 {
-                tokio::task::yield_now().await;
-            }
-        })
-        .await
-        .expect("the cancelled owner's rename fanout should drain");
-        second
-            .await
-            .expect("second overwrite task should join")
-            .expect("second overwrite should commit after the cancelled owner reaches publication");
-
-        let mut reader = set_disks
-            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
-            .await
-            .expect("the latest overwrite should be readable");
-        let mut body = Vec::new();
-        reader.stream.read_to_end(&mut body).await.expect("latest body should drain");
-        assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
-    }
-
     #[tokio::test]
     async fn put_object_no_lock_aborts_after_outer_namespace_lock_loss() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/heal/tests/endpoint_index_test.rs

@@ -14,8 +14,6 @@
 
 //! test endpoint index settings
 
-#![recursion_limit = "256"]
-
 use std::net::SocketAddr;
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -22,8 +22,6 @@
 //! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
 //! in its own process so the OnceCell never collides.
 
-#![recursion_limit = "256"]
-
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -21,8 +21,6 @@
 //! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks.
 //! Every test is `#[serial]`; under `cargo nextest` each runs in its own process.
 
-#![recursion_limit = "256"]
-
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::storage::{

crates/heal/tests/heal_integration_test.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![recursion_limit = "256"]
-
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/io-metrics/src/lib.rs

@@ -812,17 +812,6 @@ pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize,
         .record(metadata_fanout_count_to_f64(non_valid));
 }
 
-/// Record task lifecycle shape for one GetObject metadata fanout.
-#[inline(always)]
-pub fn record_get_object_metadata_fanout_lifecycle(path: &'static str, scheduled: usize, completed: usize, cancelled: usize) {
-    if !get_stage_metrics_enabled() {
-        return;
-    }
-    histogram!("rustfs_io_get_object_metadata_fanout_scheduled", "path" => path).record(metadata_fanout_count_to_f64(scheduled));
-    histogram!("rustfs_io_get_object_metadata_fanout_completed", "path" => path).record(metadata_fanout_count_to_f64(completed));
-    histogram!("rustfs_io_get_object_metadata_fanout_cancelled", "path" => path).record(metadata_fanout_count_to_f64(cancelled));
-}
-
 /// Record a guarded metadata early-stop hit for GetObject.
 #[inline(always)]
 pub fn record_get_object_metadata_early_stop_hit(path: &'static str, reason: &'static str) {
@@ -2709,7 +2698,6 @@ mod tests {
         record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
         record_get_object_metadata_response("legacy_duplex", "valid");
         record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
-        record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
         record_get_object_metadata_early_stop_hit("legacy_duplex", "valid_quorum");
         record_get_object_metadata_early_stop_miss("legacy_duplex", "insufficient_quorum");
         record_get_object_metadata_early_stop_saved_responses("legacy_duplex", 1);
@@ -2780,38 +2768,6 @@ mod tests {
         assert!(remote_scheduled >= remote_avoid_potential);
     }
 
-    #[test]
-    fn metadata_fanout_lifecycle_records_named_histograms() {
-        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
-        let recorder = DebuggingRecorder::new();
-        let snapshotter = recorder.snapshotter();
-
-        metrics::with_local_recorder(&recorder, || {
-            set_get_stage_metrics_enabled(true);
-            record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
-            set_get_stage_metrics_enabled(false);
-        });
-
-        let metrics = snapshotter.snapshot().into_vec();
-        for (name, expected) in [
-            ("rustfs_io_get_object_metadata_fanout_scheduled", 4.0),
-            ("rustfs_io_get_object_metadata_fanout_completed", 3.0),
-            ("rustfs_io_get_object_metadata_fanout_cancelled", 1.0),
-        ] {
-            let value = metrics.iter().find_map(|(composite, _, _, value)| {
-                let has_path = composite
-                    .key()
-                    .labels()
-                    .any(|label| label.key() == "path" && label.value() == "legacy_duplex");
-                (composite.kind() == MetricKind::Histogram && composite.key().name() == name && has_path).then_some(value)
-            });
-            assert!(
-                matches!(value, Some(DebugValue::Histogram(values)) if values.len() == 1 && values[0].0 == expected),
-                "{name} must record the exact fanout lifecycle sample"
-            );
-        }
-    }
-
     #[test]
     fn test_record_get_object_fill_metrics() {
         record_get_object_fill_queued("codec_streaming", "single_inflight", 1);

crates/protocols/tests/swift_metadata_persistence.rs

@@ -23,7 +23,6 @@
 //! two are tested together because a reload is the only way to tell a real
 //! merge from one that happened to look right in the cache.
 
-#![recursion_limit = "256"]
 #![cfg(feature = "swift")]
 
 use std::collections::HashMap;

crates/scanner/src/lib.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![recursion_limit = "256"]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(
     // missing_docs,

crates/scanner/tests/lifecycle_integration_test.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![recursion_limit = "256"]
-
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -12,6 +12,7 @@ use datafusion::{
 use std::sync::Arc;
 
 use crate::table_catalog::test_support::{
+    TestCatalogObjectBackend as TestTableCatalogObjectBackend, TestCatalogObjectRecord,
     manifest_avro_bytes as test_manifest_avro_bytes,
     manifest_avro_bytes_with_nullable_sequences as test_manifest_avro_bytes_with_nullable_sequences,
     manifest_list_avro_bytes as test_manifest_list_avro_bytes, manifest_list_avro_entries as test_manifest_list_avro_entries,
@@ -2017,7 +2018,7 @@ fn format_upgrade_assigns_v1_snapshot_sequences_and_rejects_v3() {
 #[tokio::test]
 async fn create_table_response_writes_initial_metadata_for_standard_request() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -2094,7 +2095,7 @@ async fn create_table_holds_bucket_fence_from_metadata_write_through_registratio
     let barrier = Arc::new(tokio::sync::Barrier::new(2));
     let metadata_backend = TestTableCatalogObjectBackend {
         put_object_barrier: Some(Arc::clone(&barrier)),
-        ..Default::default()
+        ..TestTableCatalogObjectBackend::content_addressed()
     };
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(store.as_ref(), "warehouse", true)
@@ -2128,7 +2129,7 @@ async fn create_table_holds_bucket_fence_from_metadata_write_through_registratio
         create_table_response(create_store.as_ref(), &create_backend, "warehouse", &create_namespace, request, true).await
     });
     tokio::time::timeout(StdDuration::from_secs(2), async {
-        while metadata_backend.objects.lock().await.is_empty() {
+        while metadata_backend.state.lock().await.objects.is_empty() {
             tokio::task::yield_now().await;
         }
     })
@@ -2172,7 +2173,7 @@ async fn create_table_holds_bucket_fence_from_metadata_write_through_registratio
 
 #[tokio::test]
 async fn create_table_response_recreates_dropped_identifier_without_overwriting_retained_metadata() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
@@ -2308,9 +2309,9 @@ async fn create_table_response_recreates_dropped_identifier_without_overwriting_
 
 #[tokio::test]
 async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct_metadata() {
-    let catalog_backend = TestTableCatalogObjectBackend::default();
+    let catalog_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(catalog_backend);
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -2404,7 +2405,7 @@ async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct
 #[tokio::test]
 async fn standard_commit_applies_updates_and_writes_next_metadata() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_uuid = created.metadata["table-uuid"]
@@ -2535,7 +2536,7 @@ fn table_metadata_file_name_scoping_is_bounded_and_identity_sensitive() {
 
 #[tokio::test]
 async fn renamed_and_recreated_tables_with_the_same_commit_id_use_disjoint_metadata_files() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::StrongTableCatalogStore::new(metadata_backend.clone());
     let source_namespace = crate::table_catalog::Namespace::parse("analytics").expect("source namespace should parse");
     let destination_namespace = crate::table_catalog::Namespace::parse("curated").expect("destination namespace should parse");
@@ -2952,7 +2953,7 @@ async fn standard_commit_recovers_matching_table_scoped_metadata_orphan() {
 
 #[tokio::test]
 async fn concurrent_identical_commits_reuse_table_scoped_metadata_winner() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::StrongTableCatalogStore::new(metadata_backend.clone());
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
@@ -3192,7 +3193,7 @@ async fn standard_commit_rejects_fallback_readback_mismatch() {
 #[tokio::test]
 async fn standard_commit_uses_client_uuid_commit_id_in_metadata_file_name() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
 
@@ -3241,7 +3242,7 @@ async fn standard_commit_uses_client_uuid_commit_id_in_metadata_file_name() {
 #[tokio::test]
 async fn standard_commit_accepts_non_uuid_client_commit_id_without_using_it_in_metadata_file_name() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
 
@@ -3284,7 +3285,7 @@ async fn standard_commit_accepts_non_uuid_client_commit_id_without_using_it_in_m
 #[tokio::test]
 async fn commit_publication_uses_idempotency_key_as_retry_identity() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let idempotency_key = Uuid::now_v7().to_string();
@@ -3319,7 +3320,7 @@ async fn commit_publication_replays_historical_standard_commit_across_backings()
         crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
         crate::table_catalog::TableCatalogBackingMode::DurableStrong,
     ] {
-        let metadata_backend = TestTableCatalogObjectBackend::default();
+        let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
         let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(metadata_backend.clone(), mode);
         let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
         create_standard_events_table(&store, &metadata_backend, &namespace).await;
@@ -3435,7 +3436,7 @@ async fn commit_publication_replays_historical_standard_commit_across_backings()
 
 #[tokio::test]
 async fn staged_standard_commit_retry_revalidates_referenced_objects() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
@@ -3545,7 +3546,7 @@ async fn staged_standard_commit_retry_revalidates_referenced_objects() {
 #[tokio::test]
 async fn commit_publication_denies_generated_metadata_write_before_pointer_advance() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let before = store
@@ -3605,7 +3606,7 @@ async fn commit_publication_denies_generated_metadata_write_before_pointer_advan
 #[tokio::test]
 async fn commit_publication_authorizes_referenced_objects() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let entry = store
@@ -3672,7 +3673,7 @@ async fn commit_publication_authorizes_referenced_objects() {
 #[tokio::test]
 async fn commit_publication_denies_referenced_data_read_before_pointer_advance() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let before = store
@@ -3738,7 +3739,7 @@ async fn commit_publication_holds_referenced_object_locks_until_pointer_publish(
         commit_table_pause: Some(pause.clone()),
         ..Default::default()
     });
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(store.as_ref(), &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -3846,7 +3847,7 @@ async fn commit_publication_holds_referenced_object_locks_until_pointer_publish(
 
 #[tokio::test]
 async fn rolling_upgrade_commit_retains_legacy_data_file_guard_until_publication_completes() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let data_file = "tables/table-id/data/part-00001.parquet";
     metadata_backend.put_bytes("warehouse", data_file, b"data".to_vec()).await;
     let commit_backend = TableCommitObjectBackend::rolling_upgrade(metadata_backend.clone());
@@ -3908,7 +3909,7 @@ async fn rolling_upgrade_initial_publication_fences_old_and_new_data_plane_write
     .expect("namespace should seed");
 
     let data_file = "tables/table-id/data/part-00001.parquet";
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     metadata_backend.put_bytes("warehouse", data_file, b"data".to_vec()).await;
     let publication_backend = TableCommitObjectBackend::rolling_upgrade(metadata_backend.clone());
     assert!(
@@ -4004,7 +4005,7 @@ async fn warehouse_relocation_holds_bucket_fence_before_catalog_publication() {
         commit_table_pause: Some(pause.clone()),
         ..Default::default()
     });
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(store.as_ref(), &metadata_backend, &namespace).await;
     let current = store
@@ -4079,7 +4080,7 @@ async fn warehouse_relocation_holds_bucket_fence_before_catalog_publication() {
 
 #[tokio::test]
 async fn commit_publication_lock_order_remains_compatible_with_old_maintenance_nodes() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
     let current_metadata = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
@@ -4163,7 +4164,7 @@ async fn commit_publication_lock_order_remains_compatible_with_old_maintenance_n
 
 #[tokio::test]
 async fn commit_publication_acquires_discovered_object_locks_in_key_order() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let first = "metadata/a.json";
     let last = "metadata/z.json";
     metadata_backend.put_bytes("warehouse", first, b"a".to_vec()).await;
@@ -4238,7 +4239,7 @@ async fn commit_publication_acquires_discovered_object_locks_in_key_order() {
 
 #[tokio::test]
 async fn commit_publication_revalidates_objects_after_ordered_lock_acquisition() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let object = "metadata/current.json";
     metadata_backend.put_bytes("warehouse", object, b"before".to_vec()).await;
     let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
@@ -4260,16 +4261,16 @@ async fn commit_publication_revalidates_objects_after_ordered_lock_acquisition()
 
 #[tokio::test]
 async fn commit_publication_binds_fingerprint_to_returned_bytes() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let object = "metadata/current.json";
     let original = b"original".to_vec();
     let replacement = b"replacement".to_vec();
     let original_etag = hex_sha256(&original, str::to_string);
-    metadata_backend.objects.lock().await.insert(
+    metadata_backend.state.lock().await.objects.insert(
         ("warehouse".to_string(), object.to_string()),
-        crate::table_catalog::TableCatalogObject {
+        TestCatalogObjectRecord {
             data: replacement,
-            etag: Some(original_etag.clone()),
+            etag: original_etag.clone(),
             mod_time: None,
         },
     );
@@ -4278,11 +4279,11 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
         .await
         .expect("replacement bytes should be discovered")
         .expect("replacement object should exist");
-    metadata_backend.objects.lock().await.insert(
+    metadata_backend.state.lock().await.objects.insert(
         ("warehouse".to_string(), object.to_string()),
-        crate::table_catalog::TableCatalogObject {
+        TestCatalogObjectRecord {
             data: original,
-            etag: Some(original_etag),
+            etag: original_etag,
             mod_time: None,
         },
     );
@@ -4302,7 +4303,7 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
 #[tokio::test]
 async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -4330,15 +4331,15 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
         )
         .await;
     {
-        let mut objects = metadata_backend.objects.lock().await;
+        let mut state = metadata_backend.state.lock().await;
         let data = vec![1];
         let etag = hex_sha256(&data, str::to_string);
         for file in &data_files {
-            objects.insert(
+            state.objects.insert(
                 ("warehouse".to_string(), test_snapshot_object_key("warehouse", file)),
-                crate::table_catalog::TableCatalogObject {
+                TestCatalogObjectRecord {
                     data: data.clone(),
-                    etag: Some(etag.clone()),
+                    etag: etag.clone(),
                     mod_time: None,
                 },
             );
@@ -4390,7 +4391,7 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
 
 #[tokio::test]
 async fn commit_publication_rejects_recreated_object_observed_by_exists() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let object = "data/part-00001.parquet";
     metadata_backend.put_bytes("warehouse", object, b"before".to_vec()).await;
     let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
@@ -4419,7 +4420,7 @@ async fn commit_publication_rejects_recreated_object_observed_by_exists() {
 #[tokio::test]
 async fn standard_commit_ignores_generation_only_orphan_metadata_file() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
     metadata_backend
@@ -4468,15 +4469,15 @@ async fn standard_commit_ignores_generation_only_orphan_metadata_file() {
 #[tokio::test]
 async fn concurrent_standard_commits_write_distinct_metadata_files_before_pointer_conflict() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
 
     let barrier = Arc::new(tokio::sync::Barrier::new(2));
     let metadata_backend = TestTableCatalogObjectBackend {
-        objects: Arc::clone(&metadata_backend.objects),
+        state: Arc::clone(&metadata_backend.state),
         put_object_barrier: Some(barrier),
-        ..Default::default()
+        ..TestTableCatalogObjectBackend::content_addressed()
     };
     let first_commit_id = "33333333-3333-4333-8333-333333333333";
     let second_commit_id = "44444444-4444-4444-8444-444444444444";
@@ -4537,7 +4538,7 @@ async fn concurrent_standard_commits_write_distinct_metadata_files_before_pointe
 #[tokio::test]
 async fn standard_commit_accepts_legacy_catalog_uuid_when_current_metadata_matches() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -4607,7 +4608,7 @@ async fn standard_commit_accepts_legacy_catalog_uuid_when_current_metadata_match
 #[tokio::test]
 async fn metadata_location_api_accepts_legacy_catalog_uuid_when_target_matches_current_metadata() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -4673,7 +4674,7 @@ async fn metadata_location_api_accepts_legacy_catalog_uuid_when_target_matches_c
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_runs_dry_run_and_delete() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -4847,7 +4848,7 @@ async fn table_metadata_maintenance_helper_runs_dry_run_and_delete() {
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_commits_snapshot_expiration() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5003,7 +5004,7 @@ async fn table_metadata_maintenance_helper_commits_snapshot_expiration() {
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_commits_compaction_through_publication_observer() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5095,7 +5096,7 @@ async fn table_metadata_maintenance_helper_commits_compaction_through_publicatio
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_rejects_snapshot_expiration_manual_review_commit() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5173,7 +5174,7 @@ async fn table_metadata_maintenance_helper_rejects_snapshot_expiration_manual_re
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_rejects_stale_snapshot_expiration_plan() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5257,7 +5258,7 @@ async fn table_metadata_maintenance_helper_rejects_stale_snapshot_expiration_pla
 
 #[tokio::test]
 async fn table_metadata_maintenance_helper_rejects_delete_with_snapshot_expiration_commit() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
 
@@ -5286,7 +5287,7 @@ async fn table_metadata_maintenance_helper_rejects_delete_with_snapshot_expirati
 
 #[tokio::test]
 async fn table_refs_response_reports_current_and_user_defined_refs() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5327,7 +5328,7 @@ async fn table_refs_response_reports_current_and_user_defined_refs() {
 
 #[tokio::test]
 async fn external_catalog_bridge_response_lists_supported_operator_bridges() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend);
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5362,7 +5363,7 @@ async fn external_catalog_bridge_response_lists_supported_operator_bridges() {
 
 #[tokio::test]
 async fn external_catalog_bridge_persists_identity_and_boundary() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend);
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5428,7 +5429,7 @@ async fn external_catalog_bridge_persists_identity_and_boundary() {
 
 #[tokio::test]
 async fn external_catalog_bridge_sync_registers_missing_table_from_snapshot() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5506,7 +5507,7 @@ async fn external_catalog_bridge_sync_registers_missing_table_from_snapshot() {
 
 #[tokio::test]
 async fn external_catalog_bridge_sync_commits_existing_table_pointer() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5563,7 +5564,7 @@ async fn external_catalog_bridge_sync_commits_existing_table_pointer() {
 
 #[tokio::test]
 async fn external_catalog_bridge_sync_denies_metadata_reads_before_pointer_publish() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5632,7 +5633,7 @@ async fn external_catalog_bridge_sync_denies_metadata_reads_before_pointer_publi
 
 #[tokio::test]
 async fn external_catalog_bridge_sync_conflicts_leave_pointer_unchanged() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -5823,7 +5824,7 @@ fn snapshot_conflict_rejects_unknown_snapshot_operations() {
 #[tokio::test]
 async fn row_level_conflict_allows_overwrite_when_deleted_file_is_current() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -5959,7 +5960,7 @@ async fn row_level_conflict_allows_overwrite_when_deleted_file_is_current() {
 #[tokio::test]
 async fn row_level_conflict_allows_v1_manifest_snapshot() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6069,7 +6070,7 @@ async fn row_level_conflict_allows_v1_manifest_snapshot() {
 #[tokio::test]
 async fn row_level_conflict_inherits_manifest_list_sequence_numbers() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6122,7 +6123,7 @@ async fn row_level_conflict_inherits_manifest_list_sequence_numbers() {
 #[tokio::test]
 async fn row_level_conflict_allows_inherited_manifests_on_append() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6228,7 +6229,7 @@ async fn row_level_conflict_allows_inherited_manifests_on_append() {
 #[tokio::test]
 async fn row_level_conflict_rejects_changed_inherited_manifest_identity() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6330,7 +6331,7 @@ async fn row_level_conflict_rejects_changed_inherited_manifest_identity() {
 #[tokio::test]
 async fn row_level_conflict_rejects_stale_new_manifest_sequence() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6389,7 +6390,7 @@ async fn row_level_conflict_rejects_stale_new_manifest_sequence() {
 #[tokio::test]
 async fn row_level_conflict_rejects_stale_added_entry_sequence() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6448,7 +6449,7 @@ async fn row_level_conflict_rejects_stale_added_entry_sequence() {
 #[tokio::test]
 async fn row_level_conflict_rejects_historical_change_in_new_manifest() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6507,7 +6508,7 @@ async fn row_level_conflict_rejects_historical_change_in_new_manifest() {
 #[tokio::test]
 async fn row_level_conflict_allows_add_only_overwrite_snapshot() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6618,7 +6619,7 @@ async fn row_level_conflict_allows_add_only_overwrite_snapshot() {
 #[tokio::test]
 async fn row_level_conflict_rejects_delete_of_non_current_file() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6737,7 +6738,7 @@ async fn row_level_conflict_rejects_delete_of_non_current_file() {
 #[tokio::test]
 async fn row_level_conflict_rejects_append_with_delete_files() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6794,7 +6795,7 @@ async fn row_level_conflict_rejects_append_with_delete_files() {
 #[tokio::test]
 async fn row_level_conflict_rejects_missing_manifest_before_pointer_update() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -6900,7 +6901,7 @@ async fn row_level_conflict_rejects_missing_manifest_before_pointer_update() {
 #[tokio::test]
 async fn row_level_conflict_rejects_manifest_outside_table_warehouse() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -7133,7 +7134,7 @@ fn create_view_request_accepts_deep_warehouse_location() {
 #[tokio::test]
 async fn view_catalog_responses_persist_replace_and_drop_view_metadata() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -7297,7 +7298,7 @@ async fn view_catalog_responses_persist_replace_and_drop_view_metadata() {
 #[tokio::test]
 async fn table_ref_write_responses_use_commit_guard_and_protect_deletes() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let table_location = created.metadata["location"]
@@ -7949,90 +7950,6 @@ impl TestCatalogPublishPause {
     }
 }
 
-type TestTableCatalogObjectLocks = Arc<tokio::sync::Mutex<BTreeMap<(String, String), Arc<tokio::sync::Mutex<()>>>>>;
-
-#[derive(Clone, Default)]
-struct TestTableCatalogObjectBackend {
-    objects: Arc<tokio::sync::Mutex<BTreeMap<(String, String), crate::table_catalog::TableCatalogObject>>>,
-    put_object_barrier: Option<Arc<tokio::sync::Barrier>>,
-    fail_put_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
-    corrupt_put_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
-    missing_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
-    fail_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
-    locks: TestTableCatalogObjectLocks,
-    lock_attempts: Arc<tokio::sync::Mutex<Vec<(String, String)>>>,
-}
-
-impl TestTableCatalogObjectBackend {
-    async fn put_bytes(&self, bucket: &str, object: &str, data: Vec<u8>) {
-        let etag = hex_sha256(&data, str::to_string);
-        self.objects.lock().await.insert(
-            (bucket.to_string(), object.to_string()),
-            crate::table_catalog::TableCatalogObject {
-                data,
-                etag: Some(etag),
-                mod_time: None,
-            },
-        );
-    }
-
-    async fn put_json(&self, bucket: &str, object: &str, value: serde_json::Value) {
-        self.put_json_with_mod_time(bucket, object, value, None).await;
-    }
-
-    async fn put_gzip_json(&self, bucket: &str, object: &str, value: serde_json::Value) {
-        use std::io::Write;
-
-        let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
-        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
-        encoder.write_all(&data).expect("metadata JSON should compress");
-        self.put_bytes(bucket, object, encoder.finish().expect("metadata gzip stream should finish"))
-            .await;
-    }
-
-    async fn put_json_with_mod_time(
-        &self,
-        bucket: &str,
-        object: &str,
-        value: serde_json::Value,
-        mod_time: Option<OffsetDateTime>,
-    ) {
-        let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
-        let etag = hex_sha256(&data, str::to_string);
-        self.objects.lock().await.insert(
-            (bucket.to_string(), object.to_string()),
-            crate::table_catalog::TableCatalogObject {
-                data,
-                etag: Some(etag),
-                mod_time,
-            },
-        );
-    }
-
-    async fn write_lock_is_held(&self, bucket: &str, object: &str) -> bool {
-        let lock = self
-            .locks
-            .lock()
-            .await
-            .get(&(bucket.to_string(), object.to_string()))
-            .cloned();
-        lock.is_some_and(|lock| lock.try_lock_owned().is_err())
-    }
-
-    async fn wait_for_lock_attempts(&self, count: usize) {
-        tokio::time::timeout(StdDuration::from_secs(2), async {
-            loop {
-                if self.lock_attempts.lock().await.len() >= count {
-                    return;
-                }
-                tokio::task::yield_now().await;
-            }
-        })
-        .await
-        .expect("lock acquisition attempts should be observable");
-    }
-}
-
 fn trusted_table_commit_backend(
     backend: &TestTableCatalogObjectBackend,
 ) -> TableCommitObjectBackend<TestTableCatalogObjectBackend> {
@@ -8210,7 +8127,7 @@ async fn standard_commit_foreign_primary_fixture() -> (
     String,
 ) {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let current = store
@@ -8252,7 +8169,7 @@ async fn standard_commit_primary_fixture(
     String,
 ) {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let current = store
@@ -8401,130 +8318,6 @@ async fn seed_object_table_for_metadata_maintenance(
         .await;
 }
 
-#[async_trait::async_trait]
-impl crate::table_catalog::TableCatalogObjectBackend for TestTableCatalogObjectBackend {
-    async fn read_object(
-        &self,
-        bucket: &str,
-        object: &str,
-    ) -> crate::table_catalog::TableCatalogStoreResult<Option<crate::table_catalog::TableCatalogObject>> {
-        let mut missing_read_object_path = self.missing_read_object_path.lock().await;
-        if missing_read_object_path.as_deref() == Some(object) {
-            missing_read_object_path.take();
-            return Ok(None);
-        }
-        drop(missing_read_object_path);
-
-        let mut fail_read_object_path = self.fail_read_object_path.lock().await;
-        if fail_read_object_path.as_deref() == Some(object) {
-            fail_read_object_path.take();
-            return Err(crate::table_catalog::TableCatalogStoreError::Internal(
-                "private generated metadata read failure".to_string(),
-            ));
-        }
-        drop(fail_read_object_path);
-
-        Ok(self
-            .objects
-            .lock()
-            .await
-            .get(&(bucket.to_string(), object.to_string()))
-            .cloned())
-    }
-
-    async fn object_exists(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<bool> {
-        Ok(self
-            .objects
-            .lock()
-            .await
-            .contains_key(&(bucket.to_string(), object.to_string())))
-    }
-
-    async fn put_object(
-        &self,
-        bucket: &str,
-        object: &str,
-        data: Vec<u8>,
-        precondition: crate::table_catalog::TableCatalogPutPrecondition,
-    ) -> crate::table_catalog::TableCatalogStoreResult<()> {
-        let mut fail_put_object_path = self.fail_put_object_path.lock().await;
-        if fail_put_object_path.as_deref() == Some(object) {
-            fail_put_object_path.take();
-            return Err(crate::table_catalog::TableCatalogStoreError::Internal(
-                "injected metadata write failure".to_string(),
-            ));
-        }
-        drop(fail_put_object_path);
-
-        let mut corrupt_put_object_path = self.corrupt_put_object_path.lock().await;
-        let data = if corrupt_put_object_path.as_deref() == Some(object) {
-            corrupt_put_object_path.take();
-            b"{}".to_vec()
-        } else {
-            data
-        };
-        drop(corrupt_put_object_path);
-
-        let key = (bucket.to_string(), object.to_string());
-        let mut objects = self.objects.lock().await;
-        let result = if matches!(precondition, crate::table_catalog::TableCatalogPutPrecondition::IfAbsent)
-            && objects.contains_key(&key)
-        {
-            Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
-                "object already exists: {object}"
-            )))
-        } else {
-            let etag = hex_sha256(&data, str::to_string);
-            objects.insert(
-                key,
-                crate::table_catalog::TableCatalogObject {
-                    data,
-                    etag: Some(etag),
-                    mod_time: None,
-                },
-            );
-            Ok(())
-        };
-        drop(objects);
-        if let Some(barrier) = &self.put_object_barrier {
-            barrier.wait().await;
-        }
-        result
-    }
-
-    async fn delete_object(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
-        self.objects.lock().await.remove(&(bucket.to_string(), object.to_string()));
-        Ok(())
-    }
-
-    async fn list_objects(&self, bucket: &str, prefix: &str) -> crate::table_catalog::TableCatalogStoreResult<Vec<String>> {
-        Ok(self
-            .objects
-            .lock()
-            .await
-            .keys()
-            .filter(|(object_bucket, object)| object_bucket == bucket && object.starts_with(prefix))
-            .map(|(_, object)| object.clone())
-            .collect())
-    }
-
-    async fn acquire_write_lock(
-        &self,
-        bucket: &str,
-        object: &str,
-    ) -> crate::table_catalog::TableCatalogStoreResult<Box<dyn Send>> {
-        self.lock_attempts.lock().await.push((bucket.to_string(), object.to_string()));
-        let lock = {
-            let mut locks = self.locks.lock().await;
-            locks
-                .entry((bucket.to_string(), object.to_string()))
-                .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
-                .clone()
-        };
-        Ok(Box::new(lock.lock_owned().await))
-    }
-}
-
 #[async_trait::async_trait]
 impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore {
     async fn get_table_bucket(
@@ -9040,7 +8833,7 @@ async fn namespace_helpers_call_catalog_store() {
 #[tokio::test]
 async fn table_helpers_call_catalog_store() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9172,7 +8965,7 @@ async fn table_helpers_call_catalog_store() {
 #[tokio::test]
 async fn register_table_response_adopts_metadata_table_uuid() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9223,7 +9016,7 @@ async fn register_table_response_adopts_metadata_table_uuid() {
 #[tokio::test]
 async fn register_table_denies_metadata_read_before_catalog_publication() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let (namespace, metadata_location) = seed_events_registration_target(&store, &metadata_backend).await;
     let authorized = Arc::new(tokio::sync::Mutex::new(Vec::new()));
     let guarded_backend =
@@ -9266,7 +9059,7 @@ async fn register_table_denies_metadata_read_before_catalog_publication() {
 #[tokio::test]
 async fn catalog_import_denies_metadata_read_before_catalog_publication() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let (namespace, metadata_location) = seed_events_registration_target(&store, &metadata_backend).await;
     let authorized = Arc::new(tokio::sync::Mutex::new(Vec::new()));
     let guarded_backend =
@@ -9308,7 +9101,7 @@ async fn catalog_import_denies_metadata_read_before_catalog_publication() {
 
 #[tokio::test]
 async fn register_table_rejects_metadata_replaced_before_catalog_publication() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = Arc::new(crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()));
     let (namespace, metadata_location) = seed_events_registration_target(store.as_ref(), &metadata_backend).await;
     let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
@@ -9376,7 +9169,7 @@ async fn register_table_rejects_metadata_replaced_before_catalog_publication() {
 
 #[tokio::test]
 async fn catalog_import_rejects_metadata_replaced_before_catalog_publication() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = Arc::new(crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()));
     let (namespace, metadata_location) = seed_events_registration_target(store.as_ref(), &metadata_backend).await;
     let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
@@ -9445,7 +9238,7 @@ async fn catalog_import_rejects_metadata_replaced_before_catalog_publication() {
 #[tokio::test]
 async fn register_table_response_rejects_metadata_without_format_version() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9501,7 +9294,7 @@ async fn register_table_response_rejects_metadata_without_format_version() {
 #[tokio::test]
 async fn metadata_location_api_loads_and_updates_current_pointer() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9577,7 +9370,7 @@ async fn metadata_location_api_loads_and_updates_current_pointer() {
 #[tokio::test]
 async fn metadata_location_api_accepts_gzip_table_metadata() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let current = store
@@ -9616,7 +9409,7 @@ async fn metadata_location_api_accepts_gzip_table_metadata() {
 #[tokio::test]
 async fn metadata_location_api_validates_snapshot_graph_before_commit() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
     let current = store
@@ -9682,7 +9475,7 @@ async fn metadata_location_api_validates_snapshot_graph_before_commit() {
 
 #[tokio::test]
 async fn metadata_location_api_validates_relocated_snapshot_graph_under_target_warehouse() {
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
@@ -9741,7 +9534,7 @@ async fn metadata_location_api_validates_relocated_snapshot_graph_under_target_w
 #[tokio::test]
 async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9819,7 +9612,7 @@ async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() {
 #[tokio::test]
 async fn metadata_location_api_rejects_mismatched_table_uuid_before_commit() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await
@@ -9898,7 +9691,7 @@ async fn metadata_location_api_rejects_mismatched_table_uuid_before_commit() {
 
 #[tokio::test]
 async fn catalog_import_and_rollback_use_register_and_commit_paths() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -9993,7 +9786,7 @@ async fn catalog_import_and_rollback_use_register_and_commit_paths() {
 
 #[tokio::test]
 async fn rollback_denies_metadata_reads_before_pointer_publish() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -10055,7 +9848,7 @@ async fn rollback_denies_metadata_reads_before_pointer_publish() {
 
 #[tokio::test]
 async fn rollback_rejects_invalid_target_metadata_before_commit() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -10144,7 +9937,7 @@ async fn rollback_rejects_invalid_target_metadata_before_commit() {
 
 #[tokio::test]
 async fn rollback_rejects_mismatched_table_uuid_before_commit() {
-    let backend = TestTableCatalogObjectBackend::default();
+    let backend = TestTableCatalogObjectBackend::content_addressed();
     let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
     let bucket = "warehouse";
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -10229,7 +10022,7 @@ async fn rollback_rejects_mismatched_table_uuid_before_commit() {
 #[tokio::test]
 async fn legacy_commit_rejects_mismatched_table_uuid_before_commit() {
     let store = TestTableCatalogStore::default();
-    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     ensure_table_bucket_entry(&store, "warehouse", true)
         .await

rustfs/src/app/object_usecase.rs

@@ -86,7 +86,7 @@ use super::storage_api::object_usecase::options::{
     namespace_reserved_user_metadata, normalize_content_encoding_for_storage, preserve_unclassified_user_metadata,
     put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
-use super::storage_api::object_usecase::request_context::{self, spawn_traced, spawn_traced_join};
+use super::storage_api::object_usecase::request_context::{self, spawn_traced};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
 use super::storage_api::object_usecase::set_disk::{
     get_lock_acquire_timeout, get_object_disk_read_timeout, is_valid_storage_class,
@@ -7506,50 +7506,31 @@ impl DefaultObjectUsecase {
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
 
-        let copy_commit = spawn_traced_join({
-            let store = Arc::clone(&store);
-            let src_bucket = src_bucket.clone();
-            let src_key = src_key.clone();
-            let bucket = bucket.clone();
-            let key = key.clone();
-            let src_opts = src_opts.clone();
-            let dst_opts = dst_opts.clone();
-            async move {
-                let _source_bucket_lifecycle_guard = source_bucket_lifecycle_guard;
-                let _destination_bucket_lifecycle_guard_storage = destination_bucket_lifecycle_guard_storage;
-                let _self_copy_lock_guard = _self_copy_lock_guard;
+        let oi = store
+            .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
+            .await
+            .map_err(ApiError::from)?;
+        drop(_self_copy_lock_guard);
 
-                let oi = store
-                    .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
-                    .await
-                    .map_err(ApiError::from)?;
+        // Reuse the single pre-commit replication decision (see `dsc` above) so
+        // the persisted pending marker and the schedule always agree, mirroring
+        // the PUT path.
+        if dsc.replicate_any() {
+            schedule_object_replication(oi.clone(), store.clone(), dsc).await;
+        }
 
-                // Reuse the single pre-commit replication decision (see `dsc` above) so
-                // the persisted pending marker and the schedule always agree, mirroring
-                // the PUT path.
-                if dsc.replicate_any() {
-                    schedule_object_replication(oi.clone(), Arc::clone(&store), dsc).await;
-                }
+        maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
+        let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
 
-                maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
-                let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
-
-                let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-                if has_bucket_metadata {
-                    if dest_versioned {
-                        record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
-                    } else {
-                        record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
-                    }
-                }
-
-                rustfs_scanner::record_dirty_usage_bucket(&bucket);
-                Ok::<_, S3Error>((oi, dest_versioned))
+        let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
+        // Update quota tracking after successful copy
+        if has_bucket_metadata {
+            if dest_versioned {
+                record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+            } else {
+                record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
             }
-        });
-        let (oi, dest_versioned) = copy_commit.await.map_err(|err| {
-            S3Error::with_message(S3ErrorCode::InternalError, format!("copy object commit owner task failed: {err}"))
-        })??;
+        }
 
         let raw_dest_version = oi.version_id.map(|v| v.to_string());
         let dest_version = if dest_versioned { raw_dest_version } else { None };
@@ -7597,7 +7578,7 @@ impl DefaultObjectUsecase {
             }
         }
         let copy_object_result = CopyObjectResult {
-            e_tag: oi.etag.as_ref().map(|etag| to_s3s_etag(etag)),
+            e_tag: oi.etag.map(|etag| to_s3s_etag(&etag)),
             last_modified: oi.mod_time.map(Timestamp::from),
             checksum_crc32: response_checksums.crc32,
             checksum_crc32c: response_checksums.crc32c,
@@ -7628,6 +7609,7 @@ impl DefaultObjectUsecase {
 
         let result = Ok(S3Response::new(output));
         let _ = helper.complete(&result);
+        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 

rustfs/src/app/storage_api.rs

@@ -998,7 +998,7 @@ pub(crate) mod options {
 }
 
 pub(crate) mod request_context {
-    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced, spawn_traced_join};
+    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
 }
 
 pub(crate) mod sse {

rustfs/src/storage/request_context.rs

@@ -257,15 +257,6 @@ where
     tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()));
 }
 
-/// Spawn a request-internal task and return its join handle to the caller.
-pub fn spawn_traced_join<F>(fut: F) -> tokio::task::JoinHandle<F::Output>
-where
-    F: std::future::Future + Send + 'static,
-    F::Output: Send + 'static,
-{
-    tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()))
-}
-
 #[cfg(test)]
 #[allow(unused_imports)]
 mod tests {

rustfs/src/storage/storage_api.rs

@@ -203,9 +203,7 @@ pub(crate) mod options_consumer {
 }
 
 pub(crate) mod request_context_consumer {
-    pub(crate) use super::super::request_context::{
-        RequestContext, extract_request_id_from_headers, spawn_traced, spawn_traced_join,
-    };
+    pub(crate) use super::super::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
 }
 
 pub(crate) mod rpc_consumer {

rustfs/src/table_catalog/test_support.rs

@@ -22,6 +22,15 @@
 //! fixed values (sequence 7 / snapshot 20), which keeps every produced byte
 //! identical to the pre-extraction fixtures.
 
+use std::collections::{BTreeMap, BTreeSet};
+use std::sync::Arc;
+use time::OffsetDateTime;
+
+use super::{
+    StrongTableCatalogRuntime, TableCatalogObject, TableCatalogObjectBackend, TableCatalogObjectMetadata,
+    TableCatalogPutPrecondition, TableCatalogStoreError, TableCatalogStoreResult, TableCommitPublication,
+};
+
 pub(crate) fn table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
     serde_json::json!({
         "format-version": 2,
@@ -226,3 +235,707 @@ pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i
     }
     writer.into_inner().expect("manifest avro bytes should flush")
 }
+
+// --- Stateful object backend shared by the store and admin handler tests
+// (backlog#1837 PR2). Superset instrumentation lands here incrementally;
+// this is the store-side fake moved verbatim.
+
+#[derive(Clone, Default)]
+pub(crate) struct TestCatalogObjectBackend {
+    pub(crate) state: Arc<tokio::sync::Mutex<TestCatalogObjectState>>,
+    pub(crate) locks: TestCatalogObjectLocks,
+    pub(crate) strong_runtime: Option<StrongTableCatalogRuntime>,
+    // One-shot, path-keyed injection knobs from the admin handler tests'
+    // former TestTableCatalogObjectBackend (backlog#1837 PR2): each fires
+    // once for the named object and clears itself, mirroring the original
+    // semantics exactly. They compose with (and run before) the store tests'
+    // attempt-indexed injection maps above.
+    pub(crate) put_object_barrier: Option<Arc<tokio::sync::Barrier>>,
+    pub(crate) fail_put_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
+    pub(crate) corrupt_put_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
+    pub(crate) missing_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
+    pub(crate) fail_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
+    pub(crate) lock_attempts: Arc<tokio::sync::Mutex<Vec<(String, String)>>>,
+    /// Content-addressed (sha256) etags instead of the store fake's counter.
+    /// The admin handler tests observe an object's etag and expect rewriting
+    /// identical bytes to reproduce it, so their fixtures set this.
+    pub(crate) content_addressed_etags: bool,
+}
+
+pub(crate) type TestCatalogObjectLockKey = (String, String);
+pub(crate) type TestCatalogObjectLock = Arc<tokio::sync::RwLock<()>>;
+pub(crate) type TestCatalogObjectLocks = Arc<tokio::sync::Mutex<BTreeMap<TestCatalogObjectLockKey, TestCatalogObjectLock>>>;
+
+#[derive(Clone, Default)]
+pub(crate) struct TestCatalogObjectPause {
+    started: Arc<tokio::sync::Notify>,
+    release: Arc<tokio::sync::Notify>,
+}
+
+impl TestCatalogObjectPause {
+    pub(crate) async fn wait_started(&self) {
+        self.started.notified().await;
+    }
+
+    pub(crate) fn release(&self) {
+        self.release.notify_one();
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct BlockingObjectPublication {
+    backend: TestCatalogObjectBackend,
+    object: String,
+    started: Arc<tokio::sync::Notify>,
+    guard: Arc<parking_lot::Mutex<Option<Box<dyn Send>>>>,
+}
+
+impl BlockingObjectPublication {
+    pub(crate) fn new(backend: TestCatalogObjectBackend, object: impl Into<String>) -> Self {
+        Self {
+            backend,
+            object: object.into(),
+            started: Arc::new(tokio::sync::Notify::new()),
+            guard: Arc::new(parking_lot::Mutex::new(None)),
+        }
+    }
+
+    pub(crate) async fn wait_started(&self) {
+        self.started.notified().await;
+    }
+}
+
+#[derive(Default)]
+pub(crate) struct UnserializedTestPublication;
+
+#[async_trait::async_trait]
+impl TableCommitPublication for UnserializedTestPublication {
+    async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
+    async fn prepare(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
+    fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
+        true
+    }
+
+    fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
+        true
+    }
+
+    fn complete(&self) {}
+}
+
+#[async_trait::async_trait]
+impl TableCommitPublication for BlockingObjectPublication {
+    async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
+    async fn prepare(&self, table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
+        self.started.notify_one();
+        let guard = self.backend.acquire_read_lock(table_bucket, &self.object).await?;
+        *self.guard.lock() = Some(guard);
+        Ok(())
+    }
+
+    fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
+        true
+    }
+
+    fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
+        self.guard.lock().is_some()
+    }
+
+    fn complete(&self) {
+        drop(self.guard.lock().take());
+    }
+}
+
+#[derive(Default)]
+pub(crate) struct TestCatalogObjectState {
+    pub(crate) objects: BTreeMap<(String, String), TestCatalogObjectRecord>,
+    pub(crate) etagless_objects: BTreeSet<(String, String)>,
+    pub(crate) fail_read_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
+    pub(crate) pause_before_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
+    pub(crate) pause_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
+    pub(crate) read_attempts: BTreeMap<(String, String), usize>,
+    pub(crate) read_limits: Vec<((String, String), usize)>,
+    pub(crate) fail_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
+    pub(crate) fail_after_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
+    pub(crate) pause_put_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
+    pub(crate) fail_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
+    pub(crate) fail_after_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
+    pub(crate) put_attempts: BTreeMap<(String, String), usize>,
+    pub(crate) delete_attempts: BTreeMap<(String, String), usize>,
+    pub(crate) write_lock_acquisitions: BTreeMap<(String, String), usize>,
+    pub(crate) read_lock_acquisitions: BTreeMap<(String, String), usize>,
+    pub(crate) read_calls: usize,
+    pub(crate) metadata_calls: usize,
+    pub(crate) list_calls: usize,
+    pub(crate) next_etag: u64,
+}
+
+#[derive(Clone)]
+pub(crate) struct TestCatalogObjectRecord {
+    pub(crate) data: Vec<u8>,
+    pub(crate) etag: String,
+    pub(crate) mod_time: Option<OffsetDateTime>,
+}
+
+impl TestCatalogObjectBackend {
+    pub(crate) async fn seed_object(&self, bucket: &str, object: &str, data: Vec<u8>) {
+        self.seed_object_with_mod_time(bucket, object, data, Some(OffsetDateTime::UNIX_EPOCH))
+            .await;
+    }
+
+    pub(crate) async fn seed_object_with_mod_time(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: Vec<u8>,
+        mod_time: Option<OffsetDateTime>,
+    ) {
+        let mut state = self.state.lock().await;
+        let etag = state.next_etag();
+        state
+            .objects
+            .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
+    }
+
+    pub(crate) async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) {
+        let mut state = self.state.lock().await;
+        state
+            .fail_put_attempts
+            .entry((bucket.to_string(), object.to_string()))
+            .or_default()
+            .insert(attempt);
+    }
+
+    pub(crate) async fn fail_delete_attempt(&self, bucket: &str, object: &str, attempt: usize) {
+        let mut state = self.state.lock().await;
+        state
+            .fail_delete_attempts
+            .entry((bucket.to_string(), object.to_string()))
+            .or_default()
+            .insert(attempt);
+    }
+
+    pub(crate) async fn list_call_count(&self) -> usize {
+        self.state.lock().await.list_calls
+    }
+
+    pub(crate) async fn read_call_count(&self) -> usize {
+        self.state.lock().await.read_calls
+    }
+
+    pub(crate) async fn metadata_call_count(&self) -> usize {
+        self.state.lock().await.metadata_calls
+    }
+
+    pub(crate) async fn reset_call_counts(&self) {
+        let mut state = self.state.lock().await;
+        state.read_calls = 0;
+        state.metadata_calls = 0;
+        state.list_calls = 0;
+    }
+
+    pub(crate) async fn write_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
+        self.state
+            .lock()
+            .await
+            .write_lock_acquisitions
+            .get(&(bucket.to_string(), object.to_string()))
+            .copied()
+            .unwrap_or_default()
+    }
+
+    pub(crate) async fn read_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
+        self.state
+            .lock()
+            .await
+            .read_lock_acquisitions
+            .get(&(bucket.to_string(), object.to_string()))
+            .copied()
+            .unwrap_or_default()
+    }
+
+    pub(crate) async fn fail_next_read(&self, bucket: &str, object: &str) {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
+        state.fail_read_attempts.entry(key).or_default().insert(next_attempt);
+    }
+
+    pub(crate) async fn pause_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
+        let pause = TestCatalogObjectPause::default();
+        state
+            .pause_read_attempts
+            .entry(key)
+            .or_default()
+            .insert(next_attempt, pause.clone());
+        pause
+    }
+
+    pub(crate) async fn pause_before_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
+        let pause = TestCatalogObjectPause::default();
+        state
+            .pause_before_read_attempts
+            .entry(key)
+            .or_default()
+            .insert(next_attempt, pause.clone());
+        pause
+    }
+
+    pub(crate) async fn omit_etag_for_object(&self, bucket: &str, object: &str) {
+        self.state
+            .lock()
+            .await
+            .etagless_objects
+            .insert((bucket.to_string(), object.to_string()));
+    }
+
+    pub(crate) async fn last_read_limit(&self, bucket: &str, object: &str) -> Option<usize> {
+        let key = (bucket.to_string(), object.to_string());
+        self.state
+            .lock()
+            .await
+            .read_limits
+            .iter()
+            .rev()
+            .find_map(|(read_key, limit)| (read_key == &key).then_some(*limit))
+    }
+
+    pub(crate) async fn fail_next_put(&self, bucket: &str, object: &str) {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
+        state.fail_put_attempts.entry(key).or_default().insert(next_attempt);
+    }
+
+    pub(crate) async fn fail_after_next_put(&self, bucket: &str, object: &str) {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
+        state.fail_after_put_attempts.entry(key).or_default().insert(next_attempt);
+    }
+
+    pub(crate) async fn fail_after_next_delete(&self, bucket: &str, object: &str) {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.delete_attempts.get(&key).copied().unwrap_or_default() + 1;
+        state.fail_after_delete_attempts.entry(key).or_default().insert(next_attempt);
+    }
+
+    pub(crate) async fn pause_next_put(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
+        let pause = TestCatalogObjectPause::default();
+        state
+            .pause_put_attempts
+            .entry(key)
+            .or_default()
+            .insert(next_attempt, pause.clone());
+        pause
+    }
+
+    pub(crate) async fn put_attempt_count(&self, bucket: &str, object: &str) -> usize {
+        self.state
+            .lock()
+            .await
+            .put_attempts
+            .get(&(bucket.to_string(), object.to_string()))
+            .copied()
+            .unwrap_or_default()
+    }
+}
+
+impl TestCatalogObjectState {
+    pub(crate) fn next_etag(&mut self) -> String {
+        self.next_etag += 1;
+        format!("etag-{}", self.next_etag)
+    }
+}
+
+#[async_trait::async_trait]
+impl TableCatalogObjectBackend for TestCatalogObjectBackend {
+    fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
+        self.strong_runtime.clone()
+    }
+
+    async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        let mut missing_read_object_path = self.missing_read_object_path.lock().await;
+        if missing_read_object_path.as_deref() == Some(object) {
+            missing_read_object_path.take();
+            return Ok(None);
+        }
+        drop(missing_read_object_path);
+
+        let mut fail_read_object_path = self.fail_read_object_path.lock().await;
+        if fail_read_object_path.as_deref() == Some(object) {
+            fail_read_object_path.take();
+            return Err(TableCatalogStoreError::Internal("private generated metadata read failure".to_string()));
+        }
+        drop(fail_read_object_path);
+
+        let key = (bucket.to_string(), object.to_string());
+        let (attempt, pause_before) = {
+            let mut state = self.state.lock().await;
+            state.read_calls += 1;
+            let attempt = {
+                let attempts = state.read_attempts.entry(key.clone()).or_default();
+                *attempts += 1;
+                *attempts
+            };
+            if state
+                .fail_read_attempts
+                .get(&key)
+                .is_some_and(|attempts| attempts.contains(&attempt))
+            {
+                return Err(TableCatalogStoreError::Internal(format!(
+                    "injected read failure for {object} attempt {attempt}"
+                )));
+            }
+            let pause = state
+                .pause_before_read_attempts
+                .get_mut(&key)
+                .and_then(|attempts| attempts.remove(&attempt));
+            (attempt, pause)
+        };
+        if let Some(pause) = pause_before {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
+        let (result, pause) = {
+            let mut state = self.state.lock().await;
+            let etagless = state.etagless_objects.contains(&key);
+            let result = state.objects.get(&key).map(|record| TableCatalogObject {
+                data: record.data.clone(),
+                etag: (!etagless).then(|| record.etag.clone()),
+                mod_time: record.mod_time,
+            });
+            let pause = state
+                .pause_read_attempts
+                .get_mut(&key)
+                .and_then(|attempts| attempts.remove(&attempt));
+            (result, pause)
+        };
+        if let Some(pause) = pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
+        Ok(result)
+    }
+
+    async fn read_object_limited(
+        &self,
+        bucket: &str,
+        object: &str,
+        max_size: usize,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        self.state
+            .lock()
+            .await
+            .read_limits
+            .push(((bucket.to_string(), object.to_string()), max_size));
+        let result = self.read_object(bucket, object).await?;
+        if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
+            return Err(TableCatalogStoreError::Invalid(format!(
+                "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
+            )));
+        }
+        Ok(result)
+    }
+
+    async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        let mut state = self.state.lock().await;
+        state.metadata_calls += 1;
+        let key = (bucket.to_string(), object.to_string());
+        let etagless = state.etagless_objects.contains(&key);
+        Ok(state.objects.get(&key).map(|record| TableCatalogObjectMetadata {
+            etag: (!etagless).then(|| record.etag.clone()),
+            mod_time: record.mod_time,
+        }))
+    }
+
+    async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
+        let state = self.state.lock().await;
+        Ok(state.objects.contains_key(&(bucket.to_string(), object.to_string())))
+    }
+
+    async fn put_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: Vec<u8>,
+        precondition: TableCatalogPutPrecondition,
+    ) -> TableCatalogStoreResult<()> {
+        let mut fail_put_object_path = self.fail_put_object_path.lock().await;
+        if fail_put_object_path.as_deref() == Some(object) {
+            fail_put_object_path.take();
+            return Err(TableCatalogStoreError::Internal("injected metadata write failure".to_string()));
+        }
+        drop(fail_put_object_path);
+
+        let mut corrupt_put_object_path = self.corrupt_put_object_path.lock().await;
+        let data = if corrupt_put_object_path.as_deref() == Some(object) {
+            corrupt_put_object_path.take();
+            b"{}".to_vec()
+        } else {
+            data
+        };
+        drop(corrupt_put_object_path);
+
+        let key = (bucket.to_string(), object.to_string());
+        let (attempt, pause) = {
+            let mut state = self.state.lock().await;
+            let attempt = {
+                let attempts = state.put_attempts.entry(key.clone()).or_default();
+                *attempts += 1;
+                *attempts
+            };
+            if state
+                .fail_put_attempts
+                .get(&key)
+                .is_some_and(|attempts| attempts.contains(&attempt))
+            {
+                return Err(TableCatalogStoreError::Internal(format!(
+                    "injected put failure for {object} attempt {attempt}"
+                )));
+            }
+            let pause = state
+                .pause_put_attempts
+                .get_mut(&key)
+                .and_then(|attempts| attempts.remove(&attempt));
+            (attempt, pause)
+        };
+        if let Some(pause) = pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
+
+        let result = {
+            let mut state = self.state.lock().await;
+            let precondition_failure = match &precondition {
+                TableCatalogPutPrecondition::IfAbsent if state.objects.contains_key(&key) => {
+                    Some(TableCatalogStoreError::Conflict(format!("object already exists: {object}")))
+                }
+                TableCatalogPutPrecondition::IfMatch(expected) => match state.objects.get(&key) {
+                    None => Some(TableCatalogStoreError::Conflict(format!("object is missing: {object}"))),
+                    Some(current) if &current.etag != expected => {
+                        Some(TableCatalogStoreError::Conflict(format!("object changed: {object}")))
+                    }
+                    Some(_) => None,
+                },
+                _ => None,
+            };
+            if let Some(err) = precondition_failure {
+                Err(err)
+            } else {
+                let etag = if self.content_addressed_etags {
+                    content_etag(&data)
+                } else {
+                    state.next_etag()
+                };
+                state.objects.insert(
+                    key.clone(),
+                    TestCatalogObjectRecord {
+                        data,
+                        etag,
+                        mod_time: Some(OffsetDateTime::now_utc()),
+                    },
+                );
+                if state
+                    .fail_after_put_attempts
+                    .get(&key)
+                    .is_some_and(|attempts| attempts.contains(&attempt))
+                {
+                    Err(TableCatalogStoreError::Internal(format!(
+                        "injected post-commit put failure for {object} attempt {attempt}"
+                    )))
+                } else {
+                    Ok(())
+                }
+            }
+        };
+        if let Some(barrier) = &self.put_object_barrier {
+            barrier.wait().await;
+        }
+        result
+    }
+
+    async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
+        let mut state = self.state.lock().await;
+        let key = (bucket.to_string(), object.to_string());
+        let attempt = {
+            let attempts = state.delete_attempts.entry(key.clone()).or_default();
+            *attempts += 1;
+            *attempts
+        };
+        if state
+            .fail_delete_attempts
+            .get(&key)
+            .is_some_and(|attempts| attempts.contains(&attempt))
+        {
+            return Err(TableCatalogStoreError::Internal(format!(
+                "injected delete failure for {object} attempt {attempt}"
+            )));
+        }
+        state.objects.remove(&key);
+        if state
+            .fail_after_delete_attempts
+            .get(&key)
+            .is_some_and(|attempts| attempts.contains(&attempt))
+        {
+            return Err(TableCatalogStoreError::Internal(format!(
+                "injected post-commit delete failure for {object} attempt {attempt}"
+            )));
+        }
+        Ok(())
+    }
+
+    async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
+        let mut state = self.state.lock().await;
+        state.list_calls += 1;
+        Ok(state
+            .objects
+            .keys()
+            .filter(|(entry_bucket, object)| entry_bucket == bucket && object.starts_with(prefix))
+            .map(|(_, object)| object.clone())
+            .collect())
+    }
+
+    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+        self.lock_attempts.lock().await.push((bucket.to_string(), object.to_string()));
+        {
+            let mut state = self.state.lock().await;
+            *state
+                .write_lock_acquisitions
+                .entry((bucket.to_string(), object.to_string()))
+                .or_default() += 1;
+        }
+        let lock = {
+            let mut locks = self.locks.lock().await;
+            locks
+                .entry((bucket.to_string(), object.to_string()))
+                .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
+                .clone()
+        };
+        Ok(Box::new(lock.write_owned().await))
+    }
+
+    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+        // The admin fake implemented only acquire_write_lock, so the trait's
+        // default read->write delegation made read acquisitions observable in
+        // lock_attempts as well; keep that (backlog#1837 PR2).
+        self.lock_attempts.lock().await.push((bucket.to_string(), object.to_string()));
+        {
+            let mut state = self.state.lock().await;
+            *state
+                .read_lock_acquisitions
+                .entry((bucket.to_string(), object.to_string()))
+                .or_default() += 1;
+        }
+        let lock = {
+            let mut locks = self.locks.lock().await;
+            locks
+                .entry((bucket.to_string(), object.to_string()))
+                .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
+                .clone()
+        };
+        Ok(Box::new(lock.read_owned().await))
+    }
+}
+
+fn content_etag(data: &[u8]) -> String {
+    use sha2::Digest;
+    hex_simd::encode_to_string(sha2::Sha256::digest(data), hex_simd::AsciiCase::Lower)
+}
+
+/// Admin-handler-test conveniences carried over from the former
+/// TestTableCatalogObjectBackend (backlog#1837 PR2): content-addressed etags
+/// (sha256), direct record insertion, and lock observability.
+impl TestCatalogObjectBackend {
+    /// Fake with the admin fixtures' content-addressed etag semantics.
+    pub(crate) fn content_addressed() -> Self {
+        Self {
+            content_addressed_etags: true,
+            ..Self::default()
+        }
+    }
+
+    pub(crate) async fn put_bytes(&self, bucket: &str, object: &str, data: Vec<u8>) {
+        let etag = content_etag(&data);
+        self.state.lock().await.objects.insert(
+            (bucket.to_string(), object.to_string()),
+            TestCatalogObjectRecord {
+                data,
+                etag,
+                mod_time: None,
+            },
+        );
+    }
+
+    pub(crate) async fn put_json(&self, bucket: &str, object: &str, value: serde_json::Value) {
+        self.put_json_with_mod_time(bucket, object, value, None).await;
+    }
+
+    pub(crate) async fn put_gzip_json(&self, bucket: &str, object: &str, value: serde_json::Value) {
+        use std::io::Write;
+
+        let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        encoder.write_all(&data).expect("metadata JSON should compress");
+        self.put_bytes(bucket, object, encoder.finish().expect("metadata gzip stream should finish"))
+            .await;
+    }
+
+    pub(crate) async fn put_json_with_mod_time(
+        &self,
+        bucket: &str,
+        object: &str,
+        value: serde_json::Value,
+        mod_time: Option<OffsetDateTime>,
+    ) {
+        let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
+        let etag = content_etag(&data);
+        self.state
+            .lock()
+            .await
+            .objects
+            .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
+    }
+
+    pub(crate) async fn write_lock_is_held(&self, bucket: &str, object: &str) -> bool {
+        let lock = self
+            .locks
+            .lock()
+            .await
+            .get(&(bucket.to_string(), object.to_string()))
+            .cloned();
+        lock.is_some_and(|lock| lock.try_write_owned().is_err())
+    }
+
+    pub(crate) async fn wait_for_lock_attempts(&self, count: usize) {
+        tokio::time::timeout(std::time::Duration::from_secs(2), async {
+            loop {
+                if self.lock_attempts.lock().await.len() >= count {
+                    return;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("lock acquisition attempts should be observable");
+    }
+}

rustfs/src/table_catalog/tests.rs

@@ -3,6 +3,7 @@ use super::identifier::{
     default_table_lifecycle_path, default_table_marker_path, default_table_root_prefix, is_valid_table_metadata_file_name,
     namespace_name_from_marker_path, table_name_from_marker_path, validate_object_mutation,
 };
+use super::test_support::{BlockingObjectPublication, TestCatalogObjectBackend, UnserializedTestPublication};
 use super::*;
 use datafusion::{
     arrow::{
@@ -1077,312 +1078,6 @@ fn catalog_object_entry_paths_use_internal_root_and_hashed_untrusted_ids() {
     }
 }
 
-#[derive(Clone, Default)]
-struct TestCatalogObjectBackend {
-    state: Arc<tokio::sync::Mutex<TestCatalogObjectState>>,
-    locks: TestCatalogObjectLocks,
-    strong_runtime: Option<StrongTableCatalogRuntime>,
-}
-
-type TestCatalogObjectLockKey = (String, String);
-type TestCatalogObjectLock = Arc<tokio::sync::RwLock<()>>;
-type TestCatalogObjectLocks = Arc<tokio::sync::Mutex<BTreeMap<TestCatalogObjectLockKey, TestCatalogObjectLock>>>;
-
-#[derive(Clone, Default)]
-struct TestCatalogObjectPause {
-    started: Arc<tokio::sync::Notify>,
-    release: Arc<tokio::sync::Notify>,
-}
-
-impl TestCatalogObjectPause {
-    async fn wait_started(&self) {
-        self.started.notified().await;
-    }
-
-    fn release(&self) {
-        self.release.notify_one();
-    }
-}
-
-#[derive(Clone)]
-struct BlockingObjectPublication {
-    backend: TestCatalogObjectBackend,
-    object: String,
-    started: Arc<tokio::sync::Notify>,
-    guard: Arc<parking_lot::Mutex<Option<Box<dyn Send>>>>,
-}
-
-impl BlockingObjectPublication {
-    fn new(backend: TestCatalogObjectBackend, object: impl Into<String>) -> Self {
-        Self {
-            backend,
-            object: object.into(),
-            started: Arc::new(tokio::sync::Notify::new()),
-            guard: Arc::new(parking_lot::Mutex::new(None)),
-        }
-    }
-
-    async fn wait_started(&self) {
-        self.started.notified().await;
-    }
-}
-
-#[derive(Default)]
-struct UnserializedTestPublication;
-
-#[async_trait::async_trait]
-impl TableCommitPublication for UnserializedTestPublication {
-    async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
-        Ok(())
-    }
-
-    async fn prepare(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
-        Ok(())
-    }
-
-    fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
-        true
-    }
-
-    fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
-        true
-    }
-
-    fn complete(&self) {}
-}
-
-#[async_trait::async_trait]
-impl TableCommitPublication for BlockingObjectPublication {
-    async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
-        Ok(())
-    }
-
-    async fn prepare(&self, table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
-        self.started.notify_one();
-        let guard = self.backend.acquire_read_lock(table_bucket, &self.object).await?;
-        *self.guard.lock() = Some(guard);
-        Ok(())
-    }
-
-    fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
-        true
-    }
-
-    fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
-        self.guard.lock().is_some()
-    }
-
-    fn complete(&self) {
-        drop(self.guard.lock().take());
-    }
-}
-
-#[derive(Default)]
-struct TestCatalogObjectState {
-    objects: BTreeMap<(String, String), TestCatalogObjectRecord>,
-    etagless_objects: BTreeSet<(String, String)>,
-    fail_read_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
-    pause_before_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
-    pause_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
-    read_attempts: BTreeMap<(String, String), usize>,
-    read_limits: Vec<((String, String), usize)>,
-    fail_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
-    fail_after_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
-    pause_put_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
-    fail_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
-    fail_after_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
-    put_attempts: BTreeMap<(String, String), usize>,
-    delete_attempts: BTreeMap<(String, String), usize>,
-    write_lock_acquisitions: BTreeMap<(String, String), usize>,
-    read_lock_acquisitions: BTreeMap<(String, String), usize>,
-    read_calls: usize,
-    metadata_calls: usize,
-    list_calls: usize,
-    next_etag: u64,
-}
-
-#[derive(Clone)]
-struct TestCatalogObjectRecord {
-    data: Vec<u8>,
-    etag: String,
-    mod_time: Option<OffsetDateTime>,
-}
-
-impl TestCatalogObjectBackend {
-    async fn seed_object(&self, bucket: &str, object: &str, data: Vec<u8>) {
-        self.seed_object_with_mod_time(bucket, object, data, Some(OffsetDateTime::UNIX_EPOCH))
-            .await;
-    }
-
-    async fn seed_object_with_mod_time(&self, bucket: &str, object: &str, data: Vec<u8>, mod_time: Option<OffsetDateTime>) {
-        let mut state = self.state.lock().await;
-        let etag = state.next_etag();
-        state
-            .objects
-            .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
-    }
-
-    async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) {
-        let mut state = self.state.lock().await;
-        state
-            .fail_put_attempts
-            .entry((bucket.to_string(), object.to_string()))
-            .or_default()
-            .insert(attempt);
-    }
-
-    async fn fail_delete_attempt(&self, bucket: &str, object: &str, attempt: usize) {
-        let mut state = self.state.lock().await;
-        state
-            .fail_delete_attempts
-            .entry((bucket.to_string(), object.to_string()))
-            .or_default()
-            .insert(attempt);
-    }
-
-    async fn list_call_count(&self) -> usize {
-        self.state.lock().await.list_calls
-    }
-
-    async fn read_call_count(&self) -> usize {
-        self.state.lock().await.read_calls
-    }
-
-    async fn metadata_call_count(&self) -> usize {
-        self.state.lock().await.metadata_calls
-    }
-
-    async fn reset_call_counts(&self) {
-        let mut state = self.state.lock().await;
-        state.read_calls = 0;
-        state.metadata_calls = 0;
-        state.list_calls = 0;
-    }
-
-    async fn write_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
-        self.state
-            .lock()
-            .await
-            .write_lock_acquisitions
-            .get(&(bucket.to_string(), object.to_string()))
-            .copied()
-            .unwrap_or_default()
-    }
-
-    async fn read_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
-        self.state
-            .lock()
-            .await
-            .read_lock_acquisitions
-            .get(&(bucket.to_string(), object.to_string()))
-            .copied()
-            .unwrap_or_default()
-    }
-
-    async fn fail_next_read(&self, bucket: &str, object: &str) {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
-        state.fail_read_attempts.entry(key).or_default().insert(next_attempt);
-    }
-
-    async fn pause_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
-        let pause = TestCatalogObjectPause::default();
-        state
-            .pause_read_attempts
-            .entry(key)
-            .or_default()
-            .insert(next_attempt, pause.clone());
-        pause
-    }
-
-    async fn pause_before_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
-        let pause = TestCatalogObjectPause::default();
-        state
-            .pause_before_read_attempts
-            .entry(key)
-            .or_default()
-            .insert(next_attempt, pause.clone());
-        pause
-    }
-
-    async fn omit_etag_for_object(&self, bucket: &str, object: &str) {
-        self.state
-            .lock()
-            .await
-            .etagless_objects
-            .insert((bucket.to_string(), object.to_string()));
-    }
-
-    async fn last_read_limit(&self, bucket: &str, object: &str) -> Option<usize> {
-        let key = (bucket.to_string(), object.to_string());
-        self.state
-            .lock()
-            .await
-            .read_limits
-            .iter()
-            .rev()
-            .find_map(|(read_key, limit)| (read_key == &key).then_some(*limit))
-    }
-
-    async fn fail_next_put(&self, bucket: &str, object: &str) {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
-        state.fail_put_attempts.entry(key).or_default().insert(next_attempt);
-    }
-
-    async fn fail_after_next_put(&self, bucket: &str, object: &str) {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
-        state.fail_after_put_attempts.entry(key).or_default().insert(next_attempt);
-    }
-
-    async fn fail_after_next_delete(&self, bucket: &str, object: &str) {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.delete_attempts.get(&key).copied().unwrap_or_default() + 1;
-        state.fail_after_delete_attempts.entry(key).or_default().insert(next_attempt);
-    }
-
-    async fn pause_next_put(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
-        let pause = TestCatalogObjectPause::default();
-        state
-            .pause_put_attempts
-            .entry(key)
-            .or_default()
-            .insert(next_attempt, pause.clone());
-        pause
-    }
-
-    async fn put_attempt_count(&self, bucket: &str, object: &str) -> usize {
-        self.state
-            .lock()
-            .await
-            .put_attempts
-            .get(&(bucket.to_string(), object.to_string()))
-            .copied()
-            .unwrap_or_default()
-    }
-}
-
-impl TestCatalogObjectState {
-    fn next_etag(&mut self) -> String {
-        self.next_etag += 1;
-        format!("etag-{}", self.next_etag)
-    }
-}
-
 fn maintenance_object_report<'a>(
     report: &'a TableMetadataMaintenanceReport,
     metadata_location: &str,
@@ -2466,248 +2161,6 @@ fn parquet_i32_values(data: Vec<u8>) -> Vec<i32> {
     values
 }
 
-#[async_trait::async_trait]
-impl TableCatalogObjectBackend for TestCatalogObjectBackend {
-    fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
-        self.strong_runtime.clone()
-    }
-
-    async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
-        let key = (bucket.to_string(), object.to_string());
-        let (attempt, pause_before) = {
-            let mut state = self.state.lock().await;
-            state.read_calls += 1;
-            let attempt = {
-                let attempts = state.read_attempts.entry(key.clone()).or_default();
-                *attempts += 1;
-                *attempts
-            };
-            if state
-                .fail_read_attempts
-                .get(&key)
-                .is_some_and(|attempts| attempts.contains(&attempt))
-            {
-                return Err(TableCatalogStoreError::Internal(format!(
-                    "injected read failure for {object} attempt {attempt}"
-                )));
-            }
-            let pause = state
-                .pause_before_read_attempts
-                .get_mut(&key)
-                .and_then(|attempts| attempts.remove(&attempt));
-            (attempt, pause)
-        };
-        if let Some(pause) = pause_before {
-            pause.started.notify_one();
-            pause.release.notified().await;
-        }
-        let (result, pause) = {
-            let mut state = self.state.lock().await;
-            let etagless = state.etagless_objects.contains(&key);
-            let result = state.objects.get(&key).map(|record| TableCatalogObject {
-                data: record.data.clone(),
-                etag: (!etagless).then(|| record.etag.clone()),
-                mod_time: record.mod_time,
-            });
-            let pause = state
-                .pause_read_attempts
-                .get_mut(&key)
-                .and_then(|attempts| attempts.remove(&attempt));
-            (result, pause)
-        };
-        if let Some(pause) = pause {
-            pause.started.notify_one();
-            pause.release.notified().await;
-        }
-        Ok(result)
-    }
-
-    async fn read_object_limited(
-        &self,
-        bucket: &str,
-        object: &str,
-        max_size: usize,
-    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
-        self.state
-            .lock()
-            .await
-            .read_limits
-            .push(((bucket.to_string(), object.to_string()), max_size));
-        let result = self.read_object(bucket, object).await?;
-        if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
-            )));
-        }
-        Ok(result)
-    }
-
-    async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
-        let mut state = self.state.lock().await;
-        state.metadata_calls += 1;
-        let key = (bucket.to_string(), object.to_string());
-        let etagless = state.etagless_objects.contains(&key);
-        Ok(state.objects.get(&key).map(|record| TableCatalogObjectMetadata {
-            etag: (!etagless).then(|| record.etag.clone()),
-            mod_time: record.mod_time,
-        }))
-    }
-
-    async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
-        let state = self.state.lock().await;
-        Ok(state.objects.contains_key(&(bucket.to_string(), object.to_string())))
-    }
-
-    async fn put_object(
-        &self,
-        bucket: &str,
-        object: &str,
-        data: Vec<u8>,
-        precondition: TableCatalogPutPrecondition,
-    ) -> TableCatalogStoreResult<()> {
-        let key = (bucket.to_string(), object.to_string());
-        let (attempt, pause) = {
-            let mut state = self.state.lock().await;
-            let attempt = {
-                let attempts = state.put_attempts.entry(key.clone()).or_default();
-                *attempts += 1;
-                *attempts
-            };
-            if state
-                .fail_put_attempts
-                .get(&key)
-                .is_some_and(|attempts| attempts.contains(&attempt))
-            {
-                return Err(TableCatalogStoreError::Internal(format!(
-                    "injected put failure for {object} attempt {attempt}"
-                )));
-            }
-            let pause = state
-                .pause_put_attempts
-                .get_mut(&key)
-                .and_then(|attempts| attempts.remove(&attempt));
-            (attempt, pause)
-        };
-        if let Some(pause) = pause {
-            pause.started.notify_one();
-            pause.release.notified().await;
-        }
-
-        let mut state = self.state.lock().await;
-        match precondition {
-            TableCatalogPutPrecondition::IfAbsent if state.objects.contains_key(&key) => {
-                return Err(TableCatalogStoreError::Conflict(format!("object already exists: {object}")));
-            }
-            TableCatalogPutPrecondition::IfMatch(expected) => {
-                let Some(current) = state.objects.get(&key) else {
-                    return Err(TableCatalogStoreError::Conflict(format!("object is missing: {object}")));
-                };
-                if current.etag != expected {
-                    return Err(TableCatalogStoreError::Conflict(format!("object changed: {object}")));
-                }
-            }
-            _ => {}
-        }
-
-        let etag = state.next_etag();
-        state.objects.insert(
-            key.clone(),
-            TestCatalogObjectRecord {
-                data,
-                etag,
-                mod_time: Some(OffsetDateTime::now_utc()),
-            },
-        );
-        if state
-            .fail_after_put_attempts
-            .get(&key)
-            .is_some_and(|attempts| attempts.contains(&attempt))
-        {
-            return Err(TableCatalogStoreError::Internal(format!(
-                "injected post-commit put failure for {object} attempt {attempt}"
-            )));
-        }
-        Ok(())
-    }
-
-    async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
-        let mut state = self.state.lock().await;
-        let key = (bucket.to_string(), object.to_string());
-        let attempt = {
-            let attempts = state.delete_attempts.entry(key.clone()).or_default();
-            *attempts += 1;
-            *attempts
-        };
-        if state
-            .fail_delete_attempts
-            .get(&key)
-            .is_some_and(|attempts| attempts.contains(&attempt))
-        {
-            return Err(TableCatalogStoreError::Internal(format!(
-                "injected delete failure for {object} attempt {attempt}"
-            )));
-        }
-        state.objects.remove(&key);
-        if state
-            .fail_after_delete_attempts
-            .get(&key)
-            .is_some_and(|attempts| attempts.contains(&attempt))
-        {
-            return Err(TableCatalogStoreError::Internal(format!(
-                "injected post-commit delete failure for {object} attempt {attempt}"
-            )));
-        }
-        Ok(())
-    }
-
-    async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
-        let mut state = self.state.lock().await;
-        state.list_calls += 1;
-        Ok(state
-            .objects
-            .keys()
-            .filter(|(entry_bucket, object)| entry_bucket == bucket && object.starts_with(prefix))
-            .map(|(_, object)| object.clone())
-            .collect())
-    }
-
-    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
-        {
-            let mut state = self.state.lock().await;
-            *state
-                .write_lock_acquisitions
-                .entry((bucket.to_string(), object.to_string()))
-                .or_default() += 1;
-        }
-        let lock = {
-            let mut locks = self.locks.lock().await;
-            locks
-                .entry((bucket.to_string(), object.to_string()))
-                .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
-                .clone()
-        };
-        Ok(Box::new(lock.write_owned().await))
-    }
-
-    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
-        {
-            let mut state = self.state.lock().await;
-            *state
-                .read_lock_acquisitions
-                .entry((bucket.to_string(), object.to_string()))
-                .or_default() += 1;
-        }
-        let lock = {
-            let mut locks = self.locks.lock().await;
-            locks
-                .entry((bucket.to_string(), object.to_string()))
-                .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
-                .clone()
-        };
-        Ok(Box::new(lock.read_owned().await))
-    }
-}
-
 fn test_bucket_entry(bucket: &str) -> TableBucketEntry {
     TableBucketEntry {
         version: TABLE_CATALOG_ENTRY_VERSION,