fix(s3-tables): clean up empty tables and parallelize catalog reads

crates/ecstore/src/bucket/metadata_sys.rs

@@ -683,6 +683,7 @@ pub(crate) async fn update_validated_in<F>(
     bucket: &str,
     config_file: &str,
     data: Vec<u8>,
+    expected_incarnation_id: Option<Uuid>,
     validate: F,
 ) -> Result<OffsetDateTime>
 where
@@ -690,7 +691,7 @@ where
 {
     Box::pin(async {
         let sys = require_bucket_metadata_sys_in(ctx)?;
-        let guard = acquire_config_write_guard(sys.clone(), bucket).await?;
+        let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?;
         validate()?;
         update_under_config_write_guard(sys, &guard, config_file, data, None).await
     })
@@ -5079,6 +5080,61 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn validated_config_update_rejects_a_different_bucket_incarnation() {
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = "catalog-cleanup-incarnation";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let incarnation = store
+            .bucket_incarnation_id_from_disk(bucket)
+            .await
+            .expect("read bucket incarnation");
+        let called = std::sync::atomic::AtomicBool::new(false);
+        let result = store
+            .update_bucket_metadata_config_for_incarnation_validated(
+                bucket,
+                "quota.json",
+                br#"{"quota":1}"#.to_vec(),
+                Uuid::new_v4(),
+                || {
+                    called.store(true, std::sync::atomic::Ordering::SeqCst);
+                    Ok(())
+                },
+            )
+            .await;
+        assert!(matches!(result, Err(Error::BucketNotFound(name)) if name == bucket));
+        assert!(
+            !called.load(std::sync::atomic::Ordering::SeqCst),
+            "incarnation must be checked before the external fence"
+        );
+        let (metadata, _) = get_config_from_disk_with_presence_in(&store.ctx, bucket)
+            .await
+            .expect("read unchanged bucket metadata");
+        assert!(
+            metadata.quota_config_json.is_empty(),
+            "a stale cleanup must not modify the replacement bucket"
+        );
+        store
+            .update_bucket_metadata_config_for_incarnation_validated(
+                bucket,
+                "quota.json",
+                br#"{"quota":1}"#.to_vec(),
+                incarnation,
+                || Ok(()),
+            )
+            .await
+            .expect("the current incarnation permits the validated update");
+        let (metadata, _) = get_config_from_disk_with_presence_in(&store.ctx, bucket)
+            .await
+            .expect("read updated bucket metadata");
+        assert_eq!(metadata.quota_config_json, br#"{"quota":1}"#);
+    }
+
     #[tokio::test]
     #[serial]
     async fn quota_admission_rejects_stale_cached_no_quota() {

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -977,7 +977,7 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str, root_pref
         return Err(DiskError::DiskNotFound);
     }
 
-    if !root_prefix.is_empty() {
+    if !root_prefix.is_empty() && root_prefix != DATA_MOVEMENT_MULTIPART_PREFIX {
         let upload_dirs = match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, root_prefix, -1).await {
             Ok(entries) => entries,
             Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(Vec::new()),
@@ -1002,6 +1002,13 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str, root_pref
         .into_iter()
         .map(|sha_dir| sha_dir.trim_end_matches('/').to_owned())
         .filter(|sha_dir| sha_dir.len() == 64 && sha_dir.bytes().all(|byte| byte.is_ascii_hexdigit()))
+        .map(|sha_dir| {
+            if root_prefix.is_empty() {
+                sha_dir
+            } else {
+                format!("{root_prefix}/{sha_dir}")
+            }
+        })
         .collect::<Vec<_>>();
     let listed_upload_dirs = stream::iter(sha_dirs)
         .map(|sha_dir| {
@@ -1124,6 +1131,7 @@ impl SetDisks {
         orig_bucket: &str,
         error_path: &str,
         root_prefix: &str,
+        require_all_disks: bool,
     ) -> Result<(Vec<Option<DiskStore>>, Vec<String>, usize, bool)> {
         let disks = self.disks.read().await.clone();
         if disks.is_empty() {
@@ -1165,7 +1173,8 @@ impl SetDisks {
             }
         }
 
-        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) {
+        let required_discoveries = if require_all_disks { disks.len() } else { discovery_quorum };
+        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, required_discoveries) {
             return Err(to_object_err(err.into(), vec![orig_bucket, error_path]));
         }
 
@@ -1187,7 +1196,7 @@ impl SetDisks {
 
     pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result<Option<String>> {
         let (_, paths, _, _) = self
-            .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET, "")
+            .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET, "", false)
             .await?;
         Ok(paths.into_iter().next())
     }
@@ -1200,8 +1209,9 @@ impl SetDisks {
         upload_identity: &str,
     ) -> Result<Vec<String>> {
         let expected_parent = format!("{DATA_MOVEMENT_MULTIPART_PREFIX}/{}", Self::get_multipart_sha_dir(bucket, object));
-        let (_, candidate_paths, _, has_minority_candidate) =
-            self.discover_multipart_upload_paths(bucket, object, &expected_parent).await?;
+        let (_, candidate_paths, _, has_minority_candidate) = self
+            .discover_multipart_upload_paths(bucket, object, &expected_parent, false)
+            .await?;
         if has_minority_candidate {
             return Err(Error::DecommissionCapacityBlocked {
                 message: "data movement multipart cleanup found an upload path on fewer than the discovery quorum of disks"
@@ -1750,6 +1760,54 @@ impl SetDisks {
         })
     }
 
+    pub(crate) async fn ensure_no_multipart_uploads(&self, bucket: &str) -> Result<()> {
+        // Every disk must confirm absence in both namespaces before cleanup can
+        // remove catalog protection from the bucket.
+        for root_prefix in ["", DATA_MOVEMENT_MULTIPART_PREFIX] {
+            let (disks, paths, _, has_minority_candidate) =
+                self.discover_multipart_upload_paths(bucket, "", root_prefix, true).await?;
+            if has_minority_candidate {
+                return Err(Error::ErasureReadQuorum);
+            }
+            for upload_path in paths {
+                let (sha_dir, _) = upload_path
+                    .rsplit_once('/')
+                    .filter(|(directory, upload_id)| !directory.is_empty() && !upload_id.is_empty())
+                    .ok_or(DiskError::CorruptedFormat)?;
+                let (metadata, errors) =
+                    Self::read_all_fileinfo(&disks, bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, "", false, false, false)
+                        .await?;
+                let (read_quorum, _) = Self::object_quorum_from_meta(&metadata, &errors, self.default_parity_count)?;
+                let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?;
+                if let Some(error) = reduce_read_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, read_quorum) {
+                    return Err(to_object_err(error.into(), vec![bucket, &upload_path]));
+                }
+                let (_, modified, etag) = Self::list_online_disks(&disks, &metadata, &errors, read_quorum);
+                let file = Self::pick_valid_fileinfo(&metadata, modified, etag, read_quorum)?;
+                let (Some(owner_bucket), Some(owner_object)) = (
+                    file.metadata.get(RUSTFS_MULTIPART_BUCKET_KEY),
+                    file.metadata.get(RUSTFS_MULTIPART_OBJECT_KEY),
+                ) else {
+                    return Err(to_object_err(DiskError::CorruptedFormat.into(), vec![bucket, &upload_path]));
+                };
+                if owner_bucket.is_empty()
+                    || owner_object.is_empty()
+                    || Self::get_multipart_sha_dir(owner_bucket, owner_object)
+                        != sha_dir
+                            .strip_prefix(DATA_MOVEMENT_MULTIPART_PREFIX)
+                            .and_then(|suffix| suffix.strip_prefix('/'))
+                            .unwrap_or(sha_dir)
+                {
+                    return Err(to_object_err(DiskError::CorruptedFormat.into(), vec![bucket, &upload_path]));
+                }
+                if owner_bucket == bucket {
+                    return Err(Error::BucketNotEmpty(bucket.to_string()));
+                }
+            }
+        }
+        Ok(())
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn list_multipart_uploads_for_incarnation(
         &self,
@@ -1761,7 +1819,8 @@ impl SetDisks {
         max_uploads: usize,
         expected_incarnation_id: Option<Uuid>,
     ) -> Result<ListMultipartsInfo> {
-        let (disks, candidate_paths, discovery_quorum, _) = self.discover_multipart_upload_paths(bucket, prefix, "").await?;
+        let (disks, candidate_paths, discovery_quorum, _) =
+            self.discover_multipart_upload_paths(bucket, prefix, "", false).await?;
         let listed_uploads = stream::iter(candidate_paths)
             .map(|upload_path| {
                 let disks = &disks;
@@ -11194,6 +11253,163 @@ mod tests {
         assert_eq!(upload_uuid_suffix(&recovered[0]), upload_uuid_suffix(&upload.upload_id));
     }
 
+    #[tokio::test]
+    async fn multipart_absence_inspection_distinguishes_bucket_ownership() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-cleanup-target";
+        let other = "multipart-cleanup-other";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("create target bucket");
+            disk.make_volume(other).await.expect("create other bucket");
+        }
+        set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect("an empty upload namespace permits cleanup");
+        set_disks
+            .new_multipart_upload(other, "object.bin", &ObjectOptions::default())
+            .await
+            .expect("create upload in other bucket");
+        set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect("an upload owned by another bucket permits cleanup");
+        set_disks
+            .new_multipart_upload(bucket, "object.bin", &ObjectOptions::default())
+            .await
+            .expect("create upload in target bucket");
+        assert!(
+            matches!(set_disks.ensure_no_multipart_uploads(bucket).await, Err(Error::BucketNotEmpty(name)) if name == bucket)
+        );
+    }
+
+    #[tokio::test]
+    async fn multipart_absence_inspection_requires_every_disk_online() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-cleanup-offline";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("create bucket");
+        }
+        set_disks.disks.write().await[3] = None;
+        let listed = set_disks
+            .list_multipart_uploads_for_incarnation(bucket, "", None, None, None, 1000, None)
+            .await
+            .expect("ordinary listing preserves its discovery quorum");
+        assert!(listed.uploads.is_empty());
+        let error = set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect_err("an offline disk must prevent confirmation of an empty multipart namespace");
+        assert!(
+            matches!(&error, Error::InsufficientReadQuorum(error_bucket, error_path) if error_bucket == bucket && error_path.is_empty()),
+            "the offline disk must produce a bucket-scoped read quorum error: {error:?}"
+        );
+
+        set_disks.disks.write().await[3] = Some(disk_stores[3].clone());
+        set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect("all disks can confirm the empty namespace after recovery");
+    }
+
+    #[tokio::test]
+    async fn multipart_absence_inspection_rejects_a_failed_disk_enumeration() {
+        let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-cleanup-list-error";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("create bucket");
+        }
+        let sha_dir = SetDisks::get_multipart_sha_dir(bucket, "object.bin");
+        let multipart_dir = temp_dirs[0].path().join(RUSTFS_META_MULTIPART_BUCKET);
+        tokio::fs::create_dir_all(&multipart_dir)
+            .await
+            .expect("create multipart staging directory");
+        tokio::fs::write(multipart_dir.join(&sha_dir), b"not a directory")
+            .await
+            .expect("replace one upload parent directory with a file");
+        multipart_upload_paths_on_disk(disk_stores[0].clone(), bucket, "")
+            .await
+            .expect_err("the affected disk must fail directory enumeration");
+        let listed = set_disks
+            .list_multipart_uploads_for_incarnation(bucket, "", None, None, None, 1000, None)
+            .await
+            .expect("ordinary listing still accepts the remaining discovery quorum");
+        assert!(listed.uploads.is_empty());
+        let error = set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect_err("a failed disk enumeration must prevent confirmation of an empty multipart namespace");
+        assert!(
+            matches!(&error, Error::InsufficientReadQuorum(error_bucket, error_path) if error_bucket == bucket && error_path.is_empty()),
+            "the failed enumeration must produce a bucket-scoped read quorum error: {error:?}"
+        );
+
+        tokio::fs::remove_file(temp_dirs[0].path().join(RUSTFS_META_MULTIPART_BUCKET).join(sha_dir))
+            .await
+            .expect("remove the invalid directory entry");
+        set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect("all disks can confirm the empty namespace after repair");
+    }
+
+    #[tokio::test]
+    async fn multipart_absence_inspection_rejects_data_movement_uploads() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-cleanup-data-movement";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("create bucket");
+        }
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD,
+            format!("v1:{}:{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp_nanos()),
+        );
+        set_disks
+            .new_multipart_upload(
+                bucket,
+                "object.bin",
+                &ObjectOptions {
+                    data_movement: true,
+                    user_defined: metadata,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create data movement upload");
+        assert!(
+            matches!(set_disks.ensure_no_multipart_uploads(bucket).await, Err(Error::BucketNotEmpty(name)) if name == bucket)
+        );
+    }
+
+    #[tokio::test]
+    async fn multipart_absence_inspection_rejects_unreadable_ownership() {
+        let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-cleanup-corrupt";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("create bucket");
+        }
+        set_disks
+            .new_multipart_upload(bucket, "object.bin", &ObjectOptions::default())
+            .await
+            .expect("create multipart upload");
+        let mut corrupted = 0;
+        for temp_dir in &temp_dirs {
+            for path in multipart_meta_files_on_disk(temp_dir, "xl.meta").await {
+                tokio::fs::write(path, b"invalid upload metadata")
+                    .await
+                    .expect("corrupt upload metadata");
+                corrupted += 1;
+            }
+        }
+        assert!(corrupted > 0, "upload metadata must exist before corruption");
+        set_disks
+            .ensure_no_multipart_uploads(bucket)
+            .await
+            .expect_err("unreadable upload ownership must block cleanup");
+    }
+
     /// Regression (issue #5716): a single upload directory whose `xl.meta` was
     /// destroyed (crash mid-write, torn disk state) must degrade to that upload
     /// alone. Failing the whole ListMultipartUploads turns one piece of stale

crates/ecstore/src/storage_api_contracts/mod.rs

@@ -52,7 +52,8 @@ pub(crate) mod list {
 
 pub(crate) mod multipart {
     pub(crate) use rustfs_storage_api::{
-        CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartOperations, MultipartUploadResult, PartInfo,
+        CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartOperations, MultipartUploadInspection,
+        MultipartUploadResult, PartInfo,
     };
 }
 

crates/ecstore/src/store/bucket.rs

@@ -523,7 +523,22 @@ impl ECStore {
     where
         F: FnOnce() -> Result<()> + Send,
     {
-        metadata_sys::update_validated_in(&self.ctx, bucket, config_file, data, validate).await
+        metadata_sys::update_validated_in(&self.ctx, bucket, config_file, data, None, validate).await
+    }
+
+    /// Validate an external fence and the bucket incarnation in one metadata transaction.
+    pub async fn update_bucket_metadata_config_for_incarnation_validated<F>(
+        &self,
+        bucket: &str,
+        config_file: &str,
+        data: Vec<u8>,
+        expected_incarnation_id: Uuid,
+        validate: F,
+    ) -> Result<OffsetDateTime>
+    where
+        F: FnOnce() -> Result<()> + Send,
+    {
+        metadata_sys::update_validated_in(&self.ctx, bucket, config_file, data, Some(expected_incarnation_id), validate).await
     }
 
     pub async fn bucket_incarnation_id(&self, bucket: &str) -> Result<Uuid> {

crates/ecstore/src/store/list_objects.rs

@@ -1364,6 +1364,10 @@ impl ECStore {
     ) -> Result<()> {
         check_list_objs_args(bucket, prefix, &None)?;
 
+        if opts.authoritative {
+            return self.walk_authoritatively(rx, bucket, prefix, result, opts).await;
+        }
+
         let rx = rx.child_token();
         let mut futures = Vec::new();
         let mut inputs = Vec::new();
@@ -1519,6 +1523,57 @@ impl ECStore {
 
         result
     }
+
+    async fn walk_authoritatively(
+        self: Arc<Self>,
+        cancellation: CancellationToken,
+        bucket: &str,
+        prefix: &str,
+        result: Sender<ObjectInfoOrErr>,
+        opts: WalkOptions,
+    ) -> Result<()> {
+        let cancel = cancellation.child_token();
+        let _cancel_on_drop = cancel.clone().drop_guard();
+        let mut scans = Vec::new();
+        let mut inputs = Vec::new();
+        for pool in &self.pools {
+            for set in &pool.disk_set {
+                let (sender, receiver) = mpsc::channel(1);
+                inputs.push(receiver);
+                let mut scan = recursive_walk_list_path_options(bucket.to_owned(), prefix, &opts);
+                // Internal absence checks must include reserved object prefixes.
+                scan.skip_hidden_prefix_check = true;
+                scans.push(set.list_versions_authoritatively(cancel.clone(), scan, sender));
+            }
+        }
+        if scans.is_empty() {
+            return Err(StorageError::ErasureReadQuorum);
+        }
+
+        let (sender, receiver) = mpsc::channel(100);
+        let forwarder = spawn_walk_entry_forwarder(cancel.clone(), receiver, result, opts, bucket.to_owned());
+        let merge = spawn_listing_merge(cancel.clone(), inputs, sender);
+        let scan_results = join_all(scans).await;
+        let merge_result = merge.await.map_err(Error::from)?;
+        let forward_result = forwarder.await.map_err(Error::from);
+
+        // An unavailable set cannot establish absence, even when another set
+        // completed. Only an explicitly missing directory or volume is empty.
+        for scan in scan_results {
+            match scan {
+                Ok(outcome) if !outcome.limit_reached => {}
+                Ok(_) => return Err(StorageError::ErasureReadQuorum),
+                Err(StorageError::FileNotFound | StorageError::VolumeNotFound) => {}
+                Err(err) => return Err(err),
+            }
+        }
+        merge_result?;
+        forward_result?;
+        if cancellation.is_cancelled() {
+            return Err(Error::other("authoritative object walk was cancelled"));
+        }
+        Ok(())
+    }
 }
 
 /// Forwards merged raw entries to a walk consumer as `ObjectInfoOrErr`,
@@ -2828,6 +2883,7 @@ mod test {
     use crate::error::{Result, StorageError};
     use crate::object_api::{ObjectInfo, ObjectOptions, PutObjReader};
     use crate::set_disk::{SetDisks, hermetic_set_disks_isolated};
+    use crate::storage_api_contracts::list::ListOperations as _;
     use crate::storage_api_contracts::namespace::NamespaceLocking as _;
     use crate::storage_api_contracts::object::ObjectIO as _;
     use bytes::Bytes;
@@ -6029,6 +6085,56 @@ mod test {
         Bytes::from(meta.marshal_msg().expect("encode fixture xl.meta"))
     }
 
+    #[tokio::test]
+    async fn authoritative_store_walk_includes_reserved_prefix_versions() {
+        let bucket = "catalog-cleanup-versions";
+        let name = ".rustfs-table/namespaces/ns/tables/table/metadata/metadata.json";
+        let (_dirs, store) = single_set_store(16).await;
+        let set = &store.pools[0].disk_set[0];
+        for (slot, disk) in set.disks.read().await.iter().enumerate() {
+            let disk = disk.as_ref().expect("fixture drive is online");
+            disk.make_volume(bucket).await.expect("create bucket volume");
+            disk.write_all(
+                bucket,
+                &format!("{name}/{STORAGE_FORMAT_FILE}"),
+                versioned_object_metadata(bucket, name, &[(2, 1_700_000_002), (1, 1_700_000_001)], slot),
+            )
+            .await
+            .expect("persist reserved metadata versions");
+        }
+        let (sender, mut receiver) = mpsc::channel(1);
+        let walk = store.clone().walk(
+            CancellationToken::new(),
+            bucket,
+            "",
+            sender,
+            super::WalkOptions {
+                authoritative: true,
+                ..Default::default()
+            },
+        );
+        let collect = async {
+            let mut versions = Vec::new();
+            while let Some(item) = receiver.recv().await {
+                assert!(item.err.is_none(), "reserved metadata must decode: {:?}", item.err);
+                let item = item.item.expect("walk must return an object");
+                versions.push((item.name, item.version_id));
+            }
+            versions
+        };
+        let (walk, versions) = timeout(Duration::from_secs(10), async { tokio::join!(walk, collect) })
+            .await
+            .expect("authoritative reserved-prefix walk must terminate");
+        walk.expect("authoritative walk must succeed");
+        assert_eq!(
+            versions,
+            [
+                (name.to_string(), Some(Uuid::from_u128(1))),
+                (name.to_string(), Some(Uuid::from_u128(2)))
+            ]
+        );
+    }
+
     // A versioned DeleteObject(versionId) can reach write quorum while three
     // drives are offline; they keep the deleted version after they return. A
     // different drive going offline later must not turn that sub-quorum
@@ -6099,6 +6205,22 @@ mod test {
                 .await
                 .expect("set walk must terminate");
             assert!(matches!(walk, Err(StorageError::ErasureReadQuorum)), "{walk:?}");
+            let (sender, mut receiver) = mpsc::channel(16);
+            let walk = store.clone().walk(
+                CancellationToken::new(),
+                bucket,
+                "",
+                sender,
+                super::WalkOptions {
+                    authoritative: true,
+                    ..Default::default()
+                },
+            );
+            let drain = async { while receiver.recv().await.is_some() {} };
+            let (walk, ()) = timeout(Duration::from_secs(10), async { tokio::join!(walk, drain) })
+                .await
+                .expect("authoritative store walk must terminate");
+            assert!(matches!(walk, Err(StorageError::ErasureReadQuorum)), "{walk:?}");
             let recursive_delete_scan = set
                 .clone()
                 .inner_list_object_versions_for_recursive_delete(bucket, "", None, None, 1000)

crates/ecstore/src/store/multipart.rs

@@ -23,6 +23,23 @@ use std::collections::HashSet;
 
 const MULTIPART_LIST_SET_CONCURRENCY: usize = 4;
 
+#[async_trait::async_trait]
+impl crate::storage_api_contracts::multipart::MultipartUploadInspection for ECStore {
+    type Error = Error;
+
+    async fn ensure_no_multipart_uploads(&self, bucket: &str) -> Result<()> {
+        let (_, lifecycle_guard) = self
+            .guard_multipart_bucket_incarnation(bucket, &ObjectOptions::default())
+            .await?;
+        for pool in &self.pools {
+            for set in &pool.disk_set {
+                set.ensure_no_multipart_uploads(bucket).await?;
+            }
+        }
+        ensure_multipart_bucket_lifecycle_guard_held(lifecycle_guard.as_ref(), bucket, "")
+    }
+}
+
 #[cfg(test)]
 static DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<Uuid, usize>>> =
     std::sync::OnceLock::new();

crates/storage-api/src/lib.rs

@@ -90,6 +90,7 @@ pub use bucket::BUCKET_LIFECYCLE_LOCK_OBJECT;
 pub use bucket::{BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp};
 pub use capability::{CapabilitySnapshotError, CapabilityState, CapabilityStatus};
 pub use error::{StorageErrorCode, StorageResult};
+pub use multipart::MultipartUploadInspection;
 pub use multipart::{CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartUploadResult, PartInfo};
 pub use object::DeleteAccounting;
 pub use object::ObjectLockDeleteOptions;

crates/storage-api/src/multipart.rs

@@ -15,6 +15,15 @@
 use std::collections::HashMap;
 use time::OffsetDateTime;
 
+/// Checks upload absence before removing a bucket's data protections.
+#[async_trait::async_trait]
+pub trait MultipartUploadInspection: Send + Sync {
+    type Error: std::error::Error + Send + Sync + 'static;
+
+    /// Unknown or unreadable upload ownership must fail this check.
+    async fn ensure_no_multipart_uploads(&self, bucket: &str) -> Result<(), Self::Error>;
+}
+
 #[derive(Debug, Default, Clone)]
 pub struct MultipartUploadResult {
     pub upload_id: String,

crates/storage-api/src/object.rs

@@ -366,6 +366,8 @@ pub struct WalkOptions<Filter> {
     pub filter: Option<Filter>,
     pub marker: Option<String>,
     pub latest_only: bool,
+    /// Resolve every set at its version read quorum and fail on incomplete scans.
+    pub authoritative: bool,
     pub ask_disks: String,
     pub versions_sort: WalkVersionsSortOrder,
     pub limit: usize,
@@ -380,6 +382,7 @@ impl<Filter> Default for WalkOptions<Filter> {
             filter: None,
             marker: None,
             latest_only: false,
+            authoritative: false,
             ask_disks: String::new(),
             versions_sort: WalkVersionsSortOrder::default(),
             limit: 0,
@@ -999,6 +1002,7 @@ mod tests {
         assert!(opts.filter.is_none());
         assert!(opts.marker.is_none());
         assert!(!opts.latest_only);
+        assert!(!opts.authoritative);
         assert!(opts.ask_disks.is_empty());
         assert!(matches!(opts.versions_sort, WalkVersionsSortOrder::Ascending));
         assert_eq!(opts.limit, 0);

docs/operations/s3-tables-cutover-runbook.md

@@ -91,6 +91,16 @@ Use `GET /iceberg/v1/buckets/{warehouse}` to check `enabled` and `disable-suppor
 
 Review the bucket lifecycle configuration before disabling: clearing table-bucket protection allows existing object expiration rules to run again. Objects are preserved by the disable operation itself. Catalog disablement requires no namespaces, retained table/view resources, or objects under `.rustfs-table/`; an active rename or backing migration also blocks it. Clean up through supported catalog operations; do not manually delete internal metadata to bypass these checks.
 
-The server persists an inactive catalog entry before clearing the bucket marker. If marker publication fails, retry the same DELETE; the marker retains object protections until the retry succeeds. A subsequent PUT explicitly enables the bucket again. Stale requests cannot initialize an inactive catalog entry. No table metadata or object data is deleted by either state change.
+The server persists an inactive catalog entry before clearing the bucket marker. If marker publication fails, retry the same DELETE; the marker retains object protections until the retry succeeds. A subsequent PUT explicitly enables the bucket again. Stale requests cannot initialize an inactive catalog entry. A plain DELETE and a subsequent PUT preserve table metadata and object data.
+
+### Clean Up Deleted Empty Tables
+
+For the object-backed catalog, `DELETE /iceberg/v1/buckets/{warehouse}?cleanup=true` removes the metadata and internal records left by deleted empty tables, then disables the table catalog. Ordinary table deletion continues to preserve files. The cleanup option does not change persisted catalog formats or require migration, and it can process empty tables deleted before the server upgrade.
+
+Complete the server upgrade across all serving nodes and set the existing `RUSTFS_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED=true` setting before requesting cleanup. Delete the empty tables and their namespaces through the catalog APIs, then send the cleanup request with `admin:SetTableBucket` and `s3:GetObject`/`s3:DeleteObject` permissions for the target metadata objects. The `/_iceberg/v1` alias accepts the same option. Omitting `cleanup` or setting it to `false` keeps the ordinary disable behavior; repeated or invalid values are rejected.
+
+Cleanup accepts only recognized empty-table metadata in the default managed warehouse. Active catalog resources, snapshots or other data references, user objects, object versions, incomplete multipart uploads, corrupt or unknown metadata, and unfinished catalog operations block cleanup. The server validates the complete cleanup plan and object permissions before deleting any object. Durable-strong catalogs retain their existing disable procedure.
+
+If a storage operation fails during cleanup, retry the same request. The server checks the remaining objects again and keeps the bucket's table protection enabled until cleanup and catalog disablement complete. No separate cleanup task file is created. After a successful response with `enabled: false`, delete the empty bucket through the normal S3 `DeleteBucket` operation.
 
 Before rolling back to a server version without this endpoint, re-enable disabled catalogs with PUT while the supporting version is running. Older enable handlers do not reactivate inactive catalog entries. Complete the server upgrade across the cluster before using disablement.

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

@@ -50,17 +50,23 @@ pub struct DisableTableBucketHandler {}
 
 #[async_trait::async_trait]
 impl Operation for DisableTableBucketHandler {
-    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 resource = TableCatalogResource::warehouse(&warehouse);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?;
+        let cleanup = table_bucket_cleanup_requested_from_query(&req.uri)?;
         let app_context = runtime_sources::app_context_from_extensions(&req.extensions);
         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 = disable_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
+        let publication = TableCommitObjectBackend::for_request(backend, req);
+        let response = if cleanup {
+            cleanup_empty_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?
+        } else {
+            disable_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?
+        };
         crate::app::bucket_usecase::notify_table_bucket_metadata_reload(warehouse.clone(), "disable table bucket", app_context)
             .await;
         build_json_response(StatusCode::OK, &response)

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

@@ -2092,17 +2092,25 @@ fn warehouse_from_config_query(uri: &http::Uri) -> S3Result<Option<String>> {
 }
 
 fn rest_purge_requested_from_query(uri: &http::Uri) -> S3Result<bool> {
-    let mut purge_requested = None;
+    table_catalog_boolean_query(uri, "purgeRequested")
+}
+
+fn table_bucket_cleanup_requested_from_query(uri: &http::Uri) -> S3Result<bool> {
+    table_catalog_boolean_query(uri, "cleanup")
+}
+
+fn table_catalog_boolean_query(uri: &http::Uri, parameter: &str) -> S3Result<bool> {
+    let mut requested = None;
     if let Some(query) = uri.query() {
         for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
-            if key != "purgeRequested" {
+            if key != parameter {
                 continue;
             }
-            if purge_requested.is_some() {
+            if requested.is_some() {
                 return Err(iceberg_rest_error(
                     ICEBERG_ERROR_BAD_REQUEST,
                     StatusCode::BAD_REQUEST,
-                    "purgeRequested query parameter must not be repeated",
+                    format!("{parameter} query parameter must not be repeated"),
                 ));
             }
             let value = if value.eq_ignore_ascii_case("true") {
@@ -2113,13 +2121,13 @@ fn rest_purge_requested_from_query(uri: &http::Uri) -> S3Result<bool> {
                 return Err(iceberg_rest_error(
                     ICEBERG_ERROR_BAD_REQUEST,
                     StatusCode::BAD_REQUEST,
-                    "purgeRequested query parameter must be true or false",
+                    format!("{parameter} query parameter must be true or false"),
                 ));
             };
-            purge_requested = Some(value);
+            requested = Some(value);
         }
     }
-    Ok(purge_requested.unwrap_or(false))
+    Ok(requested.unwrap_or(false))
 }
 
 fn rest_table_snapshot_selection_from_query(uri: &http::Uri) -> S3Result<RestTableSnapshotSelection> {
@@ -2572,6 +2580,61 @@ where
     table_bucket_response(store, bucket, true).await
 }
 
+async fn cleanup_empty_table_bucket_response<B>(
+    store: &crate::table_catalog::ConfiguredTableCatalogStore<B>,
+    publication: &TableCommitObjectBackend<B>,
+    object_store: &ECStore,
+    bucket: &str,
+) -> S3Result<TableBucketResponse>
+where
+    B: crate::table_catalog::TableCatalogObjectBackend,
+{
+    if !publication.publication_fence_fleet_confirmed {
+        return Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::Conflict(
+            "empty table bucket cleanup requires publication fences on every cluster node".to_string(),
+        )));
+    }
+    crate::table_catalog::TableCommitPublication::begin_table_bucket(publication, bucket)
+        .await
+        .map_err(catalog_store_error)?;
+    let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(publication);
+    let incarnation = object_store
+        .bucket_incarnation_id_from_disk(bucket)
+        .await
+        .map_err(|err| ApiError::other(format!("failed to read table bucket identity: {err}")))?;
+    let prepared = store
+        .prepare_empty_table_bucket_cleanup(table_bucket_entry_from_metadata_marker(bucket), publication)
+        .await
+        .map_err(catalog_store_error);
+    let mut cleanup = publication.finish(prepared).await?;
+    for object in cleanup.metadata_objects() {
+        if let Err(err) = publication.authorize(bucket, object, S3Action::DeleteObjectAction).await {
+            return publication.finish(Err(catalog_store_error(err))).await;
+        }
+    }
+    store
+        .execute_empty_table_bucket_cleanup(&mut cleanup, incarnation)
+        .await
+        .map_err(catalog_store_error)?;
+    let fence = cleanup.mutation_fence();
+    object_store
+        .update_bucket_metadata_config_for_incarnation_validated(
+            bucket,
+            crate::table_catalog::TABLE_BUCKET_MARKER_CONFIG,
+            Vec::new(),
+            incarnation,
+            || {
+                fence
+                    .ensure_held()
+                    .map_err(|err| crate::admin::storage_api::error::StorageError::other(err.to_string()))
+            },
+        )
+        .await
+        .map_err(|err| ApiError::other(format!("failed to disable table bucket: {err}")))?;
+    drop(cleanup);
+    table_bucket_response(store, bucket, false).await
+}
+
 async fn disable_table_bucket_response<S, B>(
     store: &S,
     publication: &TableCommitObjectBackend<B>,

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

@@ -87,6 +87,15 @@ async fn call_load_table_handler(request: S3Request<Body>) -> S3Result<S3Respons
     RestLoadTableHandler {}.call(request, matched.params).await
 }
 
+async fn call_disable_table_bucket_handler(request: S3Request<Body>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let mut router = matchit::Router::new();
+    router.insert("/iceberg/v1/buckets/{warehouse}", ()).unwrap();
+    router.insert("/_iceberg/v1/buckets/{warehouse}", ()).unwrap();
+    let path = request.uri.path().to_string();
+    let matched = router.at(&path).unwrap();
+    DisableTableBucketHandler {}.call(request, matched.params).await
+}
+
 async fn call_maintenance_worker_handler(request: S3Request<Body>) -> S3Result<S3Response<(StatusCode, Body)>> {
     let mut router = matchit::Router::new();
     router
@@ -1255,6 +1264,28 @@ fn drop_table_purge_query_is_explicit_and_strict() {
     }
 }
 
+#[test]
+fn table_bucket_cleanup_query_is_explicit_and_strict() {
+    for (uri, expected) in [
+        ("/iceberg/v1/buckets/analytics", false),
+        ("/iceberg/v1/buckets/analytics?cleanup=false", false),
+        ("/iceberg/v1/buckets/analytics?cleanup=true", true),
+        ("/_iceberg/v1/buckets/analytics?cleanup=true", true),
+    ] {
+        assert_eq!(table_bucket_cleanup_requested_from_query(&uri.parse().unwrap()).unwrap(), expected);
+    }
+    for query in [
+        "cleanup=",
+        "cleanup=1",
+        "cleanup=yes",
+        "cleanup=true&cleanup=false",
+        "cleanup=true&cleanup=true",
+    ] {
+        let uri = format!("/iceberg/v1/buckets/analytics?{query}").parse().unwrap();
+        assert!(table_bucket_cleanup_requested_from_query(&uri).is_err(), "invalid query: {query}");
+    }
+}
+
 #[test]
 fn catalog_conflicts_use_operation_specific_iceberg_errors() {
     let already_exists = catalog_store_already_exists_error(crate::table_catalog::TableCatalogStoreError::Conflict(
@@ -1689,9 +1720,7 @@ fn table_bucket_toggle_handlers_reload_peer_metadata_after_publication() {
         (enable, "enable_table_bucket_response", "enable table bucket"),
         (disable, "disable_table_bucket_response", "disable table bucket"),
     ] {
-        let response_offset = block
-            .find(&format!("let response = {response}"))
-            .expect("table bucket toggle should publish local state");
+        let response_offset = block.find(response).expect("table bucket toggle should publish local state");
         let reload = block
             .find(&format!("notify_table_bucket_metadata_reload(warehouse.clone(), \"{operation}\""))
             .expect("table bucket toggle should reload peer metadata");
@@ -16274,3 +16303,698 @@ async fn disable_table_bucket_preserves_objects_and_can_be_reenabled() {
             .table_bucket_enabled()
     );
 }
+
+async fn create_dropped_empty_tables_for_cleanup<S, B>(
+    catalog: &S,
+    backend: &B,
+    object_store: &ECStore,
+    bucket: &str,
+    table_count: usize,
+) -> Vec<String>
+where
+    S: TableCatalogStore + ?Sized,
+    B: TableCatalogObjectBackend,
+{
+    object_store.make_bucket(bucket, &MakeBucketOptions::default()).await.unwrap();
+    enable_table_bucket_response(catalog, &TableCommitObjectBackend::trusted(backend.clone()), object_store, bucket)
+        .await
+        .unwrap();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").unwrap();
+    create_namespace_response(
+        catalog,
+        bucket,
+        CreateNamespaceRequest {
+            namespace: vec!["analytics".to_string()],
+            properties: BTreeMap::new(),
+        },
+        true,
+    )
+    .await
+    .unwrap();
+    let mut metadata_objects = Vec::new();
+    for number in 0..table_count {
+        let name = format!("events_{number:02}");
+        let request = serde_json::from_value(serde_json::json!({
+            "name": name,
+            "schema": {
+                "type": "struct",
+                "schema-id": 0,
+                "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+            }
+        }))
+        .unwrap();
+        create_table_response(
+            catalog,
+            &TableCommitObjectBackend::trusted(backend.clone()),
+            bucket,
+            &namespace,
+            request,
+            true,
+        )
+        .await
+        .unwrap();
+        metadata_objects.push(
+            catalog
+                .load_table(bucket, &namespace.public_name(), &name)
+                .await
+                .unwrap()
+                .unwrap()
+                .metadata_location,
+        );
+    }
+    for number in 0..table_count {
+        catalog
+            .drop_table(bucket, &namespace.public_name(), &format!("events_{number:02}"))
+            .await
+            .unwrap();
+    }
+    catalog.drop_namespace(bucket, &namespace.public_name()).await.unwrap();
+    metadata_objects
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_reclaims_twenty_dropped_tables_and_allows_bucket_deletion() {
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-{}", Uuid::new_v4().simple());
+    let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
+        object_store.clone(),
+        crate::table_catalog::StrongTableCatalogRuntime::default(),
+    );
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    let metadata_objects = create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 20).await;
+    assert!(
+        disable_table_bucket_response(
+            &catalog,
+            &TableCommitObjectBackend::trusted(backend.clone()),
+            object_store.as_ref(),
+            &bucket,
+        )
+        .await
+        .is_err()
+    );
+    for object in &metadata_objects {
+        assert!(backend.object_exists(&bucket, object).await.unwrap());
+    }
+
+    let authorizations = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let publication = TableCommitObjectBackend::test(backend.clone(), authorizations.clone(), None);
+    let result = cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket).await;
+    assert!(!publication.finish(result).await.unwrap().enabled);
+    for object in &metadata_objects {
+        assert!(!backend.object_exists(&bucket, object).await.unwrap());
+        let authorized = authorizations.lock().await;
+        assert!(authorized.contains(&(object.clone(), S3Action::GetObjectAction)));
+        assert!(authorized.contains(&(object.clone(), S3Action::DeleteObjectAction)));
+    }
+    assert!(
+        !object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+    let publication = TableCommitObjectBackend::trusted(backend);
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_validates_authorization_and_metadata_before_deleting() {
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-check-{}", Uuid::new_v4().simple());
+    let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
+        object_store.clone(),
+        crate::table_catalog::StrongTableCatalogRuntime::default(),
+    );
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    let metadata_objects = create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 2).await;
+    let publication = TableCommitObjectBackend::rolling_upgrade(backend.clone());
+    assert!(
+        cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .is_err()
+    );
+    for object in &metadata_objects {
+        assert!(backend.object_exists(&bucket, object).await.unwrap());
+    }
+
+    let authorizations = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let denied_object = metadata_objects.last().unwrap().clone();
+    let publication = TableCommitObjectBackend::test(backend.clone(), authorizations, Some(denied_object.clone()));
+    let result = cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket).await;
+    assert_eq!(publication.finish(result).await.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+    for object in &metadata_objects {
+        assert!(backend.object_exists(&bucket, object).await.unwrap());
+    }
+
+    let original = backend.read_object(&bucket, &denied_object).await.unwrap().unwrap().data;
+    backend
+        .put_object(
+            &bucket,
+            &denied_object,
+            b"{".to_vec(),
+            crate::table_catalog::TableCatalogPutPrecondition::Any,
+        )
+        .await
+        .unwrap();
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    assert!(
+        cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .is_err()
+    );
+    for object in &metadata_objects {
+        assert!(backend.object_exists(&bucket, object).await.unwrap());
+    }
+    assert!(
+        object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+
+    backend
+        .put_object(&bucket, &denied_object, original, crate::table_catalog::TableCatalogPutPrecondition::Any)
+        .await
+        .unwrap();
+    let publication = TableCommitObjectBackend::trusted(backend);
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_handler_enforces_request_object_permissions() {
+    temp_env::async_with_vars(
+        [
+            (
+                crate::table_catalog::ENV_TABLE_CATALOG_BACKING,
+                Some(crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT),
+            ),
+            (crate::table_catalog::ENV_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED, Some("true")),
+        ],
+        async {
+            let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+            let bucket = format!("cleanup-request-{}", Uuid::new_v4().simple());
+            rustfs_iam::store::object::ObjectStore::new(object_store.clone())
+                .save_iam_config(
+                    serde_json::json!({"version": 1}),
+                    format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+                )
+                .await
+                .unwrap();
+            let iam = rustfs_iam::build_iam_sys(object_store.clone()).await.unwrap();
+            let access_key = "cleanup-request-user";
+            let secret_key = "cleanup-request-user-secret";
+            iam.create_user(
+                access_key,
+                &AddOrUpdateUserReq {
+                    secret_key: secret_key.to_string(),
+                    policy: None,
+                    status: AccountStatus::Enabled,
+                },
+            )
+            .await
+            .unwrap();
+            let context = Arc::new(AppContext::new(
+                object_store.clone(),
+                Arc::new(RequestIam { handle: iam.clone() }),
+                Arc::new(RequestKms),
+            ));
+            assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+                access_key: "cleanup-request-root".to_string(),
+                secret_key: "cleanup-request-root-secret".to_string(),
+                status: "on".to_string(),
+                ..Default::default()
+            }));
+            let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
+                object_store.clone(),
+                context.table_catalog_strong_runtime(),
+            );
+            let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+                backend.clone(),
+                crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+            );
+            let metadata_objects =
+                create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 2).await;
+            let request = |prefix: &str| {
+                let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+                request.method = Method::DELETE;
+                request.uri = format!("{prefix}/v1/buckets/{bucket}?cleanup=true").parse().unwrap();
+                request
+            };
+            let object_resource = format!("arn:aws:s3:::{bucket}/*");
+            let first_metadata_resource = format!("arn:aws:s3:::{bucket}/{}", metadata_objects[0]);
+            for (policy_name, object_statements, allowed) in [
+                ("cleanup-request-no-read", Vec::new(), false),
+                (
+                    "cleanup-request-partial-delete",
+                    vec![
+                        serde_json::json!({"Effect":"Allow","Action":["s3:GetObject"],"Resource":[object_resource]}),
+                        serde_json::json!({"Effect":"Allow","Action":["s3:DeleteObject"],"Resource":[first_metadata_resource]}),
+                    ],
+                    false,
+                ),
+                (
+                    "cleanup-request-full",
+                    vec![serde_json::json!({
+                        "Effect":"Allow",
+                        "Action":["s3:GetObject","s3:DeleteObject"],
+                        "Resource":[object_resource],
+                    })],
+                    true,
+                ),
+            ] {
+                let mut statements = vec![serde_json::json!({"Effect":"Allow","Action":["admin:SetTableBucket"]})];
+                statements.extend(object_statements);
+                let policy = serde_json::to_vec(&serde_json::json!({"Version":"2012-10-17","Statement":statements})).unwrap();
+                iam.set_policy(policy_name, Policy::parse_config(&policy).unwrap())
+                    .await
+                    .unwrap();
+                iam.policy_db_set(access_key, UserType::Reg, false, policy_name)
+                    .await
+                    .unwrap();
+
+                let result = call_disable_table_bucket_handler(request(if allowed { "/_iceberg" } else { "/iceberg" })).await;
+                if allowed {
+                    let response = result.expect("request-scoped object permissions should authorize cleanup");
+                    assert_eq!(response.output.0, StatusCode::OK);
+                    let body: serde_json::Value = serde_json::from_slice(&response.output.1.bytes().unwrap()).unwrap();
+                    assert_eq!(body["enabled"], false);
+                    for object in &metadata_objects {
+                        assert!(!backend.object_exists(&bucket, object).await.unwrap());
+                    }
+                } else {
+                    assert_eq!(result.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+                    for object in &metadata_objects {
+                        assert!(backend.object_exists(&bucket, object).await.unwrap());
+                    }
+                    assert!(
+                        object_store
+                            .get_bucket_metadata(&bucket)
+                            .await
+                            .unwrap()
+                            .table_bucket_enabled()
+                    );
+                    assert_eq!(
+                        catalog.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+                        crate::table_catalog::TableCatalogEntryState::Active,
+                    );
+                }
+            }
+            object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_retries_after_marker_write_rejection() {
+    use crate::table_catalog::TableCommitPublication;
+
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-marker-{}", Uuid::new_v4().simple());
+    let backend = TestTableCatalogObjectBackend::default();
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    let metadata_objects = create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 2).await;
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    TableCommitPublication::begin_table_bucket(&publication, &bucket)
+        .await
+        .unwrap();
+    let incarnation = object_store.bucket_incarnation_id_from_disk(&bucket).await.unwrap();
+    let mut cleanup = catalog
+        .prepare_empty_table_bucket_cleanup(table_bucket_entry_from_metadata_marker(&bucket), &publication)
+        .await
+        .unwrap();
+    catalog
+        .execute_empty_table_bucket_cleanup(&mut cleanup, incarnation)
+        .await
+        .unwrap();
+    let marker_result = object_store
+        .update_bucket_metadata_config_for_incarnation_validated(
+            &bucket,
+            crate::table_catalog::TABLE_BUCKET_MARKER_CONFIG,
+            Vec::new(),
+            incarnation,
+            || {
+                Err(crate::admin::storage_api::error::StorageError::other(
+                    "injected marker publication rejection",
+                ))
+            },
+        )
+        .await;
+    assert!(marker_result.is_err());
+    drop(cleanup);
+    TableCommitPublication::complete(&publication);
+    for object in &metadata_objects {
+        assert!(!backend.object_exists(&bucket, object).await.unwrap());
+    }
+    assert_eq!(
+        catalog.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+        crate::table_catalog::TableCatalogEntryState::Deleted,
+    );
+    assert!(
+        object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+
+    let publication = TableCommitObjectBackend::trusted(backend);
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    assert!(
+        !object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_reads_incarnation_after_publication_wait() {
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-recreated-{}", Uuid::new_v4().simple());
+    let backend = TestTableCatalogObjectBackend::default();
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 0).await;
+    let original_incarnation = object_store.bucket_incarnation_id_from_disk(&bucket).await.unwrap();
+    let publication_lock = crate::table_catalog::default_table_bucket_publication_lock_path();
+    let guard = backend.acquire_write_lock(&bucket, &publication_lock).await.unwrap();
+    let attempts = backend.write_lock_acquisition_count(&bucket, &publication_lock).await;
+    let cleanup_bucket = bucket.clone();
+    let cleanup_store = object_store.clone();
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    let cleanup = tokio::spawn(async move {
+        cleanup_empty_table_bucket_response(&catalog, &publication, cleanup_store.as_ref(), &cleanup_bucket).await
+    });
+    tokio::time::timeout(StdDuration::from_secs(5), async {
+        while backend.write_lock_acquisition_count(&bucket, &publication_lock).await == attempts {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("cleanup should wait for its publication lock");
+
+    // Replace the real bucket while the in-memory publication backend holds cleanup at its lock boundary.
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+    object_store
+        .make_bucket(&bucket, &MakeBucketOptions::default())
+        .await
+        .unwrap();
+    enable_table_bucket_marker(object_store.as_ref(), &bucket).await.unwrap();
+    let replacement_incarnation = object_store.bucket_incarnation_id_from_disk(&bucket).await.unwrap();
+    assert_ne!(replacement_incarnation, original_incarnation);
+    drop(guard);
+    let response = tokio::time::timeout(StdDuration::from_secs(10), cleanup)
+        .await
+        .expect("cleanup should resume after publication becomes available")
+        .unwrap()
+        .expect("cleanup must use the incarnation observed while holding its publication lock");
+    assert!(!response.enabled);
+    assert_eq!(
+        object_store.bucket_incarnation_id_from_disk(&bucket).await.unwrap(),
+        replacement_incarnation
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_retries_metadata_delete_failures_without_removing_protection() {
+    use crate::admin::storage_api::RUSTFS_META_BUCKET;
+    use crate::table_catalog::{TableCatalogEntryState, TableCatalogObjectPaths};
+
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    for fail_after_delete in [false, true] {
+        let bucket = format!("cleanup-retry-{}", Uuid::new_v4().simple());
+        let backend = TestTableCatalogObjectBackend::content_addressed();
+        let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+            backend.clone(),
+            crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        );
+        let mut metadata_objects =
+            create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 3).await;
+        metadata_objects.sort();
+        let paths = TableCatalogObjectPaths::default();
+        let index_prefix = paths.warehouse_index_entries_prefix(&bucket);
+        let indexes = backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap();
+        assert!(indexes.len() >= metadata_objects.len());
+        if fail_after_delete {
+            backend.fail_after_next_delete(&bucket, &metadata_objects[1]).await;
+        } else {
+            backend.fail_delete_attempt(&bucket, &metadata_objects[1], 1).await;
+        }
+
+        let publication = TableCommitObjectBackend::trusted(backend.clone());
+        let error = cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains("injected"));
+        assert!(!backend.object_exists(&bucket, &metadata_objects[0]).await.unwrap());
+        assert_eq!(backend.object_exists(&bucket, &metadata_objects[1]).await.unwrap(), !fail_after_delete);
+        assert!(backend.object_exists(&bucket, &metadata_objects[2]).await.unwrap());
+        assert_eq!(backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap(), indexes);
+        assert_eq!(
+            catalog.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+            TableCatalogEntryState::Active
+        );
+        assert!(
+            object_store
+                .get_bucket_metadata(&bucket)
+                .await
+                .unwrap()
+                .table_bucket_enabled()
+        );
+
+        let restarted = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+            backend.clone(),
+            crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        );
+        let publication = TableCommitObjectBackend::trusted(backend.clone());
+        assert!(
+            !cleanup_empty_table_bucket_response(&restarted, &publication, object_store.as_ref(), &bucket)
+                .await
+                .unwrap()
+                .enabled
+        );
+        assert!(backend.list_objects(&bucket, "").await.unwrap().is_empty());
+        assert!(
+            backend
+                .list_objects(RUSTFS_META_BUCKET, &index_prefix)
+                .await
+                .unwrap()
+                .is_empty()
+        );
+        assert_eq!(
+            restarted.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+            TableCatalogEntryState::Deleted
+        );
+        assert!(
+            !object_store
+                .get_bucket_metadata(&bucket)
+                .await
+                .unwrap()
+                .table_bucket_enabled()
+        );
+        object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+    }
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_retries_partial_warehouse_index_deletion() {
+    use crate::admin::storage_api::RUSTFS_META_BUCKET;
+    use crate::table_catalog::{TableCatalogEntryState, TableCatalogObjectPaths};
+
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-index-retry-{}", Uuid::new_v4().simple());
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 2).await;
+    let index_prefix = TableCatalogObjectPaths::default().warehouse_index_entries_prefix(&bucket);
+    let indexes = backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap();
+    assert!(indexes.len() >= 2);
+    let last_index = indexes.last().unwrap();
+    backend.fail_delete_attempt(RUSTFS_META_BUCKET, last_index, 1).await;
+
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    let error = cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+        .await
+        .unwrap_err();
+    assert!(error.to_string().contains("injected"));
+    assert!(backend.list_objects(&bucket, "").await.unwrap().is_empty());
+    assert_eq!(
+        backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap(),
+        vec![last_index.clone()]
+    );
+    assert_eq!(
+        catalog.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+        TableCatalogEntryState::Active
+    );
+    assert!(
+        object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    assert!(
+        backend
+            .list_objects(RUSTFS_META_BUCKET, &index_prefix)
+            .await
+            .unwrap()
+            .is_empty()
+    );
+    assert!(
+        !object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_stops_before_delete_when_publication_lease_is_lost() {
+    use crate::admin::storage_api::RUSTFS_META_BUCKET;
+    use crate::table_catalog::test_support::RecordingLeaseClient;
+    use crate::table_catalog::{TableCatalogEntryState, TableCatalogLockGuard, TableCatalogObjectPaths, TableCommitPublication};
+    use rustfs_lock::LockClient;
+
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = format!("cleanup-lease-{}", Uuid::new_v4().simple());
+    let backend = TestTableCatalogObjectBackend::content_addressed();
+    let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+        backend.clone(),
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+    );
+    let metadata_objects = create_dropped_empty_tables_for_cleanup(&catalog, &backend, object_store.as_ref(), &bucket, 1).await;
+    let index_prefix = TableCatalogObjectPaths::default().warehouse_index_entries_prefix(&bucket);
+    let indexes = backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap();
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    TableCommitPublication::begin_table_bucket(&publication, &bucket)
+        .await
+        .unwrap();
+    let client = Arc::new(RecordingLeaseClient {
+        inner: rustfs_lock::LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::Enabled(Arc::new(
+            rustfs_lock::FastObjectLockManager::new(),
+        )))),
+        acquired: std::sync::Mutex::new(None),
+    });
+    let lock = rustfs_lock::NamespaceLockWrapper::new(
+        rustfs_lock::NamespaceLock::with_clients_and_quorum(
+            "cleanup-publication-fence".to_string(),
+            vec![client.clone() as Arc<dyn LockClient>],
+            1,
+        ),
+        rustfs_lock::ObjectKey::new(bucket.as_str(), "cleanup-publication-fence"),
+        "cleanup-publication-owner".to_string(),
+    );
+    let guard = lock.get_read_lock(StdDuration::from_secs(5)).await.unwrap();
+    let signal = guard.lock_lost_signal().unwrap();
+    let lock_id = client.acquired.lock().unwrap().clone().unwrap();
+    publication
+        .publication
+        .lock()
+        .guards
+        .push(TableCatalogLockGuard::test_namespace(guard));
+    let mut cleanup = catalog
+        .prepare_empty_table_bucket_cleanup(table_bucket_entry_from_metadata_marker(&bucket), &publication)
+        .await
+        .unwrap();
+    let incarnation = object_store.bucket_incarnation_id_from_disk(&bucket).await.unwrap();
+    let pause = backend.pause_next_delete(&bucket, &metadata_objects[0]).await;
+    let operation = catalog.execute_empty_table_bucket_cleanup(&mut cleanup, incarnation);
+    let interfere = async {
+        tokio::time::timeout(StdDuration::from_secs(5), pause.wait_started())
+            .await
+            .unwrap();
+        assert!(client.force_release(&lock_id).await.unwrap());
+        tokio::time::timeout(StdDuration::from_secs(15), signal.notified())
+            .await
+            .expect("cleanup publication must observe lease loss");
+        pause.release();
+    };
+    let (result, ()) = tokio::join!(operation, interfere);
+    assert!(matches!(result, Err(crate::table_catalog::TableCatalogStoreError::Unavailable(_))));
+    drop(cleanup);
+    TableCommitPublication::complete(&publication);
+    assert!(backend.object_exists(&bucket, &metadata_objects[0]).await.unwrap());
+    assert_eq!(backend.list_objects(RUSTFS_META_BUCKET, &index_prefix).await.unwrap(), indexes);
+    assert_eq!(
+        catalog.get_table_bucket(&bucket).await.unwrap().unwrap().state,
+        TableCatalogEntryState::Active
+    );
+    assert!(
+        object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+
+    let publication = TableCommitObjectBackend::trusted(backend.clone());
+    assert!(
+        !cleanup_empty_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
+            .await
+            .unwrap()
+            .enabled
+    );
+    assert!(!backend.object_exists(&bucket, &metadata_objects[0]).await.unwrap());
+    assert!(
+        !object_store
+            .get_bucket_metadata(&bucket)
+            .await
+            .unwrap()
+            .table_bucket_enabled()
+    );
+    object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
+}

rustfs/src/storage_api.rs

@@ -490,6 +490,10 @@ pub(crate) mod table {
             pub(crate) use super::super::super::storage_contracts::NamespaceLocking;
         }
 
+        pub(crate) mod multipart {
+            pub(crate) use super::super::super::storage_contracts::MultipartUploadInspection;
+        }
+
         pub(crate) mod object {
             pub(crate) use super::super::super::storage_contracts::{ObjectIO, ObjectOperations};
         }

rustfs/src/table_catalog/cleanup_metadata.rs

@@ -0,0 +1,342 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+
+/// Validates one metadata object; callers must also validate the complete object inventory.
+pub(crate) fn validate_empty_table_cleanup_metadata(
+    table_bucket: &str,
+    object_key: &str,
+    metadata: &[u8],
+) -> TableCatalogStoreResult<()> {
+    let (namespace, table) = table_identity_from_metadata_object_key(object_key).ok_or_else(|| {
+        TableCatalogStoreError::Invalid("cleanup metadata must use the default table metadata directory".to_string())
+    })?;
+    let metadata = decode_table_metadata_json(object_key, metadata)?;
+    validate_supported_table_metadata(&metadata)?;
+
+    // Future metadata fields may introduce additional object references.
+    for field in metadata.as_object().expect("validated metadata is a JSON object").keys() {
+        if !matches!(
+            field.as_str(),
+            "format-version"
+                | "table-uuid"
+                | "location"
+                | "last-sequence-number"
+                | "last-updated-ms"
+                | "last-column-id"
+                | "schema"
+                | "schemas"
+                | "current-schema-id"
+                | "partition-spec"
+                | "partition-specs"
+                | "default-spec-id"
+                | "last-partition-id"
+                | "properties"
+                | "current-snapshot-id"
+                | "snapshots"
+                | "snapshot-log"
+                | "metadata-log"
+                | "sort-orders"
+                | "default-sort-order-id"
+                | "refs"
+                | "statistics"
+                | "partition-statistics"
+        ) {
+            return Err(TableCatalogStoreError::Unsupported(format!(
+                "empty table cleanup does not support metadata field {field}"
+            )));
+        }
+    }
+
+    for field in ["snapshots", "snapshot-log", "statistics", "partition-statistics"] {
+        let Some(value) = metadata.get(field) else {
+            continue;
+        };
+        let entries = value
+            .as_array()
+            .ok_or_else(|| TableCatalogStoreError::Invalid(format!("cleanup metadata field {field} must be an array")))?;
+        if !entries.is_empty() {
+            return Err(TableCatalogStoreError::Conflict(format!(
+                "empty table cleanup requires metadata field {field} to be empty"
+            )));
+        }
+    }
+    if let Some(value) = metadata.get("refs") {
+        let refs = value
+            .as_object()
+            .ok_or_else(|| TableCatalogStoreError::Invalid("cleanup metadata refs must be an object".to_string()))?;
+        if !refs.is_empty() {
+            return Err(TableCatalogStoreError::Conflict(
+                "empty table cleanup requires metadata refs to be empty".to_string(),
+            ));
+        }
+    }
+    let table_uuid = table_metadata_uuid(&metadata)?;
+    if Uuid::parse_str(table_uuid).ok().filter(|uuid| !uuid.is_nil()).is_none() {
+        return Err(TableCatalogStoreError::Invalid(
+            "cleanup metadata table-uuid must be a non-nil UUID".to_string(),
+        ));
+    }
+    let warehouse_prefix = table_warehouse_object_prefix_from_location(table_bucket, table_metadata_location(&metadata)?)?;
+    if warehouse_prefix
+        .strip_prefix("tables/")
+        .and_then(|prefix| prefix.strip_suffix('/'))
+        .and_then(|table_id| Uuid::parse_str(table_id).ok())
+        .filter(|table_id| !table_id.is_nil())
+        .is_none()
+    {
+        return Err(TableCatalogStoreError::Conflict(
+            "empty table cleanup requires the default tables/<table-id> warehouse location".to_string(),
+        ));
+    }
+    if let Some(entries) = metadata.get("metadata-log").and_then(serde_json::Value::as_array) {
+        for entry in entries {
+            let entry = entry.as_object().expect("validated metadata log entries are JSON objects");
+            if entry
+                .keys()
+                .any(|field| !matches!(field.as_str(), "timestamp-ms" | "metadata-file"))
+            {
+                return Err(TableCatalogStoreError::Unsupported(
+                    "empty table cleanup does not support unknown metadata-log fields".to_string(),
+                ));
+            }
+            let location = entry["metadata-file"]
+                .as_str()
+                .expect("validated metadata log entries contain metadata-file");
+            table_catalog_object_key_from_location(table_bucket, location)
+                .filter(|location| is_valid_table_metadata_location(&namespace, &table, location))
+                .filter(|location| location != object_key)
+                .ok_or_else(|| {
+                    TableCatalogStoreError::Invalid(
+                        "cleanup metadata-log must reference another file in the same table metadata directory".to_string(),
+                    )
+                })?;
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::table_catalog::test_support::table_metadata_json;
+    use serde_json::{Value, json};
+    use std::io::Write;
+
+    const BUCKET: &str = "warehouse";
+    const TABLE_UUID: &str = "f6dd46c2-4971-4d41-b188-452d5f5c2c01";
+    const WAREHOUSE_LOCATION: &str = "s3://warehouse/tables/a9d8f3b1-abde-4a0b-b6fb-e36a30abff16";
+    const OBJECT: &str =
+        ".rustfs-table/warehouses/default/namespaces/analytics/events/tables/orders/metadata/00002.metadata.json";
+    const PREVIOUS: &str =
+        ".rustfs-table/warehouses/default/namespaces/analytics/events/tables/orders/metadata/00001.metadata.json";
+
+    fn empty_metadata() -> Value {
+        table_metadata_json(TABLE_UUID, WAREHOUSE_LOCATION)
+    }
+
+    fn validate(metadata: &Value) -> TableCatalogStoreResult<()> {
+        validate_empty_table_cleanup_metadata(BUCKET, OBJECT, &serde_json::to_vec(metadata).unwrap())
+    }
+
+    #[test]
+    fn empty_table_cleanup_accepts_v2_and_same_directory_history() {
+        let mut metadata = empty_metadata();
+        metadata["metadata-log"] = json!([
+            {"timestamp-ms": 1, "metadata-file": PREVIOUS},
+            {"timestamp-ms": 1, "metadata-file": format!("s3://{BUCKET}/{PREVIOUS}")}
+        ]);
+        assert!(validate(&metadata).is_ok());
+    }
+
+    #[test]
+    fn empty_table_cleanup_accepts_v1_without_snapshots() {
+        let mut metadata = json!({
+            "format-version": 1,
+            "table-uuid": TABLE_UUID,
+            "location": WAREHOUSE_LOCATION,
+            "last-updated-ms": 1,
+            "last-column-id": 0,
+            "schema": {"type": "struct", "fields": []},
+            "partition-spec": [],
+            "current-snapshot-id": -1
+        });
+        assert!(validate(&metadata).is_ok());
+        metadata["snapshots"] = json!([{
+            "snapshot-id": 1,
+            "timestamp-ms": 1,
+            "manifests": ["s3://warehouse/tables/table-id/metadata/manifest.avro"]
+        }]);
+        validate_supported_table_metadata(&metadata).unwrap();
+        assert!(matches!(validate(&metadata), Err(TableCatalogStoreError::Conflict(_))));
+    }
+
+    #[test]
+    fn empty_table_cleanup_rejects_unsupported_versions_and_unknown_references() {
+        for version in [0, 3, 4] {
+            let mut metadata = empty_metadata();
+            metadata["format-version"] = json!(version);
+            assert!(validate(&metadata).is_err(), "version {version}");
+        }
+        for field in [
+            "manifest-list",
+            "manifests",
+            "data-files",
+            "delete-files",
+            "next-row-id",
+            "future-field",
+        ] {
+            let mut metadata = empty_metadata();
+            metadata[field] = json!([]);
+            assert!(validate(&metadata).is_err(), "field {field}");
+        }
+    }
+
+    #[test]
+    fn empty_table_cleanup_rejects_snapshot_and_history_references() {
+        let mut metadata = empty_metadata();
+        metadata["snapshots"] = json!([{
+            "snapshot-id": 1,
+            "sequence-number": 0,
+            "timestamp-ms": 1,
+            "manifest-list": "s3://warehouse/tables/table-id/metadata/list.avro",
+            "summary": {"operation": "append"}
+        }]);
+        validate_supported_table_metadata(&metadata).unwrap();
+        assert!(matches!(validate(&metadata), Err(TableCatalogStoreError::Conflict(_))));
+
+        let mut metadata = empty_metadata();
+        metadata["snapshot-log"] = json!([{"timestamp-ms": 1, "snapshot-id": 1}]);
+        assert!(validate(&metadata).is_err());
+        metadata = empty_metadata();
+        metadata["refs"] = json!({"main": {"snapshot-id": 1, "type": "branch"}});
+        assert!(validate(&metadata).is_err());
+        metadata = empty_metadata();
+        metadata["current-snapshot-id"] = json!(1);
+        assert!(validate(&metadata).is_err());
+    }
+
+    #[test]
+    fn empty_table_cleanup_accepts_only_absent_null_or_minus_one_current_snapshot() {
+        for version in [1, 2] {
+            let mut metadata = empty_metadata();
+            metadata["format-version"] = json!(version);
+            if version == 1 {
+                metadata["schema"] = metadata["schemas"][0].clone();
+                metadata["partition-spec"] = json!([]);
+            }
+            assert!(validate(&metadata).is_ok());
+            for value in [Value::Null, json!(-1)] {
+                metadata["current-snapshot-id"] = value;
+                assert!(validate(&metadata).is_ok(), "version {version}");
+            }
+            for value in [json!(1), json!(0), json!(-2), json!("1"), json!(false), json!([])] {
+                metadata["current-snapshot-id"] = value;
+                assert!(validate(&metadata).is_err(), "version {version}");
+            }
+        }
+    }
+
+    #[test]
+    fn empty_table_cleanup_rejects_statistics_references_and_malformed_empty_fields() {
+        for field in ["snapshots", "snapshot-log", "statistics", "partition-statistics"] {
+            for value in [Value::Null, json!({}), json!("empty"), json!([{}])] {
+                let mut metadata = empty_metadata();
+                metadata[field] = value;
+                assert!(validate(&metadata).is_err(), "field {field}");
+            }
+        }
+        for value in [Value::Null, json!([]), json!("empty")] {
+            let mut metadata = empty_metadata();
+            metadata["refs"] = value;
+            assert!(validate(&metadata).is_err());
+        }
+    }
+
+    #[test]
+    fn empty_table_cleanup_rejects_unknown_or_cross_table_history() {
+        for location in [
+            format!("s3://other/{PREVIOUS}"),
+            PREVIOUS.replace("/orders/", "/other/"),
+            PREVIOUS.replace("/analytics/events/", "/another/"),
+            PREVIOUS.replace("00001.metadata.json", "../00001.metadata.json"),
+            PREVIOUS.replace("00001.metadata.json", "%2e%2e.json"),
+            format!("{PREVIOUS}?versionId=previous"),
+            format!("/{PREVIOUS}"),
+            OBJECT.to_string(),
+        ] {
+            let mut metadata = empty_metadata();
+            metadata["metadata-log"] = json!([{"timestamp-ms": 1, "metadata-file": location}]);
+            assert!(validate(&metadata).is_err(), "location {location}");
+        }
+        let mut metadata = empty_metadata();
+        metadata["metadata-log"] = json!([{"timestamp-ms": 1, "metadata-file": PREVIOUS, "future-file": "data.parquet"}]);
+        assert!(validate(&metadata).is_err());
+    }
+
+    #[test]
+    fn empty_table_cleanup_requires_valid_schema_uuid_and_warehouse() {
+        for uuid in ["table-uuid", "00000000-0000-0000-0000-000000000000", ""] {
+            let mut metadata = empty_metadata();
+            metadata["table-uuid"] = json!(uuid);
+            assert!(validate(&metadata).is_err());
+        }
+        for location in [
+            "s3://other/tables/table-id",
+            "s3://warehouse/",
+            "s3://warehouse/.rustfs-table/metadata",
+            "file:///table",
+            "s3://warehouse/custom/table-id",
+            "s3://warehouse/tables/table-id",
+            "s3://warehouse/tables/00000000-0000-0000-0000-000000000000",
+            "s3://warehouse/tables/a9d8f3b1-abde-4a0b-b6fb-e36a30abff16/nested",
+        ] {
+            let mut metadata = empty_metadata();
+            metadata["location"] = json!(location);
+            assert!(validate(&metadata).is_err());
+        }
+        let mut metadata = empty_metadata();
+        metadata["schemas"] = json!([]);
+        assert!(validate(&metadata).is_err());
+    }
+
+    #[test]
+    fn empty_table_cleanup_rejects_non_default_metadata_paths() {
+        let bytes = serde_json::to_vec(&empty_metadata()).unwrap();
+        for object in [
+            OBJECT.replace("/default/", "/custom/"),
+            OBJECT.replace("/tables/", "/views/"),
+            OBJECT.replace("/metadata/", "/data/"),
+            OBJECT.replace("00002.metadata.json", "part.parquet"),
+            OBJECT.replace("00002.metadata.json", "nested/00002.metadata.json"),
+            format!("s3://{BUCKET}/{OBJECT}"),
+        ] {
+            assert!(validate_empty_table_cleanup_metadata(BUCKET, &object, &bytes).is_err(), "object {object}");
+        }
+    }
+
+    #[test]
+    fn empty_table_cleanup_decodes_gzip_and_rejects_invalid_json() {
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        encoder.write_all(&serde_json::to_vec(&empty_metadata()).unwrap()).unwrap();
+        let bytes = encoder.finish().unwrap();
+        let object = OBJECT.replace(".metadata.json", ".gz.metadata.json");
+        assert!(validate_empty_table_cleanup_metadata(BUCKET, &object, &bytes).is_ok());
+        assert!(validate_empty_table_cleanup_metadata(BUCKET, OBJECT, b"{}").is_err());
+        assert!(validate_empty_table_cleanup_metadata(BUCKET, OBJECT, b"[").is_err());
+        assert!(validate_empty_table_cleanup_metadata(BUCKET, &object, b"invalid gzip").is_err());
+    }
+}

rustfs/src/table_catalog/mod.rs

@@ -31,6 +31,7 @@ use crate::storage_api::table::contract::list::{
     ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info,
     ListOperations as StorageListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, WalkOptions as StorageWalkOptions,
 };
+use crate::storage_api::table::contract::multipart::MultipartUploadInspection as StorageMultipartUploadInspection;
 use crate::storage_api::table::contract::namespace::NamespaceLocking as StorageNamespaceLocking;
 use crate::storage_api::table::contract::object::{ObjectIO as StorageObjectIO, ObjectOperations as StorageObjectOperations};
 use crate::storage_api::table::contract::range::HTTPRangeSpec;
@@ -58,6 +59,7 @@ use crate::storage_api::table::{
     StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader,
 };
 
+mod cleanup_metadata;
 mod error;
 mod iceberg;
 mod identifier;
@@ -247,6 +249,7 @@ pub(crate) trait TableCatalogStorage:
         WalkCancellation = tokio_util::sync::CancellationToken,
         WalkResultSender = tokio::sync::mpsc::Sender<CatalogObjectInfoOrErr>,
     > + StorageNamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper>
+    + StorageMultipartUploadInspection<Error = EcstoreError>
 {
 }
 
@@ -275,6 +278,7 @@ impl<T> TableCatalogStorage for T where
             WalkCancellation = tokio_util::sync::CancellationToken,
             WalkResultSender = tokio::sync::mpsc::Sender<CatalogObjectInfoOrErr>,
         > + StorageNamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper>
+        + StorageMultipartUploadInspection<Error = EcstoreError>
 {
 }
 

rustfs/src/table_catalog/store/cleanup.rs

@@ -0,0 +1,267 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::object::{
+    next_table_catalog_update_time, table_warehouse_index_state_ready, validate_table_warehouse_index_entry_object,
+};
+use super::*;
+
+pub(super) struct TableBucketDisablement {
+    pub(super) entry: TableBucketEntry,
+    guards: Vec<TableCatalogLockGuard>,
+}
+
+/// An in-memory, fenced cleanup plan. No mutation occurs until every object has
+/// been validated and the caller has authorized all metadata deletions.
+pub(crate) struct EmptyTableBucketCleanup {
+    disablement: TableBucketDisablement,
+    metadata: Vec<(String, String)>,
+    indexes: Vec<(String, String)>,
+    fence: TableCatalogObjectMutationFence,
+}
+
+impl EmptyTableBucketCleanup {
+    pub(crate) fn metadata_objects(&self) -> impl Iterator<Item = &str> {
+        self.metadata.iter().map(|(key, _)| key.as_str())
+    }
+
+    pub(crate) fn ensure_held(&self) -> TableCatalogStoreResult<()> {
+        self.fence.ensure_held()
+    }
+
+    pub(crate) fn mutation_fence(&self) -> TableCatalogObjectMutationFence {
+        self.fence.clone()
+    }
+}
+
+fn cleanup_conflict(message: impl Into<String>) -> TableCatalogStoreError {
+    TableCatalogStoreError::Conflict(message.into())
+}
+
+impl<B: TableCatalogObjectBackend> ObjectTableCatalogStore<B> {
+    pub(super) async fn begin_table_bucket_disablement(
+        &self,
+        mut entry: TableBucketEntry,
+        publication: Option<&dyn TableCommitPublication>,
+    ) -> TableCatalogStoreResult<TableBucketDisablement> {
+        validate_table_bucket_entry(&entry)?;
+        let mut guards = Vec::new();
+        if publication.is_some_and(|publication| publication.catalog_migration_read_permit_status() == Some(true)) {
+            if self
+                .backend
+                .read_object(self.catalog_bucket(), &self.paths.backing_migration_global_fence_path())
+                .await?
+                .is_some()
+            {
+                return Err(cleanup_conflict("table bucket registry migration is in progress"));
+            }
+            self.ensure_object_backed_catalog_write_permit_after_lock(&entry.table_bucket)
+                .await?;
+        } else {
+            guards.push(self.acquire_table_bucket_registry_write_permit().await?);
+            guards.push(self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?);
+        }
+        let object = self.paths.table_bucket_entry_path(&entry.table_bucket);
+        guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?);
+        if let Some((current, _)) = self.read_table_bucket_with_etag_unlocked(&entry.table_bucket).await? {
+            if current.active_rename_id.is_some()
+                || !matches!(current.state, TableCatalogEntryState::Active | TableCatalogEntryState::Deleted)
+            {
+                return Err(cleanup_conflict("table bucket has an operation in progress"));
+            }
+            entry = current;
+        }
+        Ok(TableBucketDisablement { entry, guards })
+    }
+
+    pub(super) async fn finish_table_bucket_disablement(
+        &self,
+        disablement: &mut TableBucketDisablement,
+    ) -> TableCatalogStoreResult<()> {
+        let mut entry = disablement.entry.clone();
+        entry.state = TableCatalogEntryState::Deleted;
+        entry.updated_at = Some(next_table_catalog_update_time(entry.updated_at.as_deref()));
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            &self.paths.table_bucket_entry_path(&entry.table_bucket),
+            &entry,
+            TableCatalogPutPrecondition::Any,
+            disablement
+                .guards
+                .iter()
+                .flat_map(TableCatalogLockGuard::write_commit_guards)
+                .collect(),
+        )
+        .await
+    }
+
+    async fn prepare_empty_table_bucket_cleanup(
+        &self,
+        entry: TableBucketEntry,
+        publication: &impl TableCatalogObjectBackend,
+    ) -> TableCatalogStoreResult<EmptyTableBucketCleanup> {
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(cleanup_conflict("empty table bucket cleanup requires a publication fence"));
+        }
+        let disablement = self.begin_table_bucket_disablement(entry, Some(publication)).await?;
+        let bucket = &disablement.entry.table_bucket;
+        let mut fence = publication
+            .object_mutation_fence()
+            .ok_or_else(|| cleanup_conflict("cleanup requires storage fencing"))?;
+        fence
+            .signals
+            .extend(disablement.guards.iter().filter_map(TableCatalogLockGuard::lock_lost_signal));
+        fence.ensure_held()?;
+
+        let catalog_objects = self
+            .backend
+            .list_objects_for_cleanup(self.catalog_bucket(), &self.paths.table_bucket_root_prefix(bucket))
+            .await?;
+        let bucket_path = self.paths.table_bucket_entry_path(bucket);
+        let index_state_path = self.paths.warehouse_index_state_path(bucket);
+        let index_prefix = self.paths.warehouse_index_entries_prefix(bucket);
+        let mut indexes = Vec::new();
+        for object in catalog_objects {
+            if object.delete_marker
+                || object
+                    .version_id
+                    .as_deref()
+                    .is_some_and(|version| !version.is_empty() && version != "null")
+            {
+                return Err(cleanup_conflict("catalog contains versioned or deleted records"));
+            }
+            if object.key == bucket_path {
+                continue;
+            }
+            let etag = if object.key == index_state_path {
+                let (state, etag) = self
+                    .read_entry::<TableWarehouseIndexStateEntry>(self.catalog_bucket(), &object.key)
+                    .await?
+                    .ok_or_else(|| cleanup_conflict("warehouse index state changed during cleanup"))?;
+                table_warehouse_index_state_ready(&state, bucket)?;
+                if !matches!(state.state, TableCatalogEntryState::Active | TableCatalogEntryState::Deleted) {
+                    return Err(cleanup_conflict("warehouse index has an operation in progress"));
+                }
+                etag
+            } else if object.key.starts_with(&index_prefix) {
+                let (index, etag) = self
+                    .read_entry::<TableWarehouseIndexEntry>(self.catalog_bucket(), &object.key)
+                    .await?
+                    .ok_or_else(|| cleanup_conflict("warehouse index changed during cleanup"))?;
+                validate_table_warehouse_index_entry_object(&self.paths, &object.key, &index)?;
+                if index.table_bucket != *bucket || index.state != TableCatalogEntryState::Deleted {
+                    return Err(cleanup_conflict("table bucket contains an active warehouse index"));
+                }
+                Namespace::parse(&index.namespace).map_err(|err| cleanup_conflict(err.to_string()))?;
+                IdentifierSegment::parse(index.table.clone()).map_err(|err| cleanup_conflict(err.to_string()))?;
+                if Uuid::parse_str(&index.table_id).ok().filter(|id| !id.is_nil()).is_none() {
+                    return Err(cleanup_conflict("warehouse index table identity is invalid"));
+                }
+                let expected = format!("tables/{}/", index.table_id);
+                if index.warehouse_object_prefix != expected {
+                    return Err(cleanup_conflict("cleanup requires the default table warehouse location"));
+                }
+                etag
+            } else {
+                return Err(cleanup_conflict("table bucket contains namespaces or retained catalog resources"));
+            };
+            let etag = etag.ok_or_else(|| cleanup_conflict("catalog cleanup object has no etag"))?;
+            indexes.push((object.key, etag));
+        }
+        self.backend.ensure_no_multipart_uploads(bucket).await?;
+        let mut metadata = Vec::new();
+        for object in self.backend.list_objects_for_cleanup(bucket, "").await? {
+            if object.delete_marker
+                || object
+                    .version_id
+                    .as_deref()
+                    .is_some_and(|version| !version.is_empty() && version != "null")
+            {
+                return Err(cleanup_conflict("table bucket contains object versions or delete markers"));
+            }
+            let value = publication
+                .read_object_limited(bucket, &object.key, TABLE_METADATA_JSON_MAX_SIZE)
+                .await?
+                .ok_or_else(|| cleanup_conflict("metadata object changed during cleanup"))?;
+            crate::table_catalog::cleanup_metadata::validate_empty_table_cleanup_metadata(bucket, &object.key, &value.data)?;
+            let etag = value
+                .etag
+                .ok_or_else(|| cleanup_conflict("metadata cleanup object has no etag"))?;
+            metadata.push((object.key, etag));
+        }
+        fence.ensure_held()?;
+        Ok(EmptyTableBucketCleanup {
+            disablement,
+            metadata,
+            indexes,
+            fence,
+        })
+    }
+
+    async fn execute_empty_table_bucket_cleanup(
+        &self,
+        cleanup: &mut EmptyTableBucketCleanup,
+        incarnation: Uuid,
+    ) -> TableCatalogStoreResult<()> {
+        let bucket = &cleanup.disablement.entry.table_bucket;
+        for (key, etag) in &cleanup.metadata {
+            cleanup.ensure_held()?;
+            self.backend
+                .delete_object_if_match_fenced_for_incarnation(bucket, key, etag, incarnation, &cleanup.fence)
+                .await?;
+        }
+        // A retry inventories the remaining metadata. Keep warehouse protection
+        // until all user objects and uploads have been checked again.
+        if !self.backend.list_objects_for_cleanup(bucket, "").await?.is_empty() {
+            return Err(cleanup_conflict("table bucket changed during cleanup"));
+        }
+        self.backend.ensure_no_multipart_uploads(bucket).await?;
+        for (key, etag) in &cleanup.indexes {
+            cleanup.ensure_held()?;
+            self.backend
+                .delete_object_if_match_fenced(self.catalog_bucket(), key, etag, &cleanup.fence)
+                .await?;
+        }
+        cleanup.ensure_held()?;
+        self.finish_table_bucket_disablement(&mut cleanup.disablement).await
+    }
+}
+
+impl<B: TableCatalogObjectBackend> ConfiguredTableCatalogStore<B> {
+    pub(crate) async fn prepare_empty_table_bucket_cleanup(
+        &self,
+        entry: TableBucketEntry,
+        publication: &impl TableCatalogObjectBackend,
+    ) -> TableCatalogStoreResult<EmptyTableBucketCleanup> {
+        match self {
+            Self::ObjectBacked(store) => store.prepare_empty_table_bucket_cleanup(entry, publication).await,
+            Self::DurableStrong(_) => Err(TableCatalogStoreError::Unsupported(
+                "empty table bucket cleanup requires object-backed catalog storage".to_string(),
+            )),
+        }
+    }
+
+    pub(crate) async fn execute_empty_table_bucket_cleanup(
+        &self,
+        cleanup: &mut EmptyTableBucketCleanup,
+        incarnation: Uuid,
+    ) -> TableCatalogStoreResult<()> {
+        match self {
+            Self::ObjectBacked(store) => store.execute_empty_table_bucket_cleanup(cleanup, incarnation).await,
+            Self::DurableStrong(_) => Err(TableCatalogStoreError::Unsupported(
+                "empty table bucket cleanup requires object-backed catalog storage".to_string(),
+            )),
+        }
+    }
+}

rustfs/src/table_catalog/store/mod.rs

@@ -17,6 +17,7 @@ use super::*;
 use crate::storage_api::table::{WriteCommitGuard, WriteCompletion};
 
 mod backup;
+mod cleanup;
 mod migration;
 mod object;
 mod strong;
@@ -775,6 +776,14 @@ pub(crate) struct TableCatalogObjectListPage {
     pub is_truncated: bool,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct TableCatalogCleanupObject {
+    pub key: String,
+    pub version_id: Option<String>,
+    pub etag: Option<String>,
+    pub delete_marker: bool,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum TableCatalogPutPrecondition {
     Any,
@@ -934,12 +943,41 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
         fence.ensure_held()
     }
 
+    async fn delete_object_if_match_fenced_for_incarnation(
+        &self,
+        _bucket: &str,
+        _object: &str,
+        _expected_etag: &str,
+        _expected_incarnation_id: Uuid,
+        _fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        Err(TableCatalogStoreError::Unsupported(
+            "incarnation-checked cleanup is not supported by this backend".to_string(),
+        ))
+    }
+
     async fn delete_object_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
         self.delete_object(bucket, object).await
     }
 
     async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>>;
 
+    async fn list_objects_for_cleanup(
+        &self,
+        _bucket: &str,
+        _prefix: &str,
+    ) -> TableCatalogStoreResult<Vec<TableCatalogCleanupObject>> {
+        Err(TableCatalogStoreError::Unsupported(
+            "authoritative cleanup listing is not supported by this backend".to_string(),
+        ))
+    }
+
+    async fn ensure_no_multipart_uploads(&self, _bucket: &str) -> TableCatalogStoreResult<()> {
+        Err(TableCatalogStoreError::Unsupported(
+            "multipart cleanup validation is not supported by this backend".to_string(),
+        ))
+    }
+
     async fn list_objects_page(
         &self,
         bucket: &str,
@@ -2412,7 +2450,7 @@ where
     }
 
     async fn delete_object_if_match(&self, bucket: &str, object: &str, expected_etag: &str) -> TableCatalogStoreResult<()> {
-        self.delete_object_if_match_with_fence(bucket, object, expected_etag, None)
+        self.delete_object_if_match_with_fence(bucket, object, expected_etag, None, None)
             .await
     }
 
@@ -2424,7 +2462,25 @@ where
         fence: &TableCatalogObjectMutationFence,
     ) -> TableCatalogStoreResult<()> {
         fence.ensure_held()?;
-        self.delete_object_if_match_with_fence(bucket, object, expected_etag, Some(fence))
+        self.delete_object_if_match_with_fence(bucket, object, expected_etag, Some(fence), None)
+            .await
+    }
+
+    async fn delete_object_if_match_fenced_for_incarnation(
+        &self,
+        bucket: &str,
+        object: &str,
+        expected_etag: &str,
+        expected_incarnation_id: Uuid,
+        fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        if expected_incarnation_id.is_nil() {
+            return Err(TableCatalogStoreError::Invalid(
+                "cleanup requires a non-nil bucket incarnation".to_string(),
+            ));
+        }
+        fence.ensure_held()?;
+        self.delete_object_if_match_with_fence(bucket, object, expected_etag, Some(fence), Some(expected_incarnation_id))
             .await
     }
 
@@ -2498,6 +2554,58 @@ where
         })
     }
 
+    async fn list_objects_for_cleanup(
+        &self,
+        bucket: &str,
+        prefix: &str,
+    ) -> TableCatalogStoreResult<Vec<TableCatalogCleanupObject>> {
+        let cancel = tokio_util::sync::CancellationToken::new();
+        let _cancel_on_drop = cancel.clone().drop_guard();
+        let (sender, mut receiver) = tokio::sync::mpsc::channel::<CatalogObjectInfoOrErr>(100);
+        let walk = self.store.clone().walk(
+            cancel,
+            bucket,
+            prefix,
+            sender,
+            CatalogWalkOptions {
+                authoritative: true,
+                include_free_versions: true,
+                ..Default::default()
+            },
+        );
+        let collect = async move {
+            let mut objects = Vec::new();
+            while let Some(result) = receiver.recv().await {
+                if let Some(err) = result.err {
+                    return Err(storage_error_to_catalog("inspect cleanup object versions", err));
+                }
+                let object = result.item.ok_or_else(|| {
+                    TableCatalogStoreError::Internal("cleanup object walk returned an empty result".to_string())
+                })?;
+                objects.push(TableCatalogCleanupObject {
+                    key: object.name,
+                    version_id: object.version_id.filter(|id| !id.is_nil()).map(|id| id.to_string()),
+                    etag: object.etag,
+                    delete_marker: object.delete_marker,
+                });
+            }
+            Ok(objects)
+        };
+        let (walk_result, objects) = tokio::join!(walk, collect);
+        let objects = objects?;
+        walk_result.map_err(|err| storage_error_to_catalog("list cleanup object versions", err))?;
+        Ok(objects)
+    }
+
+    async fn ensure_no_multipart_uploads(&self, bucket: &str) -> TableCatalogStoreResult<()> {
+        self.store.ensure_no_multipart_uploads(bucket).await.map_err(|err| match err {
+            StorageError::BucketNotEmpty(_) => {
+                TableCatalogStoreError::Conflict(format!("table bucket {bucket} contains unfinished multipart uploads"))
+            }
+            other => storage_error_to_catalog("inspect unfinished multipart uploads", other),
+        })
+    }
+
     async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let lock = self
             .store
@@ -2534,12 +2642,14 @@ where
         object: &str,
         expected_etag: &str,
         fence: Option<&TableCatalogObjectMutationFence>,
+        expected_incarnation_id: Option<Uuid>,
     ) -> TableCatalogStoreResult<()> {
         let mut options = ObjectOptions {
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(expected_etag.to_string()),
                 ..Default::default()
             }),
+            expected_bucket_incarnation_id: expected_incarnation_id,
             ..Default::default()
         };
         if let Some(fence) = fence {

rustfs/src/table_catalog/store/object.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use super::*;
+use futures::{StreamExt, TryStreamExt, stream};
 
+const TABLE_REGISTRATION_READ_CONCURRENCY: usize = 8;
 const TABLE_MAINTENANCE_RENAME_MAX_OBJECTS: usize = 4096;
 const TABLE_MAINTENANCE_SCHEDULER_HISTORY_MAX_OBJECTS: usize = 4096;
 
@@ -261,7 +263,7 @@ fn validate_table_rename_intent_object(
     Ok(())
 }
 
-fn next_table_catalog_update_time(previous: Option<&str>) -> String {
+pub(super) fn next_table_catalog_update_time(previous: Option<&str>) -> String {
     let now = OffsetDateTime::now_utc();
     let update_time = previous
         .and_then(|value| OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).ok())
@@ -360,7 +362,7 @@ pub(super) fn validate_table_maintenance_report_owner(
     Ok(())
 }
 
-fn validate_table_warehouse_index_entry_object(
+pub(super) fn validate_table_warehouse_index_entry_object(
     paths: &TableCatalogObjectPaths,
     object: &str,
     index: &TableWarehouseIndexEntry,
@@ -374,7 +376,10 @@ fn validate_table_warehouse_index_entry_object(
     Ok(())
 }
 
-fn table_warehouse_index_state_ready(state: &TableWarehouseIndexStateEntry, table_bucket: &str) -> TableCatalogStoreResult<bool> {
+pub(super) fn table_warehouse_index_state_ready(
+    state: &TableWarehouseIndexStateEntry,
+    table_bucket: &str,
+) -> TableCatalogStoreResult<bool> {
     if state.version == 0 || state.version > TABLE_WAREHOUSE_INDEX_STATE_VERSION {
         return Err(TableCatalogStoreError::Invalid(format!(
             "unsupported warehouse index state version: {}",
@@ -1217,7 +1222,7 @@ where
             .await
     }
 
-    async fn write_entry_unlocked<T>(
+    pub(super) async fn write_entry_unlocked<T>(
         &self,
         bucket: &str,
         object: &str,
@@ -1264,7 +1269,7 @@ where
         }
     }
 
-    async fn read_table_bucket_with_etag_unlocked(
+    pub(super) async fn read_table_bucket_with_etag_unlocked(
         &self,
         table_bucket: &str,
     ) -> TableCatalogStoreResult<Option<(TableBucketEntry, String)>> {
@@ -1873,7 +1878,12 @@ where
         let candidate = table_warehouse_index_entry(entry)?;
         validate_table_entry_version_and_id(entry)?;
         for existing in self
-            .list_all_table_entries_with_limit(&candidate.table_bucket, None, locked_table_entry)
+            .list_all_table_entries_with_limit(
+                &candidate.table_bucket,
+                None,
+                locked_table_entry,
+                TABLE_REGISTRATION_READ_CONCURRENCY,
+            )
             .await?
         {
             if existing.state != TableCatalogEntryState::Active {
@@ -1896,18 +1906,20 @@ where
             }
         }
         let state_object = self.paths.warehouse_index_state_path(&candidate.table_bucket);
-        for object in self
+        let index_objects = 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 {
+            .await?;
+        let mut indexes = stream::iter(index_objects.into_iter().filter(|object| object != &state_object))
+            .map(|object| async move {
+                let entry = self
+                    .read_entry::<TableWarehouseIndexEntry>(self.catalog_bucket(), &object)
+                    .await?;
+                Ok::<_, TableCatalogStoreError>((object, entry))
+            })
+            .buffered(TABLE_REGISTRATION_READ_CONCURRENCY);
+        while let Some((object, entry)) = indexes.try_next().await? {
+            let Some((existing, _)) = entry else {
                 continue;
             };
             validate_table_warehouse_index_entry_object(&self.paths, &object, &existing)?;
@@ -2364,7 +2376,7 @@ where
     ) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
         let mut matched: Option<TableDataPlaneResource> = None;
         for table in self
-            .list_all_table_entries_with_limit(table_bucket, Some(TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_CATALOG_OBJECTS), None)
+            .list_all_table_entries_with_limit(table_bucket, Some(TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_CATALOG_OBJECTS), None, 1)
             .await?
         {
             if table.state != TableCatalogEntryState::Active {
@@ -2456,7 +2468,7 @@ where
             return Ok(());
         }
         let tables = self
-            .list_all_table_entries_with_limit(table_bucket, max_catalog_objects, None)
+            .list_all_table_entries_with_limit(table_bucket, max_catalog_objects, None, 1)
             .await?
             .into_iter()
             .filter(|table| table.state == TableCatalogEntryState::Active)
@@ -2537,7 +2549,7 @@ where
     }
 
     async fn list_all_table_entries(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>> {
-        self.list_all_table_entries_with_limit(table_bucket, None, None).await
+        self.list_all_table_entries_with_limit(table_bucket, None, None, 1).await
     }
 
     async fn list_table_entry_objects_for_data_plane_scan(
@@ -2586,8 +2598,8 @@ where
         table_bucket: &str,
         max_catalog_objects: Option<usize>,
         locked_table_entry: Option<&str>,
+        read_concurrency: usize,
     ) -> TableCatalogStoreResult<Vec<TableEntry>> {
-        let mut entries = Vec::new();
         let table_objects = match max_catalog_objects {
             Some(max_catalog_objects) => {
                 self.list_table_entry_objects_for_data_plane_scan(table_bucket, max_catalog_objects)
@@ -2601,19 +2613,24 @@ where
                 .filter(|object| object.ends_with(TABLE_ENTRY_FILE))
                 .collect(),
         };
-        for object in table_objects {
-            // Only the exact entry protected by the caller's write lock can bypass a read lock.
-            let entry = if locked_table_entry == Some(object.as_str()) {
-                self.read_entry_unlocked::<TableEntry>(self.catalog_bucket(), &object).await?
-            } else {
-                self.read_entry::<TableEntry>(self.catalog_bucket(), &object).await?
-            };
-            let Some((entry, _)) = entry else {
-                continue;
-            };
-            validate_table_entry_object(&self.paths, &object, &entry)?;
-            entries.push(entry);
-        }
+        let mut entries = stream::iter(table_objects)
+            .map(|object| async move {
+                // Only the exact entry protected by the caller's write lock can bypass a read lock.
+                let entry = if locked_table_entry == Some(object.as_str()) {
+                    self.read_entry_unlocked::<TableEntry>(self.catalog_bucket(), &object).await?
+                } else {
+                    self.read_entry::<TableEntry>(self.catalog_bucket(), &object).await?
+                };
+                let Some((entry, _)) = entry else {
+                    return Ok(None);
+                };
+                validate_table_entry_object(&self.paths, &object, &entry)?;
+                Ok::<_, TableCatalogStoreError>(Some(entry))
+            })
+            .buffered(read_concurrency)
+            .try_filter_map(|entry| async move { Ok(entry) })
+            .try_collect::<Vec<_>>()
+            .await?;
         entries.sort_by(|left, right| (&left.namespace, &left.table).cmp(&(&right.namespace, &right.table)));
         Ok(entries)
     }
@@ -6258,27 +6275,15 @@ where
         .await
     }
 
-    async fn disable_empty_table_bucket(&self, mut entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
-        validate_table_bucket_entry(&entry)?;
-        let _registry_guard = self.acquire_table_bucket_registry_write_permit().await?;
-        let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?;
-        let object = self.paths.table_bucket_entry_path(&entry.table_bucket);
-        let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?;
-        if let Some((current, _)) = self.read_table_bucket_with_etag_unlocked(&entry.table_bucket).await? {
-            if current.active_rename_id.is_some()
-                || !matches!(current.state, TableCatalogEntryState::Active | TableCatalogEntryState::Deleted)
-            {
-                return Err(TableCatalogStoreError::Conflict("table bucket has an operation in progress".to_string()));
-            }
-            entry = current;
-        }
+    async fn disable_empty_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
+        let mut disablement = self.begin_table_bucket_disablement(entry, None).await?;
         // Namespace creation holds this same bucket lock. Retained table, view, and
         // maintenance records also prevent removing their data-plane protections.
         let page = self
             .backend
             .list_objects_page(
                 self.catalog_bucket(),
-                &self.paths.namespace_entries_prefix(&entry.table_bucket),
+                &self.paths.namespace_entries_prefix(&disablement.entry.table_bucket),
                 None,
                 NonZeroUsize::MIN,
             )
@@ -6288,16 +6293,7 @@ where
                 "table bucket contains namespaces or retained catalog resources".to_string(),
             ));
         }
-        entry.state = TableCatalogEntryState::Deleted;
-        entry.updated_at = Some(next_table_catalog_update_time(entry.updated_at.as_deref()));
-        self.write_entry_unlocked(
-            self.catalog_bucket(),
-            &object,
-            &entry,
-            TableCatalogPutPrecondition::Any,
-            _guard.write_commit_guards(),
-        )
-        .await
+        self.finish_table_bucket_disablement(&mut disablement).await
     }
 
     async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {

rustfs/src/table_catalog/test_support.rs

@@ -1142,6 +1142,17 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
             .await
     }
 
+    async fn delete_object_if_match_fenced_for_incarnation(
+        &self,
+        bucket: &str,
+        object: &str,
+        expected_etag: &str,
+        _expected_incarnation_id: Uuid,
+        fence: &TableCatalogObjectMutationFence,
+    ) -> TableCatalogStoreResult<()> {
+        self.delete_object_if_match_fenced(bucket, object, expected_etag, fence).await
+    }
+
     async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
         let mut state = self.state.lock().await;
         state.list_calls += 1;
@@ -1153,6 +1164,30 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
             .collect())
     }
 
+    async fn list_objects_for_cleanup(
+        &self,
+        bucket: &str,
+        prefix: &str,
+    ) -> TableCatalogStoreResult<Vec<TableCatalogCleanupObject>> {
+        let mut state = self.state.lock().await;
+        state.list_calls += 1;
+        Ok(state
+            .objects
+            .iter()
+            .filter(|((entry_bucket, object), _)| entry_bucket == bucket && object.starts_with(prefix))
+            .map(|(key, record)| TableCatalogCleanupObject {
+                key: key.1.clone(),
+                version_id: None,
+                etag: (!state.etagless_objects.contains(key)).then(|| record.etag.clone()),
+                delete_marker: false,
+            })
+            .collect())
+    }
+
+    async fn ensure_no_multipart_uploads(&self, _bucket: &str) -> TableCatalogStoreResult<()> {
+        Ok(())
+    }
+
     async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         self.lock_attempts.lock().await.push((bucket.to_string(), object.to_string()));
         {

rustfs/src/table_catalog/tests.rs

@@ -7177,6 +7177,212 @@ async fn object_catalog_registration_scans_warehouse_prefixes_once() {
     assert_eq!(store.list_tables(bucket, &namespace.public_name()).await.unwrap().len(), 3);
 }
 
+#[tokio::test]
+async fn object_catalog_registration_bounds_concurrent_catalog_reads() {
+    const MAX_CONCURRENT_READS: usize = 8;
+
+    let backend = TestCatalogObjectBackend {
+        reject_reads_while_write_locked: true,
+        ..Default::default()
+    };
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .unwrap();
+
+    let mut table_objects = Vec::new();
+    let mut index_objects = Vec::new();
+    for number in 0..MAX_CONCURRENT_READS + 2 {
+        let name = format!("orders_{number:02}");
+        let table = IdentifierSegment::parse(&name).unwrap();
+        let mut entry = test_table_entry(
+            bucket,
+            &namespace,
+            &table,
+            default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+        );
+        entry.table_id = format!("table-{name}");
+        entry.warehouse_location = format!("s3://{bucket}/tables/{name}");
+        store.create_table(entry).await.unwrap();
+        table_objects.push(store.paths.table_entry_path(bucket, &namespace, &table));
+        index_objects.push(store.paths.warehouse_index_entry_path(bucket, &format!("tables/{name}/")));
+    }
+
+    let mut scans = Vec::new();
+    for mut objects in [table_objects, index_objects] {
+        objects.sort();
+        let mut scan = Vec::new();
+        for object in objects {
+            let previous_reads = backend
+                .state
+                .lock()
+                .await
+                .read_attempts
+                .get(&(RUSTFS_META_BUCKET.to_string(), object.clone()))
+                .copied()
+                .unwrap_or_default();
+            let pause = backend.pause_before_next_read(RUSTFS_META_BUCKET, &object).await;
+            scan.push((object, pause, previous_reads));
+        }
+        scans.push(scan);
+    }
+
+    let table = IdentifierSegment::parse("new_orders").unwrap();
+    let mut entry = test_table_entry(
+        bucket,
+        &namespace,
+        &table,
+        default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+    );
+    entry.table_id = "new-table-id".to_string();
+    entry.warehouse_location = format!("s3://{bucket}/tables/new_orders");
+    let registering_store = store.clone();
+    let registration = tokio::spawn(async move { registering_store.create_table(entry).await });
+
+    for scan in scans {
+        for (_, pause, _) in scan.iter().take(MAX_CONCURRENT_READS) {
+            tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, pause.wait_started())
+                .await
+                .expect("independent catalog reads should start concurrently");
+        }
+        let state = backend.state.lock().await;
+        let started = scan
+            .iter()
+            .filter(|(object, _, previous_reads)| {
+                state
+                    .read_attempts
+                    .get(&(RUSTFS_META_BUCKET.to_string(), object.clone()))
+                    .copied()
+                    .unwrap_or_default()
+                    > *previous_reads
+            })
+            .count();
+        assert_eq!(started, MAX_CONCURRENT_READS, "catalog reads must respect the concurrency limit");
+        drop(state);
+        for (_, pause, _) in scan {
+            pause.release();
+        }
+    }
+
+    tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, registration)
+        .await
+        .expect("registration should finish after its reads are released")
+        .unwrap()
+        .unwrap();
+    assert!(
+        store
+            .load_table(bucket, &namespace.public_name(), table.as_str())
+            .await
+            .unwrap()
+            .is_some()
+    );
+}
+
+#[tokio::test]
+async fn object_catalog_registration_propagates_catalog_scan_read_errors() {
+    for fail_index_read in [false, true] {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let namespace = Namespace::parse("sales").unwrap();
+        let existing = IdentifierSegment::parse("orders").unwrap();
+        seed_table_for_metadata_maintenance(
+            &store,
+            bucket,
+            &namespace,
+            &existing,
+            default_table_metadata_file_path(&namespace, &existing, "00001.metadata.json"),
+        )
+        .await;
+        let failed_object = if fail_index_read {
+            store.paths.warehouse_index_entry_path(bucket, "tables/table-id/")
+        } else {
+            store.paths.table_entry_path(bucket, &namespace, &existing)
+        };
+        backend.fail_next_read(RUSTFS_META_BUCKET, &failed_object).await;
+
+        let table = IdentifierSegment::parse("returns").unwrap();
+        let mut entry = test_table_entry(
+            bucket,
+            &namespace,
+            &table,
+            default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+        );
+        entry.table_id = "new-table-id".to_string();
+        entry.warehouse_location = format!("s3://{bucket}/tables/new-table-id");
+        assert_matches!(
+            store.create_table(entry.clone()).await,
+            Err(TableCatalogStoreError::Internal(message)) if message.contains(&failed_object)
+        );
+        assert!(
+            store
+                .load_table(bucket, &namespace.public_name(), table.as_str())
+                .await
+                .unwrap()
+                .is_none()
+        );
+        assert!(
+            backend
+                .read_object(
+                    RUSTFS_META_BUCKET,
+                    &store.paths.warehouse_index_entry_path(bucket, "tables/new-table-id/")
+                )
+                .await
+                .unwrap()
+                .is_none()
+        );
+        store
+            .create_table(entry)
+            .await
+            .expect("a failed scan must release its read locks for retry");
+    }
+}
+
+#[tokio::test]
+async fn object_catalog_registration_reuses_locked_deleted_entry() {
+    let backend = TestCatalogObjectBackend {
+        reject_reads_while_write_locked: true,
+        ..Default::default()
+    };
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .unwrap();
+    let mut entry = test_table_entry(
+        bucket,
+        &namespace,
+        &table,
+        default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+    );
+    entry.state = TableCatalogEntryState::Deleted;
+    backend
+        .seed_object(
+            RUSTFS_META_BUCKET,
+            &store.paths.table_entry_path(bucket, &namespace, &table),
+            serde_json::to_vec(&entry).unwrap(),
+        )
+        .await;
+    entry.state = TableCatalogEntryState::Active;
+    entry.table_id = "replacement-table-id".to_string();
+    store.create_table(entry.clone()).await.unwrap();
+    assert_eq!(
+        store
+            .load_table(bucket, &namespace.public_name(), table.as_str())
+            .await
+            .unwrap(),
+        Some(entry)
+    );
+}
+
 #[tokio::test]
 async fn catalog_backings_reject_overlapping_registered_warehouse_prefixes() {
     let backend = TestCatalogObjectBackend::default();