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).

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