fix(table-catalog): harden commit publication (#5779)

* fix(table-catalog): harden commit publication

* fix(table-catalog): make commit replay deterministic

* test(table-catalog): cover denied commit object reads

* fix(table-catalog): guard ref commits and order publication locks

* fix(table-catalog): close commit publication race gaps

* fix(table-catalog): close publication review gaps

* fix(table-catalog): isolate blocked strong publications

* fix(table-catalog): scale and fence commit publication

* fix(table-catalog): close publication compatibility gaps

* fix(table-catalog): clarify compatibility cleanup marker

* fix(table-catalog): repair publication hardening checks

* fix(table-catalog): align commit tests with publication fences

* fix(table-catalog): bind authorization to request context

* refactor(table-catalog): reuse internal error mapping

* test(storage): install request context for tag conditions

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/common/src/lib.rs

@@ -19,6 +19,7 @@ pub mod heal_channel;
 pub mod last_minute;
 pub mod metrics;
 mod readiness;
+pub mod table_catalog;
 
 pub use globals::*;
 pub use readiness::{GlobalReadiness, SystemStage};

crates/common/src/table_catalog.rs

@@ -0,0 +1,17 @@
+// 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.
+
+/// Cross-crate lock identity used to fence table-bucket publication against
+/// object mutations that bypass the S3 request authorization layer.
+pub const TABLE_BUCKET_PUBLICATION_LOCK_PATH: &str = ".rustfs-table/warehouses/default/publication.lock";

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -3318,6 +3318,9 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) {
             return;
         }
     };
+    if configs.table_bucket_enabled {
+        return;
+    }
     let Some(lifecycle) = configs.lifecycle else {
         return;
     };
@@ -3978,6 +3981,9 @@ async fn enqueue_expiry_for_existing_object_group(
 
 pub async fn enqueue_expiry_for_existing_objects(api: Arc<ECStore>, bucket: &str) -> Result<(), Error> {
     let configs = metadata_boundary::get_expiry_configs(&api, bucket).await?;
+    if configs.table_bucket_enabled {
+        return Ok(());
+    }
     let Some(lc) = configs.lifecycle else {
         return Ok(());
     };
@@ -4194,12 +4200,16 @@ pub async fn expire_transitioned_object(
     _src: &LcEventSrc,
     bucket_incarnation_id: Uuid,
 ) -> Result<ObjectInfo, std::io::Error> {
+    let publication_guard = lifecycle_expiry_publication_guard(&api, oi, bucket_incarnation_id)
+        .await
+        .ok_or_else(|| std::io::Error::other("lifecycle expiry is not allowed for this bucket"))?;
     let snapshot = lifecycle_delete_config_snapshot(&api, oi)
         .await
         .map_err(std::io::Error::other)?;
     let (versioned, version_suspended) = snapshot.versioning_config().delete_state(&oi.name);
     let mut opts = transitioned_object_delete_opts(oi, lc_event.action, versioned, version_suspended, bucket_incarnation_id)
         .map_err(std::io::Error::other)?;
+    opts.add_namespace_lock_guard(&publication_guard);
     opts.delete_replication_config_snapshot = Some(Arc::new(snapshot));
     //let tags = LcAuditEvent::new(src, lcEvent).Tags();
     if lc_event.action.delete_restored() {
@@ -4789,6 +4799,43 @@ pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, o
         .await
 }
 
+async fn lifecycle_expiry_publication_guard(
+    api: &ECStore,
+    oi: &ObjectInfo,
+    bucket_incarnation_id: Uuid,
+) -> Option<rustfs_lock::NamespaceLockGuard> {
+    let result = async {
+        let lock = api
+            .new_ns_lock(&oi.bucket, rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH)
+            .await?;
+        let guard = lock.get_read_lock(get_lock_acquire_timeout()).await.map_err(Error::other)?;
+        if guard.is_lock_lost() {
+            return Err(Error::other("table-bucket publication lock was lost before lifecycle delete admission"));
+        }
+        if !metadata_boundary::lifecycle_expiry_allowed(api, &oi.bucket, bucket_incarnation_id).await? {
+            return Ok(None);
+        }
+        Ok(Some(guard))
+    }
+    .await;
+    match result {
+        Ok(guard) => guard,
+        Err(err) => {
+            warn!(
+                event = EVENT_LIFECYCLE_DELETE_FAILED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                bucket = %oi.bucket,
+                object = %oi.name,
+                operation = "authorize_lifecycle_expiry",
+                error = %err,
+                "Lifecycle delete admission failed"
+            );
+            None
+        }
+    }
+}
+
 pub async fn apply_expiry_on_transitioned_object(
     api: Arc<ECStore>,
     oi: &ObjectInfo,
@@ -4812,6 +4859,9 @@ pub async fn apply_expiry_on_non_transitioned_objects(
     _src: &LcEventSrc,
     bucket_incarnation_id: Uuid,
 ) -> bool {
+    let Some(publication_guard) = lifecycle_expiry_publication_guard(&api, oi, bucket_incarnation_id).await else {
+        return false;
+    };
     let snapshot = match lifecycle_delete_config_snapshot(&api, oi).await {
         Ok(snapshot) => snapshot,
         Err(err) => {
@@ -4837,6 +4887,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
         expected_bucket_incarnation_id: Some(bucket_incarnation_id),
         ..Default::default()
     };
+    opts.add_namespace_lock_guard(&publication_guard);
 
     if lc_event.action.delete_versioned() {
         opts.version_id = oi.version_id.map(|v| v.to_string());
@@ -5033,7 +5084,7 @@ mod tests {
         cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at,
         enqueue_recovered_free_version_with_state, enqueue_transition_for_existing_objects_scoped,
         enqueue_transition_with_lifecycle, enqueue_transition_with_lifecycle_report, eval_action_from_lifecycle,
-        jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication,
+        get_lock_acquire_timeout, jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication,
         lifecycle_delete_all_versions_replication_scan, lifecycle_deleted_object, lifecycle_replication_blocks_action,
         lifecycle_rule_has_date_expiration, manual_transition_duration_elapsed, manual_transition_has_more_after_limit,
         manual_transition_recovery_progress_sink, manual_transition_version_marker, manual_transition_worker_failure_reason,
@@ -5090,7 +5141,6 @@ mod tests {
     #[cfg(feature = "test-util")]
     use crate::services::tier::warm_backend::WarmBackend as _;
     use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY};
-    #[cfg(feature = "test-util")]
     use crate::storage_api_contracts::namespace::NamespaceLocking as _;
     use crate::storage_api_contracts::{
         bucket::{BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions},
@@ -10319,6 +10369,85 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn queued_lifecycle_expiry_does_not_delete_from_table_bucket() {
+        let (_disk_paths, ecstore) = setup_test_env().await;
+        let bucket = format!("table-bucket-lifecycle-{}", Uuid::new_v4().simple());
+        let object = "tables/table-id/data/part-00001.parquet";
+        create_test_bucket(&ecstore, &bucket).await;
+
+        let mut reader = PutObjReader::from_vec(b"referenced table data".to_vec());
+        let object_info = ecstore
+            .put_object(&bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("table data object should be created");
+
+        let publication_lock = ecstore
+            .new_ns_lock(&bucket, rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH)
+            .await
+            .expect("table-bucket publication lock should be created");
+        let enable_guard = publication_lock
+            .get_write_lock(get_lock_acquire_timeout())
+            .await
+            .expect("table-bucket enablement should acquire the publication lock");
+        let expiry_store = ecstore.clone();
+        let expiry_object = object_info.clone();
+        let (expiry_started_tx, expiry_started_rx) = tokio::sync::oneshot::channel();
+        let mut expiry = tokio::spawn(async move {
+            let event = crate::bucket::lifecycle::lifecycle::Event {
+                action: IlmAction::DeleteAction,
+                ..Default::default()
+            };
+            let bucket_incarnation_id = expiry_store
+                .bucket_incarnation_id_from_disk(&expiry_object.bucket)
+                .await
+                .expect("bucket incarnation should be available");
+            expiry_started_tx.send(()).expect("lifecycle expiry start should be observed");
+            super::apply_expiry_on_non_transitioned_objects(
+                expiry_store,
+                &expiry_object,
+                &event,
+                &LcEventSrc::Scanner,
+                bucket_incarnation_id,
+            )
+            .await
+        });
+        expiry_started_rx.await.expect("lifecycle expiry should start");
+        assert!(
+            tokio::time::timeout(StdDuration::from_millis(100), &mut expiry)
+                .await
+                .is_err(),
+            "queued lifecycle expiry must wait for table-bucket enablement"
+        );
+
+        let sys = metadata_sys::bucket_metadata_sys_of(&ecstore.ctx).expect("metadata system should be initialized");
+        let sys = sys.read().await.clone();
+        let mut metadata = (*sys.get(&bucket).await.expect("bucket metadata should exist")).clone();
+        metadata.table_bucket_config_json = br#"{"enabled":true}"#.to_vec();
+        sys.persist_and_set(metadata)
+            .await
+            .expect("table bucket marker should be persisted");
+        sys.reload_from_store(&bucket)
+            .await
+            .expect("table bucket marker should become authoritative");
+        drop(enable_guard);
+        assert!(
+            !tokio::time::timeout(StdDuration::from_secs(2), expiry)
+                .await
+                .expect("queued lifecycle expiry should resume after enablement")
+                .expect("queued lifecycle expiry task should join"),
+            "a queued lifecycle task must be rejected after the bucket becomes table-enabled"
+        );
+        assert!(
+            ecstore
+                .get_object_info(&bucket, object, &ObjectOptions::default())
+                .await
+                .is_ok(),
+            "table data must remain readable after lifecycle admission rejects the delete"
+        );
+    }
+
     #[tokio::test]
     async fn existing_object_lifecycle_skips_current_expiration_for_explicit_legal_hold() {
         let lc = latest_expiration_lifecycle();

crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs

@@ -18,6 +18,7 @@ use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
 use time::OffsetDateTime;
 use uuid::Uuid;
 
+use crate::bucket::metadata::BucketMetadata;
 use crate::bucket::metadata_sys::{self, ObjectLockConfigState};
 use crate::error::{Error, Result};
 
@@ -26,16 +27,37 @@ pub(crate) struct LifecycleExpiryConfigs {
     pub(crate) lifecycle: Option<Arc<BucketLifecycleConfiguration>>,
     pub(crate) object_lock: Option<Arc<ObjectLockConfiguration>>,
     pub(crate) bucket_incarnation_id: Uuid,
+    pub(crate) table_bucket_enabled: bool,
 }
 
-pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str) -> Result<LifecycleExpiryConfigs> {
-    let bucket_incarnation_id = api.bucket_incarnation_id_from_disk(bucket).await?;
+async fn get_authoritative_metadata(
+    api: &crate::store::ECStore,
+    bucket: &str,
+    bucket_incarnation_id: Uuid,
+) -> Result<Arc<BucketMetadata>> {
     let sys = metadata_sys::bucket_metadata_sys_of(&api.ctx)?;
     let sys = sys.read().await.clone();
     let metadata = sys.get_authoritative_metadata(bucket).await?;
     if !metadata.bucket_incarnation_sidecar || metadata.bucket_incarnation_id != bucket_incarnation_id {
         return Err(Error::other(format!("bucket lifecycle metadata is not authoritative: {bucket}")));
     }
+    Ok(metadata)
+}
+
+pub(crate) async fn lifecycle_expiry_allowed(
+    api: &crate::store::ECStore,
+    bucket: &str,
+    bucket_incarnation_id: Uuid,
+) -> Result<bool> {
+    Ok(!get_authoritative_metadata(api, bucket, bucket_incarnation_id)
+        .await?
+        .table_bucket_enabled())
+}
+
+pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str) -> Result<LifecycleExpiryConfigs> {
+    let bucket_incarnation_id = api.bucket_incarnation_id_from_disk(bucket).await?;
+    let metadata = get_authoritative_metadata(api, bucket, bucket_incarnation_id).await?;
+    let table_bucket_enabled = metadata.table_bucket_enabled();
 
     let lifecycle = if metadata.lifecycle_config.is_none() && !metadata.lifecycle_config_xml.is_empty() {
         return Err(Error::other("persisted bucket lifecycle configuration is invalid"));
@@ -51,6 +73,7 @@ pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str
             lifecycle: None,
             object_lock: None,
             bucket_incarnation_id,
+            table_bucket_enabled,
         });
     }
     let object_lock = match metadata_sys::object_lock_config_state_from_authoritative_metadata(&metadata)? {
@@ -65,6 +88,7 @@ pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str
         lifecycle,
         object_lock,
         bucket_incarnation_id,
+        table_bucket_enabled,
     })
 }
 
@@ -125,6 +149,7 @@ mod tests {
         let lifecycle = lifecycle_config();
         metadata.lifecycle_config_xml = crate::bucket::utils::serialize(&lifecycle).unwrap();
         metadata.lifecycle_config = Some(lifecycle);
+        metadata.table_bucket_config_json = br#"{"enabled":true}"#.to_vec();
         metadata_sys::set_new_bucket_metadata_in(&store_a.ctx, metadata)
             .await
             .unwrap();
@@ -132,7 +157,14 @@ mod tests {
             .await
             .unwrap();
 
-        assert!(get_expiry_configs(&store_a, bucket).await.unwrap().lifecycle.is_some());
+        let configs = get_expiry_configs(&store_a, bucket).await.unwrap();
+        assert!(configs.lifecycle.is_some());
+        assert!(configs.table_bucket_enabled);
+        assert!(
+            !lifecycle_expiry_allowed(&store_a, bucket, configs.bucket_incarnation_id)
+                .await
+                .unwrap()
+        );
         assert!(get_expiry_configs(&store_b, bucket).await.unwrap().lifecycle.is_none());
     }
 }

crates/ecstore/src/bucket/metadata_sys.rs

@@ -288,6 +288,13 @@ pub(crate) fn bucket_metadata_sys_of(ctx: &crate::runtime::instance::InstanceCon
     get_bucket_metadata_sys()
 }
 
+pub(crate) fn require_bucket_metadata_sys_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+) -> Result<Arc<RwLock<BucketMetadataSys>>> {
+    ctx.bucket_metadata_sys()
+        .ok_or_else(|| Error::other("bucket metadata sys not initialized for this instance"))
+}
+
 pub(crate) async fn object_store_in(ctx: &crate::runtime::instance::InstanceContext) -> Result<Arc<ECStore>> {
     let sys = bucket_metadata_sys_of(ctx)?;
     Ok(sys.read().await.api.clone())
@@ -376,6 +383,15 @@ pub async fn update(bucket: &str, config_file: &str, data: Vec<u8>) -> Result<Of
     Box::pin(update_with_sys(get_bucket_metadata_sys()?, bucket, config_file, data)).await
 }
 
+pub(crate) async fn update_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    config_file: &str,
+    data: Vec<u8>,
+) -> Result<OffsetDateTime> {
+    Box::pin(update_with_sys(require_bucket_metadata_sys_in(ctx)?, bucket, config_file, data)).await
+}
+
 pub async fn delete(bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
     delete_with_sys(get_bucket_metadata_sys()?, bucket, config_file).await
 }

crates/ecstore/src/bucket/policy_sys.rs

@@ -14,6 +14,7 @@
 
 use super::metadata_sys::get_bucket_metadata_sys;
 use crate::error::{Result, StorageError};
+use crate::store::ECStore;
 use rustfs_policy::policy::{BucketPolicy, BucketPolicyArgs};
 
 pub struct PolicySys {}
@@ -27,6 +28,10 @@ impl PolicySys {
         Self::is_allowed_with_policy(args, Self::get(args.bucket).await).await
     }
 
+    pub async fn try_is_allowed_for_store(store: &ECStore, args: &BucketPolicyArgs<'_>) -> Result<bool> {
+        Self::is_allowed_with_policy(args, store.get_bucket_policy(args.bucket).await.map(|(policy, _)| policy)).await
+    }
+
     async fn is_allowed_with_policy(args: &BucketPolicyArgs<'_>, policy: Result<BucketPolicy>) -> Result<bool> {
         match policy {
             Ok(policy) => Ok(policy.is_allowed(args).await),

crates/ecstore/src/store/bucket.rs

@@ -23,6 +23,7 @@ use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, SRBucketDeleteOp};
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use futures::stream::{self, StreamExt};
+use rustfs_policy::policy::BucketPolicy;
 use std::collections::BTreeMap;
 use std::future::Future;
 
@@ -153,6 +154,31 @@ where
 }
 
 impl ECStore {
+    pub async fn get_bucket_metadata(&self, bucket: &str) -> Result<Arc<BucketMetadata>> {
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+        sys.read().await.get(bucket).await
+    }
+
+    pub async fn get_bucket_policy(&self, bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> {
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+        sys.read().await.get_bucket_policy(bucket).await
+    }
+
+    pub async fn get_bucket_policy_raw(&self, bucket: &str) -> Result<(String, OffsetDateTime)> {
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+        sys.read().await.get_bucket_policy_raw(bucket).await
+    }
+
+    pub async fn restricts_public_bucket_access(&self, bucket: &str) -> Result<bool> {
+        let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
+        let (config, _) = sys.read().await.get_public_access_block_config(bucket).await?;
+        Ok(config.restrict_public_buckets.unwrap_or(false))
+    }
+
+    pub async fn update_bucket_metadata_config(&self, bucket: &str, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
+        metadata_sys::update_in(&self.ctx, bucket, config_file, data).await
+    }
+
     pub async fn bucket_incarnation_id(&self, bucket: &str) -> Result<Uuid> {
         metadata_sys::get_cached_bucket_incarnation_id_in(&self.ctx, bucket).await
     }
@@ -1077,6 +1103,26 @@ mod tests {
         (temp_dir, ecstore)
     }
 
+    #[tokio::test]
+    async fn request_metadata_methods_fail_closed_before_instance_initialization() {
+        let (_temp_dir, store) = setup_multi_pool_scanner_listing_test_env().await;
+
+        let expected = "bucket metadata sys not initialized for this instance";
+        let errors = [
+            store.get_bucket_metadata("bucket").await.unwrap_err(),
+            store.get_bucket_policy("bucket").await.unwrap_err(),
+            store.get_bucket_policy_raw("bucket").await.unwrap_err(),
+            store.restricts_public_bucket_access("bucket").await.unwrap_err(),
+            store
+                .update_bucket_metadata_config("bucket", crate::bucket::metadata::BUCKET_POLICY_CONFIG, Vec::new())
+                .await
+                .unwrap_err(),
+        ];
+        for error in errors {
+            assert_eq!(error.to_string(), format!("Io error: {expected}"));
+        }
+    }
+
     async fn create_bucket_with_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str) {
         let generation_before_make = ecstore.scanner_namespace_mutation_generation();
         ecstore

docs/architecture/compat-cleanup-register.md

@@ -12,6 +12,7 @@ for later deletion.
 
 ## Open Items
 
+- `table-publication-fence-v1` S3 Tables publication fencing: nodes that predate table and table-bucket publication fences can mutate live files while a new node is publishing a catalog pointer. New nodes retain exact object guards until the operator confirms that every serving node uses the new fences. Fleet confirmation also requires non-overlapping active warehouse prefixes and lifecycle workers that exclude table buckets. Remove the exact live-file fallback and the fleet-confirmation gate after the minimum supported RustFS release acquires table fences for registered-table mutations and table-bucket fences for unresolved-prefix mutations.
 - `cross-pool-fence-v1` authenticated unsupported advertisement: predeployment servers recognize the versioned cross-pool fence capability probe but report support version 0, allowing a later all-peer probe to distinguish predeployment nodes without activating a second lock domain. Replace the unsupported advertisement only when composite lock acquisition, a cluster-wide activation fence, complete fleet proof, commit-time proof revalidation, and fail-closed revocation ship together.
 - `table-catalog-dotted-namespace` Iceberg REST namespace path compatibility: existing RustFS clients use dotted namespace paths, while the standard multi-level contract uses the URL-encoded unit separator `%1F`. New servers accept both forms so a rolling upgrade does not invalidate existing catalog configuration. Remove the dotted fallback after the minimum supported RustFS release advertises `%1F` and all supported clients have refreshed their catalog configuration.
 - `rustfs-5509` FileInfo positional MessagePack decoding: beta.11 serialized 28 fields, while beta.12 inserted transition-version fields in the middle and serialized an incompatible 30-field array. New releases write named maps and retain readers for both shipped array layouts so direct and rolling upgrades can read either release. Remove the positional-array readers after every supported direct-upgrade release writes named maps and no retained RPC payload can contain a pre-map FileInfo array.

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

@@ -92,6 +92,7 @@ catalog extension.
 |---|---|---|
 | Metadata retention dry-run | Supported | Reports retained metadata and deletion candidates without moving the table pointer. |
 | Metadata cleanup delete | Supported | Deletes only candidates that pass the safety window and current-pointer checks. |
+| Ordinary bucket lifecycle expiry | Disabled for table buckets | Table bucket objects are excluded from ordinary lifecycle expiration, including already queued expiry work. Snapshot expiration and orphan cleanup remain catalog maintenance operations so referenced Iceberg files cannot be deleted outside publication fencing. |
 | Snapshot expiration planning | Supported | Produces expiration plans with retained and candidate snapshots. |
 | Snapshot expiration commit | Preview / controlled | Can manually commit safe snapshot expiration through the catalog. Stale plans fail closed. |
 | Manifest/data/delete reachability cleanup | Supported | Reads manifest-list and manifest Avro references, reports reachable objects, and deletes only unreferenced table objects that pass the safety window. |
@@ -112,6 +113,7 @@ catalog extension.
 |---|---|---|
 | Single-table CAS | Supported | The table pointer advances only through expected-token and expected-metadata-location validation. |
 | Idempotent retry | Supported | Repeated commit IDs can return the already finalized result or surface recoverable finalization gaps. |
+| Commit publication fencing | Supported with rolling-upgrade gate | Existing deployments retain exact object guards so older writers cannot mutate referenced files during publication. Set `RUSTFS_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED=true` only after every serving node supports table and table-bucket publication fences. In scalable mode, active table warehouse prefixes must not overlap, ordinary lifecycle expiry remains disabled for table buckets, and first enablement, first publication, drop, and warehouse relocation are serialized by the table-bucket fence. |
 | Post-CAS finalization recovery | Supported | Diagnostics and recovery can repair stale or missing idempotency indexes without changing the current table pointer. |
 | Catalog export | Supported | Exposes table state, commit recovery state, and backing migration information for operator inspection. |
 | Strong backing state transfer | Supported | Object-backed table bucket, namespace, table, view, commit-log, and idempotency state can be materialized into the durable strong snapshot. The transfer is deterministic, ETag-CAS protected, idempotent after an interrupted finalization, and fails closed when a table or view has no owning namespace entry. |

rustfs/src/admin/auth.rs

@@ -188,6 +188,25 @@ pub async fn validate_admin_request_with_bucket_object(
     evaluate_admin_actions(iam_store, &ctx, &actions, resource.bucket, resource.object).await
 }
 
+pub(crate) async fn validate_admin_action_with_bucket_object_for_iam<S: Store>(
+    iam_store: Arc<IamSys<S>>,
+    headers: &HeaderMap,
+    cred: &Credentials,
+    is_owner: bool,
+    action: Action,
+    remote_addr: Option<std::net::SocketAddr>,
+    resource: AdminResourceScope<'_>,
+) -> S3Result<()> {
+    let ctx = AuthContext {
+        headers,
+        cred,
+        is_owner,
+        deny_only: false,
+        remote_addr,
+    };
+    evaluate_admin_actions(iam_store, &ctx, &[action], resource.bucket, resource.object).await
+}
+
 /// Admin gate for KMS endpoints that act on one key.
 ///
 /// `key_id` is the identifier as requested (before any alias resolution), so a

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

@@ -33,8 +33,12 @@ impl Operation for EnableTableBucketHandler {
         let warehouse = warehouse_from_params(&params)?;
         let resource = TableCatalogResource::warehouse(&warehouse);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?;
-        let store = table_catalog_store()?;
-        let response = enable_table_bucket_response(&store, &warehouse).await?;
+        let backend = table_catalog_backend_from_extensions(&req.extensions)?;
+        let object_store = runtime_sources::object_store_from_req(&req)
+            .ok_or_else(|| table_catalog_internal_error("request object store is not initialized"))?;
+        let store = table_catalog_store_from_backend(backend.clone())?;
+        let publication = TableCommitObjectBackend::preauthorized(backend);
+        let response = enable_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -47,8 +51,8 @@ impl Operation for GetTableBucketHandler {
         let warehouse = warehouse_from_params(&params)?;
         let resource = TableCatalogResource::warehouse(&warehouse);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableBucketAction).await?;
-        let store = table_catalog_store()?;
-        let enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
+        let enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let response = table_bucket_response(&store, &warehouse, enabled).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -62,8 +66,8 @@ impl Operation for GetTableCatalogMigrationHandler {
         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(&warehouse).await?;
-        let store = table_catalog_object_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let started = Instant::now();
         let result = store
             .plan_durable_strong_backing_migration(&warehouse)
@@ -82,8 +86,8 @@ impl Operation for MaterializeTableCatalogMigrationHandler {
     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(&warehouse).await?;
-        let store = table_catalog_object_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let started = Instant::now();
         let result = store
             .materialize_durable_strong_backing_migration(&warehouse)
@@ -102,8 +106,8 @@ impl Operation for CancelTableCatalogMigrationHandler {
     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(&warehouse).await?;
-        let store = table_catalog_object_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let started = Instant::now();
         let result = store
             .cancel_durable_strong_backing_migration(&warehouse)
@@ -125,8 +129,8 @@ impl Operation for ExternalCatalogBridgeHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_object_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let response = external_catalog_bridge_response(&store, &warehouse, &namespace, &table).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -142,9 +146,9 @@ impl Operation for PutExternalCatalogBridgeHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<ExternalCatalogBridgeRequest>(req.input).await?;
-        let store = table_catalog_object_store()?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let response = put_external_catalog_bridge_response(&store, &warehouse, &namespace, &table, request).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -154,15 +158,16 @@ pub struct SyncExternalCatalogBridgeHandler {}
 
 #[async_trait::async_trait]
 impl Operation for SyncExternalCatalogBridgeHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let metadata_backend = table_catalog_backend()?;
-        let store = table_catalog_object_store()?;
+        let principal =
+            authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         if store
             .load_table(&warehouse, &namespace.public_name(), &table)
             .await
@@ -171,18 +176,21 @@ impl Operation for SyncExternalCatalogBridgeHandler {
         {
             authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
         }
-        let request = read_json_body::<ExternalCatalogBridgeSyncRequest>(req.input).await?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
-        let response = sync_external_catalog_bridge_response(
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<ExternalCatalogBridgeSyncRequest>(std::mem::take(&mut req.input)).await?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = sync_external_catalog_bridge_response(
             &store,
-            &metadata_backend,
+            &commit_backend,
             &warehouse,
             &namespace,
             &table,
             request,
             table_bucket_enabled,
         )
-        .await?;
+        .await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -23,12 +23,12 @@ impl Operation for RestLoadCredentialsHandler {
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCredentialsAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let principal = table_catalog_request_principal(&req).await?;
-        let store = table_catalog_store()?;
-        let issuer = IamTableCredentialIssuer::from_env();
-        let response = load_credentials_response(&store, &warehouse, &namespace, &table, &issuer, Some(&principal)).await?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCredentialsAction).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
+        let issuer = IamTableCredentialIssuer::from_request(&req)?;
+        let response =
+            load_credentials_response(&store, &warehouse, &namespace, &table, &issuer, Some(&principal.credentials)).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -24,10 +24,10 @@ impl Operation for RestTableMetadataMaintenanceHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<TableMetadataMaintenanceRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
-        let store = table_catalog_object_store()?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
+        let store = table_catalog_object_store_from_extensions(&req.extensions)?;
         let response =
             table_metadata_maintenance_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
         build_json_response(StatusCode::OK, &response)
@@ -44,8 +44,8 @@ impl Operation for GetTableMaintenanceConfigHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableLifecycleAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .get_table_maintenance_config(&warehouse, &namespace.public_name(), &table)
             .await
@@ -64,9 +64,9 @@ impl Operation for PutTableMaintenanceConfigHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableLifecycleAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<crate::table_catalog::TableMaintenanceConfig>(req.input).await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .put_table_maintenance_config(&warehouse, &namespace.public_name(), &table, request)
             .await
@@ -86,8 +86,8 @@ impl Operation for GetTableMaintenanceJobHandler {
         let job = job_id_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableLifecycleAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let Some(response) = store
             .get_table_metadata_maintenance_report(&warehouse, &namespace.public_name(), &table, &job)
             .await
@@ -109,8 +109,8 @@ impl Operation for GetTableMaintenanceSchedulerHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableLifecycleAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .get_table_maintenance_scheduler_report(&warehouse, &namespace.public_name(), &table)
             .await
@@ -129,9 +129,9 @@ impl Operation for RunTableMaintenanceSchedulerHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body_or_default::<TableMaintenanceSchedulerRunRequest>(req.input).await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .run_table_maintenance_scheduler_once(
                 &warehouse,
@@ -155,9 +155,9 @@ impl Operation for RunTableMaintenanceWorkerHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<TableMaintenanceWorkerRunRequest>(req.input).await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .run_table_metadata_maintenance_worker_once(
                 &warehouse,
@@ -182,9 +182,9 @@ impl Operation for HeartbeatTableMaintenanceJobHandler {
         let job = job_id_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<TableMaintenanceHeartbeatRequest>(req.input).await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .heartbeat_table_metadata_maintenance_job(
                 &warehouse,
@@ -211,9 +211,9 @@ impl Operation for TableMaintenanceQuarantineHandler {
         let job = job_id_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<crate::table_catalog::TableMaintenanceQuarantineOperationRequest>(req.input).await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = store
             .apply_table_maintenance_quarantine_operation(&warehouse, &namespace.public_name(), &table, &job, request)
             .await

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

@@ -26,8 +26,8 @@ impl Operation for RestListNamespacesHandler {
             None => TableCatalogResource::warehouse(&warehouse),
         };
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableNamespaceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = list_namespaces_response(&store, &warehouse, parent.as_ref(), &req.uri).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -49,8 +49,8 @@ impl Operation for RestCreateNamespaceHandler {
             "namespace creation",
         )
         .await?;
-        let store = table_catalog_store()?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let response = create_namespace_response(&store, &warehouse, request, table_bucket_enabled).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -65,8 +65,8 @@ impl Operation for RestGetNamespaceHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableNamespaceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = get_namespace_response(&store, &warehouse, &namespace).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -81,8 +81,8 @@ impl Operation for RestDropNamespaceHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableNamespaceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         drop_namespace_in_store(&store, &warehouse, &namespace.public_name()).await?;
         Ok(empty_response(StatusCode::NO_CONTENT))
     }
@@ -97,7 +97,7 @@ impl Operation for RestUpdateNamespacePropertiesHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::UpdateTableNamespacePropertiesAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_bounded_json_body::<UpdateNamespacePropertiesRequest>(
             &req.headers,
             req.input,
@@ -106,7 +106,7 @@ impl Operation for RestUpdateNamespacePropertiesHandler {
             "namespace properties",
         )
         .await?;
-        let store = table_catalog_store()?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = update_namespace_properties_response(&store, &warehouse, &namespace, request).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -121,8 +121,8 @@ impl Operation for RestNamespaceExistsHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableNamespaceAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         Ok(empty_response(namespace_exists_status(&store, &warehouse, &namespace).await?))
     }
 }

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

@@ -24,8 +24,8 @@ impl Operation for ListTableRefsHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let metadata_backend = table_catalog_backend()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let response = table_refs_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
         build_json_response(StatusCode::OK, &response)
@@ -36,19 +36,21 @@ pub struct PutTableRefHandler {}
 
 #[async_trait::async_trait]
 impl Operation for PutTableRefHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let ref_name = ref_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let request = read_json_body::<PutTableRefRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let request = read_json_body::<PutTableRefRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response =
-            put_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = put_table_ref_response(&store, &commit_backend, &warehouse, &namespace, &table, &ref_name, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -57,19 +59,21 @@ pub struct DeleteTableRefHandler {}
 
 #[async_trait::async_trait]
 impl Operation for DeleteTableRefHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let ref_name = ref_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let request = read_json_body_or_default::<DeleteTableRefRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let request = read_json_body_or_default::<DeleteTableRefRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response =
-            delete_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = delete_table_ref_response(&store, &commit_backend, &warehouse, &namespace, &table, &ref_name, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -23,8 +23,8 @@ impl Operation for RestListTablesHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = list_tables_response(&store, &warehouse, &namespace, &req.uri).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -40,11 +40,12 @@ impl Operation for RestCreateTableHandler {
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?;
         let request = read_json_body::<CreateTableRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let commit_backend = TableCommitObjectBackend::preauthorized(metadata_backend);
         let response =
-            create_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
+            create_table_response(&store, &commit_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -53,17 +54,20 @@ pub struct RestRegisterTableHandler {}
 
 #[async_trait::async_trait]
 impl Operation for RestRegisterTableHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
-        let request = read_json_body::<RegisterTableRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<RegisterTableRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
-        let response =
-            register_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result =
+            register_table_response(&store, &commit_backend, &warehouse, &namespace, request, table_bucket_enabled).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -78,8 +82,8 @@ impl Operation for RestLoadTableHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let metadata_backend = table_catalog_backend()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
         build_json_response(StatusCode::OK, &response)
@@ -96,8 +100,8 @@ impl Operation for RestTableExistsHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         Ok(empty_response(table_exists_status(&store, &warehouse, &namespace, &table).await?))
     }
 }
@@ -106,17 +110,20 @@ pub struct RestCommitTableHandler {}
 
 #[async_trait::async_trait]
 impl Operation for RestCommitTableHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let request = read_json_body::<RestCommitTableRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let request = read_json_body::<RestCommitTableRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response = commit_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = commit_table_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -131,8 +138,8 @@ impl Operation for RestDropTableHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         drop_table_in_store(&store, &warehouse, &namespace, &table).await?;
         Ok(empty_response(StatusCode::NO_CONTENT))
     }
@@ -148,8 +155,8 @@ impl Operation for GetTableMetadataLocationHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataLocationAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = get_table_metadata_location_response(&store, &warehouse, &namespace, &table).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -159,18 +166,22 @@ pub struct UpdateTableMetadataLocationHandler {}
 
 #[async_trait::async_trait]
 impl Operation for UpdateTableMetadataLocationHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let request = read_json_body::<UpdateTableMetadataLocationRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal =
+            authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let request = read_json_body::<UpdateTableMetadataLocationRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response =
-            update_table_metadata_location_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result =
+            update_table_metadata_location_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -185,8 +196,8 @@ impl Operation for ExportTableCatalogHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let started = Instant::now();
         let result = store
             .export_table_catalog_entry(&warehouse, &namespace.public_name(), &table)
@@ -202,19 +213,21 @@ pub struct ImportTableCatalogHandler {}
 
 #[async_trait::async_trait]
 impl Operation for ImportTableCatalogHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
-        let request = read_json_body::<CatalogImportRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let request = read_json_body::<CatalogImportRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
-        let response =
-            catalog_import_response(&store, &metadata_backend, &warehouse, &namespace, &table, request, table_bucket_enabled)
-                .await?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result =
+            catalog_import_response(&store, &commit_backend, &warehouse, &namespace, &table, request, table_bucket_enabled).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -229,8 +242,8 @@ impl Operation for GetTableCatalogDiagnosticsHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let config = store
             .get_table_maintenance_config(&warehouse, &namespace.public_name(), &table)
             .await
@@ -263,8 +276,8 @@ impl Operation for RecoverTableCatalogHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let started = Instant::now();
         let result = store
             .recover_table_commits(&warehouse, &namespace.public_name(), &table)
@@ -280,17 +293,20 @@ pub struct RollbackTableCatalogHandler {}
 
 #[async_trait::async_trait]
 impl Operation for RollbackTableCatalogHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let request = read_json_body::<RollbackTableRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let request = read_json_body::<RollbackTableRequest>(std::mem::take(&mut req.input)).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response = rollback_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = rollback_table_response(&store, &commit_backend, &warehouse, &namespace, &table, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -23,8 +23,8 @@ impl Operation for RestListViewsHandler {
         let namespace = namespace_from_params(&params)?;
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         let response = list_views_response(&store, &warehouse, &namespace, &req.uri).await?;
         build_json_response(StatusCode::OK, &response)
     }
@@ -40,9 +40,9 @@ impl Operation for RestCreateViewHandler {
         let resource = TableCatalogResource::namespace(&warehouse, &namespace);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?;
         let request = read_json_body::<CreateViewRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?;
+        let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let response =
             create_view_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
         build_json_response(StatusCode::OK, &response)
@@ -59,8 +59,8 @@ impl Operation for RestLoadViewHandler {
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let metadata_backend = table_catalog_backend()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let response = load_view_response(&store, &metadata_backend, &warehouse, &namespace, &view).await?;
         build_json_response(StatusCode::OK, &response)
@@ -77,8 +77,8 @@ impl Operation for RestViewExistsHandler {
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         Ok(empty_response(view_exists_status(&store, &warehouse, &namespace, &view).await?))
     }
 }
@@ -93,9 +93,9 @@ impl Operation for RestReplaceViewHandler {
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_json_body::<RestCommitViewRequest>(req.input).await?;
-        let metadata_backend = table_catalog_backend()?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let response = replace_view_response(&store, &metadata_backend, &warehouse, &namespace, &view, request).await?;
         build_json_response(StatusCode::OK, &response)
@@ -112,8 +112,8 @@ impl Operation for RestDropViewHandler {
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableAction).await?;
-        ensure_table_bucket_enabled(&warehouse).await?;
-        let store = table_catalog_store()?;
+        ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let store = table_catalog_store_from_extensions(&req.extensions)?;
         drop_view_in_store(&store, &warehouse, &namespace, &view).await?;
         Ok(empty_response(StatusCode::NO_CONTENT))
     }

rustfs/src/admin/storage_api.rs

@@ -793,7 +793,9 @@ pub(crate) mod data_usage {
 }
 
 pub(crate) mod access {
-    pub(crate) use crate::storage::storage_api::access_consumer::{ReqInfo, authorize_request};
+    pub(crate) use crate::storage::storage_api::access_consumer::{
+        ReqInfo, authorize_internal_object_request, authorize_request,
+    };
     pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
 }
 

rustfs/src/auth.rs

@@ -451,7 +451,7 @@ pub async fn check_key_valid_with_context(
         cred = u.credentials;
     }
 
-    let claims = check_claims_from_token(session_token, &cred)
+    let claims = check_claims_from_token_with_context(session_token, &cred, ctx)
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {e}")))?;
 
     cred.claims = if !claims.is_empty() { Some(claims) } else { None };
@@ -461,6 +461,14 @@ pub async fn check_key_valid_with_context(
 }
 
 pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result<HashMap<String, Value>> {
+    check_claims_from_token_with_context(token, cred, None)
+}
+
+fn check_claims_from_token_with_context(
+    token: &str,
+    cred: &Credentials,
+    ctx: Option<&AppContext>,
+) -> S3Result<HashMap<String, Value>> {
     if !token.is_empty() && cred.access_key.is_empty() {
         return Err(s3_error!(InvalidRequest, "no access key"));
     }
@@ -481,7 +489,11 @@ pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result<Hash
         return Err(s3_error!(InvalidRequest, "invalid access key is temp and expired"));
     }
 
-    let Some(sys_cred) = current_action_credentials() else {
+    let sys_cred = match ctx {
+        Some(context) => context.action_credentials().get(),
+        None => current_action_credentials(),
+    };
+    let Some(sys_cred) = sys_cred else {
         return Err(s3_error!(InternalError, "action cred not init"));
     };
 
@@ -1059,14 +1071,47 @@ pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str>
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::runtime_sources::{IamInterface, KmsInterface};
     use http::{HeaderMap, HeaderValue, Uri};
     use rustfs_credentials::Credentials;
+    use rustfs_iam::{
+        store::{
+            Store as _,
+            object::{IAM_CONFIG_PREFIX, ObjectStore},
+        },
+        sys::IamSys,
+    };
+    use rustfs_kms::KmsServiceManager;
+    use rustfs_policy::auth::get_new_credentials_with_metadata;
     use rustfs_trusted_proxies::ValidationMode;
     use s3s::auth::SecretKey;
     use serde_json::json;
     use std::collections::HashMap;
+    use std::sync::Arc;
     use time::OffsetDateTime;
 
+    struct ContextIam {
+        handle: Arc<IamSys<ObjectStore>>,
+    }
+
+    impl IamInterface for ContextIam {
+        fn handle(&self) -> Arc<IamSys<ObjectStore>> {
+            self.handle.clone()
+        }
+
+        fn is_ready(&self) -> bool {
+            true
+        }
+    }
+
+    struct TestKms;
+
+    impl KmsInterface for TestKms {
+        fn handle(&self) -> Arc<KmsServiceManager> {
+            Arc::new(KmsServiceManager::new())
+        }
+    }
+
     fn create_test_credentials() -> Credentials {
         Credentials {
             access_key: "test-access-key".to_string(),
@@ -1268,6 +1313,59 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn check_claims_uses_the_explicit_context_signing_secret() {
+        let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+        ObjectStore::new(store.clone())
+            .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
+            .await
+            .expect("seed request IAM format");
+        let iam = rustfs_iam::build_iam_sys(store.clone())
+            .await
+            .expect("request IAM should initialize");
+        let matching = AppContext::new(store.clone(), Arc::new(ContextIam { handle: iam.clone() }), Arc::new(TestKms));
+        let mismatching = AppContext::new(store, Arc::new(ContextIam { handle: iam.clone() }), Arc::new(TestKms));
+        assert!(matching.publish_action_credentials(Credentials {
+            access_key: "matching-root".to_string(),
+            secret_key: "matching-signing-secret".to_string(),
+            status: "on".to_string(),
+            ..Default::default()
+        }));
+        assert!(mismatching.publish_action_credentials(Credentials {
+            access_key: "mismatching-root".to_string(),
+            secret_key: "mismatching-signing-secret".to_string(),
+            status: "on".to_string(),
+            ..Default::default()
+        }));
+        let claims = HashMap::from([
+            (
+                "exp".to_string(),
+                json!((OffsetDateTime::now_utc() + time::Duration::minutes(5)).unix_timestamp()),
+            ),
+            ("context".to_string(), json!("matching")),
+        ]);
+        let mut credential = get_new_credentials_with_metadata(&claims, "matching-signing-secret")
+            .expect("temporary credentials should be generated");
+        credential.parent_user = "matching-root".to_string();
+        iam.set_temp_user(&credential.access_key, &credential, None)
+            .await
+            .expect("temporary credentials should be stored in request IAM");
+
+        let (verified, _) = check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&matching))
+            .await
+            .expect("the matching request context should verify the token");
+        assert_eq!(
+            verified.claims.as_ref().and_then(|claims| claims.get("context")),
+            Some(&json!("matching"))
+        );
+        assert!(
+            check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&mismatching))
+                .await
+                .is_err(),
+            "a different server context must not validate the token"
+        );
+    }
+
     #[test]
     fn test_get_session_token_from_header() {
         let mut headers = HeaderMap::new();

rustfs/src/storage/ecfs.rs

@@ -34,7 +34,7 @@ use crate::storage::storage_api::ecfs_consumer::contract::{
 use crate::storage::storage_api::ecfs_consumer::object_lock::{
     parse_object_lock_legal_hold, parse_object_lock_retention, validate_bucket_object_lock_enabled,
 };
-use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
+use crate::storage::storage_api::runtime_sources_consumer::{ECStore, runtime_sources};
 use crate::table_catalog;
 use http::StatusCode;
 use metrics::{counter, histogram};
@@ -128,6 +128,15 @@ impl FS {
         let Some(store) = self.server_ctx.object_store() else {
             return Ok(std::collections::HashMap::new());
         };
+        Self::get_object_tag_conditions_for_policy_from_store(store.as_ref(), bucket, object, version_id).await
+    }
+
+    pub(crate) async fn get_object_tag_conditions_for_policy_from_store(
+        store: &ECStore,
+        bucket: &str,
+        object: &str,
+        version_id: Option<&str>,
+    ) -> S3Result<std::collections::HashMap<String, Vec<String>>> {
         let opts = ObjectOptions {
             version_id: version_id.map(String::from),
             ..Default::default()

rustfs/src/storage/mod.rs

@@ -67,15 +67,14 @@ pub(crate) use storage_api::{
     ecstore_layout, ecstore_metrics, ecstore_notification, ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk,
     ecstore_storage, ecstore_tier, encode_tags, find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config,
     get_bucket_logging_config, get_bucket_metadata, get_bucket_notification_config, get_bucket_object_lock_config,
-    get_bucket_policy_raw, get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config,
-    get_bucket_website_config, get_local_server_property, get_lock_acquire_timeout, get_public_access_block_config,
-    head_prefix_consumer, helper_consumer, init_background_replication, init_bucket_metadata_sys, init_ecstore_config,
-    init_local_disks_with_instance_ctx, init_lock_clients, is_err_bucket_not_found, is_err_object_not_found,
-    is_err_version_not_found, is_valid_storage_class, options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config,
-    record_replication_proxy, rpc_consumer, runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash,
-    to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
-    try_migrate_server_config, update_bucket_metadata_config, update_bucket_metadata_config_if_incarnation, verify_rpc_signature,
-    wrap_reader,
+    get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config,
+    get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer, init_background_replication,
+    init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
+    is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class, options_consumer,
+    prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer, runtime_sources_consumer,
+    s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities,
+    try_migrate_bucket_metadata, try_migrate_iam_config, try_migrate_server_config, update_bucket_metadata_config,
+    update_bucket_metadata_config_if_incarnation, verify_rpc_signature, wrap_reader,
 };
 
 #[cfg(test)]

rustfs/src/storage/storage_api.rs

@@ -109,9 +109,9 @@ pub(crate) use super::sse::{
 pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
         PostObjectRequestMarker, ReqInfo, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
-        authorize_request, bucket_config_mutation_incarnation, has_bypass_governance_header, load_bucket_generation_from_store,
-        log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization, recursive_force_delete_is_authorized,
-        replication_request_authorized, req_info_mut, req_info_ref,
+        authorize_internal_object_request, authorize_request, bucket_config_mutation_incarnation, has_bypass_governance_header,
+        load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization,
+        recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, req_info_ref,
     };
 }
 
@@ -1448,10 +1448,6 @@ pub(crate) async fn get_bucket_accelerate_config(
     ecstore_bucket::metadata_sys::get_accelerate_config(bucket).await
 }
 
-pub(crate) async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, time::OffsetDateTime)> {
-    ecstore_bucket::metadata_sys::get_bucket_policy_raw(bucket).await
-}
-
 pub(crate) async fn get_bucket_cors_config(bucket: &str) -> Result<(s3s::dto::CORSConfiguration, time::OffsetDateTime)> {
     ecstore_bucket::metadata_sys::get_cors_config(bucket).await
 }
@@ -1466,12 +1462,6 @@ pub(crate) async fn get_bucket_object_lock_config(
     ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
 }
 
-pub(crate) async fn get_public_access_block_config(
-    bucket: &str,
-) -> Result<(s3s::dto::PublicAccessBlockConfiguration, time::OffsetDateTime)> {
-    ecstore_bucket::metadata_sys::get_public_access_block_config(bucket).await
-}
-
 pub(crate) async fn get_bucket_replication_config(
     bucket: &str,
 ) -> Result<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {

rustfs/src/table_catalog/iceberg/commit.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::{BTreeMap, BTreeSet};
+
 use super::super::*;
 
 pub(crate) fn commit_log_matches_request(commit_log: &CommitLogEntry, request: &TableCommitRequest, table_id: &str) -> bool {
@@ -39,6 +41,49 @@ pub(crate) fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitL
         && table.version_token == commit_log.expected_version_token
 }
 
+pub(crate) struct TableCommitHistoryIndex<'a> {
+    table_id: &'a str,
+    reachable_states: BTreeSet<(&'a str, &'a str)>,
+}
+
+impl<'a> TableCommitHistoryIndex<'a> {
+    pub(crate) fn new(table: &'a TableEntry, commits: impl IntoIterator<Item = &'a CommitLogEntry>) -> Self {
+        let mut by_new_state = BTreeMap::<(&str, &str), Option<(&str, &str)>>::new();
+        for commit in commits
+            .into_iter()
+            .filter(|commit| commit.table_id == table.table_id && !matches!(commit.status, CommitLogStatus::Failed))
+        {
+            let key = (commit.new_metadata_location.as_str(), commit.new_version_token.as_str());
+            let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
+            by_new_state
+                .entry(key)
+                .and_modify(|candidate| *candidate = None)
+                .or_insert(Some(previous));
+        }
+
+        let mut reachable_states = BTreeSet::new();
+        let mut state = (table.metadata_location.as_str(), table.version_token.as_str());
+        while reachable_states.insert(state) {
+            let Some(Some(previous)) = by_new_state.get(&state) else {
+                break;
+            };
+            state = *previous;
+        }
+        Self {
+            table_id: &table.table_id,
+            reachable_states,
+        }
+    }
+
+    pub(crate) fn proves_committed(&self, target: &CommitLogEntry) -> bool {
+        self.table_id == target.table_id.as_str()
+            && !matches!(target.status, CommitLogStatus::Failed)
+            && self
+                .reachable_states
+                .contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
+    }
+}
+
 pub(crate) fn table_catalog_recovery_summary(
     metadata_status: &TableMetadataPointerStatus,
     commit_recovery: &TableCommitRecoveryReport,
@@ -76,7 +121,7 @@ pub(crate) fn table_catalog_recovery_summary(
     (metadata_status.unwrap_or(TableCatalogRecoveryStatus::Healthy), actions)
 }
 
-fn commit_logs_share_recovery_payload(left: &CommitLogEntry, right: &CommitLogEntry) -> bool {
+pub(crate) fn commit_logs_share_recovery_payload(left: &CommitLogEntry, right: &CommitLogEntry) -> bool {
     left.version == right.version
         && left.commit_id == right.commit_id
         && left.idempotency_key == right.idempotency_key
@@ -109,6 +154,7 @@ pub(crate) fn table_commit_recovery_entry(
     table: &TableEntry,
     commit_log: &CommitLogEntry,
     idempotency_commit: Option<&CommitLogEntry>,
+    historically_committed: bool,
 ) -> TableCommitRecoveryEntry {
     let idempotency_index_status = commit_idempotency_index_status(commit_log, idempotency_commit);
     let idempotency_index_present = matches!(
@@ -127,6 +173,11 @@ pub(crate) fn table_commit_recovery_entry(
             TableCommitRecoveryState::ManualReview,
             "idempotency index points at a different commit payload".to_string(),
         )
+    } else if matches!(commit_log.status, CommitLogStatus::Failed) {
+        (
+            TableCommitRecoveryState::ManualReview,
+            "failed commit log cannot be finalized automatically".to_string(),
+        )
     } else if table_matches_committed_log(table, commit_log) {
         if matches!(commit_log.status, CommitLogStatus::Committed) {
             if idempotency_index_repair_required {
@@ -146,6 +197,11 @@ pub(crate) fn table_commit_recovery_entry(
                 "current table pointer already advanced but commit log is not finalized".to_string(),
             )
         }
+    } else if matches!(commit_log.status, CommitLogStatus::Staged) && historically_committed {
+        (
+            TableCommitRecoveryState::FinalizationRequired,
+            "a later committed pointer proves this staged commit is part of table history".to_string(),
+        )
     } else if matches!(commit_log.status, CommitLogStatus::Committed) {
         if idempotency_index_repair_required {
             (

rustfs/src/table_catalog/iceberg/validation.rs

@@ -175,6 +175,46 @@ pub(crate) fn table_metadata_warehouse_location(
     metadata_warehouse_location(table_bucket, metadata_location, metadata_object, validate_table_warehouse_location)
 }
 
+pub(crate) fn canonical_json_sha256(metadata: &serde_json::Value) -> TableCatalogStoreResult<String> {
+    let canonical = serde_json::to_vec(metadata)
+        .map_err(|err| TableCatalogStoreError::Internal(format!("failed to encode metadata digest input: {err}")))?;
+    Ok(hex_simd::encode_to_string(Sha256::digest(canonical), hex_simd::AsciiCase::Lower))
+}
+
+pub(crate) fn validate_commit_metadata_digest(
+    request: &TableCommitRequest,
+    metadata_object: &TableCatalogObject,
+) -> TableCatalogStoreResult<()> {
+    let mut expected_digest = None;
+    for requirement in &request.requirements {
+        if requirement.get("type").and_then(serde_json::Value::as_str) != Some(TABLE_METADATA_DIGEST_REQUIREMENT_TYPE) {
+            continue;
+        }
+        if expected_digest.is_some() {
+            return Err(TableCatalogStoreError::Invalid(
+                "commit contains duplicate metadata digest requirements".to_string(),
+            ));
+        }
+        expected_digest = Some(
+            requirement
+                .get("sha256")
+                .and_then(serde_json::Value::as_str)
+                .filter(|digest| rustfs_utils::crypto::is_sha256_checksum(digest))
+                .ok_or_else(|| TableCatalogStoreError::Invalid("commit metadata digest is invalid".to_string()))?,
+        );
+    }
+    let Some(expected_digest) = expected_digest else {
+        return Ok(());
+    };
+    let metadata = decode_table_metadata_json(&request.new_metadata_location, &metadata_object.data)?;
+    if canonical_json_sha256(&metadata)? != expected_digest {
+        return Err(TableCatalogStoreError::Conflict(
+            "new metadata object changed after commit validation".to_string(),
+        ));
+    }
+    Ok(())
+}
+
 pub(crate) fn view_metadata_warehouse_location(
     table_bucket: &str,
     metadata_location: &str,
@@ -197,6 +237,10 @@ pub(crate) fn warehouse_index_candidate_prefixes(object: &str) -> Vec<&str> {
     prefixes
 }
 
+pub(crate) fn warehouse_object_prefixes_overlap(left: &str, right: &str) -> bool {
+    left.starts_with(right) || right.starts_with(left)
+}
+
 pub(crate) fn table_data_plane_resource_from_entry(table: TableEntry, warehouse_object_prefix: String) -> TableDataPlaneResource {
     TableDataPlaneResource {
         table_bucket: table.table_bucket,
@@ -1352,7 +1396,9 @@ where
         validate_snapshot_graph_data_file_reference(reference, format_version, manifest_sequence_number)?;
         let object_key = snapshot_graph_object_key(context, &reference.location, reference.object_kind.clone())?;
         match reference.entry_status {
-            Some(0 | 1) if budget.validated_live_objects.insert(object_key.clone()) => live_object_keys.push(object_key),
+            Some(0 | 1) if budget.validated_live_objects.insert(object_key.clone()) => {
+                live_object_keys.push(object_key);
+            }
             Some(0 | 1) => {}
             Some(2) => {}
             Some(_) => {

rustfs/src/table_catalog/identifier.rs

@@ -177,6 +177,14 @@ pub(crate) fn default_table_root_prefix(namespace: &Namespace) -> String {
     format!("{}{}/{}/", default_namespace_root_prefix(), namespace.storage_id(), TABLE_ROOT)
 }
 
+pub(crate) fn default_table_publication_lock_path(namespace: &Namespace, table: &IdentifierSegment) -> String {
+    format!("{}{}/publication.lock", default_table_root_prefix(namespace), table.as_str())
+}
+
+pub(crate) fn default_table_bucket_publication_lock_path() -> String {
+    rustfs_common::table_catalog::TABLE_BUCKET_PUBLICATION_LOCK_PATH.to_string()
+}
+
 pub(crate) fn default_table_marker_path(namespace: &Namespace, table: &IdentifierSegment) -> String {
     format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), TABLE_MARKER_FILE)
 }

rustfs/src/table_catalog/mod.rs

@@ -72,9 +72,10 @@ pub(crate) use error::{TableCatalogStoreError, TableCatalogStoreResult};
 pub(crate) use iceberg::*;
 pub use identifier::{IdentifierSegment, Namespace, is_reserved_table_object_key};
 pub(crate) use identifier::{
-    default_table_data_dir_path, default_table_delete_dir_path, default_table_metadata_dir_path,
-    default_table_metadata_file_path, default_view_metadata_file_path, is_valid_table_metadata_location,
-    is_valid_view_metadata_location, metadata_location_from_metadata_file_path, validate_bucket_object_mutation,
+    default_table_bucket_publication_lock_path, default_table_data_dir_path, default_table_delete_dir_path,
+    default_table_metadata_dir_path, default_table_metadata_file_path, default_table_publication_lock_path,
+    default_view_metadata_file_path, is_valid_table_metadata_location, is_valid_view_metadata_location,
+    metadata_location_from_metadata_file_path, validate_bucket_object_mutation,
 };
 pub(crate) use maintenance::*;
 pub(crate) use model::*;
@@ -89,12 +90,16 @@ pub(crate) const TABLE_NAMESPACE_MARKER_VERSION: u16 = 1;
 pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1;
 pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1;
 pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1;
+pub(crate) const TABLE_WAREHOUSE_INDEX_STATE_VERSION: u16 = 2;
 pub(crate) const TABLE_MAINTENANCE_CONFIG_VERSION: u16 = 1;
 pub(crate) const TABLE_EXTERNAL_CATALOG_BRIDGE_VERSION: u16 = 1;
 pub(crate) const TABLE_CATALOG_BACKING_MANIFEST_VERSION: u16 = 1;
 pub(crate) const ENV_TABLE_CATALOG_BACKING: &str = "RUSTFS_TABLE_CATALOG_BACKING";
+pub(crate) const ENV_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED: &str =
+    "RUSTFS_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED";
 pub(crate) const TABLE_CATALOG_BACKING_OBJECT: &str = "object";
 pub(crate) const TABLE_CATALOG_BACKING_DURABLE_STRONG: &str = "durable-strong";
+pub(crate) const TABLE_METADATA_DIGEST_REQUIREMENT_TYPE: &str = "assert-rustfs-metadata-sha256";
 pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128;
 pub(crate) const TABLE_METADATA_JSON_MAX_SIZE: usize = 50 * 1024 * 1024;
 pub(crate) const TABLE_MANIFEST_AVRO_MAX_SIZE: usize = 128 * 1024 * 1024;
@@ -104,7 +109,7 @@ const TABLE_MANIFEST_AVRO_MAX_HEADER_ENTRIES: usize = 1_024;
 const TABLE_COMMIT_MAX_MANIFESTS: usize = 10_000;
 const TABLE_COMMIT_MAX_AVRO_BYTES: usize = 512 * 1024 * 1024;
 const TABLE_COMMIT_MAX_FILE_REFERENCES: usize = 1_000_000;
-const TABLE_COMMIT_OBJECT_VALIDATION_CONCURRENCY: usize = 16;
+pub(crate) const TABLE_COMMIT_OBJECT_VALIDATION_CONCURRENCY: usize = 16;
 pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;
 const WAREHOUSE_ROOT: &str = "warehouses";
 const NAMESPACE_ROOT: &str = "namespaces";

rustfs/src/table_catalog/store/migration.rs

@@ -390,6 +390,23 @@ where
                 "migration snapshot contains duplicate idempotency lookup keys".to_string(),
             ));
         }
+        let mut commit_logs_by_table = BTreeMap::<&str, Vec<&CommitLogEntry>>::new();
+        for record in &snapshot.commits {
+            commit_logs_by_table
+                .entry(record.table_id.as_str())
+                .or_default()
+                .push(&record.commit);
+        }
+        let history_by_table = tables_by_id
+            .iter()
+            .map(|(table_id, table)| {
+                let commits = commit_logs_by_table
+                    .get(table_id)
+                    .into_iter()
+                    .flat_map(|commits| commits.iter().copied());
+                (*table_id, TableCommitHistoryIndex::new(table, commits))
+            })
+            .collect::<BTreeMap<_, _>>();
         for record in &snapshot.commits {
             let table = tables_by_id.get(record.table_id.as_str()).ok_or_else(|| {
                 TableCatalogStoreError::Invalid(format!("commit {} has no table in migration snapshot", record.commit.commit_id))
@@ -408,7 +425,14 @@ where
                 .idempotency_key
                 .as_deref()
                 .and_then(|idempotency_key| idempotency_by_key.get(&(record.table_id.as_str(), idempotency_key)).copied());
-            let recovery = table_commit_recovery_entry(table, &record.commit, indexed);
+            let recovery = table_commit_recovery_entry(
+                table,
+                &record.commit,
+                indexed,
+                history_by_table
+                    .get(record.table_id.as_str())
+                    .is_some_and(|history| history.proves_committed(&record.commit)),
+            );
             if recovery.recovery_state != TableCommitRecoveryState::Committed {
                 return Err(TableCatalogStoreError::Conflict(format!(
                     "commit {} requires catalog recovery before durable strong migration",

rustfs/src/table_catalog/store/mod.rs

@@ -166,6 +166,12 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
     async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>;
 
+    async fn register_table_with_publication(
+        &self,
+        entry: TableEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()>;
+
     async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
 
     async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
@@ -201,6 +207,12 @@ pub(crate) trait TableCatalogStore: Send + Sync {
     /// newly introduced or changed snapshots before invoking this persistence boundary.
     async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult>;
 
+    async fn commit_table_with_publication(
+        &self,
+        request: TableCommitRequest,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<TableCommitResult>;
+
     async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()>;
 
     async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()>;
@@ -243,6 +255,131 @@ pub(crate) trait TableCatalogStore: Send + Sync {
     ) -> TableCatalogStoreResult<Option<CommitLogEntry>>;
 }
 
+#[async_trait::async_trait]
+pub(crate) trait TableCommitPublication: Send + Sync {
+    async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()>;
+
+    async fn prepare(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()>;
+
+    fn holds_table_bucket(&self, table_bucket: &str) -> bool;
+
+    fn holds_table(&self, table_bucket: &str, namespace: &str, table: &str) -> bool;
+
+    fn complete(&self);
+}
+
+pub(crate) struct TableCommitPublicationCompletion<'a> {
+    publication: &'a (dyn TableCommitPublication + Sync),
+}
+
+impl<'a> TableCommitPublicationCompletion<'a> {
+    pub(crate) fn new(publication: &'a (dyn TableCommitPublication + Sync)) -> Self {
+        Self { publication }
+    }
+}
+
+impl Drop for TableCommitPublicationCompletion<'_> {
+    fn drop(&mut self) {
+        self.publication.complete();
+    }
+}
+
+struct TableCommitLockPublication<'a, B> {
+    backend: &'a B,
+    state: parking_lot::Mutex<TableCommitLockPublicationState>,
+}
+
+#[derive(Default)]
+struct TableCommitLockPublicationState {
+    table_bucket: Option<String>,
+    table: Option<(String, String, String)>,
+    guards: Vec<Box<dyn Send>>,
+}
+
+impl<'a, B> TableCommitLockPublication<'a, B> {
+    fn new(backend: &'a B) -> Self {
+        Self {
+            backend,
+            state: parking_lot::Mutex::new(TableCommitLockPublicationState::default()),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<'a, B> TableCommitPublication for TableCommitLockPublication<'a, B>
+where
+    B: TableCatalogObjectBackend,
+{
+    async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
+        {
+            let mut state = self.state.lock();
+            if state.table_bucket.as_deref() == Some(table_bucket) {
+                return Ok(());
+            }
+            if state.table_bucket.is_some() || state.table.is_some() {
+                return Err(TableCatalogStoreError::Internal(
+                    "table-bucket publication lock is already held for another table bucket".to_string(),
+                ));
+            }
+            state.table_bucket = Some(table_bucket.to_string());
+        }
+        let publication_lock = default_table_bucket_publication_lock_path();
+        let guard = match self.backend.acquire_write_lock(table_bucket, &publication_lock).await {
+            Ok(guard) => guard,
+            Err(err) => {
+                self.state.lock().table_bucket = None;
+                return Err(err);
+            }
+        };
+        self.state.lock().guards.push(guard);
+        Ok(())
+    }
+
+    async fn prepare(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
+        let namespace = parse_namespace_for_store(namespace)?;
+        let table = parse_table_for_store(table)?;
+        let table_key = (table_bucket.to_string(), namespace.public_name(), table.as_str().to_string());
+        {
+            let mut state = self.state.lock();
+            if state.table.as_ref() == Some(&table_key) {
+                return Ok(());
+            }
+            if state.table.is_some() {
+                return Err(TableCatalogStoreError::Internal(
+                    "table publication lock is already held for another table".to_string(),
+                ));
+            }
+            state.table = Some(table_key);
+        }
+        let publication_lock = default_table_publication_lock_path(&namespace, &table);
+        let guard = match self.backend.acquire_write_lock(table_bucket, &publication_lock).await {
+            Ok(guard) => guard,
+            Err(err) => {
+                self.state.lock().table = None;
+                return Err(err);
+            }
+        };
+        self.state.lock().guards.push(guard);
+        Ok(())
+    }
+
+    fn holds_table_bucket(&self, table_bucket: &str) -> bool {
+        self.state.lock().table_bucket.as_deref() == Some(table_bucket)
+    }
+
+    fn holds_table(&self, table_bucket: &str, namespace: &str, table: &str) -> bool {
+        self.state
+            .lock()
+            .table
+            .as_ref()
+            .is_some_and(|held| held.0 == table_bucket && held.1 == namespace && held.2 == table)
+    }
+
+    fn complete(&self) {
+        *self.state.lock() = TableCommitLockPublicationState::default();
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct TableCatalogObject {
     pub data: Vec<u8>,
@@ -317,8 +454,26 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
             }))
     }
 
+    async fn object_metadata_unlocked(
+        &self,
+        bucket: &str,
+        object: &str,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        Ok(self
+            .read_object_unlocked(bucket, object)
+            .await?
+            .map(|object| TableCatalogObjectMetadata {
+                etag: object.etag,
+                mod_time: object.mod_time,
+            }))
+    }
+
     async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool>;
 
+    async fn object_exists_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
+        self.object_exists(bucket, object).await
+    }
+
     async fn put_object(
         &self,
         bucket: &str,
@@ -370,6 +525,55 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
     }
 
     async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>>;
+
+    async fn begin_table_bucket_commit_publication(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
+    fn table_bucket_commit_publication_is_held(&self, _table_bucket: &str) -> bool {
+        false
+    }
+
+    async fn prepare_table_commit_publication(
+        &self,
+        _table_bucket: &str,
+        _namespace: &str,
+        _table: &str,
+    ) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
+    fn table_commit_publication_is_held(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
+        false
+    }
+
+    fn complete_table_commit_publication(&self) {}
+}
+
+#[async_trait::async_trait]
+impl<B> TableCommitPublication for B
+where
+    B: TableCatalogObjectBackend,
+{
+    async fn begin_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
+        self.begin_table_bucket_commit_publication(table_bucket).await
+    }
+
+    async fn prepare(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
+        self.prepare_table_commit_publication(table_bucket, namespace, table).await
+    }
+
+    fn holds_table_bucket(&self, table_bucket: &str) -> bool {
+        self.table_bucket_commit_publication_is_held(table_bucket)
+    }
+
+    fn holds_table(&self, table_bucket: &str, namespace: &str, table: &str) -> bool {
+        self.table_commit_publication_is_held(table_bucket, namespace, table)
+    }
+
+    fn complete(&self) {
+        self.complete_table_commit_publication();
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -577,6 +781,10 @@ impl TableCatalogObjectPaths {
         )
     }
 
+    pub fn warehouse_index_entries_prefix(&self, table_bucket: &str) -> String {
+        format!("{}{}/", self.table_bucket_root_prefix(table_bucket), WAREHOUSE_INDEX_ROOT)
+    }
+
     pub fn warehouse_index_entry_path(&self, table_bucket: &str, warehouse_object_prefix: &str) -> String {
         format!(
             "{}{}/{}.json",
@@ -779,6 +987,17 @@ where
         }
     }
 
+    async fn register_table_with_publication(
+        &self,
+        entry: TableEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
+        match self {
+            Self::ObjectBacked(store) => store.register_table_with_publication(entry, publication).await,
+            Self::DurableStrong(store) => store.register_table_with_publication(entry, publication).await,
+        }
+    }
+
     async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<TableEntry>> {
         match self {
             Self::ObjectBacked(store) => store.list_tables(table_bucket, namespace).await,
@@ -831,6 +1050,17 @@ where
         }
     }
 
+    async fn commit_table_with_publication(
+        &self,
+        request: TableCommitRequest,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<TableCommitResult> {
+        match self {
+            Self::ObjectBacked(store) => store.commit_table_with_publication(request, publication).await,
+            Self::DurableStrong(store) => store.commit_table_with_publication(request, publication).await,
+        }
+    }
+
     async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
         match self {
             Self::ObjectBacked(store) => store.drop_table(table_bucket, namespace, table).await,
@@ -1205,6 +1435,32 @@ where
         }
     }
 
+    async fn object_metadata_unlocked(
+        &self,
+        bucket: &str,
+        object: &str,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
+        match self
+            .store
+            .get_object_info(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+        {
+            Ok(info) => Ok(Some(TableCatalogObjectMetadata {
+                etag: info.etag,
+                mod_time: info.mod_time,
+            })),
+            Err(err) if is_missing_storage_error(&err) => Ok(None),
+            Err(err) => Err(storage_error_to_catalog("stat catalog object", err)),
+        }
+    }
+
     async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
         match self.store.get_object_info(bucket, object, &ObjectOptions::default()).await {
             Ok(_) => Ok(true),
@@ -1213,6 +1469,25 @@ where
         }
     }
 
+    async fn object_exists_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
+        match self
+            .store
+            .get_object_info(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+        {
+            Ok(_) => Ok(true),
+            Err(err) if is_missing_storage_error(&err) => Ok(false),
+            Err(err) => Err(storage_error_to_catalog("check catalog object", err)),
+        }
+    }
+
     async fn put_object(
         &self,
         bucket: &str,

rustfs/src/table_catalog/store/object.rs

@@ -519,7 +519,7 @@ where
 
     async fn write_warehouse_index_state_unlocked(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
         let state = TableWarehouseIndexStateEntry {
-            version: TABLE_CATALOG_ENTRY_VERSION,
+            version: TABLE_WAREHOUSE_INDEX_STATE_VERSION,
             table_bucket: table_bucket.to_string(),
             state: TableCatalogEntryState::Active,
         };
@@ -542,7 +542,7 @@ where
         else {
             return Ok(false);
         };
-        Ok(state.version == TABLE_CATALOG_ENTRY_VERSION
+        Ok(state.version == TABLE_WAREHOUSE_INDEX_STATE_VERSION
             && state.table_bucket == table_bucket
             && state.state == TableCatalogEntryState::Active)
     }
@@ -635,11 +635,51 @@ where
         Ok(true)
     }
 
+    async fn ensure_table_warehouse_prefix_available(&self, entry: &TableEntry) -> TableCatalogStoreResult<()> {
+        let candidate = table_warehouse_index_entry(entry)?;
+        let state_object = self.paths.warehouse_index_state_path(&candidate.table_bucket);
+        for object in self
+            .backend
+            .list_objects(self.catalog_bucket(), &self.paths.warehouse_index_entries_prefix(&candidate.table_bucket))
+            .await?
+        {
+            if object == state_object {
+                continue;
+            }
+            let Some((existing, _)) = self
+                .read_entry::<TableWarehouseIndexEntry>(self.catalog_bucket(), &object)
+                .await?
+            else {
+                continue;
+            };
+            if existing.table_id == candidate.table_id || existing.state != TableCatalogEntryState::Active {
+                continue;
+            }
+            if warehouse_object_prefixes_overlap(&existing.warehouse_object_prefix, &candidate.warehouse_object_prefix)
+                && self.warehouse_index_entry_has_active_owner(&existing).await?
+            {
+                return Err(TableCatalogStoreError::Conflict(format!(
+                    "table warehouse location overlaps an active table: {}",
+                    candidate.warehouse_object_prefix
+                )));
+            }
+        }
+        Ok(())
+    }
+
     async fn reserve_table_warehouse_index(&self, entry: &TableEntry) -> TableCatalogStoreResult<WarehouseIndexReservation> {
         let index = table_warehouse_index_entry(entry)?;
         let object = self
             .paths
             .warehouse_index_entry_path(&index.table_bucket, &index.warehouse_object_prefix);
+        if let Some((existing, _)) = self
+            .read_entry::<TableWarehouseIndexEntry>(self.catalog_bucket(), &object)
+            .await?
+            && existing == index
+        {
+            return Ok(WarehouseIndexReservation::AlreadyReserved);
+        }
+        self.ensure_table_warehouse_prefix_available(entry).await?;
         loop {
             match self
                 .write_entry(self.catalog_bucket(), &object, &index, TableCatalogPutPrecondition::IfAbsent)
@@ -745,7 +785,7 @@ where
         else {
             return Ok(false);
         };
-        Ok(state.version == TABLE_CATALOG_ENTRY_VERSION
+        Ok(state.version == TABLE_WAREHOUSE_INDEX_STATE_VERSION
             && state.table_bucket == table_bucket
             && state.state == TableCatalogEntryState::Active)
     }
@@ -868,7 +908,27 @@ where
         if self.read_warehouse_index_state_unlocked(table_bucket).await? {
             return Ok(());
         }
-        for table in self.list_all_tables(table_bucket).await? {
+        let tables = self.list_all_tables(table_bucket).await?;
+        let mut active_prefixes = tables
+            .iter()
+            .filter(|table| table.state == TableCatalogEntryState::Active)
+            .filter_map(|table| {
+                table_warehouse_object_prefix(table)
+                    .ok()
+                    .map(|prefix| (prefix, table.table_id.as_str()))
+            })
+            .collect::<Vec<_>>();
+        active_prefixes.sort_unstable_by(|left, right| left.0.cmp(&right.0));
+        if let Some(window) = active_prefixes
+            .windows(2)
+            .find(|window| window[0].1 != window[1].1 && warehouse_object_prefixes_overlap(&window[0].0, &window[1].0))
+        {
+            return Err(TableCatalogStoreError::Conflict(format!(
+                "active table warehouse locations overlap: {} and {}",
+                window[0].0, window[1].0
+            )));
+        }
+        for table in tables {
             if table.state != TableCatalogEntryState::Active {
                 continue;
             }
@@ -958,12 +1018,30 @@ where
         &self,
         entry: TableEntry,
         precondition: TableCatalogPutPrecondition,
+    ) -> TableCatalogStoreResult<()> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        self.write_table_entry_with_publication(entry, precondition, &publication)
+            .await
+    }
+
+    async fn write_table_entry_with_publication(
+        &self,
+        entry: TableEntry,
+        precondition: TableCatalogPutPrecondition,
+        publication: &(dyn TableCommitPublication + Sync),
     ) -> TableCatalogStoreResult<()> {
         validate_catalog_entry_version("table", entry.version)?;
-        self.require_table_bucket(&entry.table_bucket).await?;
         let namespace = parse_namespace_for_store(&entry.namespace)?;
         let table = parse_table_for_store(&entry.table)?;
         validate_table_warehouse_location(&entry.table_bucket, &entry.warehouse_location)?;
+        publication.begin_table_bucket(&entry.table_bucket).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "table registration requires a table-bucket publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        self.require_table_bucket(&entry.table_bucket).await?;
         let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?;
         let namespace_path = self.paths.namespace_entry_path(&entry.table_bucket, &namespace);
         let _namespace_guard = self
@@ -974,6 +1052,15 @@ where
             .await?;
         let table_path = self.paths.table_entry_path(&entry.table_bucket, &namespace, &table);
         let _table_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+        // Preserve catalog -> publication -> object lock order across rolling upgrades.
+        publication
+            .prepare(&entry.table_bucket, &entry.namespace, &entry.table)
+            .await?;
+        if !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table) {
+            return Err(TableCatalogStoreError::Internal(
+                "table registration requires a table publication fence".to_string(),
+            ));
+        }
         let reservation = self.reserve_table_warehouse_index(&entry).await?;
         let result = self
             .write_entry_unlocked(self.catalog_bucket(), &table_path, &entry, precondition)
@@ -1056,6 +1143,20 @@ where
             .map(|entry| entry.map(|(commit, _)| commit))
     }
 
+    async fn read_table_commit_logs(&self, entry: &TableEntry) -> TableCatalogStoreResult<Vec<(String, CommitLogEntry)>> {
+        let commit_prefix = self.paths.commit_log_entries_prefix(&entry.table_bucket, &entry.table_id);
+        let mut commits = Vec::new();
+        for object in self.backend.list_objects(self.catalog_bucket(), &commit_prefix).await? {
+            if !object.ends_with(".json") {
+                continue;
+            }
+            if let Some(commit_log) = self.read_commit_by_path(&object).await? {
+                commits.push((object, commit_log));
+            }
+        }
+        Ok(commits)
+    }
+
     async fn finalize_commit_log(
         &self,
         commit_path: &str,
@@ -1076,15 +1177,10 @@ where
         entry: &TableEntry,
         finalized_count: usize,
     ) -> TableCatalogStoreResult<TableCommitRecoveryReport> {
-        let commit_prefix = self.paths.commit_log_entries_prefix(&entry.table_bucket, &entry.table_id);
-        let mut commits = Vec::new();
-        for object in self.backend.list_objects(self.catalog_bucket(), &commit_prefix).await? {
-            if !object.ends_with(".json") {
-                continue;
-            }
-            let Some(commit_log) = self.read_commit_by_path(&object).await? else {
-                continue;
-            };
+        let commit_logs_with_paths = self.read_table_commit_logs(entry).await?;
+        let history = TableCommitHistoryIndex::new(entry, commit_logs_with_paths.iter().map(|(_, commit_log)| commit_log));
+        let mut commits = Vec::with_capacity(commit_logs_with_paths.len());
+        for (_, commit_log) in &commit_logs_with_paths {
             let idempotency_commit = match commit_log.idempotency_key.as_deref() {
                 Some(idempotency_key) => {
                     let idempotency_path =
@@ -1094,7 +1190,12 @@ where
                 }
                 None => None,
             };
-            commits.push(table_commit_recovery_entry(entry, &commit_log, idempotency_commit.as_ref()));
+            commits.push(table_commit_recovery_entry(
+                entry,
+                commit_log,
+                idempotency_commit.as_ref(),
+                history.proves_committed(commit_log),
+            ));
         }
         commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id));
 
@@ -1172,15 +1273,10 @@ where
             )));
         };
 
-        let commit_prefix = self.paths.commit_log_entries_prefix(table_bucket, &entry.table_id);
+        let commit_logs_with_paths = self.read_table_commit_logs(&entry).await?;
+        let history = TableCommitHistoryIndex::new(&entry, commit_logs_with_paths.iter().map(|(_, commit_log)| commit_log));
         let mut finalized_count = 0;
-        for commit_path in self.backend.list_objects(self.catalog_bucket(), &commit_prefix).await? {
-            if !commit_path.ends_with(".json") {
-                continue;
-            }
-            let Some(commit_log) = self.read_commit_by_path(&commit_path).await? else {
-                continue;
-            };
+        for (commit_path, commit_log) in &commit_logs_with_paths {
             let idempotency_path = commit_log.idempotency_key.as_deref().map(|idempotency_key| {
                 self.paths
                     .commit_idempotency_entry_path(table_bucket, &entry.table_id, idempotency_key)
@@ -1189,14 +1285,19 @@ where
                 Some(idempotency_path) => self.read_commit_by_path(idempotency_path).await?,
                 None => None,
             };
-            let recovery_entry = table_commit_recovery_entry(&entry, &commit_log, idempotency_commit.as_ref());
+            let recovery_entry = table_commit_recovery_entry(
+                &entry,
+                commit_log,
+                idempotency_commit.as_ref(),
+                history.proves_committed(commit_log),
+            );
             if matches!(
                 recovery_entry.recovery_state,
                 TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired
             ) {
-                let mut committed = commit_log;
+                let mut committed = commit_log.clone();
                 committed.status = CommitLogStatus::Committed;
-                self.finalize_commit_log(&commit_path, idempotency_path.as_deref(), &committed)
+                self.finalize_commit_log(commit_path, idempotency_path.as_deref(), &committed)
                     .await?;
                 finalized_count += 1;
             }
@@ -2427,6 +2528,21 @@ where
         table: &str,
         config: TableSnapshotExpirationConfig,
     ) -> TableCatalogStoreResult<TableSnapshotExpirationReport> {
+        self.plan_table_snapshot_expiration_with_backend(&self.backend, table_bucket, namespace, table, config)
+            .await
+    }
+
+    pub(crate) async fn plan_table_snapshot_expiration_with_backend<P>(
+        &self,
+        metadata_backend: &P,
+        table_bucket: &str,
+        namespace: &str,
+        table: &str,
+        config: TableSnapshotExpirationConfig,
+    ) -> TableCatalogStoreResult<TableSnapshotExpirationReport>
+    where
+        P: TableCatalogObjectBackend,
+    {
         validate_table_snapshot_expiration_config(&config)?;
         let namespace = parse_namespace_for_store(namespace)?;
         let table = parse_table_for_store(table)?;
@@ -2445,7 +2561,7 @@ where
             ));
         }
 
-        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        let Some(current_metadata) = read_table_metadata_value(metadata_backend, table_bucket, &entry.metadata_location).await?
         else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
@@ -2507,6 +2623,23 @@ where
         table: &str,
         config: TableCompactionPlanningConfig,
     ) -> TableCatalogStoreResult<TableCompactionPlanningReport> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        self.commit_table_compaction_with_publication(&self.backend, &publication, table_bucket, namespace, table, config)
+            .await
+    }
+
+    pub(crate) async fn commit_table_compaction_with_publication<P>(
+        &self,
+        object_backend: &P,
+        publication: &(dyn TableCommitPublication + Sync),
+        table_bucket: &str,
+        namespace: &str,
+        table: &str,
+        config: TableCompactionPlanningConfig,
+    ) -> TableCatalogStoreResult<TableCompactionPlanningReport>
+    where
+        P: TableCatalogObjectBackend,
+    {
         validate_table_compaction_planning_config(&config)?;
         let namespace = parse_namespace_for_store(namespace)?;
         let table = parse_table_for_store(table)?;
@@ -2525,7 +2658,7 @@ where
             ));
         }
 
-        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        let Some(current_metadata) = read_table_metadata_value(object_backend, table_bucket, &entry.metadata_location).await?
         else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
@@ -2533,13 +2666,13 @@ where
             )));
         };
         let mut report =
-            table_compaction_planning_report(&self.backend, table_bucket, &namespace, &table, &entry, &current_metadata, config)
+            table_compaction_planning_report(object_backend, table_bucket, &namespace, &table, &entry, &current_metadata, config)
                 .await?;
         if report.status != TableCompactionPlanningStatus::RewriteCandidates {
             return Err(TableCatalogStoreError::Invalid("compaction has no safe rewrite candidates".to_string()));
         }
         let current_data_files =
-            compaction_current_data_files(&self.backend, table_bucket, &namespace, &table, &entry, &current_metadata).await?;
+            compaction_current_data_files(object_backend, table_bucket, &namespace, &table, &entry, &current_metadata).await?;
         let current_data_files_by_key = current_data_files
             .iter()
             .map(|file| (file.object_key.as_str(), file))
@@ -2574,14 +2707,14 @@ where
             let sort_order_id = compaction_rewrite_group_sort_order(&current_data_files_by_key, rewrite_group)?;
             let mut input_files = Vec::with_capacity(rewrite_group.input_file_locations.len());
             for input_file in &rewrite_group.input_file_locations {
-                let Some(input_object) = self.backend.read_object(table_bucket, input_file).await? else {
+                let Some(input_object) = object_backend.read_object(table_bucket, input_file).await? else {
                     return Err(TableCatalogStoreError::NotFound(format!("compaction input data file {input_file}")));
                 };
                 input_files.push((input_file.clone(), input_object.data));
             }
             let compacted_file = compact_parquet_data_files(&input_files)?;
             let output_bytes = u64::try_from(compacted_file.data.len()).unwrap_or(u64::MAX);
-            self.backend
+            object_backend
                 .put_object(table_bucket, &output_file, compacted_file.data, TableCatalogPutPrecondition::IfAbsent)
                 .await?;
             rewrite_group.output_file_location = Some(output_file_path.clone());
@@ -2608,7 +2741,7 @@ where
             default_table_metadata_file_path(&namespace, &table, &format!("compaction-{compaction_id}.metadata.json"));
         let manifest_data = compacted_manifest_avro_bytes(&manifest_data_files)?;
         let manifest_length = u64::try_from(manifest_data.len()).unwrap_or(u64::MAX);
-        self.backend
+        object_backend
             .put_object(table_bucket, &new_manifest, manifest_data, TableCatalogPutPrecondition::IfAbsent)
             .await?;
         let added_files_count = compacted_files.len();
@@ -2631,7 +2764,7 @@ where
             added_rows_count,
             existing_rows_count,
         })?;
-        self.backend
+        object_backend
             .put_object(
                 table_bucket,
                 &new_manifest_list,
@@ -2648,24 +2781,27 @@ where
             &entry.metadata_location,
             now,
         )?;
-        self.backend
+        object_backend
             .put_object(table_bucket, &new_metadata, new_metadata_data, TableCatalogPutPrecondition::IfAbsent)
             .await?;
 
         let commit_result = self
-            .commit_table(TableCommitRequest {
-                table_bucket: table_bucket.to_string(),
-                namespace: namespace.public_name(),
-                table: table.as_str().to_string(),
-                commit_id: format!("compaction-{compaction_id}"),
-                idempotency_key: Some(format!("compaction-{compaction_id}")),
-                operation: "compaction".to_string(),
-                expected_version_token: entry.version_token,
-                expected_metadata_location: entry.metadata_location,
-                new_metadata_location: new_metadata.clone(),
-                requirements: Vec::new(),
-                writer: Some("rustfs-maintenance".to_string()),
-            })
+            .commit_table_with_publication(
+                TableCommitRequest {
+                    table_bucket: table_bucket.to_string(),
+                    namespace: namespace.public_name(),
+                    table: table.as_str().to_string(),
+                    commit_id: format!("compaction-{compaction_id}"),
+                    idempotency_key: Some(format!("compaction-{compaction_id}")),
+                    operation: "compaction".to_string(),
+                    expected_version_token: entry.version_token,
+                    expected_metadata_location: entry.metadata_location,
+                    new_metadata_location: new_metadata.clone(),
+                    requirements: Vec::new(),
+                    writer: Some("rustfs-maintenance".to_string()),
+                },
+                publication,
+            )
             .await?;
 
         report.status = TableCompactionPlanningStatus::Committed;
@@ -3245,6 +3381,8 @@ where
 
         let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table);
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+        let publication_lock = default_table_publication_lock_path(&namespace, &table);
+        let _publication_guard = self.backend.acquire_write_lock(table_bucket, &publication_lock).await?;
         let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "table {}/{}/{}",
@@ -3635,6 +3773,15 @@ where
         self.write_table_entry(entry, TableCatalogPutPrecondition::IfAbsent).await
     }
 
+    async fn register_table_with_publication(
+        &self,
+        entry: TableEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
+        self.write_table_entry_with_publication(entry, TableCatalogPutPrecondition::IfAbsent, publication)
+            .await
+    }
+
     async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<TableEntry>> {
         let namespace = parse_namespace_for_store(namespace)?;
         let mut entries = Vec::new();
@@ -3722,6 +3869,7 @@ where
 
         match self.backfill_table_warehouse_index(table_bucket).await {
             Ok(()) => self.resolve_table_data_plane_resource_from_index(table_bucket, object).await,
+            Err(err @ TableCatalogStoreError::Conflict(_)) => Err(err),
             Err(err) => {
                 tracing::warn!(
                     table_bucket = %table_bucket,
@@ -3734,6 +3882,16 @@ where
     }
 
     async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        publication.begin_table_bucket(&request.table_bucket).await?;
+        self.commit_table_with_publication(request, &publication).await
+    }
+
+    async fn commit_table_with_publication(
+        &self,
+        request: TableCommitRequest,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<TableCommitResult> {
         let commit_started = Instant::now();
         record_table_commit_attempt(&request.operation);
         let namespace = parse_namespace_for_store(&request.namespace)?;
@@ -3741,6 +3899,16 @@ where
         let _migration_guard = self.acquire_object_backed_catalog_write_permit(&request.table_bucket).await?;
         let table_path = self.paths.table_entry_path(&request.table_bucket, &namespace, &table);
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+        // Preserve catalog -> publication -> object lock order across rolling upgrades.
+        publication
+            .prepare(&request.table_bucket, &request.namespace, &request.table)
+            .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table) {
+            return Err(TableCatalogStoreError::Internal(
+                "table commit requires a table publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
 
         let Some((current, current_etag)) = self
             .read_table_with_etag_unlocked(&request.table_bucket, &namespace, &table)
@@ -3773,6 +3941,48 @@ where
             None => None,
         };
 
+        if let (Some(existing), Some(indexed)) = (&existing_commit, &existing_idempotency_commit)
+            && !commit_logs_share_recovery_payload(existing, indexed)
+        {
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Conflict(
+                    "commit record and idempotency index contain different payloads".to_string(),
+                )),
+            );
+        }
+        if let Some(existing) = existing_idempotency_commit.as_ref()
+            && !commit_log_matches_request(existing, &request, &current.table_id)
+        {
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Conflict("idempotency key already exists".to_string())),
+            );
+        }
+        if existing_commit.is_none() && existing_idempotency_commit.is_some() {
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Conflict(
+                    "idempotency key exists without a recoverable commit record".to_string(),
+                )),
+            );
+        }
+
         if let Some(existing) = existing_commit.as_ref() {
             if !commit_log_matches_request(existing, &request, &current.table_id) {
                 return table_commit_result(
@@ -3788,7 +3998,48 @@ where
                     ))),
                 );
             }
-            if matches!(existing.status, CommitLogStatus::Committed) || table_matches_committed_log(&current, existing) {
+            if matches!(existing.status, CommitLogStatus::Failed) {
+                return table_commit_result(
+                    &request.table_bucket,
+                    &request.namespace,
+                    &request.table,
+                    &request.commit_id,
+                    &request.operation,
+                    commit_started,
+                    Err(TableCatalogStoreError::Conflict("failed commit record cannot be replayed".to_string())),
+                );
+            }
+            if matches!(existing.status, CommitLogStatus::Committed) && table_matches_staged_base(&current, existing) {
+                return table_commit_result(
+                    &request.table_bucket,
+                    &request.namespace,
+                    &request.table,
+                    &request.commit_id,
+                    &request.operation,
+                    commit_started,
+                    Err(TableCatalogStoreError::Conflict(
+                        "committed record still matches the pre-commit table state".to_string(),
+                    )),
+                );
+            }
+            let historically_committed = if matches!(existing.status, CommitLogStatus::Staged)
+                && !table_matches_staged_base(&current, existing)
+                && !table_matches_committed_log(&current, existing)
+            {
+                let commit_logs = self
+                    .read_table_commit_logs(&current)
+                    .await?
+                    .into_iter()
+                    .map(|(_, commit_log)| commit_log)
+                    .collect::<Vec<_>>();
+                TableCommitHistoryIndex::new(&current, commit_logs.iter()).proves_committed(existing)
+            } else {
+                false
+            };
+            if matches!(existing.status, CommitLogStatus::Committed)
+                || (matches!(existing.status, CommitLogStatus::Staged)
+                    && (table_matches_committed_log(&current, existing) || historically_committed))
+            {
                 let mut committed = existing.clone();
                 committed.status = CommitLogStatus::Committed;
                 let _ = self
@@ -3821,32 +4072,6 @@ where
                 );
             }
         }
-        if let Some(existing) = existing_idempotency_commit.as_ref()
-            && !commit_log_matches_request(existing, &request, &current.table_id)
-        {
-            return table_commit_result(
-                &request.table_bucket,
-                &request.namespace,
-                &request.table,
-                &request.commit_id,
-                &request.operation,
-                commit_started,
-                Err(TableCatalogStoreError::Conflict("idempotency key already exists".to_string())),
-            );
-        }
-        if existing_commit.is_none() && existing_idempotency_commit.is_some() {
-            return table_commit_result(
-                &request.table_bucket,
-                &request.namespace,
-                &request.table,
-                &request.commit_id,
-                &request.operation,
-                commit_started,
-                Err(TableCatalogStoreError::Conflict(
-                    "idempotency key exists without a recoverable commit record".to_string(),
-                )),
-            );
-        }
 
         if current.version_token != request.expected_version_token {
             return table_commit_result(
@@ -3905,6 +4130,7 @@ where
                 ))),
             );
         };
+        validate_commit_metadata_digest(&request, &new_metadata_object)?;
         let table_bucket = request.table_bucket.clone();
         let metadata_location = request.new_metadata_location.clone();
         let next_warehouse_location = tokio::task::spawn_blocking(move || {
@@ -3912,6 +4138,23 @@ where
         })
         .await
         .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
+        if next_warehouse_location
+            .as_ref()
+            .is_some_and(|warehouse_location| warehouse_location != &current.warehouse_location)
+            && !publication.holds_table_bucket(&request.table_bucket)
+        {
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Internal(
+                    "table warehouse relocation requires a table-bucket publication fence".to_string(),
+                )),
+            );
+        }
 
         let has_existing_commit = existing_commit.is_some();
         let mut staged_commit_log = existing_commit.unwrap_or_else(|| CommitLogEntry {
@@ -4024,6 +4267,8 @@ where
     }
 
     async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        publication.begin_table_bucket(table_bucket).await?;
         let namespace = parse_namespace_for_store(namespace)?;
         let table = parse_table_for_store(table)?;
         let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?;
@@ -4034,6 +4279,17 @@ where
             .await?;
         let object = self.paths.table_entry_path(table_bucket, &namespace, &table);
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?;
+        publication
+            .prepare(table_bucket, &namespace.public_name(), table.as_str())
+            .await?;
+        if !publication.holds_table_bucket(table_bucket)
+            || !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str())
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "table drop requires table-bucket and table publication fences".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(&publication);
         let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "table {}/{}/{}",

rustfs/src/table_catalog/store/strong.rs

@@ -443,16 +443,20 @@ where
                 continue;
             };
             let table_key = (table_bucket.clone(), namespace.clone(), table.clone());
-            if let Some(existing_key) = warehouse_index
-                .entry(table_bucket.clone())
-                .or_default()
-                .insert(warehouse_object_prefix.clone(), table_key.clone())
+            let bucket_index = warehouse_index.entry(table_bucket.clone()).or_default();
+            let predecessor = bucket_index.range(..=warehouse_object_prefix.clone()).next_back();
+            let successor = bucket_index.range(warehouse_object_prefix.clone()..).next();
+            if let Some((existing_prefix, existing_key)) = predecessor
+                .into_iter()
+                .chain(successor)
+                .find(|(existing_prefix, _)| warehouse_object_prefixes_overlap(existing_prefix, &warehouse_object_prefix))
             {
                 return Err(TableCatalogStoreError::Invalid(format!(
-                    "duplicate active table warehouse location in strong catalog snapshot: {warehouse_object_prefix} is owned by {}/{}/{} and {}/{}/{}",
+                    "overlapping active table warehouse location in strong catalog snapshot: {warehouse_object_prefix} overlaps {existing_prefix} owned by {}/{}/{} and {}/{}/{}",
                     existing_key.0, existing_key.1, existing_key.2, table_key.0, table_key.1, table_key.2
                 )));
             }
+            bucket_index.insert(warehouse_object_prefix, table_key);
         }
         state.warehouse_index = warehouse_index;
         Ok(())
@@ -696,9 +700,9 @@ where
             let Ok(existing_prefix) = table_warehouse_object_prefix(existing) else {
                 continue;
             };
-            if existing_prefix == candidate_prefix {
+            if warehouse_object_prefixes_overlap(&existing_prefix, &candidate_prefix) {
                 return Err(TableCatalogStoreError::Conflict(format!(
-                    "table warehouse location is already registered: {candidate_prefix}"
+                    "table warehouse location overlaps an active table: {candidate_prefix}"
                 )));
             }
         }
@@ -709,6 +713,14 @@ where
         state: &StrongTableCatalogState,
         entry: &TableEntry,
     ) -> TableCommitRecoveryReport {
+        let history = TableCommitHistoryIndex::new(
+            entry,
+            state
+                .commits
+                .iter()
+                .filter(|((table_bucket, table_id, _), _)| table_bucket == &entry.table_bucket && table_id == &entry.table_id)
+                .map(|(_, commit_log)| commit_log),
+        );
         let mut commits = state
             .commits
             .iter()
@@ -719,7 +731,7 @@ where
                         .idempotency
                         .get(&Self::idempotency_key(&entry.table_bucket, &entry.table_id, idempotency_key))
                 });
-                table_commit_recovery_entry(entry, commit_log, idempotency_commit)
+                table_commit_recovery_entry(entry, commit_log, idempotency_commit, history.proves_committed(commit_log))
             })
             .collect::<Vec<_>>();
         commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id));
@@ -783,18 +795,11 @@ where
             .map(|idempotency_key| Self::idempotency_key(&request.table_bucket, &current.table_id, idempotency_key));
         let existing_idempotency_commit = idempotency_key.as_ref().and_then(|key| state.idempotency.get(key));
 
-        if let Some(existing) = existing_commit {
-            if !commit_log_matches_request(existing, request, &current.table_id) {
-                return Err(TableCatalogStoreError::Conflict(format!(
-                    "commit id already exists: {}",
-                    request.commit_id
-                )));
-            }
-            if matches!(existing.status, CommitLogStatus::Committed) || table_matches_committed_log(&current, existing) {
-                return Ok(current);
-            }
+        if let (Some(existing), Some(indexed)) = (existing_commit, existing_idempotency_commit)
+            && !commit_logs_share_recovery_payload(existing, indexed)
+        {
             return Err(TableCatalogStoreError::Conflict(
-                "existing commit record does not match current table state".to_string(),
+                "commit record and idempotency index contain different payloads".to_string(),
             ));
         }
         if let Some(existing) = existing_idempotency_commit
@@ -802,11 +807,55 @@ where
         {
             return Err(TableCatalogStoreError::Conflict("idempotency key already exists".to_string()));
         }
-        if existing_idempotency_commit.is_some() {
+        if existing_commit.is_none() && existing_idempotency_commit.is_some() {
             return Err(TableCatalogStoreError::Conflict(
                 "idempotency key exists without a recoverable commit record".to_string(),
             ));
         }
+
+        if let Some(existing) = existing_commit {
+            if !commit_log_matches_request(existing, request, &current.table_id) {
+                return Err(TableCatalogStoreError::Conflict(format!(
+                    "commit id already exists: {}",
+                    request.commit_id
+                )));
+            }
+            if matches!(existing.status, CommitLogStatus::Failed) {
+                return Err(TableCatalogStoreError::Conflict("failed commit record cannot be replayed".to_string()));
+            }
+            if matches!(existing.status, CommitLogStatus::Committed) && table_matches_staged_base(&current, existing) {
+                return Err(TableCatalogStoreError::Conflict(
+                    "committed record still matches the pre-commit table state".to_string(),
+                ));
+            }
+            let historically_committed = if matches!(existing.status, CommitLogStatus::Staged)
+                && !table_matches_staged_base(&current, existing)
+                && !table_matches_committed_log(&current, existing)
+            {
+                TableCommitHistoryIndex::new(
+                    &current,
+                    state
+                        .commits
+                        .iter()
+                        .filter(|((table_bucket, table_id, _), _)| {
+                            table_bucket == &request.table_bucket && table_id == &current.table_id
+                        })
+                        .map(|(_, commit_log)| commit_log),
+                )
+                .proves_committed(existing)
+            } else {
+                false
+            };
+            if matches!(existing.status, CommitLogStatus::Committed)
+                || (matches!(existing.status, CommitLogStatus::Staged)
+                    && (table_matches_committed_log(&current, existing) || historically_committed))
+            {
+                return Ok(current);
+            }
+            return Err(TableCatalogStoreError::Conflict(
+                "existing commit record does not match current table state".to_string(),
+            ));
+        }
         if current.version_token != request.expected_version_token {
             return Err(TableCatalogStoreError::Conflict(
                 "current table version token does not match expected token".to_string(),
@@ -831,15 +880,38 @@ where
         current: TableEntry,
     ) -> Option<TableCommitResult> {
         let commit_key = Self::commit_key(&request.table_bucket, &current.table_id, &request.commit_id);
-        let existing = state.commits.get(&commit_key)?;
-        if !commit_log_matches_request(existing, request, &current.table_id) {
+        let existing = state.commits.get(&commit_key)?.clone();
+        if !commit_log_matches_request(&existing, request, &current.table_id) {
             return None;
         }
-        if !matches!(existing.status, CommitLogStatus::Committed) && !table_matches_committed_log(&current, existing) {
+        let historically_committed = if matches!(existing.status, CommitLogStatus::Staged)
+            && !table_matches_staged_base(&current, &existing)
+            && !table_matches_committed_log(&current, &existing)
+        {
+            TableCommitHistoryIndex::new(
+                &current,
+                state
+                    .commits
+                    .iter()
+                    .filter(|((table_bucket, table_id, _), _)| {
+                        table_bucket == &request.table_bucket && table_id == &current.table_id
+                    })
+                    .map(|(_, commit_log)| commit_log),
+            )
+            .proves_committed(&existing)
+        } else {
+            false
+        };
+        if matches!(existing.status, CommitLogStatus::Failed)
+            || (matches!(existing.status, CommitLogStatus::Committed) && table_matches_staged_base(&current, &existing))
+            || (!matches!(existing.status, CommitLogStatus::Committed)
+                && !table_matches_committed_log(&current, &existing)
+                && !historically_committed)
+        {
             return None;
         }
 
-        let mut committed = existing.clone();
+        let mut committed = existing;
         committed.status = CommitLogStatus::Committed;
         state.commits.insert(commit_key, committed.clone());
         if let Some(idempotency_key) = committed.idempotency_key.as_deref() {
@@ -1202,12 +1274,36 @@ where
     }
 
     async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> {
-        let _write_guard = self.write_lock.lock().await;
-        self.hydrate_state().await?;
+        let publication = TableCommitLockPublication::new(&self.object_backend);
+        self.register_table_with_publication(entry, &publication).await
+    }
+
+    async fn register_table_with_publication(
+        &self,
+        entry: TableEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
         validate_catalog_entry_version("table", entry.version)?;
         let namespace = parse_namespace_for_store(&entry.namespace)?;
         let table = parse_table_for_store(&entry.table)?;
         table_warehouse_object_prefix(&entry)?;
+        publication.begin_table_bucket(&entry.table_bucket).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "table registration requires a table-bucket publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        publication
+            .prepare(&entry.table_bucket, &entry.namespace, &entry.table)
+            .await?;
+        if !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table) {
+            return Err(TableCatalogStoreError::Internal(
+                "table registration requires a table publication fence".to_string(),
+            ));
+        }
+        let _write_guard = self.write_lock.lock().await;
+        self.hydrate_state().await?;
         let key = Self::table_key(&entry.table_bucket, &namespace, &table);
         let (snapshot, precondition) = {
             let state = self.state.lock().await;
@@ -1328,12 +1424,31 @@ where
     }
 
     async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult> {
-        let _write_guard = self.write_lock.lock().await;
-        self.hydrate_state().await?;
+        let publication = TableCommitLockPublication::new(&self.object_backend);
+        publication.begin_table_bucket(&request.table_bucket).await?;
+        self.commit_table_with_publication(request, &publication).await
+    }
+
+    async fn commit_table_with_publication(
+        &self,
+        request: TableCommitRequest,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<TableCommitResult> {
         let commit_started = Instant::now();
         record_table_commit_attempt(&request.operation);
         let namespace = parse_namespace_for_store(&request.namespace)?;
         let table = parse_table_for_store(&request.table)?;
+        publication
+            .prepare(&request.table_bucket, &request.namespace, &request.table)
+            .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table) {
+            return Err(TableCatalogStoreError::Internal(
+                "table commit requires a table publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        let _write_guard = self.write_lock.lock().await;
+        self.hydrate_state().await?;
         let key = Self::table_key(&request.table_bucket, &namespace, &table);
 
         let committed_existing_result = {
@@ -1396,6 +1511,7 @@ where
                 ))),
             );
         };
+        validate_commit_metadata_digest(&request, &new_metadata_object)?;
         let table_bucket = request.table_bucket.clone();
         let metadata_location = request.new_metadata_location.clone();
         let next_warehouse_location = tokio::task::spawn_blocking(move || {
@@ -1403,6 +1519,36 @@ where
         })
         .await
         .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
+        let current_warehouse_location = {
+            let state = self.state.lock().await;
+            state
+                .tables
+                .get(&key)
+                .map(|entry| entry.warehouse_location.clone())
+                .ok_or_else(|| {
+                    TableCatalogStoreError::NotFound(format!(
+                        "table {}/{}/{}",
+                        request.table_bucket, request.namespace, request.table
+                    ))
+                })?
+        };
+        if next_warehouse_location
+            .as_ref()
+            .is_some_and(|warehouse_location| warehouse_location != &current_warehouse_location)
+            && !publication.holds_table_bucket(&request.table_bucket)
+        {
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Internal(
+                    "table warehouse relocation requires a table-bucket publication fence".to_string(),
+                )),
+            );
+        }
 
         let cas_started = Instant::now();
         let prepared_result = {
@@ -1433,6 +1579,15 @@ where
     }
 
     async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> {
+        let publication = TableCommitLockPublication::new(&self.object_backend);
+        publication.begin_table_bucket(table_bucket).await?;
+        publication.prepare(table_bucket, namespace, table).await?;
+        if !publication.holds_table_bucket(table_bucket) || !publication.holds_table(table_bucket, namespace, table) {
+            return Err(TableCatalogStoreError::Internal(
+                "table drop requires table-bucket and table publication fences".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(&publication);
         let _write_guard = self.write_lock.lock().await;
         self.hydrate_state().await?;
         let namespace = parse_namespace_for_store(namespace)?;