test(e2e): stabilize mixed-version coordinator assertion against leader-lock refresh gaps (#9103)

Replace the immediate post-replacement leader identity comparison with a
polling stabilization loop (up to 20 x 500ms). A leader-lock refresh gap
during old-binary replacement can make leader() briefly observe a different
node; the loop waits for the measured new coordinator to be confirmed. A
persistent change to an old-binary node still fails with the observed
identity in the error message.

Removes the redundant immediate assertion that the stabilization loop
supersedes. Also consolidates the non-mixed leader/pid resolution to avoid
a double leader query.

crates/e2e_test/src/distributed/scanner_peer_bucket_test.rs

@@ -1636,20 +1636,53 @@ async fn hot_window_with_old_followers(
     let mixed_initial_physical = if mixed { json!(converged_shards(&run)?) } else { Value::Null };
     let scheduler_config = run.assert_default_cadence().await?;
     let baseline = run.quiet_baseline().await?;
-    let leader = run.leader().await?;
-    let leader_pid = run.pid(leader)?;
+    // A leader-lock refresh gap during old-binary replacement can make
+    // `leader()` briefly observe a different node. Poll until the coordinator
+    // is stable before comparing against the measured fleet. A persistent
+    // change to an old-binary node is a genuine election outcome and fails.
+    let (leader, leader_pid) = if mixed {
+        let expected_leader = usize::try_from(
+            mixed_fleet["leader"]
+                .as_u64()
+                .ok_or("mixed fleet omitted its measured leader")?,
+        )?;
+        let expected_pid = mixed_fleet["leader_pid"]
+            .as_u64()
+            .ok_or("mixed fleet omitted its measured leader pid")?;
+        let mut resolved = None;
+        for _ in 0..20 {
+            let candidate = run.leader().await?;
+            let candidate_pid = u64::from(run.pid(candidate)?);
+            if Some(u64::try_from(candidate)?) == mixed_fleet["leader"].as_u64() && Some(candidate_pid) == Some(expected_pid) {
+                resolved = Some((candidate, run.pid(candidate)?));
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+        }
+        match resolved {
+            Some(pair) => pair,
+            None => {
+                let current = match run.leader().await {
+                    Ok(node) => Some((node, run.pid(node).ok())),
+                    Err(_) => None,
+                };
+                return Err(format!(
+                    "the hot window did not retain its measured new coordinator after stabilization \
+                     (expected node {expected_leader} pid {expected_pid}, observed {current:?})"
+                )
+                .into());
+            }
+        }
+    } else {
+        let leader = run.leader().await?;
+        (leader, run.pid(leader)?)
+    };
     let producer = (leader + 1) % 4;
     let producers = if fault.is_some() {
         vec![producer, (leader + 2) % 4]
     } else {
         vec![producer]
     };
-    if mixed
-        && (mixed_fleet["leader"].as_u64() != Some(u64::try_from(leader)?)
-            || mixed_fleet["leader_pid"].as_u64() != Some(u64::from(leader_pid)))
-    {
-        return Err("the hot window did not retain its measured new coordinator".into());
-    }
     let producer_pid = run.pid(producer)?;
     let producer_address = run.peer_address(producer);
     let [cold_a, cold_b, hot] = run.buckets.clone();

crates/ecstore/src/cache_value/metacache_set.rs

@@ -747,6 +747,16 @@ async fn list_path_raw_inner(
             // Start every missing head read in the same round so one stalled
             // disk cannot multiply the wait budget by the erasure-set width.
             // Outcomes are still consumed below in stable disk-index order.
+            if !revjob_rx.is_cancelled() {
+                for (i, reader) in readers.iter_mut().enumerate() {
+                    if errs[i].is_none()
+                        && pending_entries[i].is_none()
+                        && let Some(Ok(Some(entry))) = reader.try_peek_buffered()
+                    {
+                        pending_entries[i] = Some(entry);
+                    }
+                }
+            }
             let concurrent_peeks = readers.iter_mut().enumerate().filter_map(|(i, reader)| {
                 if errs[i].is_some() || pending_entries[i].is_some() {
                     return None;
@@ -878,7 +888,7 @@ async fn list_path_raw_inner(
                 }
                 if entry.name == current.name {
                     // If exact match, we agree.
-                    if current.matches_strict(&entry) {
+                    if current.matches_strict_borrowed(&entry) {
                         top_entries[i] = Some(entry);
                         agree += 1;
                     } else {
@@ -986,7 +996,9 @@ async fn list_path_raw_inner(
 
             if agree == readers.len() && !opts.preserve_replica_metadata {
                 for r in readers.iter_mut() {
-                    let _ = r.skip(1).await;
+                    if !r.skip_current() {
+                        let _ = r.skip(1).await;
+                    }
                 }
 
                 if let Some(agreed_fn) = opts.agreed.as_ref() {
@@ -1009,7 +1021,7 @@ async fn list_path_raw_inner(
             }
 
             for (i, r) in readers.iter_mut().enumerate() {
-                if top_entries[i].is_some() {
+                if top_entries[i].is_some() && !r.skip_current() {
                     let _ = r.skip(1).await;
                 }
             }

crates/ecstore/src/config/com.rs

@@ -3506,6 +3506,7 @@ mod tests {
                 replication_decision: String::new(),
                 checksum: None,
                 multipart_completion_replayed: false,
+                backfilled_old_current_size: None,
             }
         }
     }

crates/ecstore/src/disk/local.rs

@@ -459,22 +459,62 @@ fn read_listing_metadata_entry_std(
     root: &os::PublicationRoot,
     object_dir: &Path,
 ) -> core::result::Result<ListingMetadataRead, ReadAllError> {
-    let metadata_path = object_dir.join(STORAGE_FORMAT_FILE);
-    let (bytes, _) = read_listing_metadata_bytes_std(root.open_readonly(&metadata_path))?;
-    let (file_meta, data_dirs, counts_toward_limit) = parse_listing_metadata_bytes(&bytes);
-    #[cfg(not(test))]
-    let _ = file_meta;
-    let has_namespace_child_candidate =
-        os::has_non_data_dir_entry(object_dir, &data_dirs).map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
-
-    Ok(ListingMetadataRead {
-        bytes,
+    #[cfg(target_os = "linux")]
+    {
+        let (dir_fd, meta_file) = root
+            .open_directory_and_readonly_child(object_dir, STORAGE_FORMAT_FILE)
+            .map_err(ReadAllError::Open)?;
+        let (bytes, _) = read_listing_metadata_bytes_std(Ok(meta_file))?;
+        let mut parsed_counts: Option<bool> = None;
+        let (data_dirs, has_namespace_child_candidate) = os::has_non_data_dir_entry_lazy_from_fd(&dir_fd, || {
+            let (_, data_dirs, counts) = parse_listing_metadata_bytes(&bytes);
+            parsed_counts = Some(counts);
+            data_dirs
+        })
+        .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
+        let counts_toward_limit = parsed_counts.unwrap_or_else(|| FileMeta::bytes_count_toward_listing_limit(&bytes));
+        Ok(ListingMetadataRead {
+            #[cfg(test)]
+            file_meta: FileMeta::load(&bytes).ok(),
+            bytes,
+            data_dirs,
+            has_namespace_child_candidate,
+            counts_toward_limit,
+        })
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        let metadata_path = object_dir.join(STORAGE_FORMAT_FILE);
+        let (bytes, _) = read_listing_metadata_bytes_std(root.open_readonly(&metadata_path))?;
         #[cfg(test)]
-        file_meta,
-        data_dirs,
-        has_namespace_child_candidate,
-        counts_toward_limit,
-    })
+        let (file_meta, data_dirs, counts_toward_limit, has_namespace_child_candidate) = {
+            let (file_meta, data_dirs, counts_toward_limit) = parse_listing_metadata_bytes(&bytes);
+            let has_namespace_child_candidate = os::has_non_data_dir_entry(object_dir, &data_dirs)
+                .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
+            (file_meta, data_dirs, counts_toward_limit, has_namespace_child_candidate)
+        };
+        #[cfg(not(test))]
+        let (data_dirs, counts_toward_limit, has_namespace_child_candidate) = {
+            let mut parsed_counts: Option<bool> = None;
+            let (data_dirs, has_namespace_child_candidate) = os::has_non_data_dir_entry_lazy(object_dir, || {
+                let (_, data_dirs, counts) = parse_listing_metadata_bytes(&bytes);
+                parsed_counts = Some(counts);
+                data_dirs
+            })
+            .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
+            let counts_toward_limit = parsed_counts.unwrap_or_else(|| FileMeta::bytes_count_toward_listing_limit(&bytes));
+            (data_dirs, counts_toward_limit, has_namespace_child_candidate)
+        };
+
+        Ok(ListingMetadataRead {
+            bytes,
+            #[cfg(test)]
+            file_meta,
+            data_dirs,
+            has_namespace_child_candidate,
+            counts_toward_limit,
+        })
+    }
 }
 
 pub(crate) fn inline_metadata_rollback_dir(version_id: Uuid, meta: &FileMeta) -> Uuid {
@@ -3563,9 +3603,9 @@ async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option
 
     loop {
         let result_limit = if remaining > 0 {
-            (remaining as usize).min(WALK_DIR_BATCH_SIZE)
+            (remaining as usize).min(READ_DIR_BATCH_SIZE)
         } else {
-            WALK_DIR_BATCH_SIZE
+            READ_DIR_BATCH_SIZE
         };
         let (next_entries, next_names, reached_eof) = read_dir_entry_batch_with_walk_stall(entries, result_limit, stall).await?;
         entries = next_entries;
@@ -3579,7 +3619,11 @@ async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option
     Ok(names)
 }
 
+#[cfg(test)]
 const WALK_DIR_BATCH_SIZE: usize = 32;
+#[cfg(not(test))]
+const WALK_DIR_BATCH_SIZE: usize = 256;
+const READ_DIR_BATCH_SIZE: usize = 512;
 
 async fn read_dir_with_walk_stall(path: PathBuf, stall: Option<Duration>) -> Result<std::fs::ReadDir> {
     let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
@@ -3598,7 +3642,7 @@ async fn read_dir_entry_batch_with_walk_stall(
     let batch = spawn_blocking_with_disk_io_permit(io_permit, move || {
         let mut names = Vec::with_capacity(result_limit);
         let mut reached_eof = false;
-        for _ in 0..WALK_DIR_BATCH_SIZE {
+        for _ in 0..READ_DIR_BATCH_SIZE {
             if names.len() == result_limit {
                 break;
             }
@@ -3606,8 +3650,8 @@ async fn read_dir_entry_batch_with_walk_stall(
                 reached_eof = true;
                 break;
             };
-            let _ = progress_tx.send(());
             let entry = entry.map_err(to_file_error)?;
+            let _ = progress_tx.send(());
             let name = entry.file_name().to_string_lossy().to_string();
             if name.is_empty() || name == "." || name == ".." {
                 continue;
@@ -9574,13 +9618,8 @@ impl LocalDisk {
             .and_then(|v| v.strip_suffix(".meta"))
             .and_then(|v| v.parse::<usize>().ok())
             .unwrap_or_default();
-        let data_path = self.io_get_object_path(
-            bucket,
-            &path_join_buf(&[
-                path.parent().unwrap_or_else(|| Path::new("")).to_string_lossy().as_ref(),
-                &format!("part.{num}"),
-            ]),
-        )?;
+        let parent_str = path.parent().unwrap_or_else(|| Path::new("")).to_string_lossy();
+        let data_path = self.io_get_object_path(bucket, &path_join_buf(&[parent_str.as_ref(), &format!("part.{num}")]))?;
         let metadata_path = self.io_get_object_path(bucket, path.to_string_lossy().as_ref());
         // A part's existence check, metadata read and decode share one dispatch.
         // Keep open errors unmapped for the existing missing-volume fallback.
@@ -9646,12 +9685,29 @@ impl LocalDisk {
             stream_listing_metadata_batch(&object_names, tx, |object_name| {
                 let (bucket_path, object_dir) = build_local_disk_object_path(&io_root, &volume_str, object_name);
                 #[cfg(target_os = "linux")]
-                let path_check = check_local_disk_valid_object_path_at(&io_root, &mount_lease, &bucket_path, &object_dir);
+                {
+                    if let Err(err) = check_local_disk_object_path_components(&io_root, &bucket_path, &object_dir) {
+                        return Err(ReadAllError::Disk(err));
+                    }
+                    match read_listing_metadata_entry_std(&root, &object_dir) {
+                        Ok(read) => Ok(read),
+                        Err(err) => {
+                            if let Err(path_err) =
+                                check_local_disk_valid_object_path_at(&io_root, &mount_lease, &bucket_path, &object_dir)
+                            {
+                                Err(ReadAllError::Disk(path_err))
+                            } else {
+                                Err(err)
+                            }
+                        }
+                    }
+                }
                 #[cfg(not(target_os = "linux"))]
-                let path_check = check_local_disk_valid_object_path(&io_root, &bucket_path, &object_dir);
-                match path_check {
-                    Ok(()) => read_listing_metadata_entry_std(&root, &object_dir),
-                    Err(err) => Err(ReadAllError::Disk(err)),
+                {
+                    match check_local_disk_valid_object_path(&io_root, &bucket_path, &object_dir) {
+                        Ok(()) => read_listing_metadata_entry_std(&root, &object_dir),
+                        Err(err) => Err(ReadAllError::Disk(err)),
+                    }
                 }
             });
         }));
@@ -10928,14 +10984,25 @@ impl LocalDisk {
                     }
 
                     let (pop, skip_object, dir_to_skip, scan_required) = dir_stack.pop().expect("operation should succeed");
-                    write_metacache_obj(
-                        out,
-                        &MetaCacheEntry {
+                    let more_ready_in_batch = !(opts.recursive && scan_required)
+                        && prefetched_batch.as_ref().is_some_and(|b| b.remaining > 0 && !b.rx.is_empty());
+                    if more_ready_in_batch {
+                        out.write_obj_buffered(&MetaCacheEntry {
                             name: pop.clone(),
                             ..Default::default()
-                        },
-                    )
-                    .await?;
+                        })
+                        .await
+                        .map_err(metacache_write_error)?;
+                    } else {
+                        write_metacache_obj(
+                            out,
+                            &MetaCacheEntry {
+                                name: pop.clone(),
+                                ..Default::default()
+                            },
+                        )
+                        .await?;
+                    }
 
                     let scan_path = pop.clone();
                     if opts.recursive && scan_required {
@@ -10992,7 +11059,14 @@ impl LocalDisk {
                         let mut spawned = false;
                         let (volume_dir, raw_res) = loop {
                             if let Some(batch) = prefetched_batch.as_mut() {
-                                if let Some(raw_res) = with_walk_stall_deadline(stall, batch.rx.recv()).await? {
+                                let next_res = match batch.rx.try_recv() {
+                                    Ok(res) => Some(res),
+                                    Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
+                                        with_walk_stall_deadline(stall, batch.rx.recv()).await?
+                                    }
+                                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => None,
+                                };
+                                if let Some(raw_res) = next_res {
                                     batch.remaining -= 1;
                                     let stop = batch.remaining == 0
                                         || raw_res.as_ref().map_or(true, |r| r.has_namespace_child_candidate);
@@ -11030,7 +11104,13 @@ impl LocalDisk {
                             };
                             prefetched_batch = Some(self.spawn_listing_metadata_batch(&opts.bucket, batch_names, stall).await?);
                         };
-                        with_walk_stall_timeout(stall, self.resolve_read_all_result(&opts.bucket, &volume_dir, raw_res)).await
+                        match raw_res {
+                            Ok(read) => Ok(read),
+                            Err(err) => {
+                                with_walk_stall_timeout(stall, self.resolve_read_all_result(&opts.bucket, &volume_dir, Err(err)))
+                                    .await
+                            }
+                        }
                     }
                     .await
                 } else {
@@ -11063,12 +11143,21 @@ impl LocalDisk {
 
                         meta.metadata = read.bytes;
 
-                        write_metacache_obj(out, &meta).await?;
-
                         if opts.limit <= 0 || read.counts_toward_limit {
                             *objs_returned += 1;
                         }
 
+                        let reached_limit = opts.limit > 0 && *objs_returned >= opts.limit;
+                        let more_ready_in_batch = !reached_limit
+                            && !read.has_namespace_child_candidate
+                            && dir_stack.is_empty()
+                            && prefetched_batch.as_ref().is_some_and(|b| b.remaining > 0 && !b.rx.is_empty());
+                        if more_ready_in_batch {
+                            out.write_obj_buffered(&meta).await.map_err(metacache_write_error)?;
+                        } else {
+                            write_metacache_obj(out, &meta).await?;
+                        }
+
                         let has_namespace_child_candidate = read.has_namespace_child_candidate;
                         let dir_to_skip: HashSet<String> = if has_namespace_child_candidate || !dir_objes.is_empty() {
                             read.data_dirs.into_iter().map(|dir| dir.to_string()).collect()
@@ -11550,7 +11639,7 @@ fn is_root_path(path: impl AsRef<Path>) -> bool {
 }
 
 fn metadata_counts_toward_limit(metadata: &[u8]) -> bool {
-    FileMeta::load(metadata).map_or(true, |meta| file_meta_counts_toward_limit(&meta))
+    FileMeta::bytes_count_toward_listing_limit(metadata)
 }
 
 fn file_meta_counts_toward_limit(meta: &FileMeta) -> bool {
@@ -13227,7 +13316,7 @@ impl DiskAPI for LocalDisk {
 
         let durability = effective_durability(dst_volume);
         let part = ObjectPartInfo::unmarshal(&meta)?;
-        if let Some(integrity) = part.integrity {
+        let integrity_prep = if let Some(integrity) = part.integrity {
             integrity.validate()?;
             if usize::try_from(integrity.number).map_err(|_| DiskError::FileCorrupt)? != part.number
                 || usize::try_from(integrity.size).map_err(|_| DiskError::FileCorrupt)? != part.size
@@ -13240,55 +13329,84 @@ impl DiskAPI for LocalDisk {
             let destination = dst_file_path.parent().ok_or(DiskError::FileCorrupt)?.join(&proof_name);
             check_path_length(source.to_string_lossy().as_ref())?;
             check_path_length(destination.to_string_lossy().as_ref())?;
-            let stat = fs::symlink_metadata(&source).await.map_err(to_file_error)?;
-            if !stat.is_file() || stat.len() != u64::try_from(integrity.index_size()?).map_err(|_| DiskError::FileCorrupt)? {
-                return Err(DiskError::FileCorrupt);
+            let expected_len = u64::try_from(integrity.index_size()?).map_err(|_| DiskError::FileCorrupt)?;
+            Some((integrity, source, destination, expected_len))
+        } else {
+            None
+        };
+        let sync = durability.syncs_data_shards() || durability.syncs_commit_metadata();
+        let publication_root = self.publication_root.clone();
+        let lease = os::acquire_namespace_mutation_lease_with_owner(&transaction_path, source_owner).await;
+        os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<Result<()>> {
+            if let Some(ref admission) = admission
+                && let Err(err) = admission.validate()
+            {
+                return Ok(Err(to_file_error(err).into()));
             }
-            if durability.syncs_data_shards() {
-                let source = source.clone();
-                let retained_owner = source_owner.clone();
-                tokio::task::spawn_blocking(move || {
-                    let _owner = retained_owner;
-                    os::sync_file(&source)
-                })
-                .await
-                .map_err(DiskError::from)?
-                .map_err(to_file_error)?;
-            }
-            // Publish the immutable generation before preparing the part switch.
-            // Rollback retains the old generation; an unreferenced new index is
-            // reclaimed with the upload directory if preparation is interrupted.
-            os::rename_all_with_precondition_and_parent_sync(
-                &source,
-                &destination,
-                &dst_volume_dir,
-                &self.publication_root,
-                source_owner.clone(),
-                admission.as_ref().map(|check| check.precondition()),
-                durability.syncs_commit_metadata(),
-            )
-            .await?;
-        }
-        let retained_owner = source_owner.clone();
-        tokio::task::spawn_blocking(move || {
-            let _owner = retained_owner;
-            if let Some(admission) = admission {
-                admission.validate().map_err(to_file_error)?;
-            }
-            let source = std::fs::symlink_metadata(&src_file_path).map_err(to_file_error)?;
+            let source = match std::fs::symlink_metadata(&src_file_path) {
+                Ok(source) => source,
+                Err(err) => return Ok(Err(to_file_error(err).into())),
+            };
             if !source.is_file() {
-                return Err(DiskError::FileAccessDenied);
+                return Ok(Err(DiskError::FileAccessDenied));
             }
             if transaction_path.exists() {
-                return Err(DiskError::FileAccessDenied);
+                return Ok(Err(DiskError::FileAccessDenied));
+            }
+            if let Some((integrity, source, destination, expected_len)) = integrity_prep {
+                let stat = match std::fs::symlink_metadata(&source) {
+                    Ok(stat) => stat,
+                    Err(err) => return Ok(Err(to_file_error(err).into())),
+                };
+                if !stat.is_file() || stat.len() != expected_len {
+                    return Ok(Err(DiskError::FileCorrupt));
+                }
+                let verify_and_sync = (|| -> std::io::Result<bool> {
+                    let mut file = std::fs::File::open(&source)?;
+                    let mut header = [0u8; rustfs_filemeta::shard_integrity::INDEX_HEADER_SIZE];
+                    std::io::Read::read_exact(&mut file, &mut header)?;
+                    if header != integrity.index_header() {
+                        return Ok(false);
+                    }
+                    if durability.syncs_data_shards() {
+                        file.sync_all()?;
+                    }
+                    Ok(true)
+                })();
+                match verify_and_sync {
+                    Ok(true) => {}
+                    Ok(false) => return Ok(Err(DiskError::FileCorrupt)),
+                    Err(err) => return Ok(Err(to_file_error(err).into())),
+                }
+                let precondition = admission.as_ref().map(|check| check.precondition());
+                // Publish the immutable generation before preparing the part switch.
+                // Rollback retains the old generation; an unreferenced new index is
+                // reclaimed with the upload directory if preparation is interrupted.
+                // Defer the parent directory fsync to the transaction_parent fsync
+                // below when both live in the same parent directory.
+                let same_parent = destination.parent() == transaction_path.parent();
+                if let Err(err) = os::rename_all_with_precondition_and_parent_sync_std(
+                    &source,
+                    &destination,
+                    &dst_volume_dir,
+                    &publication_root,
+                    precondition.as_deref(),
+                    durability.syncs_commit_metadata() && !same_parent,
+                ) {
+                    return Ok(Err(to_file_error(err).into()));
+                }
             }
 
             let Some(transaction_parent) = transaction_path.parent() else {
-                return Err(DiskError::InvalidPath);
+                return Ok(Err(DiskError::InvalidPath));
             };
-            std::fs::create_dir_all(transaction_parent).map_err(to_file_error)?;
+            if let Err(err) = std::fs::create_dir_all(transaction_parent) {
+                return Ok(Err(to_file_error(err).into()));
+            }
             let staging_path = transaction_parent.join(format!(".part-txn-{}", Uuid::new_v4()));
-            std::fs::create_dir(&staging_path).map_err(to_file_error)?;
+            if let Err(err) = std::fs::create_dir(&staging_path) {
+                return Ok(Err(to_file_error(err).into()));
+            }
 
             let prepare_result = (|| -> std::io::Result<()> {
                 snapshot_part_transaction_file(
@@ -13325,13 +13443,13 @@ impl DiskAPI for LocalDisk {
 
             if let Err(err) = prepare_result {
                 let _ = remove_dir_all_if_exists(&staging_path);
-                return Err(to_file_error(err).into());
+                return Ok(Err(to_file_error(err).into()));
             }
 
-            Ok(())
+            Ok(Ok(()))
         })
         .await
-        .map_err(DiskError::from)?
+        .map_err(to_file_error)?
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -13346,47 +13464,57 @@ impl DiskAPI for LocalDisk {
             check_path_length(candidate.to_string_lossy().as_ref())?;
         }
         let durability = effective_durability(volume);
+        let sync = durability.syncs_commit_metadata();
+        let lease = os::acquire_namespace_mutation_lease_with_owner(&transaction_path, source_owner).await;
 
-        tokio::task::spawn_blocking(move || {
-            let _owner = source_owner;
+        os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<Result<()>> {
             match std::fs::symlink_metadata(&transaction_path) {
                 Ok(metadata) if metadata.is_dir() => {}
-                Ok(_) => return Err(DiskError::FileCorrupt),
-                Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
-                Err(err) => return Err(to_file_error(err).into()),
+                Ok(_) => return Ok(Err(DiskError::FileCorrupt)),
+                Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(())),
+                Err(err) => return Ok(Err(to_file_error(err).into())),
             }
 
-            if let Some(admission) = admission {
-                admission.validate().map_err(to_file_error)?;
+            if let Some(admission) = admission
+                && let Err(err) = admission.validate()
+            {
+                return Ok(Err(to_file_error(err).into()));
             }
             if action == PartTransactionAction::Rollback {
-                std::fs::write(transaction_path.join(PART_TRANSACTION_ROLLBACK), []).map_err(to_file_error)?;
-                if durability.syncs_commit_metadata() {
-                    os::fsync_dir_std(&transaction_path).map_err(to_file_error)?;
+                if let Err(err) = std::fs::write(transaction_path.join(PART_TRANSACTION_ROLLBACK), []) {
+                    return Ok(Err(to_file_error(err).into()));
                 }
-                restore_part_transaction_file(
+                if durability.syncs_commit_metadata()
+                    && let Err(err) = os::fsync_dir_std(&transaction_path)
+                {
+                    return Ok(Err(to_file_error(err).into()));
+                }
+                if let Err(err) = restore_part_transaction_file(
                     &current_data_path,
                     &transaction_path.join(PART_TRANSACTION_OLD_DATA),
                     &transaction_path.join(PART_TRANSACTION_OLD_DATA_ABSENT),
                     &transaction_path.join("restore.data"),
-                )
-                .map_err(to_file_error)?;
-                restore_part_transaction_file(
+                ) {
+                    return Ok(Err(to_file_error(err).into()));
+                }
+                if let Err(err) = restore_part_transaction_file(
                     &current_meta_path,
                     &transaction_path.join(PART_TRANSACTION_OLD_META),
                     &transaction_path.join(PART_TRANSACTION_OLD_META_ABSENT),
                     &transaction_path.join("restore.meta"),
-                )
-                .map_err(to_file_error)?;
+                ) {
+                    return Ok(Err(to_file_error(err).into()));
+                }
                 if durability.syncs_commit_metadata()
                     && let Some(parent) = current_data_path.parent()
+                    && let Err(err) = os::fsync_dir_std(parent)
                 {
-                    os::fsync_dir_std(parent).map_err(to_file_error)?;
+                    return Ok(Err(to_file_error(err).into()));
                 }
             }
 
             let Some(parent) = transaction_path.parent() else {
-                return Err(DiskError::InvalidPath);
+                return Ok(Err(DiskError::InvalidPath));
             };
             // Delete only the generation made obsolete by this settled switch.
             // Check the published metadata before removing either proof, so a
@@ -13413,25 +13541,27 @@ impl DiskAPI for LocalDisk {
             {
                 let obsolete_path = parent.join(integrity.file_name());
                 match std::fs::remove_file(&obsolete_path) {
-                    Ok(()) => {
-                        if durability.syncs_commit_metadata() {
-                            os::fsync_dir_std(parent).map_err(to_file_error)?;
-                        }
-                    }
+                    Ok(()) => {}
                     Err(error) if error.kind() == ErrorKind::NotFound => {}
-                    Err(error) => return Err(to_file_error(error).into()),
+                    Err(error) => return Ok(Err(to_file_error(error).into())),
                 }
             }
             let cleanup_path = parent.join(format!(".part-txn-settled-{}", Uuid::new_v4()));
-            std::fs::rename(&transaction_path, &cleanup_path).map_err(to_file_error)?;
-            if durability.syncs_commit_metadata() {
-                os::fsync_dir_std(parent).map_err(to_file_error)?;
+            if let Err(err) = std::fs::rename(&transaction_path, &cleanup_path) {
+                return Ok(Err(to_file_error(err).into()));
             }
-            remove_dir_all_if_exists(&cleanup_path).map_err(to_file_error)?;
-            Ok(())
+            if durability.syncs_commit_metadata()
+                && let Err(err) = os::fsync_dir_std(parent)
+            {
+                return Ok(Err(to_file_error(err).into()));
+            }
+            if let Err(err) = remove_dir_all_if_exists(&cleanup_path) {
+                return Ok(Err(to_file_error(err).into()));
+            }
+            Ok(Ok(()))
         })
         .await
-        .map_err(DiskError::from)??;
+        .map_err(to_file_error)??;
 
         self.io_backend.invalidate_cached_fd(volume, path).await;
         Ok(())
@@ -13776,7 +13906,7 @@ impl DiskAPI for LocalDisk {
         }
         pool_meta_native::reject_raw_path(&self.root, volume, path)?;
         crate::hp_guard!("LocalDisk::create_file");
-        if !origvolume.is_empty() {
+        if !origvolume.is_empty() && !(volume == RUSTFS_META_TMP_BUCKET && path.ends_with("/part.1")) {
             let origvolume_dir = self.io_get_bucket_path(origvolume)?;
             if !skip_access_checks(origvolume) {
                 cached_access(origvolume_dir)
@@ -15850,7 +15980,7 @@ mod test {
         std::fs::write(&xl_path, old_metadata).expect("old metadata should be written");
         set_local_inline_rollback_hardlink_failure(&xl_path);
 
-        let rollback_path = create_local_inline_rollback_backup(&xl_path, &staging_path, old_metadata)
+        let rollback_path = create_local_inline_rollback_backup(&xl_path, &staging_path, old_metadata, false)
             .expect("copy fallback should create rollback backup");
         let backup = std::fs::read(&rollback_path).expect("fallback backup should be readable");
 

crates/ecstore/src/disk/local/commit.rs

@@ -175,17 +175,28 @@ pub(super) fn create_local_inline_rollback_backup(
     dst_file_path: &Path,
     staging_file_path: &Path,
     old_metadata: &[u8],
+    sync: bool,
 ) -> std::io::Result<PathBuf> {
     let Some(staging_parent) = staging_file_path.parent() else {
         return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"));
     };
     let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP);
     remove_file_if_exists(&backup_path)?;
-    if (should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err())
-        && let Err(err) = std::fs::write(&backup_path, old_metadata)
-    {
-        let _ = remove_file_if_exists(&backup_path);
-        return Err(err);
+    if should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err() {
+        let write_res = (|| -> std::io::Result<()> {
+            let mut file = std::fs::File::create(&backup_path)?;
+            std::io::Write::write_all(&mut file, old_metadata)?;
+            if sync && !old_metadata.is_empty() {
+                file.sync_data()?;
+            }
+            Ok(())
+        })();
+        if let Err(err) = write_res {
+            let _ = remove_file_if_exists(&backup_path);
+            return Err(err);
+        }
+    } else if sync && !old_metadata.is_empty() {
+        std::fs::File::open(&backup_path)?.sync_data()?;
     }
     Ok(backup_path)
 }
@@ -785,9 +796,10 @@ impl LocalDisk {
                     &dst_volume_dir_owned,
                     &publication_root_owned,
                 )?;
-                // An existing xl.meta can share versions or require rollback
-                // restoration when a post-rename directory sync fails.
-                if is_cancelled() || (sync_commit_metadata && has_dst_buf.is_some()) {
+                // An existing xl.meta that replaces an existing version slot
+                // requires a rollback directory; appending a new version keeps
+                // an inline rollback backup for post-rename directory sync errors.
+                if is_cancelled() {
                     return Ok(FusedCommitAttempt::Fallback(guard, has_dst_buf, fi_moved));
                 }
                 let mut xlmeta = FileMeta::new();
@@ -834,6 +846,13 @@ impl LocalDisk {
                     os::create_prepared_rename_source_with_commit_guard(&src_file_path_owned, &dst_file_path_owned, &guard)?;
                 prepared_metadata_source.write_all(&new_dst_buf, sync_commit_metadata)?;
 
+                let mut staged_rollback_path = None;
+                if sync_commit_metadata && let Some(ref old_metadata) = has_dst_buf {
+                    let backup_path =
+                        create_local_inline_rollback_backup(&dst_file_path_owned, &src_file_path_owned, old_metadata, true)?;
+                    staged_rollback_path = Some(StagedRollbackBackup(backup_path));
+                }
+
                 if no_inline {
                     if sync_data_shards && let Some((src_data_path, _)) = has_data_dir_path_owned.as_ref() {
                         let sync_result = if super::take_presynced_tmp_data_dir(src_data_path) {
@@ -846,12 +865,18 @@ impl LocalDisk {
                         if let Err(err) = sync_result
                             && err.kind() != ErrorKind::NotFound
                         {
+                            if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
+                                let _ = remove_file_if_exists(backup_path);
+                            }
                             return Err(err);
                         }
                     }
                     if let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref()
                         && let Err(err) = os::rename_all_with_commit_guard_on_thread(src_data_path, dst_data_path, &guard)
                     {
+                        if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
+                            let _ = remove_file_if_exists(backup_path);
+                        }
                         return Ok(FusedCommitAttempt::CommitErr(err));
                     }
                     #[cfg(test)]
@@ -868,6 +893,9 @@ impl LocalDisk {
                     &publication_root_owned,
                     &guard,
                 ) {
+                    if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
+                        let _ = remove_file_if_exists(backup_path);
+                    }
                     if no_inline && let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref() {
                         let _ = std::fs::rename(dst_data_path, src_data_path);
                     }
@@ -875,13 +903,22 @@ impl LocalDisk {
                 }
 
                 if sync_commit_metadata
-                    && let Err(err) =
-                        fsync_inline_commit_std(&dst_file_path_owned, &dst_volume_dir_owned, destination_was_absent, None, &guard)
+                    && let Err(err) = fsync_inline_commit_std(
+                        &dst_file_path_owned,
+                        &dst_volume_dir_owned,
+                        destination_was_absent,
+                        staged_rollback_path.as_ref(),
+                        &guard,
+                    )
                 {
                     if no_inline && let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref() {
                         let _ = std::fs::rename(dst_data_path, src_data_path);
                     }
-                    return Err(err);
+                    return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
+                }
+
+                if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
+                    let _ = remove_file_if_exists(backup_path);
                 }
 
                 if let Some(parent) = src_file_path_owned.parent() {
@@ -1053,52 +1090,6 @@ impl LocalDisk {
             // after the commit rename, and a crash before the rename means the
             // PUT was never acknowledged. A metadata commit: relaxed tiers
             // leave it to the page cache.
-            if let Some(publication) = multipart_publication.as_ref() {
-                pin_multipart_future(|| publication.prepare_metadata(&new_dst_buf, has_dst_buf.as_deref())).await?;
-            }
-            let tmp_meta_sync = if durability.syncs_commit_metadata() || multipart_publication.is_some() {
-                SyncMode::FileOnly
-            } else {
-                SyncMode::None
-            };
-            // The tmp xl.meta write and the shard-file fdatasync are independent
-            // (disjoint paths) and both only need to be durable before the commit
-            // renames below, so run them concurrently to drop a blocking
-            // round-trip from the PUT commit critical path (rustfs/backlog#922
-            // step 2). The "contents durable -> rename -> dst dir fsync" ordering
-            // is unchanged — both futures complete before any rename — which the
-            // rename_data crash-consistency harness (backlog#935) exercises.
-            //
-            // Shard durability: once rename_data succeeds the write is
-            // acknowledged, so data must not live only in the page cache.
-            // Multipart parts were already synced during rename_part, so their
-            // fdatasync here is a cheap no-op. A missing source dir is left for the
-            // rename below to report through the existing rollback path. Payload
-            // durability is kept by both strict and relaxed.
-            let tmp_meta_write = {
-                let src_file_path = multipart_publication
-                    .as_ref()
-                    .map_or_else(|| commit.source_metadata.to_path_buf(), |publication| publication.prepared_path.clone());
-                let dst_file_path = dst_file_path.clone();
-                let owned_directory_guard = commit.directory_guard.clone();
-                let owned_mutation_lease = commit.mutation_lease.clone();
-                async move {
-                    os::run_blocking_namespace_operation(owned_mutation_lease, move || {
-                        #[cfg(test)]
-                        run_owned_file_before_open(&src_file_path);
-                        let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(
-                            &src_file_path,
-                            &dst_file_path,
-                            &owned_directory_guard,
-                        )?;
-                        prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?;
-                        Ok(prepared_metadata_source)
-                    })
-                    .await
-                    .map_err(to_file_error)
-                    .map_err(DiskError::from)
-                }
-            };
             let shard_sync = async {
                 if durability.syncs_data_shards()
                     && let Some((src_data_path, _)) = has_data_dir_path.as_ref()
@@ -1109,15 +1100,55 @@ impl LocalDisk {
                 }
                 Ok(())
             };
-            let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync);
-            // Surface a tmp-meta failure first (its prior serial position), then a
-            // shard-sync failure; either aborts before any rename, exactly as the
-            // sequential version did.
-            let prepared_metadata_source = tmp_meta_res?;
-            shard_sync_res?;
-            let prepared_metadata_source = match multipart_publication.as_ref() {
-                Some(publication) => pin_multipart_future(|| publication.install_head(prepared_metadata_source)).await?,
-                None => prepared_metadata_source,
+            let prepared_metadata_source = if let Some(publication) = multipart_publication.as_ref() {
+                let dst_file_path_ref = &dst_file_path;
+                let dir_guard_ref = &commit.directory_guard;
+                let lease_clone = commit.mutation_lease.clone();
+                let prepare_fut = pin_multipart_future(|| {
+                    publication.prepare_and_install_head(
+                        &new_dst_buf,
+                        has_dst_buf.as_deref(),
+                        dst_file_path_ref,
+                        dir_guard_ref,
+                        lease_clone,
+                    )
+                });
+                let (tmp_meta_res, shard_sync_res) = tokio::join!(prepare_fut, shard_sync);
+                let prepared_metadata_source = tmp_meta_res?;
+                shard_sync_res?;
+                prepared_metadata_source
+            } else {
+                let tmp_meta_sync = if durability.syncs_commit_metadata() {
+                    SyncMode::FileOnly
+                } else {
+                    SyncMode::None
+                };
+                let tmp_meta_write = {
+                    let src_file_path = commit.source_metadata.to_path_buf();
+                    let dst_file_path = dst_file_path.clone();
+                    let owned_directory_guard = commit.directory_guard.clone();
+                    let owned_mutation_lease = commit.mutation_lease.clone();
+                    async move {
+                        os::run_blocking_namespace_operation(owned_mutation_lease, move || {
+                            #[cfg(test)]
+                            run_owned_file_before_open(&src_file_path);
+                            let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(
+                                &src_file_path,
+                                &dst_file_path,
+                                &owned_directory_guard,
+                            )?;
+                            prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?;
+                            Ok(prepared_metadata_source)
+                        })
+                        .await
+                        .map_err(to_file_error)
+                        .map_err(DiskError::from)
+                    }
+                };
+                let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync);
+                let prepared_metadata_source = tmp_meta_res?;
+                shard_sync_res?;
+                prepared_metadata_source
             };
             commit.directory_guard = remove_dst_base_before_commit(
                 dst_path,
@@ -1663,13 +1694,8 @@ impl LocalDisk {
                     #[cfg(not(windows))]
                     let backup_path = {
                         let _backup = os::StageTimer::new(native_timing, "pool_meta_native_backup_create");
-                        create_local_inline_rollback_backup(&dst, &src, old_metadata)?
+                        create_local_inline_rollback_backup(&dst, &src, old_metadata, sync)?
                     };
-                    #[cfg(not(windows))]
-                    if sync {
-                        let _backup_sync = os::StageTimer::new(native_timing, "pool_meta_native_backup_file_sync");
-                        std::fs::File::open(&backup_path)?.sync_data()?;
-                    }
                     staged_rollback_path = Some(StagedRollbackBackup(backup_path));
                 }
 

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -2132,12 +2132,13 @@ impl LocalDisk {
                     Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(None)),
                     Err(err) => return Err(err),
                 };
-                let (old_dir, journal_raw) = match (|| -> Result<Option<(Option<Uuid>, Vec<u8>)>> {
+                type PreparedDeleteTuple = (Option<Uuid>, Option<Vec<u8>>, Vec<u8>);
+                let (old_dir, target, journal_raw) = match (|| -> Result<Option<PreparedDeleteTuple>> {
                     let mut meta = FileMeta::load(&buf)?;
-                    let deleted_versions: Vec<_> = meta
+                    let mut deleted_headers: Vec<_> = meta
                         .versions
                         .iter()
-                        .map(|v| v.header.version_id.unwrap_or_default())
+                        .map(|v| (v.header.version_id, std::mem::discriminant(&v.header.version_type)))
                         .collect();
                     let original_versions: Vec<_> = meta
                         .versions
@@ -2156,9 +2157,14 @@ impl LocalDisk {
                             return Ok(None);
                         }
                     }
-                    if !meta.versions.is_empty() {
-                        return Ok(None);
-                    }
+                    let surviving: HashSet<_> = meta
+                        .versions
+                        .iter()
+                        .map(|v| (v.header.version_id, std::mem::discriminant(&v.header.version_type)))
+                        .collect();
+                    deleted_headers.retain(|header| !surviving.contains(header));
+                    let deleted_versions = deleted_headers.into_iter().map(|(id, _)| id.unwrap_or_default()).collect();
+                    let target = (!meta.versions.is_empty()).then(|| meta.marshal_msg()).transpose()?;
                     let mut predecessors = BTreeSet::new();
                     if let Ok(entries) = std::fs::read_dir(&file_path) {
                         for (visited, entry) in entries.enumerate() {
@@ -2216,16 +2222,16 @@ impl LocalDisk {
                         transaction,
                         bucket_incarnation: incarnation,
                         original_digest: Some(digest(&buf)),
-                        target_digest: None,
+                        target_digest: target.as_deref().map(digest),
                         data_dirs: old_dir.into_iter().collect(),
                         deleted_versions,
                         requests: vec![sparse],
                         original_versions,
                         predecessors: predecessors.into_iter().collect(),
                     };
-                    Ok(Some((old_dir, journal_bytes(&journal)?)))
+                    Ok(Some((old_dir, target, journal_bytes(&journal)?)))
                 })() {
-                    Ok(Some(pair)) => pair,
+                    Ok(Some(tuple)) => tuple,
                     Ok(None) => return Ok(Ok(None)),
                     Err(err) => return Ok(Err(err)),
                 };
@@ -2233,10 +2239,18 @@ impl LocalDisk {
                 let staged = tempfile::Builder::new()
                     .prefix("delete-preparation-")
                     .tempdir_in(&staging_root)?;
-                for (name, bytes) in [
-                    (STORAGE_FORMAT_FILE_BACKUP, buf.as_slice()),
-                    (PREPARED, journal_raw.as_slice()),
-                ] {
+                let backup_path = staged.path().join(STORAGE_FORMAT_FILE_BACKUP);
+                if std::fs::hard_link(&xl_path, &backup_path).is_err() {
+                    let mut file = std::fs::File::create(&backup_path)?;
+                    std::io::Write::write_all(&mut file, &buf)?;
+                    if sync && !buf.is_empty() {
+                        file.sync_data()?;
+                    }
+                }
+                for (name, bytes) in [(TARGET, target.as_deref()), (PREPARED, Some(journal_raw.as_slice()))] {
+                    let Some(bytes) = bytes else {
+                        continue;
+                    };
                     let mut file = std::fs::File::create(staged.path().join(name))?;
                     std::io::Write::write_all(&mut file, bytes)?;
                     if sync && !bytes.is_empty() {
@@ -2261,23 +2275,57 @@ impl LocalDisk {
                     std::fs::create_dir_all(parent)?;
                 }
                 std::fs::rename(staged.path(), &destination)?;
-                if sync {
-                    os::fsync_dir_std(&file_path)?;
+                if sync && let Err(err) = os::fsync_dir_std(&file_path) {
+                    let _ = std::fs::remove_dir_all(&destination);
+                    return Err(err);
                 }
                 let mut reserved_data_path = None;
                 if let Some(uuid) = old_dir {
                     let old_path = file_path.join(uuid.to_string());
                     if old_path.is_dir() {
                         let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
-                        let marker = std::fs::File::create(&reserved_marker)?;
-                        if sync {
-                            marker.sync_all()?;
-                            os::fsync_dir_std(&old_path)?;
+                        let reserve_res = (|| -> std::io::Result<()> {
+                            let marker = std::fs::File::create(&reserved_marker)?;
+                            if sync {
+                                marker.sync_all()?;
+                                os::fsync_dir_std(&old_path)?;
+                            }
+                            Ok(())
+                        })();
+                        if let Err(err) = reserve_res {
+                            let _ = std::fs::remove_file(&reserved_marker);
+                            let _ = std::fs::remove_dir_all(&destination);
+                            return Err(err);
                         }
                         reserved_data_path = Some((old_path, reserved_marker));
                     }
                 }
-                if let Err(err) = std::fs::remove_file(&xl_path)
+                if let Some(ref target_bytes) = target {
+                    let mut staged_xl = match tempfile::NamedTempFile::new_in(&staging_root) {
+                        Ok(staged_xl) => staged_xl,
+                        Err(err) => {
+                            if let Some((_, reserved_marker)) = &reserved_data_path {
+                                let _ = std::fs::remove_file(reserved_marker);
+                            }
+                            let _ = std::fs::remove_dir_all(&destination);
+                            return Err(err);
+                        }
+                    };
+                    let write_res = (|| -> std::io::Result<()> {
+                        std::io::Write::write_all(staged_xl.as_file_mut(), target_bytes)?;
+                        if sync && !target_bytes.is_empty() {
+                            staged_xl.as_file().sync_data()?;
+                        }
+                        std::fs::rename(staged_xl.path(), &xl_path)
+                    })();
+                    if let Err(err) = write_res {
+                        if let Some((_, reserved_marker)) = &reserved_data_path {
+                            let _ = std::fs::remove_file(reserved_marker);
+                        }
+                        let _ = std::fs::remove_dir_all(&destination);
+                        return Err(err);
+                    }
+                } else if let Err(err) = std::fs::remove_file(&xl_path)
                     && err.kind() != ErrorKind::NotFound
                 {
                     if let Some((_, reserved_marker)) = &reserved_data_path {
@@ -2360,12 +2408,18 @@ impl LocalDisk {
                 {
                     return Ok(Err(DiskError::FileCorrupt));
                 }
-                let current_exists = match regular_file_exists(&root.join(STORAGE_FORMAT_FILE)) {
-                    Ok(exists) => exists,
-                    Err(e) => return Ok(Err(e)),
+                let target_satisfied = match journal.target_digest {
+                    None => match regular_file_exists(&root.join(STORAGE_FORMAT_FILE)) {
+                        Ok(exists) => !exists,
+                        Err(e) => return Ok(Err(e)),
+                    },
+                    Some(expected_target) => match regular_bytes(&root.join(STORAGE_FORMAT_FILE)) {
+                        Ok(Some(bytes)) => digest(&bytes) == expected_target,
+                        Ok(None) => false,
+                        Err(e) => return Ok(Err(e)),
+                    },
                 };
-                let unversioned_target_absent = journal.target_digest.is_none() && !current_exists;
-                if !unversioned_target_absent {
+                if !target_satisfied {
                     return Ok(Ok(None));
                 }
                 let mut sole_transaction = journal.predecessors.is_empty();
@@ -2382,8 +2436,15 @@ impl LocalDisk {
                                     && id != transaction
                                     && !journal.data_dirs.contains(&id)
                                 {
-                                    sole_transaction = false;
-                                    break;
+                                    let dir_path = entry.path();
+                                    let is_surviving_data_dir = journal.original_versions.iter().any(|v| v.data_dir == Some(id))
+                                        && !dir_path.join(PREPARED).exists()
+                                        && !dir_path.join(DELETE_QUORUM_COMMITTED_FILE).exists()
+                                        && !dir_path.join(STORAGE_FORMAT_FILE_BACKUP).exists();
+                                    if !is_surviving_data_dir {
+                                        sole_transaction = false;
+                                        break;
+                                    }
                                 }
                             }
                         }
@@ -2406,7 +2467,7 @@ impl LocalDisk {
                 if sync {
                     os::fsync_dir_std(&rollback)?;
                 }
-                Ok(Ok(Some((journal, sole_transaction, unversioned_target_absent))))
+                Ok(Ok(Some((journal, sole_transaction, target_satisfied))))
             },
         )
         .await
@@ -2458,12 +2519,15 @@ impl LocalDisk {
         }
 
         let bucket_path = self.io_get_bucket_path(volume)?;
+        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
         let root = self.io_get_object_path(volume, object)?;
         let rollback = root.join(transaction.to_string());
         let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
         let sync = effective_durability(volume).syncs_commit_metadata();
-        let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
         let data_dirs = journal.data_dirs.clone();
+        let has_target = journal.target_digest.is_some();
+        let journal_for_receipt = has_target.then(|| journal.clone());
+        let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
         let prune_paths: Vec<_> = rollback
             .ancestors()
             .take_while(|dir| *dir != bucket_path.as_path() && dir.starts_with(&bucket_path))
@@ -2522,13 +2586,25 @@ impl LocalDisk {
             if !data_dirs.is_empty() && sync && root.is_dir() {
                 os::fsync_dir_std(&root)?;
             }
-            for name in [
-                STORAGE_FORMAT_FILE_BACKUP,
-                DELETE_MARKER_ROLLBACK_FILE,
-                TARGET,
-                PREPARED,
-                FINISHED,
-            ] {
+            for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] {
+                ignore_not_found(std::fs::remove_file(rollback.join(name)))?;
+            }
+            if let Some(journal) = journal_for_receipt {
+                let finished_receipt = CommitReceipt { journal, finished: true };
+                let receipt_bytes = serde_json::to_vec(&finished_receipt).map_err(std::io::Error::other)?;
+                std::fs::create_dir_all(&staging_root)?;
+                let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?;
+                std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?;
+                if sync && !receipt_bytes.is_empty() {
+                    staged_receipt.as_file().sync_data()?;
+                }
+                std::fs::rename(staged_receipt.path(), &receipt_path)?;
+                let _ = std::fs::File::create(rollback.join(FINISHED))?;
+                if sync {
+                    os::fsync_dir_std(&rollback)?;
+                }
+            }
+            for name in [TARGET, PREPARED, FINISHED] {
                 ignore_not_found(std::fs::remove_file(rollback.join(name)))?;
             }
             if sync {
@@ -6282,6 +6358,157 @@ mod tests {
             b"outside bytes"
         );
     }
+
+    #[tokio::test]
+    async fn delete_recovery_fused_versioned_delete_marker_and_publish_fast_path() {
+        let (_temp, disk, info, _) = fixture().await;
+        let root = disk.io_get_object_path("recovery", "object").expect("object root");
+        let transaction = Uuid::new_v4();
+        let mut marker = FileInfo {
+            name: "object".to_owned(),
+            version_id: Some(Uuid::new_v4()),
+            deleted: true,
+            mark_deleted: true,
+            mod_time: info.mod_time.map(|t| t + time::Duration::seconds(1)),
+            ..Default::default()
+        };
+        let opts = DeleteOptions {
+            old_data_dir: Some(transaction),
+            recoverable_delete: true,
+            ..Default::default()
+        };
+        let fused = disk
+            .try_fused_recoverable_delete_version("recovery", "object", &marker, &opts, transaction, None)
+            .await
+            .expect("fused delete marker preparation must succeed");
+        assert_eq!(fused, Some(()), "versioned delete marker must use fused path");
+        assert!(root.join(transaction.to_string()).join(TARGET).is_file());
+        assert!(root.join(STORAGE_FORMAT_FILE).is_file());
+
+        let published = disk
+            .try_publish_prepared_delete_local("recovery", "object", transaction, None)
+            .await
+            .expect("local publish must succeed")
+            .expect("versioned target must satisfy local publish");
+        assert!(published.1, "sole transaction must be true");
+        assert!(published.2, "target_satisfied must be true");
+
+        let finalized = disk
+            .try_finish_and_finalize_prepared_delete_local("recovery", "object", transaction, &published.0, None)
+            .await
+            .expect("local finalize must succeed");
+        assert!(finalized);
+        assert!(!root.join(transaction.to_string()).exists());
+        let final_meta = FileMeta::load(&fs::read(root.join(STORAGE_FORMAT_FILE)).await.expect("read final xl.meta"))
+            .expect("decode final xl.meta");
+        assert_eq!(final_meta.versions.len(), 2);
+        assert_eq!(final_meta.versions[0].header.version_id, marker.version_id);
+        marker.deleted = false;
+    }
+
+    #[tokio::test]
+    async fn delete_recovery_fused_versioned_delete_with_surviving_version_and_interrupted_target_cleanup() {
+        let (_temp, disk, mut v1, _) = fixture().await;
+        let root = disk.io_get_object_path("recovery", "object").expect("object path");
+
+        // Configure v1 and v2 with distinct physical data directories so deleting v2
+        // exercises both `old_dir` reservation and `is_surviving_data_dir`.
+        v1.data = None;
+        rustfs_utils::http::metadata_compat::remove_str(
+            &mut v1.metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_INLINE_DATA,
+        );
+        let v1_dir = Uuid::new_v4();
+        v1.data_dir = Some(v1_dir);
+        fs::create_dir_all(root.join(v1_dir.to_string()))
+            .await
+            .expect("create v1 dir");
+        fs::write(root.join(v1_dir.to_string()).join("part.1"), b"v1-data")
+            .await
+            .expect("write v1 part");
+
+        let mut v2 = v1.clone();
+        v2.version_id = Some(Uuid::new_v4());
+        let v2_dir = Uuid::new_v4();
+        v2.data_dir = Some(v2_dir);
+        fs::create_dir_all(root.join(v2_dir.to_string()))
+            .await
+            .expect("create v2 dir");
+        fs::write(root.join(v2_dir.to_string()).join("part.1"), b"v2-data")
+            .await
+            .expect("write v2 part");
+
+        let mut meta = FileMeta::new();
+        meta.add_version(v1.clone()).expect("add v1");
+        meta.add_version(v2.clone()).expect("add v2");
+        fs::write(root.join(STORAGE_FORMAT_FILE), meta.marshal_msg().expect("marshal meta"))
+            .await
+            .expect("write xl.meta");
+
+        let transaction = Uuid::new_v4();
+        let opts = DeleteOptions {
+            recoverable_delete: true,
+            ..Default::default()
+        };
+        let fused = disk
+            .try_fused_recoverable_delete_version("recovery", "object", &v2, &opts, transaction, None)
+            .await
+            .expect("fused versioned delete preparation must succeed");
+        assert_eq!(fused, Some(()));
+        assert!(
+            root.join(v2_dir.to_string())
+                .join(format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"))
+                .is_file(),
+            "deleted version data_dir must have recovery marker"
+        );
+
+        let (journal, sole_transaction, target_satisfied) = disk
+            .try_publish_prepared_delete_local("recovery", "object", transaction, None)
+            .await
+            .expect("local publish must succeed")
+            .expect("versioned target must satisfy local publish");
+        assert!(sole_transaction, "surviving v1 data_dir must not clear sole_transaction");
+        assert!(target_satisfied);
+        assert!(journal.target_digest.is_some());
+
+        // Simulate interruption during finish-before-finalize right after TARGET is removed
+        // (after finished receipt + FINISHED marker were persisted, before receipt_path is removed).
+        let rollback = root.join(transaction.to_string());
+        let finished_receipt = CommitReceipt {
+            journal: journal.clone(),
+            finished: true,
+        };
+        fs::write(
+            rollback.join(DELETE_QUORUM_COMMITTED_FILE),
+            serde_json::to_vec(&finished_receipt).expect("serialize finished receipt"),
+        )
+        .await
+        .expect("write finished receipt");
+        fs::write(rollback.join(FINISHED), []).await.expect("write FINISHED");
+        fs::remove_file(rollback.join(STORAGE_FORMAT_FILE_BACKUP))
+            .await
+            .expect("remove backup");
+        fs::remove_file(rollback.join(TARGET)).await.expect("remove TARGET");
+
+        let inspected = disk
+            .inspect_delete_transaction("recovery", "object", transaction)
+            .await
+            .expect("interrupted post-TARGET state must remain inspectable without FileCorrupt");
+        assert_eq!(inspected.state, DeleteRecoveryState::CommitFinished);
+
+        // Complete cleanup via try_finish_and_finalize_prepared_delete_local.
+        let finalized = disk
+            .try_finish_and_finalize_prepared_delete_local("recovery", "object", transaction, &journal, None)
+            .await
+            .expect("local finalize must succeed after interruption");
+        assert!(finalized);
+        assert!(!rollback.exists(), "rollback directory must be pruned");
+        assert!(!root.join(v2_dir.to_string()).exists(), "deleted v2 data_dir must be removed");
+        assert!(
+            root.join(v1_dir.to_string()).join("part.1").is_file(),
+            "surviving v1 data_dir must remain intact"
+        );
+    }
 }
 
 #[cfg(test)]

crates/ecstore/src/disk/local/multipart_source_state.rs

@@ -1323,41 +1323,6 @@ impl LocalDisk {
 }
 
 impl SourceStore {
-    /// Only native source validation may create provenance. Ordinary TMP/heal
-    /// publication can retain a valid old local anchor, never issue/rebind one.
-    pub async fn write_retirement_anchor(
-        &self,
-        state: &SourceState,
-        publication: &PublicationIdentity,
-        directory: &Path,
-        lease: Arc<os::NamespaceMutationLease>,
-    ) -> Result<()> {
-        if state.lineage != publication.lineage
-            || state.generation != publication.source_generation
-            || state.disk_id != self.disk_id
-            || state.physical_slot != self.physical_slot
-        {
-            return Err(DiskError::ConditionalDeleteConflict);
-        }
-        let record = NativeRetirementAnchor {
-            protocol_version: MULTIPART_SOURCE_PROTOCOL_VERSION,
-            disk_id: self.disk_id,
-            physical_slot: self.physical_slot,
-            publication: publication.clone(),
-            source_original_sha256: state.original_sha256,
-            source_validation_receipt_id: state.backup_owner,
-        };
-        let bytes = encode(&record)?;
-        let directory = directory.to_path_buf();
-        let base = self.base.clone();
-        let root = self.publication_root.clone();
-        os::run_blocking_namespace_operation(lease, move || {
-            let _parent = os::mkdir_all_below_existing_base_std(&directory, &base, &root)?;
-            os::replace_multipart_source_metadata(&directory.join(RETIREMENT_ANCHOR), &bytes, &base, &root)
-        })
-        .await
-        .map_err(DiskError::from)
-    }
     /// This caller already owns SourceRW and destination metadata ownership.
     /// An Aborting survivor can confirm a historical target only with its
     /// independent, same-disk publication anchor. Closure never creates one.
@@ -1569,7 +1534,31 @@ impl LocalDisk {
         };
         // Copy the small source FI to an independent immutable inode. Its
         // canonical inode is never truncated to build the destination proxy.
-        store.write_owned_backup_and_persist(owner, false, original, &state).await?;
+        let directory = staging.private_directory(publication.final_data_dir);
+        let anchor_record = NativeRetirementAnchor {
+            protocol_version: MULTIPART_SOURCE_PROTOCOL_VERSION,
+            disk_id: store.disk_id,
+            physical_slot: store.physical_slot,
+            publication: publication.clone(),
+            source_original_sha256: state.original_sha256,
+            source_validation_receipt_id: state.backup_owner,
+        };
+        let anchor_bytes = encode(&anchor_record)?;
+        state.validate(store.disk_id, &store.path)?;
+        let state_raw = encode(&state)?;
+        let source_dir = store.source.clone();
+        let base_dir = store.base.clone();
+        let pub_root = store.publication_root.clone();
+        let terminal = terminal_record_path(&source_dir, &state);
+        let original_bytes = original.to_vec();
+        os::run_blocking_namespace_operation_with_sync(staging.lease.clone(), true, move || {
+            SourceStore::write_owned_backup_std(&source_dir, &base_dir, &pub_root, owner, false, &original_bytes)?;
+            SourceStore::persist_state_std(&source_dir, &base_dir, &pub_root, &state_raw, terminal.as_deref(), false)?;
+            let _parent = os::mkdir_all_below_existing_base_std(&directory, &base_dir, &pub_root)?;
+            os::replace_multipart_source_metadata(&directory.join(RETIREMENT_ANCHOR), &anchor_bytes, &base_dir, &pub_root)
+        })
+        .await
+        .map_err(DiskError::from)?;
         let prepared_path = store.source.join(format!(".mpu-prepared-{owner}"));
         let context = NativeMultipartPublication {
             store,
@@ -1580,12 +1569,6 @@ impl LocalDisk {
             validated_prepared: parking_lot::Mutex::new(None),
             legacy,
         };
-        let directory = staging.private_directory(context.publication.final_data_dir);
-        let reserved = context.state.lock().clone();
-        context
-            .store
-            .write_retirement_anchor(&reserved, &context.publication, &directory, staging.lease.clone())
-            .await?;
         Ok(context)
     }
 }
@@ -1594,18 +1577,15 @@ impl NativeMultipartPublication {
     pub fn canonical_head(&self) -> PathBuf {
         self.store.source.join(STORAGE_FORMAT_FILE)
     }
+    #[cfg(test)]
     pub async fn prepare_metadata(&self, new_metadata: &[u8], old_metadata: Option<&[u8]>) -> Result<()> {
         let mut state = self.state.lock().clone();
-        let current = self.store.load().await?;
-        let current_state = current.state.as_ref().ok_or(DiskError::ConditionalDeleteConflict)?;
-        if encode(current_state)? != encode(&state)? || !current.head_matches_state {
-            return Err(DiskError::ConditionalDeleteConflict);
-        }
         // Validate the actual persisted representation once, after FileMeta's
         // canonicalization. No unvalidated input can install a source proxy.
         if !self.store.target_matches(&self.publication, new_metadata)? {
             return Err(DiskError::ConditionalDeleteConflict);
         }
+        let expected_state_raw = encode(&state)?;
         let validated = ValidatedPreparedPublication {
             publication: self.publication.clone(),
             sha256: raw_sha(new_metadata),
@@ -1618,17 +1598,109 @@ impl NativeMultipartPublication {
         state.prepared_sha256 = Some(validated.sha256);
         state.prepared_length = Some(validated.length);
         state.advance(MultipartSourcePhase::Prepared)?;
-        if let Some(previous) = old_metadata {
-            self.store
-                .write_owned_backup_and_persist(state.backup_owner, true, previous, &state)
-                .await?;
-        } else {
-            self.store.persist(&state).await?;
-        }
+        state.validate(self.store.disk_id, &self.store.path)?;
+        let next_state_raw = encode(&state)?;
+        let source = self.store.source.clone();
+        let base = self.store.base.clone();
+        let root = self.store.publication_root.clone();
+        let path = self.store.path.clone();
+        let disk_id = self.store.disk_id;
+        let physical_slot = self.store.physical_slot;
+        let disk_ids = self.store.disk_ids.clone();
+        let backup_owner = state.backup_owner;
+        let old_bytes = old_metadata.map(|b| b.to_vec());
+        let terminal = terminal_record_path(&source, &state);
+        os::run_blocking_namespace_operation_with_sync(self.store.lease.clone(), true, move || {
+            let current = load_source(&source, &base, &root, &path, disk_id, physical_slot, &disk_ids).map_err(IoError::other)?;
+            let current_state = current
+                .state
+                .as_ref()
+                .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+            let current_raw = encode(current_state).map_err(IoError::other)?;
+            if current_raw != expected_state_raw || !current.head_matches_state {
+                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+            }
+            if let Some(previous) = old_bytes.as_deref() {
+                SourceStore::write_owned_backup_std(&source, &base, &root, backup_owner, true, previous)?;
+            }
+            SourceStore::persist_state_std(&source, &base, &root, &next_state_raw, terminal.as_deref(), false)
+        })
+        .await
+        .map_err(DiskError::from)?;
         *self.state.lock() = state;
         *self.validated_prepared.lock() = Some(validated);
         Ok(())
     }
+    pub async fn prepare_and_install_head(
+        &self,
+        new_metadata: &[u8],
+        old_metadata: Option<&[u8]>,
+        dst_file_path: &Path,
+        directory_guard: &os::RenameCommitGuard,
+        mutation_lease: Arc<os::NamespaceMutationLease>,
+    ) -> Result<os::PreparedRenameSource> {
+        let mut state = self.state.lock().clone();
+        if !self.store.target_matches(&self.publication, new_metadata)? {
+            return Err(DiskError::ConditionalDeleteConflict);
+        }
+        let expected_state_raw = encode(&state)?;
+        let validated = ValidatedPreparedPublication {
+            publication: self.publication.clone(),
+            sha256: raw_sha(new_metadata),
+            length: new_metadata.len() as u64,
+        };
+        if let Some(previous) = old_metadata {
+            state.previous_target_sha256 = Some(raw_sha(previous));
+            state.previous_target_length = Some(previous.len() as u64);
+        }
+        state.prepared_sha256 = Some(validated.sha256);
+        state.prepared_length = Some(validated.length);
+        state.advance(MultipartSourcePhase::Prepared)?;
+        state.validate(self.store.disk_id, &self.store.path)?;
+        let next_state_raw = encode(&state)?;
+        let source = self.store.source.clone();
+        let base = self.store.base.clone();
+        let root = self.store.publication_root.clone();
+        let path = self.store.path.clone();
+        let disk_id = self.store.disk_id;
+        let physical_slot = self.store.physical_slot;
+        let disk_ids = self.store.disk_ids.clone();
+        let backup_owner = state.backup_owner;
+        let old_bytes = old_metadata.map(|b| b.to_vec());
+        let new_bytes = new_metadata.to_vec();
+        let terminal = terminal_record_path(&source, &state);
+        let prepared_path = self.prepared_path.clone();
+        let canonical = self.canonical_head();
+        let dst_file_path = dst_file_path.to_path_buf();
+        let owned_directory_guard = directory_guard.clone();
+        let prepared = os::run_blocking_namespace_operation_with_sync(mutation_lease, true, move || {
+            let current = load_source(&source, &base, &root, &path, disk_id, physical_slot, &disk_ids).map_err(IoError::other)?;
+            let current_state = current
+                .state
+                .as_ref()
+                .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+            let current_raw = encode(current_state).map_err(IoError::other)?;
+            if current_raw != expected_state_raw || !current.head_matches_state {
+                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+            }
+            if let Some(previous) = old_bytes.as_deref() {
+                SourceStore::write_owned_backup_std(&source, &base, &root, backup_owner, true, previous)?;
+            }
+            SourceStore::persist_state_std(&source, &base, &root, &next_state_raw, terminal.as_deref(), false)?;
+            #[cfg(test)]
+            super::run_owned_file_before_open(&prepared_path);
+            let mut prepared_metadata_source =
+                os::create_prepared_rename_source_with_commit_guard(&prepared_path, &dst_file_path, &owned_directory_guard)?;
+            prepared_metadata_source.write_all(&new_bytes, true)?;
+            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root)
+        })
+        .await
+        .map_err(DiskError::from)?;
+        *self.state.lock() = state;
+        *self.validated_prepared.lock() = Some(validated);
+        Ok(prepared)
+    }
+    #[cfg(test)]
     pub async fn install_head(&self, prepared: os::PreparedRenameSource) -> Result<os::PreparedRenameSource> {
         {
             let guard = self.validated_prepared.lock();
@@ -1701,11 +1773,7 @@ impl NativeMultipartPublication {
     }
     pub async fn finish(&self, target_lease: Arc<os::NamespaceMutationLease>) -> Result<MultipartPublicationResult> {
         let mut state = self.state.lock().clone();
-        let current = self.store.load().await?;
-        let current_state = current.state.as_ref().ok_or(DiskError::ConditionalDeleteConflict)?;
-        if current.head.is_some() || encode(current_state)? != encode(&state)? {
-            return Err(DiskError::ConditionalDeleteConflict);
-        }
+        let expected_state_raw = encode(&state)?;
         let (expected_sha256, expected_length) = {
             let guard = self.validated_prepared.lock();
             let validated = guard.as_ref().ok_or(DiskError::ConditionalDeleteConflict)?;
@@ -1717,16 +1785,52 @@ impl NativeMultipartPublication {
             }
             (validated.sha256, validated.length)
         };
-        self.store
-            .read_and_sync_target_with_lease(&self.publication, target_lease, move |target| {
-                if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
-                    return Err(DiskError::ConditionalDeleteConflict);
-                }
-                Ok(())
-            })
-            .await?;
+        let target_dir = self.store.target_path(&self.publication)?;
         state.advance(MultipartSourcePhase::ConsumedUnsealed)?;
-        let snapshot = self.store.persist_with_source_sync(&state, true).await?;
+        state.validate(self.store.disk_id, &self.store.path)?;
+        let next_state_raw = encode(&state)?;
+        let snapshot = snapshot_for_state(self.store.disk_id, &state, &next_state_raw, None)?;
+        let source = self.store.source.clone();
+        let base = self.store.base.clone();
+        let root = self.store.publication_root.clone();
+        let path = self.store.path.clone();
+        let disk_id = self.store.disk_id;
+        let physical_slot = self.store.physical_slot;
+        let disk_ids = self.store.disk_ids.clone();
+        let terminal = terminal_record_path(&source, &state);
+        os::run_blocking_namespace_operation_with_sync(target_lease, true, move || {
+            let current = load_source(&source, &base, &root, &path, disk_id, physical_slot, &disk_ids).map_err(IoError::other)?;
+            let current_state = current
+                .state
+                .as_ref()
+                .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+            let current_raw = encode(current_state).map_err(IoError::other)?;
+            if current.head.is_some() || current_raw != expected_state_raw {
+                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+            }
+            os::read_and_sync_multipart_target_metadata(
+                &target_dir,
+                std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
+                &base,
+                &root,
+                |target| {
+                    if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
+                        return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                    }
+                    Ok(())
+                },
+            )
+            .map_err(|error| {
+                if error.kind() == std::io::ErrorKind::NotFound {
+                    IoError::other(DiskError::ConditionalDeleteConflict)
+                } else {
+                    error
+                }
+            })?;
+            SourceStore::persist_state_std(&source, &base, &root, &next_state_raw, terminal.as_deref(), true)
+        })
+        .await
+        .map_err(DiskError::from)?;
         *self.state.lock() = state;
         let proof = self
             .store

crates/ecstore/src/disk/os.rs

@@ -2752,22 +2752,114 @@ pub(crate) fn read_dir_probe(path: impl AsRef<Path>, entry_limit: usize) -> io::
     })
 }
 
+#[cfg(test)]
 pub(crate) fn has_non_data_dir_entry(path: impl AsRef<Path>, data_dirs: &[uuid::Uuid]) -> io::Result<bool> {
-    match has_non_data_dir_entry_inner(path.as_ref(), data_dirs) {
-        Ok(res) => Ok(res),
-        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
-        Err(err) => Err(err),
-    }
+    has_non_data_dir_entry_lazy(path, || data_dirs.to_vec()).map(|(_, res)| res)
 }
 
-fn has_non_data_dir_entry_inner(path: &Path, data_dirs: &[uuid::Uuid]) -> io::Result<bool> {
-    let mut dir = std::fs::read_dir(path)?;
-    let entry_limit = data_dirs.len().saturating_add(2);
+#[cfg(target_os = "linux")]
+pub(crate) fn has_non_data_dir_entry_lazy_from_fd(
+    dir_fd: &std::fs::File,
+    load_data_dirs: impl FnOnce() -> Vec<uuid::Uuid>,
+) -> io::Result<(Vec<uuid::Uuid>, bool)> {
+    use rustix::fs::{AtFlags, FileType, RawDir, statat};
+
+    let mut raw_buf = [std::mem::MaybeUninit::<u8>::uninit(); 1024];
+    let mut dir = RawDir::new(dir_fd, &mut raw_buf);
+    let mut data_dirs: Option<Vec<uuid::Uuid>> = None;
+    let mut load = Some(load_data_dirs);
     let mut valid_entries = 0usize;
 
     for _ in 0..READ_DIR_PROBE_RAW_LIMIT {
         let Some(entry) = dir.next() else {
-            return Ok(false);
+            return Ok((data_dirs.unwrap_or_default(), false));
+        };
+        let entry = match entry {
+            Ok(entry) => entry,
+            Err(rustix::io::Errno::NOENT) => return Ok((data_dirs.unwrap_or_default(), false)),
+            Err(err) => return Err(io::Error::from(err)),
+        };
+        let c_name = entry.file_name();
+        let name_bytes = c_name.to_bytes();
+        if name_bytes.is_empty() || name_bytes == b"." || name_bytes == b".." {
+            continue;
+        }
+
+        let file_type = match entry.file_type() {
+            FileType::Unknown => match statat(dir_fd, c_name, AtFlags::SYMLINK_NOFOLLOW) {
+                Ok(st) => FileType::from_raw_mode(st.st_mode),
+                Err(rustix::io::Errno::NOENT) => continue,
+                Err(err) => return Err(io::Error::from(err)),
+            },
+            ft => ft,
+        };
+        if file_type == FileType::Directory {
+            let parsed_uuid = if name_bytes.len() == uuid::fmt::Hyphenated::LENGTH {
+                std::str::from_utf8(name_bytes).ok().and_then(|s| {
+                    let parsed = uuid::Uuid::try_parse(s).ok()?;
+                    let mut buf = [0u8; uuid::fmt::Hyphenated::LENGTH];
+                    (parsed.hyphenated().encode_lower(&mut buf) == s).then_some(parsed)
+                })
+            } else {
+                None
+            };
+            let Some(parsed) = parsed_uuid else {
+                let dirs = data_dirs
+                    .take()
+                    .unwrap_or_else(|| (load.take().expect("load_data_dirs called once"))());
+                return Ok((dirs, true));
+            };
+            let dirs = data_dirs.get_or_insert_with(|| (load.take().expect("load_data_dirs called once"))());
+            if dirs.binary_search(&parsed).is_err() {
+                return Ok((std::mem::take(dirs), true));
+            }
+            valid_entries += 1;
+        } else if file_type == FileType::RegularFile {
+            valid_entries += 1;
+        } else {
+            continue;
+        }
+
+        let entry_limit = data_dirs.as_ref().map_or(2, |dirs| dirs.len().saturating_add(2));
+        if valid_entries == entry_limit {
+            let dirs = data_dirs.get_or_insert_with(|| (load.take().expect("load_data_dirs called once"))());
+            if valid_entries == dirs.len().saturating_add(2) {
+                return Ok((std::mem::take(dirs), true));
+            }
+        }
+    }
+
+    let dirs = data_dirs
+        .take()
+        .unwrap_or_else(|| (load.take().expect("load_data_dirs called once"))());
+    Ok((dirs, true))
+}
+
+#[cfg(any(not(target_os = "linux"), test))]
+pub(crate) fn has_non_data_dir_entry_lazy(
+    path: impl AsRef<Path>,
+    load_data_dirs: impl FnOnce() -> Vec<uuid::Uuid>,
+) -> io::Result<(Vec<uuid::Uuid>, bool)> {
+    match has_non_data_dir_entry_lazy_inner(path.as_ref(), load_data_dirs) {
+        Ok(res) => Ok(res),
+        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok((Vec::new(), false)),
+        Err(err) => Err(err),
+    }
+}
+
+#[cfg(any(not(target_os = "linux"), test))]
+fn has_non_data_dir_entry_lazy_inner(
+    path: &Path,
+    load_data_dirs: impl FnOnce() -> Vec<uuid::Uuid>,
+) -> io::Result<(Vec<uuid::Uuid>, bool)> {
+    let mut dir = std::fs::read_dir(path)?;
+    let mut data_dirs: Option<Vec<uuid::Uuid>> = None;
+    let mut load = Some(load_data_dirs);
+    let mut valid_entries = 0usize;
+
+    for _ in 0..READ_DIR_PROBE_RAW_LIMIT {
+        let Some(entry) = dir.next() else {
+            return Ok((data_dirs.unwrap_or_default(), false));
         };
         let entry = entry?;
         let file_name = entry.file_name();
@@ -2777,18 +2869,24 @@ fn has_non_data_dir_entry_inner(path: &Path, data_dirs: &[uuid::Uuid]) -> io::Re
 
         let file_type = entry.file_type()?;
         if file_type.is_dir() {
-            let is_data_dir = file_name.len() == uuid::fmt::Hyphenated::LENGTH
-                && file_name
-                    .to_str()
-                    .and_then(|s| {
-                        let parsed = uuid::Uuid::try_parse(s).ok()?;
-                        let mut buf = [0u8; uuid::fmt::Hyphenated::LENGTH];
-                        (parsed.hyphenated().encode_lower(&mut buf) == s && data_dirs.binary_search(&parsed).is_ok())
-                            .then_some(())
-                    })
-                    .is_some();
-            if !is_data_dir {
-                return Ok(true);
+            let parsed_uuid = if file_name.len() == uuid::fmt::Hyphenated::LENGTH {
+                file_name.to_str().and_then(|s| {
+                    let parsed = uuid::Uuid::try_parse(s).ok()?;
+                    let mut buf = [0u8; uuid::fmt::Hyphenated::LENGTH];
+                    (parsed.hyphenated().encode_lower(&mut buf) == s).then_some(parsed)
+                })
+            } else {
+                None
+            };
+            let Some(parsed) = parsed_uuid else {
+                let dirs = data_dirs
+                    .take()
+                    .unwrap_or_else(|| (load.take().expect("load_data_dirs called once"))());
+                return Ok((dirs, true));
+            };
+            let dirs = data_dirs.get_or_insert_with(|| (load.take().expect("load_data_dirs called once"))());
+            if dirs.binary_search(&parsed).is_err() {
+                return Ok((std::mem::take(dirs), true));
             }
             valid_entries += 1;
         } else if file_type.is_file() {
@@ -2797,12 +2895,19 @@ fn has_non_data_dir_entry_inner(path: &Path, data_dirs: &[uuid::Uuid]) -> io::Re
             continue;
         }
 
+        let entry_limit = data_dirs.as_ref().map_or(2, |dirs| dirs.len().saturating_add(2));
         if valid_entries == entry_limit {
-            return Ok(true);
+            let dirs = data_dirs.get_or_insert_with(|| (load.take().expect("load_data_dirs called once"))());
+            if valid_entries == dirs.len().saturating_add(2) {
+                return Ok((std::mem::take(dirs), true));
+            }
         }
     }
 
-    Ok(true)
+    let dirs = data_dirs
+        .take()
+        .unwrap_or_else(|| (load.take().expect("load_data_dirs called once"))());
+    Ok((dirs, true))
 }
 
 // read_dir  count read limit. when count == 0 unlimit.
@@ -4832,6 +4937,17 @@ impl PublicationRoot {
         )
     }
 
+    #[cfg(target_os = "linux")]
+    pub(crate) fn open_directory_and_readonly_child(
+        &self,
+        dir_path: &Path,
+        child_name: &str,
+    ) -> io::Result<(std::fs::File, std::fs::File)> {
+        let dir = self.open_directory(dir_path)?;
+        let child = open_linux_relative(&dir, Path::new(child_name), rustix::fs::OFlags::RDONLY, rustix::fs::Mode::empty())?;
+        Ok((dir, child))
+    }
+
     #[cfg(target_os = "linux")]
     fn relative_path<'a>(&self, path: &'a Path) -> io::Result<&'a Path> {
         path.strip_prefix(self.io_path.as_path())
@@ -10985,6 +11101,54 @@ mod tests {
         assert_eq!(err.kind(), io::ErrorKind::NotFound);
     }
 
+    #[test]
+    fn has_non_data_dir_entry_lazy_skips_loader_when_only_xl_meta_exists() {
+        let temp_dir = tempdir().expect("create temp dir");
+        std::fs::write(temp_dir.path().join("xl.meta"), b"meta").expect("write xl.meta");
+        let mut called = 0usize;
+        let (dirs, has_child) = has_non_data_dir_entry_lazy(temp_dir.path(), || {
+            called += 1;
+            vec![uuid::Uuid::new_v4()]
+        })
+        .expect("lazy check must succeed");
+        assert_eq!(called, 0, "inline object dir without subdirs must not invoke loader");
+        assert!(dirs.is_empty());
+        assert!(!has_child);
+
+        #[cfg(target_os = "linux")]
+        {
+            let pub_root = PublicationRoot::new(temp_dir.path()).expect("open pub root");
+            let (dir_fd, _meta) = pub_root
+                .open_directory_and_readonly_child(temp_dir.path(), "xl.meta")
+                .expect("open dir and child");
+            let mut fd_called = 0usize;
+            let (fd_dirs, fd_has_child) = has_non_data_dir_entry_lazy_from_fd(&dir_fd, || {
+                fd_called += 1;
+                vec![uuid::Uuid::new_v4()]
+            })
+            .expect("fd lazy check must succeed");
+            assert_eq!(fd_called, 0);
+            assert!(fd_dirs.is_empty());
+            assert!(!fd_has_child);
+        }
+
+        let data_dir = uuid::Uuid::new_v4();
+        std::fs::create_dir(temp_dir.path().join(data_dir.to_string())).expect("create data dir");
+        let (dirs, has_child) = has_non_data_dir_entry_lazy(temp_dir.path(), || {
+            called += 1;
+            vec![data_dir]
+        })
+        .expect("lazy check with data dir must succeed");
+        assert_eq!(called, 1, "data dir presence must invoke loader once");
+        assert_eq!(dirs, vec![data_dir]);
+        assert!(!has_child);
+
+        std::fs::create_dir(temp_dir.path().join("nested-key")).expect("create nested child dir");
+        let (dirs, has_child) = has_non_data_dir_entry_lazy(temp_dir.path(), || vec![data_dir]).expect("lazy check child");
+        assert_eq!(dirs, vec![data_dir], "child presence must still return data_dirs for skip set");
+        assert!(has_child);
+    }
+
     #[tokio::test]
     #[serial_test::serial(file_sync_probe)]
     async fn resource_grouped_fdatasync_retired_limiter_keeps_its_budget_until_worker_cleanup() {

crates/ecstore/src/io_support/bitrot.rs

@@ -828,13 +828,27 @@ pub async fn create_bitrot_writer(
     length: i64,
     shard_size: usize,
     checksum_algo: HashAlgorithm,
+) -> disk::error::Result<BitrotWriterWrapper> {
+    create_bitrot_writer_with_orig_volume(is_inline_buffer, disk, "", volume, path, length, shard_size, checksum_algo).await
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn create_bitrot_writer_with_orig_volume(
+    is_inline_buffer: bool,
+    disk: Option<&DiskStore>,
+    origvolume: &str,
+    volume: &str,
+    path: &str,
+    length: i64,
+    shard_size: usize,
+    checksum_algo: HashAlgorithm,
 ) -> disk::error::Result<BitrotWriterWrapper> {
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
         let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
 
-        let file = disk.create_file("", volume, path, length).await?;
+        let file = disk.create_file(origvolume, volume, path, length).await?;
         #[cfg(feature = "hotpath")]
         let file = instrument_raw_shard_writer(file, disk.is_local());
         CustomWriter::new_tokio_writer(file)

crates/ecstore/src/object_api/types.rs

@@ -1543,6 +1543,8 @@ pub struct ObjectInfo {
     /// published by the same upload. Callers must not repeat quota, replication,
     /// lifecycle, or object-created side effects for that response.
     pub multipart_completion_replayed: bool,
+    /// Previous current version's size observed during `rename_data` commit.
+    pub backfilled_old_current_size: Option<crate::disk::OldCurrentSize>,
 }
 
 impl Default for ObjectInfo {
@@ -1585,6 +1587,7 @@ impl Default for ObjectInfo {
             replication_decision: String::new(),
             checksum: None,
             multipart_completion_replayed: false,
+            backfilled_old_current_size: None,
         }
     }
 }
@@ -1628,6 +1631,7 @@ impl Clone for ObjectInfo {
             replication_decision: self.replication_decision.clone(),
             checksum: self.checksum.clone(),
             multipart_completion_replayed: self.multipart_completion_replayed,
+            backfilled_old_current_size: self.backfilled_old_current_size,
             expires: self.expires,
         }
     }

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -5598,10 +5598,11 @@ impl SetDisks {
         write_quorum: usize,
         quorum_context: Option<MultipartWriteQuorumContext<'_>>,
     ) -> disk::error::Result<Vec<Option<DiskStore>>> {
-        self.recover_part_transaction(dst_object, write_quorum).await?;
         let part = ObjectPartInfo::unmarshal(&meta)?;
         let integrity = part.integrity.map(Arc::new);
 
+        self.recover_part_transaction(dst_object, write_quorum).await?;
+
         let src_bucket = Arc::new(src_bucket.to_string());
         let src_object = Arc::new(src_object.to_string());
         let dst_bucket = Arc::new(dst_bucket.to_string());
@@ -5623,7 +5624,9 @@ impl SetDisks {
                 if let Err(error) = prepared {
                     return Some(Err(error));
                 }
-                if let Some(integrity) = integrity {
+                if let Some(integrity) = integrity
+                    && !disk.is_local()
+                {
                     let Some((directory, _)) = dst_object.rsplit_once('/') else {
                         return Some(Err(DiskError::FileCorrupt));
                     };
@@ -5740,7 +5743,36 @@ impl SetDisks {
             return Err(err);
         }
 
-        let committed = self.recover_part_transaction(&dst_object, write_quorum).await?;
+        let all_prepared_succeeded = errs
+            .iter()
+            .zip(prepared_disks.iter())
+            .all(|(err, disk)| disk.is_none() || err.is_none());
+        let committed = if all_prepared_succeeded {
+            let settles = prepared_disks.iter().map(|disk| {
+                let disk = disk.clone();
+                let dst_object = dst_object.clone();
+                async move {
+                    let Some(disk) = disk else {
+                        return Err(DiskError::DiskNotFound);
+                    };
+                    disk.settle_part_transaction(RUSTFS_META_MULTIPART_BUCKET, &dst_object, PartTransactionAction::Commit)
+                        .await
+                }
+            });
+            let settle_results = join_all(settles).await;
+            if settle_results.iter().filter(|res| res.is_ok()).count() >= write_quorum
+                && settle_results
+                    .iter()
+                    .zip(prepared_disks.iter())
+                    .all(|(res, disk)| disk.is_none() || res.is_ok())
+            {
+                true
+            } else {
+                self.recover_part_transaction(&dst_object, write_quorum).await?
+            }
+        } else {
+            self.recover_part_transaction(&dst_object, write_quorum).await?
+        };
         if !committed {
             let err = DiskError::ErasureWriteQuorum;
             if let Some(context) = quorum_context {

crates/ecstore/src/set_disk/mod.rs

@@ -80,7 +80,7 @@ use crate::erasure::codec::bridge::{
 use crate::erasure::coding;
 use crate::error::{Error, Result, is_err_version_not_found};
 use crate::error::{GenericError, ObjectApiError, is_err_object_not_found};
-use crate::io_support::bitrot::{create_bitrot_reader_from_bytes, create_bitrot_writer};
+use crate::io_support::bitrot::{create_bitrot_reader_from_bytes, create_bitrot_writer, create_bitrot_writer_with_orig_volume};
 use crate::object_api::ObjectOptions;
 use crate::object_api::get_object_body_cache_hook;
 use crate::object_api::object_api_utils::get_raw_etag;

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

@@ -3471,10 +3471,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         bind_multipart_proof_snapshot(&disks, &self.multipart_proof_verification_context())?;
 
         let part_path = format!("{}/{}/", upload_id_path, fi.data_dir.unwrap_or(Uuid::nil()));
-        self.recover_part_transactions(&part_path, read_quorum, write_quorum)
-            .await
-            .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?;
-
         let part_meta_paths = uploaded_parts
             .iter()
             .map(|v| format!("{part_path}part.{0}.meta", v.part_num))
@@ -3482,6 +3478,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let part_numbers = uploaded_parts.iter().map(|v| v.part_num).collect::<Vec<usize>>();
 
+        self.recover_part_transactions(&part_path, read_quorum, write_quorum)
+            .await
+            .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?;
+
         let object_parts = Self::read_parts(&disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum)
             .await
             .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?;
@@ -4389,6 +4389,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             let needs_immediate_heal = rename_commit.needs_immediate_heal();
             let op_old_dir = rename_commit.data_dir;
             let cleanup_disks = rename_commit.cleanup_disks;
+            let old_current_size = rename_commit.old_current_size;
             let committed_file_info = rename_commit.committed_file_info;
 
             if needs_immediate_heal {
@@ -4503,7 +4504,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
             drop(_upload_guard.take());
 
-            Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
+            let mut obj_info = ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned);
+            obj_info.backfilled_old_current_size = old_current_size;
+            Ok(obj_info)
         };
 
         let result = if detach_commit_owner {

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

@@ -46,9 +46,9 @@ use super::super::{
     build_get_object_info, build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
     check_object_lock_delete, check_object_lock_for_deletion_with_state, check_object_lock_retention_update,
     classify_get_codec_streaming_object_class, classify_put_write_path, classify_storage_error,
-    collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug, delete_file_info_version_id,
-    disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker, finish_set_disk_read_lock,
-    get_codec_streaming_reader_gate_with_plan, get_object_body_cache_hook, get_raw_etag,
+    collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer_with_orig_volume, debug,
+    delete_file_info_version_id, disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker,
+    finish_set_disk_read_lock, get_codec_streaming_reader_gate_with_plan, get_object_body_cache_hook, get_raw_etag,
     get_small_object_direct_memory_decision_with_threshold_and_plan, get_small_object_direct_memory_threshold,
     get_stage_timer_if_enabled, get_str, get_transitioned_object_reader_with_tier_manager, inline_admission_shard_size,
     inline_erasure_shard_file_offset, inline_erasure_shard_size, insert_str, is_deadlock_detection_enabled,
@@ -4459,9 +4459,10 @@ impl SetDisks {
                             if let Some(disk) = disk_op
                                 && disk.is_online().await
                             {
-                                match create_bitrot_writer(
+                                match create_bitrot_writer_with_orig_volume(
                                     is_inline_buffer,
                                     Some(disk),
+                                    bucket,
                                     RUSTFS_META_TMP_BUCKET,
                                     &tmp_obj,
                                     shard_file_size,
@@ -11064,7 +11065,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         if let Some(err) = &gerr
             && goi.name.is_empty()
         {
-            if should_force_delete_marker_for_missing_version(&opts) {
+            if (is_err_object_not_found(err) || is_err_version_not_found(err))
+                && should_force_delete_marker_for_missing_version(&opts)
+            {
                 version_found = false;
             } else {
                 if is_err_object_not_found(err) {

crates/ecstore/src/store/list_objects.rs

@@ -1292,6 +1292,37 @@ impl ECStore {
         );
 
         let rx = rx.child_token();
+        if self.pools.len() == 1 && self.pools[0].disk_set.len() == 1 {
+            rustfs_io_metrics::record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 1, 1);
+            let res = self.pools[0].disk_set[0].list_path(rx, opts, sender).await;
+            let mut outcome = ListPathRawOutcome::default();
+            let err = match res {
+                Ok(set_outcome) => {
+                    outcome.limit_reached = set_outcome.limit_reached;
+                    None
+                }
+                Err(err) => Some(err),
+            };
+            let errs = [err];
+            if is_all_not_found(&errs) {
+                return Ok(outcome);
+            }
+            if let Some(err) = errs.into_iter().flatten().next() {
+                return Err(err);
+            }
+            rustfs_io_metrics::record_stage_duration(
+                "store_list_objects_list_merged",
+                merge_started.elapsed().as_secs_f64() * 1000.0,
+            );
+            debug!(
+                set_pair_count = 1,
+                limit_reached = outcome.limit_reached,
+                error_count = 0,
+                "store list_merged finished"
+            );
+            return Ok(outcome);
+        }
+
         let mut futures = Vec::new();
 
         let mut inputs = Vec::new();
@@ -1811,7 +1842,7 @@ async fn gather_results(
         if opts.versioned
             && is_object
             && entry
-                .xl_meta()
+                .xl_meta_ref()
                 .is_ok_and(|meta| !meta.versions.is_empty() && meta.versions.iter().all(|version| version.header.free_version()))
         {
             continue;

crates/ecstore/src/store/object.rs

@@ -5637,13 +5637,11 @@ impl ECStore {
         }
 
         if self.single_pool()
-            && !creates_latest_marker
             && !opts.data_movement
             && opts.http_preconditions.is_none()
             && !opts.replication_request
-            && !opts.versioned
             && !opts.version_suspended
-            && opts.version_id.is_none()
+            && !opts.delete_marker
             && !self.is_suspended(0).await
             && !self.is_pool_rebalancing(0).await
         {
@@ -5655,6 +5653,10 @@ impl ECStore {
                 .await
             {
                 Ok(mut res) => {
+                    #[cfg(test)]
+                    if creates_latest_marker {
+                        pause_versioned_delete_marker_after_commit(bucket, object).await;
+                    }
                     res.name = decode_dir_object(res.name.as_str());
                     return Ok(res);
                 }

crates/filemeta/src/filemeta.rs

@@ -2070,6 +2070,53 @@ mod test {
         assert!(meta.versions.iter().all(|version| !version.header.free_version()));
     }
 
+    #[test]
+    fn bytes_count_toward_listing_limit_matches_fileinfo_visibility() {
+        let mut meta = FileMeta::new();
+        let mut obj = FileInfo::new("object", 2, 2);
+        obj.version_id = Some(Uuid::new_v4());
+        obj.mod_time = Some(OffsetDateTime::now_utc());
+        meta.add_version(obj.clone()).expect("add object version");
+        let obj_bytes = meta.marshal_msg().expect("marshal object");
+        assert!(FileMeta::bytes_count_toward_listing_limit(&obj_bytes));
+
+        let dm = FileInfo {
+            name: "object".to_string(),
+            version_id: Some(Uuid::new_v4()),
+            deleted: true,
+            mark_deleted: true,
+            mod_time: obj.mod_time.map(|t| t + time::Duration::seconds(1)),
+            ..Default::default()
+        };
+        meta.delete_version(&dm).expect("add delete marker");
+        let dm_bytes = meta.marshal_msg().expect("marshal delete marker");
+        assert!(!FileMeta::bytes_count_toward_listing_limit(&dm_bytes));
+
+        let (_, mut source) = tier_overwrite_fixture(crate::TransitionVersionState::Exact);
+        source.version_id = Some(Uuid::new_v4());
+        let mut transitioned = FileMeta::new();
+        transitioned.add_version(source.clone()).expect("seed transitioned version");
+        let mut del_req = FileInfo {
+            name: "object".to_string(),
+            version_id: source.version_id,
+            mod_time: source.mod_time,
+            ..Default::default()
+        };
+        del_req.set_tier_free_version_id(&Uuid::new_v4().to_string());
+        transitioned
+            .delete_version(&del_req)
+            .expect("delete transitioned version leaving free version");
+        assert!(!FileMeta::bytes_count_toward_listing_limit(
+            &transitioned.marshal_msg().expect("marshal free-only")
+        ));
+
+        transitioned.add_version(obj).expect("add live object version");
+        assert!(FileMeta::bytes_count_toward_listing_limit(
+            &transitioned.marshal_msg().expect("marshal free + live object")
+        ));
+        assert!(FileMeta::bytes_count_toward_listing_limit(b"corrupt-xl-meta"));
+    }
+
     #[test]
     fn add_version_filemata_uses_canonical_equal_time_order() {
         let mod_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid test timestamp");

crates/filemeta/src/filemeta/codec.rs

@@ -324,6 +324,63 @@ impl FileMeta {
         true
     }
 
+    pub(crate) fn buffers_match_strict(self_buf: &[u8], other_buf: &[u8]) -> bool {
+        let Ok((s_meta, _)) = Self::parse_validated_xl_envelope(self_buf) else {
+            return false;
+        };
+        let Ok((o_meta, _)) = Self::parse_validated_xl_envelope(other_buf) else {
+            return false;
+        };
+
+        if s_meta.is_empty() || o_meta.is_empty() {
+            return s_meta.is_empty() && o_meta.is_empty();
+        }
+
+        let Ok((s_len, s_header_ver, _s_meta_ver, s_meta)) = Self::decode_validated_xl_headers(s_meta) else {
+            return false;
+        };
+        let Ok((o_len, o_header_ver, _o_meta_ver, o_meta)) = Self::decode_validated_xl_headers(o_meta) else {
+            return false;
+        };
+
+        if s_len != o_len {
+            return false;
+        }
+
+        let mut s_cur: Cursor<&[u8]> = Cursor::new(s_meta);
+        let mut o_cur: Cursor<&[u8]> = Cursor::new(o_meta);
+        for _ in 0..s_len {
+            let Ok((s_header_raw, _)) = Self::read_next_raw_slices(s_meta, &mut s_cur) else {
+                return false;
+            };
+            let Ok((o_header_raw, _)) = Self::read_next_raw_slices(o_meta, &mut o_cur) else {
+                return false;
+            };
+            if s_header_ver == XL_HEADER_VERSION
+                && o_header_ver == XL_HEADER_VERSION
+                && s_header_raw.len() == 38
+                && o_header_raw.len() == 38
+                && s_header_raw == o_header_raw
+                && parse_canonical_v3_header_fields(s_header_raw).is_some()
+            {
+                continue;
+            }
+            let mut s_header = FileMetaVersionHeader::default();
+            if s_header.unmarshal_v(s_header_ver, s_header_raw).is_err() {
+                return false;
+            }
+            let mut o_header = FileMetaVersionHeader::default();
+            if o_header.unmarshal_v(o_header_ver, o_header_raw).is_err() {
+                return false;
+            }
+            if !Self::headers_match_strict(&s_header, &o_header) {
+                return false;
+            }
+        }
+
+        true
+    }
+
     #[inline]
     fn header_matches_canonical_v3_raw_strict(s_header: &FileMetaVersionHeader, buf: &[u8]) -> Option<bool> {
         if buf.len() != 38 {
@@ -625,6 +682,49 @@ impl FileMeta {
         }
     }
 
+    pub fn bytes_count_toward_listing_limit(buf: &[u8]) -> bool {
+        let Ok((meta, _inline_buf)) = Self::parse_validated_xl_envelope(buf) else {
+            return true;
+        };
+        if meta.is_empty() {
+            return false;
+        }
+        let Ok((versions_len, header_ver, _meta_ver, meta)) = Self::decode_validated_xl_headers(meta) else {
+            return true;
+        };
+        if versions_len == 0 {
+            return false;
+        }
+        let mut cur = Cursor::new(meta);
+        let mut selected: Option<(FileMetaVersionHeader, &[u8])> = None;
+        let mut any_non_hidden = false;
+        for _ in 0..versions_len {
+            let Ok((header, ver_meta)) = Self::read_next_shallow_slice(meta, &mut cur, header_ver) else {
+                return true;
+            };
+            if header.version_type != VersionType::Delete && !header.free_version() {
+                any_non_hidden = true;
+            }
+            if selected.is_some() {
+                continue;
+            }
+            if header.free_version() && header.version_id.is_some_and(|id| !id.is_nil()) {
+                continue;
+            }
+            selected = Some((header, ver_meta));
+        }
+        let Some((header, ver_meta)) = selected else {
+            return false;
+        };
+        if !header.free_version() && header.version_type != VersionType::Delete {
+            return true;
+        }
+        if !any_non_hidden {
+            return false;
+        }
+        FileMetaVersion::decode_fileinfo_from_slice(ver_meta, "", "", true, false).is_err()
+    }
+
     #[hotpath::measure(impl_type = "FileMeta")]
     pub fn marshal_msg(&self) -> Result<Vec<u8>> {
         let mut wr =

crates/filemeta/src/metacache.rs

@@ -352,6 +352,22 @@ impl MetaCacheEntry {
         (prefer, true)
     }
 
+    pub fn matches_strict_borrowed(&mut self, other: &MetaCacheEntry) -> bool {
+        if self.name != other.name {
+            return false;
+        }
+
+        if other.is_dir() || self.is_dir() {
+            return other.is_dir() == self.is_dir();
+        }
+
+        if self.cached.is_none() && other.cached.is_none() {
+            return FileMeta::buffers_match_strict(&self.metadata, &other.metadata);
+        }
+
+        self.matches_strict(other)
+    }
+
     pub fn matches_strict(&mut self, other: &MetaCacheEntry) -> bool {
         if self.name != other.name {
             return false;
@@ -989,6 +1005,17 @@ impl<W: AsyncWrite + Unpin> MetacacheWriter<W> {
         Ok(())
     }
 
+    pub async fn write_obj_buffered(&mut self, obj: &MetaCacheEntry) -> Result<()> {
+        self.init().await?;
+
+        Self::encode_obj(&mut self.buf, obj)?;
+        if self.buf.len() >= METACACHE_READ_CHUNK_SIZE {
+            self.flush().await?;
+        }
+
+        Ok(())
+    }
+
     pub async fn close(&mut self) -> Result<()> {
         self.close_with_limit_reached(false).await
     }
@@ -1282,6 +1309,85 @@ impl<R: AsyncRead + Unpin> MetacacheReader<R> {
         Ok(u32::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
     }
 
+    pub fn skip_current(&mut self) -> bool {
+        if self.has_current && self.init && self.err.is_none() {
+            self.has_current = false;
+            true
+        } else {
+            false
+        }
+    }
+
+    pub fn try_peek_buffered(&mut self) -> Option<Result<Option<MetaCacheEntry>>> {
+        if !self.init || self.err.is_some() || self.terminated {
+            return None;
+        }
+
+        let avail = &self.buf[self.pos..];
+        let &first = avail.first()?;
+        if first != 0xc3 {
+            return None;
+        }
+
+        let str_marker = *avail.get(1)?;
+        let (str_hdr, str_len) = match str_marker {
+            0xa0..=0xbf => (1usize, usize::from(str_marker & 0x1f)),
+            0xd9 => (2usize, usize::from(*avail.get(2)?)),
+            0xda => (3usize, usize::from(u16::from_be_bytes([*avail.get(2)?, *avail.get(3)?]))),
+            0xdb => (
+                5usize,
+                usize::try_from(u32::from_be_bytes([*avail.get(2)?, *avail.get(3)?, *avail.get(4)?, *avail.get(5)?])).ok()?,
+            ),
+            _ => return None,
+        };
+
+        let bin_offset = 1usize.checked_add(str_hdr)?.checked_add(str_len)?;
+        let bin_marker = *avail.get(bin_offset)?;
+        let (bin_hdr, bin_len) = match bin_marker {
+            0xc4 => (2usize, usize::from(*avail.get(bin_offset + 1)?)),
+            0xc5 => (
+                3usize,
+                usize::from(u16::from_be_bytes([*avail.get(bin_offset + 1)?, *avail.get(bin_offset + 2)?])),
+            ),
+            0xc6 => (
+                5usize,
+                usize::try_from(u32::from_be_bytes([
+                    *avail.get(bin_offset + 1)?,
+                    *avail.get(bin_offset + 2)?,
+                    *avail.get(bin_offset + 3)?,
+                    *avail.get(bin_offset + 4)?,
+                ]))
+                .ok()?,
+            ),
+            _ => return None,
+        };
+
+        let total_len = bin_offset.checked_add(bin_hdr)?.checked_add(bin_len)?;
+        if avail.len() < total_len {
+            return None;
+        }
+
+        let Ok(decoded) = from_utf8(&avail[1 + str_hdr..bin_offset]) else {
+            return None;
+        };
+        let name = decoded.to_owned();
+        let metadata = if bin_len == 0 {
+            Vec::new()
+        } else {
+            avail[bin_offset + bin_hdr..total_len].to_vec()
+        };
+
+        self.pos += total_len;
+        self.reset();
+        self.has_current = true;
+        Some(Ok(Some(MetaCacheEntry {
+            name,
+            metadata,
+            cached: None,
+            reusable: false,
+        })))
+    }
+
     pub async fn skip(&mut self, size: usize) -> Result<()> {
         self.check_init().await?;
 
@@ -4257,4 +4363,95 @@ mod tests {
             .expect("resolve mixed empty");
         assert_eq!(res_mixed_empty.cached.expect("cached").versions.len(), 1);
     }
+
+    #[test]
+    fn matches_strict_borrowed_matches_shallow_headers_and_rejects_corrupt_or_divergent_buffers() {
+        let base_bytes = create_real_xlmeta().expect("create real xlmeta");
+        let mut peer_fm = FileMeta::load(&base_bytes).expect("load base xlmeta");
+        // Mutate inner version payload (ver.meta) without changing shallow header.
+        let mut ver = FileMetaVersion::try_from(peer_fm.versions[1].meta.as_slice()).expect("decode version");
+        ver.object.as_mut().expect("object version").erasure_index += 1;
+        peer_fm.versions[1].meta = ver.marshal_msg().expect("marshal peer version");
+        let peer_bytes = peer_fm.marshal_msg().expect("marshal peer xlmeta");
+        assert_ne!(base_bytes, peer_bytes, "peer shard has distinct erasure index payload");
+
+        let mut e1 = MetaCacheEntry {
+            name: "obj-a".to_owned(),
+            metadata: base_bytes.clone(),
+            cached: None,
+            reusable: false,
+        };
+        let e1_clone = e1.clone();
+        let e2 = MetaCacheEntry {
+            name: "obj-a".to_owned(),
+            metadata: peer_bytes,
+            cached: None,
+            reusable: false,
+        };
+        assert!(e1.matches_strict_borrowed(&e2));
+        assert!(e1.matches_strict_borrowed(&e1_clone));
+
+        // Different shallow header must not match.
+        let mut newer_fm = FileMeta::load(&base_bytes).expect("load base xlmeta");
+        newer_fm.versions[0].header.version_id = Some(Uuid::new_v4());
+        let newer_entry = MetaCacheEntry {
+            name: "obj-a".to_owned(),
+            metadata: newer_fm.marshal_msg().expect("marshal newer xlmeta"),
+            cached: None,
+            reusable: false,
+        };
+        assert!(!e1.matches_strict_borrowed(&newer_entry));
+
+        // Identical corrupt buffers must return false, never true via byte-equality shortcut.
+        let mut corrupt = MetaCacheEntry {
+            name: "obj-a".to_owned(),
+            metadata: b"XL2 corrupt-payload".to_vec(),
+            cached: None,
+            reusable: false,
+        };
+        let corrupt_peer = corrupt.clone();
+        assert!(!corrupt.matches_strict_borrowed(&corrupt_peer));
+    }
+
+    #[tokio::test]
+    async fn try_peek_buffered_and_skip_current_match_async_peek_and_skip() {
+        let entries: Vec<_> = (0..5)
+            .map(|i| MetaCacheEntry {
+                name: format!("buffered-key-{i}"),
+                metadata: vec![i as u8; 16],
+                cached: None,
+                reusable: false,
+            })
+            .collect();
+        let mut buf = Cursor::new(Vec::new());
+        let mut writer = MetacacheWriter::new(&mut buf);
+        writer.write(&entries).await.expect("write entries");
+        writer.close_with_limit_reached(false).await.expect("close stream");
+
+        let mut reader = MetacacheReader::new(Cursor::new(buf.into_inner()));
+        assert!(reader.try_peek_buffered().is_none(), "uninitialized reader must defer to async peek");
+        assert!(!reader.skip_current(), "no current entry before first peek");
+
+        let first = reader.peek().await.expect("first peek").expect("entry 0");
+        assert_eq!(first, entries[0]);
+        assert!(reader.skip_current(), "skip_current must consume the peeked entry");
+        assert!(!reader.skip_current(), "second skip_current without peek must return false");
+
+        for expected in &entries[1..] {
+            let next = reader
+                .try_peek_buffered()
+                .expect("remaining entries are already buffered")
+                .expect("valid entry")
+                .expect("present entry");
+            assert_eq!(&next, expected);
+            assert!(reader.skip_current());
+        }
+
+        assert!(
+            reader.try_peek_buffered().is_none(),
+            "stream terminator must defer to async peek for trailer handling"
+        );
+        assert!(reader.peek().await.expect("eof peek").is_none());
+        assert_eq!(reader.limit_reached(), Some(false));
+    }
 }

rustfs/src/admin/console_test.rs

@@ -43,9 +43,9 @@ mod tests {
             .map(|value| value.to_str().expect("allow-origin header must be valid UTF-8").to_string())
     }
 
-    /// Runs a request through a router that returns a brotli-encoded response without
-    /// `Content-Length` and returns the response's `content-length` header value, if any.
-    async fn brotli_response_content_length(has_content_length: bool) -> Option<String> {
+    /// Sends a brotli-encoded response through `buffer_brotli_responses` and returns
+    /// the `Content-Length` header value and body bytes.
+    async fn run_brotli_middleware(body: &'static str, has_content_length: bool) -> (Option<String>, bytes::Bytes) {
         use crate::admin::console::buffer_brotli_responses;
         use axum::middleware;
         use axum::{Router, body::Body, routing::get};
@@ -57,9 +57,9 @@ mod tests {
                 get(move || async move {
                     let mut builder = http::Response::builder().header(http::header::CONTENT_ENCODING, "br");
                     if has_content_length {
-                        builder = builder.header(http::header::CONTENT_LENGTH, "4");
+                        builder = builder.header(http::header::CONTENT_LENGTH, body.len().to_string());
                     }
-                    builder.body(Body::from("data")).expect("response must build")
+                    builder.body(Body::from(body)).expect("response must build")
                 }),
             )
             .layer(middleware::from_fn(buffer_brotli_responses));
@@ -74,27 +74,130 @@ mod tests {
             .await
             .expect("request must not fail");
 
-        response
+        let content_length = response
             .headers()
             .get(http::header::CONTENT_LENGTH)
-            .map(|value| value.to_str().expect("content-length header must be valid UTF-8").to_string())
+            .map(|value| value.to_str().expect("content-length header must be valid UTF-8").to_string());
+        let body_bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
+            .await
+            .expect("body must be readable");
+        (content_length, body_bytes)
     }
 
     #[tokio::test]
     #[serial]
     async fn test_buffer_brotli_responses() {
-        // Brotli response without Content-Length must have Content-Length added.
-        assert!(
-            brotli_response_content_length(false).await.is_some(),
-            "brotli response without Content-Length must have Content-Length added"
+        // Brotli response without Content-Length must have Content-Length added with correct value.
+        let (content_length, body) = run_brotli_middleware("fake-brotli-data", false).await;
+        assert_eq!(
+            content_length.as_deref(),
+            Some("16"),
+            "brotli response without Content-Length must have Content-Length set to the body length"
+        );
+        assert_eq!(
+            body.as_ref(),
+            b"fake-brotli-data",
+            "body bytes must be preserved through the buffer round-trip"
         );
 
         // Brotli response that already carries Content-Length must pass through unchanged.
+        let (content_length, body) = run_brotli_middleware("data", true).await;
         assert_eq!(
-            brotli_response_content_length(true).await.as_deref(),
+            content_length.as_deref(),
             Some("4"),
             "brotli response that already has Content-Length must not be re-buffered"
         );
+        assert_eq!(body.as_ref(), b"data", "body bytes must be preserved when passing through unchanged");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_buffer_brotli_responses_non_brotli_passthrough() {
+        use crate::admin::console::buffer_brotli_responses;
+        use axum::middleware;
+        use axum::{Router, body::Body, routing::get};
+        use tower::ServiceExt;
+
+        // Non-brotli response must pass through with no modification.
+        let app = Router::new()
+            .route(
+                "/",
+                get(|| async {
+                    http::Response::builder()
+                        .header(http::header::CONTENT_ENCODING, "gzip")
+                        .body(Body::from("gzip-data"))
+                        .expect("response must build")
+                }),
+            )
+            .layer(middleware::from_fn(buffer_brotli_responses));
+
+        let response = app
+            .oneshot(
+                http::Request::builder()
+                    .uri("/")
+                    .body(Body::empty())
+                    .expect("request must build"),
+            )
+            .await
+            .expect("request must not fail");
+
+        assert_eq!(
+            response
+                .headers()
+                .get(http::header::CONTENT_ENCODING)
+                .and_then(|v| v.to_str().ok()),
+            Some("gzip"),
+            "non-brotli Content-Encoding must pass through unchanged"
+        );
+        let body_bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
+            .await
+            .expect("body must be readable");
+        assert_eq!(body_bytes.as_ref(), b"gzip-data", "body bytes must pass through unchanged");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_buffer_brotli_responses_body_stream_error() {
+        use crate::admin::console::buffer_brotli_responses;
+        use axum::middleware;
+        use axum::{Router, body::Body, routing::get};
+        use tower::ServiceExt;
+
+        // A brotli response whose body stream errors must have Content-Encoding stripped
+        // so the client does not attempt to decode an empty body as brotli.
+        let app = Router::new()
+            .route(
+                "/",
+                get(|| async {
+                    let broken = futures::stream::once(async {
+                        Err::<bytes::Bytes, std::io::Error>(std::io::Error::other("simulated stream error"))
+                    });
+                    http::Response::builder()
+                        .header(http::header::CONTENT_ENCODING, "br")
+                        .body(Body::from_stream(broken))
+                        .expect("response must build")
+                }),
+            )
+            .layer(middleware::from_fn(buffer_brotli_responses));
+
+        let response = app
+            .oneshot(
+                http::Request::builder()
+                    .uri("/")
+                    .body(Body::empty())
+                    .expect("request must build"),
+            )
+            .await
+            .expect("request must not fail");
+
+        assert!(
+            response.headers().get(http::header::CONTENT_ENCODING).is_none(),
+            "Content-Encoding must be stripped when body buffering fails so the client does not decode an empty body as brotli"
+        );
+        let body_bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
+            .await
+            .expect("body must be readable");
+        assert!(body_bytes.is_empty(), "body must be empty after buffering failure");
     }
 
     #[tokio::test]

rustfs/src/app/multipart_usecase.rs

@@ -39,6 +39,7 @@ use super::storage_api::multipart_usecase::contract::object::ObjectOperations as
 use super::storage_api::multipart_usecase::contract::range::HTTPRangeSpec;
 use super::storage_api::multipart_usecase::data_usage::{
     quota_object_size, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
+    record_bucket_object_write_unknown_previous_memory,
 };
 use super::storage_api::multipart_usecase::error::{
     StorageError, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found,
@@ -690,25 +691,45 @@ impl DefaultMultipartUsecase {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
-        let current_opts = internal_object_info_lookup_opts(
-            get_opts(&bucket, &key, None, None, &req.headers)
-                .await
-                .map_err(ApiError::from)?,
-        );
+        let quota_metadata_sys = self.bucket_metadata_sys();
+        let quota_tracking = quota_metadata_sys.is_some();
+        let mut quota_enabled = false;
+        if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
+            let quota_checker = QuotaChecker::new(metadata_sys.clone());
+            let check_result =
+                map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)?;
+            quota_enabled = check_result.quota_limit.is_some();
+            apply_quota_admission(&mut opts, &check_result)?;
+        }
+
         let object_lock_config_state = Box::pin(load_bucket_object_lock_config_state(&bucket)).await?;
-        let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
-            Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)?;
-                let physical_size = existing_obj_info.size.max(0) as u64;
-                let logical_size = quota_object_size(&existing_obj_info);
-                Some((physical_size, logical_size))
-            }
-            Err(err) => {
-                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
-                    return Err(ApiError::from(err).into());
+        let prelookup_required = quota_enabled
+            || (!super::object::shared::put_like_write_creates_new_version(&opts)
+                && super::object::shared::object_lock_checks_required_for_state(&object_lock_config_state));
+        let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
+            let current_opts = internal_object_info_lookup_opts(
+                get_opts(&bucket, &key, None, None, &req.headers)
+                    .await
+                    .map_err(ApiError::from)?,
+            );
+            match store.get_object_info(&bucket, &key, &current_opts).await {
+                Ok(existing_obj_info) => {
+                    validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)?;
+                    Some(Some(if quota_enabled {
+                        quota_object_size(&existing_obj_info).map_err(ApiError::from)?
+                    } else {
+                        existing_obj_info.size.max(0) as u64
+                    }))
+                }
+                Err(err) => {
+                    if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                        return Err(ApiError::from(err).into());
+                    }
+                    Some(None)
                 }
-                None
             }
+        } else {
+            None
         };
 
         // A lost Complete response retries the same upload id after staging is
@@ -746,24 +767,6 @@ impl DefaultMultipartUsecase {
         let upload_user_defined = multipart_info.as_ref().map(|info| info.user_defined.clone());
         let table_data_plane_publication_guards = retained_table_data_plane_publication_guards(&req);
 
-        let quota_metadata_sys = self.bucket_metadata_sys();
-        let quota_tracking = quota_metadata_sys.is_some();
-        let mut quota_enabled = false;
-        if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
-            let quota_checker = QuotaChecker::new(metadata_sys.clone());
-            let check_result =
-                map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)?;
-            quota_enabled = check_result.quota_limit.is_some();
-            apply_quota_admission(&mut opts, &check_result)?;
-        }
-
-        let previous_current_size = match previous_current_sizes {
-            Some((_, Ok(logical_size))) if quota_enabled => Some(logical_size),
-            Some((_, Err(err))) if quota_enabled => return Err(ApiError::from(err).into()),
-            Some((physical_size, _)) => Some(physical_size),
-            None => None,
-        };
-
         // Replication admission is decided at the object-creation commit, not
         // at multipart-session creation. Configuration and matching targets
         // can change while parts are uploaded, so persist the exact decision's
@@ -831,11 +834,20 @@ impl DefaultMultipartUsecase {
                 if !obj_info.multipart_completion_replayed {
                     if quota_tracking {
                         let committed_size = quota_accounting_object_size(&obj_info, quota_enabled)?;
-
-                        if versioned {
-                            record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
-                        } else {
-                            record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
+                        match prelookup_previous_current_size
+                            .or_else(|| super::object::previous_current_size_from_backfill(obj_info.backfilled_old_current_size))
+                        {
+                            Some(previous_current_size) => {
+                                if versioned {
+                                    record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size)
+                                        .await;
+                                } else {
+                                    record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
+                                }
+                            }
+                            None => {
+                                record_bucket_object_write_unknown_previous_memory(&bucket, committed_size, versioned).await;
+                            }
                         }
                     }
 

rustfs/src/app/object/mod.rs

@@ -230,7 +230,10 @@ pub(crate) use self::on_demand_migration_put::*;
 #[cfg(test)]
 pub(crate) use self::put::PollCountingBody;
 use self::put::*;
-pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout, reject_oversize_single_upload};
+pub(crate) use self::put::{
+    guard_put_object_body_read_timeout, previous_current_size_from_backfill, put_object_body_read_timeout,
+    reject_oversize_single_upload,
+};
 #[cfg(test)]
 pub(crate) use self::restore::RestoreStatusCommitBarrier;
 pub(crate) use self::shared::*;

rustfs/src/app/object/put.rs

@@ -2324,7 +2324,7 @@ impl DefaultObjectUsecase {
 /// `previous_current_size` value the usage-accounting helpers expect. Outer
 /// `None` = unknown (no quorum agreement, or a peer predates the field) — the
 /// caller must fall back to the degraded accounting path.
-pub(super) fn previous_current_size_from_backfill(backfill: Option<OldCurrentSize>) -> Option<Option<u64>> {
+pub(crate) fn previous_current_size_from_backfill(backfill: Option<OldCurrentSize>) -> Option<Option<u64>> {
     backfill.map(|observation| match observation {
         OldCurrentSize::Present(size) => Some(size.max(0) as u64),
         OldCurrentSize::Absent => None,

rustfs/src/app/object/shared.rs

@@ -874,7 +874,7 @@ pub(crate) fn build_put_like_object_lock_metadata(
     Ok(Some(eval_metadata))
 }
 
-pub(super) fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool {
+pub(in crate::app) fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool {
     opts.version_id.is_none() && opts.versioned && !opts.version_suspended
 }
 
@@ -997,7 +997,7 @@ pub(crate) async fn object_lock_checks_required(bucket: &str) -> bool {
         .map_or(true, |metadata| metadata.object_lock_checks_required())
 }
 
-pub(super) fn object_lock_checks_required_for_state(state: &metadata_sys::ObjectLockConfigState) -> bool {
+pub(in crate::app) fn object_lock_checks_required_for_state(state: &metadata_sys::ObjectLockConfigState) -> bool {
     match state {
         metadata_sys::ObjectLockConfigState::Configured { .. } | metadata_sys::ObjectLockConfigState::Fabricated => true,
         metadata_sys::ObjectLockConfigState::ConfirmedAbsent => false,