Merge branch 'fix/ant-request-layer-batch' into fix/ant-table-catalog-batch

Scan findings in the Iceberg table catalog:

- Metadata maintenance deleted every unreferenced *.metadata.json in a
  table's directory without checking whether a renamed table in another
  namespace still owned those files (backlog#3060); it now resolves the
  owning table through the directory index before deleting.
- Manifest lists and manifests were decoded with the embedded Avro
  writer schema and no recursion depth limit; a self-referencing record
  schema overflowed the stack (backlog#3080). Schemas are validated for
  depth before decoding.
- Compaction decoded every input Parquet file into Arrow batches held in
  memory; a decompression bomb OOM-killed the server (backlog#3085). Row
  groups are streamed through a byte budget and oversized inputs are
  refused.
- An S3 request for a key outside an indexed table prefix listed the
  bucket's whole namespaces/ catalog prefix and failed once it held more
  than 4096 objects (backlog#3086); a per-bucket directory index answers
  index misses without the listing.
- The maintenance planner read whole orphan files to obtain their
  modification time (backlog#3087); it reads metadata only.
- A client disconnect between persisting active_rename_id and clearing it
  left the bucket's catalog entry locked at 503 (backlog#3088); the rename
  runs detached from the request future and clears the flag on every exit.
- Table commits held the bucket publication fence until publication,
  starving every object write (backlog#3111); commits now hold it under a
  deadline and object writes are admitted between commits.

The directory index is managed through three new admin routes under
{warehouse}/catalog/directory, registered in the admin route policy.

The durable-strong store is left as is (its snapshot partitioning for
backlog#3109 and the strong-path deadline are not ported): that backing
is being retired separately, so the new tests cover the object backing.

docs/architecture/s3-tables-support-matrix.md

@@ -112,6 +112,8 @@ metadata or warehouse directory, including a recreated source name after rename.
 Such tables require disjoint cleanup roots before maintenance can run; enabling
 deletion does not override this ownership check.
 
+Create, register, and commit publication share one preparation deadline derived from the existing lock-acquisition timeout. Migration and bucket-lock queueing, reference HEAD validation, ownership inventories, metadata parsing, and snapshot preparation consume that same interval; repeated preparation does not renew it. Expiry before a possible durable catalog write returns HTTP 503 and releases the preparation locks. A valid large operation that cannot finish within the interval may continue to fail on retry; smaller operations or less contention are required. Once a catalog write may have begun, this deadline no longer cancels that transaction. A preceding rename recovery completes under its existing protocol, then the new operation checks the original deadline. Individual PUT completion and persisted commit/rename recovery retain their existing cancellation guarantees; the entire multi-write catalog operation is not newly detached from client cancellation. This is a preparation limit, not a hard bound on disk I/O or the durable tail: snapshot write/reload and rename maintenance copy verification/cleanup can still scale with stored state. Rename maintenance inventory isolation remains a separate limitation.
+
 Durable-strong rename preserves maintenance configuration, job identifiers,
 current/latest aliases, history, audit events, and quarantine state. A persisted
 rename intent protects sidecar migration across interruption: the source remains

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

@@ -293,3 +293,46 @@ impl Operation for SyncExternalCatalogBridgeHandler {
         build_json_response(StatusCode::OK, &response)
     }
 }
+
+pub struct GetTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for GetTableDirectoryHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        let resource = TableCatalogResource::warehouse(&warehouse);
+        authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCatalogAction).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+        let response = store.table_directory_status(&warehouse).await.map_err(catalog_store_error)?;
+        build_json_response(StatusCode::OK, &response)
+    }
+}
+
+pub struct AdvanceTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for AdvanceTableDirectoryHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        authorize_table_catalog_request(&req, AdminAction::MigrateTableCatalogAction).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+        let response = store.advance_table_directory(&warehouse).await.map_err(catalog_store_error)?;
+        build_json_response(StatusCode::OK, &response)
+    }
+}
+
+pub struct DisableTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for DisableTableDirectoryHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        authorize_table_catalog_request(&req, AdminAction::MigrateTableCatalogAction).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+        let response = store.disable_table_directory(&warehouse).await.map_err(catalog_store_error)?;
+        build_json_response(StatusCode::OK, &response)
+    }
+}

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

@@ -1303,6 +1303,13 @@ async fn authorize_table_catalog_s3_actions(
 
     let mut result = Ok(());
     for action in actions {
+        #[cfg(test)]
+        if *action == S3Action::PutObjectAction
+            && let Ok(Some(pause)) =
+                crate::table_catalog::test_support::CATALOG_PUT_AUTHORIZATION_PAUSE.try_with(|slot| slot.get().cloned())
+        {
+            pause.pause().await;
+        }
         if let Err(err) = authorize_internal_object_request(req, Action::S3Action(*action)).await {
             result = Err(err);
             break;
@@ -1364,8 +1371,14 @@ enum TableCommitObjectIdentity {
     ContentSha256(String),
 }
 
+// Legacy objects without a write-unique body identity retain SHA validation,
+// with a total byte budget independent of the manifest graph's larger limit.
+const TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES: usize = 2 * MAX_ADMIN_REQUEST_BODY_SIZE;
+
 #[derive(Clone, PartialEq, Eq)]
 struct TableCommitObservedObject {
+    body_revision: Option<Uuid>,
+    content_size: usize,
     identity: TableCommitObjectIdentity,
     max_size: Option<usize>,
 }
@@ -1377,6 +1390,7 @@ struct TableCommitPublicationState {
     table_fence: Option<(String, String, String)>,
     catalog_migration_read_permit: bool,
     observed_objects: BTreeMap<(String, String), TableCommitObservedObject>,
+    legacy_revalidation_bytes: usize,
     guards: Vec<crate::table_catalog::TableCatalogLockGuard>,
 }
 
@@ -1387,6 +1401,7 @@ struct TableCommitObjectBackend<B> {
     authorization_error: Arc<tokio::sync::Mutex<Option<S3Error>>>,
     publication: Arc<parking_lot::Mutex<TableCommitPublicationState>>,
     publication_fence_fleet_confirmed: bool,
+    preparation: Arc<crate::table_catalog::TablePublicationPreparation>,
 }
 
 impl<B> TableCommitObjectBackend<B>
@@ -1400,6 +1415,7 @@ where
             authorization_error: Arc::new(tokio::sync::Mutex::new(None)),
             publication: Arc::new(parking_lot::Mutex::new(TableCommitPublicationState::default())),
             publication_fence_fleet_confirmed,
+            preparation: Arc::new(crate::table_catalog::TablePublicationPreparation::default()),
         }
     }
 
@@ -1445,6 +1461,45 @@ where
         )
     }
 
+    async fn prepare_put_observation(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &[u8],
+    ) -> crate::table_catalog::TableCatalogStoreResult<TableCommitObservedObject> {
+        let fence = {
+            let publication = self.publication.lock();
+            publication.bucket_fence.as_ref().map(|_| {
+                crate::table_catalog::TableCatalogObjectMutationFence::from_signals(
+                    publication
+                        .guards
+                        .iter()
+                        .map(crate::table_catalog::TableCatalogLockGuard::lock_lost_signal),
+                )
+            })
+        };
+        let authorize = self.authorize(bucket, object, S3Action::PutObjectAction);
+        if fence.is_some() {
+            self.preparation.read(self, authorize).await?;
+        } else {
+            authorize.await?;
+        }
+        self.ensure_observation_allowed(bucket, object)?;
+        let observation = TableCommitObservedObject {
+            identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(data, str::to_string)),
+            body_revision: None,
+            content_size: data.len(),
+            max_size: Some(data.len()),
+        };
+        if let Some(fence) = fence {
+            // An unfenced candidate write precedes the publication transaction. Only a PUT
+            // under the bucket fence commits this preparation to finishing its durable tail.
+            fence.ensure_held()?;
+            self.preparation.begin_mutation()?;
+        }
+        Ok(observation)
+    }
+
     async fn authorize(&self, bucket: &str, object: &str, action: S3Action) -> crate::table_catalog::TableCatalogStoreResult<()> {
         let result = match &self.authorization {
             TableCommitObjectAuthorization::Request(req) => {
@@ -1502,31 +1557,62 @@ where
         bucket: &str,
         object: &str,
         observation: TableCommitObservedObject,
+        enforce_legacy_budget: bool,
     ) -> crate::table_catalog::TableCatalogStoreResult<()> {
         let key = (bucket.to_string(), object.to_string());
         let mut publication = self.publication.lock();
         if publication.phase == TableCommitPublicationPhase::Complete {
             return Ok(());
         }
+        let previous_legacy_bytes = publication.observed_objects.get(&key).map_or(0, |observed| {
+            if observed.body_revision.is_none() {
+                observed.content_size
+            } else {
+                0
+            }
+        });
+        let legacy_bytes = publication.legacy_revalidation_bytes - previous_legacy_bytes;
+        let legacy_bytes = legacy_bytes.checked_add(if observation.body_revision.is_none() {
+            observation.content_size
+        } else {
+            0
+        });
+        let Some(legacy_bytes) =
+            legacy_bytes.filter(|bytes| !enforce_legacy_budget || *bytes <= TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES)
+        else {
+            return Err(crate::table_catalog::TableCatalogStoreError::Unavailable(
+                "legacy catalog objects exceed the publication validation budget; rewrite the referenced objects and retry"
+                    .to_string(),
+            ));
+        };
         let Some(existing) = publication.observed_objects.get_mut(&key) else {
             if publication.phase != TableCommitPublicationPhase::Discovering {
                 return Err(crate::table_catalog::TableCatalogStoreError::Internal(format!(
                     "table commit accessed an unvalidated object after publication preparation: {bucket}/{object}"
                 )));
             }
+            publication.legacy_revalidation_bytes = legacy_bytes;
             publication.observed_objects.insert(key, observation);
             return Ok(());
         };
-        if existing.identity != observation.identity {
+        if existing.identity != observation.identity
+            || existing
+                .body_revision
+                .zip(observation.body_revision)
+                .is_some_and(|(before, after)| before != after)
+        {
             return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
                 "table commit object changed during validation: {bucket}/{object}"
             )));
         }
+        existing.body_revision = observation.body_revision;
+        existing.content_size = observation.content_size;
         existing.max_size = match (existing.max_size, observation.max_size) {
             (Some(expected), Some(actual)) => Some(expected.min(actual)),
             (Some(expected), None) => Some(expected),
             (None, actual) => actual,
         };
+        publication.legacy_revalidation_bytes = legacy_bytes;
         Ok(())
     }
 
@@ -1538,7 +1624,12 @@ where
             None => TableCommitObjectIdentity::Missing,
             Some(object) => TableCommitObjectIdentity::ContentSha256(hex_sha256(&object.data, str::to_string)),
         };
-        TableCommitObservedObject { identity, max_size }
+        TableCommitObservedObject {
+            identity,
+            max_size,
+            body_revision: object.and_then(|object| object.body_revision).filter(|id| !id.is_nil()),
+            content_size: object.map_or(0, |object| object.data.len()),
+        }
     }
 
     fn observed_metadata(
@@ -1560,10 +1651,18 @@ where
         Ok(TableCommitObservedObject {
             identity,
             max_size: None,
+            body_revision: None,
+            content_size: 0,
         })
     }
 
     async fn begin_bucket_publication(&self, table_bucket: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
+        self.preparation
+            .read(self, self.begin_bucket_publication_inner(table_bucket))
+            .await
+    }
+
+    async fn begin_bucket_publication_inner(&self, table_bucket: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
         {
             let mut publication = self.publication.lock();
             if publication.phase != TableCommitPublicationPhase::Discovering {
@@ -1610,6 +1709,17 @@ where
         table_bucket: &str,
         namespace: &str,
         table: &str,
+    ) -> crate::table_catalog::TableCatalogStoreResult<()> {
+        self.preparation
+            .read(self, self.prepare_publication_inner(table_bucket, namespace, table))
+            .await
+    }
+
+    async fn prepare_publication_inner(
+        &self,
+        table_bucket: &str,
+        namespace: &str,
+        table: &str,
     ) -> crate::table_catalog::TableCatalogStoreResult<()> {
         let namespace = crate::table_catalog::Namespace::parse(namespace)
             .map_err(|err| crate::table_catalog::TableCatalogStoreError::Invalid(format!("invalid namespace: {err}")))?;
@@ -1619,6 +1729,12 @@ where
         let publication_lock = crate::table_catalog::default_table_publication_lock_path(&namespace, &table);
         let expected = {
             let mut publication = self.publication.lock();
+            if publication.legacy_revalidation_bytes > TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES {
+                return Err(crate::table_catalog::TableCatalogStoreError::Unavailable(
+                    "legacy catalog objects exceed the publication validation budget; rewrite the referenced objects and retry"
+                        .to_string(),
+                ));
+            }
             match publication.phase {
                 TableCommitPublicationPhase::Prepared if publication.table_fence.as_ref() == Some(&table_fence) => {
                     return Ok(());
@@ -1668,10 +1784,19 @@ where
                 if !matches!(&expected_object.identity, TableCommitObjectIdentity::ContentSha256(_)) {
                     continue;
                 }
-                let actual_object = match expected_object.max_size {
-                    Some(max_size) => self.backend.read_object_unlocked_limited(bucket, object, max_size).await?,
-                    None => self.backend.read_object_unlocked(bucket, object).await?,
-                };
+                if let Some(revision) = expected_object.body_revision {
+                    let actual = self.backend.object_metadata_unlocked(bucket, object).await?;
+                    if actual.and_then(|metadata| metadata.body_revision) != Some(revision) {
+                        return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
+                            "table commit object changed before catalog publication: {bucket}/{object}"
+                        )));
+                    }
+                    continue;
+                }
+                let max_size = expected_object
+                    .max_size
+                    .map_or(expected_object.content_size, |limit| limit.min(expected_object.content_size));
+                let actual_object = self.backend.read_object_unlocked_limited(bucket, object, max_size).await?;
                 let actual = Self::observed_content(actual_object.as_ref(), expected_object.max_size);
                 if actual.identity != expected_object.identity {
                     return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
@@ -1762,7 +1887,7 @@ where
         self.authorize(bucket, object, S3Action::GetObjectAction).await?;
         self.ensure_observation_allowed(bucket, object)?;
         let result = self.backend.read_object(bucket, object).await?;
-        self.record_observation(bucket, object, Self::observed_content(result.as_ref(), None))?;
+        self.record_observation(bucket, object, Self::observed_content(result.as_ref(), None), true)?;
         Ok(result)
     }
 
@@ -1775,7 +1900,7 @@ where
         self.authorize(bucket, object, S3Action::GetObjectAction).await?;
         self.ensure_observation_allowed(bucket, object)?;
         let result = self.backend.read_object_limited(bucket, object, max_size).await?;
-        self.record_observation(bucket, object, Self::observed_content(result.as_ref(), Some(max_size)))?;
+        self.record_observation(bucket, object, Self::observed_content(result.as_ref(), Some(max_size)), true)?;
         Ok(result)
     }
 
@@ -1802,7 +1927,7 @@ where
         let metadata = self.backend.object_metadata(bucket, object).await?;
         let observation = Self::observed_metadata(bucket, object, metadata)?;
         let exists = observation.identity != TableCommitObjectIdentity::Missing;
-        self.record_observation(bucket, object, observation)?;
+        self.record_observation(bucket, object, observation, true)?;
         Ok(exists)
     }
 
@@ -1817,14 +1942,12 @@ where
         data: Vec<u8>,
         precondition: crate::table_catalog::TableCatalogPutPrecondition,
     ) -> crate::table_catalog::TableCatalogStoreResult<()> {
-        self.authorize(bucket, object, S3Action::PutObjectAction).await?;
-        self.ensure_observation_allowed(bucket, object)?;
-        let observation = TableCommitObservedObject {
-            identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(&data, str::to_string)),
-            max_size: None,
-        };
+        let observation = self.prepare_put_observation(bucket, object, &data).await?;
         self.backend.put_object(bucket, object, data, precondition).await?;
-        self.record_observation(bucket, object, observation)
+        // The existing generated-metadata readback binds the written bytes to
+        // their revision. Until then, keep the original SHA expectation and
+        // enforce the fallback budget before publication preparation.
+        self.record_observation(bucket, object, observation, false)
     }
 
     async fn put_object_unlocked(
@@ -1835,16 +1958,11 @@ where
         precondition: crate::table_catalog::TableCatalogPutPrecondition,
         write_guards: Vec<crate::admin::storage_api::object::WriteCommitGuard>,
     ) -> crate::table_catalog::TableCatalogStoreResult<()> {
-        self.authorize(bucket, object, S3Action::PutObjectAction).await?;
-        self.ensure_observation_allowed(bucket, object)?;
-        let observation = TableCommitObservedObject {
-            identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(&data, str::to_string)),
-            max_size: None,
-        };
+        let observation = self.prepare_put_observation(bucket, object, &data).await?;
         self.backend
             .put_object_unlocked(bucket, object, data, precondition, write_guards)
             .await?;
-        self.record_observation(bucket, object, observation)
+        self.record_observation(bucket, object, observation, false)
     }
 
     async fn delete_object(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
@@ -1890,6 +2008,10 @@ where
         self.backend.acquire_write_lock(bucket, object).await
     }
 
+    fn table_publication_preparation_budget(&self) -> Arc<crate::table_catalog::TablePublicationPreparation> {
+        self.preparation.clone()
+    }
+
     async fn begin_table_bucket_commit_publication(
         &self,
         table_bucket: &str,
@@ -2522,6 +2644,29 @@ where
         .map_err(catalog_store_error)
 }
 
+/// Recheck the bucket registered before the publication lock. A missing entry
+/// here can indicate retirement while queued; do not recreate it under the fence.
+async fn require_table_bucket_for_publication<S>(
+    store: &S,
+    bucket: &str,
+    publication: &(dyn crate::table_catalog::TableCommitPublication + Sync),
+) -> S3Result<()>
+where
+    S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
+    let entry = publication
+        .preparation_budget()
+        .read(publication, store.get_table_bucket(bucket))
+        .await
+        .map_err(catalog_store_error)?;
+    if entry.is_some_and(|entry| entry.state == crate::table_catalog::TableCatalogEntryState::Active) {
+        return Ok(());
+    }
+    Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::NotFound(format!(
+        "table bucket {bucket}"
+    ))))
+}
+
 async fn table_bucket_response<S>(store: &S, bucket: &str, enabled: bool) -> S3Result<TableBucketResponse>
 where
     S: crate::table_catalog::TableCatalogStore + ?Sized,
@@ -6014,7 +6159,7 @@ where
         )));
     }
     let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
-    ensure_table_bucket_entry(store, bucket, true).await?;
+    require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
     let metadata_data = serde_json::to_vec(&metadata)
         .map_err(|err| s3_error!(InternalError, "failed to serialize initial table metadata: {}", err))?;
     metadata_backend
@@ -6055,7 +6200,7 @@ where
         )));
     }
     let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
-    ensure_table_bucket_entry(store, bucket, true).await?;
+    require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
     let metadata_data = serde_json::to_vec(&metadata)
         .map_err(|err| s3_error!(InternalError, "failed to serialize initial view metadata: {}", err))?;
     metadata_backend
@@ -6478,15 +6623,18 @@ async fn publish_table_commit<S>(
 where
     S: crate::table_catalog::TableCatalogStore + ?Sized,
 {
-    let _bucket_publication_completion = if table_bucket_fence_required {
-        metadata_backend
-            .begin_table_bucket_commit_publication(&request.table_bucket)
-            .await
-            .map_err(catalog_store_error)?;
-        Some(crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend))
-    } else {
-        None
-    };
+    // Real request backends acquire the migration permit here, before the store chooses its
+    // inline/child route. Nested publication reuses that same permit and preparation budget.
+    let _bucket_publication_completion =
+        if table_bucket_fence_required || metadata_backend.acquires_catalog_migration_read_permit() {
+            metadata_backend
+                .begin_table_bucket_commit_publication(&request.table_bucket)
+                .await
+                .map_err(catalog_store_error)?;
+            Some(crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend))
+        } else {
+            None
+        };
     store
         .commit_table_with_publication(request, metadata_backend)
         .await
@@ -6843,6 +6991,7 @@ where
         .map_err(catalog_store_error)?;
     validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?;
 
+    ensure_table_bucket_entry(store, bucket, true).await?;
     crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket)
         .await
         .map_err(catalog_store_error)?;
@@ -6852,7 +7001,7 @@ where
         )));
     }
     let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
-    ensure_table_bucket_entry(store, bucket, true).await?;
+    require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
     let metadata_data = serde_json::to_vec(&metadata).map_err(|err| {
         S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize initial table metadata: {err}"))
     })?;

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

@@ -39,6 +39,21 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
         format!("{prefix}/buckets/{{warehouse}}").as_str(),
         AdminOperation(&GET_TABLE_BUCKET_HANDLER),
     )?;
+    r.insert(
+        Method::GET,
+        format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+        AdminOperation(&GetTableDirectoryHandler {}),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+        AdminOperation(&AdvanceTableDirectoryHandler {}),
+    )?;
+    r.insert(
+        Method::DELETE,
+        format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+        AdminOperation(&DisableTableDirectoryHandler {}),
+    )?;
     r.insert(
         Method::GET,
         format!("{prefix}/{{warehouse}}/catalog/migration").as_str(),

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

@@ -56,6 +56,25 @@ impl Operation for RestRenameTableHandler {
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
 
         let store = table_catalog_store_from_extensions(&req.extensions)?;
+        rename_table_response(store, warehouse, source_namespace, source_table, destination_namespace, destination_table).await?;
+        Ok(empty_response(StatusCode::NO_CONTENT))
+    }
+}
+
+// Authorization has completed before ownership moves to the mutation task. A
+// disconnected request must not strand the persisted bucket-wide rename fence.
+pub(super) async fn rename_table_response<S>(
+    store: S,
+    warehouse: String,
+    source_namespace: crate::table_catalog::Namespace,
+    source_table: crate::table_catalog::IdentifierSegment,
+    destination_namespace: crate::table_catalog::Namespace,
+    destination_table: crate::table_catalog::IdentifierSegment,
+) -> S3Result<()>
+where
+    S: crate::table_catalog::TableCatalogStore + Send + 'static,
+{
+    crate::admin::handlers::supervise_admin_mutation("table rename", async move {
         store
             .rename_table(
                 &warehouse,
@@ -65,9 +84,9 @@ impl Operation for RestRenameTableHandler {
                 destination_table.as_str(),
             )
             .await
-            .map_err(catalog_store_error)?;
-        Ok(empty_response(StatusCode::NO_CONTENT))
-    }
+            .map_err(catalog_store_error)
+    })
+    .await
 }
 
 pub struct RestCreateTableHandler {}

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

@@ -277,6 +277,41 @@ async fn table_catalog_authentication_and_credentials_use_the_request_context()
     assert!(Arc::ptr_eq(&resolved_store, &store));
 
     let mut request = request;
+    iam.set_policy(policy_name, Policy::parse_config(br#"{
+        "Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["admin:GetTable","admin:CreatePluginDelegation"]},
+            {"Effect":"Deny","Action":["admin:GetTable","admin:CreatePluginDelegation"],"Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}}
+        ]
+    }"#).unwrap()).await.unwrap();
+    request
+        .extensions
+        .insert(Some(crate::server::RemoteAddr("10.1.2.3:4567".parse().unwrap())));
+    request.headers.insert("x-forwarded-for", "10.1.2.3".parse().unwrap());
+    for (client, allowed) in [
+        (None, true),
+        (Some("203.0.113.17:7890"), false),
+        (Some("10.8.7.6:7890"), true),
+    ] {
+        request.extensions.remove::<rustfs_trusted_proxies::ClientInfo>();
+        if let Some(client) = client {
+            request
+                .extensions
+                .insert(rustfs_trusted_proxies::ClientInfo::direct(client.parse().unwrap()));
+        }
+        let unscoped = authorize_table_catalog_request(&request, AdminAction::GetTableAction).await;
+        let scoped = authorize_table_catalog_resource_request(&request, &resource, AdminAction::GetTableAction).await;
+        let delegated =
+            crate::admin::handlers::plugin_delegation::owner(&request, AdminAction::CreatePluginDelegationAction).await;
+        if allowed {
+            unscoped.expect("verified private address authorizes table catalog access");
+            scoped.expect("verified private address authorizes scoped table access");
+            delegated.expect("verified private address authorizes delegation");
+        } else {
+            assert_eq!(unscoped.expect_err("external client is denied").code(), &S3ErrorCode::AccessDenied);
+            assert_eq!(scoped.err().expect("external client is denied").code(), &S3ErrorCode::AccessDenied);
+            assert_eq!(delegated.err().expect("external client is denied").code(), &S3ErrorCode::AccessDenied);
+        }
+    }
     request.method = Method::POST;
     request.uri = "/iceberg/v1/request-warehouse/namespaces/analytics/tables/events/maintenance/worker/run"
         .parse()
@@ -1473,6 +1508,7 @@ fn table_catalog_handlers_require_table_admin_actions() {
         2
     );
     assert_eq!(rename_block.matches("AdminAction::SetTableAction").count(), 2);
+    assert!(rename_block.contains("rename_table_response("));
 
     let migration_block = operation_block(&src, "GetTableCatalogMigrationHandler");
     assert!(
@@ -1731,7 +1767,7 @@ fn table_bucket_toggle_handlers_reload_peer_metadata_after_publication() {
 #[test]
 fn table_commit_publication_acquires_migration_permits_before_publication_lock() {
     let src = table_catalog_handler_source();
-    let block = function_block(&src, "async fn begin_bucket_publication");
+    let block = function_block(&src, "async fn begin_bucket_publication_inner");
     let migration_permits = block
         .find("acquire_catalog_migration_read_guards(table_bucket)")
         .expect("publication should acquire catalog migration permits");
@@ -7128,6 +7164,371 @@ async fn commit_publication_acquires_discovered_object_locks_in_key_order() {
     crate::table_catalog::TableCatalogObjectBackend::complete_table_commit_publication(&commit_backend);
 }
 
+#[tokio::test]
+async fn catalog_publication_reads_graph_once_for_registration_and_relocation() {
+    for relocate in [false, true] {
+        let backend = TestTableCatalogObjectBackend::content_addressed();
+        let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace");
+        let (metadata_location, mut metadata, version_token) = if relocate {
+            let created = create_standard_events_table(&store, &backend, &namespace).await;
+            let current = store
+                .load_table("warehouse", "analytics", "events")
+                .await
+                .expect("load table")
+                .expect("table exists");
+            let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table");
+            (
+                crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002-relocation.metadata.json"),
+                created.metadata,
+                Some(current.version_token),
+            )
+        } else {
+            let (_, location) = seed_events_registration_target(&store, &backend).await;
+            (
+                location,
+                test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id"),
+                None,
+            )
+        };
+        let warehouse = "s3://warehouse/tables/revision-graph";
+        let manifest_list = format!("{warehouse}/metadata/snap-10.avro");
+        let manifest = format!("{warehouse}/metadata/manifest-snap-10.avro");
+        let data_file = format!("{warehouse}/data/part-10.parquet");
+        metadata["location"] = serde_json::json!(warehouse);
+        metadata["last-sequence-number"] = serde_json::json!(1);
+        metadata["snapshots"] = serde_json::json!([{
+            "snapshot-id": 10, "sequence-number": 1, "timestamp-ms": 10,
+            "manifest-list": manifest_list, "summary": {"operation": "append"}
+        }]);
+        metadata["current-snapshot-id"] = serde_json::json!(10);
+        metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+        backend.put_json("warehouse", &metadata_location, metadata).await;
+        seed_test_snapshot_manifest(&backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await;
+        let commit = TableCommitObjectBackend::trusted(backend.clone());
+        if let Some(version_token) = version_token {
+            update_table_metadata_location_response(
+                &store,
+                &commit,
+                "warehouse",
+                &namespace,
+                "events",
+                UpdateTableMetadataLocationRequest {
+                    metadata_location: table_metadata_location_for_client("warehouse", &metadata_location),
+                    version_token,
+                    commit_id: Some("revision-relocation".to_string()),
+                    idempotency_key: None,
+                },
+            )
+            .await
+            .expect("relocation publishes the validated graph");
+        } else {
+            register_table_response(
+                &store,
+                &commit,
+                "warehouse",
+                &namespace,
+                RegisterTableRequest {
+                    name: "events".to_string(),
+                    metadata_location: metadata_location.clone(),
+                    overwrite: false,
+                },
+                true,
+            )
+            .await
+            .expect("registration publishes the validated graph");
+        }
+        let published = store
+            .load_table("warehouse", "analytics", "events")
+            .await
+            .expect("load published table")
+            .expect("table exists");
+        assert_eq!(published.warehouse_location, warehouse);
+        assert_eq!(published.metadata_location, metadata_location);
+        let state = backend.state.lock().await;
+        for location in [&manifest_list, &manifest] {
+            let key = ("warehouse".to_string(), test_snapshot_object_key("warehouse", location));
+            assert_eq!(
+                state.read_attempts.get(&key),
+                Some(&1),
+                "manifest bodies must be read once, before the bucket fence: {location}"
+            );
+            assert_eq!(
+                state.read_lock_acquisitions.get(&key),
+                Some(&1),
+                "publication must retain the manifest object read lock"
+            );
+        }
+    }
+}
+
+#[tokio::test]
+async fn catalog_create_binds_generated_metadata_before_publication() {
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace");
+    let created = create_standard_events_table(&store, &backend, &namespace).await;
+    let object = table_metadata_location_for_catalog("warehouse", &created.metadata_location).expect("metadata key");
+    let state = backend.state.lock().await;
+    let key = ("warehouse".to_string(), object);
+    assert_eq!(
+        state.read_attempts.get(&key),
+        Some(&1),
+        "initial create keeps its generated metadata SHA read bounded"
+    );
+    assert!(
+        state
+            .objects
+            .get(&key)
+            .expect("created metadata exists")
+            .body_revision
+            .is_some()
+    );
+    assert_eq!(
+        state.read_lock_acquisitions.get(&key),
+        Some(&1),
+        "create keeps its metadata body stable until publication"
+    );
+}
+
+#[tokio::test]
+async fn commit_publication_revalidates_legacy_bytes_with_a_total_budget() {
+    use crate::table_catalog::{TableCatalogObjectBackend as _, TableCatalogStoreError};
+
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let first = "metadata/legacy-first.avro";
+    let second = "metadata/legacy-second.avro";
+    let payload = vec![b'x'; TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES / 2 + 1];
+    for object in [first, second] {
+        backend.put_bytes("warehouse", object, payload.clone()).await;
+        backend
+            .state
+            .lock()
+            .await
+            .objects
+            .get_mut(&("warehouse".to_string(), object.to_string()))
+            .expect("legacy fixture exists")
+            .body_revision = None;
+    }
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    commit
+        .read_object("warehouse", first)
+        .await
+        .expect("first legacy object fits");
+    commit
+        .read_object("warehouse", first)
+        .await
+        .expect("repeated observations do not consume the budget twice");
+    let error = commit
+        .read_object("warehouse", second)
+        .await
+        .expect_err("combined legacy payload exceeds the budget");
+    assert!(matches!(error, TableCatalogStoreError::Unavailable(_)));
+    assert!(
+        backend.lock_attempts.lock().await.is_empty(),
+        "oversized legacy validation fails before publication locks"
+    );
+
+    // A supported small legacy object still uses the original body fingerprint,
+    // even when metadata and the absent revision cannot distinguish an overwrite.
+    let legacy = "metadata/legacy-small.json";
+    backend.put_bytes("warehouse", legacy, b"before".to_vec()).await;
+    backend
+        .state
+        .lock()
+        .await
+        .objects
+        .get_mut(&("warehouse".to_string(), legacy.to_string()))
+        .expect("small legacy fixture exists")
+        .body_revision = None;
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    commit.read_object("warehouse", legacy).await.expect("read legacy object");
+    backend
+        .state
+        .lock()
+        .await
+        .objects
+        .get_mut(&("warehouse".to_string(), legacy.to_string()))
+        .expect("legacy object remains")
+        .data = b"after!".to_vec();
+    let error = commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect_err("legacy SHA validation detects an equal-length overwrite with unchanged metadata");
+    assert!(matches!(error, TableCatalogStoreError::Conflict(_)));
+
+    // Rewriting old objects supplies fresh server revisions and removes the
+    // legacy-only size restriction without removing publication validation.
+    for object in [first, second] {
+        backend.put_bytes("warehouse", object, payload.clone()).await;
+    }
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    for object in [first, second] {
+        commit
+            .read_object("warehouse", object)
+            .await
+            .expect("rewritten object can be validated");
+    }
+    let reads = backend.read_call_count().await;
+    commit
+        .begin_table_bucket_commit_publication("warehouse")
+        .await
+        .expect("take bucket fence");
+    commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect("publish rewritten objects");
+    assert_eq!(
+        backend.read_call_count().await,
+        reads,
+        "revision validation must not reread bodies under the bucket fence"
+    );
+    commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_bounds_unread_generated_metadata() {
+    use crate::table_catalog::{TableCatalogObjectBackend as _, TableCatalogPutPrecondition, TableCatalogStoreError};
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let payload = vec![b'x'; TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES + 1];
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    commit
+        .put_object("warehouse", "metadata/generated.json", payload, TableCatalogPutPrecondition::IfAbsent)
+        .await
+        .expect("write generated metadata before its normal readback");
+    let error = commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect_err("publication cannot read oversized unbound generated metadata under its locks");
+    assert!(matches!(error, TableCatalogStoreError::Unavailable(_)));
+    assert!(backend.lock_attempts.lock().await.is_empty());
+    assert_eq!(backend.read_call_count().await, 0);
+    commit
+        .read_object("warehouse", "metadata/generated.json")
+        .await
+        .expect("the existing generated metadata readback binds the revision");
+    let reads = backend.read_call_count().await;
+    commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect("bound generated metadata can publish");
+    assert_eq!(backend.read_call_count().await, reads);
+    commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_accepts_metadata_only_changes_with_same_body_revision() {
+    use crate::table_catalog::TableCatalogObjectBackend as _;
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    backend
+        .put_bytes("warehouse", "metadata/current.json", b"unchanged body".to_vec())
+        .await;
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    commit
+        .read_object("warehouse", "metadata/current.json")
+        .await
+        .expect("observe body");
+    {
+        let mut state = backend.state.lock().await;
+        let object = state
+            .objects
+            .get_mut(&("warehouse".to_string(), "metadata/current.json".to_string()))
+            .expect("object exists");
+        object.etag = "metadata-only-update".to_string();
+        object.mod_time = Some(time::OffsetDateTime::now_utc());
+    }
+    commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect("unchanged body remains valid after a metadata-only update");
+    assert_eq!(backend.read_call_count().await, 1);
+    commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_accepts_unchanged_legacy_bytes() {
+    use crate::table_catalog::TableCatalogObjectBackend as _;
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    backend
+        .put_bytes("warehouse", "metadata/legacy.json", b"unchanged".to_vec())
+        .await;
+    backend
+        .state
+        .lock()
+        .await
+        .objects
+        .get_mut(&("warehouse".to_string(), "metadata/legacy.json".to_string()))
+        .expect("legacy fixture exists")
+        .body_revision = None;
+    let commit = TableCommitObjectBackend::trusted(backend.clone());
+    commit
+        .read_object("warehouse", "metadata/legacy.json")
+        .await
+        .expect("observe legacy body");
+    commit
+        .prepare_table_commit_publication("warehouse", "analytics", "events")
+        .await
+        .expect("small legacy object remains supported");
+    assert_eq!(
+        backend.read_call_count().await,
+        2,
+        "legacy body must still be verified under its object lock"
+    );
+    commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn catalog_body_revision_changes_on_real_identical_body_overwrite() {
+    use crate::table_catalog::{
+        EcStoreTableCatalogObjectBackend, StrongTableCatalogRuntime, TableCatalogObjectBackend as _, TableCatalogPutPrecondition,
+    };
+    let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = "catalog-body-revision";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let backend = EcStoreTableCatalogObjectBackend::new_with_strong_runtime(store, StrongTableCatalogRuntime::default());
+    backend
+        .put_object(bucket, "manifest.avro", b"same bytes".to_vec(), TableCatalogPutPrecondition::IfAbsent)
+        .await
+        .expect("write object");
+    let before = backend
+        .read_object(bucket, "manifest.avro")
+        .await
+        .expect("read body")
+        .expect("object exists");
+    let before_stat = backend
+        .object_metadata(bucket, "manifest.avro")
+        .await
+        .expect("stat body")
+        .expect("object exists");
+    assert!(before.body_revision.is_some());
+    assert_eq!(before.body_revision, before_stat.body_revision);
+    backend
+        .put_object(
+            bucket,
+            "manifest.avro",
+            b"same bytes".to_vec(),
+            TableCatalogPutPrecondition::IfMatch(before.etag.clone().expect("etag exists")),
+        )
+        .await
+        .expect("overwrite same bytes");
+    let after = backend
+        .read_object(bucket, "manifest.avro")
+        .await
+        .expect("read overwritten body")
+        .expect("object exists");
+    let after_stat = backend
+        .object_metadata(bucket, "manifest.avro")
+        .await
+        .expect("stat overwritten body")
+        .expect("object exists");
+    assert_eq!(before.etag, after.etag);
+    assert_ne!(before.body_revision, after.body_revision);
+    assert_eq!(after.body_revision, after_stat.body_revision);
+}
+
 #[tokio::test]
 async fn commit_publication_revalidates_objects_after_ordered_lock_acquisition() {
     let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
@@ -7160,6 +7561,7 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
     metadata_backend.state.lock().await.objects.insert(
         ("warehouse".to_string(), object.to_string()),
         TestCatalogObjectRecord {
+            body_revision: Some(Uuid::new_v4()),
             data: replacement,
             etag: original_etag.clone(),
             mod_time: None,
@@ -7173,6 +7575,7 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
     metadata_backend.state.lock().await.objects.insert(
         ("warehouse".to_string(), object.to_string()),
         TestCatalogObjectRecord {
+            body_revision: Some(Uuid::new_v4()),
             data: original,
             etag: original_etag,
             mod_time: None,
@@ -7226,6 +7629,7 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
             state.objects.insert(
                 ("warehouse".to_string(), test_snapshot_object_key("warehouse", file)),
                 TestCatalogObjectRecord {
+                    body_revision: Some(Uuid::new_v4()),
                     data: data.clone(),
                     etag: etag.clone(),
                     mod_time: None,
@@ -7256,7 +7660,7 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
         ]
     }))
     .expect("boundary commit request should parse");
-    let commit_backend = TableCommitObjectBackend::trusted(metadata_backend);
+    let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
 
     let result = standard_commit_table_response(&store, &commit_backend, "warehouse", &namespace, "events", request).await;
     let response = commit_backend
@@ -7265,6 +7669,14 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
         .expect("a valid snapshot with more than ten thousand live files should publish");
 
     assert_eq!(response.metadata["current-snapshot-id"], 20);
+    for location in [&manifest_list, &manifest] {
+        let key = ("warehouse".to_string(), test_snapshot_object_key("warehouse", location));
+        assert_eq!(
+            metadata_backend.state.lock().await.read_attempts.get(&key),
+            Some(&2),
+            "graph and conflict validation each read manifests once; publication must not read them again"
+        );
+    }
     assert_eq!(
         store
             .load_table("warehouse", "analytics", "events")
@@ -7565,6 +7977,78 @@ async fn metadata_location_api_rejects_unbound_legacy_catalog_identity() {
     assert_events_table_entry_unchanged(&store, &legacy_entry).await;
 }
 
+#[tokio::test]
+async fn rename_request_cancellation_finishes_the_persisted_bucket_fence() {
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+    let namespace = crate::table_catalog::Namespace::parse("analytics").unwrap();
+    let source = crate::table_catalog::IdentifierSegment::parse("events").unwrap();
+    let destination = crate::table_catalog::IdentifierSegment::parse("renamed").unwrap();
+    create_standard_events_table(&store, &backend, &namespace).await;
+    let original = store.load_table("warehouse", "analytics", "events").await.unwrap().unwrap();
+    let paths = crate::table_catalog::TableCatalogObjectPaths::default();
+    let source_path = paths.table_entry_path("warehouse", &namespace, &source);
+    let attempt = backend
+        .put_attempt_count(crate::admin::storage_api::RUSTFS_META_BUCKET, &source_path)
+        .await
+        + 2;
+    let pause = backend
+        .pause_put_attempt(crate::admin::storage_api::RUSTFS_META_BUCKET, &source_path, attempt)
+        .await;
+    let request = tokio::spawn(table::rename_table_response(
+        store.clone(),
+        "warehouse".to_string(),
+        namespace.clone(),
+        source.clone(),
+        namespace.clone(),
+        destination.clone(),
+    ));
+    tokio::time::timeout(std::time::Duration::from_secs(10), pause.wait_started())
+        .await
+        .expect("rename reaches persisted fence");
+    assert!(
+        store
+            .get_table_bucket("warehouse")
+            .await
+            .unwrap()
+            .unwrap()
+            .active_rename_id
+            .is_some()
+    );
+    request.abort();
+    assert!(request.await.unwrap_err().is_cancelled());
+    pause.release();
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        while store
+            .get_table_bucket("warehouse")
+            .await
+            .unwrap()
+            .unwrap()
+            .active_rename_id
+            .is_some()
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("detached mutation clears the durable fence");
+    let renamed = store.load_table("warehouse", "analytics", "renamed").await.unwrap().unwrap();
+    assert_eq!(renamed.table_id, original.table_id);
+    assert_eq!(renamed.metadata_location, original.metadata_location);
+    assert!(store.load_table("warehouse", "analytics", "events").await.unwrap().is_none());
+    assert!(
+        store
+            .resolve_table_data_plane_resource("warehouse", "ordinary.txt")
+            .await
+            .unwrap()
+            .is_none()
+    );
+    table::rename_table_response(store.clone(), "warehouse".to_string(), namespace.clone(), destination, namespace, source)
+        .await
+        .expect("subsequent ordinary rename succeeds");
+    assert!(store.load_table("warehouse", "analytics", "events").await.unwrap().is_some());
+}
+
 #[tokio::test]
 async fn table_metadata_maintenance_helper_runs_dry_run_and_delete() {
     let backend = TestTableCatalogObjectBackend::content_addressed();
@@ -15963,6 +16447,7 @@ struct ColdStrongRegistrationBackend<B> {
     armed: Arc<std::sync::atomic::AtomicBool>,
     reader_ready: Arc<tokio::sync::Notify>,
     resume: Arc<tokio::sync::Notify>,
+    backup_writer_waiting: Arc<tokio::sync::Notify>,
 }
 
 impl<B> ColdStrongRegistrationBackend<B> {
@@ -16012,7 +16497,16 @@ impl<B: crate::table_catalog::TableCatalogObjectBackend> crate::table_catalog::T
         bucket: &str,
         object: &str,
     ) -> crate::table_catalog::TableCatalogStoreResult<crate::table_catalog::TableCatalogLockGuard> {
-        self.inner.acquire_write_lock(bucket, object).await
+        let mut acquire = self.inner.acquire_write_lock(bucket, object);
+        if bucket == crate::admin::storage_api::RUSTFS_META_BUCKET
+            && object == crate::table_catalog::TableCatalogObjectPaths::default().backing_migration_global_fence_lock_path()
+        {
+            match futures::poll!(acquire.as_mut()) {
+                std::task::Poll::Ready(result) => return result,
+                std::task::Poll::Pending => self.backup_writer_waiting.notify_one(),
+            }
+        }
+        acquire.await
     }
 
     async fn read_object(
@@ -16116,6 +16610,7 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
         armed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
         reader_ready: Arc::new(tokio::sync::Notify::new()),
         resume: Arc::new(tokio::sync::Notify::new()),
+        backup_writer_waiting: Arc::new(tokio::sync::Notify::new()),
     };
     let catalog = StrongTableCatalogStore::new(backend.clone());
     for bucket in [&bucket, &backup_bucket] {
@@ -16176,17 +16671,18 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
         .await
         .expect("registration holds first real global reader");
     // The backup targets another bucket, so its documented writer-drain precondition is met.
-    let backup_catalog = StrongTableCatalogStore::new(inner);
-    let mut backup = Box::pin(backup_catalog.create_durable_catalog_backup(&backup_bucket, None));
-    assert!(
-        futures::poll!(backup.as_mut()).is_pending(),
-        "actual backup writer must queue behind held registration reader"
-    );
+    let backup_catalog = StrongTableCatalogStore::new(backend.clone());
+    // Keep driving route discovery and snapshot I/O until the global write lock itself is pending.
+    let backup = tokio::spawn(async move { backup_catalog.create_durable_catalog_backup(&backup_bucket, None).await });
+    tokio::time::timeout(StdDuration::from_secs(5), backend.backup_writer_waiting.notified())
+        .await
+        .expect("actual backup writer must queue behind held registration reader");
     backend.resume.notify_one();
     let result = tokio::time::timeout(StdDuration::from_secs(2), &mut registration).await;
     if result.is_err() {
         // Prove that the queued backup, rather than ECStore I/O latency, caused the stall.
-        drop(backup);
+        backup.abort();
+        assert!(backup.await.unwrap_err().is_cancelled());
         tokio::time::timeout(StdDuration::from_secs(5), &mut registration)
             .await
             .expect("registration must resume once only the queued backup is cancelled")
@@ -16201,6 +16697,7 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
     tokio::time::timeout(StdDuration::from_secs(5), backup)
         .await
         .expect("the other bucket backup should resume when registration releases its reader")
+        .expect("backup task joins")
         .expect("the other bucket backup succeeds");
 }
 
@@ -16528,6 +17025,59 @@ async fn cleanup_empty_table_bucket_reclaims_twenty_dropped_tables_and_allows_bu
     object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
 }
 
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_validates_directory_candidate_path_and_body_before_deletion() {
+    use crate::admin::storage_api::RUSTFS_META_BUCKET;
+    use crate::table_catalog::{TableCatalogObjectPaths, TableCatalogPutPrecondition};
+
+    let (_temp, _disks, storage) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-directory-{}", Uuid::new_v4().simple());
+    let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
+        storage.clone(),
+        crate::table_catalog::StrongTableCatalogRuntime::default(),
+    );
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    let metadata = create_dropped_empty_tables_for_cleanup(&catalog, &backend, &storage, &bucket, 1).await;
+    let paths = TableCatalogObjectPaths::default();
+    let bucket_entry = paths.table_bucket_entry_path(&bucket);
+    let directory = format!("{}/table-entry-directory/v1/", bucket_entry.rsplit_once('/').unwrap().0);
+    let real = format!("{directory}analytics/tables/events_00/table-entry.json");
+    assert!(backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+    for (suffix, target_suffix) in [
+        ("bad segment/tables/x/table-entry.json", "bad segment/tables/x/table-entry.json"),
+        ("analytics/tables/forged/table-entry.json", "analytics/tables/events_00/table-entry.json"),
+    ] {
+        let unknown = format!("{directory}{suffix}");
+        let target = format!("{}{target_suffix}", paths.namespace_entries_prefix(&bucket));
+        backend
+            .put_object(RUSTFS_META_BUCKET, &unknown, target.into_bytes(), TableCatalogPutPrecondition::IfAbsent)
+            .await
+            .unwrap();
+        let publication = TableCommitObjectBackend::trusted(backend.clone());
+        assert!(
+            cleanup_empty_table_bucket_response(&catalog, &publication, &storage, &bucket)
+                .await
+                .is_err()
+        );
+        assert!(backend.object_exists(RUSTFS_META_BUCKET, &unknown).await.unwrap());
+        assert!(backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+        assert!(backend.object_exists(&bucket, &metadata[0]).await.unwrap());
+        backend.delete_object(RUSTFS_META_BUCKET, &unknown).await.unwrap();
+    }
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, &storage, &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    assert!(!backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+}
+
 #[tokio::test]
 #[serial_test::serial]
 async fn cleanup_empty_table_bucket_validates_authorization_and_metadata_before_deleting() {
@@ -16705,7 +17255,7 @@ async fn cleanup_empty_table_bucket_handler_enforces_request_object_permissions(
                         assert!(!backend.object_exists(&bucket, object).await.unwrap());
                     }
                 } else {
-                    assert_eq!(result.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+                    assert_eq!(result.expect_err("request must be denied").code(), &S3ErrorCode::AccessDenied);
                     for object in &metadata_objects {
                         assert!(backend.object_exists(&bucket, object).await.unwrap());
                     }
@@ -17099,3 +17649,202 @@ async fn cleanup_empty_table_bucket_stops_before_delete_when_publication_lease_i
     );
     object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
 }
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_first_fenced_metadata_put_finishes_after_budget() {
+    use crate::table_catalog::{TableCatalogObjectBackend, TableCatalogPutPrecondition, TableCommitPublication};
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    publication.begin_table_bucket("warehouse").await.unwrap();
+    let key = "generated.metadata.json";
+    let pause = backend.pause_next_put("warehouse", key).await;
+    let worker = publication.clone();
+    let put = tokio::spawn(async move {
+        worker
+            .put_object("warehouse", key, b"{}".to_vec(), TableCatalogPutPrecondition::IfAbsent)
+            .await
+    });
+    pause.wait_started().await;
+    tokio::time::advance(crate::storage_api::table::get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    assert!(publication.holds_table_bucket("warehouse"));
+    assert!(!put.is_finished());
+    pause.release();
+    put.await.unwrap().unwrap();
+    publication
+        .preparation
+        .read(&publication, backend.object_metadata("warehouse", key))
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(publication.holds_table_bucket("warehouse"));
+    publication.complete();
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_slow_head_clears_partial_http_preparation() {
+    use crate::table_catalog::test_support::TestCatalogObjectPause;
+    use crate::table_catalog::{TableCatalogObjectBackend, TableCommitPublication};
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    backend.seed_object("warehouse", "data/file.parquet", b"data".to_vec()).await;
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    publication.object_exists("warehouse", "data/file.parquet").await.unwrap();
+    publication.begin_table_bucket("warehouse").await.unwrap();
+    let pause = TestCatalogObjectPause::default();
+    backend
+        .state
+        .lock()
+        .await
+        .pause_metadata
+        .insert(("warehouse".to_string(), "data/file.parquet".to_string()), pause.clone());
+    let preparing = {
+        let publication = publication.clone();
+        tokio::spawn(async move { publication.prepare("warehouse", "analytics", "events").await })
+    };
+    pause.wait_started().await;
+    tokio::time::advance(crate::storage_api::table::get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    assert!(
+        matches!(preparing.await.unwrap(), Err(crate::table_catalog::TableCatalogStoreError::Unavailable(message)) if message.contains("preparation deadline"))
+    );
+    assert!(!publication.holds_table_bucket("warehouse"));
+    assert!(!publication.holds_table("warehouse", "analytics", "events"));
+    assert!(publication.publication.lock().observed_objects.is_empty());
+    assert!(backend.state.lock().await.put_attempts.is_empty());
+    pause.release();
+    let _writer = backend
+        .acquire_read_lock("warehouse", &crate::table_catalog::default_table_bucket_publication_lock_path())
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn directory_management_handlers_enforce_migration_permission_before_mutation() {
+    temp_env::async_with_vars([
+        (crate::table_catalog::ENV_TABLE_CATALOG_BACKING, Some("object")),
+        ("RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX", Some("true")),
+        ("RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX_FLEET_CONFIRMED", Some("true")),
+    ], async {
+
+    let (_temp_dir, _disks, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    object_store
+        .make_bucket("warehouse", &MakeBucketOptions::default())
+        .await
+        .unwrap();
+    rustfs_iam::store::object::ObjectStore::new(object_store.clone())
+        .save_iam_config(
+            serde_json::json!({"version": 1}),
+            format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+        )
+        .await
+        .unwrap();
+    let iam = rustfs_iam::build_iam_sys(object_store.clone()).await.unwrap();
+    let access_key = "directory-upgrade-user";
+    let secret_key = "directory-upgrade-secret";
+    iam.create_user(
+        access_key,
+        &AddOrUpdateUserReq {
+            secret_key: secret_key.to_string(),
+            policy: None,
+            status: AccountStatus::Enabled,
+        },
+    )
+    .await
+    .unwrap();
+    let context = Arc::new(AppContext::new(
+        object_store.clone(),
+        Arc::new(RequestIam { handle: iam.clone() }),
+        Arc::new(RequestKms),
+    ));
+    assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+        access_key: "directory-root".to_string(),
+        secret_key: "directory-root-secret".to_string(),
+        status: "on".to_string(),
+        ..Default::default()
+    }));
+    let setup = table_catalog_handler_request(context.clone(), access_key, secret_key);
+    let backend = table_catalog_backend_from_extensions(&setup.extensions).unwrap();
+    let configured = table_catalog_store_from_backend(backend.clone()).unwrap();
+    enable_table_bucket_marker(object_store.as_ref(), "warehouse").await.unwrap();
+    ensure_table_bucket_entry(&configured, "warehouse", true).await.unwrap();
+
+    for (name, policy, allowed) in [
+        (
+            "directory-maintenance",
+            serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:RunTableMaintenance"]}]}),
+            false,
+        ),
+        (
+            "directory-denied",
+            serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:*"]},{"Effect":"Deny","Action":["admin:MigrateTableCatalog"]}]}),
+            false,
+        ),
+        (
+            "directory-migrator",
+            serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:MigrateTableCatalog"]}]}),
+            true,
+        ),
+    ] {
+        iam.set_policy(name, Policy::parse_config(&serde_json::to_vec(&policy).unwrap()).unwrap())
+            .await
+            .unwrap();
+        iam.policy_db_set(access_key, UserType::Reg, false, name).await.unwrap();
+        for prefix in ["/iceberg/v1", "/_iceberg/v1"] {
+            let mut router = matchit::Router::new();
+            router
+                .insert(format!("{prefix}/{{warehouse}}/catalog/directory"), ())
+                .unwrap();
+            let uri = format!("{prefix}/warehouse/catalog/directory");
+            let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+            request.method = Method::DELETE;
+            request.uri = uri.parse().unwrap();
+            let matched = router.at(&uri).unwrap();
+            let result = DisableTableDirectoryHandler {}.call(request, matched.params).await;
+            if allowed {
+                assert_eq!(result.unwrap().output.0, StatusCode::OK);
+            } else {
+                assert_eq!(result.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+                let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+                request.method = Method::POST;
+                request.uri = uri.parse().unwrap();
+                let matched = router.at(&uri).unwrap();
+                assert_eq!(
+                    AdvanceTableDirectoryHandler {}
+                        .call(request, matched.params)
+                        .await
+                        .unwrap_err()
+                        .code(),
+                    &S3ErrorCode::AccessDenied
+                );
+            }
+            if allowed {
+                for expected_phase in ["building", "ready"] {
+                    let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+                    request.method = Method::POST;
+                    request.uri = uri.parse().unwrap();
+                    let matched = router.at(&uri).unwrap();
+                    let response = AdvanceTableDirectoryHandler {}.call(request, matched.params).await.unwrap();
+                    assert_eq!(response.output.0, StatusCode::OK);
+                    let bytes = http_body_util::BodyExt::collect(response.output.1).await.unwrap().to_bytes();
+                    let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+                    assert_eq!(value["phase"], expected_phase);
+                }
+                let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+                request.method = Method::DELETE;
+                request.uri = uri.parse().unwrap();
+                let matched = router.at(&uri).unwrap();
+                assert_eq!(DisableTableDirectoryHandler {}.call(request, matched.params).await.unwrap().output.0, StatusCode::OK);
+            }
+            let object_catalog = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+            assert!(
+                object_catalog
+                    .table_directory_status("warehouse")
+                    .await
+                    .unwrap()
+                    .phase
+                    .is_none(),
+                "denied requests and idempotent rollback must not activate a directory"
+            );
+        }
+    }
+    }).await;
+}

rustfs/src/admin/route_policy.rs

@@ -1132,6 +1132,24 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         GET_TABLE_BUCKET,
         RouteRiskLevel::Sensitive,
     ),
+    admin(
+        HttpMethod::Get,
+        "/iceberg/v1/{warehouse}/catalog/directory",
+        GET_TABLE_CATALOG,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/iceberg/v1/{warehouse}/catalog/directory",
+        MIGRATE_TABLE_CATALOG,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Delete,
+        "/iceberg/v1/{warehouse}/catalog/directory",
+        MIGRATE_TABLE_CATALOG,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/iceberg/v1/{warehouse}/catalog/migration",
@@ -1440,6 +1458,24 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         GET_TABLE_BUCKET,
         RouteRiskLevel::Sensitive,
     ),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/catalog/directory",
+        GET_TABLE_CATALOG,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/catalog/directory",
+        MIGRATE_TABLE_CATALOG,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Delete,
+        "/_iceberg/v1/{warehouse}/catalog/directory",
+        MIGRATE_TABLE_CATALOG,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/_iceberg/v1/{warehouse}/catalog/migration",
@@ -2125,7 +2161,7 @@ mod tests {
         let table_specs = ADMIN_ROUTE_POLICY_SPECS
             .iter()
             .filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
-        assert_eq!(table_specs.count(), 106);
+        assert_eq!(table_specs.count(), 112);
         assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
         assert_action(HttpMethod::Delete, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
         assert_action(HttpMethod::Delete, "/_iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
@@ -2304,6 +2340,12 @@ mod tests {
             "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
             RUN_TABLE_MAINTENANCE,
         );
+        assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/catalog/directory", GET_TABLE_CATALOG);
+        assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+        assert_action(HttpMethod::Delete, "/iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+        assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/catalog/directory", GET_TABLE_CATALOG);
+        assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+        assert_action(HttpMethod::Delete, "/_iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
         assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
         assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
         assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/catalog/migration", MIGRATE_TABLE_CATALOG);

rustfs/src/storage_api.rs

@@ -483,6 +483,7 @@ pub(crate) mod workload {
 pub(crate) mod table {
     pub(crate) use crate::storage::storage_api::ecstore_bucket::utils::check_valid_bucket_name_strict;
     pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};
+    pub(crate) use crate::storage::storage_api::request_context_consumer::spawn_traced_join;
     pub(crate) mod contract {
         pub(crate) mod http {
             pub(crate) use super::super::super::storage_contracts::HTTPPreconditions;
@@ -518,6 +519,13 @@ pub(crate) mod table {
         StorageObjectInfo, StorageObjectOptions, StorageObjectToDelete, StoragePutObjReader, get_bucket_metadata,
         get_lock_acquire_timeout, table_catalog_path_hash,
     };
+
+    #[cfg(test)]
+    pub(crate) mod test_support {
+        pub(crate) use crate::storage::storage_api::contract::bucket::{BucketOperations, MakeBucketOptions};
+        pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::init_bucket_metadata_sys;
+        pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
+    }
 }
 
 pub(crate) mod on_demand_migration {

rustfs/src/table_catalog/iceberg/manifest.rs

@@ -17,6 +17,7 @@ use std::io::Read;
 use super::super::*;
 
 const AVRO_ZSTANDARD_MAX_WINDOW_LOG: u32 = 27;
+const AVRO_MANIFEST_MAX_SCHEMA_DEPTH: usize = 64;
 
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct ManifestDataFileReference {
@@ -182,6 +183,7 @@ pub(crate) fn decode_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<
     let decoded_size = validate_avro_container(data)?;
     let reader = apache_avro::Reader::new(data)
         .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list Avro: {err}")))?;
+    validate_manifest_writer_schema(reader.writer_schema())?;
     let format_version =
         avro_record_format_version(reader.writer_schema(), &["sequence_number", "min_sequence_number"], "manifest list")?;
     if format_version == 2 {
@@ -276,6 +278,7 @@ pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Decod
     let decoded_size = validate_avro_container(data)?;
     let reader = apache_avro::Reader::new(data)
         .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest Avro: {err}")))?;
+    validate_manifest_writer_schema(reader.writer_schema())?;
     let format_version =
         avro_record_format_version(reader.writer_schema(), &["sequence_number", "file_sequence_number"], "manifest")?;
     let partition_spec_id = reader
@@ -365,6 +368,79 @@ enum AvroContainerCodec {
     Zstandard,
 }
 
+// Iceberg manifests do not use recursive types. Bound resolved depth before any
+// value decoding; byte and record limits cannot bound recursive Avro values.
+fn validate_manifest_writer_schema(schema: &apache_avro::Schema) -> TableCatalogStoreResult<()> {
+    use apache_avro::schema::{Name, NamesRef, ResolvedSchema};
+    use std::collections::{HashMap, HashSet};
+
+    fn depth(
+        schema: &apache_avro::Schema,
+        names: &NamesRef<'_>,
+        active: &mut HashSet<Name>,
+        heights: &mut HashMap<Name, usize>,
+        nesting: usize,
+    ) -> TableCatalogStoreResult<usize> {
+        use apache_avro::Schema;
+        if nesting > AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+            return Err(TableCatalogStoreError::Invalid(
+                "manifest Avro schema exceeds the maximum nesting depth".to_string(),
+            ));
+        }
+        if let Schema::Ref { name } = schema {
+            let target = names.get(name).ok_or_else(|| {
+                TableCatalogStoreError::Invalid("manifest Avro schema has an unresolved type reference".to_string())
+            })?;
+            return depth(target, names, active, heights, nesting);
+        }
+        if let Some(name) = schema.name() {
+            if let Some(height) = heights.get(name) {
+                if nesting + height - 1 > AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+                    return Err(TableCatalogStoreError::Invalid(
+                        "manifest Avro schema exceeds the maximum nesting depth".to_string(),
+                    ));
+                }
+                return Ok(*height);
+            }
+            if !active.insert(name.clone()) {
+                return Err(TableCatalogStoreError::Invalid(
+                    "manifest Avro schema contains a recursive type".to_string(),
+                ));
+            }
+        }
+        let mut height = 1;
+        let mut visit = |child| -> TableCatalogStoreResult<()> {
+            height = height.max(1 + depth(child, names, active, heights, nesting + 1)?);
+            Ok(())
+        };
+        match schema {
+            Schema::Record(record) => {
+                for field in &record.fields {
+                    visit(&field.schema)?;
+                }
+            }
+            Schema::Array(array) => visit(&array.items)?,
+            Schema::Map(map) => visit(&map.types)?,
+            Schema::Union(union) => {
+                for variant in union.variants() {
+                    visit(variant)?;
+                }
+            }
+            _ => {}
+        }
+        if let Some(name) = schema.name() {
+            active.remove(name);
+            heights.insert(name.clone(), height);
+        }
+        Ok(height)
+    }
+
+    let resolved = ResolvedSchema::try_from(schema)
+        .map_err(|err| TableCatalogStoreError::Invalid(format!("invalid manifest Avro schema: {err}")))?;
+    depth(schema, resolved.get_names(), &mut HashSet::new(), &mut HashMap::new(), 1)?;
+    Ok(())
+}
+
 fn validate_avro_container(data: &[u8]) -> TableCatalogStoreResult<usize> {
     if !data.starts_with(b"Obj\x01") {
         return Err(TableCatalogStoreError::Invalid("Avro container has invalid header magic".to_string()));
@@ -665,6 +741,96 @@ mod tests {
     use super::*;
     use std::assert_matches;
 
+    fn schema_container(schema: &serde_json::Value, record: &[u8]) -> Vec<u8> {
+        fn long(bytes: &mut Vec<u8>, value: usize) {
+            let mut encoded = u64::try_from(value).expect("fixture length") << 1;
+            while encoded >= 128 {
+                bytes.push((encoded as u8 & 0x7f) | 0x80);
+                encoded >>= 7;
+            }
+            bytes.push(encoded as u8);
+        }
+        let schema = serde_json::to_vec(schema).expect("schema JSON");
+        let mut bytes = b"Obj\x01".to_vec();
+        long(&mut bytes, 1);
+        long(&mut bytes, b"avro.schema".len());
+        bytes.extend_from_slice(b"avro.schema");
+        long(&mut bytes, schema.len());
+        bytes.extend_from_slice(&schema);
+        long(&mut bytes, 0);
+        bytes.extend_from_slice(&[0; 16]);
+        long(&mut bytes, 1);
+        long(&mut bytes, record.len());
+        bytes.extend_from_slice(record);
+        bytes.extend_from_slice(&[0; 16]);
+        bytes
+    }
+
+    #[test]
+    fn manifest_decoders_reject_recursive_writer_schemas_before_decoding_values() {
+        for child in [
+            serde_json::json!(["null", "example.manifest_file"]),
+            serde_json::json!({"type":"array", "items":"example.manifest_file"}),
+            serde_json::json!({"type":"map", "values":"example.manifest_file"}),
+            serde_json::json!({"type":"record", "name":"child", "fields":[{"name":"parent", "type":"example.manifest_file"}]}),
+        ] {
+            let schema = serde_json::json!({
+                "type":"record", "name":"manifest_file", "namespace":"example",
+                "fields":[{"name":"manifest_path","type":"string"}, {"name":"next", "type":child}]
+            });
+            // A valid self-referential union datum can drive thousands of decoder frames.
+            let mut record = [0, 2].repeat(10_000);
+            record.extend_from_slice(&[0, 0]);
+            let data = schema_container(&schema, &record);
+            assert!(
+                matches!(decode_manifest_list_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("recursive type"))
+            );
+            assert!(
+                matches!(decode_manifest_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("recursive type"))
+            );
+        }
+    }
+
+    #[test]
+    fn manifest_decoders_bound_resolved_schema_depth() {
+        let mut fields = vec![serde_json::json!({"name":"manifest_path", "type":"string"})];
+        for index in 0..AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+            let children = if index == 0 {
+                vec![]
+            } else {
+                vec![serde_json::json!({"name":"previous", "type":format!("node{}", index - 1)})]
+            };
+            fields.push(serde_json::json!({"name":format!("field{index}"), "type":{"type":"record", "name":format!("node{index}"), "fields":children}}));
+        }
+        let data = schema_container(&serde_json::json!({"type":"record", "name":"manifest_file", "fields":fields}), &[0]);
+        assert!(
+            matches!(decode_manifest_list_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("maximum nesting depth"))
+        );
+        assert!(
+            matches!(decode_manifest_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("maximum nesting depth"))
+        );
+    }
+
+    #[test]
+    fn manifest_decoder_accepts_reused_nonrecursive_named_types() {
+        let schema = serde_json::json!({
+            "type":"record", "name":"manifest_file", "namespace":"example",
+            "fields":[
+                {"name":"manifest_path", "type":"string"},
+                {"name":"first", "type":{"type":"record", "name":"detail", "fields":[{"name":"value", "type":"string"}]}},
+                {"name":"second", "type":"detail"}
+            ]
+        });
+        let path = "manifest.avro";
+        let mut record = vec![u8::try_from(path.len() * 2).expect("short path")];
+        record.extend_from_slice(path.as_bytes());
+        record.extend_from_slice(&[0, 0]);
+        let decoded =
+            decode_manifest_list_avro(&schema_container(&schema, &record)).expect("reused named types are not recursive");
+        assert_eq!(decoded.references.len(), 1);
+        assert_eq!(decoded.references[0].manifest_path, path);
+    }
+
     #[test]
     fn rejects_v2_manifest_lists_without_required_count_fields() {
         let schema = apache_avro::Schema::parse_str(

rustfs/src/table_catalog/maintenance/compaction.rs

@@ -0,0 +1,666 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::super::*;
+use datafusion::arrow::array::{
+    Array, BinaryArray, BinaryViewArray, FixedSizeListArray, LargeBinaryArray, LargeListArray, LargeListViewArray,
+    LargeStringArray, ListArray, ListViewArray, MapArray, StringArray, StringViewArray, StructArray, downcast_dictionary_array,
+};
+use datafusion::arrow::datatypes::{DataType, Field, Schema};
+use datafusion::parquet::arrow::{
+    add_encoded_arrow_schema_to_metadata, arrow_reader::ArrowReaderOptions, arrow_writer::ArrowWriterOptions,
+};
+use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties};
+use std::io::Write;
+use std::sync::OnceLock;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+
+const COMPACTION_WRITE_BYTES: usize = 8 * 1024 * 1024;
+const COMPACTION_BATCH_ROWS: usize = 256;
+const COMPACTION_ROW_GROUP_ROWS: usize = 65_536;
+const COMPACTION_MAX_ROW_GROUPS: usize = 256;
+const COMPACTION_MAX_COLUMN_CHUNKS: usize = 16_384;
+const COMPACTION_SCHEMA_DEPTH: usize = 64;
+const COMPACTION_SCHEMA_FIELDS: usize = 1024;
+
+pub(crate) fn try_admit_compaction() -> TableCatalogStoreResult<Arc<OwnedSemaphorePermit>> {
+    static ADMISSION: OnceLock<Arc<Semaphore>> = OnceLock::new();
+    ADMISSION
+        .get_or_init(|| Arc::new(Semaphore::new(1)))
+        .clone()
+        .try_acquire_owned()
+        .map(Arc::new)
+        .map_err(|_| TableCatalogStoreError::Unavailable("table compaction capacity is full".to_owned()))
+}
+
+fn budget_error() -> TableCatalogStoreError {
+    TableCatalogStoreError::Invalid("table compaction exceeds the server memory or expanded-value budget".to_owned())
+}
+
+fn add_budget(total: &mut usize, amount: usize, limit: usize) -> TableCatalogStoreResult<()> {
+    *total = total
+        .checked_add(amount)
+        .filter(|value| *value <= limit)
+        .ok_or_else(budget_error)?;
+    Ok(())
+}
+
+fn view_field(field: &Field, depth: usize, fields: &mut usize) -> TableCatalogStoreResult<Arc<Field>> {
+    add_budget(fields, 1, COMPACTION_SCHEMA_FIELDS)?;
+    Ok(Arc::new(field.clone().with_data_type(view_type(field.data_type(), depth, fields)?)))
+}
+
+fn view_type(data_type: &DataType, depth: usize, fields: &mut usize) -> TableCatalogStoreResult<DataType> {
+    if depth > COMPACTION_SCHEMA_DEPTH {
+        return Err(budget_error());
+    }
+    Ok(match data_type {
+        DataType::Utf8 | DataType::LargeUtf8 => DataType::Utf8View,
+        DataType::Binary | DataType::LargeBinary => DataType::BinaryView,
+        DataType::List(field) => DataType::List(view_field(field, depth + 1, fields)?),
+        DataType::LargeList(field) => DataType::LargeList(view_field(field, depth + 1, fields)?),
+        DataType::ListView(field) => DataType::ListView(view_field(field, depth + 1, fields)?),
+        DataType::LargeListView(field) => DataType::LargeListView(view_field(field, depth + 1, fields)?),
+        DataType::FixedSizeList(field, size) => DataType::FixedSizeList(view_field(field, depth + 1, fields)?, *size),
+        DataType::Map(field, sorted) => DataType::Map(view_field(field, depth + 1, fields)?, *sorted),
+        DataType::Struct(children) => DataType::Struct(
+            children
+                .iter()
+                .map(|field| view_field(field, depth + 1, fields))
+                .collect::<TableCatalogStoreResult<Vec<_>>>()?
+                .into(),
+        ),
+        DataType::Dictionary(_, value) => {
+            // The pinned reader does not support Dictionary<ByteView>. Keep
+            // its already shared values, while still checking nested schema limits.
+            let _ = view_type(value, depth + 1, fields)?;
+            data_type.clone()
+        }
+        other => other.clone(),
+    })
+}
+
+fn view_schema(schema: &Schema) -> TableCatalogStoreResult<SchemaRef> {
+    let mut fields = 0;
+    let mapped = schema
+        .fields()
+        .iter()
+        .map(|field| view_field(field, 0, &mut fields))
+        .collect::<TableCatalogStoreResult<Vec<_>>>()?;
+    Ok(Arc::new(Schema::new_with_metadata(mapped, schema.metadata().clone())))
+}
+
+/// Charge the values the writer may expand, not the small view/dictionary
+/// buffers. Even a null or empty nested cell consumes structural capacity.
+fn charge_value(array: &dyn Array, row: usize, depth: usize, total: &mut usize, limit: usize) -> TableCatalogStoreResult<()> {
+    if depth > COMPACTION_SCHEMA_DEPTH {
+        return Err(budget_error());
+    }
+    add_budget(total, 8, limit)?;
+    if array.is_null(row) {
+        return Ok(());
+    }
+    macro_rules! scalar_bytes {
+        ($ty:ty) => {
+            add_budget(
+                total,
+                array
+                    .as_any()
+                    .downcast_ref::<$ty>()
+                    .ok_or_else(budget_error)?
+                    .value(row)
+                    .len(),
+                limit,
+            )
+        };
+    }
+    macro_rules! nested_values {
+        ($ty:ty) => {{
+            let value = array.as_any().downcast_ref::<$ty>().ok_or_else(budget_error)?.value(row);
+            for index in 0..value.len() {
+                charge_value(value.as_ref(), index, depth + 1, total, limit)?;
+            }
+            Ok(())
+        }};
+    }
+    match array.data_type() {
+        DataType::Utf8 => scalar_bytes!(StringArray),
+        DataType::LargeUtf8 => scalar_bytes!(LargeStringArray),
+        DataType::Utf8View => scalar_bytes!(StringViewArray),
+        DataType::Binary => scalar_bytes!(BinaryArray),
+        DataType::LargeBinary => scalar_bytes!(LargeBinaryArray),
+        DataType::BinaryView => scalar_bytes!(BinaryViewArray),
+        DataType::Boolean | DataType::Null => Ok(()),
+        DataType::FixedSizeBinary(size) => add_budget(total, usize::try_from(*size).map_err(|_| budget_error())?, limit),
+        DataType::List(_) => nested_values!(ListArray),
+        DataType::LargeList(_) => nested_values!(LargeListArray),
+        DataType::ListView(_) => nested_values!(ListViewArray),
+        DataType::LargeListView(_) => nested_values!(LargeListViewArray),
+        DataType::FixedSizeList(_, _) => nested_values!(FixedSizeListArray),
+        DataType::Map(_, _) => {
+            let value = array.as_any().downcast_ref::<MapArray>().ok_or_else(budget_error)?.value(row);
+            for index in 0..value.len() {
+                charge_value(&value, index, depth + 1, total, limit)?;
+            }
+            Ok(())
+        }
+        DataType::Struct(_) => {
+            for column in array
+                .as_any()
+                .downcast_ref::<StructArray>()
+                .ok_or_else(budget_error)?
+                .columns()
+            {
+                charge_value(column.as_ref(), row, depth + 1, total, limit)?;
+            }
+            Ok(())
+        }
+        DataType::Dictionary(_, _) => downcast_dictionary_array!(array => {
+            if let Some(key) = array.key(row) { charge_value(array.values().as_ref(), key, depth + 1, total, limit)?; }
+            Ok(())
+        }, _ => Err(budget_error())),
+        data_type => add_budget(total, data_type.primitive_width().ok_or_else(budget_error)?, limit),
+    }
+}
+
+struct BoundedOutput {
+    bytes: Vec<u8>,
+    limit: usize,
+}
+impl Write for BoundedOutput {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        let Some(required) = self.bytes.len().checked_add(bytes.len()).filter(|size| *size <= self.limit) else {
+            return Err(std::io::Error::other("compaction output exceeds the server size limit"));
+        };
+        if required > self.bytes.capacity() {
+            // Parquet writes successive pages and buffered footer fragments.
+            // Amortize their copies without allocating past the output budget.
+            let capacity = required.max(self.bytes.capacity().saturating_mul(2)).min(self.limit);
+            self.bytes
+                .try_reserve_exact(capacity - self.bytes.len())
+                .map_err(std::io::Error::other)?;
+        }
+        self.bytes.extend_from_slice(bytes);
+        Ok(bytes.len())
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+pub(crate) fn compact_parquet_data_files(input_files: Vec<(String, Vec<u8>)>) -> TableCatalogStoreResult<CompactedParquetFile> {
+    compact_parquet_with_limits(input_files, TABLE_COMPACTION_MAX_DECODED_BYTES, COMPACTION_WRITE_BYTES)
+}
+
+fn compact_parquet_with_limits(
+    input_files: Vec<(String, Vec<u8>)>,
+    limit: usize,
+    write_limit: usize,
+) -> TableCatalogStoreResult<CompactedParquetFile> {
+    compact_parquet_with_budgets(input_files, limit, write_limit, COMPACTION_MAX_ROW_GROUPS, COMPACTION_ROW_GROUP_ROWS)
+}
+
+fn compact_parquet_with_budgets(
+    input_files: Vec<(String, Vec<u8>)>,
+    limit: usize,
+    write_limit: usize,
+    group_limit: usize,
+    row_group_rows: usize,
+) -> TableCatalogStoreResult<CompactedParquetFile> {
+    let mut schema: Option<SchemaRef> = None;
+    let mut declared = 0;
+    let mut stored = 0;
+    let mut files = Vec::new();
+    let mut leaf_columns = 0;
+    // Validate every footer before decoding any page or creating output.
+    for (location, data) in input_files {
+        add_budget(&mut stored, data.len(), TABLE_COMPACTION_MAX_STORED_BYTES)?;
+        let data = Bytes::from(data);
+        let builder = ParquetRecordBatchReaderBuilder::try_new(data.clone())
+            .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read compaction parquet {location}: {err}")))?;
+        let file_schema = builder.schema().clone();
+        leaf_columns = leaf_columns.max(builder.metadata().file_metadata().schema_descr().num_columns());
+        match &schema {
+            Some(expected) if expected.as_ref() != file_schema.as_ref() => {
+                return Err(TableCatalogStoreError::Invalid("compaction input parquet schemas must match".to_owned()));
+            }
+            None => schema = Some(file_schema),
+            _ => {}
+        }
+        for group in builder.metadata().row_groups() {
+            let mut columns = 0;
+            for column in group.columns() {
+                add_budget(
+                    &mut columns,
+                    usize::try_from(column.uncompressed_size()).map_err(|_| budget_error())?,
+                    limit,
+                )?;
+            }
+            let group_bytes = usize::try_from(group.total_byte_size()).map_err(|_| budget_error())?;
+            add_budget(&mut declared, group_bytes.max(columns), limit)?;
+        }
+        files.push((location, data));
+    }
+    let schema = schema
+        .ok_or_else(|| TableCatalogStoreError::Invalid("compaction requires at least one parquet input file".to_owned()))?;
+    let read_schema = view_schema(&schema)?;
+    let group_limit = group_limit.min(COMPACTION_MAX_COLUMN_CHUNKS / leaf_columns.max(1));
+    // These bounds include metadata retained until the footer is serialized.
+    // Page/offset indexes would add a separate, page-count-dependent inventory.
+    let mut properties = WriterProperties::builder()
+        .set_max_row_group_row_count(Some(row_group_rows))
+        .set_statistics_enabled(EnabledStatistics::Chunk)
+        .set_statistics_truncate_length(Some(64))
+        .set_offset_index_disabled(true)
+        .build();
+    // The writer requires nested array types to match its input schema exactly.
+    // Views share the same Parquet physical/logical types; retain the original
+    // Arrow schema in the footer so readers recover its types and field metadata.
+    // Do not cast sliced lists/maps: Arrow casts their entire child buffers.
+    add_encoded_arrow_schema_to_metadata(&schema, &mut properties);
+    let options = ArrowWriterOptions::new()
+        .with_properties(properties)
+        .with_skip_arrow_metadata(true);
+    let mut writer = ArrowWriter::try_new_with_options(
+        BoundedOutput {
+            bytes: Vec::new(),
+            limit,
+        },
+        read_schema.clone(),
+        options,
+    )
+    .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to build compaction writer: {err}")))?;
+    let mut record_count = 0_u64;
+    let mut decoded = 0;
+    for (location, data) in files {
+        let options = ArrowReaderOptions::new().with_schema(read_schema.clone());
+        let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(data, options)
+            .and_then(|builder| builder.with_batch_size(COMPACTION_BATCH_ROWS).build())
+            .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to build compaction reader {location}: {err}")))?;
+        for batch in reader {
+            let batch = batch
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to decode compaction batch {location}: {err}")))?;
+            add_budget(&mut decoded, batch.get_array_memory_size(), limit)?;
+            let mut start = 0;
+            while start < batch.num_rows() {
+                let mut end = start;
+                let mut expanded = 0_usize;
+                let remaining_rows = row_group_rows.saturating_sub(writer.in_progress_rows());
+                while end < batch.num_rows() && end - start < remaining_rows {
+                    let mut row_bytes = 0;
+                    for column in batch.columns() {
+                        charge_value(column.as_ref(), end, 0, &mut row_bytes, write_limit)?;
+                    }
+                    if expanded.checked_add(row_bytes).is_none_or(|size| size > write_limit) {
+                        break;
+                    }
+                    expanded += row_bytes;
+                    end += 1;
+                }
+                if end == start {
+                    return Err(budget_error());
+                }
+                if writer
+                    .memory_size()
+                    .checked_add(expanded)
+                    .is_none_or(|size| size > write_limit)
+                {
+                    writer.flush().map_err(|err| {
+                        TableCatalogStoreError::Invalid(format!("failed to flush bounded compaction output: {err}"))
+                    })?;
+                }
+                if writer.flushed_row_groups().len() >= group_limit {
+                    return Err(TableCatalogStoreError::Invalid(
+                        "compaction output row-group metadata exceeds the server limit".to_owned(),
+                    ));
+                }
+                writer.write(&batch.slice(start, end - start)).map_err(|err| {
+                    TableCatalogStoreError::Invalid(format!("failed to write bounded compaction output: {err}"))
+                })?;
+                record_count = record_count.checked_add((end - start) as u64).ok_or_else(budget_error)?;
+                start = end;
+            }
+        }
+    }
+    let output = writer
+        .into_inner()
+        .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to finalize bounded compaction output: {err}")))?;
+    Ok(CompactedParquetFile {
+        data: output.bytes,
+        record_count,
+    })
+}
+
+pub(crate) async fn run_compaction_worker(
+    input_files: Vec<(String, Vec<u8>)>,
+    admission: Arc<OwnedSemaphorePermit>,
+) -> TableCatalogStoreResult<CompactedParquetFile> {
+    #[cfg(test)]
+    let barrier = COMPACTION_WORKER_BARRIER.try_with(Arc::clone).ok();
+    tokio::task::spawn_blocking(move || {
+        let _admission = admission;
+        #[cfg(test)]
+        if let Some(barrier) = barrier {
+            barrier.entered.notify_one();
+            let mut released = barrier.released.lock().unwrap();
+            while !*released {
+                released = barrier.release.wait(released).unwrap();
+            }
+        }
+        compact_parquet_data_files(input_files)
+    })
+    .await
+    .map_err(|_| TableCatalogStoreError::Internal("compaction decoder worker failed".to_owned()))?
+}
+
+pub(crate) async fn run_compaction_output<F>(operation: F, admission: Arc<OwnedSemaphorePermit>) -> TableCatalogStoreResult<()>
+where
+    F: std::future::Future<Output = TableCatalogStoreResult<()>> + Send + 'static,
+{
+    let credentials = rustfs_keystone::KEYSTONE_CREDENTIALS.try_with(Clone::clone).ok().flatten();
+    let operation = rustfs_keystone::KEYSTONE_CREDENTIALS.scope(credentials, async move {
+        let _admission = admission;
+        operation.await
+    });
+    let operation = rustfs_io_metrics::s3_http_metrics::with_s3_http_operation_context(
+        rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_context(),
+        operation,
+    );
+    crate::storage_api::table::spawn_traced_join(operation)
+        .await
+        .map_err(|_| TableCatalogStoreError::Internal("compaction output worker failed".to_owned()))?
+}
+
+#[cfg(test)]
+struct WorkerBarrier {
+    entered: tokio::sync::Notify,
+    released: std::sync::Mutex<bool>,
+    release: std::sync::Condvar,
+}
+#[cfg(test)]
+tokio::task_local! { static COMPACTION_WORKER_BARRIER: Arc<WorkerBarrier>; }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use datafusion::arrow::record_batch::RecordBatch;
+    use datafusion::parquet::basic::{Compression, ZstdLevel};
+
+    fn parquet_strings(value: &str, rows: usize, dictionary: bool) -> Vec<u8> {
+        let schema = Arc::new(Schema::new_with_metadata(
+            vec![Field::new("value", DataType::Utf8, false)],
+            [("custom-key".to_owned(), "preserved".to_owned())].into_iter().collect(),
+        ));
+        let values = StringArray::from(vec![value; rows]);
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(values)]).unwrap();
+        let properties = WriterProperties::builder()
+            .set_dictionary_enabled(dictionary)
+            .set_compression(Compression::ZSTD(ZstdLevel::default()))
+            .build();
+        let mut writer = ArrowWriter::try_new(Vec::new(), schema, Some(properties)).unwrap();
+        writer.write(&batch).unwrap();
+        writer.into_inner().unwrap()
+    }
+
+    #[test]
+    fn compaction_rejects_compressed_expansion_before_decoding_pages() {
+        let file = parquet_strings(&"x".repeat(64 * 1024), 32, false);
+        assert!(file.len() < 16 * 1024, "fixture must really be highly compressed");
+        let error = compact_parquet_with_limits(vec![("bomb.parquet".to_owned(), file)], 128 * 1024, 64 * 1024)
+            .err()
+            .expect("footer must reject decoded expansion before allocation");
+        assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+    }
+
+    #[test]
+    fn compaction_dictionary_views_bound_each_writer_input_and_preserve_schema() {
+        let value = "x".repeat(16 * 1024);
+        let file = parquet_strings(&value, 32, true);
+        let original_schema = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(file.clone()))
+            .unwrap()
+            .schema()
+            .clone();
+        let output =
+            compact_parquet_with_limits(vec![("dictionary.parquet".to_owned(), file)], 2 * 1024 * 1024, 32 * 1024).unwrap();
+        assert_eq!(output.record_count, 32);
+        let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(output.data)).unwrap();
+        assert_eq!(builder.schema(), &original_schema);
+        assert!(builder.metadata().num_row_groups() <= 32);
+        for batch in builder.build().unwrap() {
+            let batch = batch.unwrap();
+            let values = batch.column(0).as_any().downcast_ref::<StringArray>().unwrap();
+            assert!(values.iter().all(|item| item == Some(value.as_str())));
+        }
+    }
+
+    #[test]
+    fn compaction_rejects_a_single_expanded_value_before_writer_allocation() {
+        let file = parquet_strings(&"x".repeat(4096), 16, true);
+        let error = compact_parquet_with_limits(vec![("single-row.parquet".to_owned(), file)], 1024 * 1024, 1024)
+            .err()
+            .expect("one expanded row cannot exceed the writer input budget");
+        assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+    }
+
+    #[test]
+    fn compaction_nested_views_keep_field_and_schema_metadata() {
+        use datafusion::arrow::array::{ListBuilder, StringBuilder};
+        let child = Arc::new(
+            Field::new("text", DataType::Utf8, false)
+                .with_metadata([("PARQUET:field_id".to_owned(), "2".to_owned())].into_iter().collect()),
+        );
+        let array = StructArray::from(vec![(child.clone(), Arc::new(StringArray::from(vec!["nested"; 4])) as Arc<dyn Array>)]);
+        let value = "nested".repeat(32);
+        let mut lists = ListBuilder::new(StringBuilder::new());
+        for _ in 0..4 {
+            lists.values().append_value(&value);
+            lists.append(true);
+        }
+        let lists = lists.finish();
+        let schema = Arc::new(Schema::new_with_metadata(
+            vec![
+                Field::new("outer", DataType::Struct(vec![child].into()), false),
+                Field::new("list", lists.data_type().clone(), false),
+            ],
+            [("source".to_owned(), "nested-compaction".to_owned())].into_iter().collect(),
+        ));
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(array), Arc::new(lists)]).unwrap();
+        let mut writer = ArrowWriter::try_new(Vec::new(), schema.clone(), None).unwrap();
+        writer.write(&batch).unwrap();
+        let file = writer.into_inner().unwrap();
+        let rejected = compact_parquet_with_limits(vec![("nested.parquet".to_owned(), file.clone())], 1024 * 1024, 200)
+            .err()
+            .expect("a nested row must fit before reaching the writer");
+        assert!(matches!(rejected, TableCatalogStoreError::Invalid(message) if message.contains("expanded-value budget")));
+        // Force slices within the decoded nested batch. They must not require
+        // expanding all values in the retained child buffer before writing.
+        let output = compact_parquet_with_limits(vec![("nested.parquet".to_owned(), file)], 1024 * 1024, 300).unwrap();
+        let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(output.data)).unwrap();
+        assert_eq!(builder.schema(), &schema);
+        let batches = builder.build().unwrap().collect::<Result<Vec<_>, _>>().unwrap();
+        assert_eq!(batches.iter().map(|batch| batch.num_rows()).sum::<usize>(), 4);
+        for batch in batches {
+            let outer = batch.column(0).as_any().downcast_ref::<StructArray>().unwrap();
+            let texts = outer.column(0).as_any().downcast_ref::<StringArray>().unwrap();
+            assert!(texts.iter().all(|text| text == Some("nested")));
+            let lists = batch.column(1).as_any().downcast_ref::<ListArray>().unwrap();
+            for row in 0..batch.num_rows() {
+                let values = lists.value(row);
+                let values = values.as_any().downcast_ref::<StringArray>().unwrap();
+                assert_eq!(values.len(), 1);
+                assert_eq!(values.value(0), value);
+            }
+        }
+    }
+
+    #[test]
+    fn compaction_arrow_dictionary_schema_round_trips_through_view_reader() {
+        use datafusion::arrow::array::{DictionaryArray, Int8Array};
+        use datafusion::arrow::datatypes::Int8Type;
+        let values = Arc::new(StringArray::from(vec!["dictionary-value".repeat(1024)]));
+        let array = DictionaryArray::<Int8Type>::try_new(Int8Array::from(vec![0; 16]), values).unwrap();
+        let field = Arc::new(Field::new("dictionary", array.data_type().clone(), false));
+        let array = Arc::new(array);
+        let nested = StructArray::from(vec![(field.clone(), array.clone() as Arc<dyn Array>)]);
+        let schema = Arc::new(Schema::new(vec![
+            field.as_ref().clone(),
+            Field::new("nested", nested.data_type().clone(), false),
+        ]));
+        let batch = RecordBatch::try_new(schema.clone(), vec![array, Arc::new(nested)]).unwrap();
+        let mut writer = ArrowWriter::try_new(Vec::new(), schema, None).unwrap();
+        writer.write(&batch).unwrap();
+        let file = writer.into_inner().unwrap();
+        let expected = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(file.clone()))
+            .unwrap()
+            .schema()
+            .clone();
+        let output =
+            compact_parquet_with_limits(vec![("arrow-dictionary.parquet".to_owned(), file)], 1024 * 1024, 40 * 1024).unwrap();
+        assert_eq!(output.record_count, 16);
+        let reader = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(output.data)).unwrap();
+        assert_eq!(reader.schema(), &expected);
+        let batches = reader.build().unwrap().collect::<Result<Vec<_>, _>>().unwrap();
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 16);
+        for batch in batches {
+            let nested = batch.column(1).as_any().downcast_ref::<StructArray>().unwrap();
+            for column in [batch.column(0), nested.column(0)] {
+                let dictionary = column.as_any().downcast_ref::<DictionaryArray<Int8Type>>().unwrap();
+                let values = dictionary.values().as_any().downcast_ref::<StringArray>().unwrap();
+                for row in 0..dictionary.len() {
+                    assert_eq!(values.value(dictionary.key(row).unwrap()), "dictionary-value".repeat(1024));
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn compaction_retained_row_group_metadata_is_bounded_before_another_group() {
+        let file = parquet_strings("", 1024, true);
+        let output =
+            compact_parquet_with_budgets(vec![("rle.parquet".to_owned(), file)], 1024 * 1024, 64 * 1024, 2, 512).unwrap();
+        let reader = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(output.data)).unwrap();
+        assert_eq!(
+            reader.metadata().num_row_groups(),
+            2,
+            "256-row input batches must not force separate row groups"
+        );
+        assert_eq!(output.record_count, 1024);
+        let file = parquet_strings("", 1025, true);
+        let error = compact_parquet_with_budgets(vec![("rle.parquet".to_owned(), file)], 1024 * 1024, 64 * 1024, 2, 512)
+            .err()
+            .expect("a third row group must not allocate retained metadata");
+        assert!(matches!(error, TableCatalogStoreError::Invalid(message) if message.contains("row-group metadata")));
+    }
+
+    #[test]
+    fn compaction_output_sink_rejects_before_growing() {
+        let mut output = BoundedOutput {
+            bytes: Vec::new(),
+            limit: 4,
+        };
+        output.write_all(b"1234").unwrap();
+        let capacity = output.bytes.capacity();
+        assert!(output.write_all(b"5").is_err());
+        assert_eq!(&output.bytes, b"1234");
+        assert_eq!(output.bytes.capacity(), capacity, "rejected writes must not allocate");
+    }
+
+    #[test]
+    fn compaction_output_sink_amortizes_small_writes_within_the_capacity_limit() {
+        // Include a non-power-of-two cap so the last growth must stop at the
+        // limit instead of using Vec's unconstrained doubling policy.
+        for chunk_size in [1, 7, 64] {
+            let mut output = BoundedOutput {
+                bytes: Vec::new(),
+                limit: 4099,
+            };
+            let mut growths = 0;
+            let mut capacity = 0;
+            while output.bytes.len() < output.limit {
+                let count = chunk_size.min(output.limit - output.bytes.len());
+                output.write_all(&[42; 64][..count]).expect("bounded fragment");
+                if output.bytes.capacity() != capacity {
+                    growths += 1;
+                    capacity = output.bytes.capacity();
+                }
+                assert!(capacity >= output.bytes.len() && capacity <= output.limit);
+            }
+            assert!(growths <= 14, "small writes must require logarithmic capacity growth, got {growths}");
+            assert_eq!(output.bytes, vec![42; output.limit]);
+            assert!(output.write_all(b"x").is_err());
+            assert_eq!(output.bytes.capacity(), capacity);
+            assert_eq!(output.bytes.len(), output.limit);
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(table_compaction)]
+    async fn compaction_output_worker_preserves_keystone_authorization_context() {
+        let credentials = rustfs_credentials::Credentials {
+            access_key: "keystone:compaction".to_owned(),
+            ..Default::default()
+        };
+        let admission = try_admit_compaction().unwrap();
+        rustfs_keystone::KEYSTONE_CREDENTIALS
+            .scope(Some(credentials), async {
+                run_compaction_output(
+                    async {
+                        assert_eq!(
+                            rustfs_keystone::KEYSTONE_CREDENTIALS
+                                .with(|credentials| credentials.as_ref().map(|value| value.access_key.clone())),
+                            Some("keystone:compaction".to_owned())
+                        );
+                        Ok(())
+                    },
+                    admission,
+                )
+                .await
+                .unwrap();
+            })
+            .await;
+        assert!(rustfs_keystone::KEYSTONE_CREDENTIALS.try_with(Clone::clone).is_err());
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(table_compaction)]
+    async fn compaction_cancelled_caller_keeps_actual_decoder_worker_admitted() {
+        let barrier = Arc::new(WorkerBarrier {
+            entered: tokio::sync::Notify::new(),
+            released: std::sync::Mutex::new(false),
+            release: std::sync::Condvar::new(),
+        });
+        let admission = try_admit_compaction().unwrap();
+        let file = parquet_strings("valid", 4, true);
+        let scope_barrier = barrier.clone();
+        let caller = tokio::spawn(COMPACTION_WORKER_BARRIER.scope(scope_barrier, async move {
+            run_compaction_worker(vec![("valid.parquet".to_owned(), file)], admission).await
+        }));
+        barrier.entered.notified().await;
+        caller.abort();
+        let _ = caller.await;
+        assert!(matches!(try_admit_compaction(), Err(TableCatalogStoreError::Unavailable(_))));
+        *barrier.released.lock().unwrap() = true;
+        barrier.release.notify_one();
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
+            loop {
+                if try_admit_compaction().is_ok() {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("worker completion must return the permit");
+    }
+}

rustfs/src/table_catalog/maintenance/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod compaction;
 mod config;
 mod periodic;
 mod periodic_runtime;
@@ -19,6 +20,7 @@ mod planner;
 mod recovery;
 mod worker;
 
+pub(crate) use compaction::{run_compaction_output, run_compaction_worker, try_admit_compaction};
 pub(crate) use config::*;
 pub(crate) use periodic_runtime::{PeriodicMaintenanceConfig, PeriodicMaintenanceRuntime};
 pub(crate) use planner::*;

rustfs/src/table_catalog/maintenance/planner.rs

@@ -402,10 +402,10 @@ where
                     "compaction data file must be inside the table data directory".to_string(),
                 ));
             }
-            let Some(data_object) = backend.read_object(table_bucket, &data_key).await? else {
+            let Some(data_metadata) = backend.object_metadata(table_bucket, &data_key).await? else {
                 return Err(TableCatalogStoreError::NotFound(format!("compaction data file {data_key}")));
             };
-            let size_bytes = u64::try_from(data_object.data.len()).unwrap_or(u64::MAX);
+            let size_bytes = data_metadata.size;
             if size_bytes <= config.small_file_threshold_bytes {
                 planning.candidates.push(TableCompactionDataFileCandidate {
                     rewrite_prefix: compaction_data_file_rewrite_prefix_for_entry(entry, warehouse_object_prefix, &data_key)
@@ -447,7 +447,7 @@ where
         ));
     }
 
-    let object_exists = backend.read_object(table_bucket, &delete_key).await?.is_some();
+    let object_exists = backend.object_metadata(table_bucket, &delete_key).await?.is_some();
     planning.status = TableRowLevelMaintenancePlanningStatus::ManualReviewRequired;
     planning.delete_file_count = planning.delete_file_count.saturating_add(1);
     planning.manual_review_count = planning.manual_review_count.saturating_add(1);
@@ -519,6 +519,7 @@ pub(crate) async fn compaction_current_data_files<B>(
     table: &IdentifierSegment,
     entry: &TableEntry,
     current_metadata: &serde_json::Value,
+    admission: &Arc<tokio::sync::OwnedSemaphorePermit>,
 ) -> TableCatalogStoreResult<Vec<CompactedDataFile>>
 where
     B: TableCatalogObjectBackend,
@@ -603,7 +604,7 @@ where
                     "compaction data file must be inside the table data directory".to_string(),
                 ));
             }
-            let Some(data_object) = backend.read_object(table_bucket, &data_key).await? else {
+            let Some(data_metadata) = backend.object_metadata(table_bucket, &data_key).await? else {
                 return Err(TableCatalogStoreError::NotFound(format!("compaction data file {data_key}")));
             };
             let snapshot_id = reference
@@ -625,14 +626,24 @@ where
                     TableCatalogStoreError::Invalid("compaction manifest data file missing file sequence number".to_string())
                 })?;
             data_files.push(CompactedDataFile {
-                object_key: data_key,
+                object_key: data_key.clone(),
                 file_path: reference.location,
-                file_size_bytes: reference
-                    .file_size_bytes
-                    .unwrap_or_else(|| u64::try_from(data_object.data.len()).unwrap_or(u64::MAX)),
+                file_size_bytes: reference.file_size_bytes.unwrap_or(data_metadata.size),
                 record_count: match reference.record_count {
                     Some(record_count) => record_count,
-                    None => parquet_record_count(&data_object.data)?,
+                    None => {
+                        let object = backend
+                            .read_object_limited(table_bucket, &data_key, TABLE_COMPACTION_MAX_STORED_BYTES)
+                            .await?
+                            .ok_or_else(|| TableCatalogStoreError::NotFound(format!("compaction data file {data_key}")))?;
+                        let admission = Arc::clone(admission);
+                        tokio::task::spawn_blocking(move || {
+                            let _admission = admission;
+                            parquet_record_count(object.data)
+                        })
+                        .await
+                        .map_err(|_| TableCatalogStoreError::Internal("compaction footer worker failed".to_owned()))??
+                    }
                 },
                 partition_spec_id: manifest_reference.partition_spec_id.unwrap_or(0),
                 partition: reference.partition,
@@ -836,58 +847,6 @@ pub(crate) fn compaction_manifest_partition_spec_id(data_files: &[CompactedDataF
     Ok(partition_spec_id)
 }
 
-pub(crate) fn compact_parquet_data_files(input_files: &[(String, Vec<u8>)]) -> TableCatalogStoreResult<CompactedParquetFile> {
-    let mut schema: Option<SchemaRef> = None;
-    let mut batches = Vec::new();
-    let mut record_count = 0_u64;
-
-    for (location, data) in input_files {
-        let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(data.clone())).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to read compaction input parquet {location}: {err}"))
-        })?;
-        let file_schema = builder.schema().clone();
-        match schema.as_ref() {
-            Some(expected_schema) if expected_schema.as_ref() != file_schema.as_ref() => {
-                return Err(TableCatalogStoreError::Invalid("compaction input parquet schemas must match".to_string()));
-            }
-            Some(_) => {}
-            None => schema = Some(file_schema),
-        }
-
-        let reader = builder.build().map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to build compaction parquet reader {location}: {err}"))
-        })?;
-        for batch in reader {
-            let batch = batch.map_err(|err| {
-                TableCatalogStoreError::Invalid(format!("failed to read compaction parquet batch {location}: {err}"))
-            })?;
-            record_count = record_count.saturating_add(u64::try_from(batch.num_rows()).unwrap_or(u64::MAX));
-            batches.push(batch);
-        }
-    }
-
-    let Some(schema) = schema else {
-        return Err(TableCatalogStoreError::Invalid(
-            "compaction requires at least one parquet input file".to_string(),
-        ));
-    };
-    let mut data = Vec::new();
-    {
-        let mut writer = ArrowWriter::try_new(&mut data, schema, None)
-            .map_err(|err| TableCatalogStoreError::Internal(format!("failed to build compaction parquet writer: {err}")))?;
-        for batch in batches {
-            writer
-                .write(&batch)
-                .map_err(|err| TableCatalogStoreError::Internal(format!("failed to write compaction parquet batch: {err}")))?;
-        }
-        writer
-            .close()
-            .map_err(|err| TableCatalogStoreError::Internal(format!("failed to close compaction parquet writer: {err}")))?;
-    }
-
-    Ok(CompactedParquetFile { data, record_count })
-}
-
 fn validate_compaction_manifest_entry_status(entry_status: Option<i32>) -> TableCatalogStoreResult<()> {
     match entry_status {
         Some(0 | 1) => Ok(()),
@@ -901,8 +860,8 @@ fn validate_compaction_manifest_entry_status(entry_status: Option<i32>) -> Table
     }
 }
 
-fn parquet_record_count(data: &[u8]) -> TableCatalogStoreResult<u64> {
-    let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::copy_from_slice(data))
+fn parquet_record_count(data: Vec<u8>) -> TableCatalogStoreResult<u64> {
+    let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(data))
         .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read compaction parquet metadata: {err}")))?;
     u64::try_from(builder.metadata().file_metadata().num_rows())
         .map_err(|_| TableCatalogStoreError::Invalid("compaction parquet record count must not be negative".to_string()))

rustfs/src/table_catalog/maintenance/recovery.rs

@@ -72,6 +72,22 @@ fn insert_referenced_object_report(
     report.reasons.insert(reason);
 }
 
+pub(crate) fn maintenance_cleanup_roots(entry: &TableEntry) -> TableCatalogStoreResult<Vec<String>> {
+    let metadata_dir = table_metadata_dir_path_for_entry(entry)?;
+    let table_root = metadata_dir
+        .strip_suffix(&format!("/{METADATA_DIR}"))
+        .ok_or_else(|| TableCatalogStoreError::Invalid("maintenance metadata directory has no table root".to_string()))?;
+    let warehouse = table_warehouse_object_prefix(entry)?;
+    Ok(vec![
+        format!("{metadata_dir}/"),
+        format!("{table_root}/{DATA_DIR}/"),
+        format!("{table_root}/{DELETE_DIR}/"),
+        format!("{warehouse}{METADATA_DIR}/"),
+        format!("{warehouse}{DATA_DIR}/"),
+        format!("{warehouse}{DELETE_DIR}/"),
+    ])
+}
+
 #[derive(Clone, Copy)]
 enum TableMaintenanceObjectScope<'a> {
     Logical {
@@ -795,7 +811,7 @@ where
             table_metadata_maintenance_reason_for_object_kind(&object_kind),
             TableMetadataMaintenanceReason::NoCurrentReachability,
         ]);
-        let state = match backend.read_object(table_bucket, &object_location).await? {
+        let state = match backend.object_metadata(table_bucket, &object_location).await? {
             Some(object) if metadata_candidate_is_past_safety_window(object.mod_time, now) => {
                 reasons.insert(TableMetadataMaintenanceReason::SafetyWindowSatisfied);
                 cleanup_candidate_locations.push(object_location.clone());

rustfs/src/table_catalog/mod.rs

@@ -130,6 +130,8 @@ const TABLE_COMMIT_MAX_AVRO_BYTES: usize = 512 * 1024 * 1024;
 const TABLE_COMMIT_MAX_FILE_REFERENCES: usize = 1_000_000;
 const TABLE_COMMIT_MAX_STATISTICS_OBJECTS: usize = 1_024;
 const TABLE_COMMIT_MAX_STATISTICS_BYTES: usize = 512 * 1024 * 1024;
+const TABLE_COMPACTION_MAX_STORED_BYTES: usize = 512 * 1024 * 1024;
+const TABLE_COMPACTION_MAX_DECODED_BYTES: usize = 512 * 1024 * 1024;
 const TABLE_STATISTICS_FILE_MAX_SIZE: usize = 128 * 1024 * 1024;
 pub(crate) const TABLE_COMMIT_OBJECT_VALIDATION_CONCURRENCY: usize = 16;
 pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;

rustfs/src/table_catalog/store/cleanup.rs

@@ -131,6 +131,7 @@ impl<B: TableCatalogObjectBackend> ObjectTableCatalogStore<B> {
         let bucket_path = self.paths.table_bucket_entry_path(bucket);
         let index_state_path = self.paths.warehouse_index_state_path(bucket);
         let index_prefix = self.paths.warehouse_index_entries_prefix(bucket);
+        let directory_prefix = self.paths.table_directory_prefix(bucket);
         let mut indexes = Vec::new();
         for object in catalog_objects {
             if object.delete_marker
@@ -173,6 +174,14 @@ impl<B: TableCatalogObjectBackend> ObjectTableCatalogStore<B> {
                     return Err(cleanup_conflict("cleanup requires the default table warehouse location"));
                 }
                 etag
+            } else if object.key.starts_with(&directory_prefix) {
+                // Registration/rename share the held bucket publication fence.
+                // Remove only the exact immutable candidate of a vanished entry.
+                let (target, etag) = self.directory_candidate_target(bucket, &object.key).await?;
+                if self.backend.object_exists(self.catalog_bucket(), &target).await? {
+                    return Err(cleanup_conflict("table directory candidate references a retained table entry"));
+                }
+                etag
             } else {
                 return Err(cleanup_conflict("table bucket contains namespaces or retained catalog resources"));
             };

rustfs/src/table_catalog/store/migration.rs

@@ -23,6 +23,20 @@ use super::strong::{
 };
 use super::*;
 
+async fn run_owned_backing_migration<T: Send + 'static>(
+    operation: impl std::future::Future<Output = TableCatalogStoreResult<T>> + Send + 'static,
+) -> TableCatalogStoreResult<T> {
+    let credentials = rustfs_keystone::KEYSTONE_CREDENTIALS.try_with(Clone::clone).ok().flatten();
+    let operation = rustfs_keystone::KEYSTONE_CREDENTIALS.scope(credentials, operation);
+    let operation = rustfs_io_metrics::s3_http_metrics::with_s3_http_operation_context(
+        rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_context(),
+        operation,
+    );
+    crate::storage_api::table::spawn_traced_join(operation)
+        .await
+        .map_err(|_| TableCatalogStoreError::Internal("catalog migration worker failed".to_owned()))?
+}
+
 pub(super) fn table_catalog_backing_manifest(
     paths: &TableCatalogObjectPaths,
     namespace: &Namespace,
@@ -307,6 +321,11 @@ where
         &self,
         table_bucket: &str,
     ) -> TableCatalogStoreResult<()> {
+        // The directory upgrade keeps legacy entries authoritative and requires
+        // every compatible writer to publish its candidate before the entry.
+        if self.read_table_directory_state(table_bucket).await?.is_some() {
+            return Ok(());
+        }
         if self.read_backing_migration_fence(table_bucket).await?.is_some() {
             return Err(TableCatalogStoreError::Conflict(format!(
                 "object-backed catalog writes are fenced while table bucket {table_bucket} is prepared for durable strong cutover"
@@ -318,6 +337,7 @@ where
     async fn ensure_global_backing_migration_fence(
         &self,
         fence_path: &str,
+        mutation_fence: &TableCatalogObjectMutationFence,
     ) -> TableCatalogStoreResult<TableCatalogBackingMigrationGlobalFence> {
         if let Some((fence, _)) = self
             .read_entry::<TableCatalogBackingMigrationGlobalFence>(self.catalog_bucket(), fence_path)
@@ -330,13 +350,17 @@ where
             version: TABLE_CATALOG_MIGRATION_VERSION,
             migration_id: Uuid::new_v4().to_string(),
         };
-        self.write_entry(self.catalog_bucket(), fence_path, &fence, TableCatalogPutPrecondition::IfAbsent)
+        self.write_migration_entry(fence_path, &fence, TableCatalogPutPrecondition::IfAbsent, mutation_fence)
             .await?;
         Ok(fence)
     }
 
-    async fn clear_global_backing_migration_fence_if_unused(&self, fence_path: &str) -> TableCatalogStoreResult<()> {
-        let Some((fence, _)) = self
+    async fn clear_global_backing_migration_fence_if_unused(
+        &self,
+        fence_path: &str,
+        mutation_fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        let Some((fence, etag)) = self
             .read_entry::<TableCatalogBackingMigrationGlobalFence>(self.catalog_bucket(), fence_path)
             .await?
         else {
@@ -353,7 +377,24 @@ where
         {
             return Ok(());
         }
-        self.backend.delete_object(self.catalog_bucket(), fence_path).await
+        let etag = etag.ok_or_else(|| TableCatalogStoreError::Conflict("migration fence has no ETag".to_owned()))?;
+        self.backend
+            .delete_object_if_match_fenced(self.catalog_bucket(), fence_path, &etag, mutation_fence)
+            .await
+    }
+
+    async fn write_migration_entry<T: Serialize>(
+        &self,
+        path: &str,
+        entry: &T,
+        precondition: TableCatalogPutPrecondition,
+        fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        let data = serde_json::to_vec(entry)
+            .map_err(|err| TableCatalogStoreError::Internal(format!("cannot serialize migration entry: {err}")))?;
+        self.backend
+            .put_object_fenced(self.catalog_bucket(), path, data, precondition, fence)
+            .await
     }
 
     async fn collect_bucket_snapshot_with_locks(
@@ -907,6 +948,16 @@ where
     pub(crate) async fn materialize_durable_strong_backing_migration(
         &self,
         table_bucket: &str,
+    ) -> TableCatalogStoreResult<TableCatalogBackingMigrationExecutionReport> {
+        let store = self.clone();
+        let table_bucket = table_bucket.to_owned();
+        run_owned_backing_migration(async move { store.materialize_durable_strong_backing_migration_owned(&table_bucket).await })
+            .await
+    }
+
+    async fn materialize_durable_strong_backing_migration_owned(
+        &self,
+        table_bucket: &str,
     ) -> TableCatalogStoreResult<TableCatalogBackingMigrationExecutionReport> {
         let fence_path = self.paths.backing_migration_fence_path(table_bucket);
         let fence_lock_path = self.paths.backing_migration_fence_lock_path(table_bucket);
@@ -929,6 +980,10 @@ where
             .backend
             .acquire_write_lock(self.catalog_bucket(), &fence_lock_path)
             .await?;
+        let mut mutation_fence = TableCatalogObjectMutationFence::from_signals([
+            _global_fence_guard.lock_lost_signal(),
+            _fence_guard.lock_lost_signal(),
+        ]);
         let existing_fence = self.read_backing_migration_fence(table_bucket).await?;
         let strong_store = StrongTableCatalogStore::new(self.backend.clone());
         if let Some((fence, _)) = existing_fence.as_ref()
@@ -955,6 +1010,10 @@ where
         let source = self
             .collect_bucket_snapshot_with_locks(table_bucket, &mut source_guards)
             .await?;
+        mutation_fence
+            .signals
+            .extend(source_guards.iter().filter_map(TableCatalogLockGuard::lock_lost_signal));
+        mutation_fence.ensure_held()?;
         let source_fingerprint = table_catalog_bucket_snapshot_fingerprint(&source)?;
         if let Some((fence, _)) = existing_fence.as_ref()
             && fence.status == TableCatalogBackingMigrationFenceStatus::Materialized
@@ -965,7 +1024,8 @@ where
             )));
         }
 
-        self.ensure_global_backing_migration_fence(&global_fence_path).await?;
+        self.ensure_global_backing_migration_fence(&global_fence_path, &mutation_fence)
+            .await?;
         let (target_fingerprint, target_snapshot_etag) = Self::observe_durable_strong_migration_target(
             &strong_store,
             table_bucket,
@@ -985,12 +1045,14 @@ where
                 source_fingerprint: None,
                 target_snapshot_etag,
             };
-            self.write_entry(self.catalog_bucket(), &fence_path, &fence, TableCatalogPutPrecondition::IfAbsent)
+            self.write_migration_entry(&fence_path, &fence, TableCatalogPutPrecondition::IfAbsent, &mutation_fence)
                 .await?;
             (fence.migration_id, target_bucket_existed)
         };
 
-        let (target_snapshot_etag, created) = strong_store.materialize_bucket_snapshot(source.clone()).await?;
+        let (target_snapshot_etag, created) = strong_store
+            .materialize_bucket_snapshot_with_fence(source.clone(), &mutation_fence)
+            .await?;
         let completed_fence = TableCatalogBackingMigrationFence {
             version: TABLE_CATALOG_MIGRATION_VERSION,
             table_bucket: table_bucket.to_string(),
@@ -1000,7 +1062,7 @@ where
             source_fingerprint: Some(source_fingerprint.clone()),
             target_snapshot_etag: Some(target_snapshot_etag.clone()),
         };
-        self.write_entry(self.catalog_bucket(), &fence_path, &completed_fence, TableCatalogPutPrecondition::Any)
+        self.write_migration_entry(&fence_path, &completed_fence, TableCatalogPutPrecondition::Any, &mutation_fence)
             .await?;
 
         drop(source_guards);
@@ -1030,6 +1092,15 @@ where
     pub(crate) async fn cancel_durable_strong_backing_migration(
         &self,
         table_bucket: &str,
+    ) -> TableCatalogStoreResult<TableCatalogBackingMigrationCancelReport> {
+        let store = self.clone();
+        let table_bucket = table_bucket.to_owned();
+        run_owned_backing_migration(async move { store.cancel_durable_strong_backing_migration_owned(&table_bucket).await }).await
+    }
+
+    async fn cancel_durable_strong_backing_migration_owned(
+        &self,
+        table_bucket: &str,
     ) -> TableCatalogStoreResult<TableCatalogBackingMigrationCancelReport> {
         let fence_path = self.paths.backing_migration_fence_path(table_bucket);
         let fence_lock_path = self.paths.backing_migration_fence_lock_path(table_bucket);
@@ -1043,8 +1114,12 @@ where
             .backend
             .acquire_write_lock(self.catalog_bucket(), &fence_lock_path)
             .await?;
-        let Some((fence, _)) = self.read_backing_migration_fence(table_bucket).await? else {
-            self.clear_global_backing_migration_fence_if_unused(&global_fence_path)
+        let mut mutation_fence = TableCatalogObjectMutationFence::from_signals([
+            _global_fence_guard.lock_lost_signal(),
+            _fence_guard.lock_lost_signal(),
+        ]);
+        let Some((fence, fence_etag)) = self.read_backing_migration_fence(table_bucket).await? else {
+            self.clear_global_backing_migration_fence_if_unused(&global_fence_path, &mutation_fence)
                 .await?;
             return Ok(TableCatalogBackingMigrationCancelReport {
                 table_bucket: table_bucket.to_string(),
@@ -1052,12 +1127,17 @@ where
                 object_backed_writes_fenced: false,
             });
         };
-        self.ensure_global_backing_migration_fence(&global_fence_path).await?;
+        self.ensure_global_backing_migration_fence(&global_fence_path, &mutation_fence)
+            .await?;
 
         let mut source_guards = Vec::new();
         let source = self
             .collect_bucket_snapshot_with_locks(table_bucket, &mut source_guards)
             .await?;
+        mutation_fence
+            .signals
+            .extend(source_guards.iter().filter_map(TableCatalogLockGuard::lock_lost_signal));
+        mutation_fence.ensure_held()?;
         let source_fingerprint = table_catalog_bucket_snapshot_fingerprint(&source)?;
         if fence.status == TableCatalogBackingMigrationFenceStatus::Materialized
             && fence.source_fingerprint.as_deref() != Some(source_fingerprint.as_str())
@@ -1086,11 +1166,14 @@ where
         }
         if !fence.target_bucket_existed {
             strong_store
-                .remove_bucket_snapshot_if_unchanged(table_bucket, &source_fingerprint)
+                .remove_bucket_snapshot_if_unchanged_with_fence(table_bucket, &source_fingerprint, &mutation_fence)
                 .await?;
         }
-        self.backend.delete_object(self.catalog_bucket(), &fence_path).await?;
-        self.clear_global_backing_migration_fence_if_unused(&global_fence_path)
+        let fence_etag = fence_etag.ok_or_else(|| TableCatalogStoreError::Conflict("migration fence has no ETag".to_owned()))?;
+        self.backend
+            .delete_object_if_match_fenced(self.catalog_bucket(), &fence_path, &fence_etag, &mutation_fence)
+            .await?;
+        self.clear_global_backing_migration_fence_if_unused(&global_fence_path, &mutation_fence)
             .await?;
         Ok(TableCatalogBackingMigrationCancelReport {
             table_bucket: table_bucket.to_string(),

rustfs/src/table_catalog/store/mod.rs

@@ -18,9 +18,15 @@ use crate::storage_api::table::{WriteCommitGuard, WriteCompletion};
 
 mod backup;
 mod cleanup;
+mod directory_index;
 mod migration;
 mod object;
+mod publication_deadline;
 mod registration_cache;
+pub(crate) use publication_deadline::TablePublicationPreparation;
+use publication_deadline::{
+    begin_publication_mutation, publication_preparation, publication_recovery, retain_publication_preparation_fence,
+};
 #[cfg(test)]
 mod registration_cache_tests;
 pub(crate) use registration_cache::TableCatalogRegistrationCache;
@@ -464,6 +470,10 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
 #[async_trait::async_trait]
 pub(crate) trait TableCommitPublication: Send + Sync {
+    fn preparation_budget(&self) -> Arc<TablePublicationPreparation> {
+        Arc::new(TablePublicationPreparation::default())
+    }
+
     async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()>;
 
     async fn prepare(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()>;
@@ -507,6 +517,7 @@ impl Drop for TableCommitPublicationCompletion<'_> {
 
 struct TableCommitLockPublication<'a, B> {
     backend: &'a B,
+    preparation: Arc<TablePublicationPreparation>,
     state: parking_lot::Mutex<TableCommitLockPublicationState>,
 }
 
@@ -522,6 +533,7 @@ impl<'a, B> TableCommitLockPublication<'a, B> {
     fn new(backend: &'a B) -> Self {
         Self {
             backend,
+            preparation: Arc::new(TablePublicationPreparation::default()),
             state: parking_lot::Mutex::new(TableCommitLockPublicationState::default()),
         }
     }
@@ -571,7 +583,13 @@ impl<'a, B> TableCommitPublication for TableCommitLockPublication<'a, B>
 where
     B: TableCatalogObjectBackend,
 {
+    fn preparation_budget(&self) -> Arc<TablePublicationPreparation> {
+        self.preparation.clone()
+    }
+
     async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
+        // Includes an earlier acquisition even when a caller prepares before entering the store.
+        let _ = self.preparation.deadline();
         {
             let mut state = self.state.lock();
             if state
@@ -677,12 +695,15 @@ where
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct TableCatalogObject {
     pub data: Vec<u8>,
+    /// Server-generated identity of the body returned by this read.
+    pub body_revision: Option<Uuid>,
     pub etag: Option<String>,
     pub mod_time: Option<OffsetDateTime>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct TableCatalogObjectMetadata {
+    pub body_revision: Option<Uuid>,
     pub etag: Option<String>,
     pub mod_time: Option<OffsetDateTime>,
     pub size: u64,
@@ -788,6 +809,11 @@ pub(crate) struct TableCatalogCleanupObject {
     pub delete_marker: bool,
 }
 
+pub(crate) struct TableCatalogCleanupPage {
+    pub objects: Vec<TableCatalogCleanupObject>,
+    pub is_truncated: bool,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum TableCatalogPutPrecondition {
     Any,
@@ -872,6 +898,7 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
             .read_object(bucket, object)
             .await?
             .map(|object| TableCatalogObjectMetadata {
+                body_revision: object.body_revision,
                 etag: object.etag,
                 mod_time: object.mod_time,
                 size: u64::try_from(object.data.len()).unwrap_or(u64::MAX),
@@ -887,6 +914,7 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
             .read_object_unlocked(bucket, object)
             .await?
             .map(|object| TableCatalogObjectMetadata {
+                body_revision: object.body_revision,
                 etag: object.etag,
                 mod_time: object.mod_time,
                 size: u64::try_from(object.data.len()).unwrap_or(u64::MAX),
@@ -1010,6 +1038,20 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
         ))
     }
 
+    /// GET-equivalent, bounded enumeration for publishing authoritative indexes.
+    /// Backends without this contract must not fall back to an ordinary S3 list.
+    async fn list_catalog_objects_page(
+        &self,
+        _bucket: &str,
+        _prefix: &str,
+        _start_after: Option<&str>,
+        _limit: NonZeroUsize,
+    ) -> TableCatalogStoreResult<TableCatalogCleanupPage> {
+        Err(TableCatalogStoreError::Unsupported(
+            "authoritative catalog pagination is not supported by this backend".to_string(),
+        ))
+    }
+
     async fn list_objects_page(
         &self,
         bucket: &str,
@@ -1042,6 +1084,10 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
         Ok(())
     }
 
+    fn table_publication_preparation_budget(&self) -> Arc<TablePublicationPreparation> {
+        Arc::new(TablePublicationPreparation::default())
+    }
+
     async fn begin_table_bucket_commit_publication(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
         Ok(())
     }
@@ -1178,6 +1224,10 @@ impl<B> TableCommitPublication for B
 where
     B: TableCatalogObjectBackend,
 {
+    fn preparation_budget(&self) -> Arc<TablePublicationPreparation> {
+        self.table_publication_preparation_budget()
+    }
+
     async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
         self.begin_table_bucket_commit_publication(table_bucket).await
     }
@@ -2333,6 +2383,8 @@ where
     ) -> TableCatalogStoreResult<Vec<TableCatalogLockGuard>> {
         // Handlers discover and publish objects through one publication object. Keep the
         // migration read permits ahead of its publication lock, matching store mutations.
+        #[cfg(test)]
+        crate::table_catalog::test_support::pause_before_catalog_migration_read().await;
         let paths = TableCatalogObjectPaths::default();
         let global = self
             .acquire_read_lock(RUSTFS_META_BUCKET, &paths.backing_migration_global_fence_lock_path())
@@ -2390,8 +2442,17 @@ where
     }
 
     async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        #[cfg(test)]
+        if bucket == RUSTFS_META_BUCKET
+            && object == StrongTableCatalogStore::<Self>::snapshot_object_path()
+            && let Ok(Some(pause)) =
+                crate::table_catalog::test_support::CATALOG_ROOT_HEAD_PAUSE.try_with(|slot| slot.get().cloned())
+        {
+            pause.pause().await;
+        }
         match self.store.get_object_info(bucket, object, &ObjectOptions::default()).await {
             Ok(info) => Ok(Some(TableCatalogObjectMetadata {
+                body_revision: info.data_dir.filter(|id| !id.is_nil()),
                 etag: info.etag,
                 mod_time: info.mod_time,
                 size: u64::try_from(info.size.max(0)).unwrap_or(u64::MAX),
@@ -2419,6 +2480,7 @@ where
             .await
         {
             Ok(info) => Ok(Some(TableCatalogObjectMetadata {
+                body_revision: info.data_dir.filter(|id| !id.is_nil()),
                 etag: info.etag,
                 mod_time: info.mod_time,
                 size: u64::try_from(info.size.max(0)).unwrap_or(u64::MAX),
@@ -2644,6 +2706,36 @@ where
         })
     }
 
+    async fn list_catalog_objects_page(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        start_after: Option<&str>,
+        limit: NonZeroUsize,
+    ) -> TableCatalogStoreResult<TableCatalogCleanupPage> {
+        let max_keys = i32::try_from(limit.get())
+            .map_err(|_| TableCatalogStoreError::Invalid("catalog page size exceeds storage API range".to_string()))?;
+        let page = self
+            .store
+            .clone()
+            .list_unversioned_objects_authoritative(bucket, prefix, start_after.map(str::to_string), max_keys)
+            .await
+            .map_err(|err| storage_error_to_catalog("list authoritative catalog page", err))?;
+        Ok(TableCatalogCleanupPage {
+            objects: page
+                .objects
+                .into_iter()
+                .map(|object| TableCatalogCleanupObject {
+                    key: object.name,
+                    version_id: object.version_id.filter(|id| !id.is_nil()).map(|id| id.to_string()),
+                    etag: object.etag,
+                    delete_marker: object.delete_marker,
+                })
+                .collect(),
+            is_truncated: page.is_truncated,
+        })
+    }
+
     async fn list_objects_for_cleanup(
         &self,
         bucket: &str,
@@ -2783,6 +2875,7 @@ where
                 )));
             }
         }
+        let body_revision = reader.object_info.data_dir.filter(|id| !id.is_nil());
         let etag = reader.object_info.etag.clone();
         let mod_time = reader.object_info.mod_time;
         let mut data = Vec::new();
@@ -2801,7 +2894,12 @@ where
                 TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}"))
             })?;
         }
-        Ok(Some(TableCatalogObject { data, etag, mod_time }))
+        Ok(Some(TableCatalogObject {
+            data,
+            body_revision,
+            etag,
+            mod_time,
+        }))
     }
 
     async fn put_object_with_options(
@@ -2926,33 +3024,52 @@ mod tests {
             )))),
             acquired: Mutex::new(None),
         });
-        let lock = rustfs_lock::NamespaceLockWrapper::new(
-            rustfs_lock::NamespaceLock::with_clients_and_quorum(
-                "maintenance-probe".to_string(),
-                vec![client.clone() as Arc<dyn LockClient>],
-                1,
-            ),
-            rustfs_lock::ObjectKey::new(
-                RUSTFS_META_BUCKET,
-                TableCatalogObjectPaths::default().backing_migration_global_fence_lock_path(),
-            ),
-            "maintenance-migration-probe".to_string(),
+        let lock = rustfs_lock::NamespaceLock::with_clients_and_quorum(
+            "maintenance-probe".to_string(),
+            vec![client.clone() as Arc<dyn LockClient>],
+            1,
         );
-        let guard = TableCatalogLockGuard::namespace(lock.get_read_lock(Duration::from_secs(5)).await.unwrap());
-        let publication = TableCommitLockPublication::new(&backend);
-        publication.attach_catalog_migration_read_guard(guard).unwrap();
-        publication.begin_table_bucket(bucket).await.unwrap();
-        publication.prepare(bucket, "sales", "orders").await.unwrap();
-        assert!(publication.holds_table(bucket, "sales", "orders"));
+        let resource = rustfs_lock::ObjectKey::new(
+            RUSTFS_META_BUCKET,
+            TableCatalogObjectPaths::default().backing_migration_global_fence_lock_path(),
+        );
+        // Observe real lease loss before the separate publication preparation
+        // deadline. Keep the lease TTL unchanged; only speed up this probe.
+        let request =
+            rustfs_lock::LockRequest::new(resource.clone(), rustfs_lock::LockType::Shared, "maintenance-migration-probe")
+                .with_acquire_timeout(Duration::from_secs(5))
+                .with_ttl(Duration::from_secs(30))
+                .with_refresh_interval(Duration::from_millis(20));
+        let guard = TableCatalogLockGuard::namespace(lock.acquire_guard(&request).await.unwrap().unwrap());
         let id = client
             .acquired
             .lock()
             .unwrap()
             .clone()
-            .expect("actual migration lease acquired");
+            .expect("actual publication migration lease acquired");
+        // Completion clears the publication's guards. Keep another real lease
+        // to test rejection of an already-lost guard after that cleanup.
+        let attach_request = rustfs_lock::LockRequest::new(resource, rustfs_lock::LockType::Shared, "maintenance-attach-probe")
+            .with_acquire_timeout(Duration::from_secs(5))
+            .with_ttl(Duration::from_secs(30))
+            .with_refresh_interval(Duration::from_millis(20));
+        let lost = TableCatalogLockGuard::namespace(lock.acquire_guard(&attach_request).await.unwrap().unwrap());
+        let attach_id = client
+            .acquired
+            .lock()
+            .unwrap()
+            .clone()
+            .expect("actual attachment migration lease acquired");
+        assert_ne!(id, attach_id, "the attachment probe must use its own lease");
+        let publication = TableCommitLockPublication::new(&backend);
+        publication.attach_catalog_migration_read_guard(guard).unwrap();
+        publication.begin_table_bucket(bucket).await.unwrap();
+        publication.prepare(bucket, "sales", "orders").await.unwrap();
+        assert!(publication.holds_table(bucket, "sales", "orders"));
         assert!(client.force_release(&id).await.unwrap());
-        tokio::time::timeout(Duration::from_secs(15), async {
-            while publication.catalog_migration_read_permit_status() != Some(false) {
+        assert!(client.force_release(&attach_id).await.unwrap());
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while publication.catalog_migration_read_permit_status() != Some(false) || !lost.is_lock_lost() {
                 tokio::time::sleep(Duration::from_millis(10)).await;
             }
         })

rustfs/src/table_catalog/store/publication_deadline.rs

@@ -0,0 +1,170 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use std::{
+    future::Future,
+    sync::{
+        OnceLock,
+        atomic::{AtomicBool, Ordering},
+    },
+    task::Poll,
+};
+
+/// One admission budget, retained across repeated preparation and completed recovery.
+/// It never cancels a transaction after its first possible durable side effect.
+#[derive(Default)]
+pub(crate) struct TablePublicationPreparation {
+    deadline: OnceLock<tokio::time::Instant>,
+    mutation_started: AtomicBool,
+    fence: parking_lot::Mutex<Option<TableCatalogObjectMutationFence>>,
+}
+
+tokio::task_local! {
+    static PUBLICATION_PREPARATION: Arc<TablePublicationPreparation>;
+}
+
+impl TablePublicationPreparation {
+    pub(super) fn deadline(&self) -> tokio::time::Instant {
+        *self
+            .deadline
+            .get_or_init(|| tokio::time::Instant::now() + get_lock_acquire_timeout())
+    }
+
+    fn expired() -> TableCatalogStoreError {
+        TableCatalogStoreError::Unavailable("table publication preparation deadline exceeded; retry the operation".to_string())
+    }
+
+    fn check(&self) -> TableCatalogStoreResult<()> {
+        if !self.mutation_started.load(Ordering::Acquire) && tokio::time::Instant::now() >= self.deadline() {
+            return Err(Self::expired());
+        }
+        Ok(())
+    }
+
+    pub(crate) fn begin_mutation(&self) -> TableCatalogStoreResult<()> {
+        self.check()?;
+        if let Some(fence) = self.fence.lock().as_ref() {
+            fence.ensure_held()?;
+        }
+        self.mutation_started.store(true, Ordering::Release);
+        Ok(())
+    }
+
+    // Only use for pure acquisition/validation. The publication is cleared on
+    // timeout or caller cancellation, including partially acquired guards.
+    pub(crate) async fn read<T>(
+        &self,
+        publication: &(dyn TableCommitPublication + Sync),
+        future: impl Future<Output = TableCatalogStoreResult<T>>,
+    ) -> TableCatalogStoreResult<T> {
+        let mut cleanup = PreparationReadCleanup {
+            publication,
+            armed: true,
+        };
+        let result = if self.mutation_started.load(Ordering::Acquire) {
+            future.await
+        } else {
+            tokio::time::timeout_at(self.deadline(), future)
+                .await
+                .map_err(|_| Self::expired())?
+        };
+        if result.is_ok() {
+            self.check()?;
+        }
+        cleanup.armed = false;
+        result
+    }
+
+    async fn run<T>(&self, future: impl Future<Output = TableCatalogStoreResult<T>>) -> TableCatalogStoreResult<T> {
+        let timer = tokio::time::sleep_until(self.deadline());
+        tokio::pin!(timer, future);
+        std::future::poll_fn(|cx| {
+            // Check before polling new work. A mutation boundary also checks
+            // synchronously, so CPU work cannot publish after an expired timer.
+            if !self.mutation_started.load(Ordering::Acquire) && timer.as_mut().poll(cx).is_ready() {
+                return Poll::Ready(Err(Self::expired()));
+            }
+            match future.as_mut().poll(cx) {
+                Poll::Ready(result) => Poll::Ready(self.check().and(result)),
+                Poll::Pending => Poll::Pending,
+            }
+        })
+        .await
+    }
+}
+
+struct PreparationReadCleanup<'a> {
+    publication: &'a (dyn TableCommitPublication + Sync),
+    armed: bool,
+}
+
+impl Drop for PreparationReadCleanup<'_> {
+    fn drop(&mut self) {
+        if self.armed {
+            self.publication.complete();
+        }
+    }
+}
+
+pub(super) async fn publication_preparation<T>(
+    publication: &(dyn TableCommitPublication + Sync),
+    future: impl Future<Output = TableCatalogStoreResult<T>>,
+) -> TableCatalogStoreResult<T> {
+    let _completion = TableCommitPublicationCompletion::new(publication);
+    let preparation = publication.preparation_budget();
+    PUBLICATION_PREPARATION
+        .scope(preparation.clone(), preparation.run(future))
+        .await
+}
+
+pub(super) fn retain_publication_preparation_fence(publication: &(dyn TableCommitPublication + Sync)) {
+    let _ = PUBLICATION_PREPARATION.try_with(|preparation| {
+        *preparation.fence.lock() = publication.object_mutation_fence();
+    });
+}
+
+/// Call immediately before the first possible persistent side effect, after
+/// read-only validation and existing lease checks. A failed/ambiguous write is
+/// still a mutation: recovery must retain the original cancellation semantics.
+pub(super) fn begin_publication_mutation() -> TableCatalogStoreResult<()> {
+    PUBLICATION_PREPARATION
+        .try_with(|preparation| preparation.begin_mutation())
+        .unwrap_or(Ok(()))
+}
+
+/// Recovery is a preceding transaction, not permission for the new transaction
+/// to bypass admission. Complete its original protocol once a write starts,
+/// then resume the unchanged deadline before any new preparation.
+pub(super) async fn publication_recovery<T>(
+    future: impl Future<Output = TableCatalogStoreResult<T>>,
+) -> TableCatalogStoreResult<T> {
+    let preparation = PUBLICATION_PREPARATION.try_with(Arc::clone).ok();
+    let already_mutating = preparation
+        .as_ref()
+        .is_some_and(|budget| budget.mutation_started.load(Ordering::Acquire));
+    let result = future.await;
+    if result.is_ok()
+        && !already_mutating
+        && let Some(preparation) = preparation
+    {
+        preparation.mutation_started.store(false, Ordering::Release);
+        preparation.check()?;
+    }
+    result
+}
+
+#[cfg(test)]
+#[path = "publication_deadline_tests.rs"]
+mod tests;

rustfs/src/table_catalog/store/publication_deadline_tests.rs

@@ -0,0 +1,331 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::table_catalog::store::registration_cache_tests::{cache_fixture, registration_entry};
+use crate::table_catalog::test_support::TestCatalogObjectBackend;
+use crate::table_catalog::tests::{test_bucket_entry, test_namespace_entry, test_table_entry};
+
+fn is_preparation_timeout<T>(result: &TableCatalogStoreResult<T>) -> bool {
+    matches!(result, Err(TableCatalogStoreError::Unavailable(message)) if message.contains("preparation deadline"))
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_registration_inventory_releases_fence_without_writes() {
+    let (backend, store, namespace) = cache_fixture().await;
+    backend.registration_cache.clear();
+    let path = store
+        .paths
+        .table_entry_path("analytics", &namespace, &IdentifierSegment::parse("orders").unwrap());
+    let pause = backend.pause_before_next_read(RUSTFS_META_BUCKET, &path).await;
+    let before = backend.state.lock().await.put_attempts.clone();
+    let registering = {
+        let store = store.clone();
+        let entry = registration_entry(&namespace, "candidate");
+        tokio::spawn(async move { store.register_table(entry).await })
+    };
+    pause.wait_started().await;
+    let writer = {
+        let backend = backend.clone();
+        tokio::spawn(async move {
+            // Data-plane mutations acquire this same read fence before writing.
+            let _guard = backend
+                .acquire_read_lock("analytics", &default_table_bucket_publication_lock_path())
+                .await?;
+            backend
+                .put_object(
+                    "analytics",
+                    "tables/other-tenant/data/new",
+                    b"new".to_vec(),
+                    TableCatalogPutPrecondition::IfAbsent,
+                )
+                .await
+        })
+    };
+    tokio::task::yield_now().await;
+    assert!(!writer.is_finished());
+    tokio::time::advance(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    assert!(is_preparation_timeout(&registering.await.unwrap()));
+    writer.await.unwrap().unwrap();
+    pause.release();
+    assert!(store.load_table("analytics", "sales", "candidate").await.unwrap().is_none());
+    let after = backend.state.lock().await.put_attempts.clone();
+    for (key, attempts) in after {
+        if key.0 == RUSTFS_META_BUCKET {
+            assert_eq!(Some(&attempts), before.get(&key), "timed-out preparation must not stage catalog writes");
+        }
+    }
+    store
+        .create_table(registration_entry(&namespace, "after_timeout"))
+        .await
+        .unwrap();
+}
+
+async fn commit_fixture(strong: bool) -> (TestCatalogObjectBackend, Arc<dyn TableCatalogStore>, TableCommitRequest) {
+    let backend = TestCatalogObjectBackend::default();
+    let store: Arc<dyn TableCatalogStore> = if strong {
+        Arc::new(StrongTableCatalogStore::new(backend.clone()))
+    } else {
+        Arc::new(ObjectTableCatalogStore::new(backend.clone()))
+    };
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    let next = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+    store.put_table_bucket(test_bucket_entry("analytics")).await.unwrap();
+    store
+        .create_namespace(test_namespace_entry("analytics", &namespace))
+        .await
+        .unwrap();
+    store
+        .create_table(test_table_entry("analytics", &namespace, &table, current.clone()))
+        .await
+        .unwrap();
+    backend.seed_object("analytics", &next, b"{}".to_vec()).await;
+    (
+        backend,
+        store,
+        TableCommitRequest {
+            table_bucket: "analytics".to_string(),
+            namespace: "sales".to_string(),
+            table: "orders".to_string(),
+            commit_id: "deadline-commit".to_string(),
+            idempotency_key: Some("deadline-idempotency".to_string()),
+            operation: "append".to_string(),
+            expected_version_token: "token-v1".to_string(),
+            expected_metadata_location: current,
+            new_metadata_location: next,
+            requirements: Vec::new(),
+            writer: None,
+        },
+    )
+}
+
+// The durable-strong backing does not take part in the preparation deadline; it is retired separately,
+// so only the object backing is exercised here.
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_commit_metadata_read_does_not_publish_object_backing() {
+    let (backend, store, request) = commit_fixture(false).await;
+    let pause = backend
+        .pause_before_next_read("analytics", &request.new_metadata_location)
+        .await;
+    let before = backend.state.lock().await.put_attempts.clone();
+    let committing = {
+        let store = store.clone();
+        let request = request.clone();
+        tokio::spawn(async move { store.commit_table(request).await })
+    };
+    pause.wait_started().await;
+    tokio::time::advance(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    assert!(is_preparation_timeout(&committing.await.unwrap()));
+    assert_eq!(backend.state.lock().await.put_attempts, before);
+    assert_eq!(
+        store
+            .load_table("analytics", "sales", "orders")
+            .await
+            .unwrap()
+            .unwrap()
+            .generation,
+        1
+    );
+    pause.release();
+    // Positive retry exercises parsing a nontrivial valid metadata body.
+    backend
+        .seed_object(
+            "analytics",
+            &request.new_metadata_location,
+            serde_json::to_vec(&serde_json::json!({"properties": {"padding": "x".repeat(1024 * 1024)}})).unwrap(),
+        )
+        .await;
+    tokio::time::resume();
+    let committed = store.commit_table(request.clone()).await.unwrap();
+    assert_eq!(committed.table.generation, 2);
+    assert_eq!(
+        store.commit_table(request).await.unwrap().table.generation,
+        2,
+        "retry must remain idempotent"
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_create_finishes_after_first_write_on_either_backing() {
+    for strong in [false, true] {
+        let (backend, store, _) = commit_fixture(strong).await;
+        let namespace = Namespace::parse("sales").unwrap();
+        let candidate = registration_entry(&namespace, "candidate");
+        let paths = TableCatalogObjectPaths::default();
+        let first_write = if strong {
+            StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path()
+        } else {
+            paths.warehouse_index_entry_path("analytics", "tables/id-candidate/")
+        };
+        let pause = backend.pause_next_put(RUSTFS_META_BUCKET, &first_write).await;
+        let creating = {
+            let store = store.clone();
+            tokio::spawn(async move { store.create_table(candidate).await })
+        };
+        pause.wait_started().await;
+        tokio::time::advance(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+        tokio::task::yield_now().await;
+        assert!(!creating.is_finished(), "deadline must not cancel a possible durable write");
+        pause.release();
+        creating.await.unwrap().unwrap();
+        assert!(store.load_table("analytics", "sales", "candidate").await.unwrap().is_some());
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_queueing_and_repeated_prepare_share_one_budget() {
+    let backend = TestCatalogObjectBackend::default();
+    let publication = TableCommitLockPublication::new(&backend);
+    let budget = publication.preparation_budget();
+    let first_deadline = budget.deadline();
+    tokio::time::advance(get_lock_acquire_timeout() - StdDuration::from_millis(1)).await;
+    publication.begin_table_bucket("analytics").await.unwrap();
+    publication.prepare("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(budget.deadline(), first_deadline);
+    let result = publication_preparation(&publication, async {
+        publication.prepare("analytics", "sales", "orders").await?;
+        tokio::time::sleep(StdDuration::from_millis(2)).await;
+        begin_publication_mutation()
+    })
+    .await;
+    assert!(is_preparation_timeout(&result));
+    assert!(!publication.holds_table_bucket("analytics"));
+    assert_eq!(budget.deadline(), first_deadline, "cleanup must not renew the admission budget");
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_completed_recovery_cannot_exempt_a_new_transaction() {
+    let backend = TestCatalogObjectBackend::default();
+    let publication = TableCommitLockPublication::new(&backend);
+    let result = publication_preparation(&publication, async {
+        publication.begin_table_bucket("analytics").await?;
+        publication_recovery(async {
+            begin_publication_mutation()?;
+            backend
+                .put_object("analytics", "recovery", b"recovered".to_vec(), TableCatalogPutPrecondition::IfAbsent)
+                .await?;
+            tokio::time::sleep(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+            Ok(())
+        })
+        .await?;
+        backend
+            .put_object(
+                "analytics",
+                "new-transaction",
+                b"unexpected".to_vec(),
+                TableCatalogPutPrecondition::IfAbsent,
+            )
+            .await
+    })
+    .await;
+    assert!(is_preparation_timeout(&result));
+    assert!(backend.object_exists("analytics", "recovery").await.unwrap());
+    assert!(!backend.object_exists("analytics", "new-transaction").await.unwrap());
+    assert!(!publication.holds_table_bucket("analytics"));
+}
+
+// The durable-strong backing does not take part in the preparation deadline; it is retired separately,
+// so only the object backing is exercised here.
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_create_queue_wait_cannot_restart_after_admission() {
+    let (backend, store, _) = commit_fixture(false).await;
+    let key = default_table_bucket_publication_lock_path();
+    let blocker = backend.acquire_read_lock("analytics", &key).await.unwrap();
+    let attempts = backend.write_lock_acquisition_count("analytics", &key).await;
+    let candidate = registration_entry(&Namespace::parse("sales").unwrap(), "queued");
+    let creating = {
+        let store = store.clone();
+        tokio::spawn(async move { store.create_table(candidate).await })
+    };
+    while backend.write_lock_acquisition_count("analytics", &key).await == attempts {
+        tokio::task::yield_now().await;
+    }
+    tokio::time::advance(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    assert!(is_preparation_timeout(&creating.await.unwrap()));
+    drop(blocker);
+    assert!(store.load_table("analytics", "sales", "queued").await.unwrap().is_none());
+    store
+        .create_table(registration_entry(&Namespace::parse("sales").unwrap(), "fresh"))
+        .await
+        .unwrap();
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_active_rename_recovery_finishes_before_rejecting_new_registration() {
+    let (backend, store, namespace) = cache_fixture().await;
+    let source = store
+        .paths
+        .table_entry_path("analytics", &namespace, &IdentifierSegment::parse("orders").unwrap());
+    backend.fail_next_put(RUSTFS_META_BUCKET, &source).await;
+    assert!(
+        store
+            .rename_table("analytics", "sales", "orders", "sales", "renamed_orders")
+            .await
+            .is_err()
+    );
+    assert!(
+        store
+            .get_table_bucket("analytics")
+            .await
+            .unwrap()
+            .unwrap()
+            .active_rename_id
+            .is_some()
+    );
+    let pause = backend.pause_next_put(RUSTFS_META_BUCKET, &source).await;
+    let registering = {
+        let store = store.clone();
+        let candidate = registration_entry(&namespace, "after_recovery");
+        tokio::spawn(async move { store.register_table(candidate).await })
+    };
+    pause.wait_started().await;
+    tokio::time::advance(get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+    tokio::task::yield_now().await;
+    assert!(
+        !registering.is_finished(),
+        "in-flight recovery must not be cancelled by the preparation deadline"
+    );
+    pause.release();
+    assert!(is_preparation_timeout(&registering.await.unwrap()));
+    assert!(
+        store
+            .get_table_bucket("analytics")
+            .await
+            .unwrap()
+            .unwrap()
+            .active_rename_id
+            .is_none()
+    );
+    assert!(store.load_table("analytics", "sales", "orders").await.unwrap().is_none());
+    assert!(
+        store
+            .load_table("analytics", "sales", "renamed_orders")
+            .await
+            .unwrap()
+            .is_some()
+    );
+    assert!(
+        store
+            .load_table("analytics", "sales", "after_recovery")
+            .await
+            .unwrap()
+            .is_none()
+    );
+    store
+        .register_table(registration_entry(&namespace, "fresh_request"))
+        .await
+        .unwrap();
+}

rustfs/src/table_catalog/store/registration_cache.rs

@@ -171,10 +171,44 @@ impl<B: TableCatalogObjectBackend> ObjectTableCatalogStore<B> {
         let index_prefix = self.paths.warehouse_index_entries_prefix(&entry.table_bucket);
         // The backend must preserve system-bucket committed-read semantics as
         // well as completeness; ordinary S3 LIST is insufficient for cache hits.
-        let mut tables = self
-            .backend
-            .list_objects_for_cleanup(self.catalog_bucket(), &namespace_prefix)
-            .await?;
+        let directory = self.table_directory_objects(&entry.table_bucket, None).await?;
+        let directory_epoch = directory.as_ref().map(|scan| scan.activation_id);
+        let mut tables = match directory {
+            Some(scan) => {
+                // The directory proves which keys to inspect, but cached owners
+                // still require the current GET-equivalent committed ETag. Do
+                // not replace this exact version probe with an ordinary stat.
+                stream::iter(scan.objects)
+                    .map(|key| async move {
+                        let page = self
+                            .backend
+                            .list_catalog_objects_page(
+                                self.catalog_bucket(),
+                                &key,
+                                None,
+                                NonZeroUsize::new(2).ok_or_else(|| {
+                                    TableCatalogStoreError::Internal("exact inventory page size must be positive".to_string())
+                                })?,
+                            )
+                            .await?;
+                        if page.is_truncated || page.objects.len() > 1 || page.objects.iter().any(|object| object.key != key) {
+                            return Err(TableCatalogStoreError::Invalid(
+                                "table directory target has an ambiguous authoritative inventory".to_string(),
+                            ));
+                        }
+                        Ok(page.objects.into_iter().next())
+                    })
+                    .buffered(TABLE_REGISTRATION_READ_CONCURRENCY)
+                    .try_filter_map(|object| async move { Ok(object) })
+                    .try_collect::<Vec<_>>()
+                    .await?
+            }
+            None => {
+                self.backend
+                    .list_objects_for_cleanup(self.catalog_bucket(), &namespace_prefix)
+                    .await?
+            }
+        };
         let mut indexes = self
             .backend
             .list_objects_for_cleanup(self.catalog_bucket(), &index_prefix)
@@ -252,6 +286,9 @@ impl<B: TableCatalogObjectBackend> ObjectTableCatalogStore<B> {
                 })?;
             table_warehouse_index_state_ready(&state, &entry.table_bucket)?;
         }
+        if let Some(epoch) = directory_epoch {
+            self.ensure_table_directory_epoch(&entry.table_bucket, epoch).await?;
+        }
         if !publication.holds_table_bucket(&entry.table_bucket) {
             return Err(TableCatalogStoreError::Unavailable("registration publication fence was lost".to_string()));
         }

rustfs/src/table_catalog/store/strong.rs

@@ -1532,9 +1532,19 @@ where
         }
     }
 
+    #[cfg(test)]
     pub(in crate::table_catalog) async fn materialize_bucket_snapshot(
         &self,
         source: StrongTableCatalogBucketSnapshot,
+    ) -> TableCatalogStoreResult<(String, bool)> {
+        self.materialize_bucket_snapshot_with_fence(source, &TableCatalogObjectMutationFence::default())
+            .await
+    }
+
+    pub(in crate::table_catalog) async fn materialize_bucket_snapshot_with_fence(
+        &self,
+        source: StrongTableCatalogBucketSnapshot,
+        fence: &TableCatalogObjectMutationFence,
     ) -> TableCatalogStoreResult<(String, bool)> {
         let _write_guard = self.write_lock.lock().await;
         self.hydrate_state().await?;
@@ -1562,7 +1572,8 @@ where
                 postcondition,
             )
         };
-        self.finalize_snapshot_write(snapshot, precondition, postcondition).await?;
+        self.finalize_snapshot_write_with_fence(snapshot, precondition, postcondition, Some(fence))
+            .await?;
         self.hydrate_state().await?;
         let state = self.state.lock().await;
         let snapshot_etag = state
@@ -1572,11 +1583,27 @@ where
         Ok((snapshot_etag, true))
     }
 
+    #[cfg(test)]
     pub(in crate::table_catalog) async fn remove_bucket_snapshot_if_unchanged(
         &self,
         table_bucket: &str,
         expected_fingerprint: &str,
     ) -> TableCatalogStoreResult<()> {
+        self.remove_bucket_snapshot_if_unchanged_with_fence(
+            table_bucket,
+            expected_fingerprint,
+            &TableCatalogObjectMutationFence::default(),
+        )
+        .await
+    }
+
+    pub(in crate::table_catalog) async fn remove_bucket_snapshot_if_unchanged_with_fence(
+        &self,
+        table_bucket: &str,
+        expected_fingerprint: &str,
+        fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        fence.ensure_held()?;
         let _write_guard = self.write_lock.lock().await;
         self.hydrate_state().await?;
         let (snapshot, precondition, postcondition) = {
@@ -1597,7 +1624,8 @@ where
                 StrongSnapshotWritePostcondition::TableBucketAbsent(table_bucket.to_string()),
             )
         };
-        self.finalize_snapshot_write(snapshot, precondition, postcondition).await
+        self.finalize_snapshot_write_with_fence(snapshot, precondition, postcondition, Some(fence))
+            .await
     }
 
     pub(super) async fn restore_absent_migration_snapshot_baseline(&self) -> TableCatalogStoreResult<()> {

rustfs/src/table_catalog/test_support.rs

@@ -442,6 +442,11 @@ pub(crate) struct TestCatalogObjectPause {
 }
 
 impl TestCatalogObjectPause {
+    pub(crate) async fn pause(&self) {
+        self.started.notify_one();
+        self.release.notified().await;
+    }
+
     pub(crate) async fn wait_started(&self) {
         self.started.notified().await;
     }
@@ -451,6 +456,18 @@ impl TestCatalogObjectPause {
     }
 }
 
+tokio::task_local! {
+    pub(crate) static CATALOG_MIGRATION_READ_PAUSE: TestCatalogObjectPause;
+    pub(crate) static CATALOG_ROOT_HEAD_PAUSE: Arc<std::sync::OnceLock<TestCatalogObjectPause>>;
+    pub(crate) static CATALOG_PUT_AUTHORIZATION_PAUSE: Arc<std::sync::OnceLock<TestCatalogObjectPause>>;
+}
+
+pub(crate) async fn pause_before_catalog_migration_read() {
+    if let Ok(pause) = CATALOG_MIGRATION_READ_PAUSE.try_with(Clone::clone) {
+        pause.pause().await;
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct BlockingObjectPublication {
     backend: TestCatalogObjectBackend,
@@ -524,8 +541,16 @@ impl TableCommitPublication for BlockingObjectPublication {
     }
 }
 
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) struct TestCatalogWorkerContext {
+    pub(crate) access_key: Option<String>,
+    pub(crate) operation: Option<&'static str>,
+    pub(crate) span_name: Option<&'static str>,
+}
+
 #[derive(Default)]
 pub(crate) struct TestCatalogObjectState {
+    pub(crate) observed_worker_contexts: Option<Vec<TestCatalogWorkerContext>>,
     pub(crate) objects: BTreeMap<(String, String), TestCatalogObjectRecord>,
     pub(crate) etagless_objects: BTreeSet<(String, String)>,
     pub(crate) fail_read_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
@@ -546,6 +571,10 @@ pub(crate) struct TestCatalogObjectState {
     pub(crate) read_calls: usize,
     pub(crate) read_bytes: usize,
     pub(crate) metadata_calls: usize,
+    pub(crate) pause_metadata: BTreeMap<(String, String), TestCatalogObjectPause>,
+    pub(crate) pause_after_metadata: BTreeMap<(String, String), TestCatalogObjectPause>,
+    pub(crate) fail_put_prefix: Option<(String, String)>,
+    pub(crate) corrupt_put_prefix: Option<(String, String)>,
     pub(crate) list_calls: usize,
     pub(crate) failed_inventory_prefixes: BTreeSet<String>,
     pub(crate) inventory_overrides: BTreeMap<String, Vec<TableCatalogCleanupObject>>,
@@ -554,6 +583,7 @@ pub(crate) struct TestCatalogObjectState {
 
 #[derive(Clone)]
 pub(crate) struct TestCatalogObjectRecord {
+    pub(crate) body_revision: Option<Uuid>,
     pub(crate) data: Vec<u8>,
     pub(crate) etag: String,
     pub(crate) mod_time: Option<OffsetDateTime>,
@@ -574,9 +604,15 @@ impl TestCatalogObjectBackend {
     ) {
         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 });
+        state.objects.insert(
+            (bucket.to_string(), object.to_string()),
+            TestCatalogObjectRecord {
+                body_revision: Some(Uuid::new_v4()),
+                data,
+                etag,
+                mod_time,
+            },
+        );
     }
 
     pub(crate) async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) {
@@ -781,6 +817,23 @@ impl TestCatalogObjectBackend {
         if let Some(fence) = fence {
             fence.ensure_held()?;
         }
+        let (fail_prefix, corrupt_prefix) = {
+            let mut state = self.state.lock().await;
+            let matches = |key: &(String, String)| key.0 == bucket && object.starts_with(&key.1);
+            let fail = state.fail_put_prefix.as_ref().is_some_and(matches);
+            let corrupt = state.corrupt_put_prefix.as_ref().is_some_and(matches);
+            if fail {
+                state.fail_put_prefix.take();
+            }
+            if corrupt {
+                state.corrupt_put_prefix.take();
+            }
+            (fail, corrupt)
+        };
+        if fail_prefix {
+            return Err(TableCatalogStoreError::Internal("injected candidate write failure".to_owned()));
+        }
+        let data = if corrupt_prefix { b"{}".to_vec() } else { data };
         let mut fail_put_object_path = self.fail_put_object_path.lock().await;
         if fail_put_object_path.as_deref() == Some(object) {
             fail_put_object_path.take();
@@ -854,6 +907,7 @@ impl TestCatalogObjectBackend {
                 state.objects.insert(
                     key.clone(),
                     TestCatalogObjectRecord {
+                        body_revision: Some(Uuid::new_v4()),
                         data,
                         etag,
                         mod_time: Some(OffsetDateTime::now_utc()),
@@ -995,6 +1049,16 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
         let (attempt, pause_before) = {
             let mut state = self.state.lock().await;
             state.read_calls += 1;
+            if let Some(observations) = state.observed_worker_contexts.as_mut() {
+                observations.push(TestCatalogWorkerContext {
+                    access_key: rustfs_keystone::KEYSTONE_CREDENTIALS
+                        .try_with(|credentials| credentials.as_ref().map(|credentials| credentials.access_key.clone()))
+                        .ok()
+                        .flatten(),
+                    operation: rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_name(),
+                    span_name: tracing::Span::current().metadata().map(|metadata| metadata.name()),
+                });
+            }
             let attempt = {
                 let attempts = state.read_attempts.entry(key.clone()).or_default();
                 *attempts += 1;
@@ -1023,6 +1087,7 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
             let mut state = self.state.lock().await;
             let etagless = state.etagless_objects.contains(&key);
             let result = state.objects.get(&key).map(|record| TableCatalogObject {
+                body_revision: record.body_revision,
                 data: record.data.clone(),
                 etag: (!etagless).then(|| record.etag.clone()),
                 mod_time: record.mod_time,
@@ -1068,15 +1133,42 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
     }
 
     async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        let pause = self
+            .state
+            .lock()
+            .await
+            .pause_metadata
+            .remove(&(bucket.to_string(), object.to_string()));
+        if let Some(pause) = pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
         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 {
+        let metadata = state.objects.get(&key).map(|record| TableCatalogObjectMetadata {
+            body_revision: record.body_revision,
             etag: (!etagless).then(|| record.etag.clone()),
             mod_time: record.mod_time,
             size: record.data.len() as u64,
-        }))
+        });
+        let pause = state.pause_after_metadata.remove(&key);
+        drop(state);
+        if let Some(pause) = pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
+        Ok(metadata)
+    }
+
+    async fn object_metadata_unlocked(
+        &self,
+        bucket: &str,
+        object: &str,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        // Match the ECStore stat path while publication already holds the object lock.
+        self.object_metadata(bucket, object).await
     }
 
     async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
@@ -1202,6 +1294,25 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
             .collect())
     }
 
+    async fn list_catalog_objects_page(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        start_after: Option<&str>,
+        limit: NonZeroUsize,
+    ) -> TableCatalogStoreResult<TableCatalogCleanupPage> {
+        // Only this in-memory fixture materializes the complete inventory.
+        // The production adapter uses a bounded authoritative storage page.
+        let mut objects = self.list_objects_for_cleanup(bucket, prefix).await?;
+        objects.sort_by(|left, right| left.key.cmp(&right.key));
+        objects.retain(|object| start_after.is_none_or(|after| object.key.as_str() > after));
+        // Production stops on a full page without consuming lookahead. Only
+        // a subsequent shorter (possibly empty) page proves EOF.
+        let is_truncated = objects.len() >= limit.get();
+        objects.truncate(limit.get());
+        Ok(TableCatalogCleanupPage { objects, is_truncated })
+    }
+
     async fn ensure_no_multipart_uploads(&self, _bucket: &str) -> TableCatalogStoreResult<()> {
         Ok(())
     }
@@ -1281,6 +1392,7 @@ impl TestCatalogObjectBackend {
         self.state.lock().await.objects.insert(
             (bucket.to_string(), object.to_string()),
             TestCatalogObjectRecord {
+                body_revision: Some(Uuid::new_v4()),
                 data,
                 etag,
                 mod_time: None,
@@ -1311,11 +1423,15 @@ impl TestCatalogObjectBackend {
     ) {
         let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
         let etag = content_etag(&data);
-        self.state
-            .lock()
-            .await
-            .objects
-            .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
+        self.state.lock().await.objects.insert(
+            (bucket.to_string(), object.to_string()),
+            TestCatalogObjectRecord {
+                body_revision: Some(Uuid::new_v4()),
+                data,
+                etag,
+                mod_time,
+            },
+        );
     }
 
     pub(crate) async fn write_lock_is_held(&self, bucket: &str, object: &str) -> bool {

rustfs/src/table_catalog/tests.rs

@@ -4874,6 +4874,7 @@ async fn maintenance_test_object_versions(backend: &TestCatalogObjectBackend) ->
             (
                 key.clone(),
                 TableCatalogObject {
+                    body_revision: object.body_revision,
                     data: object.data.clone(),
                     etag: Some(object.etag.clone()),
                     mod_time: object.mod_time,
@@ -7047,6 +7048,7 @@ async fn table_metadata_data_plane_resource_resolves_only_current_metadata() {
 
     seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
     backend.reset_call_counts().await;
+    backend.state.lock().await.read_attempts.clear();
 
     let resource = table_metadata_data_plane_resource_for_object(&store, bucket, &current)
         .await
@@ -7055,7 +7057,20 @@ async fn table_metadata_data_plane_resource_resolves_only_current_metadata() {
     assert_eq!(resource.namespace, "sales");
     assert_eq!(resource.table, "orders");
     assert_eq!(backend.list_call_count().await, 1);
-    assert_eq!(backend.read_call_count().await, 2);
+    let paths = TableCatalogObjectPaths::default();
+    // One bucket generation read, one directory mode probe, and one authoritative
+    // table entry. The legacy fixture must never load a candidate object.
+    assert_eq!(
+        backend.state.lock().await.read_attempts,
+        [
+            paths.table_bucket_entry_path(bucket),
+            paths.backing_migration_fence_path(bucket),
+            paths.table_entry_path(bucket, &namespace, &table),
+        ]
+        .into_iter()
+        .map(|key| ((RUSTFS_META_BUCKET.to_owned(), key), 1))
+        .collect()
+    );
     assert_eq!(backend.metadata_call_count().await, 1);
 
     let historical = default_table_metadata_file_path(&namespace, &table, "00000.metadata.json");
@@ -13031,6 +13046,207 @@ async fn maintenance_reachability_fails_closed_when_retained_metadata_is_unreada
     );
 }
 
+#[tokio::test]
+async fn object_maintenance_refuses_roots_shared_after_cross_namespace_rename() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let destination = Namespace::parse("private").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let original_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    let replacement_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, original_metadata.clone()).await;
+    backend
+        .seed_object(bucket, &original_metadata, br#"{"metadata-log":[]}"#.to_vec())
+        .await;
+    store
+        .create_namespace(test_namespace_entry(bucket, &destination))
+        .await
+        .unwrap();
+    store
+        .rename_table(bucket, "sales", "orders", "private", "orders")
+        .await
+        .unwrap();
+    let mut replacement = test_table_entry(bucket, &namespace, &table, replacement_metadata.clone());
+    replacement.table_id = "replacement-id".to_string();
+    replacement.table_uuid = "replacement-uuid".to_string();
+    replacement.warehouse_location = format!("s3://{bucket}/tables/replacement-id");
+    backend
+        .seed_object(bucket, &replacement_metadata, br#"{"metadata-log":[]}"#.to_vec())
+        .await;
+    store.create_table(replacement).await.unwrap();
+    assert_matches!(store.plan_table_metadata_maintenance(bucket, "sales", "orders", 0).await,
+        Err(TableCatalogStoreError::Conflict(message)) if message.contains("exclusive metadata"));
+    assert_matches!(store.delete_table_metadata_maintenance_candidates(bucket, "sales", "orders", 0).await,
+        Err(TableCatalogStoreError::Conflict(message)) if message.contains("exclusive metadata"));
+    assert!(backend.object_exists(bucket, &original_metadata).await.unwrap());
+    assert_eq!(
+        store
+            .load_table(bucket, "private", "orders")
+            .await
+            .unwrap()
+            .unwrap()
+            .metadata_location,
+        original_metadata
+    );
+    assert!(backend.state.lock().await.delete_attempts.is_empty());
+}
+
+#[tokio::test]
+async fn object_maintenance_rechecks_root_ownership_under_publication_fence() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let other_namespace = Namespace::parse("private").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+    let orphan = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
+    backend
+        .seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
+        .await;
+    backend.seed_object(bucket, &orphan, br#"{"metadata-log":[]}"#.to_vec()).await;
+    let report = store
+        .plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
+        .await
+        .unwrap();
+    assert_eq!(report.deletable_metadata_locations, vec![orphan.clone()]);
+    store
+        .create_namespace(test_namespace_entry(bucket, &other_namespace))
+        .await
+        .unwrap();
+    let mut other = test_table_entry(bucket, &other_namespace, &table, orphan.clone());
+    other.table_id = "other-id".to_string();
+    other.table_uuid = "other-uuid".to_string();
+    other.warehouse_location = format!("s3://{bucket}/tables/other-id");
+    store.create_table(other.clone()).await.unwrap();
+    assert_matches!(store.delete_table_metadata_maintenance_report(bucket, "sales", "orders", report.clone()).await,
+        Err(TableCatalogStoreError::Conflict(message)) if message.contains("exclusive metadata"));
+    assert!(backend.object_exists(bucket, &orphan).await.unwrap());
+    store.drop_table(bucket, "private", "orders").await.unwrap();
+
+    let pause = backend.pause_next_delete(bucket, &orphan).await;
+    let deleting_store = store.clone();
+    let deleting = tokio::spawn(async move {
+        deleting_store
+            .delete_table_metadata_maintenance_report(bucket, "sales", "orders", report)
+            .await
+    });
+    tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, pause.wait_started())
+        .await
+        .unwrap();
+    let publication_lock = default_table_bucket_publication_lock_path();
+    let attempts = backend.write_lock_acquisition_count(bucket, &publication_lock).await;
+    let creating_store = store.clone();
+    let creating = tokio::spawn(async move { creating_store.create_table(other).await });
+    tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, async {
+        while backend.write_lock_acquisition_count(bucket, &publication_lock).await == attempts {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    assert!(!creating.is_finished(), "new owner cannot publish between root validation and deletion");
+    creating.abort();
+    let _ = creating.await;
+    pause.release();
+    let deleted = tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, deleting)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(deleted.deletable_metadata_locations, vec![orphan.clone()]);
+    assert!(!backend.object_exists(bucket, &orphan).await.unwrap());
+    assert!(backend.object_exists(bucket, &current).await.unwrap());
+}
+
+#[tokio::test]
+async fn object_maintenance_rejects_a_report_from_a_replaced_table_incarnation() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+    let orphan = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
+    backend
+        .seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
+        .await;
+    backend.seed_object(bucket, &orphan, br#"{"metadata-log":[]}"#.to_vec()).await;
+    let report = store
+        .plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
+        .await
+        .unwrap();
+    store.drop_table(bucket, "sales", "orders").await.unwrap();
+    let mut replacement = test_table_entry(bucket, &namespace, &table, current.clone());
+    replacement.table_id = "new-incarnation".to_string();
+    replacement.table_uuid = "new-uuid".to_string();
+    replacement.warehouse_location = format!("s3://{bucket}/tables/new-incarnation");
+    store.create_table(replacement).await.unwrap();
+    assert_matches!(store.delete_table_metadata_maintenance_report(bucket, "sales", "orders", report).await,
+        Err(TableCatalogStoreError::Invalid(message)) if message.contains("identity does not match"));
+    assert!(backend.object_exists(bucket, &current).await.unwrap());
+    assert!(backend.object_exists(bucket, &orphan).await.unwrap());
+}
+
+#[tokio::test]
+async fn maintenance_orphan_planning_and_deletion_never_read_object_bodies() {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+    seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
+    backend
+        .seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
+        .await;
+    let root = format!("{}{}/", default_table_root_prefix(&namespace), table.as_str());
+    let old = format!("{root}data/old.parquet");
+    let fresh = format!("{root}delete/fresh.parquet");
+    let unknown = format!("{root}metadata/unknown.avro");
+    backend.seed_object(bucket, &old, b"body must not be read".to_vec()).await;
+    backend
+        .seed_object_with_mod_time(bucket, &fresh, vec![1], Some(OffsetDateTime::now_utc()))
+        .await;
+    backend.seed_object_with_mod_time(bucket, &unknown, vec![2], None).await;
+    for object in [&old, &fresh, &unknown] {
+        backend.fail_next_read(bucket, object).await;
+    }
+    let report = store
+        .plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
+        .await
+        .expect("stat-only dry run");
+    assert_eq!(report.deletable_object_locations, vec![old.clone()]);
+    assert_eq!(
+        object_cleanup_report(&report, &fresh).state,
+        TableMetadataMaintenanceObjectState::PendingSafetyWindow
+    );
+    assert_eq!(
+        object_cleanup_report(&report, &unknown).state,
+        TableMetadataMaintenanceObjectState::PendingSafetyWindow
+    );
+    let deleted = store
+        .delete_table_metadata_maintenance_candidates(bucket, "sales", "orders", 0)
+        .await
+        .expect("stat-only deletion recheck");
+    assert_eq!(deleted.deletable_object_locations, vec![old.clone()]);
+    assert!(!backend.object_exists(bucket, &old).await.unwrap());
+    assert!(backend.object_exists(bucket, &fresh).await.unwrap());
+    assert!(backend.object_exists(bucket, &unknown).await.unwrap());
+    let state = backend.state.lock().await;
+    for object in [&old, &fresh, &unknown] {
+        assert_eq!(
+            state.read_attempts.get(&(bucket.to_string(), object.to_string())),
+            None,
+            "orphan body must never be fetched"
+        );
+    }
+}
+
 #[tokio::test]
 async fn maintenance_dry_run_reports_unreachable_manifest_data_and_delete_candidates() {
     let backend = TestCatalogObjectBackend::default();
@@ -13570,6 +13786,7 @@ async fn compaction_plan_reports_manifest_reader_gap_without_rewrite_candidates(
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_plan_reports_row_level_delete_files_without_rewrite_candidates() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -13744,6 +13961,7 @@ async fn compaction_plan_requires_current_snapshot_metadata() {
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_commit_rewrites_small_data_files_and_advances_pointer() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -13918,6 +14136,7 @@ async fn compaction_commit_rewrites_small_data_files_and_advances_pointer() {
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_commit_keeps_partition_rewrite_groups_isolated() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -14099,6 +14318,7 @@ async fn compaction_commit_keeps_partition_rewrite_groups_isolated() {
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_commit_preserves_sort_order_and_keeps_groups_isolated() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -14238,6 +14458,7 @@ async fn compaction_commit_preserves_sort_order_and_keeps_groups_isolated() {
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_commit_rejects_schema_mismatch_without_advancing_pointer() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -14313,6 +14534,7 @@ async fn compaction_commit_rejects_schema_mismatch_without_advancing_pointer() {
 }
 
 #[tokio::test]
+#[serial_test::serial(table_compaction)]
 async fn compaction_commit_rejects_deleted_manifest_entries_without_advancing_pointer() {
     let backend = TestCatalogObjectBackend::default();
     let store = ObjectTableCatalogStore::new(backend.clone());
@@ -18514,7 +18736,8 @@ async fn strong_catalog_snapshot_rejects_unknown_fields_and_versions() {
         .as_object_mut()
         .expect("snapshot should be an object")
         .remove("future-field");
-    value["version"] = serde_json::Value::from(STRONG_TABLE_CATALOG_SNAPSHOT_VERSION.saturating_add(1));
+    // Physical version 3 identifies the bucket registry. Version 4 is unknown.
+    value["version"] = serde_json::Value::from(4);
     backend
         .seed_object(
             RUSTFS_META_BUCKET,
@@ -22089,58 +22312,108 @@ fn table_metadata_pointer_json_round_trips() {
 
 #[test]
 fn object_mutation_entrypoints_call_reserved_prefix_guard() {
-    let source = [
-        include_str!("../app/object/mod.rs"),
-        include_str!("../app/object/shared.rs"),
-        include_str!("../app/object/extract.rs"),
-        include_str!("../app/object/put.rs"),
-        include_str!("../app/object/copy.rs"),
-        include_str!("../app/object/delete.rs"),
-        include_str!("../app/object/head.rs"),
-        include_str!("../app/object/restore.rs"),
-    ]
-    .concat();
-    let delete_module = include_str!("../app/object/delete.rs");
-    let delete_object = delete_module
-        .split_once("pub async fn execute_delete_object")
-        .map(|(_, remainder)| {
-            remainder
-                .split_once("\nmod tests")
-                .map_or(remainder, |(entrypoints, _)| entrypoints)
-        })
-        .expect("delete object entrypoints should remain in app/object/delete.rs");
-
-    for expected in [
-        "validate_new_object_key(&key, request_method_name)?;\n        validate_table_catalog_object_mutation(&bucket, &key).await?;",
-        "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await",
-        "validate_table_catalog_object_mutation(&bucket, &object).await?;",
-        "validate_new_object_key(&key, \"PUT\")?;\n        validate_table_catalog_object_mutation(&bucket, &key).await?;",
-        "extract_try!(validate_table_catalog_object_mutation(&bucket, &fpath).await);",
-    ] {
-        assert!(source.contains(expected), "missing object mutation guard: {expected}");
+    // Scope every assertion to one implementation method: a guard in a sibling
+    // method (or its tests) must not satisfy another entrypoint's contract.
+    fn method<'a>(source: &'a str, signature: &str) -> &'a str {
+        source
+            .split_once(signature)
+            .unwrap_or_else(|| panic!("missing object mutation method: {signature}"))
+            .1
+            .split_once("\n    }\n")
+            .unwrap_or_else(|| panic!("missing method boundary: {signature}"))
+            .0
     }
-    let copy_object = include_str!("../app/object/copy.rs")
-        .split_once("async fn execute_copy_object_body(")
-        .map(|(_, body)| body.split_once("\n#[cfg(test)]").map_or(body, |(entrypoint, _)| entrypoint))
-        .expect("copy object entrypoint should remain in app/object/copy.rs");
-    let copy_key = copy_object
-        .find("validate_new_object_key(&key, \"COPY (dest)\")?;")
-        .expect("copy destination key validation");
-    let copy_guard = copy_object
-        .find("validate_table_catalog_object_mutation(&bucket, &key).await?;")
-        .expect("copy destination catalog guard");
-    let copy_store = copy_object
-        .find("let Some(store) = self.object_store()")
-        .expect("copy object store resolution");
-    assert!(
-        copy_key < copy_guard && copy_guard < copy_store,
-        "copy must guard the destination before store access"
-    );
 
-    assert!(
-        delete_object.contains("validate_object_key(&key, \"DELETE\")?;")
-            && delete_object.contains("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
-        "delete object entrypoint must validate the object key and reserved catalog prefix"
+    fn assert_ordered(body: &str, expected: &[&str]) {
+        let mut remainder = body;
+        for statement in expected {
+            remainder = remainder
+                .split_once(statement)
+                .unwrap_or_else(|| panic!("missing or out-of-order object mutation guard: {statement}"))
+                .1;
+        }
+    }
+
+    let put = include_str!("../app/object/put.rs");
+    let copy = include_str!("../app/object/copy.rs");
+    let delete = include_str!("../app/object/delete.rs");
+    let extract = include_str!("../app/object/extract.rs");
+    let restore = include_str!("../app/object/restore.rs");
+
+    for (source, signature, delegate) in [
+        (put, "fn execute_put_object(", "self.execute_put_object_boxed(_fs, req)"),
+        (put, "fn execute_put_object_boxed<'a>(", "self.execute_put_object_inner(_fs, req)"),
+        (put, "fn execute_put_object_inner(", "self.execute_put_object_extract(req)"),
+        (copy, "fn execute_copy_object(", "self.execute_copy_object_body_boxed(req)"),
+        (copy, "fn execute_copy_object_body_boxed(", "self.execute_copy_object_body(req)"),
+        (delete, "fn execute_delete_object(", "owner.execute_delete_object_owned(req)"),
+        (delete, "fn execute_delete_objects(", "owner.execute_delete_objects_owned(req)"),
+        (extract, "fn execute_put_object_extract(", "self.execute_put_object_extract_boxed(req)"),
+        (
+            extract,
+            "fn execute_put_object_extract_boxed(",
+            "self.execute_put_object_extract_inner(req)",
+        ),
+    ] {
+        assert!(
+            method(source, signature).contains(delegate),
+            "{signature} must call its guarded implementation"
+        );
+    }
+
+    let key_guard = "validate_table_catalog_object_mutation(&bucket, &key).await?;";
+    let store_resolution = "let Some(store) = self.object_store()";
+    for (source, signature, key_validation, mutation_boundary) in [
+        (
+            put,
+            "fn execute_put_object_inner(",
+            "validate_new_object_key(&key, request_method_name)?;",
+            "self.put_object_core(write, body, start_time).await?;",
+        ),
+        (
+            copy,
+            "fn execute_copy_object_body(",
+            "validate_new_object_key(&key, \"COPY (dest)\")?;",
+            store_resolution,
+        ),
+        (
+            delete,
+            "fn execute_delete_object_owned(",
+            "validate_object_key(&key, \"DELETE\")?;",
+            store_resolution,
+        ),
+        (
+            extract,
+            "fn execute_put_object_extract_inner(",
+            "validate_new_object_key(&key, \"PUT\")?;",
+            store_resolution,
+        ),
+    ] {
+        assert_ordered(method(source, signature), &[key_validation, key_guard, mutation_boundary]);
+    }
+    assert_ordered(
+        method(restore, "fn execute_restore_object("),
+        &[
+            "validate_table_catalog_object_mutation(&bucket, &object).await?;",
+            store_resolution,
+        ],
+    );
+    assert_ordered(
+        method(delete, "fn execute_delete_objects_owned("),
+        &[
+            "validate_object_key(&obj_id.key, \"DeleteObjects\")",
+            "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await {",
+            "continue;\n            }",
+            "authorized_deletes.push(AuthorizedDelete { idx, object });",
+        ],
+    );
+    assert_ordered(
+        method(extract, "fn execute_put_object_extract_inner("),
+        &[
+            "validate_extract_member_key(&fpath, extract_limits)",
+            "extract_try!(validate_table_catalog_object_mutation(&bucket, &fpath).await);",
+            "batch.push(ExtractPreparedMember {",
+        ],
     );
 }
 
@@ -25256,3 +25529,82 @@ async fn disable_empty_table_bucket_failure_keeps_active_catalog() {
         restarted.disable_empty_table_bucket(test_bucket_entry(bucket)).await.unwrap();
     }
 }
+
+#[tokio::test]
+async fn compaction_planner_stats_data_files_without_reading_their_bodies() {
+    let backend = TestCatalogObjectBackend::default();
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let metadata_dir = default_table_metadata_dir_path(&namespace, &table);
+    let current = default_table_metadata_file_path(&namespace, &table, "00004.metadata.json");
+    let entry = test_table_entry(bucket, &namespace, &table, current);
+    let data_dir = default_table_data_dir_path(&namespace, &table);
+    let files = [format!("{data_dir}/left.parquet"), format!("{data_dir}/right.parquet")];
+    let manifest = format!("{metadata_dir}/manifest-20.avro");
+    let manifest_list = format!("{metadata_dir}/snap-20.avro");
+    let manifest_bytes = manifest_avro_bytes(&[(&files[0], 0), (&files[1], 0)]);
+    backend
+        .seed_object(bucket, &manifest_list, manifest_list_avro_bytes(&[(&manifest, manifest_bytes.len())]))
+        .await;
+    backend.seed_object(bucket, &manifest, manifest_bytes).await;
+    for file in &files {
+        backend.seed_object(bucket, file, vec![0; 8192]).await;
+        backend.fail_next_read(bucket, file).await;
+    }
+    let metadata = serde_json::json!({
+        "current-snapshot-id": 20,
+        "snapshots": [{"snapshot-id": 20, "manifest-list": manifest_list}],
+        "refs": {"main": {"snapshot-id": 20, "type": "branch"}}
+    });
+    backend.reset_call_counts().await;
+    let report = table_compaction_planning_report(
+        &backend,
+        bucket,
+        &namespace,
+        &table,
+        &entry,
+        &metadata,
+        TableCompactionPlanningConfig {
+            target_file_size_bytes: 65536,
+            small_file_threshold_bytes: 16384,
+            min_input_files: 2,
+            max_rewrite_bytes_per_job: 65536,
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(report.status, TableCompactionPlanningStatus::RewriteCandidates);
+    assert_eq!(report.candidate_file_count, 2);
+    assert_eq!(backend.metadata_call_count().await, 2);
+    assert_eq!(backend.read_call_count().await, 2, "only the manifest list and manifest may be read");
+}
+
+#[tokio::test]
+async fn compaction_backend_read_limit_is_enforced_on_real_storage() {
+    use crate::storage_api::table::test_support::{
+        BucketOperations as _, MakeBucketOptions, init_bucket_metadata_sys, isolated_store_over_temp_disks,
+    };
+    let (_disks, store) = Box::pin(isolated_store_over_temp_disks()).await;
+    init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+    store
+        .make_bucket("bounded-compaction", &MakeBucketOptions::default())
+        .await
+        .unwrap();
+    let backend = EcStoreTableCatalogObjectBackend::new_with_strong_runtime(store, StrongTableCatalogRuntime::default());
+    backend
+        .put_object("bounded-compaction", "data.parquet", vec![7; 8192], TableCatalogPutPrecondition::IfAbsent)
+        .await
+        .unwrap();
+    let error = backend
+        .read_object_limited("bounded-compaction", "data.parquet", 1024)
+        .await
+        .expect_err("oversized stored input");
+    assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+    let data = backend
+        .read_object_limited("bounded-compaction", "data.parquet", 8192)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(data.data, vec![7; 8192]);
+}