fix(ecstore): keep the single-block bound on every inline candidate (#7954)

.config/nextest.toml

@@ -132,6 +132,12 @@ test-group = 'ecstore-serial-flaky'
 filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))'
 test-group = 'embedded-test-ports'
 
+# The real object probe includes bucket creation and cleanup in its short
+# measurement budget. Keep competing storage fixtures outside that budget.
+[[profile.default.overrides]]
+filter = 'package(rustfs) & binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)'
+threads-required = "num-test-threads"
+
 # Serialize the durable manual-transition checkpoint test across nextest's
 # process boundary; it mutates bucket lifecycle metadata and is not quarantined.
 [[profile.default.overrides]]
@@ -327,6 +333,10 @@ test-group = 'ecstore-serial-flaky'
 filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))'
 test-group = 'embedded-test-ports'
 
+[[profile.ci.overrides]]
+filter = 'package(rustfs) & binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)'
+threads-required = "num-test-threads"
+
 # Serialize the durable manual-transition checkpoint test under the ci profile
 # too. No retries: failures stay visible.
 [[profile.ci.overrides]]

Cargo.lock

@@ -3790,6 +3790,23 @@ dependencies = [
  "cipher 0.5.2",
 ]
 
+[[package]]
+name = "dial9"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49b11bce55173f77d9003cce8f129b4af889a4639005c53c3498cef933bc895c"
+dependencies = [
+ "bon",
+ "dial9-core",
+ "dial9-macro",
+ "dial9-metrique",
+ "dial9-perf-self-profile",
+ "dial9-tokio-telemetry",
+ "dial9-trace-format",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "dial9-core"
 version = "0.5.0"
@@ -3812,6 +3829,31 @@ dependencies = [
  "ulid",
 ]
 
+[[package]]
+name = "dial9-macro"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4177b8f8bf95d8e3769c1aad596f5d6fb231953a45c4c3d2a5946fe8e4ec327b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "dial9-metrique"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d42fe0e3e14e8780c4ecf9ee14c183dbcb9b1ae3612cc9a689e7dc83b13adef1"
+dependencies = [
+ "dial9-core",
+ "dial9-trace-format",
+ "metrique",
+ "metrique-writer",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "dial9-perf-self-profile"
 version = "0.5.0"
@@ -3827,6 +3869,7 @@ dependencies = [
  "libc",
  "perf-event-data",
  "perf-event-open-sys2",
+ "tokio",
  "tracing",
 ]
 
@@ -10345,7 +10388,7 @@ dependencies = [
  "crossbeam-channel",
  "crossbeam-deque",
  "crossbeam-utils",
- "dial9-tokio-telemetry",
+ "dial9",
  "flate2",
  "futures-util",
  "glob",

Cargo.toml

@@ -351,7 +351,7 @@ zstd = "0.14.0"
 # Observability and Metrics
 metrics = "0.24.6"
 metrics-util = "0.20"
-dial9-tokio-telemetry = "0.5.0"
+dial9 = { version = "0.5.0", default-features = false }
 opentelemetry = { version = "0.32.0" }
 opentelemetry-appender-tracing = { version = "0.32.0" }
 opentelemetry-otlp = { version = "0.32.0" }

crates/config/src/constants/runtime.rs

@@ -39,7 +39,7 @@ pub const ENV_RUNTIME_DIAL9_S3_BUCKET: &str = "RUSTFS_RUNTIME_DIAL9_S3_BUCKET";
 /// Accepted but not honoured; see [`ENV_RUNTIME_DIAL9_S3_BUCKET`].
 pub const ENV_RUNTIME_DIAL9_S3_PREFIX: &str = "RUSTFS_RUNTIME_DIAL9_S3_PREFIX";
 // Note: there are deliberately no task-dump knobs. dial9 only captures task
-// dumps for futures spawned through `dial9_tokio_telemetry::spawn`, and RustFS
+// dumps for futures spawned through `dial9::spawn`, and RustFS
 // spawns with `tokio::spawn` throughout, so the switch could never do anything.
 // Measured: 0 dumps via tokio::spawn vs 14709 via dial9::spawn on an identical
 // workload. See rustfs/backlog#1157 (D9-16).

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

@@ -515,20 +515,26 @@ impl ExpiryStats {
         Self::add_nonnegative(&self.missed_tier_journal_tasks, 1);
     }
 
+    // The pending and active gauges are balanced by design: every increment
+    // has exactly one matching decrement. They must not saturate at zero on
+    // update, because a worker can dequeue (and decrement) before the
+    // enqueuing side has recorded its increment. Clamping that transient -1
+    // to 0 turns the later +1 into a phantom task that never drains
+    // (rustfs#7921). Readers clamp negative snapshots instead.
     fn increment_pending_tasks(&self) {
-        Self::add_nonnegative(&self.pending_tasks, 1);
+        self.pending_tasks.fetch_add(1, Ordering::AcqRel);
     }
 
     fn decrement_pending_tasks(&self) {
-        Self::add_nonnegative(&self.pending_tasks, -1);
+        self.pending_tasks.fetch_sub(1, Ordering::AcqRel);
     }
 
     fn increment_active_tasks(&self) {
-        Self::add_nonnegative(&self.active_tasks, 1);
+        self.active_tasks.fetch_add(1, Ordering::AcqRel);
     }
 
     fn decrement_active_tasks(&self) {
-        Self::add_nonnegative(&self.active_tasks, -1);
+        self.active_tasks.fetch_sub(1, Ordering::AcqRel);
     }
 
     fn increment_workers(&self) {
@@ -1074,9 +1080,13 @@ impl ExpiryState {
     }
 
     fn send_expiry_task(&self, wrkr: Sender<Option<ExpiryOpType>>, task: ExpiryOpType) -> bool {
+        // Account for the task before a worker can observe it. The worker
+        // decrements on dequeue, so incrementing after `try_send` would let a
+        // fast dequeue run the gauge through zero first.
+        self.stats.increment_pending_tasks();
         let queued = wrkr.try_send(Some(task)).is_ok();
-        if queued {
-            self.stats.increment_pending_tasks();
+        if !queued {
+            self.stats.decrement_pending_tasks();
         }
         queued
     }
@@ -1444,11 +1454,14 @@ async fn enqueue_recovered_free_version_with_state(state: &Arc<RwLock<ExpiryStat
         return false;
     };
 
+    // Same ordering rule as `ExpiryState::send_expiry_task`: count first, so
+    // a worker that dequeues immediately cannot decrement before this
+    // increment lands.
+    stats.increment_pending_tasks();
     let queued = wrkr.try_send(Some(Box::new(task))).is_ok();
     if !queued {
+        stats.decrement_pending_tasks();
         stats.increment_missed_freevers_tasks();
-    } else {
-        stats.increment_pending_tasks();
     }
     stats.record_scanner_expiry_state();
     queued
@@ -3757,11 +3770,8 @@ pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) {
     }
 }
 
-pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) {
-    let Some(api) = runtime_sources::object_store_handle() else {
-        return;
-    };
-    let configs = match metadata_boundary::get_expiry_configs(&api, &oi.bucket).await {
+pub(crate) async fn enqueue_immediate_expiry(api: Arc<ECStore>, oi: &ObjectInfo, src: LcEventSrc, opts: &ObjectOptions) {
+    let configs = match metadata_boundary::get_expiry_configs_for_options(&api, &oi.bucket, opts).await {
         Ok(configs) => configs,
         Err(err) => {
             observe_lifecycle_observability_event(EVENT_LIFECYCLE_EVALUATION_FAILED, "failed", Some("metadata_unavailable"));
@@ -7572,6 +7582,65 @@ mod tests {
         assert!(recovery_notify.notified().now_or_never().is_some());
     }
 
+    #[tokio::test]
+    async fn expiry_pending_gauge_survives_dequeue_landing_before_enqueue_accounting() {
+        // A worker may dequeue and decrement before the enqueuing side records
+        // its increment. The gauge must return to zero afterwards instead of
+        // clamping the transient -1 away and reporting a phantom pending task
+        // that no idle check can ever drain (rustfs#7921).
+        let state = ExpiryState::new();
+        let stats = Arc::clone(&state.read().await.stats);
+
+        stats.decrement_pending_tasks();
+        stats.increment_pending_tasks();
+        assert_eq!(stats.pending_tasks(), 0);
+        assert_eq!(state.read().await.pending_tasks(), 0);
+
+        stats.decrement_active_tasks();
+        stats.increment_active_tasks();
+        assert_eq!(stats.active_tasks(), 0);
+        assert_eq!(state.read().await.active_tasks(), 0);
+    }
+
+    #[tokio::test]
+    async fn free_version_enqueue_rolls_back_pending_when_queue_full() {
+        // Single-threaded, so this cannot observe the count-before-publish
+        // ordering itself; it pins the rollback that ordering requires: a
+        // rejected send must not leave its speculative increment behind.
+        let state = ExpiryState::new_with_unconsumed_worker_channel(1);
+        let oi = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            transitioned_object: TransitionedObject {
+                name: "remote/object".to_string(),
+                version_id: "remote-version".to_string(),
+                tier: "WARM".to_string(),
+                free_version: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        assert!(state.read().await.enqueue_free_version(oi.clone()));
+        assert_eq!(state.read().await.stats.pending_tasks(), 1);
+
+        // The single-slot queue is full for both enqueue paths.
+        assert!(!enqueue_recovered_free_version_with_state(&state, oi.clone()).await);
+        assert_eq!(state.read().await.stats.pending_tasks(), 1);
+        assert_eq!(state.read().await.stats.missed_free_vers_tasks(), 1);
+
+        assert!(!state.read().await.enqueue_free_version(oi));
+        assert_eq!(state.read().await.stats.pending_tasks(), 1);
+        assert_eq!(state.read().await.stats.missed_free_vers_tasks(), 2);
+
+        // Draining the one real task returns the gauge to zero.
+        let receiver = state.read().await.tasks_rx[0].clone();
+        let task = receiver.lock().await.recv().await.expect("queued task");
+        assert!(task.is_some());
+        state.read().await.stats.decrement_pending_tasks();
+        assert_eq!(state.read().await.pending_tasks(), 0);
+    }
+
     #[tokio::test]
     async fn enqueue_recovered_free_version_reports_false_without_worker_channel() {
         let state = ExpiryState::new();
@@ -12703,7 +12772,8 @@ mod tests {
                 .push((event, state, reason));
         });
 
-        super::enqueue_immediate_expiry(&object_info, LcEventSrc::S3PutObject).await;
+        super::enqueue_immediate_expiry(Arc::clone(&ecstore), &object_info, LcEventSrc::S3PutObject, &ObjectOptions::default())
+            .await;
 
         assert!(
             observed.lock().expect("observed events should not poison").contains(&(

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

@@ -55,7 +55,15 @@ pub(crate) async fn lifecycle_expiry_allowed(
 }
 
 pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str) -> Result<LifecycleExpiryConfigs> {
-    let bucket_incarnation_id = api.bucket_incarnation_id_from_disk(bucket).await?;
+    get_expiry_configs_for_options(api, bucket, &crate::object_api::ObjectOptions::default()).await
+}
+
+pub(crate) async fn get_expiry_configs_for_options(
+    api: &crate::store::ECStore,
+    bucket: &str,
+    opts: &crate::object_api::ObjectOptions,
+) -> Result<LifecycleExpiryConfigs> {
+    let bucket_incarnation_id = metadata_sys::get_bucket_incarnation_id_for_options_in(&api.ctx, bucket, opts).await?;
     let metadata = get_authoritative_metadata(api, bucket, bucket_incarnation_id).await?;
     let table_bucket_enabled = metadata.table_bucket_enabled();
 

crates/ecstore/src/bucket/metadata_sys.rs

@@ -24,6 +24,7 @@ use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_p
 use crate::bucket::utils::is_meta_bucketname;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result, is_err_bucket_not_found, is_err_strict_volume_not_found};
+use crate::object_api::ObjectOptions;
 use crate::runtime::sources as runtime_sources;
 use crate::storage_api_contracts::heal::HealOperations as _;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
@@ -490,6 +491,18 @@ pub(crate) async fn get_bucket_incarnation_id_in(ctx: &crate::runtime::instance:
     sys.get_bucket_incarnation_id_from_disk(bucket).await
 }
 
+/// Validate against disk while retaining an already-held Object Lock fence.
+pub(crate) async fn get_bucket_incarnation_id_for_options_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<Uuid> {
+    let sys = bucket_metadata_sys_of(ctx)?;
+    let sys = sys.read().await.clone();
+    let guard = acquire_bucket_metadata_transaction_read_lock_for_options_in(ctx, bucket, opts).await?;
+    sys.get_bucket_incarnation_id_under_transaction_lock(bucket, &guard).await
+}
+
 pub(crate) async fn get_cached_bucket_incarnation_id_in(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
@@ -1055,6 +1068,24 @@ pub(crate) async fn acquire_bucket_metadata_transaction_read_lock_in(
     Ok(lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?)
 }
 
+/// Readers already holding an Object Lock snapshot must share its guard:
+/// a fresh read acquisition can queue behind a writer waiting for that snapshot.
+pub(crate) async fn acquire_bucket_metadata_transaction_read_lock_for_options_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<Arc<rustfs_lock::NamespaceLockGuard>> {
+    if let Some(snapshot) = opts.object_lock_config_snapshot.as_ref() {
+        let store = object_store_in(ctx).await?;
+        return snapshot
+            .metadata_transaction_guard_for(store.id, bucket, opts.expected_bucket_incarnation_id)
+            .ok_or_else(|| {
+                Error::other("Object Lock snapshot does not hold a valid metadata transaction fence for this bucket")
+            });
+    }
+    Ok(Arc::new(acquire_bucket_metadata_transaction_read_lock_in(ctx, bucket).await?))
+}
+
 async fn acquire_transaction_lock_with_sys(
     sys: &Arc<RwLock<BucketMetadataSys>>,
     bucket: &str,
@@ -2357,8 +2388,17 @@ impl BucketMetadataSys {
         let _transaction_guard = transaction_lock
             .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
             .await?;
+        self.get_bucket_incarnation_id_under_transaction_lock(bucket, &_transaction_guard)
+            .await
+    }
+
+    async fn get_bucket_incarnation_id_under_transaction_lock(
+        &self,
+        bucket: &str,
+        transaction_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<Uuid> {
         let incarnation_id = load_bucket_incarnation(self.object_store(), bucket).await?;
-        if _transaction_guard.is_lock_lost() {
+        if transaction_guard.is_lock_lost() {
             return Err(Error::other(format!("bucket incarnation metadata transaction lock was lost: {bucket}")));
         }
         match incarnation_id {

crates/ecstore/src/bucket/quota/reservation.rs

@@ -38,6 +38,7 @@ const MAX_ORPHANS_REAPED_PER_WRITE: usize = 64;
 const MAX_ORPHAN_PROBES_PER_WRITE: usize = 128;
 const ORPHAN_PROBE_CONCURRENCY: usize = 32;
 const EVENT_QUOTA_LEDGER_SETTLEMENT: &str = "quota_ledger_settlement";
+const EVENT_QUOTA_ADMISSION: &str = "quota_admission";
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_QUOTA: &str = "quota";
 
@@ -46,6 +47,8 @@ static FAIL_NEXT_LEDGER_SAVE: std::sync::atomic::AtomicBool = std::sync::atomic:
 
 // Lock order: caller-held destination object/upload, bucket metadata
 // transaction (read), operation reservation, then quota ledger.
+// When Object Lock already holds the metadata transaction read lock, reuse
+// that guard: reacquiring it behind a waiting metadata writer would deadlock.
 
 #[cfg(not(any(test, feature = "test-util")))]
 const ORPHAN_MIN_AGE_SECONDS: i64 = 30;
@@ -210,7 +213,7 @@ pub(crate) struct QuotaContext {
     capability_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     snapshot_admission: Option<QuotaAdmission>,
     legacy_data_movement: bool,
-    metadata_guard: Option<NamespaceLockGuard>,
+    metadata_guard: Option<Arc<NamespaceLockGuard>>,
     pool_index: Option<usize>,
     set_index: Option<usize>,
 }
@@ -295,7 +298,7 @@ impl QuotaContext {
                     limit: quota_limit,
                 });
             }
-            if operation_guard.is_lock_lost() || metadata_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) {
+            if operation_guard.is_lock_lost() || metadata_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
                 return Err(StorageError::NamespaceLockQuorumUnavailable {
                     mode: "quota_reservation",
                     bucket: ledger_data.bucket.clone(),
@@ -332,7 +335,7 @@ struct LedgerReservationData {
 pub(crate) struct QuotaReservation {
     ledger: Option<LedgerReservationData>,
     operation_guard: Option<NamespaceLockGuard>,
-    metadata_guard: Option<NamespaceLockGuard>,
+    metadata_guard: Option<Arc<NamespaceLockGuard>>,
     capability_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     state: ReservationState,
 }
@@ -346,7 +349,7 @@ enum ReservationState {
 }
 
 impl QuotaReservation {
-    fn unlimited(metadata_guard: Option<NamespaceLockGuard>) -> Self {
+    fn unlimited(metadata_guard: Option<Arc<NamespaceLockGuard>>) -> Self {
         Self {
             ledger: None,
             operation_guard: None,
@@ -358,7 +361,7 @@ impl QuotaReservation {
 
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.operation_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
-            || self.metadata_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
+            || self.metadata_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
     }
 
     pub(crate) fn capability_proof_matches(&self) -> bool {
@@ -438,11 +441,12 @@ pub(crate) async fn begin(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
     object: &str,
-    snapshot_admission: Option<QuotaAdmission>,
-    data_movement: bool,
+    opts: &ObjectOptions,
     pool_index: usize,
     set_index: usize,
 ) -> Result<QuotaContext> {
+    let snapshot_admission = opts.quota_admission;
+    let data_movement = opts.data_movement;
     if crate::bucket::utils::is_meta_bucketname(bucket) {
         return Ok(QuotaContext {
             store: None,
@@ -497,7 +501,7 @@ pub(crate) async fn begin(
         });
     }
 
-    let metadata_guard = metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(ctx, bucket).await?;
+    let metadata_guard = metadata_sys::acquire_bucket_metadata_transaction_read_lock_for_options_in(ctx, bucket, opts).await?;
     let (quota, bucket_incarnation, quota_revision) =
         metadata_sys::get_quota_config_and_incarnation_from_disk_in(ctx, bucket).await?;
     if metadata_guard.is_lock_lost() {
@@ -513,6 +517,7 @@ pub(crate) async fn begin(
         .as_ref()
         .is_some_and(|quota| quota.has_unsupported_reservation_protocol())
     {
+        log_admission_rejected(bucket, object, "unsupported_reservation_protocol");
         return Err(StorageError::PartMissingOrCorrupt);
     }
     let durable_quota = quota.as_ref().filter(|quota| quota.uses_durable_reservations());
@@ -529,7 +534,16 @@ pub(crate) async fn begin(
         Some(quota) => match (quota.quota, snapshot_admission) {
             (Some(limit), Some(admission)) if admission.quota_limit() == limit => Some(admission),
             (Some(_), None) if data_movement => None,
-            (Some(_), _) => return Err(StorageError::PartMissingOrCorrupt),
+            (Some(_), _) => {
+                // A snapshot-protocol quota requires the request handler's
+                // admission on every write. Missing or mismatched admission
+                // means a caller rebuilt `ObjectOptions` without carrying it
+                // over (rustfs/rustfs#7674 lost it on CopyObject); fail closed
+                // but leave a diagnosable trace, because the storage error is
+                // the generic `PartMissingOrCorrupt`.
+                log_admission_rejected(bucket, object, "snapshot_quota_admission_missing");
+                return Err(StorageError::PartMissingOrCorrupt);
+            }
             (None, _) => None,
         },
         None => None,
@@ -904,6 +918,18 @@ pub fn fail_next_quota_ledger_save_for_test() {
     FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
 }
 
+fn log_admission_rejected(bucket: &str, object: &str, state: &'static str) {
+    warn!(
+        event = EVENT_QUOTA_ADMISSION,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_QUOTA,
+        state,
+        bucket = %bucket,
+        object = %object,
+        "quota admission rejected the write before commit"
+    );
+}
+
 fn log_deferred_settlement(data: &LedgerReservationData, state: &'static str, err: &StorageError) {
     warn!(
         event = EVENT_QUOTA_LEDGER_SETTLEMENT,

crates/ecstore/src/core/pools_test.rs

@@ -481,6 +481,9 @@ mod decommission_lock_order_tests {
         if let Some(deployment_id) = other_store.ctx.deployment_id() {
             ctx.set_deployment_id(deployment_id);
         }
+        for pool in &mut pools {
+            Arc::make_mut(pool).set_instance_ctx_for_test(Arc::clone(&ctx));
+        }
         let store = Arc::new(crate::store::ECStore {
             id: uuid::Uuid::new_v4(),
             disk_map: other_store.disk_map.clone(),
@@ -6645,7 +6648,19 @@ mod decommission_lock_order_tests {
 
         let restore_get_barrier = if matches!(mutation, ExternalObjectMutation::Restore) {
             let tier_name = format!("ORDERRESTORE{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-            let backend = register_mock_tier(&store.pools[2].instance_ctx().tier_config_mgr(), &tier_name).await;
+            let source_tiers = store.tier_config_mgr();
+            let restore_tiers = other_store.tier_config_mgr();
+            let backend = register_mock_tier(&source_tiers, &tier_name).await;
+            // Both peers must resolve the same persisted backend identity, including its prefix.
+            let tier_config = source_tiers.read().await.tiers[&tier_name].clone();
+            restore_tiers.write().await.tiers.insert(tier_name.clone(), tier_config);
+            crate::services::tier::tier::TierConfigMgr::install_test_driver_in(
+                &restore_tiers,
+                &tier_name,
+                Box::new(backend.clone()),
+            )
+            .await
+            .expect("install the same mock tier on the restoring peer");
             store.pools[2]
                 .transition_object(
                     &bucket,
@@ -6919,7 +6934,10 @@ mod decommission_lock_order_tests {
             }
         });
         if let Some(get_barrier) = restore_get_barrier.as_ref() {
-            get_barrier.wait_until_paused().await;
+            tokio::select! {
+                () = get_barrier.wait_until_paused() => {}
+                result = &mut ordinary_mutation => panic!("restore finished before the tier GET barrier: {result:?}"),
+            }
             let read_opts = ObjectOptions {
                 skip_decommissioned: true,
                 ..Default::default()

crates/ecstore/src/core/sets.rs

@@ -324,6 +324,15 @@ impl Sets {
         &self.ctx
     }
 
+    /// Keep simulated peers' metadata ownership separate while sharing disks and lock clients.
+    #[cfg(test)]
+    pub(crate) fn set_instance_ctx_for_test(&mut self, ctx: Arc<InstanceContext>) {
+        for set in &mut self.disk_set {
+            Arc::make_mut(set).set_instance_ctx_for_test(Arc::clone(&ctx));
+        }
+        self.ctx = ctx;
+    }
+
     async fn monitor_and_connect_endpoints_task(sets: Weak<Sets>, mut rx: Receiver<()>) {
         let startup_delay = tokio::time::sleep(Duration::from_secs(5));
         tokio::pin!(startup_delay);
@@ -790,6 +799,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for Sets {
             version_id: dst_opts.version_id.clone(),
             mod_time: dst_opts.mod_time,
             http_preconditions: dst_opts.http_preconditions.clone(),
+            quota_admission: dst_opts.quota_admission,
             ..Default::default()
         };
 

crates/ecstore/src/object_api/types.rs

@@ -41,7 +41,7 @@ impl Debug for NamespaceLockFence {
 }
 
 impl NamespaceLockFence {
-    fn new() -> Self {
+    pub(crate) fn new() -> Self {
         Self {
             signals: Arc::default(),
             #[cfg(test)]
@@ -153,7 +153,7 @@ pub struct ObjectLockConfigSnapshot {
     state: crate::bucket::metadata_sys::ObjectLockConfigState,
     lifecycle_fence: NamespaceLockFence,
     _lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
-    metadata_transaction_guard: Option<rustfs_lock::NamespaceLockGuard>,
+    metadata_transaction_guard: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
 }
 
 impl ObjectLockConfigSnapshot {
@@ -170,6 +170,7 @@ impl ObjectLockConfigSnapshot {
         }
     }
 
+    #[cfg(test)]
     pub(crate) fn for_store_bucket(
         store_id: Uuid,
         bucket: &str,
@@ -210,7 +211,7 @@ impl ObjectLockConfigSnapshot {
             state,
             lifecycle_fence,
             _lifecycle_guard: Some(lifecycle_guard),
-            metadata_transaction_guard: Some(metadata_transaction_guard),
+            metadata_transaction_guard: Some(Arc::new(metadata_transaction_guard)),
         }
     }
 
@@ -221,7 +222,7 @@ impl ObjectLockConfigSnapshot {
         config_revision: OffsetDateTime,
         state: crate::bucket::metadata_sys::ObjectLockConfigState,
         lifecycle_fence: NamespaceLockFence,
-        metadata_transaction_guard: rustfs_lock::NamespaceLockGuard,
+        metadata_transaction_guard: Arc<rustfs_lock::NamespaceLockGuard>,
     ) -> Self {
         Self {
             store_id: Some(store_id),
@@ -265,6 +266,23 @@ impl ObjectLockConfigSnapshot {
                 .is_some_and(|guard| !guard.is_lock_lost())
     }
 
+    /// Share the held transaction lock without queuing another reader behind
+    /// a metadata writer that is itself waiting for this snapshot to drop.
+    pub(crate) fn metadata_transaction_guard_for(
+        &self,
+        store_id: Uuid,
+        bucket: &str,
+        expected_incarnation_id: Option<Uuid>,
+    ) -> Option<Arc<rustfs_lock::NamespaceLockGuard>> {
+        let bucket_incarnation_id = self.bucket_incarnation_id?;
+        if expected_incarnation_id.is_some_and(|expected| expected != bucket_incarnation_id)
+            || !self.is_valid_for_destructive_put(store_id, bucket, bucket_incarnation_id)
+        {
+            return None;
+        }
+        self.metadata_transaction_guard.clone()
+    }
+
     pub(crate) fn add_lock_fences(&self, opts: &mut ObjectOptions) {
         opts.bucket_lifecycle_lock_fence
             .get_or_insert_with(NamespaceLockFence::new)

crates/ecstore/src/services/rebalance/mod.rs

@@ -222,10 +222,14 @@ async fn test_pool_stores_with_contexts(
         std::sync::Arc::clone(&ctx)
     };
     let make_store = |store_ctx: std::sync::Arc<crate::runtime::instance::InstanceContext>| {
+        let mut store_pools = pools.clone();
+        for pool in &mut store_pools {
+            std::sync::Arc::make_mut(pool).set_instance_ctx_for_test(std::sync::Arc::clone(&store_ctx));
+        }
         std::sync::Arc::new(crate::store::ECStore {
             id: uuid::Uuid::new_v4(),
             disk_map: std::collections::HashMap::new(),
-            pools: pools.clone(),
+            pools: store_pools,
             peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, std::sync::Arc::clone(&store_ctx)),
             pool_meta: tokio::sync::RwLock::new(pool_meta.clone()),
             rebalance_meta: tokio::sync::RwLock::new(rebalance_meta.clone()),

crates/ecstore/src/set_disk/mod.rs

@@ -4703,6 +4703,12 @@ impl SetDisks {
         &self.ctx
     }
 
+    #[cfg(test)]
+    pub(crate) fn set_instance_ctx_for_test(&mut self, ctx: Arc<InstanceContext>) {
+        self.local_lock_manager = ctx.lock_manager();
+        self.ctx = ctx;
+    }
+
     /// Read the persisted bucket identity through this set's metadata owner.
     /// Missing or non-authoritative legacy identities remain errors.
     pub async fn bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Uuid> {

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

@@ -2613,16 +2613,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         fi.parts = Vec::with_capacity(uploaded_parts.len());
 
-        let quota_context = reservation::begin(
-            &self.ctx,
-            bucket,
-            object,
-            opts.quota_admission,
-            opts.data_movement,
-            self.pool_index,
-            self.set_index,
-        )
-        .await?;
+        let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?;
         let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
         let preserve_replication_ciphertext = opts.replication_request
             && contains_key_str(&fi.metadata, rustfs_utils::http::SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT);

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

@@ -3427,8 +3427,8 @@ impl SetDisks {
         Ok(removed)
     }
 
-    async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid) -> Result<()> {
-        let current = metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+    async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid, opts: &ObjectOptions) -> Result<()> {
+        let current = metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, opts).await?;
         if current != expected {
             return Err(StorageError::BucketNotFound(bucket.to_string()));
         }
@@ -3636,13 +3636,22 @@ impl SetDisks {
             // Transformed streams (unknown stored size) are admitted by their
             // plaintext size and then take the streaming encode with inline
             // buffer writers, since the single-block fast path needs a known length.
-            // Protected inline writes retain the single-block bound on proof metadata.
+            // Every inline candidate keeps the single-block bound on its admission
+            // size: the inline buffer writer grows without a cap, so an explicit
+            // INLINE_BLOCK budget above `block_size` must not admit multi-block
+            // payloads into memory and xl.meta. Protected inline writes also need
+            // a known stored length for their proof metadata, so they never fall
+            // back to the plaintext size.
+            let inline_admission_size = if put_object_size >= 0 || protect_write {
+                put_object_size
+            } else {
+                data.actual_size()
+            };
             let is_inline_buffer = storage_class_config.should_inline(
                 inline_admission_shard_size(&erasure, put_object_size, data.actual_size()),
                 erasure.data_shards,
                 opts.versioned,
-            ) && (!protect_write
-                || usize::try_from(put_object_size).is_ok_and(|size| size <= erasure.block_size));
+            ) && usize::try_from(inline_admission_size).is_ok_and(|size| size <= erasure.block_size);
 
             let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
             let shard_file_size = shard_file_size_raw;
@@ -4234,16 +4243,7 @@ impl SetDisks {
                 None
             };
 
-            let quota_context = reservation::begin(
-                &self.ctx,
-                bucket,
-                object,
-                opts.quota_admission,
-                opts.data_movement,
-                self.pool_index,
-                self.set_index,
-            )
-            .await?;
+            let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?;
             let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
             let mut replication_quota_size = None;
 
@@ -8869,7 +8869,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 return Ok(ObjectInfo::default());
             }
             if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id {
-                self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+                self.validate_bucket_incarnation(bucket, expected_incarnation_id, &opts)
+                    .await?;
             }
             ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
             begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
@@ -9283,7 +9284,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         check_object_lock_retention_update(bucket, object, &obj_info, opts)?;
 
         if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id {
-            self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+            self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts)
+                .await?;
         }
         if _lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
             || opts
@@ -9863,7 +9865,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 .await?
                 .acquire_bucket_lifecycle_read_lock(bucket)
                 .await?;
-            self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+            self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts)
+                .await?;
             Some(guard)
         } else {
             None
@@ -11801,6 +11804,54 @@ mod inline_put_commit_path_tests {
         }
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn unprotected_puts_keep_multi_block_inline_candidates_external() {
+        for transformed in [true, false] {
+            let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let storage_class =
+                temp_env::with_var(INLINE_BLOCK_ENV, Some("16MiB"), || lookup_config_for_pools(&KVS::new(), &[4]))
+                    .expect("large inline budget should resolve");
+            set_disks.set_test_storage_class_config(storage_class);
+            let bucket = "unprotected-inline-candidates";
+            let object = "object.bin";
+            make_bucket(&disk_stores, bucket).await;
+            // Above the single erasure block: an explicit INLINE_BLOCK budget must
+            // not pull a multi-block payload into the unbounded inline buffer.
+            let plaintext = compressible_payload(1024 * 1024 + 17);
+            let (mut reader, opts) = if transformed {
+                transformed_put(plaintext.clone())
+            } else {
+                (PutObjReader::from_vec(plaintext.clone()), ObjectOptions::default())
+            };
+            temp_env::async_with_vars(
+                [(
+                    crate::set_disk::core::io_primitives::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE,
+                    Some("false"),
+                )],
+                set_disks.put_object(bucket, object, &mut reader, &opts),
+            )
+            .await
+            .expect("multi-block PUT should commit with part files");
+
+            for (disk_index, disk) in disk_stores.iter().enumerate() {
+                let file_info = disk
+                    .read_version("", bucket, object, "", &ReadOptions::default())
+                    .await
+                    .unwrap_or_else(|err| panic!("disk {disk_index} should persist metadata: {err}"));
+                assert!(
+                    !file_info.inline_data(),
+                    "disk {disk_index}: a payload above block_size must stay external (transformed={transformed})"
+                );
+            }
+            assert!(
+                count_part_files(&temp_dirs) > 0,
+                "a multi-block object must keep part files (transformed={transformed})"
+            );
+            assert_eq!(read_back(&set_disks, bucket, object).await, plaintext);
+        }
+    }
+
     #[tokio::test]
     async fn transformed_put_above_inline_budget_keeps_part_files() {
         let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/ecstore/src/store/heal.rs

@@ -2776,7 +2776,16 @@ mod tests {
 
         let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
         store.pools[0]
-            .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
+            .put_object(
+                &bucket,
+                active_object,
+                &mut active_reader,
+                &ObjectOptions {
+                    // Drain the rename tail before removing a physical shard.
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
             .await
             .expect("active owner object should be written");
         let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
@@ -2812,6 +2821,7 @@ mod tests {
                     &mut duplicate_reader,
                     &ObjectOptions {
                         mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
+                        write_completion: crate::object_api::WriteCompletion::TailDrained,
                         ..Default::default()
                     },
                 )

crates/ecstore/src/store/init.rs

@@ -2905,8 +2905,17 @@ mod tests {
                     if !version.is_empty() {
                         insert_str(&mut metadata, SUFFIX_TRANSITIONED_VERSION_ID, version.to_string());
                     }
+                    // The commit path acknowledges at write quorum and drains
+                    // the remaining rename fan-out in the background. This
+                    // fixture "crashes" the store right after the overwrite,
+                    // so it must wait for that tail: a disk left with the old
+                    // live transitioned source makes exact cleanup fail closed
+                    // (a minority live owner still references the remote
+                    // tuple) until heal repairs it, which this fixture never
+                    // runs (rustfs#7921).
                     let options = ObjectOptions {
                         version_suspended: suspended,
+                        write_completion: crate::object_api::WriteCompletion::TailDrained,
                         ..Default::default()
                     };
                     store
@@ -2978,6 +2987,7 @@ mod tests {
                     assert_eq!(free.len(), 1, "{state:?}, suspended={suspended}, copy={self_copy}");
                     assert_eq!(free[0].transitioned_objname, remote);
                     assert_eq!(free[0].transition_version_state, state);
+                    assert_every_disk_holds_only_the_cleanup_owner(&set, &bucket, object, &remote).await;
                     assert!(backend.contains(&remote).await, "commit must not delete remote bytes before cleanup");
                     let removed_before = backend.remove_count().await;
 
@@ -3149,6 +3159,120 @@ mod tests {
         shutdown.cancel();
     }
 
+    /// rustfs/rustfs#7674: a bucket carrying a legacy snapshot-protocol quota
+    /// (written before durable reservations existed, so `quota.json` has no
+    /// `reservation_protocol`) must accept a cross-key CopyObject whose
+    /// destination options carry the handler's quota admission. The storage
+    /// layer fails closed with `PartMissingOrCorrupt` when a quota-enforced
+    /// bucket sees a write without admission, so dropping the admission while
+    /// rebuilding the destination options turns every copy into a
+    /// deterministic "part missing or corrupt" failure even though the source
+    /// object is perfectly readable.
+    #[cfg(feature = "test-util")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(storage_class_env)]
+    async fn copy_object_forwards_quota_admission_on_legacy_snapshot_quota_bucket() {
+        let temp_dir = tempfile::tempdir().expect("create legacy quota copy store dir");
+        let (ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-quota-copy", &[1])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+
+        let bucket = format!("legacy-quota-copy-{}", Uuid::new_v4());
+        let source_object = "docker/registry/v2/repositories/example/_uploads/upload-id/data";
+        let target_object = "docker/registry/v2/blobs/sha256/a5/digest/data";
+        let payload = vec![0x5A; 8178];
+        let quota_limit = 1u64 << 30;
+
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket for legacy quota copy");
+        // Legacy quota shape: no `reservation_protocol`, so the storage layer
+        // requires the handler-supplied snapshot admission on every write.
+        let legacy_quota = format!(r#"{{"quota":{quota_limit},"quota_type":"Hard"}}"#);
+        crate::bucket::metadata_sys::update_in(
+            &ctx,
+            &bucket,
+            crate::bucket::metadata::BUCKET_QUOTA_CONFIG_FILE,
+            legacy_quota.into_bytes(),
+        )
+        .await
+        .expect("persist legacy snapshot quota");
+        let (quota, _, _) = crate::bucket::metadata_sys::get_quota_config_and_incarnation_from_disk_in(&ctx, &bucket)
+            .await
+            .expect("legacy quota should load");
+        let quota = quota.expect("legacy quota must be persisted");
+        assert_eq!(quota.quota, Some(quota_limit));
+        assert!(!quota.uses_durable_reservations(), "fixture must stay on the snapshot protocol");
+
+        let mut write_opts = ObjectOptions::default();
+        assert!(write_opts.set_quota_admission(0, quota_limit));
+
+        let upload = store
+            .new_multipart_upload(&bucket, source_object, &write_opts)
+            .await
+            .expect("create source multipart upload");
+        let mut part_reader = PutObjReader::from_vec(payload.clone());
+        let part = store
+            .put_object_part(&bucket, source_object, &upload.upload_id, 1, &mut part_reader, &write_opts)
+            .await
+            .expect("stage multipart source part");
+        store
+            .clone()
+            .complete_multipart_upload(
+                &bucket,
+                source_object,
+                &upload.upload_id,
+                vec![crate::storage_api_contracts::multipart::CompletePart {
+                    part_num: part.part_num,
+                    etag: part.etag,
+                    ..Default::default()
+                }],
+                &write_opts,
+            )
+            .await
+            .expect("complete the multipart source under the legacy quota");
+
+        let source_reader = store
+            .get_object_reader(&bucket, source_object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("completed multipart source should be readable");
+        let mut copy_info = source_reader.object_info.clone();
+        let actual_size = copy_info.get_actual_size().expect("copy source logical size should resolve");
+        assert_eq!(actual_size, payload.len() as i64);
+        let copy_reader = rustfs_rio::HashReader::from_stream(source_reader.stream, actual_size, actual_size, None, None, false)
+            .expect("copy source hash reader should build");
+        copy_info.put_object_reader = Some(PutObjReader::new(copy_reader));
+
+        let mut dst_opts = ObjectOptions::default();
+        assert!(dst_opts.set_quota_admission(payload.len() as u64, quota_limit));
+        store
+            .copy_object(
+                &bucket,
+                source_object,
+                &bucket,
+                target_object,
+                &mut copy_info,
+                &ObjectOptions::default(),
+                &dst_opts,
+            )
+            .await
+            .expect("CopyObject must forward the handler quota admission to the destination write");
+
+        let mut target_reader = store
+            .get_object_reader(&bucket, target_object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("copied target should be readable");
+        let mut target_body = Vec::new();
+        target_reader
+            .stream
+            .read_to_end(&mut target_body)
+            .await
+            .expect("target body should stream");
+        assert_eq!(target_body, payload);
+        shutdown.cancel();
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     #[serial_test::serial(storage_class_env)]
@@ -9712,14 +9836,7 @@ mod tests {
             .await
             .expect("manual task receipt path should resolve");
         let target_task_set = store.pools[1].get_disks_by_key(&manual_task_receipt_path);
-        let original_target_task_disks = {
-            let mut disks = target_task_set.disks.write().await;
-            let original = disks.clone();
-            for disk in disks.iter_mut().take(2) {
-                *disk = None;
-            }
-            original
-        };
+        let offline_target_task_disks = force_set_disk_range_offline_for_test(&target_task_set, 0..2).await;
         let receipt_quorum_error = store
             .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(
                 0,
@@ -9728,7 +9845,7 @@ mod tests {
             )
             .await
             .expect_err("target read quorum without receipt write quorum must retain the source");
-        *target_task_set.disks.write().await = original_target_task_disks;
+        drop(offline_target_task_disks);
         let receipt_quorum_error = receipt_quorum_error.to_string();
         assert!(receipt_quorum_error.contains("receipt"));
         assert!(receipt_quorum_error.contains(&manual_task_path));
@@ -12648,10 +12765,45 @@ mod tests {
         }
     }
 
+    /// Every physical copy must carry the replacement plus its tier
+    /// free-version owner, and none may still hold the pre-overwrite live
+    /// transitioned source. A stale minority copy is exactly what a lost
+    /// early-ACK rename tail leaves behind, and exact cleanup refuses to
+    /// delete remote bytes while such a live reference exists.
+    #[cfg(feature = "test-util")]
+    async fn assert_every_disk_holds_only_the_cleanup_owner(
+        set: &crate::set_disk::SetDisks,
+        bucket: &str,
+        object: &str,
+        remote: &str,
+    ) {
+        use crate::disk::DiskAPI as _;
+
+        let disk_object = rustfs_utils::path::encode_dir_object(object);
+        for (index, disk) in set.disk_inventory().await.into_iter().enumerate() {
+            let disk = disk.unwrap_or_else(|| panic!("disk{index} should be online"));
+            let raw = disk
+                .read_xl(bucket, &disk_object, false)
+                .await
+                .unwrap_or_else(|err| panic!("disk{index} xl.meta should be readable after the overwrite: {err:?}"));
+            let versions = rustfs_filemeta::FileMeta::load(&raw.buf)
+                .and_then(|meta| meta.get_all_file_info_versions(bucket, object, true))
+                .unwrap_or_else(|err| panic!("disk{index} xl.meta should decode: {err:?}"));
+            let all: Vec<_> = versions.versions.iter().chain(versions.free_versions.iter()).collect();
+            let owners = all.iter().filter(|fi| fi.tier_free_version()).count();
+            let live_sources = all
+                .iter()
+                .filter(|fi| !fi.tier_free_version() && fi.transitioned_objname == remote)
+                .count();
+            assert_eq!(owners, 1, "disk{index} must hold exactly one cleanup owner after the drained overwrite");
+            assert_eq!(live_sources, 0, "disk{index} must not retain the pre-overwrite live transitioned source");
+        }
+    }
+
     #[cfg(feature = "test-util")]
     async fn wait_for_expiry_workers_idle(store: &crate::store::ECStore) {
         let expiry_state = store.ctx.expiry_state();
-        tokio::time::timeout(Duration::from_secs(30), async {
+        let idle = tokio::time::timeout(Duration::from_secs(30), async {
             loop {
                 let idle = {
                     let state = expiry_state.read().await;
@@ -12671,8 +12823,15 @@ mod tests {
                 }
             }
         })
-        .await
-        .expect("lifecycle expiry workers should become idle");
+        .await;
+        if idle.is_err() {
+            let state = expiry_state.read().await;
+            panic!(
+                "lifecycle expiry workers should become idle: pending={} active={}",
+                state.pending_tasks(),
+                state.active_tasks()
+            );
+        }
     }
 
     #[cfg(feature = "test-util")]

crates/ecstore/src/store/mod.rs

@@ -395,12 +395,19 @@ async fn scan_metadata_less_residue_with_budget(
     Ok(scan)
 }
 
-async fn enqueue_transition_after_write(result: Result<ObjectInfo>, src: LcEventSrc) -> Result<ObjectInfo> {
+async fn enqueue_transition_after_write(
+    store: &ECStore,
+    result: Result<ObjectInfo>,
+    src: LcEventSrc,
+    opts: &ObjectOptions,
+) -> Result<ObjectInfo> {
     match result {
         Ok(oi) => {
             if should_enqueue_transition_immediately(&oi) {
                 enqueue_transition_immediate(&oi, src.clone()).await;
-                enqueue_immediate_expiry(&oi, src).await;
+                if let Ok(api) = metadata_sys::object_store_in(&store.ctx).await {
+                    enqueue_immediate_expiry(api, &oi, src, opts).await;
+                }
             }
             Ok(oi)
         }
@@ -1281,9 +1288,11 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
         let result = match self.handle_put_object(bucket, object, data, opts).await {
-            Ok((object_info, old_current_size)) => enqueue_transition_after_write(Ok(object_info), LcEventSrc::S3PutObject)
-                .await
-                .map(|object_info| (object_info, old_current_size)),
+            Ok((object_info, old_current_size)) => {
+                enqueue_transition_after_write(self, Ok(object_info), LcEventSrc::S3PutObject, opts)
+                    .await
+                    .map(|object_info| (object_info, old_current_size))
+            }
             Err(err) => Err(err),
         };
         if result.is_ok() {
@@ -1356,9 +1365,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
         dst_opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         let result = enqueue_transition_after_write(
+            self,
             self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
                 .await,
             LcEventSrc::S3CopyObject,
+            dst_opts,
         )
         .await;
         if result.is_ok() {
@@ -1625,10 +1636,12 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for ECStore {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         let result = enqueue_transition_after_write(
+            self.as_ref(),
             self.clone()
                 .handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
                 .await,
             LcEventSrc::S3CompleteMultipartUpload,
+            opts,
         )
         .await;
         if result.is_ok() {

crates/ecstore/src/store/multipart.rs

@@ -355,7 +355,7 @@ impl ECStore {
         if let Some(guard) = guard.as_ref() {
             opts.add_bucket_lifecycle_lock_guard(guard);
         }
-        let current = crate::bucket::metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+        let current = crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
         if opts.expected_bucket_incarnation_id != Some(current) {
             return Err(StorageError::BucketNotFound(bucket.to_string()));
         }
@@ -1047,6 +1047,7 @@ impl ECStore {
         opts: &ObjectOptions,
         publication_fence: Option<RemoteTuplePublicationFence>,
     ) -> Result<ObjectInfo> {
+        let request_opts = opts;
         let (target_pool_idx, mutation_fence) = target;
         check_complete_multipart_args(bucket, object, upload_id)?;
         if !opts.data_movement {
@@ -1132,7 +1133,8 @@ impl ECStore {
             )
             .await;
         drop(publication_guard);
-        let result = enqueue_transition_after_write(result, LcEventSrc::S3CompleteMultipartUpload).await;
+        let result =
+            enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, request_opts).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self.as_ref(), bucket).await;
         }

crates/ecstore/src/store/object.rs

@@ -29,8 +29,8 @@ use crate::bucket::lifecycle::{
     },
 };
 use crate::bucket::metadata_sys::{
-    acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_in, get_cached_bucket_incarnation_id_in,
-    get_object_lock_config_and_incarnation_from_disk_in,
+    acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_for_options_in,
+    get_cached_bucket_incarnation_id_in, get_object_lock_config_and_incarnation_from_disk_in,
 };
 use crate::bucket::object_lock::objectlock_sys::{
     check_object_lock_for_deletion_with_state, ensure_recursive_force_delete_allowed_for_state,
@@ -2837,7 +2837,7 @@ impl ECStore {
             config_revision,
             state,
             lifecycle_fence.clone(),
-            metadata_guard,
+            Arc::new(metadata_guard),
         )))
     }
 
@@ -4296,6 +4296,7 @@ impl ECStore {
         if !opts.data_movement {
             return Err(Error::other("data movement PUT requires data_movement options"));
         }
+        let request_opts = opts;
         let (object, mut opts) = self.prepare_put_object(bucket, object, opts).await?;
         ensure_decommission_capacity_mutation_id(bucket, &object, &mut opts);
         let idx = self
@@ -4338,7 +4339,7 @@ impl ECStore {
                 },
             )
             .await;
-        let result = enqueue_transition_after_write(result, LcEventSrc::S3PutObject).await;
+        let result = enqueue_transition_after_write(self, result, LcEventSrc::S3PutObject, request_opts).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self, bucket).await;
         }
@@ -4445,7 +4446,7 @@ impl ECStore {
         };
         let current_bucket_incarnation_id = if let Some(guard) = _bucket_lifecycle_guard.as_ref() {
             dst_opts.add_bucket_lifecycle_lock_guard(guard);
-            let current_incarnation_id = get_bucket_incarnation_id_in(&self.ctx, dst_bucket).await?;
+            let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, dst_bucket, &dst_opts).await?;
             if dst_opts
                 .expected_bucket_incarnation_id
                 .is_some_and(|expected| expected != current_incarnation_id)
@@ -4564,6 +4565,7 @@ impl ECStore {
                     namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
                     bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
                     object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
+                    quota_admission: dst_opts.quota_admission,
                     ..Default::default()
                 };
                 if !self.single_pool() {
@@ -4606,6 +4608,7 @@ impl ECStore {
                         namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
                         bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
                         object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
+                        quota_admission: dst_opts.quota_admission,
                         ..Default::default()
                     };
                     if !self.single_pool() {
@@ -4658,6 +4661,7 @@ impl ECStore {
             namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
             bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
             object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
+            quota_admission: dst_opts.quota_admission,
             ..Default::default()
         };
         if !self.single_pool() {
@@ -4823,22 +4827,26 @@ impl ECStore {
             get_cached_bucket_incarnation_id_in(&self.ctx, bucket).await?;
         }
         let _object_lock_metadata_guard = if !is_meta_bucketname(bucket) {
-            Some(acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?)
+            Some(Arc::new(acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?))
         } else {
             None
         };
         if let Some(guard) = _object_lock_metadata_guard.as_ref() {
             opts.add_namespace_lock_guard(guard);
         }
-        let current_bucket_incarnation_id = if _object_lock_metadata_guard.is_some() {
+        let current_bucket_incarnation_id = if let Some(metadata_guard) = _object_lock_metadata_guard.as_ref() {
             let (state, incarnation_id, config_revision) =
                 get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await?;
-            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket(
+            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket_under_lifecycle_fence(
                 self.id,
                 bucket,
                 incarnation_id,
                 config_revision,
                 state,
+                opts.bucket_lifecycle_lock_fence
+                    .clone()
+                    .unwrap_or_else(NamespaceLockFence::new),
+                Arc::clone(metadata_guard),
             )));
             Some(incarnation_id)
         } else {
@@ -5257,26 +5265,32 @@ impl ECStore {
         let _object_lock_metadata_guard = if is_meta_bucketname(bucket) {
             None
         } else {
-            Some(match acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await {
-                Ok(guard) => guard,
-                Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
-            })
+            Some(Arc::new(
+                match acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await {
+                    Ok(guard) => guard,
+                    Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
+                },
+            ))
         };
         if let Some(guard) = _object_lock_metadata_guard.as_ref() {
             opts.add_namespace_lock_guard(guard);
         }
-        let current_bucket_incarnation_id = if _object_lock_metadata_guard.is_some() {
+        let current_bucket_incarnation_id = if let Some(metadata_guard) = _object_lock_metadata_guard.as_ref() {
             let (state, incarnation_id, config_revision) =
                 match get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await {
                     Ok(snapshot) => snapshot,
                     Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
                 };
-            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket(
+            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket_under_lifecycle_fence(
                 self.id,
                 bucket,
                 incarnation_id,
                 config_revision,
                 state,
+                opts.bucket_lifecycle_lock_fence
+                    .clone()
+                    .unwrap_or_else(NamespaceLockFence::new),
+                Arc::clone(metadata_guard),
             )));
             Some(incarnation_id)
         } else {
@@ -5617,7 +5631,7 @@ impl ECStore {
             opts.add_bucket_lifecycle_lock_guard(guard);
         }
         if !is_meta_bucketname(bucket) {
-            let current_incarnation_id = get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+            let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
             if opts.expected_bucket_incarnation_id != Some(current_incarnation_id) {
                 return Err(StorageError::BucketNotFound(bucket.to_string()));
             }
@@ -5690,7 +5704,7 @@ impl ECStore {
             None
         } else {
             let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
-            let current_incarnation_id = get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+            let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
             if opts
                 .expected_bucket_incarnation_id
                 .is_some_and(|expected| expected != current_incarnation_id)

crates/ecstore/tests/metadata_read_lock_reuse_test.rs

@@ -0,0 +1,288 @@
+// 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.
+
+#![cfg(feature = "test-util")]
+
+mod storage_api;
+
+use std::sync::Arc;
+use std::time::Duration;
+use storage_api::metadata_lock::{
+    BucketOperations, CompletePart, Error, MakeBucketOptions, MultipartOperations, NamespaceLocking, ObjectIO, ObjectOperations,
+    ObjectOptions, PutObjReader, PutObjectCommitBarrier, PutObjectCommitPause, init_bucket_metadata_sys,
+    isolated_store_over_temp_disks,
+};
+use tokio::io::AsyncReadExt;
+use tokio::time::timeout;
+use uuid::Uuid;
+
+#[tokio::test]
+async fn replica_put_completes_while_metadata_writer_waits_for_its_snapshot() {
+    replica_write_with_waiting_metadata_writer(false).await;
+}
+
+#[tokio::test]
+async fn replica_multipart_completes_while_metadata_writer_waits_for_its_snapshot() {
+    replica_write_with_waiting_metadata_writer(true).await;
+}
+
+async fn replica_write_with_waiting_metadata_writer(multipart: bool) {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "replica-metadata-read-reuse";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    store
+        .update_bucket_metadata_config(
+            bucket,
+            "lifecycle.xml",
+            br#"<LifecycleConfiguration><Rule><ID>expire-old-versions</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule></LifecycleConfiguration>"#.to_vec(),
+        )
+        .await
+        .expect("configure post-write lifecycle evaluation");
+    let opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        replication_request: true,
+        expected_bucket_incarnation_id: Some(store.bucket_incarnation_id(bucket).await.expect("load incarnation")),
+        object_lock_config_snapshot: Some(
+            store
+                .object_lock_config_snapshot(bucket)
+                .await
+                .expect("capture Object Lock snapshot"),
+        ),
+        ..Default::default()
+    };
+    let upload = if multipart {
+        let upload = store
+            .new_multipart_upload(bucket, "object", &opts)
+            .await
+            .expect("stage upload");
+        let mut data = PutObjReader::from_vec(b"replicated version".to_vec());
+        let part = store
+            .put_object_part(bucket, "object", &upload.upload_id, 1, &mut data, &opts)
+            .await
+            .expect("stage part");
+        Some((
+            upload.upload_id,
+            vec![CompletePart {
+                part_num: 1,
+                etag: part.etag,
+                ..Default::default()
+            }],
+        ))
+    } else {
+        None
+    };
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create metadata writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    // Leave the writer registered while the replica continues. A second read
+    // acquisition would queue behind this writer, which needs our first read.
+    assert!(futures::poll!(&mut writer).is_pending());
+    timeout(Duration::from_secs(2), async {
+        if let Some((upload_id, parts)) = upload {
+            store
+                .clone()
+                .complete_multipart_upload(bucket, "object", &upload_id, parts, &opts)
+                .await
+        } else {
+            let mut data = PutObjReader::from_vec(b"replicated version".to_vec());
+            store.put_object(bucket, "object", &mut data, &opts).await
+        }
+    })
+    .await
+    .expect("replica must reuse its metadata read lock instead of waiting behind the writer")
+    .expect("replica PUT should commit");
+    assert!(futures::poll!(&mut writer).is_pending(), "the snapshot must still fence metadata updates");
+    let read_opts = ObjectOptions {
+        version_id: opts.version_id.clone(),
+        ..Default::default()
+    };
+    drop(opts);
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("metadata writer must proceed after the snapshot drops")
+        .expect("acquire metadata writer");
+    let mut reader = store
+        .get_object_reader(bucket, "object", None, Default::default(), &read_opts)
+        .await
+        .expect("read committed replica version");
+    let mut bytes = Vec::new();
+    reader.stream.read_to_end(&mut bytes).await.expect("read replica body");
+    assert_eq!(bytes, b"replicated version");
+}
+
+#[tokio::test]
+async fn reused_metadata_lock_still_enforces_quota() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-quota";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    store
+        .update_bucket_metadata_config(bucket, "quota.json", br#"{"quota":1}"#.to_vec())
+        .await
+        .expect("set quota");
+    let mut opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        expected_bucket_incarnation_id: Some(store.bucket_incarnation_id(bucket).await.expect("load incarnation")),
+        object_lock_config_snapshot: Some(store.object_lock_config_snapshot(bucket).await.expect("capture snapshot")),
+        ..Default::default()
+    };
+    assert!(opts.set_quota_admission(0, 1));
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    assert!(futures::poll!(&mut writer).is_pending());
+    let mut data = PutObjReader::from_vec(b"too large".to_vec());
+    let error = timeout(Duration::from_secs(2), store.put_object(bucket, "object", &mut data, &opts))
+        .await
+        .expect("quota admission must not reacquire the lock")
+        .expect_err("quota must reject growth");
+    assert!(matches!(error, Error::QuotaExceeded { limit: 1, .. }), "{error:?}");
+    drop(opts);
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("release metadata reader after denial")
+        .expect("acquire writer");
+}
+
+#[tokio::test]
+async fn cancelled_put_releases_shared_metadata_guard() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-cancel";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        object_lock_config_snapshot: Some(store.object_lock_config_snapshot(bucket).await.expect("capture snapshot")),
+        ..Default::default()
+    };
+    let barrier = PutObjectCommitBarrier::install(bucket, "object", PutObjectCommitPause::AfterQuotaReservation);
+    let put_store = Arc::clone(&store);
+    let put = tokio::spawn(async move {
+        let mut data = PutObjReader::from_vec(b"cancel before commit".to_vec());
+        put_store.put_object(bucket, "object", &mut data, &opts).await
+    });
+    barrier.wait_until_paused().await;
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    assert!(futures::poll!(&mut writer).is_pending());
+    put.abort();
+    assert!(put.await.expect_err("PUT should be cancelled").is_cancelled());
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("all metadata reader owners must drop after cancellation")
+        .expect("acquire writer");
+}
+
+#[tokio::test]
+async fn multipart_rejects_metadata_snapshots_from_another_scope() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let (_other_dirs, other_store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&other_store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-scope";
+    for target in [&store, &other_store] {
+        target
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+    }
+    store
+        .make_bucket("other-bucket", &MakeBucketOptions::default())
+        .await
+        .expect("create other bucket");
+    let other_store_incarnation = other_store
+        .bucket_incarnation_id(bucket)
+        .await
+        .expect("load other store incarnation");
+    let other_bucket_incarnation = store
+        .bucket_incarnation_id("other-bucket")
+        .await
+        .expect("load other bucket incarnation");
+    for (snapshot_store, snapshot_bucket, expected) in [
+        (&other_store, bucket, other_store_incarnation),
+        (&store, "other-bucket", other_bucket_incarnation),
+        (&store, bucket, Uuid::new_v4()),
+    ] {
+        let opts = ObjectOptions {
+            expected_bucket_incarnation_id: Some(expected),
+            object_lock_config_snapshot: Some(
+                snapshot_store
+                    .object_lock_config_snapshot(snapshot_bucket)
+                    .await
+                    .expect("capture snapshot"),
+            ),
+            ..Default::default()
+        };
+        let error = store
+            .new_multipart_upload(bucket, "object", &opts)
+            .await
+            .expect_err("foreign snapshot must be rejected");
+        assert!(error.to_string().contains("valid metadata transaction fence"), "{error:?}");
+    }
+}
+
+#[tokio::test]
+async fn delete_prefix_reuses_its_metadata_snapshot_for_generation_validation() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-delete";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let mut data = PutObjReader::from_vec(b"remove prefix".to_vec());
+    store
+        .put_object(bucket, "prefix/object", &mut data, &ObjectOptions::default())
+        .await
+        .expect("seed object");
+    store
+        .delete_object(
+            bucket,
+            "prefix/",
+            ObjectOptions {
+                delete_prefix: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("delete prefix with an internally captured metadata snapshot");
+    let error = store
+        .get_object_info(bucket, "prefix/object", &ObjectOptions::default())
+        .await
+        .expect_err("prefix object must be deleted");
+    assert!(matches!(error, Error::ObjectNotFound(..)), "{error:?}");
+}

crates/ecstore/tests/storage_api.rs

@@ -6,6 +6,17 @@ pub(crate) use rustfs_ecstore::api::erasure::Erasure;
 pub(crate) use rustfs_ecstore::api::object::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
 pub(crate) use rustfs_ecstore::api::{error::Error, set_disk::SetDisks, storage::ECStore};
 use rustfs_storage_api as storage_contracts;
+#[cfg(feature = "test-util")]
+pub(crate) mod metadata_lock {
+    pub(crate) use super::storage_contracts::{
+        BucketOperations, CompletePart, MakeBucketOptions, MultipartOperations, NamespaceLocking, ObjectIO, ObjectOperations,
+    };
+    pub(crate) use super::{Error, ObjectOptions, PutObjReader};
+    pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{
+        init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks,
+    };
+    pub(crate) use rustfs_ecstore::api::set_disk::test_util::{PutObjectCommitBarrier, PutObjectCommitPause};
+}
 
 pub(crate) mod contract_compat {
     pub(crate) use super::storage_contracts::{

crates/heal/src/heal/outcome.rs

@@ -105,13 +105,17 @@ pub struct HealObjectReceipt {
 
 impl HealObjectReceipt {
     pub(crate) fn verified_for(&self, expected: &HealObjectIdentity) -> bool {
-        matches!(
-            self.disposition,
+        let disposition_verifies = match self.disposition {
             HealObjectDisposition::Repaired
-                | HealObjectDisposition::VerifiedHealthy
-                | HealObjectDisposition::MetadataHealthy
-                | HealObjectDisposition::AuthoritativelyAbsent
-        ) && self.identity.kind == expected.kind
+            | HealObjectDisposition::VerifiedHealthy
+            | HealObjectDisposition::AuthoritativelyAbsent => true,
+            // A metadata/presence proof never certifies payload bytes, so it
+            // cannot discharge a request that exists to decode the payload.
+            HealObjectDisposition::MetadataHealthy => expected.kind != HealObjectKind::Decode,
+            _ => false,
+        };
+        disposition_verifies
+            && self.identity.kind == expected.kind
             && self.identity.bucket == expected.bucket
             && self.identity.object == expected.object
             && self.identity.version_id == expected.version_id
@@ -530,6 +534,32 @@ mod canonical_outcome_tests {
         assert!(serde_json::from_value::<HealTaskOutcome>(value).is_err());
     }
 
+    #[test]
+    fn metadata_health_receipt_never_discharges_a_payload_decode_request() {
+        let incarnation = Uuid::new_v4();
+        let mut expected = HealObjectIdentity {
+            bucket_incarnation_id: Some(incarnation),
+            ..item(HealObjectDisposition::Unknown).identity
+        };
+        let mut receipt = HealObjectReceipt {
+            identity: expected.clone(),
+            disposition: HealObjectDisposition::MetadataHealthy,
+        };
+        assert!(
+            receipt.verified_for(&expected),
+            "a presence proof still settles a metadata-level object request"
+        );
+
+        expected.kind = HealObjectKind::Decode;
+        receipt.identity.kind = HealObjectKind::Decode;
+        assert!(
+            !receipt.verified_for(&expected),
+            "a presence-only proof must not clear a payload decode responsibility"
+        );
+        receipt.disposition = HealObjectDisposition::VerifiedHealthy;
+        assert!(receipt.verified_for(&expected));
+    }
+
     #[test]
     fn canonical_outcome_categories_have_one_terminal_count() {
         let mut outcome = HealTaskOutcome::default();

crates/obs/Cargo.toml

@@ -74,11 +74,11 @@ hotpath-cpu = [
 # Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the
 # build script fails the compile when that flag is missing. Off by default so
 # ordinary builds neither pay for nor depend on Tokio's unstable API.
-dial9 = ["dep:dial9-tokio-telemetry", "dial9-tokio-telemetry/process-resource"]
+dial9 = ["dep:dial9"]
 #
 # NOTE: there is deliberately no `dial9-taskdump` feature. dial9 only captures a
 # task dump for futures it wrapped itself, i.e. those spawned via
-# `dial9_tokio_telemetry::spawn`. RustFS spawns with `tokio::spawn` throughout,
+# `dial9::spawn`. RustFS spawns with `tokio::spawn` throughout,
 # so enabling `tokio/taskdump` would cost a Linux-only build constraint and
 # record nothing. Measured on an identical workload: 0 dumps via `tokio::spawn`,
 # 14709 via `dial9::spawn`. Re-adding this feature only makes sense together
@@ -149,7 +149,7 @@ tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "traci
 tokio = { workspace = true, features = ["sync", "rt-multi-thread", "time", "macros"] }
 tokio-util = { workspace = true, features = ["io", "compat"] }
 url.workspace = true
-dial9-tokio-telemetry = { workspace = true, optional = true }
+dial9 = { workspace = true, optional = true, features = ["tokio", "process-resource"] }
 thiserror = { workspace = true }
 zstd = { workspace = true, features = ["zstdmt"] }
 sysinfo = { workspace = true }

crates/obs/build.rs

@@ -14,7 +14,7 @@
 
 //! Enforces that the `dial9` feature is paired with `--cfg tokio_unstable`.
 //!
-//! `dial9-tokio-telemetry` hooks Tokio's unstable runtime instrumentation. The
+//! `dial9` hooks Tokio's unstable runtime instrumentation. The
 //! flag lives in `RUSTFLAGS`, which a caller can silently clobber by exporting
 //! their own value — that used to turn telemetry off with no diagnostic. Fail
 //! the build instead.

crates/obs/src/telemetry/dial9/config.rs

@@ -14,10 +14,10 @@
 
 //! Configuration for dial9 Tokio runtime telemetry.
 //!
-//! This module carries no dependency on the `dial9-tokio-telemetry` crate, so
-//! it compiles identically whether or not the `dial9` feature is enabled. That
-//! lets callers read the configured state (and export metrics about it) from a
-//! binary that was built without telemetry support.
+//! This module carries no dependency on the optional `dial9` crate, so it
+//! compiles identically whether or not the `dial9` feature is enabled. That lets
+//! callers read the configured state (and export metrics about it) from a binary
+//! that was built without telemetry support.
 
 use super::state::dial9_runtime_state;
 use rustfs_config::{

crates/obs/src/telemetry/dial9/disabled.rs

@@ -20,7 +20,7 @@
 
 use crate::TelemetryError;
 
-/// Placeholder for `dial9_tokio_telemetry::telemetry::TelemetryGuard`.
+/// Placeholder for `dial9::TelemetryGuard`.
 ///
 /// Never constructed: [`build_traced_runtime`] always fails without the
 /// `dial9` feature.

crates/obs/src/telemetry/dial9/enabled.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! `dial9-tokio-telemetry` integration, compiled when the `dial9` feature is on.
+//! `dial9` integration, compiled when the `dial9` feature is on.
 //!
 //! Captures Tokio runtime-level events (poll start/end, worker park/unpark,
 //! task spawn/terminate) into rotating binary trace segments. This is an
@@ -23,9 +23,8 @@ use super::config::Dial9Config;
 use super::state::{dial9_runtime_state, measure_disk_usage_bytes};
 use super::{EVENT_DIAL9_STATE, LOG_COMPONENT_OBS, LOG_SUBSYSTEM_DIAL9};
 use crate::TelemetryError;
-use dial9_tokio_telemetry::telemetry::{
-    Dial9Handle, Dial9HandleTokioExt, DiskBuffer, ProcessResourceUsageConfig, RecorderPerfExt, TokioAttachOptions, recorder,
-};
+use dial9::process::ProcessResourceUsageConfig;
+use dial9::{Dial9Handle, Dial9HandleTokioExt, DiskBuffer, RecorderPerfExt, TokioAttachOptions, recorder};
 use std::time::Duration;
 use tracing::{info, warn};
 
@@ -134,7 +133,7 @@ pub fn build_traced_runtime(
         .map(|runtime| (runtime, guard, shutdown));
 
     // No task dumps here. dial9 captures a task dump only for futures it
-    // wrapped itself, i.e. those spawned via `dial9_tokio_telemetry::spawn`;
+    // wrapped itself, i.e. those spawned via `dial9::spawn`;
     // `tokio::spawn` gets no wrapper. RustFS spawns with `tokio::spawn`
     // throughout, so enabling task dumps records nothing. Measured on an
     // identical workload: 0 dumps via `tokio::spawn`, 14709 via `dial9::spawn`.

crates/obs/src/telemetry/dial9/mod.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Tokio runtime-level telemetry via `dial9-tokio-telemetry`.
+//! Tokio runtime-level telemetry via `dial9`.
 //!
 //! This is an on-demand profiler for executor-level faults — long polls that
 //! stall a worker, park/unpark storms, tasks that never yield — which are

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

@@ -39,9 +39,9 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
 | Client or engine | Status | Claim |
 |---|---|---|
 | PyIceberg | Automated | Namespace and table create, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional vended credentials with an exact-prefix data-plane scope check. |
-| Spark Iceberg REST catalog | Manual/live harness | Pinned package inputs, catalog properties, SQL, expected `row_count=2`, and a CI opt-in gate for create/append/refresh/count/cleanup. Live execution and commit-conflict probing remain manual unless enabled in the runner. |
+| Spark Iceberg REST catalog | Manual/live harness | Pinned package inputs, catalog properties, atomic CTAS SQL through staged create, expected `row_count=2`, and a CI opt-in gate for create/refresh/count/cleanup. Live execution and commit-conflict probing remain manual unless enabled in the runner. |
 | Trino Iceberg REST catalog | Manual/live harness | Catalog properties and a read-only `SELECT COUNT(*)` against a PyIceberg- or Spark-created table. Write compatibility not claimed. |
-| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` covers metadata-location read, single-table create/insert/update/delete/merge, schema evolution, snapshots, concurrent writers, drop, PyIceberg cross-read, and both signing profiles. Staged create, purge-on-drop, and format v3 are verified fail-closed. Two-table mode runs without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES` and vended-credential integration not claimed. |
+| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` covers metadata-location read, single-table create/insert/update/delete/merge, schema evolution, snapshots, concurrent writers, drop, PyIceberg cross-read, and both signing profiles. Purge-on-drop and format v3 are verified fail-closed. Two-table mode runs without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES`, staged-create client integration, and vended-credential integration are not claimed. |
 | StarRocks Iceberg REST catalog | Documented, not automated | External catalog read-path reference only. |
 | Databend | Manual/live harness | S3 stage read probe for table data files only; Iceberg REST integration not claimed. |
 | Snowflake Open Catalog / Iceberg integrations | Generated harness | Operator-adapted external volume/catalog SQL template only; live interop not claimed. |
@@ -62,7 +62,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
 | Catalog config | Supported | `GET /v1/config` advertises defaults and only the supported OpenAPI REST paths in `endpoints`; RustFS extensions (administration, maintenance, migration, diagnostics, refs, metadata-location) are not presented as standard endpoints. |
 | Table bucket discovery | Supported | `PUT` / `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. |
 | Namespaces | Supported | Create, list, load, exists, drop on both prefixes; `pageSize`/`pageToken` pagination with context-bound tokens; identifiers limited to 512 ASCII characters. |
-| Tables | Supported | Create, register, list, load, exists, rename, commit, metadata-location get/update, drop on both prefixes. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. |
+| Tables | Supported | Direct and staged create, register, list, load, exists, rename, commit, metadata-location get/update, and drop on both prefixes. Staged create is stateless until an `assert-create` commit atomically publishes the initial metadata and table entry. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. |
 | Commit CAS | Supported | Single-table commits validate base metadata, version token, referenced object existence, warehouse scope, and Iceberg requirements before advancing the pointer; external metadata transitions preserve monotonic assignment watermarks and immutable retained definitions. `idempotency-key-lifetime` is not advertised; mutation-wide `Idempotency-Key` semantics are unsupported. |
 | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. |
 | Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. |

docs/operations/dial9-runtime-profiling.md

@@ -3,9 +3,8 @@
 **Use this when:** you need Tokio runtime-level evidence (which task held a worker, long polls, park/unpark behaviour) that Prometheus metrics and `tracing` spans cannot provide, or you are building or running the opt-in `dial9` profiling binary.
 **Source of truth:** `crates/obs/src/telemetry/dial9/mod.rs` (session setup), `crates/obs/src/metrics/collectors/dial9.rs` (metrics), `crates/config/src/constants/runtime.rs` (`RUSTFS_RUNTIME_DIAL9_*` and defaults), `.config/make/build.mak` (`build-profiling`), `crates/obs/build.rs` (feature/cfg pairing check).
 
-`dial9-tokio-telemetry` records Tokio runtime-level events — poll start/end,
-worker park/unpark, task spawn/terminate, and optionally async backtraces of
-stalled tasks — into binary trace segments.
+`dial9` records Tokio runtime-level events — poll start/end, worker park/unpark,
+task spawn/terminate — into binary trace segments.
 
 It answers questions that Prometheus metrics and `tracing` spans cannot:
 
@@ -29,7 +28,7 @@ For drive stalls use the `rustfs_io_*` metrics and the drive-stall budget. dial9
 answers a different question: which task held a worker, and for how long.
 
 **It cannot tell you where a task was stuck.** That would need a task dump, and
-dial9 only captures those for futures spawned through `dial9_tokio_telemetry::spawn`.
+dial9 only captures those for futures spawned through `dial9::spawn`.
 RustFS spawns with `tokio::spawn` throughout, so no task dump is ever recorded and
 no configuration exposes one. Tracked as D9-16 in rustfs/backlog#1157.
 

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

@@ -910,6 +910,16 @@ struct RestLoadTableResponse {
     storage_credentials: Vec<RestStorageCredential>,
 }
 
+#[derive(Debug, Serialize)]
+struct RestCreateTableResponse {
+    #[serde(rename = "metadata-location")]
+    metadata_location: Option<String>,
+    metadata: serde_json::Value,
+    config: BTreeMap<String, String>,
+    #[serde(rename = "storage-credentials")]
+    storage_credentials: Vec<RestStorageCredential>,
+}
+
 #[derive(Debug, Serialize)]
 struct RestLoadCredentialsResponse {
     config: BTreeMap<String, String>,
@@ -2706,6 +2716,25 @@ fn load_table_response_from_entry(entry: crate::table_catalog::TableEntry, metad
     }
 }
 
+fn create_table_response_from_entry(
+    entry: crate::table_catalog::TableEntry,
+    metadata: serde_json::Value,
+    staged: bool,
+) -> RestCreateTableResponse {
+    let RestLoadTableResponse {
+        metadata_location,
+        metadata,
+        config,
+        storage_credentials,
+    } = load_table_response_from_entry(entry, metadata);
+    RestCreateTableResponse {
+        metadata_location: (!staged).then_some(metadata_location),
+        metadata,
+        config,
+        storage_credentials,
+    }
+}
+
 fn load_view_response_from_entry(entry: crate::table_catalog::ViewEntry, metadata: serde_json::Value) -> RestLoadViewResponse {
     let mut config = BTreeMap::new();
     let warehouse_location = entry.warehouse_location.clone();
@@ -3104,16 +3133,9 @@ fn table_entry_from_create_table_request(
         mut schema,
         mut partition_spec,
         mut write_order,
-        stage_create,
+        stage_create: _,
         mut properties,
     } = request;
-    if stage_create {
-        return Err(iceberg_rest_error(
-            ICEBERG_ERROR_UNSUPPORTED_OPERATION,
-            StatusCode::NOT_ACCEPTABLE,
-            "stage-create is not supported",
-        ));
-    }
 
     let table = crate::table_catalog::IdentifierSegment::parse(name)
         .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
@@ -3490,6 +3512,26 @@ fn validate_table_commit_requirements(metadata: &serde_json::Value, requirements
     Ok(())
 }
 
+fn request_has_assert_create_requirement(request: &RestCommitTableRequest) -> bool {
+    request
+        .requirements
+        .iter()
+        .any(|requirement| requirement.get("type").and_then(serde_json::Value::as_str) == Some("assert-create"))
+}
+
+fn validate_create_table_commit_requirements(requirements: &[serde_json::Value]) -> S3Result<()> {
+    if requirements.is_empty()
+        || requirements
+            .iter()
+            .any(|requirement| requirement.get("type").and_then(serde_json::Value::as_str) != Some("assert-create"))
+    {
+        return Err(S3Error::from(ApiError::invalid_request(
+            "create table commit requires only assert-create requirements",
+        )));
+    }
+    Ok(())
+}
+
 fn validate_i64_requirement(
     metadata: &serde_json::Value,
     requirement: &serde_json::Value,
@@ -3555,20 +3597,104 @@ fn apply_table_commit_updates(
 }
 
 fn apply_table_commit_updates_at(
-    mut metadata: serde_json::Value,
+    metadata: serde_json::Value,
     updates: &[serde_json::Value],
     previous_metadata_location: &str,
     commit_timestamp_ms: i64,
+) -> S3Result<serde_json::Value> {
+    apply_table_updates_at(
+        metadata,
+        updates,
+        TableMetadataUpdateMode::Existing {
+            previous_metadata_location,
+        },
+        commit_timestamp_ms,
+    )
+}
+
+fn apply_table_create_updates_at(updates: &[serde_json::Value], commit_timestamp_ms: i64) -> S3Result<serde_json::Value> {
+    let format_version = updates
+        .iter()
+        .find(|update| update.get("action").and_then(serde_json::Value::as_str) == Some("upgrade-format-version"))
+        .and_then(|update| update.get("format-version"))
+        .and_then(serde_json::Value::as_i64)
+        .unwrap_or(2);
+    let mut metadata = serde_json::json!({
+        "format-version": format_version,
+        "last-updated-ms": commit_timestamp_ms,
+        "last-column-id": 0,
+        "schemas": [],
+        "current-schema-id": -1,
+        "partition-specs": [],
+        "default-spec-id": -1,
+        "last-partition-id": 999,
+        "sort-orders": [],
+        "default-sort-order-id": -1,
+        "properties": {},
+        "current-snapshot-id": -1,
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": [],
+        "refs": {}
+    });
+    if format_version == 2 {
+        metadata_object_mut(&mut metadata)?.insert("last-sequence-number".to_string(), serde_json::Value::from(0));
+    }
+    apply_table_updates_at(metadata, updates, TableMetadataUpdateMode::Create, commit_timestamp_ms)
+}
+
+#[derive(Debug, Clone, Copy)]
+enum TableMetadataUpdateMode<'a> {
+    Create,
+    Existing { previous_metadata_location: &'a str },
+}
+
+impl TableMetadataUpdateMode<'_> {
+    fn initial_catalog_id(self, array_key: &str) -> Option<i64> {
+        match (self, array_key) {
+            (Self::Create, "schemas" | "partition-specs") => Some(0),
+            (Self::Create, "sort-orders") => Some(1),
+            (Self::Create, _) | (Self::Existing { .. }, _) => None,
+        }
+    }
+}
+
+fn apply_table_updates_at(
+    mut metadata: serde_json::Value,
+    updates: &[serde_json::Value],
+    mode: TableMetadataUpdateMode<'_>,
+    commit_timestamp_ms: i64,
 ) -> S3Result<serde_json::Value> {
     if !metadata.is_object() {
         return Err(s3_error!(InvalidRequest, "current table metadata must be a JSON object"));
     }
-    if metadata.get("format-version").is_some() {
+    if matches!(mode, TableMetadataUpdateMode::Existing { .. }) && metadata.get("format-version").is_some() {
         crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?;
     }
-    let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id")?;
-    let mut next_spec_id = next_catalog_id_for_updates(&metadata, updates, "add-spec", "partition-specs", "spec-id")?;
-    let mut next_sort_order_id = next_catalog_id_for_updates(&metadata, updates, "add-sort-order", "sort-orders", "order-id")?;
+    let mut next_schema_id = next_catalog_id_for_updates(
+        &metadata,
+        updates,
+        "add-schema",
+        "schemas",
+        "schema-id",
+        mode.initial_catalog_id("schemas"),
+    )?;
+    let mut next_spec_id = next_catalog_id_for_updates(
+        &metadata,
+        updates,
+        "add-spec",
+        "partition-specs",
+        "spec-id",
+        mode.initial_catalog_id("partition-specs"),
+    )?;
+    let mut next_sort_order_id = next_catalog_id_for_updates(
+        &metadata,
+        updates,
+        "add-sort-order",
+        "sort-orders",
+        "order-id",
+        mode.initial_catalog_id("sort-orders"),
+    )?;
     let mut last_added_schema_id = None;
     let mut last_added_spec_id = None;
     let mut last_added_sort_order_id = None;
@@ -3658,8 +3784,16 @@ fn apply_table_commit_updates_at(
     if metadata.get("format-version").is_some() {
         crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?;
     }
-    append_previous_metadata_log(&mut metadata, previous_metadata_location)?;
+    if let TableMetadataUpdateMode::Existing {
+        previous_metadata_location,
+    } = mode
+    {
+        append_previous_metadata_log(&mut metadata, previous_metadata_location)?;
+    }
     metadata_object_mut(&mut metadata)?.insert("last-updated-ms".to_string(), serde_json::Value::from(commit_timestamp_ms));
+    if matches!(mode, TableMetadataUpdateMode::Create) {
+        crate::table_catalog::validate_supported_table_metadata(&metadata).map_err(catalog_store_error)?;
+    }
     Ok(metadata)
 }
 
@@ -3726,7 +3860,7 @@ fn apply_view_commit_updates_at(
     if !metadata.is_object() {
         return Err(s3_error!(InvalidRequest, "current view metadata must be a JSON object"));
     }
-    let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id")?;
+    let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id", None)?;
     let mut last_added_schema_id = None;
     let mut last_added_view_version_id = None;
     let mut added_view_version_timestamps = BTreeMap::new();
@@ -5084,11 +5218,12 @@ fn next_catalog_id_for_updates(
     action: &str,
     array_key: &str,
     id_key: &str,
+    initial_id: Option<i64>,
 ) -> S3Result<Option<i64>> {
     updates
         .iter()
         .any(|update| update.get("action").and_then(serde_json::Value::as_str) == Some(action))
-        .then(|| next_array_object_i64(metadata, array_key, id_key))
+        .then(|| next_array_object_i64(metadata, array_key, id_key, initial_id))
         .transpose()
 }
 
@@ -5103,10 +5238,22 @@ fn take_catalog_assigned_id(next_id: &mut Option<i64>, label: &str) -> S3Result<
     Ok(assigned_id)
 }
 
-fn next_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str) -> S3Result<i64> {
-    last_array_object_i64(metadata, array_key, id_key)?
-        .checked_add(1)
-        .ok_or_else(|| s3_error!(InvalidRequest, "metadata field {array_key} {id_key} exceeds the signed 64-bit range"))
+fn next_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str, initial_id: Option<i64>) -> S3Result<i64> {
+    match last_array_object_i64(metadata, array_key, id_key) {
+        Ok(last_id) => last_id
+            .checked_add(1)
+            .ok_or_else(|| s3_error!(InvalidRequest, "metadata field {array_key} {id_key} exceeds the signed 64-bit range")),
+        Err(_)
+            if metadata
+                .get(array_key)
+                .and_then(serde_json::Value::as_array)
+                .is_some_and(Vec::is_empty) =>
+        {
+            initial_id
+                .ok_or_else(|| S3Error::from(ApiError::invalid_request(format!("metadata field {array_key} has no {id_key}"))))
+        }
+        Err(err) => Err(err),
+    }
 }
 
 fn last_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str) -> S3Result<i64> {
@@ -5435,11 +5582,40 @@ async fn create_table_response<S>(
     namespace: &crate::table_catalog::Namespace,
     request: CreateTableRequest,
     table_bucket_enabled: bool,
-) -> S3Result<RestLoadTableResponse>
+) -> S3Result<RestCreateTableResponse>
 where
     S: crate::table_catalog::TableCatalogStore + ?Sized,
 {
+    let staged = request.stage_create;
     let (entry, metadata) = table_entry_from_create_table_request(bucket, namespace, request)?;
+    if staged {
+        if !table_bucket_enabled {
+            return Err(S3Error::from(ApiError::invalid_request(format!("bucket {bucket} is not table-enabled"))));
+        }
+        get_namespace_response(store, bucket, namespace).await?;
+        if store
+            .load_table(bucket, &namespace.public_name(), &entry.table)
+            .await
+            .map_err(catalog_store_error)?
+            .is_some()
+            || store
+                .load_view(bucket, &namespace.public_name(), &entry.table)
+                .await
+                .map_err(catalog_store_error)?
+                .is_some()
+        {
+            return Err(iceberg_rest_error(
+                ICEBERG_ERROR_ALREADY_EXISTS,
+                StatusCode::CONFLICT,
+                format!("catalog object already exists: {}/{}/{}", bucket, namespace.public_name(), entry.table),
+            ));
+        }
+        store
+            .ensure_table_warehouse_location_available(&entry)
+            .await
+            .map_err(catalog_store_already_exists_error)?;
+        return Ok(create_table_response_from_entry(entry, metadata, true));
+    }
     ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?;
     crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket)
         .await
@@ -5465,7 +5641,7 @@ where
         .register_table_with_publication(entry.clone(), metadata_backend)
         .await
         .map_err(catalog_store_already_exists_error)?;
-    Ok(load_table_response_from_entry(entry, metadata))
+    Ok(create_table_response_from_entry(entry, metadata, false))
 }
 
 async fn create_view_response<S>(
@@ -6080,11 +6256,14 @@ async fn standard_commit_table_response<S>(
 where
     S: crate::table_catalog::TableCatalogStore + ?Sized,
 {
-    let Some(current) = store
+    let current = store
         .load_table(bucket, &namespace.public_name(), table)
         .await
-        .map_err(catalog_store_error)?
-    else {
+        .map_err(catalog_store_error)?;
+    let Some(current) = current else {
+        if request_has_assert_create_requirement(&request) {
+            return commit_staged_table_create_response(store, metadata_backend, bucket, namespace, table, request).await;
+        }
         return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found"));
     };
     if let Some(response) =
@@ -6194,6 +6373,104 @@ where
     Ok(commit_table_response_from_result(result, next_metadata))
 }
 
+async fn commit_staged_table_create_response<S>(
+    store: &S,
+    metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    table: &str,
+    request: RestCommitTableRequest,
+) -> S3Result<RestCommitTableResponse>
+where
+    S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
+    validate_create_table_commit_requirements(&request.requirements)?;
+    get_namespace_response(store, bucket, namespace).await?;
+    if store
+        .load_view(bucket, &namespace.public_name(), table)
+        .await
+        .map_err(catalog_store_error)?
+        .is_some()
+    {
+        return Err(iceberg_rest_error(
+            ICEBERG_ERROR_ALREADY_EXISTS,
+            StatusCode::CONFLICT,
+            format!("catalog object already exists: {}/{}/{}", bucket, namespace.public_name(), table),
+        ));
+    }
+    let commit_timestamp_ms = current_time_millis();
+    let metadata = apply_table_create_updates_at(&request.updates, commit_timestamp_ms)?;
+    validate_metadata_table_location_in_bucket(bucket, &metadata)?;
+
+    let table = crate::table_catalog::IdentifierSegment::parse(table.to_string())
+        .map_err(|err| S3Error::from(ApiError::invalid_request(format!("invalid table name: {err}"))))?;
+    let (commit_id, _) = standard_commit_ids(request.commit_id.or(request.idempotency_key));
+    let table_id = Uuid::new_v4().to_string();
+    let metadata_location =
+        crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
+    let properties = serde_json::from_value::<BTreeMap<String, String>>(
+        metadata
+            .get("properties")
+            .cloned()
+            .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())),
+    )
+    .map_err(|_| S3Error::from(ApiError::invalid_request("table property values must be strings")))?;
+    let mut entry = crate::table_catalog::TableEntry {
+        version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
+        table_bucket: bucket.to_string(),
+        namespace: namespace.public_name(),
+        table: table.as_str().to_string(),
+        table_id,
+        table_uuid: Uuid::new_v4().to_string(),
+        format: "ICEBERG".to_string(),
+        format_version: 1,
+        warehouse_location: metadata_table_location(&metadata)?.to_string(),
+        metadata_location,
+        version_token: format!("token-{}", Uuid::new_v4()),
+        generation: 1,
+        state: crate::table_catalog::TableCatalogEntryState::Active,
+        properties,
+        created_at: None,
+        updated_at: None,
+    };
+    adopt_registered_metadata_identity(&mut entry, &metadata)?;
+    validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?;
+
+    crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket)
+        .await
+        .map_err(catalog_store_error)?;
+    if !crate::table_catalog::TableCommitPublication::holds_table_bucket(metadata_backend, bucket) {
+        return Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::Internal(
+            "staged table creation requires a table-bucket publication fence".to_string(),
+        )));
+    }
+    let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
+    let metadata_data = serde_json::to_vec(&metadata).map_err(|err| {
+        S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize initial table metadata: {err}"))
+    })?;
+    metadata_backend
+        .put_object(
+            bucket,
+            &entry.metadata_location,
+            metadata_data,
+            crate::table_catalog::TableCatalogPutPrecondition::IfAbsent,
+        )
+        .await
+        .map_err(catalog_store_already_exists_error)?;
+    store
+        .register_table_with_publication(entry.clone(), metadata_backend)
+        .await
+        .map_err(catalog_store_already_exists_error)?;
+
+    Ok(RestCommitTableResponse {
+        metadata_location: table_metadata_location_for_client(bucket, &entry.metadata_location),
+        metadata: table_metadata_for_client(bucket, metadata),
+        version_token: entry.version_token,
+        generation: entry.generation,
+        commit_id,
+    })
+}
+
 async fn table_commit_for_retry<S>(
     store: &S,
     bucket: &str,

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

@@ -203,6 +203,10 @@ impl Operation for RestCommitTableHandler {
         install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let request = read_rest_commit_table_request(std::mem::take(&mut req.input)).await?;
+        if request_has_assert_create_requirement(&request) {
+            let namespace_resource = TableCatalogResource::namespace(&warehouse, &namespace);
+            authorize_table_catalog_resource_request(&req, &namespace_resource, AdminAction::CreateTableAction).await?;
+        }
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);

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

@@ -668,6 +668,11 @@ fn table_catalog_handlers_require_table_admin_actions() {
         );
     }
 
+    let commit_table_block = operation_block(&src, "RestCommitTableHandler");
+    assert!(commit_table_block.contains("request_has_assert_create_requirement(&request)"));
+    assert!(commit_table_block.contains("TableCatalogResource::namespace(&warehouse, &namespace)"));
+    assert!(commit_table_block.contains("AdminAction::CreateTableAction"));
+
     let sync_bridge_block = operation_block(&src, "SyncExternalCatalogBridgeHandler");
     assert!(
         sync_bridge_block.contains("AdminAction::RegisterTableAction"),
@@ -2825,7 +2830,7 @@ async fn commit_request_readers_require_standard_arrays_and_preserve_legacy_poin
 }
 
 #[test]
-fn unsupported_create_and_register_modes_return_iceberg_errors() {
+fn register_overwrite_returns_iceberg_unsupported_error() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
     let register_error = table_entry_from_register_request(
         "warehouse",
@@ -2839,17 +2844,6 @@ fn unsupported_create_and_register_modes_return_iceberg_errors() {
     .expect_err("register overwrite should remain unsupported");
     assert_eq!(register_error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into()));
     assert_eq!(register_error.status_code(), Some(StatusCode::NOT_ACCEPTABLE));
-
-    let create_request: CreateTableRequest = serde_json::from_value(serde_json::json!({
-        "name": "events",
-        "schema": {"type": "struct", "schema-id": 0, "fields": []},
-        "stage-create": true
-    }))
-    .expect("stage-create request should parse");
-    let create_error = table_entry_from_create_table_request("warehouse", &namespace, create_request)
-        .expect_err("staged create should remain unsupported");
-    assert_eq!(create_error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into()));
-    assert_eq!(create_error.status_code(), Some(StatusCode::NOT_ACCEPTABLE));
 }
 
 #[test]
@@ -2973,10 +2967,10 @@ async fn create_table_response_writes_initial_metadata_for_standard_request() {
         .expect("table should exist");
     assert_eq!(
         response.metadata_location,
-        format!(
+        Some(format!(
             "s3://warehouse/.rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001-{}.metadata.json",
             entry.table_id
-        )
+        ))
     );
     assert_eq!(response.metadata["table-uuid"], entry.table_uuid);
     assert!(
@@ -3380,7 +3374,7 @@ async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct
     let winner_entry = &tables[0];
     assert_eq!(
         winner.metadata_location,
-        table_metadata_location_for_client("warehouse", &winner_entry.metadata_location)
+        Some(table_metadata_location_for_client("warehouse", &winner_entry.metadata_location))
     );
     let metadata_prefix = winner_entry
         .metadata_location
@@ -3415,6 +3409,423 @@ async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct
     assert!(table_uuids.contains(&winner_entry.table_uuid));
 }
 
+#[tokio::test]
+async fn staged_create_is_invisible_until_assert_create_commit_across_backings() {
+    for mode in [
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        crate::table_catalog::TableCatalogBackingMode::DurableStrong,
+    ] {
+        let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+        let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(metadata_backend.clone(), mode);
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+        ensure_table_bucket_entry(&store, "warehouse", true)
+            .await
+            .expect("table bucket entry should be seeded");
+        create_namespace_response(
+            &store,
+            "warehouse",
+            CreateNamespaceRequest {
+                namespace: vec!["analytics".to_string()],
+                properties: BTreeMap::new(),
+            },
+            true,
+        )
+        .await
+        .expect("namespace should be created");
+        let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
+
+        let staged =
+            create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true)
+                .await
+                .expect("stage-create should return initialized metadata");
+
+        assert_eq!(staged.metadata_location, None, "{mode:?}");
+        let staged_json = serde_json::to_value(&staged).expect("staged response should serialize");
+        assert!(staged_json["metadata-location"].is_null(), "{mode:?}");
+        assert!(
+            store
+                .load_table("warehouse", "analytics", "events")
+                .await
+                .expect("table lookup should succeed")
+                .is_none(),
+            "{mode:?}"
+        );
+        let metadata_prefix = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/";
+        assert!(
+            metadata_backend
+                .list_objects("warehouse", metadata_prefix)
+                .await
+                .expect("metadata listing should succeed")
+                .is_empty(),
+            "{mode:?}"
+        );
+
+        let committed = commit_table_response(
+            &store,
+            &commit_backend,
+            "warehouse",
+            &namespace,
+            "events",
+            staged_create_commit_request(&staged.metadata, []),
+        )
+        .await
+        .expect("assert-create commit should publish the table");
+
+        assert_eq!(committed.generation, 1, "{mode:?}");
+        assert_eq!(committed.metadata["table-uuid"], staged.metadata["table-uuid"], "{mode:?}");
+        assert_eq!(committed.metadata["metadata-log"], serde_json::json!([]), "{mode:?}");
+        let entry = store
+            .load_table("warehouse", "analytics", "events")
+            .await
+            .expect("table lookup should succeed")
+            .expect("committed table should be visible");
+        assert_eq!(entry.table_uuid, staged.metadata["table-uuid"].as_str().unwrap(), "{mode:?}");
+        assert_eq!(
+            committed.metadata_location,
+            table_metadata_location_for_client("warehouse", &entry.metadata_location),
+            "{mode:?}"
+        );
+        assert!(
+            metadata_backend
+                .object_exists("warehouse", &entry.metadata_location)
+                .await
+                .expect("metadata lookup should succeed"),
+            "{mode:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn staged_create_commit_accepts_initial_snapshot_for_spark_ctas() {
+    let store = TestTableCatalogStore::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    ensure_table_bucket_entry(&store, "warehouse", true)
+        .await
+        .expect("table bucket entry should be seeded");
+    create_namespace_response(
+        &store,
+        "warehouse",
+        CreateNamespaceRequest {
+            namespace: vec!["analytics".to_string()],
+            properties: BTreeMap::new(),
+        },
+        true,
+    )
+    .await
+    .expect("namespace should be created");
+    let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
+    let staged = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true)
+        .await
+        .expect("stage-create should succeed");
+    let table_location = staged.metadata["location"]
+        .as_str()
+        .expect("table location should be present");
+    let manifest_list = format!("{table_location}/metadata/snap-10.avro");
+    let data_file = format!("{table_location}/data/part-10.parquet");
+    seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await;
+
+    let committed = commit_table_response(
+        &store,
+        &commit_backend,
+        "warehouse",
+        &namespace,
+        "events",
+        staged_create_commit_request(
+            &staged.metadata,
+            [
+                serde_json::json!({
+                    "action": "add-snapshot",
+                    "snapshot": {
+                        "snapshot-id": 10,
+                        "sequence-number": 1,
+                        "timestamp-ms": 1234,
+                        "manifest-list": manifest_list,
+                        "summary": {"operation": "append"}
+                    }
+                }),
+                serde_json::json!({
+                    "action": "set-snapshot-ref",
+                    "ref-name": "main",
+                    "snapshot-id": 10,
+                    "type": "branch"
+                }),
+            ],
+        ),
+    )
+    .await
+    .expect("Spark CTAS-style create commit should succeed");
+
+    assert_eq!(committed.metadata["current-snapshot-id"], 10);
+    assert_eq!(committed.metadata["last-sequence-number"], 1);
+    assert_eq!(committed.metadata["refs"]["main"]["snapshot-id"], 10);
+}
+
+#[test]
+fn staged_create_commit_defaults_to_format_version_two_without_upgrade() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let (_, staged_metadata) = table_entry_from_create_table_request("warehouse", &namespace, staged_events_create_request())
+        .expect("staged create metadata should initialize");
+    let mut request = staged_create_commit_request(&staged_metadata, []);
+    request
+        .updates
+        .retain(|update| update.get("action").and_then(serde_json::Value::as_str) != Some("upgrade-format-version"));
+
+    let metadata = apply_table_create_updates_at(&request.updates, 1234)
+        .expect("create updates without an explicit upgrade should use the Iceberg default");
+
+    assert_eq!(metadata["format-version"], 2);
+    assert_eq!(metadata["last-sequence-number"], 0);
+}
+
+#[tokio::test]
+async fn staged_create_rejects_an_active_table_warehouse_location() {
+    for mode in [
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        crate::table_catalog::TableCatalogBackingMode::DurableStrong,
+    ] {
+        let catalog_backend = TestTableCatalogObjectBackend::content_addressed();
+        let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+        let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(catalog_backend, mode);
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+        ensure_table_bucket_entry(&store, "warehouse", true)
+            .await
+            .expect("table bucket entry should be seeded");
+        create_namespace_response(
+            &store,
+            "warehouse",
+            CreateNamespaceRequest {
+                namespace: vec!["analytics".to_string()],
+                properties: BTreeMap::new(),
+            },
+            true,
+        )
+        .await
+        .expect("namespace should be created");
+        let commit_backend = TableCommitObjectBackend::trusted(metadata_backend);
+        let active_request = serde_json::from_value::<CreateTableRequest>(serde_json::json!({
+            "name": "active_events",
+            "location": "s3://warehouse/shared/events",
+            "schema": {"type": "struct", "fields": []}
+        }))
+        .expect("active create request should parse");
+        create_table_response(&store, &commit_backend, "warehouse", &namespace, active_request, true)
+            .await
+            .expect("active table should be created");
+        let staged_request = serde_json::from_value::<CreateTableRequest>(serde_json::json!({
+            "name": "staged_events",
+            "location": "s3://warehouse/shared/events/child",
+            "schema": {"type": "struct", "fields": []},
+            "stage-create": true
+        }))
+        .expect("staged create request should parse");
+
+        let error = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_request, true)
+            .await
+            .expect_err("stage-create must reject a location owned by an active table");
+
+        assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_ALREADY_EXISTS.into()), "{mode:?}");
+        assert_eq!(error.status_code(), Some(StatusCode::CONFLICT), "{mode:?}");
+        assert!(
+            store
+                .load_table("warehouse", "analytics", "staged_events")
+                .await
+                .expect("table lookup should succeed")
+                .is_none(),
+            "{mode:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn concurrent_staged_create_commits_publish_exactly_one_table() {
+    for mode in [
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        crate::table_catalog::TableCatalogBackingMode::DurableStrong,
+    ] {
+        let catalog_backend = TestTableCatalogObjectBackend::content_addressed();
+        let barrier = Arc::new(tokio::sync::Barrier::new(2));
+        let metadata_backend = TestTableCatalogObjectBackend {
+            put_object_barrier: Some(Arc::clone(&barrier)),
+            ..TestTableCatalogObjectBackend::content_addressed()
+        };
+        let store = Arc::new(crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(catalog_backend, mode));
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+        ensure_table_bucket_entry(store.as_ref(), "warehouse", true)
+            .await
+            .expect("table bucket entry should be seeded");
+        create_namespace_response(
+            store.as_ref(),
+            "warehouse",
+            CreateNamespaceRequest {
+                namespace: vec!["analytics".to_string()],
+                properties: BTreeMap::new(),
+            },
+            true,
+        )
+        .await
+        .expect("namespace should be created");
+        let staged = create_table_response(
+            store.as_ref(),
+            &TableCommitObjectBackend::trusted(metadata_backend.clone()),
+            "warehouse",
+            &namespace,
+            staged_events_create_request(),
+            true,
+        )
+        .await
+        .expect("stage-create should succeed");
+        let first_store = Arc::clone(&store);
+        let first_namespace = namespace.clone();
+        let first_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
+        let first_request = staged_create_commit_request(&staged.metadata, []);
+        let first = tokio::spawn(async move {
+            commit_table_response(
+                first_store.as_ref(),
+                &first_backend,
+                "warehouse",
+                &first_namespace,
+                "events",
+                first_request,
+            )
+            .await
+        });
+        tokio::time::timeout(StdDuration::from_secs(2), async {
+            while metadata_backend.state.lock().await.objects.is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("first metadata write should reach the publication pause");
+        metadata_backend.lock_attempts.lock().await.clear();
+        let second_store = Arc::clone(&store);
+        let second_namespace = namespace.clone();
+        let second_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
+        let second_request = staged_create_commit_request(&staged.metadata, []);
+        let second = tokio::spawn(async move {
+            commit_table_response(
+                second_store.as_ref(),
+                &second_backend,
+                "warehouse",
+                &second_namespace,
+                "events",
+                second_request,
+            )
+            .await
+        });
+        metadata_backend.wait_for_lock_attempts(1).await;
+        assert!(
+            !second.is_finished(),
+            "second create must wait after observing the table as absent: {mode:?}"
+        );
+
+        barrier.wait().await;
+        let _winner = tokio::time::timeout(StdDuration::from_secs(2), first)
+            .await
+            .expect("first assert-create commit should complete")
+            .expect("first assert-create task should join")
+            .expect("first assert-create commit should win");
+        tokio::time::timeout(StdDuration::from_secs(2), async {
+            while metadata_backend.state.lock().await.objects.len() < 2 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("second metadata write should reach the publication pause");
+        barrier.wait().await;
+        let loser = tokio::time::timeout(StdDuration::from_secs(2), second)
+            .await
+            .expect("second assert-create commit should complete")
+            .expect("second assert-create task should join")
+            .expect_err("second assert-create commit must lose at atomic registration");
+
+        assert_eq!(loser.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_ALREADY_EXISTS.into()), "{mode:?}");
+        assert_eq!(loser.status_code(), Some(StatusCode::CONFLICT), "{mode:?}");
+        assert_eq!(
+            store
+                .list_tables("warehouse", "analytics")
+                .await
+                .expect("table listing should succeed")
+                .len(),
+            1,
+            "{mode:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn malformed_or_conflicting_assert_create_does_not_replace_a_table() {
+    let store = TestTableCatalogStore::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    ensure_table_bucket_entry(&store, "warehouse", true)
+        .await
+        .expect("table bucket entry should be seeded");
+    create_namespace_response(
+        &store,
+        "warehouse",
+        CreateNamespaceRequest {
+            namespace: vec!["analytics".to_string()],
+            properties: BTreeMap::new(),
+        },
+        true,
+    )
+    .await
+    .expect("namespace should be created");
+    let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
+    let malformed = serde_json::from_value(serde_json::json!({
+        "requirements": [{"type": "assert-create"}],
+        "updates": []
+    }))
+    .expect("malformed create commit should parse");
+    let error = commit_table_response(&store, &commit_backend, "warehouse", &namespace, "events", malformed)
+        .await
+        .expect_err("incomplete create updates must fail");
+    assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST));
+    assert!(
+        store
+            .load_table("warehouse", "analytics", "events")
+            .await
+            .expect("table lookup should succeed")
+            .is_none()
+    );
+
+    let staged = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true)
+        .await
+        .expect("stage-create should succeed");
+    let first = commit_table_response(
+        &store,
+        &commit_backend,
+        "warehouse",
+        &namespace,
+        "events",
+        staged_create_commit_request(&staged.metadata, []),
+    )
+    .await
+    .expect("first assert-create should succeed");
+    let error = commit_table_response(
+        &store,
+        &commit_backend,
+        "warehouse",
+        &namespace,
+        "events",
+        staged_create_commit_request(&staged.metadata, []),
+    )
+    .await
+    .expect_err("assert-create must fail after the table exists");
+    assert_eq!(error.status_code(), Some(StatusCode::CONFLICT));
+    let current = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should remain visible");
+    assert_eq!(current.generation, first.generation);
+    assert_eq!(
+        table_metadata_location_for_client("warehouse", &current.metadata_location),
+        first.metadata_location
+    );
+}
+
 #[tokio::test]
 async fn standard_commit_applies_updates_and_writes_next_metadata() {
     let store = TestTableCatalogStore::default();
@@ -10868,6 +11279,68 @@ async fn seed_test_manifest_data_files(
     }
 }
 
+fn staged_events_create_request() -> CreateTableRequest {
+    serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [
+                {"id": 1, "name": "id", "required": true, "type": "long"},
+                {"id": 2, "name": "payload", "required": false, "type": "string"}
+            ]
+        },
+        "stage-create": true,
+        "properties": {"write.format.default": "parquet"}
+    }))
+    .expect("staged create table request should parse")
+}
+
+fn staged_create_commit_request(
+    staged_metadata: &serde_json::Value,
+    additional_updates: impl IntoIterator<Item = serde_json::Value>,
+) -> RestCommitTableRequest {
+    let mut updates = vec![
+        serde_json::json!({
+            "action": "assign-uuid",
+            "uuid": staged_metadata["table-uuid"]
+        }),
+        serde_json::json!({
+            "action": "upgrade-format-version",
+            "format-version": staged_metadata["format-version"]
+        }),
+        serde_json::json!({
+            "action": "add-schema",
+            "schema": staged_metadata["schemas"][0]
+        }),
+        serde_json::json!({"action": "set-current-schema", "schema-id": -1}),
+        serde_json::json!({
+            "action": "add-spec",
+            "spec": staged_metadata["partition-specs"][0]
+        }),
+        serde_json::json!({"action": "set-default-spec", "spec-id": -1}),
+        serde_json::json!({
+            "action": "add-sort-order",
+            "sort-order": staged_metadata["sort-orders"][0]
+        }),
+        serde_json::json!({"action": "set-default-sort-order", "sort-order-id": -1}),
+        serde_json::json!({
+            "action": "set-location",
+            "location": staged_metadata["location"]
+        }),
+        serde_json::json!({
+            "action": "set-properties",
+            "updates": staged_metadata["properties"]
+        }),
+    ];
+    updates.extend(additional_updates);
+    serde_json::from_value(serde_json::json!({
+        "requirements": [{"type": "assert-create"}],
+        "updates": updates
+    }))
+    .expect("staged create commit request should parse")
+}
+
 async fn create_standard_events_table<S>(
     store: &S,
     metadata_backend: &TestTableCatalogObjectBackend,
@@ -10907,9 +11380,17 @@ where
     }))
     .expect("standard create table request should parse");
     let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
-    create_table_response(store, &commit_backend, "warehouse", namespace, create_request, true)
+    let response = create_table_response(store, &commit_backend, "warehouse", namespace, create_request, true)
         .await
-        .expect("table should be created")
+        .expect("table should be created");
+    RestLoadTableResponse {
+        metadata_location: response
+            .metadata_location
+            .expect("direct create should return metadata location"),
+        metadata: response.metadata,
+        config: response.config,
+        storage_credentials: response.storage_credentials,
+    }
 }
 
 async fn create_standard_recent_events_view<S>(

rustfs/src/connect/diagnostics/perf_object.rs

@@ -64,6 +64,7 @@ const SIGNATURE_PATH: &str = "envelope.sig";
 const RESULT_PATH: &str = "result.json";
 const MAX_OBJECT_RESPONSE_BYTES: usize = 16_384;
 const CLEANUP_RESERVE_MAX: Duration = Duration::from_millis(250);
+const CLEANUP_TIMEOUT: Duration = Duration::from_secs(2);
 const OUTPUT_MODE: u32 = 0o600;
 
 static OBJECT_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false);
@@ -374,7 +375,7 @@ impl S3ObjectProbe {
         let mut builder = Client::builder()
             .no_proxy()
             .redirect(reqwest::redirect::Policy::none())
-            .timeout(timeout);
+            .timeout(timeout.max(CLEANUP_TIMEOUT));
         if let Some(root_ca_pem) = root_ca_pem {
             let certificate =
                 reqwest::Certificate::from_pem(root_ca_pem).map_err(|_| ObjectPerformanceError::InvalidRootCertificate)?;
@@ -403,7 +404,6 @@ impl S3ObjectProbe {
         let started = Instant::now();
         let reserve = CLEANUP_RESERVE_MAX.min(request.duration / 10);
         let operation_deadline = started + request.duration.saturating_sub(reserve);
-        let cleanup_deadline = started + request.duration;
         let bucket = format!("rustfs-connect-perf-{}", request.artifact_uid.replace('-', ""));
         let object = "synthetic-object";
         let bucket_url = self.object_url(&bucket, None)?;
@@ -427,6 +427,9 @@ impl S3ObjectProbe {
             }
             Err(error) => Err(error),
         };
+        // Cleanup owns two sequential requests and must not inherit only the
+        // fractional tail of the caller's measurement window.
+        let cleanup_deadline = Instant::now() + CLEANUP_TIMEOUT;
         let cleanup = self.cleanup(object_url, bucket_url, cleanup_deadline).await;
         if cleanup.is_err() {
             return Err(ObjectProbeError::CleanupFailed);

rustfs/src/table_catalog/store/mod.rs

@@ -245,6 +245,26 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
     async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
 
+    /// Checks whether an active table already owns any part of the candidate warehouse location.
+    ///
+    /// This is a preflight for clients that write data before catalog registration. Registration
+    /// remains the authoritative atomic check for concurrent creators.
+    async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> {
+        let candidate_prefix = table_warehouse_object_prefix(candidate)?;
+        for existing in self.list_all_tables(&candidate.table_bucket).await? {
+            if existing.state != TableCatalogEntryState::Active || existing.table_id == candidate.table_id {
+                continue;
+            }
+            let existing_prefix = table_warehouse_object_prefix(&existing)?;
+            if warehouse_object_prefixes_overlap(&existing_prefix, &candidate_prefix) {
+                return Err(TableCatalogStoreError::Conflict(format!(
+                    "table warehouse location overlaps an active table: {candidate_prefix}"
+                )));
+            }
+        }
+        Ok(())
+    }
+
     async fn list_tables_page(
         &self,
         table_bucket: &str,
@@ -1242,6 +1262,13 @@ where
         }
     }
 
+    async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> {
+        match self {
+            Self::ObjectBacked(store) => store.ensure_table_warehouse_location_available(candidate).await,
+            Self::DurableStrong(store) => store.ensure_table_warehouse_location_available(candidate).await,
+        }
+    }
+
     async fn list_tables_page(
         &self,
         table_bucket: &str,

rustfs/src/table_catalog/store/object.rs

@@ -5168,6 +5168,10 @@ where
         Ok(entries)
     }
 
+    async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> {
+        self.ensure_table_warehouse_prefix_available(candidate).await
+    }
+
     async fn list_tables_page(
         &self,
         table_bucket: &str,

rustfs/src/table_catalog/store/strong.rs

@@ -2270,6 +2270,16 @@ where
             .collect())
     }
 
+    async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> {
+        validate_table_entry_version_and_id(candidate)?;
+        let namespace = parse_namespace_for_store(&candidate.namespace)?;
+        let table = parse_table_for_store(&candidate.table)?;
+        self.hydrate_state().await?;
+        let state = self.state.lock().await;
+        let key = Self::table_key(&candidate.table_bucket, &namespace, &table);
+        Self::ensure_table_warehouse_prefix_available_locked(&state, candidate, &key)
+    }
+
     async fn list_tables_page(
         &self,
         table_bucket: &str,

rustfs/tests/connect_perf_object.rs

@@ -396,6 +396,61 @@ async fn oversized_response_is_rejected_without_buffering_it() {
     server.await.expect("response server");
 }
 
+#[tokio::test]
+async fn cleanup_uses_a_budget_independent_of_the_measurement_window() {
+    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+    let _guard = TEST_LOCK.lock().await;
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("response listener");
+    let address = listener.local_addr().expect("listener address");
+    let server = tokio::spawn(async move {
+        for (index, response) in [
+            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".as_slice(),
+            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".as_slice(),
+            b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\n\xa5\xa5\xa5\xa5".as_slice(),
+            b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(),
+            b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(),
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            let (mut socket, _) = listener.accept().await.expect("object connection");
+            let mut request = vec![0_u8; 16 * 1024];
+            let _ = socket.read(&mut request).await.expect("request headers");
+            if index >= 3 {
+                tokio::time::sleep(Duration::from_millis(600)).await;
+            }
+            socket.write_all(response).await.expect("response");
+        }
+    });
+    let probe = S3ObjectProbe::new(
+        &format!("http://{address}"),
+        None,
+        None,
+        Zeroizing::new("access".to_owned()),
+        Zeroizing::new("secret".to_owned()),
+        Zeroizing::new(String::new()),
+        Duration::from_secs(1),
+    )
+    .expect("object probe");
+    let mut request = request(ObjectOperation::GetObject);
+    request.traffic_bytes = 4;
+
+    let measurement = measure_object(&request, &probe, &CancellationToken::new())
+        .await
+        .expect("object measurement");
+    assert_eq!(
+        measurement.result.outcome(),
+        ObjectOutcome::Succeeded,
+        "target result: {:?}",
+        measurement.target
+    );
+    tokio::time::timeout(Duration::from_secs(3), server)
+        .await
+        .expect("cleanup requests should complete")
+        .expect("response server");
+}
+
 #[tokio::test]
 async fn signed_result_is_saved_without_overwrite() {
     let _guard = TEST_LOCK.lock().await;
@@ -463,6 +518,7 @@ fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put() {
 }
 
 async fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put_body() {
+    let _guard = TEST_LOCK.lock().await;
     let port = match find_available_port() {
         Ok(port) => port,
         Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,

scripts/table-catalog/README.md

@@ -649,8 +649,8 @@ python3 scripts/table-catalog/engine_compatibility.py \
   --cleanup
 ```
 
-The generated SQL covers namespace creation, table creation, append, refresh,
-count, and optional cleanup. Until Spark execution is enabled in CI through the
+The generated SQL covers namespace creation, atomic CTAS through the Iceberg
+REST staged-create flow, refresh, count, and optional cleanup. Until Spark execution is enabled in CI through the
 explicit live-conformance gate, do not claim Spark support beyond a manually
 verified run with the exact RustFS build, Spark version, Iceberg version, and
 expected output recorded.

scripts/table-catalog/duckdb_smoke.py

@@ -287,17 +287,6 @@ def negative_sql(args: argparse.Namespace, *, kind: str, seed_table: str, write_
     bootstrap = f"bootstrap_{kind}"
     sql = profile_sql(args, catalog=bootstrap, table=seed_table)
     sql += f"DETACH {engine_compatibility.quote_double_identifier(bootstrap)};\n"
-    if kind == "stage-create":
-        catalog = "stage_default"
-        sql += attach_sql(
-            args,
-            catalog=catalog,
-            rest_path="/iceberg",
-            signing_name="s3",
-            compatibility_options=False,
-        )
-        sql += f"CREATE TABLE {table_identifier(catalog, args.namespace, table_name(args.table, 'stage'))} (id BIGINT);\n"
-        return sql
     if kind == "purge":
         catalog = "purge_requested"
         sql += attach_sql(
@@ -435,9 +424,8 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du
     write_table = table_name(args.table, "write")
     purge_table = table_name(args.table, "purge")
     drop_table = table_name(args.table, "drop")
-    stage_table = table_name(args.table, "stage")
     v3_table = table_name(args.table, "v3")
-    smoke_tables = [seed_table, write_table, purge_table, drop_table, stage_table, v3_table]
+    smoke_tables = [seed_table, write_table, purge_table, drop_table, v3_table]
     catalog = pyiceberg_smoke.load_rest_catalog(iceberg_args, deps)
     namespace_preexisting = bool(catalog.namespace_exists(args.namespace))
     prepare_smoke_tables(catalog, args.namespace, smoke_tables, args.replace)
@@ -489,7 +477,6 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du
         checks["s3tables_alias"] = "pass"
 
         negatives = [
-            ("stage-create", "stage-create is not supported"),
             ("purge", "purgeRequested=true is not supported"),
             ("format-v3", "unsupported Iceberg table format-version: 3"),
         ]
@@ -507,7 +494,7 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du
             )
             require_duckdb_error(execution, kind, expected_error)
             checks[kind] = "failed-closed"
-        if catalog.table_exists((args.namespace, stage_table)) or catalog.table_exists((args.namespace, v3_table)):
+        if catalog.table_exists((args.namespace, v3_table)):
             raise RuntimeError("a failed DuckDB create probe left a catalog table behind")
         if not catalog.table_exists((args.namespace, purge_table)):
             raise RuntimeError("purgeRequested=true removed a table despite the expected failure")

scripts/table-catalog/engine_compatibility.py

@@ -136,8 +136,8 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
             "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
             "scenarios": [
                 scenario("create-namespace", "manual-live-harness", "CREATE NAMESPACE IF NOT EXISTS"),
-                scenario("create-table", "manual-live-harness", "CREATE TABLE USING iceberg"),
-                scenario("append", "manual-live-harness", "INSERT INTO"),
+                scenario("create-table", "manual-live-harness", "CREATE TABLE USING iceberg AS SELECT"),
+                scenario("staged-create", "manual-live-harness", "atomic CTAS through the REST staged-create flow"),
                 scenario("reload-table", "manual-live-harness", "REFRESH TABLE and SELECT COUNT"),
                 scenario("drop-table", "manual-live-harness", "DROP TABLE and optional DROP NAMESPACE"),
                 scenario("commit-conflict", "manual-validation-required", "requires a two-writer Spark or REST conflict harness"),
@@ -162,7 +162,7 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
                 scenario("catalog-attach", "automated", "attach `/iceberg` with s3 signing and `/_iceberg` with s3tables signing"),
                 scenario("read-table", "automated", "read a PyIceberg-created table through the attached catalog"),
                 scenario("write-table", "automated", "exercise single-table DDL, DML, schema evolution, snapshots, and PyIceberg cross-read"),
-                scenario("unsupported-boundaries", "automated", "verify staged create, purge, and format v3 fail closed"),
+                scenario("unsupported-boundaries", "automated", "verify purge and format v3 fail closed"),
                 scenario("multi-table-mode", "automated", "verify DuckDB can avoid the multi-table commit endpoint without claiming cross-table atomicity"),
             ],
         },
@@ -363,8 +363,8 @@ def spark_sql_smoke(
     statements = [
         f"CREATE NAMESPACE IF NOT EXISTS {namespace_identifier};",
         f"DROP TABLE IF EXISTS {table_identifier};",
-        f"CREATE TABLE {table_identifier} (id BIGINT, payload STRING) USING iceberg;",
-        f"INSERT INTO {table_identifier} VALUES (1, 'alpha'), (2, 'beta');",
+        f"CREATE TABLE {table_identifier} USING iceberg AS "
+        "SELECT * FROM VALUES (1L, 'alpha'), (2L, 'beta') AS source(id, payload);",
         f"REFRESH TABLE {table_identifier};",
         f"SELECT COUNT(*) AS row_count FROM {table_identifier};",
     ]
@@ -665,7 +665,7 @@ def live_conformance_evidence(
                     OrderedDict(
                         [
                             ("client", "Spark Iceberg REST catalog"),
-                            ("scenario", "create-namespace-create-table-append-refresh-count-cleanup"),
+                            ("scenario", "create-namespace-staged-ctas-refresh-count-cleanup"),
                             ("expected_status", "pass"),
                             ("expected_row_count", 2),
                             ("claim_after_pass", "manual-live-verified"),

scripts/table-catalog/test_duckdb_smoke.py

@@ -145,17 +145,6 @@ class DuckDBSmokeTest(unittest.TestCase):
 
     def test_boundary_sql_records_required_compatibility_options(self) -> None:
         args = self.args()
-        stage_sql = duckdb_smoke.negative_sql(
-            args,
-            kind="stage-create",
-            seed_table="events_seed",
-            write_table="events_write",
-            purge_table="events_purge",
-        )
-        stage_attach = stage_sql.split('DETACH "bootstrap_stage-create";', 1)[1]
-        self.assertNotIn("STAGE_CREATE_TABLES false", stage_attach)
-        self.assertIn("CREATE TABLE", stage_attach)
-
         purge_sql = duckdb_smoke.negative_sql(
             args,
             kind="purge",

scripts/table-catalog/test_engine_compatibility.py

@@ -32,7 +32,7 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertEqual(spark["status"], "manual-live-harness")
         self.assertContainsScenario(spark, "create-namespace", "manual-live-harness")
         self.assertContainsScenario(spark, "create-table", "manual-live-harness")
-        self.assertContainsScenario(spark, "append", "manual-live-harness")
+        self.assertContainsScenario(spark, "staged-create", "manual-live-harness")
         self.assertContainsScenario(spark, "reload-table", "manual-live-harness")
         self.assertContainsScenario(spark, "drop-table", "manual-live-harness")
         self.assertContainsScenario(spark, "commit-conflict", "manual-validation-required")
@@ -142,7 +142,7 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertEqual(config["spark.sql.catalog.rustfs.s3.secret-access-key"], "rustfsadmin")
         self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")
 
-    def test_spark_sql_smoke_covers_lifecycle_append_reload_and_cleanup(self) -> None:
+    def test_spark_sql_smoke_covers_staged_ctas_reload_and_cleanup(self) -> None:
         sql = engine_compatibility.spark_sql_smoke(
             catalog_name="rustfs",
             namespace="sales",
@@ -152,8 +152,9 @@ class EngineCompatibilityTest(unittest.TestCase):
 
         self.assertIn("CREATE NAMESPACE IF NOT EXISTS rustfs.`sales`", sql)
         self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql)
-        self.assertIn("CREATE TABLE rustfs.`sales`.`orders`", sql)
-        self.assertIn("INSERT INTO rustfs.`sales`.`orders`", sql)
+        self.assertIn("CREATE TABLE rustfs.`sales`.`orders` USING iceberg AS", sql)
+        self.assertIn("SELECT * FROM VALUES (1L, 'alpha'), (2L, 'beta')", sql)
+        self.assertNotIn("INSERT INTO rustfs.`sales`.`orders`", sql)
         self.assertIn("REFRESH TABLE rustfs.`sales`.`orders`", sql)
         self.assertIn("SELECT COUNT(*) AS row_count FROM rustfs.`sales`.`orders`", sql)
         self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql)
@@ -547,7 +548,7 @@ class EngineCompatibilityTest(unittest.TestCase):
         record = engine_compatibility.live_conformance_evidence_record(
             client_name="Spark Iceberg REST catalog",
             client_version="3.5.4/iceberg-1.7.1",
-            scenario="create-namespace-create-table-append-refresh-count-cleanup",
+            scenario="create-namespace-staged-ctas-refresh-count-cleanup",
             rustfs_build="rustfs-test",
             git_sha="abc123",
             catalog_backing="durable-strong",