diff --git a/rustfs/src/table_catalog/test_support.rs b/rustfs/src/table_catalog/test_support.rs
index 374d18ce8..6b2f955c8 100644
--- a/rustfs/src/table_catalog/test_support.rs
+++ b/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,561 @@ 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>,
+}
+
+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 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))
+    }
+}
diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs
index afe497697..f8a0346c2 100644
--- a/rustfs/src/table_catalog/tests.rs
+++ b/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,