fix(tables): retain dropped warehouse protection (#7675)

rustfs/src/table_catalog/iceberg/validation.rs

@@ -270,6 +270,29 @@ pub(crate) fn table_data_plane_resource_from_entry(table: TableEntry, warehouse_
     }
 }
 
+pub(crate) fn table_data_plane_resource_from_warehouse_index(
+    index: &TableWarehouseIndexEntry,
+) -> TableCatalogStoreResult<TableDataPlaneResource> {
+    if index.table_bucket.is_empty() || index.table_id.is_empty() {
+        return Err(TableCatalogStoreError::Invalid(
+            "warehouse index has an empty table bucket or table id".to_string(),
+        ));
+    }
+    parse_namespace_for_store(&index.namespace)?;
+    parse_table_for_store(&index.table)?;
+    let normalized = normalize_warehouse_object_prefix(&index.warehouse_object_prefix, Some(WAREHOUSE_INDEX_MAX_PREFIX_DEPTH))?;
+    if normalized != index.warehouse_object_prefix {
+        return Err(TableCatalogStoreError::Invalid("warehouse index prefix is not canonical".to_string()));
+    }
+    Ok(TableDataPlaneResource {
+        table_bucket: index.table_bucket.clone(),
+        namespace: index.namespace.clone(),
+        table: index.table.clone(),
+        table_id: index.table_id.clone(),
+        warehouse_object_prefix: index.warehouse_object_prefix.clone(),
+    })
+}
+
 pub(crate) async fn table_data_plane_resource_for_object<S>(
     store: &S,
     bucket: &str,

rustfs/src/table_catalog/model.rs

@@ -258,10 +258,11 @@ pub(crate) struct TableWarehouseIndexStateEntry {
     pub(super) state: TableCatalogEntryState,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum WarehouseIndexReservation {
     Created,
     AlreadyReserved,
+    Replaced(TableWarehouseIndexEntry),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

rustfs/src/table_catalog/store/object.rs

@@ -239,6 +239,12 @@ struct ActiveNamespaceEvidence {
     explicit_entry: Option<NamespaceEntry>,
 }
 
+enum TableWarehouseIndexResolution {
+    Active(TableDataPlaneResource),
+    Deleted(TableDataPlaneResource),
+    Missing,
+}
+
 #[derive(Clone)]
 pub(crate) struct ObjectTableCatalogStore<B> {
     pub(in crate::table_catalog) backend: B,
@@ -1272,15 +1278,14 @@ where
         Ok(())
     }
 
-    async fn replace_stale_table_warehouse_index(
+    async fn replace_table_warehouse_index(
         &self,
         object: &str,
         stale: &TableWarehouseIndexEntry,
         replacement: &TableWarehouseIndexEntry,
-        reason: &'static str,
     ) -> TableCatalogStoreResult<bool> {
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), object).await?;
-        let Some((current, _)) = self
+        let Some((current, current_etag)) = self
             .read_entry_unlocked::<TableWarehouseIndexEntry>(self.catalog_bucket(), object)
             .await?
         else {
@@ -1290,9 +1295,16 @@ where
         if current != *stale {
             return Ok(false);
         }
-        self.delete_warehouse_index_object_unlocked(object, stale, reason).await?;
-        self.write_entry_unlocked(self.catalog_bucket(), object, replacement, TableCatalogPutPrecondition::IfAbsent)
-            .await?;
+        let current_etag = current_etag
+            .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog warehouse index has no etag: {object}")))?;
+        // A missing index would expose residual data during failed prefix reuse.
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            object,
+            replacement,
+            TableCatalogPutPrecondition::IfMatch(current_etag),
+        )
+        .await?;
         Ok(true)
     }
 
@@ -1398,11 +1410,8 @@ where
                             index.warehouse_object_prefix
                         )));
                     }
-                    if self
-                        .replace_stale_table_warehouse_index(&object, &existing, &index, "stale reservation conflict")
-                        .await?
-                    {
-                        return Ok(WarehouseIndexReservation::Created);
+                    if self.replace_table_warehouse_index(&object, &existing, &index).await? {
+                        return Ok(WarehouseIndexReservation::Replaced(existing));
                     }
                 }
                 Err(err) => return Err(err),
@@ -1422,12 +1431,12 @@ where
             .map_err(|err| TableCatalogStoreError::Internal(format!("failed to delete stale warehouse index {object}: {err}")))
     }
 
-    async fn fail_closed_for_broken_warehouse_index(
+    async fn fail_closed_for_broken_warehouse_index<T>(
         &self,
         object: &str,
         index: &TableWarehouseIndexEntry,
         reason: &'static str,
-    ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
+    ) -> TableCatalogStoreResult<Option<T>> {
         Err(TableCatalogStoreError::Internal(format!(
             "active warehouse index {object} for {}/{}/{} ({}) is inconsistent: {reason}",
             index.table_bucket, index.namespace, index.table, index.table_id
@@ -1438,11 +1447,15 @@ where
         &self,
         index_object: &str,
         index: TableWarehouseIndexEntry,
-    ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
+    ) -> TableCatalogStoreResult<Option<(TableCatalogEntryState, TableDataPlaneResource)>> {
         validate_table_warehouse_index_entry_object(&self.paths, index_object, &index)?;
+        if index.state == TableCatalogEntryState::Deleted {
+            let resource = table_data_plane_resource_from_warehouse_index(&index)?;
+            return Ok(Some((TableCatalogEntryState::Deleted, resource)));
+        }
         if index.state != TableCatalogEntryState::Active {
             return Err(TableCatalogStoreError::Internal(format!(
-                "warehouse index {index_object} for {}/{}/{} is inactive while the index is authoritative",
+                "warehouse index {index_object} for {}/{}/{} has an invalid transient state while the index is authoritative",
                 index.table_bucket, index.namespace, index.table
             )));
         }
@@ -1472,7 +1485,10 @@ where
                 .fail_closed_for_broken_warehouse_index(index_object, &index, "referenced table identity changed")
                 .await;
         }
-        Ok(Some(table_data_plane_resource_from_entry(table, current_prefix)))
+        Ok(Some((
+            TableCatalogEntryState::Active,
+            table_data_plane_resource_from_entry(table, current_prefix),
+        )))
     }
 
     async fn read_warehouse_index_state_unlocked(&self, table_bucket: &str) -> TableCatalogStoreResult<bool> {
@@ -1488,17 +1504,41 @@ where
         table_warehouse_index_state_ready(&state, table_bucket)
     }
 
-    async fn delete_created_table_warehouse_index(
+    async fn rollback_table_warehouse_index_reservation(
         &self,
         entry: &TableEntry,
         reservation: WarehouseIndexReservation,
         reason: &'static str,
     ) {
-        if reservation != WarehouseIndexReservation::Created {
-            return;
+        let result = async {
+            match reservation {
+                WarehouseIndexReservation::AlreadyReserved => Ok(()),
+                WarehouseIndexReservation::Created => self.delete_table_warehouse_index(entry).await,
+                WarehouseIndexReservation::Replaced(previous) => {
+                    let index = table_warehouse_index_entry(entry)?;
+                    // An error response can follow a committed catalog write.
+                    if let Some(current) = self
+                        .load_table_entry(&entry.table_bucket, &entry.namespace, &entry.table)
+                        .await?
+                        && table_warehouse_index_entry(&current)? == index
+                    {
+                        return Ok(());
+                    }
+                    self.replace_table_warehouse_index(
+                        &self
+                            .paths
+                            .warehouse_index_entry_path(&index.table_bucket, &index.warehouse_object_prefix),
+                        &index,
+                        &previous,
+                    )
+                    .await
+                    .map(|_| ())
+                }
+            }
         }
+        .await;
         let warehouse_object_prefix = table_warehouse_object_prefix(entry).ok();
-        if let Err(err) = self.delete_table_warehouse_index(entry).await {
+        if let Err(err) = result {
             tracing::warn!(
                 table_bucket = %entry.table_bucket,
                 namespace = %entry.namespace,
@@ -1527,28 +1567,46 @@ where
         .map(|_| ())
     }
 
-    async fn delete_owned_table_warehouse_index_for_drop(&self, entry: &TableEntry) -> TableCatalogStoreResult<()> {
+    pub(in crate::table_catalog) async fn tombstone_table_warehouse_index_for_drop(
+        &self,
+        entry: &TableEntry,
+        replace_deleted_owner: bool,
+    ) -> TableCatalogStoreResult<()> {
         let index = table_warehouse_index_entry(entry)?;
+        let mut tombstone = index.clone();
+        tombstone.state = TableCatalogEntryState::Deleted;
         let object = self
             .paths
             .warehouse_index_entry_path(&index.table_bucket, &index.warehouse_object_prefix);
         validate_table_warehouse_index_entry_object(&self.paths, &object, &index)?;
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?;
-        let Some((current, _)) = self
+        let Some((current, current_etag)) = self
             .read_entry_unlocked::<TableWarehouseIndexEntry>(self.catalog_bucket(), &object)
             .await?
         else {
-            return Ok(());
+            return self
+                .write_entry_unlocked(self.catalog_bucket(), &object, &tombstone, TableCatalogPutPrecondition::IfAbsent)
+                .await;
         };
         validate_table_warehouse_index_entry_object(&self.paths, &object, &current)?;
-        if current != index {
+        if current == tombstone {
+            return Ok(());
+        }
+        if current != index && !(replace_deleted_owner && current.state == TableCatalogEntryState::Deleted) {
             return Err(TableCatalogStoreError::Conflict(format!(
                 "table warehouse index owner changed before drop: {}",
                 index.warehouse_object_prefix
             )));
         }
-        self.delete_warehouse_index_object_unlocked(&object, &index, "table warehouse index owner dropped")
-            .await
+        let current_etag = current_etag
+            .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog warehouse index has no etag: {object}")))?;
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            &object,
+            &tombstone,
+            TableCatalogPutPrecondition::IfMatch(current_etag),
+        )
+        .await
     }
 
     async fn restore_table_warehouse_index_after_failed_drop(&self, entry: &TableEntry, reason: &'static str) {
@@ -1592,8 +1650,10 @@ where
         &self,
         table_bucket: &str,
         object: &str,
-    ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
-        let mut matched: Option<TableDataPlaneResource> = None;
+    ) -> TableCatalogStoreResult<TableWarehouseIndexResolution> {
+        let mut active: Option<TableDataPlaneResource> = None;
+        let mut deleted: Option<TableDataPlaneResource> = None;
+        let mut deleted_overlap = None;
         for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) {
             let index_object = self.paths.warehouse_index_entry_path(table_bucket, warehouse_object_prefix);
             let Some((index, _)) = self
@@ -1607,19 +1667,107 @@ where
                     "warehouse index entry does not match indexed prefix: {index_object}"
                 )));
             }
-            if let Some(resource) = self
+            if let Some((state, resource)) = self
                 .resolve_table_data_plane_resource_from_index_entry(&index_object, index)
                 .await?
             {
+                let matched = match &state {
+                    TableCatalogEntryState::Active => &mut active,
+                    TableCatalogEntryState::Deleted => {
+                        if let Some(current) = deleted.as_ref() {
+                            deleted_overlap =
+                                Some((current.warehouse_object_prefix.clone(), resource.warehouse_object_prefix.clone()));
+                            continue;
+                        }
+                        &mut deleted
+                    }
+                    TableCatalogEntryState::Renaming | TableCatalogEntryState::Deleting => {
+                        unreachable!("transient indexes are rejected above")
+                    }
+                };
                 if let Some(current) = matched.as_ref() {
                     return Err(TableCatalogStoreError::Invalid(format!(
                         "object {object} matches overlapping active table warehouse indexes {} and {}",
                         current.warehouse_object_prefix, resource.warehouse_object_prefix
                     )));
                 }
-                matched = Some(resource);
+                *matched = Some(resource);
             }
         }
+        if let Some(resource) = active {
+            Ok(TableWarehouseIndexResolution::Active(resource))
+        } else if let Some((left, right)) = deleted_overlap {
+            Err(TableCatalogStoreError::Invalid(format!(
+                "object {object} matches overlapping deleted table warehouse indexes {left} and {right}"
+            )))
+        } else if let Some(resource) = deleted {
+            Ok(TableWarehouseIndexResolution::Deleted(resource))
+        } else {
+            Ok(TableWarehouseIndexResolution::Missing)
+        }
+    }
+
+    async fn resolve_table_data_plane_resource_with_scan(
+        &self,
+        table_bucket: &str,
+        object: &str,
+    ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
+        match self
+            .resolve_table_data_plane_resource_from_index(table_bucket, object)
+            .await?
+        {
+            TableWarehouseIndexResolution::Active(resource) => Ok(Some(resource)),
+            TableWarehouseIndexResolution::Deleted(tombstone) => {
+                Ok(scan_table_data_plane_resource_for_object(self, table_bucket, object)
+                    .await?
+                    .or(Some(tombstone)))
+            }
+            TableWarehouseIndexResolution::Missing => scan_table_data_plane_resource_for_object(self, table_bucket, object).await,
+        }
+    }
+
+    pub(in crate::table_catalog) async fn resolve_deleted_table_data_plane_resource_from_index(
+        &self,
+        table_bucket: &str,
+        object: &str,
+    ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
+        let mut matched: Option<TableDataPlaneResource> = None;
+        for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) {
+            let index_object = self.paths.warehouse_index_entry_path(table_bucket, warehouse_object_prefix);
+            let Some((index, _)) = self
+                .read_entry::<TableWarehouseIndexEntry>(self.catalog_bucket(), &index_object)
+                .await?
+            else {
+                continue;
+            };
+            validate_table_warehouse_index_entry_object(&self.paths, &index_object, &index)?;
+            if index.table_bucket != table_bucket || index.warehouse_object_prefix != warehouse_object_prefix {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "warehouse index entry does not match indexed prefix: {index_object}"
+                )));
+            }
+            match &index.state {
+                TableCatalogEntryState::Active => {
+                    return Err(TableCatalogStoreError::Internal(format!(
+                        "active warehouse index {index_object} is absent from the authoritative strong catalog"
+                    )));
+                }
+                TableCatalogEntryState::Deleted => {}
+                TableCatalogEntryState::Renaming | TableCatalogEntryState::Deleting => {
+                    return Err(TableCatalogStoreError::Internal(format!(
+                        "warehouse index {index_object} has an incomplete transient state"
+                    )));
+                }
+            }
+            let resource = table_data_plane_resource_from_warehouse_index(&index)?;
+            if let Some(current) = matched.as_ref() {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "object {object} matches overlapping deleted table warehouse indexes {} and {}",
+                    current.warehouse_object_prefix, resource.warehouse_object_prefix
+                )));
+            }
+            matched = Some(resource);
+        }
         Ok(matched)
     }
 
@@ -1872,7 +2020,7 @@ where
         if !publication.holds_table_bucket(&entry.table_bucket)
             || !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table)
         {
-            self.delete_created_table_warehouse_index(&entry, reservation, "table publication fence lost")
+            self.rollback_table_warehouse_index_reservation(&entry, reservation, "table publication fence lost")
                 .await;
             return Err(TableCatalogStoreError::Internal(
                 "table registration publication fence was lost before catalog update".to_string(),
@@ -1882,7 +2030,7 @@ where
             .write_entry_unlocked(self.catalog_bucket(), &table_path, &entry, precondition)
             .await;
         if result.is_err() {
-            self.delete_created_table_warehouse_index(&entry, reservation, "table entry write failed")
+            self.rollback_table_warehouse_index_reservation(&entry, reservation, "table entry write failed")
                 .await;
         }
         result
@@ -5119,29 +5267,17 @@ where
         let read_version = self.table_data_plane_read_version(table_bucket).await?;
 
         let resource = if self.warehouse_index_ready(table_bucket).await? {
-            match self
-                .resolve_table_data_plane_resource_from_index(table_bucket, object)
-                .await?
-            {
-                Some(resource) => Ok(Some(resource)),
-                None => scan_table_data_plane_resource_for_object(self, table_bucket, object).await,
-            }
+            self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await
         } else {
             match self.backfill_table_warehouse_index(table_bucket).await {
-                Ok(()) => match self
-                    .resolve_table_data_plane_resource_from_index(table_bucket, object)
-                    .await?
-                {
-                    Some(resource) => Ok(Some(resource)),
-                    None => scan_table_data_plane_resource_for_object(self, table_bucket, object).await,
-                },
+                Ok(()) => self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await,
                 Err(err @ TableCatalogStoreError::Internal(_)) => {
                     tracing::warn!(
                         table_bucket = %table_bucket,
                         error = %err,
                         "failed to backfill table warehouse index; falling back to catalog scan"
                     );
-                    scan_table_data_plane_resource_for_object(self, table_bucket, object).await
+                    self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await
                 }
                 Err(err) => Err(err),
             }
@@ -5524,7 +5660,7 @@ where
         }
         .await;
         if let Err(err) = staged_write_result {
-            self.delete_created_table_warehouse_index(&next, reservation, "commit staging failed")
+            self.rollback_table_warehouse_index_reservation(&next, reservation, "commit staging failed")
                 .await;
             return table_commit_result(
                 &request.table_bucket,
@@ -5540,7 +5676,7 @@ where
         if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table)
             || (warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket))
         {
-            self.delete_created_table_warehouse_index(&next, reservation, "table publication fence lost")
+            self.rollback_table_warehouse_index_reservation(&next, reservation, "table publication fence lost")
                 .await;
             return table_commit_result(
                 &request.table_bucket,
@@ -5566,7 +5702,7 @@ where
             .await;
         record_table_commit_cas_result(&request.operation, cas_started, &cas_result);
         if let Err(err) = cas_result {
-            self.delete_created_table_warehouse_index(&next, reservation, "table pointer CAS failed")
+            self.rollback_table_warehouse_index_reservation(&next, reservation, "table pointer CAS failed")
                 .await;
             return table_commit_result(
                 &request.table_bucket,
@@ -5632,7 +5768,7 @@ where
                 table.as_str()
             )));
         };
-        self.delete_owned_table_warehouse_index_for_drop(&entry).await?;
+        self.tombstone_table_warehouse_index_for_drop(&entry, false).await?;
         if !publication.holds_table_bucket(table_bucket)
             || !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str())
         {

rustfs/src/table_catalog/store/strong.rs

@@ -2416,26 +2416,32 @@ where
         }
 
         self.hydrate_state().await?;
-        let state = self.state.lock().await;
-        self.require_data_plane_ready_locked(&state, table_bucket)?;
+        let active_resource = {
+            let state = self.state.lock().await;
+            self.require_data_plane_ready_locked(&state, table_bucket)?;
 
-        let Some(bucket_index) = state.warehouse_index.get(table_bucket) else {
-            return Ok(None);
-        };
-
-        for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) {
-            if let Some(table_key) = bucket_index.get(warehouse_object_prefix) {
-                Self::ensure_identifier_is_unambiguous_locked(&state, table_key)?;
-                let Some(table) = state.tables.get(table_key) else {
-                    continue;
-                };
-                return Ok(Some(table_data_plane_resource_from_entry(
-                    table.clone(),
-                    warehouse_object_prefix.to_string(),
-                )));
+            let mut resource = None;
+            if let Some(bucket_index) = state.warehouse_index.get(table_bucket) {
+                for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) {
+                    if let Some(table_key) = bucket_index.get(warehouse_object_prefix) {
+                        Self::ensure_identifier_is_unambiguous_locked(&state, table_key)?;
+                        let Some(table) = state.tables.get(table_key) else {
+                            continue;
+                        };
+                        resource = Some(table_data_plane_resource_from_entry(table.clone(), warehouse_object_prefix.to_string()));
+                        break;
+                    }
+                }
             }
+            resource
+        };
+        if active_resource.is_some() {
+            return Ok(active_resource);
         }
-        Ok(None)
+
+        ObjectTableCatalogStore::new(self.object_backend.clone())
+            .resolve_deleted_table_data_plane_resource_from_index(table_bucket, object)
+            .await
     }
 
     async fn resolve_table_metadata_data_plane_resource(
@@ -2674,11 +2680,22 @@ where
         let namespace = parse_namespace_for_store(namespace)?;
         let table = parse_table_for_store(table)?;
         let key = Self::table_key(table_bucket, &namespace, &table);
+        let dropped = {
+            let state = self.state.lock().await;
+            state.tables.get(&key).cloned().ok_or_else(|| {
+                TableCatalogStoreError::NotFound(format!("table {}/{}/{}", table_bucket, namespace.public_name(), table.as_str()))
+            })?
+        };
+        if dropped.state == TableCatalogEntryState::Active {
+            ObjectTableCatalogStore::new(self.object_backend.clone())
+                .tombstone_table_warehouse_index_for_drop(&dropped, true)
+                .await?;
+        }
         let (snapshot, precondition, postcondition) = {
             let state = self.state.lock().await;
-            if !state.tables.contains_key(&key) {
-                return Err(TableCatalogStoreError::NotFound(format!(
-                    "table {}/{}/{}",
+            if state.tables.get(&key) != Some(&dropped) {
+                return Err(TableCatalogStoreError::Conflict(format!(
+                    "table changed while preparing drop: {}/{}/{}",
                     table_bucket,
                     namespace.public_name(),
                     table.as_str()

rustfs/src/table_catalog/tests.rs

@@ -4534,7 +4534,7 @@ async fn table_data_plane_resource_scans_when_an_index_disappears_during_backfil
 }
 
 #[tokio::test]
-async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() {
+async fn active_catalog_owner_takes_precedence_over_a_warehouse_tombstone() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
     let bucket = "analytics";
@@ -4544,7 +4544,7 @@ async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() {
     let prefix = "tables/table-id/";
 
     seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
-    let inactive = TableWarehouseIndexEntry {
+    let tombstone = TableWarehouseIndexEntry {
         version: TABLE_CATALOG_ENTRY_VERSION,
         table_bucket: bucket.to_string(),
         namespace: namespace.public_name(),
@@ -4557,16 +4557,58 @@ async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() {
         .write_entry(
             store.catalog_bucket(),
             &store.paths.warehouse_index_entry_path(bucket, prefix),
-            &inactive,
+            &tombstone,
             TableCatalogPutPrecondition::Any,
         )
         .await
-        .expect("inactive warehouse index should be seeded");
+        .expect("warehouse tombstone should be seeded");
 
-    assert_matches!(
-        table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet").await,
-        Err(TableCatalogStoreError::Internal(message)) if message.contains("inactive while the index is authoritative")
-    );
+    let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
+        .await
+        .expect("active catalog owner should resolve ahead of the tombstone")
+        .expect("active catalog owner should retain table-aware protection");
+
+    assert_eq!(resource.table_id, "table-id");
+    assert_eq!(resource.warehouse_object_prefix, prefix);
+}
+
+#[tokio::test]
+async fn table_data_plane_resource_fails_closed_for_a_transient_ready_index() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend);
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let table = IdentifierSegment::parse("orders").expect("table should parse");
+    let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    let prefix = "tables/table-id/";
+
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
+    for state in [TableCatalogEntryState::Renaming, TableCatalogEntryState::Deleting] {
+        let index = TableWarehouseIndexEntry {
+            version: TABLE_CATALOG_ENTRY_VERSION,
+            table_bucket: bucket.to_string(),
+            namespace: namespace.public_name(),
+            table: table.as_str().to_string(),
+            table_id: "table-id".to_string(),
+            warehouse_object_prefix: prefix.to_string(),
+            state,
+        };
+        store
+            .write_entry(
+                store.catalog_bucket(),
+                &store.paths.warehouse_index_entry_path(bucket, prefix),
+                &index,
+                TableCatalogPutPrecondition::Any,
+            )
+            .await
+            .expect("transient warehouse index should be seeded");
+
+        assert_matches!(
+            table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet").await,
+            Err(TableCatalogStoreError::Internal(message))
+                if message.contains("invalid transient state while the index is authoritative")
+        );
+    }
 }
 
 #[tokio::test]
@@ -5391,19 +5433,130 @@ async fn object_table_catalog_store_rolls_back_warehouse_index_when_table_entry_
 }
 
 #[tokio::test]
-async fn object_table_catalog_store_keeps_table_when_drop_index_delete_fails() {
+async fn object_catalog_prefix_reuse_preserves_protection_on_write_failure() {
+    for failure in ["index", "table", "rollback", "ambiguous"] {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let namespace = Namespace::parse("sales").unwrap();
+        let first = IdentifierSegment::parse("orders").unwrap();
+        let second = IdentifierSegment::parse("returns").unwrap();
+        let current = default_table_metadata_file_path(&namespace, &first, "00001.metadata.json");
+        seed_table_for_metadata_maintenance(&store, bucket, &namespace, &first, current).await;
+        store
+            .drop_table(bucket, &namespace.public_name(), first.as_str())
+            .await
+            .unwrap();
+
+        let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+        let table_path = store.paths.table_entry_path(bucket, &namespace, &second);
+        let index_attempts = backend.put_attempt_count(RUSTFS_META_BUCKET, &index_path).await;
+        if failure == "index" {
+            backend
+                .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, index_attempts + 2)
+                .await;
+        } else if failure == "ambiguous" {
+            backend.fail_after_next_put(RUSTFS_META_BUCKET, &table_path).await;
+        } else {
+            backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await;
+            if failure == "rollback" {
+                backend
+                    .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, index_attempts + 3)
+                    .await;
+            }
+        }
+
+        let metadata = default_table_metadata_file_path(&namespace, &second, "00001.metadata.json");
+        let mut entry = test_table_entry(bucket, &namespace, &second, metadata);
+        entry.table_id = "replacement-table-id".to_string();
+        entry.warehouse_location = format!("s3://{bucket}/tables/table-id");
+        assert_matches!(store.create_table(entry.clone()).await, Err(TableCatalogStoreError::Internal(_)));
+        assert_eq!(
+            store
+                .load_table(bucket, &namespace.public_name(), second.as_str())
+                .await
+                .unwrap()
+                .is_some(),
+            failure == "ambiguous"
+        );
+
+        let restarted = ObjectTableCatalogStore::new(backend.clone());
+        let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet").await;
+        if failure == "rollback" {
+            assert_matches!(resource, Err(TableCatalogStoreError::Internal(_)));
+        } else {
+            let resource = resource
+                .expect("lookup should succeed")
+                .expect("failed prefix reuse must retain protection");
+            assert_eq!(
+                resource.table_id,
+                if failure == "ambiguous" {
+                    "replacement-table-id"
+                } else {
+                    "table-id"
+                },
+                "failure at {failure}"
+            );
+        }
+        let (index, _) = restarted
+            .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+            .await
+            .unwrap()
+            .expect("failed prefix reuse must not remove the warehouse index");
+        assert_eq!(
+            index.state,
+            if matches!(failure, "rollback" | "ambiguous") {
+                TableCatalogEntryState::Active
+            } else {
+                TableCatalogEntryState::Deleted
+            }
+        );
+
+        if failure == "rollback" {
+            let third = IdentifierSegment::parse("retry_returns").unwrap();
+            let mut retry = entry.clone();
+            retry.table = third.as_str().to_string();
+            retry.table_id = "retry-table-id".to_string();
+            retry.metadata_location = default_table_metadata_file_path(&namespace, &third, "00001.metadata.json");
+            backend
+                .fail_next_put(RUSTFS_META_BUCKET, &store.paths.table_entry_path(bucket, &namespace, &third))
+                .await;
+            assert_matches!(restarted.create_table(retry).await, Err(TableCatalogStoreError::Internal(_)));
+            assert_matches!(
+                table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet").await,
+                Err(TableCatalogStoreError::Internal(_))
+            );
+        }
+        if failure != "ambiguous" {
+            restarted
+                .create_table(entry)
+                .await
+                .expect("prefix reuse should remain retryable");
+        }
+        restarted
+            .drop_table(bucket, &namespace.public_name(), second.as_str())
+            .await
+            .expect("replacement table should remain droppable");
+        let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet")
+            .await
+            .unwrap()
+            .expect("replacement tombstone should protect the reused prefix");
+        assert_eq!(resource.table_id, "replacement-table-id");
+    }
+}
+
+#[tokio::test]
+async fn object_table_catalog_store_keeps_table_when_drop_index_tombstone_write_fails() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
     let bucket = "analytics";
     let namespace = Namespace::parse("sales").expect("namespace should parse");
     let table = IdentifierSegment::parse("orders").expect("table should parse");
     let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
-    let table_path = store.paths.table_entry_path(bucket, &namespace, &table);
     let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
 
     seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
-    backend.fail_delete_attempt(RUSTFS_META_BUCKET, &index_path, 1).await;
-    backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await;
+    backend.fail_next_put(RUSTFS_META_BUCKET, &index_path).await;
 
     let error = store
         .drop_table(bucket, &namespace.public_name(), table.as_str())
@@ -5467,6 +5620,50 @@ async fn object_table_catalog_store_rejects_drop_when_warehouse_index_owner_chan
     assert_eq!(retained_index, conflicting_index);
 }
 
+#[tokio::test]
+async fn object_table_catalog_store_rejects_drop_when_deleted_warehouse_index_owner_changed() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend);
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let table = IdentifierSegment::parse("orders").expect("table should parse");
+    let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
+    let conflicting_index = TableWarehouseIndexEntry {
+        version: TABLE_CATALOG_ENTRY_VERSION,
+        table_bucket: bucket.to_string(),
+        namespace: "finance".to_string(),
+        table: "returns".to_string(),
+        table_id: "other-table-id".to_string(),
+        warehouse_object_prefix: "tables/table-id/".to_string(),
+        state: TableCatalogEntryState::Deleted,
+    };
+    store
+        .write_entry(store.catalog_bucket(), &index_path, &conflicting_index, TableCatalogPutPrecondition::Any)
+        .await
+        .expect("conflicting warehouse tombstone should be seeded");
+
+    assert_matches!(
+        store.drop_table(bucket, &namespace.public_name(), table.as_str()).await,
+        Err(TableCatalogStoreError::Conflict(message)) if message.contains("owner changed")
+    );
+    assert!(
+        store
+            .load_table(bucket, &namespace.public_name(), table.as_str())
+            .await
+            .expect("retained table lookup should succeed")
+            .is_some()
+    );
+    let (retained_index, _) = store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("conflicting warehouse tombstone lookup should succeed")
+        .expect("conflicting warehouse tombstone should remain present");
+    assert_eq!(retained_index, conflicting_index);
+}
+
 #[tokio::test]
 async fn object_table_catalog_store_drops_table_when_warehouse_index_is_missing() {
     let backend = TestCatalogObjectBackend::default();
@@ -5506,6 +5703,76 @@ async fn object_table_catalog_store_drops_table_when_warehouse_index_is_missing(
             .expect("dropped table lookup should succeed")
             .is_none()
     );
+    let (tombstone, _) = store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("warehouse tombstone lookup should succeed")
+        .expect("warehouse tombstone should be created");
+    assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
+    let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
+        .await
+        .expect("dropped table data-plane lookup should succeed")
+        .expect("dropped warehouse prefix should remain protected");
+    assert_eq!(resource.table_id, "table-id");
+}
+
+#[tokio::test]
+async fn active_table_warehouse_prefix_takes_precedence_over_a_deleted_parent() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend);
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let parent = IdentifierSegment::parse("orders").expect("table should parse");
+    let child = IdentifierSegment::parse("returns").expect("table should parse");
+    let parent_metadata = default_table_metadata_file_path(&namespace, &parent, "00001.metadata.json");
+
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &parent, parent_metadata).await;
+    store
+        .drop_table(bucket, &namespace.public_name(), parent.as_str())
+        .await
+        .expect("parent table should be dropped");
+
+    let child_metadata = default_table_metadata_file_path(&namespace, &child, "00001.metadata.json");
+    let mut child_entry = test_table_entry(bucket, &namespace, &child, child_metadata);
+    child_entry.table_id = "child-table-id".to_string();
+    child_entry.warehouse_location = format!("s3://{bucket}/tables/table-id/child");
+    store.create_table(child_entry).await.expect("child table should be created");
+
+    let child_resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/data/file.parquet")
+        .await
+        .expect("child data-plane lookup should succeed")
+        .expect("active child table should protect its prefix");
+    assert_eq!(child_resource.table_id, "child-table-id");
+
+    let parent_resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/orphan/file.parquet")
+        .await
+        .expect("deleted parent data-plane lookup should succeed")
+        .expect("deleted parent should keep protecting its remaining prefix");
+    assert_eq!(parent_resource.table_id, "table-id");
+
+    store
+        .drop_table(bucket, &namespace.public_name(), child.as_str())
+        .await
+        .unwrap();
+    assert_matches!(
+        table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/data/file.parquet").await,
+        Err(TableCatalogStoreError::Invalid(message)) if message.contains("overlapping deleted table warehouse indexes")
+    );
+    let grandchild = IdentifierSegment::parse("nested_returns").unwrap();
+    let mut entry = test_table_entry(
+        bucket,
+        &namespace,
+        &grandchild,
+        default_table_metadata_file_path(&namespace, &grandchild, "00001.metadata.json"),
+    );
+    entry.table_id = "grandchild-table-id".to_string();
+    entry.warehouse_location = format!("s3://{bucket}/tables/table-id/child/nested");
+    store.create_table(entry).await.unwrap();
+    let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/nested/data/file.parquet")
+        .await
+        .unwrap()
+        .expect("active owner should take precedence over both tombstones");
+    assert_eq!(resource.table_id, "grandchild-table-id");
 }
 
 #[tokio::test]
@@ -5538,6 +5805,7 @@ async fn object_table_catalog_store_restores_index_when_table_entry_delete_fails
         .expect("restored warehouse index lookup should succeed")
         .expect("warehouse index should be restored");
     assert_eq!(restored_index.table_id, "table-id");
+    assert_eq!(restored_index.state, TableCatalogEntryState::Active);
     let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
         .await
         .expect("restored index lookup should succeed")
@@ -5546,7 +5814,7 @@ async fn object_table_catalog_store_restores_index_when_table_entry_delete_fails
 }
 
 #[tokio::test]
-async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_fails() {
+async fn object_table_catalog_store_uses_tombstone_when_drop_index_restore_fails() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
     let bucket = "analytics";
@@ -5558,7 +5826,10 @@ async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_f
 
     seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
     backend.fail_delete_attempt(RUSTFS_META_BUCKET, &table_path, 1).await;
-    backend.fail_next_put(RUSTFS_META_BUCKET, &index_path).await;
+    let restore_attempt = backend.put_attempt_count(RUSTFS_META_BUCKET, &index_path).await + 2;
+    backend
+        .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, restore_attempt)
+        .await;
 
     assert_matches!(
         store.drop_table(bucket, &namespace.public_name(), table.as_str()).await,
@@ -5570,18 +5841,16 @@ async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_f
         .expect("retained table lookup should succeed")
         .expect("table entry should remain present");
     assert_eq!(retained.table_id, "table-id");
-    assert!(
-        store
-            .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
-            .await
-            .expect("warehouse index lookup should succeed")
-            .is_none(),
-        "the injected index restore failure must leave the scan fallback under test"
-    );
+    let (tombstone, _) = store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("warehouse index lookup should succeed")
+        .expect("the failed restore should retain the warehouse tombstone");
+    assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
     let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
         .await
-        .expect("catalog scan fallback should succeed")
-        .expect("catalog scan fallback should keep data-plane protection");
+        .expect("tombstone fallback should succeed")
+        .expect("tombstone fallback should keep data-plane protection");
     assert_eq!(resource.table, "orders");
 }
 
@@ -5609,12 +5878,11 @@ async fn object_table_catalog_store_accepts_ambiguous_delete_when_table_is_absen
             .expect("dropped table lookup should succeed")
             .is_none()
     );
-    assert!(
-        table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
-            .await
-            .expect("dropped table data-plane lookup should succeed")
-            .is_none()
-    );
+    let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet")
+        .await
+        .expect("dropped table data-plane lookup should succeed")
+        .expect("dropped warehouse prefix should remain protected");
+    assert_eq!(resource.table_id, "table-id");
 }
 
 #[tokio::test]
@@ -12609,6 +12877,196 @@ async fn strong_catalog_table_metadata_data_plane_resource_resolves_only_current
     );
 }
 
+#[tokio::test]
+async fn strong_catalog_drop_keeps_warehouse_tombstone_after_restart() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = StrongTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let table = IdentifierSegment::parse("orders").expect("table should parse");
+    let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+
+    store
+        .put_table_bucket(test_bucket_entry(bucket))
+        .await
+        .expect("table bucket should be created");
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .expect("namespace should be created");
+    store
+        .create_table(test_table_entry(bucket, &namespace, &table, current))
+        .await
+        .expect("table should be created");
+    store
+        .drop_table(bucket, &namespace.public_name(), table.as_str())
+        .await
+        .expect("table should be dropped");
+
+    let restarted = StrongTableCatalogStore::new(backend.clone());
+    assert!(
+        restarted
+            .load_table(bucket, &namespace.public_name(), table.as_str())
+            .await
+            .expect("dropped table lookup should succeed")
+            .is_none()
+    );
+    let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet")
+        .await
+        .expect("dropped table data-plane lookup should succeed")
+        .expect("dropped warehouse prefix should remain protected");
+    assert_eq!(resource.table_id, "table-id");
+
+    let object_store = ObjectTableCatalogStore::new(backend);
+    let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+    let (tombstone, _) = object_store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("warehouse tombstone lookup should succeed")
+        .expect("warehouse tombstone should remain durable");
+    assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
+}
+
+#[tokio::test]
+async fn strong_catalog_failed_drop_keeps_active_table_ahead_of_tombstone() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = StrongTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let table = IdentifierSegment::parse("orders").expect("table should parse");
+    let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+
+    store
+        .put_table_bucket(test_bucket_entry(bucket))
+        .await
+        .expect("table bucket should be created");
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .expect("namespace should be created");
+    store
+        .create_table(test_table_entry(bucket, &namespace, &table, current))
+        .await
+        .expect("table should be created");
+    let snapshot_path = StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path();
+    backend.fail_next_put(RUSTFS_META_BUCKET, &snapshot_path).await;
+
+    assert_matches!(
+        store.drop_table(bucket, &namespace.public_name(), table.as_str()).await,
+        Err(TableCatalogStoreError::Internal(_))
+    );
+    assert!(
+        store
+            .load_table(bucket, &namespace.public_name(), table.as_str())
+            .await
+            .expect("retained table lookup should succeed")
+            .is_some()
+    );
+    let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/file.parquet")
+        .await
+        .expect("retained table data-plane lookup should succeed")
+        .expect("active table should remain protected after the failed drop");
+    assert_eq!(resource.table_id, "table-id");
+
+    let object_store = ObjectTableCatalogStore::new(backend);
+    let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+    let (tombstone, _) = object_store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("warehouse tombstone lookup should succeed")
+        .expect("failed snapshot write should retain the warehouse tombstone");
+    assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
+}
+
+#[tokio::test]
+async fn strong_catalog_reused_warehouse_prefix_replaces_the_old_tombstone_on_drop() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = StrongTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").expect("namespace should parse");
+    let first = IdentifierSegment::parse("orders").expect("table should parse");
+    let second = IdentifierSegment::parse("returns").expect("table should parse");
+    let first_metadata = default_table_metadata_file_path(&namespace, &first, "00001.metadata.json");
+
+    store
+        .put_table_bucket(test_bucket_entry(bucket))
+        .await
+        .expect("table bucket should be created");
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .expect("namespace should be created");
+    store
+        .create_table(test_table_entry(bucket, &namespace, &first, first_metadata))
+        .await
+        .expect("first table should be created");
+    store
+        .drop_table(bucket, &namespace.public_name(), first.as_str())
+        .await
+        .expect("first table should be dropped");
+
+    let second_metadata = default_table_metadata_file_path(&namespace, &second, "00001.metadata.json");
+    let mut second_entry = test_table_entry(bucket, &namespace, &second, second_metadata);
+    second_entry.table_id = "second-table-id".to_string();
+    second_entry.warehouse_location = format!("s3://{bucket}/tables/table-id");
+    store
+        .create_table(second_entry)
+        .await
+        .expect("second table should reuse the prefix");
+    store
+        .drop_table(bucket, &namespace.public_name(), second.as_str())
+        .await
+        .expect("second table should replace the old tombstone when dropped");
+
+    let object_store = ObjectTableCatalogStore::new(backend);
+    let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+    let (tombstone, _) = object_store
+        .read_entry::<TableWarehouseIndexEntry>(RUSTFS_META_BUCKET, &index_path)
+        .await
+        .expect("warehouse tombstone lookup should succeed")
+        .expect("warehouse tombstone should remain durable");
+    assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
+    assert_eq!(tombstone.table_id, "second-table-id");
+    assert_eq!(tombstone.table, "returns");
+}
+
+#[tokio::test]
+async fn strong_catalog_fails_closed_for_an_unowned_active_external_warehouse_index() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = StrongTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    store
+        .put_table_bucket(test_bucket_entry(bucket))
+        .await
+        .expect("table bucket should be created");
+
+    let object_store = ObjectTableCatalogStore::new(backend);
+    let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
+    let unowned = TableWarehouseIndexEntry {
+        version: TABLE_CATALOG_ENTRY_VERSION,
+        table_bucket: bucket.to_string(),
+        namespace: "sales".to_string(),
+        table: "orders".to_string(),
+        table_id: "table-id".to_string(),
+        warehouse_object_prefix: "tables/table-id/".to_string(),
+        state: TableCatalogEntryState::Active,
+    };
+    object_store
+        .write_entry(
+            object_store.catalog_bucket(),
+            &index_path,
+            &unowned,
+            TableCatalogPutPrecondition::IfAbsent,
+        )
+        .await
+        .expect("unowned active warehouse index should be seeded");
+
+    assert_matches!(
+        table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/file.parquet").await,
+        Err(TableCatalogStoreError::Internal(message)) if message.contains("authoritative strong catalog")
+    );
+}
+
 #[tokio::test]
 async fn strong_catalog_data_plane_fails_closed_for_missing_bucket_snapshot() {
     let store = StrongTableCatalogStore::new(TestCatalogObjectBackend::default());