Merge branch 'main' into cxymds/fix-1853-lock-lease-reaper

crates/log-analyzer/src/rules/seed/heal.rs

@@ -97,7 +97,7 @@ pub(super) fn rules() -> Vec<Rule> {
             )
         },
         Rule {
-            anchors: strings(["Heal task execution failed", "Heal manager is not running"]),
+            anchors: strings(["Heal task execution failed"]),
             ..base(
                 "heal-task-failure",
                 P2Degraded,

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,