fix(ecstore): preserve multipart shards within write budgets

.docker/observability/otel-collector-config.yaml

@@ -28,15 +28,6 @@ processors:
     check_interval: 1s
     limit_mib: 1024
     spike_limit_mib: 256
-  transform/metrics:
-    error_mode: ignore
-    metric_statements:
-      # RustFS process metrics already expose process_pid as a metric label.
-      # Remove only the duplicate datapoint label and keep the shared resource
-      # process.pid available to sibling metrics and lock histograms.
-      - context: datapoint
-        statements:
-          - delete_key(attributes, "process_pid") where IsMatch(metric.name, "^rustfs_system_process_")
   transform/logs:
     error_mode: ignore
     log_statements:
@@ -121,7 +112,7 @@ service:
       exporters: [ otlp/tempo, otlp/jaeger ]
     metrics:
       receivers: [ otlp ]
-      processors: [ transform/metrics, memory_limiter, batch ]
+      processors: [ batch ]
       exporters: [ prometheus ]
     logs:
       receivers: [ otlp ]

crates/ecstore/src/core/pool_meta_recovery_codec.rs

@@ -315,7 +315,7 @@ pub(super) async fn encode_inline_pool_meta_replicas(
         .encode_with_shard_integrity(
             Cursor::new(Bytes::copy_from_slice(canonical)),
             &mut writers,
-            raw.len(),
+            crate::erasure::coding::encode::ShardWritePolicy::object(raw.len()),
             1,
             IntegrityEncodeMode::Inline(canonical.len()),
             protected,

crates/ecstore/src/disk/local.rs

@@ -3159,7 +3159,6 @@ pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
             == Some(dst_path)
         || os::has_dir_fsync_test_failures()
         || os::file_sync_probe::is_active()
-        || os::prepared_publication_test_hooks::has_any_hooks()
 }
 
 #[cfg(all(not(test), feature = "test-util", target_os = "linux"))]
@@ -3170,7 +3169,6 @@ pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
         .iter()
         .any(|target| target == dst_path)
         || os::has_dir_fsync_test_failures()
-        || os::prepared_publication_test_hooks::has_any_hooks()
 }
 
 #[cfg(any(not(any(test, feature = "test-util")), not(target_os = "linux")))]
@@ -4200,43 +4198,48 @@ fn reclaim_std_file_cache(file: &std::fs::File, reclaim_len: usize, reclaim_on_s
 #[cfg(not(target_os = "linux"))]
 fn reclaim_std_file_cache(_file: &std::fs::File, _reclaim_len: usize, _reclaim_on_shutdown: bool) {}
 
-const PRESYNCED_TMP_DATA_DIRS_PER_SHARD: usize = 64;
-const PRESYNCED_TMP_FILES_PER_SHARD: usize = 128;
+static PRESYNCED_TMP_DATA_DIRS: std::sync::LazyLock<ParkingLotMutex<HashSet<PathBuf>>> =
+    std::sync::LazyLock::new(|| ParkingLotMutex::new(HashSet::new()));
 
-static PRESYNCED_TMP_DATA_DIRS: std::sync::LazyLock<os::ShardedPathCache<()>> =
-    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PRESYNCED_TMP_DATA_DIRS_PER_SHARD));
+static PRESYNCED_TMP_FILES: std::sync::LazyLock<ParkingLotMutex<HashSet<PathBuf>>> =
+    std::sync::LazyLock::new(|| ParkingLotMutex::new(HashSet::new()));
 
-static PRESYNCED_TMP_FILES: std::sync::LazyLock<os::ShardedPathCache<()>> =
-    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PRESYNCED_TMP_FILES_PER_SHARD));
-
-fn mark_presynced_tmp_data_dir(dir: impl AsRef<Path>) {
-    PRESYNCED_TMP_DATA_DIRS.insert(dir.as_ref(), ());
+fn mark_presynced_tmp_data_dir(dir: PathBuf) {
+    let mut set = PRESYNCED_TMP_DATA_DIRS.lock();
+    if set.len() >= 4096 {
+        set.clear();
+    }
+    set.insert(dir);
 }
 
 fn unmark_presynced_tmp_data_dir(dir: &Path) -> bool {
-    PRESYNCED_TMP_DATA_DIRS.remove(dir).is_some()
+    PRESYNCED_TMP_DATA_DIRS.lock().remove(dir)
 }
 
 #[cfg(target_os = "linux")]
 pub(in crate::disk) fn is_presynced_tmp_data_dir(dir: &Path) -> bool {
-    PRESYNCED_TMP_DATA_DIRS.with_get(dir, |()| ()).is_some()
+    PRESYNCED_TMP_DATA_DIRS.lock().contains(dir)
 }
 
 #[cfg(target_os = "linux")]
 pub(in crate::disk) fn take_presynced_tmp_data_dir(dir: &Path) -> bool {
-    PRESYNCED_TMP_DATA_DIRS.remove(dir).is_some()
+    PRESYNCED_TMP_DATA_DIRS.lock().remove(dir)
 }
 
-fn mark_presynced_tmp_file(file: impl AsRef<Path>) {
-    PRESYNCED_TMP_FILES.insert(file.as_ref(), ());
+fn mark_presynced_tmp_file(file: PathBuf) {
+    let mut set = PRESYNCED_TMP_FILES.lock();
+    if set.len() >= 8192 {
+        set.clear();
+    }
+    set.insert(file);
 }
 
 fn unmark_presynced_tmp_file(file: &Path) {
-    let _ = PRESYNCED_TMP_FILES.remove(file);
+    PRESYNCED_TMP_FILES.lock().remove(file);
 }
 
 fn take_presynced_tmp_file(file: &Path) -> bool {
-    PRESYNCED_TMP_FILES.remove(file).is_some()
+    PRESYNCED_TMP_FILES.lock().remove(file)
 }
 
 fn open_lazy_local_file_std(file_path: &Path, volume_dir: &Path, _reclaim_on_shutdown: bool) -> std::io::Result<std::fs::File> {
@@ -4322,7 +4325,7 @@ impl LazyLocalFileWriter {
                         drop(file);
                         if sync_on_close {
                             if mark_dir_presynced && let Some(parent) = file_path.parent() {
-                                mark_presynced_tmp_data_dir(parent);
+                                mark_presynced_tmp_data_dir(parent.to_path_buf());
                             }
                             mark_presynced_tmp_file(file_path);
                         }
@@ -4344,7 +4347,7 @@ impl LazyLocalFileWriter {
                         drop(file);
                         if sync_on_close {
                             if mark_dir_presynced && let Some(parent) = file_path.parent() {
-                                mark_presynced_tmp_data_dir(parent);
+                                mark_presynced_tmp_data_dir(parent.to_path_buf());
                             }
                             mark_presynced_tmp_file(file_path);
                         }
@@ -4472,7 +4475,7 @@ impl AsyncWrite for LazyLocalFileWriter {
                         drop(file);
                         if sync_on_close {
                             if mark_dir_presynced && let Some(parent) = file_path.parent() {
-                                mark_presynced_tmp_data_dir(parent);
+                                mark_presynced_tmp_data_dir(parent.to_path_buf());
                             }
                             mark_presynced_tmp_file(file_path);
                         }
@@ -22180,7 +22183,7 @@ mod test {
                 .get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}"))
                 .expect("staged shard directory should resolve");
             assert!(
-                PRESYNCED_TMP_DATA_DIRS.with_get(&staged_dir, |()| ()).is_none(),
+                !PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir),
                 "an unknown destination must defer payload sync until commit"
             );
 
@@ -22233,10 +22236,7 @@ mod test {
             let staged_dir = disk
                 .get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}"))
                 .expect("staged shard directory should resolve");
-            assert!(
-                PRESYNCED_TMP_DATA_DIRS.with_get(&staged_dir, |()| ()).is_some(),
-                "the writer must pre-sync the shard"
-            );
+            assert!(PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir), "the writer must pre-sync the shard");
             assert!(!os::fsync_dir_recorder::was_fsynced(&staged_dir));
 
             let mut fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None);
@@ -22247,7 +22247,7 @@ mod test {
                 .await
                 .expect("rename_data should commit");
             assert!(
-                PRESYNCED_TMP_DATA_DIRS.with_get(&staged_dir, |()| ()).is_none(),
+                !PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir),
                 "the fused commit must consume the pre-sync marker"
             );
             assert!(
@@ -22438,141 +22438,6 @@ mod test {
         assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(1)));
     }
 
-    #[cfg(target_os = "linux")]
-    #[tokio::test]
-    #[allow(clippy::await_holding_lock)]
-    async fn test_rename_data_fused_overwrite_and_rollback_double_failure() {
-        const CHILD: &str = "RUSTFS_TEST_FUSED_OVERWRITE_CHILD";
-        const TEST: &str = "disk::local::test::test_rename_data_fused_overwrite_and_rollback_double_failure";
-
-        // Other tests install process-wide hooks that deliberately force rename_data off the fused path.
-        if std::env::var_os(CHILD).is_none() {
-            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
-                .args(["--exact", TEST, "--nocapture", "--test-threads=1"])
-                .env(CHILD, "1")
-                .env("RUSTFS_EXPERIMENTAL_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE", "false")
-                .env("RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_ENABLE", "false")
-                .output()
-                .expect("fused overwrite subprocess");
-            let stdout = String::from_utf8_lossy(&output.stdout);
-            let stderr = String::from_utf8_lossy(&output.stderr);
-            assert!(output.status.success(), "{stdout}{stderr}");
-            assert!(stdout.contains("FUSED_OVERWRITE_OK"), "fused overwrite test did not run: {stdout}");
-            return;
-        }
-
-        assert!(!os::dst_dir_fsync_group_commit_enabled());
-        assert!(!os::file_fdatasync_group_commit_enabled());
-        let _mode = durability_mode_override::set(DurabilityMode::Strict);
-        let dir = tempfile::tempdir().expect("temp dir");
-        let endpoint = Endpoint::try_from(dir.path().to_str().expect("utf8")).expect("endpoint");
-        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
-        let (bucket, object, version_id) = ("fused-overwrite-bucket", "fused-object", Uuid::nil());
-        let (data_dir_v1, data_dir_v2) = (Uuid::new_v4(), Uuid::new_v4());
-        ensure_test_volume(&disk, bucket).await;
-        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
-
-        for (tmp, data_dir, payload) in [("tmp-v1", data_dir_v1, b"v1-d"), ("tmp-v2", data_dir_v2, b"v2-d")] {
-            let mut w = disk
-                .create_file(bucket, RUSTFS_META_TMP_BUCKET, &format!("{tmp}/{data_dir}/part.1"), 4)
-                .await
-                .expect("create shard");
-            w.write_all(payload).await.expect("write");
-            w.shutdown().await.expect("close");
-        }
-        let mut fi_v1 = test_file_info(object, version_id, Some(data_dir_v1), None);
-        (fi_v1.size, fi_v1.parts[0].size, fi_v1.parts[0].actual_size) = (4, 4, 4);
-        let resp1 = disk
-            .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-v1", fi_v1, bucket, object)
-            .await
-            .expect("commit v1");
-        assert_eq!((resp1.old_data_dir, resp1.rollback_data_dir), (None, None));
-        let dst_object_dir = disk.io_get_object_path(bucket, object).expect("dst object path");
-        let v1_meta_bytes = fs::read(dst_object_dir.join(STORAGE_FORMAT_FILE))
-            .await
-            .expect("read v1 xl.meta");
-
-        let mut fi_v2 = test_file_info(object, version_id, Some(data_dir_v2), None);
-        (fi_v2.size, fi_v2.parts[0].size, fi_v2.parts[0].actual_size) = (4, 4, 4);
-        assert!(!has_rename_data_test_failpoints(object));
-        let resp2 = disk
-            .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-v2", fi_v2, bucket, object)
-            .await
-            .expect("fused overwrite v2");
-        assert_eq!(
-            (resp2.old_data_dir, resp2.rollback_data_dir, resp2.cleanup_data_dir),
-            (Some(data_dir_v1), Some(data_dir_v1), Some(data_dir_v1))
-        );
-        assert_eq!(
-            fs::read(dst_object_dir.join(data_dir_v1.to_string()).join(STORAGE_FORMAT_FILE_BACKUP))
-                .await
-                .expect("v1 backup"),
-            v1_meta_bytes
-        );
-
-        fs::create_dir_all(dir.path().join(RUSTFS_META_TMP_BUCKET).join("tmp-v3"))
-            .await
-            .expect("create tmp-v3");
-        let fi_v3 = test_file_info(object, version_id, None, Some(Bytes::from_static(b"v3-inline")));
-        let resp3 = disk
-            .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-v3", fi_v3, bucket, object)
-            .await
-            .expect("fused inline overwrite v3");
-        assert_eq!(
-            (resp3.old_data_dir, resp3.rollback_data_dir, resp3.cleanup_data_dir),
-            (Some(data_dir_v2), Some(data_dir_v2), Some(data_dir_v2))
-        );
-        assert!(
-            dst_object_dir
-                .join(data_dir_v2.to_string())
-                .join(STORAGE_FORMAT_FILE_BACKUP)
-                .is_file()
-        );
-
-        let dst_meta_path = dst_object_dir.join(STORAGE_FORMAT_FILE);
-        let dst_vol_dir = disk.io_get_bucket_path(bucket).expect("bucket path");
-        let lease = os::acquire_namespace_mutation_lease_with_owner(&dst_meta_path, None).await;
-        let guard =
-            commit::lock_rename_commit_directories(&dst_object_dir, &dst_object_dir, &dst_vol_dir, &disk.publication_root, lease)
-                .await
-                .expect("lock commit guard");
-        let replacement_dir = dst_object_dir.join(Uuid::new_v4().to_string());
-        fs::create_dir_all(&replacement_dir).await.expect("create replacement_dir");
-        fs::write(replacement_dir.join("part.1"), b"live").await.expect("write part");
-
-        os::fsync_dir_recorder::set_failure(&dst_object_dir, std::io::ErrorKind::Other);
-        assert!(
-            commit::fsync_inline_commit_std(
-                &dst_meta_path,
-                &dst_vol_dir,
-                false,
-                Some(Uuid::new_v4()),
-                None,
-                Some(&replacement_dir),
-                &guard,
-            )
-            .is_err()
-        );
-        assert!(replacement_dir.join("part.1").is_file());
-
-        os::fsync_dir_recorder::set_failure(&dst_object_dir, std::io::ErrorKind::Other);
-        assert!(
-            commit::fsync_inline_commit_std(
-                &dst_meta_path,
-                &dst_vol_dir,
-                false,
-                Some(data_dir_v1),
-                None,
-                Some(&replacement_dir),
-                &guard,
-            )
-            .is_err()
-        );
-        assert!(!replacement_dir.exists());
-        assert_eq!(fs::read(&dst_meta_path).await.expect("restored"), v1_meta_bytes);
-        println!("FUSED_OVERWRITE_OK");
-    }
-
     /// rustfs/backlog#1009: `observe_old_current_size` must mirror the pre-PUT
     /// `get_object_info` semantics bit for bit — latest version's
     /// `ObjectInfo.size` (0 for a delete-marker latest, which that lookup

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

@@ -15,8 +15,6 @@
 //! Single-disk object rename publication and rollback. The shared execution core
 //! retains instrumentation, mutation leases, and commit guards through the syscall.
 
-#[cfg(target_os = "linux")]
-use super::has_rename_data_test_failpoints;
 #[cfg(all(test, windows))]
 use super::run_destination_commit_directory_preparation;
 #[cfg(any(not(windows), test))]
@@ -98,7 +96,7 @@ struct RenameCommitContext<'a> {
     data_paths: Option<&'a (PathBuf, PathBuf)>,
 }
 
-pub(super) struct StagedRollbackBackup(PathBuf);
+struct StagedRollbackBackup(PathBuf);
 
 impl StagedRollbackBackup {
     fn path(&self) -> &Path {
@@ -235,13 +233,11 @@ fn rollback_inline_metadata_std(
     }
 }
 
-pub(super) fn fsync_inline_commit_std(
+fn fsync_inline_commit_std(
     dst: &Path,
     dst_volume_dir: &Path,
     destination_was_absent: bool,
-    rollback_data_dir: Option<Uuid>,
     local_backup: Option<&StagedRollbackBackup>,
-    new_data_dir: Option<&Path>,
     commit_guard: &os::RenameCommitGuard,
 ) -> std::io::Result<()> {
     if let Some(dst_parent) = dst.parent() {
@@ -251,10 +247,7 @@ pub(super) fn fsync_inline_commit_std(
                 rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                 fsync_started,
             );
-            rollback_inline_metadata_std(dst, rollback_data_dir, local_backup)?;
-            if let Some(new_data_path) = new_data_dir {
-                remove_dir_all_if_exists(new_data_path)?;
-            }
+            rollback_inline_metadata_std(dst, None, local_backup)?;
             return Err(err);
         }
         rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -275,10 +268,7 @@ pub(super) fn fsync_inline_commit_std(
                     rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                     fsync_started,
                 );
-                rollback_inline_metadata_std(dst, rollback_data_dir, local_backup)?;
-                if let Some(new_data_path) = new_data_dir {
-                    remove_dir_all_if_exists(new_data_path)?;
-                }
+                rollback_inline_metadata_std(dst, None, local_backup)?;
                 return Err(err);
             }
             rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -837,7 +827,7 @@ impl LocalDisk {
                         None
                     }
                 });
-                if rollback_data_dir.is_some() && has_rename_data_test_failpoints(&dst_path_str) {
+                if rollback_data_dir.is_some() {
                     return Ok(FusedCommitAttempt::Fallback(guard, has_dst_buf, Box::new(fi_moved)));
                 }
                 #[cfg(test)]
@@ -860,9 +850,6 @@ impl LocalDisk {
                     &dst_path_str,
                     version_id,
                 );
-                if let Some(old_data_dir) = has_old_data_dir.as_ref() {
-                    let _ = xlmeta.data.remove_two(version_id, *old_data_dir);
-                }
                 xlmeta.add_version(fi_moved)?;
                 let version_signature = rename_data_versions_signature(&xlmeta);
                 let new_dst_buf = xlmeta.marshal_msg().map_err(std::io::Error::other)?;
@@ -882,9 +869,8 @@ impl LocalDisk {
                 }
 
                 let mut staged_rollback_path = None;
-                if (!no_inline || rollback_data_dir.is_none())
-                    && let Some(ref old_metadata) = has_dst_buf
-                    && (rollback_data_dir.is_some() || sync_commit_metadata || cfg!(test))
+                if let Some(ref old_metadata) = has_dst_buf
+                    && (sync_commit_metadata || cfg!(test))
                 {
                     let backup_path = create_local_inline_rollback_backup(
                         &dst_file_path_owned,
@@ -985,52 +971,9 @@ impl LocalDisk {
                         }
                         return Ok(FusedCommitAttempt::RestoreDataSource(guard, DiskError::Unexpected));
                     }
-                    if let (Some(old_data_dir), Some(dst_buf)) = (rollback_data_dir, has_dst_buf.as_ref()) {
-                        let backup_parent = dst_parent_owned.join(old_data_dir.to_string());
-                        if let Err(err) =
-                            os::mkdir_all_below_existing_base_std(&backup_parent, &skip_parent, &publication_root_owned)
-                        {
-                            return Ok(FusedCommitAttempt::RestoreDataSource(guard, to_file_error(err).into()));
-                        }
-                        let backup_path_guard = match guard.create_destination_directory_for_path_access(&backup_parent) {
-                            Ok(g) => g,
-                            Err(err) => return Ok(FusedCommitAttempt::RestoreDataSource(guard, to_file_error(err).into())),
-                        };
-                        let backup_path = backup_parent.join(STORAGE_FORMAT_FILE_BACKUP);
-                        if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) {
-                            return Ok(FusedCommitAttempt::RestoreDataSource(guard, err));
-                        }
-                        #[cfg(test)]
-                        run_owned_file_before_open(&backup_path);
-                        if let Err(err) = backup_path_guard.write_file_for_path_access(
-                            &backup_path,
-                            dst_buf.as_ref(),
-                            sync_commit_metadata,
-                            sync_commit_metadata,
-                        ) {
-                            return Ok(FusedCommitAttempt::RestoreDataSource(guard, to_file_error(err).into()));
-                        }
-                    }
                     if crash_inject::should_crash_at(CrashPoint::RenameAfterBackupBeforeMetaCommit, &dst_path_str) {
                         return Ok(FusedCommitAttempt::CommitErr(DiskError::Unexpected));
                     }
-                } else if let (Some(old_data_dir), Some(staged_backup)) =
-                    (rollback_data_dir, staged_rollback_path.as_ref().map(StagedRollbackBackup::path))
-                {
-                    let backup_parent = dst_parent_owned.join(old_data_dir.to_string());
-                    let backup_path = backup_parent.join(STORAGE_FORMAT_FILE_BACKUP);
-                    if let Err(err) = os::rename_all_with_precondition_and_parent_sync_std(
-                        staged_backup,
-                        &backup_path,
-                        &dst_volume_dir_owned,
-                        &publication_root_owned,
-                        None,
-                        sync_commit_metadata,
-                    ) {
-                        let _ = remove_file_if_exists(staged_backup);
-                        return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
-                    }
-                    staged_rollback_path = None;
                 }
 
                 let rename_res = if should_fail_commit_rename(&dst_path_str) {
@@ -1058,16 +1001,15 @@ impl LocalDisk {
 
                 if should_fail_after_metadata_commit(&dst_path_str) {
                     if no_inline {
-                        rollback_inline_metadata_std(&dst_file_path_owned, rollback_data_dir, staged_rollback_path.as_ref())?;
+                        rollback_inline_metadata_std(&dst_file_path_owned, None, staged_rollback_path.as_ref())?;
                         if let Some((_, new_data_path)) = has_data_dir_path_owned.as_ref() {
                             remove_dir_all_if_exists(new_data_path)?;
                         }
                         return Ok(FusedCommitAttempt::CommitErr(DiskError::Unexpected));
                     }
-                    let err =
-                        rollback_inline_metadata_std(&dst_file_path_owned, rollback_data_dir, staged_rollback_path.as_ref())
-                            .err()
-                            .unwrap_or_else(|| std::io::Error::other("test fail after metadata commit"));
+                    let err = rollback_inline_metadata_std(&dst_file_path_owned, None, staged_rollback_path.as_ref())
+                        .err()
+                        .unwrap_or_else(|| std::io::Error::other("test fail after metadata commit"));
                     return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
                 }
 
@@ -1080,16 +1022,14 @@ impl LocalDisk {
                         &dst_file_path_owned,
                         &dst_volume_dir_owned,
                         destination_was_absent,
-                        rollback_data_dir,
                         staged_rollback_path.as_ref(),
-                        has_data_dir_path_owned
-                            .as_ref()
-                            .filter(|_| no_inline)
-                            .map(|(_, new_data_path)| new_data_path.as_path()),
                         &guard,
                     )
                 {
                     if no_inline {
+                        if let Some((_, new_data_path)) = has_data_dir_path_owned.as_ref() {
+                            remove_dir_all_if_exists(new_data_path)?;
+                        }
                         return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
                     }
                     return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
@@ -1105,9 +1045,9 @@ impl LocalDisk {
 
                 Ok(FusedCommitAttempt::Committed(Box::new(RenameDataResp {
                     multipart_source: None,
-                    old_data_dir: has_old_data_dir,
-                    rollback_data_dir,
-                    cleanup_data_dir: has_old_data_dir,
+                    old_data_dir: None,
+                    rollback_data_dir: None,
+                    cleanup_data_dir: None,
                     sign: version_signature,
                     old_current_size,
                     old_current_source_checked,
@@ -1596,9 +1536,8 @@ impl LocalDisk {
                     let dst_file_path = dst_file_path.clone();
                     let owned_directory_guard = commit.directory_guard.clone();
                     let owned_mutation_lease = commit.mutation_lease.clone();
-                    let sync = tmp_meta_sync != SyncMode::None;
                     async move {
-                        os::run_blocking_namespace_operation_with_sync(owned_mutation_lease, sync, 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(
@@ -1606,7 +1545,7 @@ impl LocalDisk {
                                 &dst_file_path,
                                 &owned_directory_guard,
                             )?;
-                            prepared_metadata_source.write_all(&new_dst_buf, sync)?;
+                            prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?;
                             Ok(prepared_metadata_source)
                         })
                         .await
@@ -1769,15 +1708,13 @@ impl LocalDisk {
                 // sync finish. A detached spawn_blocking writer could survive
                 // cancellation and later truncate a newer transaction's
                 // deterministic rollback backup.
-                let lease = commit.mutation_lease.clone();
-                let sync = backup_sync != SyncMode::None;
-                let write_result = os::run_blocking_namespace_operation_with_sync(lease, sync, move || {
+                let write_result = os::run_blocking_namespace_operation(commit.mutation_lease.clone(), move || {
                     #[cfg(test)]
                     run_owned_file_before_open(&backup_path);
                     backup_path_guard.write_file_for_path_access(
                         &backup_path,
                         backup_bytes.as_ref(),
-                        sync,
+                        backup_sync != SyncMode::None,
                         backup_sync == SyncMode::FileAndDir,
                     )
                 })
@@ -2217,9 +2154,7 @@ impl LocalDisk {
                             &dst,
                             &dst_volume_dir_for_commit,
                             destination_was_absent,
-                            rollback_data_dir,
                             staged_rollback_path.as_ref(),
-                            None,
                             &rename_commit_guard_for_preparation,
                         )
                     {

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

@@ -271,9 +271,6 @@ pub(in crate::disk) fn generation(mut canonical: FileInfo) -> Result<crate::disk
         compat::SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
         compat::SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
     ] {
-        if canonical.metadata.is_empty() {
-            break;
-        }
         if !compat::contains_key_str(&canonical.metadata, suffix) {
             continue;
         }
@@ -304,24 +301,22 @@ pub(in crate::disk) fn generation(mut canonical: FileInfo) -> Result<crate::disk
         }
         compat::insert_str(&mut canonical.metadata, suffix, value);
     }
-    fn sort_json_in_place(value: &mut serde_json::Value) {
+    fn ordered(value: serde_json::Value) -> serde_json::Value {
         match value {
             serde_json::Value::Object(map) => {
-                map.sort_keys();
-                map.values_mut().for_each(sort_json_in_place);
+                let sorted: std::collections::BTreeMap<_, _> = map.into_iter().collect();
+                serde_json::Value::Object(sorted.into_iter().map(|(key, value)| (key, ordered(value))).collect())
             }
-            serde_json::Value::Array(values) => values.iter_mut().for_each(sort_json_in_place),
-            _ => {}
+            serde_json::Value::Array(values) => serde_json::Value::Array(values.into_iter().map(ordered).collect()),
+            other => other,
         }
     }
-    let mut json_value = serde_json::to_value(&canonical)?;
-    sort_json_in_place(&mut json_value);
     Ok(crate::disk::DeleteRecoveryVersionIdentity {
         version_id: canonical.version_id,
         mod_time: canonical.mod_time,
         data_dir: canonical.data_dir,
         deleted: canonical.deleted,
-        metadata_digest: digest(&serde_json::to_vec(&json_value)?),
+        metadata_digest: digest(&serde_json::to_vec(&ordered(serde_json::to_value(canonical)?))?),
     })
 }
 
@@ -432,13 +427,6 @@ fn journal_bytes(journal: &DeleteRecoveryJournal) -> Result<Vec<u8>> {
     if bytes.len() > usize::try_from(MAX_METADATA).map_err(|_| DiskError::FileCorrupt)? - 128 {
         return Err(DiskError::FileCorrupt);
     }
-    // MessagePack encoding of `DeleteRecoveryRpcReply::Inspected` is more compact
-    // than JSON; only construct and serialize `projected` when the JSON size plus
-    // `current_versions` and fixed envelope could approach the RPC payload limit.
-    let rpc_budget = rustfs_protos::DELETE_RECOVERY_MAX_PAYLOAD_BYTES.saturating_sub(4096);
-    if bytes.len().saturating_mul(2).saturating_add(4096) <= rpc_budget {
-        return Ok(bytes);
-    }
     let projected = crate::disk::DeleteRecoveryRpcReply::Inspected(DeleteRecoveryInspection {
         protocol_version: DELETE_RECOVERY_PROTOCOL_VERSION,
         id: DeleteTransactionId {
@@ -455,7 +443,7 @@ fn journal_bytes(journal: &DeleteRecoveryJournal) -> Result<Vec<u8>> {
         explicit_delete_safe: false,
         fingerprint: [0; 32],
     });
-    if rmp_serde::to_vec_named(&projected)?.len() > rpc_budget {
+    if rmp_serde::to_vec_named(&projected)?.len() > rustfs_protos::DELETE_RECOVERY_MAX_PAYLOAD_BYTES - 4096 {
         return Err(DiskError::FileCorrupt);
     }
     Ok(bytes)
@@ -7180,86 +7168,6 @@ mod tests {
         assert!(!root.join(data_dir.to_string()).exists(), "physical data_dir must be removed");
         assert!(!root.exists(), "empty object directory must be pruned");
     }
-
-    #[cfg(all(any(test, feature = "test-util"), not(windows)))]
-    #[tokio::test]
-    async fn delete_recovery_receipt_formatting_and_post_finished_unlink_interruption() {
-        let (_temp, disk, info, _) = fixture().await;
-        let root = disk.io_get_object_path("recovery", "object").expect("object path");
-
-        let mut v1 = info.clone();
-        v1.version_id = Some(Uuid::new_v4());
-        let mut v2 = info.clone();
-        v2.version_id = Some(Uuid::new_v4());
-        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 mut rev_info = v1.clone();
-        rev_info
-            .metadata
-            .extend([("x-amz-meta-z".into(), "1".into()), ("x-amz-meta-a".into(), "2".into())]);
-        let mut fwd_info = v1.clone();
-        fwd_info
-            .metadata
-            .extend([("x-amz-meta-a".into(), "2".into()), ("x-amz-meta-z".into(), "1".into())]);
-        assert_eq!(
-            generation(rev_info).expect("rev").metadata_digest,
-            generation(fwd_info).expect("fwd").metadata_digest
-        );
-
-        let transaction = Uuid::new_v4();
-        let opts = DeleteOptions {
-            recoverable_delete: true,
-            ..Default::default()
-        };
-        assert_eq!(
-            disk.try_fused_recoverable_delete_version("recovery", "object", &v2, &opts, transaction, None)
-                .await
-                .expect("fused versioned delete preparation"),
-            Some(())
-        );
-        let (journal, sole_transaction, target_satisfied) = disk
-            .try_publish_prepared_delete_local("recovery", "object", transaction, None)
-            .await
-            .expect("local publish")
-            .expect("versioned target satisfied");
-        assert!(sole_transaction && target_satisfied && journal.target_digest.is_some());
-
-        let rollback = root.join(transaction.to_string());
-        // Real finalizer failure at the second fsync_dir_std(&rollback) (after [TARGET, PREPARED, FINISHED] unlink)
-        // must leave a durable finished=true receipt that inspects as CommitFinished.
-        let rollback_for_hook = rollback.clone();
-        let _hook = os::prepared_publication_test_hooks::install_at(
-            os::prepared_publication_test_hooks::Stage::Remove,
-            &rollback.join(FINISHED),
-            move || {
-                os::fsync_dir_recorder::set_failure(&rollback_for_hook, std::io::ErrorKind::Other);
-            },
-        );
-        assert!(
-            disk.try_finish_and_finalize_prepared_delete_local("recovery", "object", transaction, &journal, None)
-                .await
-                .is_err()
-        );
-        assert!(!rollback.join(FINISHED).exists());
-        assert_eq!(
-            disk.inspect_delete_transaction("recovery", "object", transaction)
-                .await
-                .expect("post-FINISHED-unlink failure")
-                .state,
-            DeleteRecoveryState::CommitFinished
-        );
-        assert!(
-            disk.try_finish_and_finalize_prepared_delete_local("recovery", "object", transaction, &journal, None)
-                .await
-                .expect("finalize")
-        );
-        assert!(!rollback.exists());
-    }
 }
 
 #[cfg(test)]

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

@@ -398,9 +398,7 @@ impl LocalDisk {
         let source_path = source_path.to_owned();
         let incoming = incoming.clone();
         let sync_directory = effective_durability(destination_volume).syncs_data_shards();
-        let sync_lane = sync_directory && incoming.data.is_none() && !incoming.is_remote();
-        let lease = Arc::clone(&staging.lease);
-        let fresh_head = os::run_cancellable_blocking_namespace_operation_with_sync(lease, sync_lane, move |cancelled| {
+        let fresh_head = os::run_cancellable_blocking_namespace_operation(Arc::clone(&staging.lease), move |cancelled| {
             if cancelled() {
                 return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "multipart staging cancelled"));
             }

crates/ecstore/src/disk/os.rs

@@ -3808,7 +3808,7 @@ pub(crate) async fn rename_reconciled_metadata(
     authority: Arc<crate::bucket::lifecycle::legacy_transition_state_reconcile::TransitionStateReconcileAuthority>,
 ) -> Result<()> {
     let lease = acquire_namespace_mutation_lease_with_owner(&destination, owner).await;
-    run_blocking_namespace_operation_with_sync(lease, true, move || {
+    run_blocking_namespace_operation(lease, move || {
         let preparation = prepare_rename_with_retry(&source, &destination, &base_dir, &publication_root)?;
         #[cfg(all(any(test, feature = "test-util"), not(windows)))]
         prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &destination);
@@ -4808,23 +4808,6 @@ pub(crate) async fn run_cancellable_blocking_namespace_operation<T: Send + 'stat
         .await
 }
 
-pub(crate) async fn run_cancellable_blocking_namespace_operation_with_sync<T: Send + 'static>(
-    lease: Arc<NamespaceMutationLease>,
-    sync: bool,
-    operation: impl FnOnce(&dyn Fn() -> bool) -> io::Result<T> + Send + 'static,
-) -> io::Result<T> {
-    if !sync {
-        return run_cancellable_blocking_namespace_operation(lease, operation).await;
-    }
-    let queue = Arc::clone(&lease.blocking_queue);
-    queue
-        .run_fsync_cancellable(move |is_cancelled| {
-            let _lease = lease;
-            operation(is_cancelled)
-        })
-        .await
-}
-
 /// Admit one strict inline commit under the disk sync limit. The caller already
 /// owns the namespace lease, establishing namespace -> disk ordering. Holding
 /// admission across adjacent durability barriers prevents one transaction from

crates/ecstore/src/erasure/coding/encode.rs

@@ -326,6 +326,40 @@ impl Default for WriteProgressPolicy {
     }
 }
 
+/// Caller-specific completion behavior for an erasure write.
+///
+/// A multipart upload must not discard a different healthy shard from each
+/// part merely because another quorum finished first: the final object needs
+/// a quorum of disks carrying every part. Pending writers still obey the
+/// existing progress and absolute deadlines and never count as completed.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ShardWritePolicy {
+    quorum: usize,
+    post_quorum: PostQuorumPolicy,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PostQuorumPolicy {
+    Grace,
+    ExistingBudget,
+}
+
+impl ShardWritePolicy {
+    pub(crate) fn object(quorum: usize) -> Self {
+        Self {
+            quorum,
+            post_quorum: PostQuorumPolicy::Grace,
+        }
+    }
+
+    pub(crate) fn multipart(quorum: usize) -> Self {
+        Self {
+            quorum,
+            post_quorum: PostQuorumPolicy::ExistingBudget,
+        }
+    }
+}
+
 /// After write quorum is already in hand, how long a slower shard may still
 /// finish before it is dropped. The per-shard stall budget (30s by default)
 /// is what a black-hole peer is allowed to consume *before* quorum exists.
@@ -338,13 +372,15 @@ fn post_quorum_straggler_grace(stall_timeout: Option<std::time::Duration>) -> Op
     stall_timeout.map(|stall| stall.min(WRITE_QUORUM_STRAGGLER_GRACE))
 }
 
-/// Drive shard ops until write quorum has succeeded and stragglers have had
-/// [`post_quorum_straggler_grace`] to finish. Returns `true` for each slot
-/// whose future completed; `false` means the caller must fail that slot.
+/// Drive shard operations within their existing deadlines. Ordinary object
+/// writes additionally stop after the post-quorum grace; multipart writes keep
+/// polling all candidates within the same operation budget. A `false` slot
+/// never completed and must be excluded from commit.
 async fn await_write_quorum<F>(
     write_quorum: usize,
     stall_timeout: Option<std::time::Duration>,
     slot_count: usize,
+    post_quorum: PostQuorumPolicy,
     mut futures: FuturesUnordered<F>,
 ) -> Vec<bool>
 where
@@ -376,6 +412,7 @@ where
         if succeeded {
             successes = successes.saturating_add(1);
             if successes >= write_quorum
+                && post_quorum == PostQuorumPolicy::Grace
                 && grace.is_none()
                 && let Some(delay) = post_quorum_straggler_grace(stall_timeout)
             {
@@ -404,6 +441,7 @@ pub(crate) struct MultiWriter<'a> {
     writers: &'a mut [Option<BitrotWriterWrapper>],
     integrity: Option<&'a mut IntegrityBuilder>,
     write_quorum: usize,
+    post_quorum: PostQuorumPolicy,
     errs: Vec<Option<Error>>,
     policy: WriteProgressPolicy,
     /// Absolute deadline for the whole object, armed lazily on the first
@@ -412,16 +450,21 @@ pub(crate) struct MultiWriter<'a> {
 }
 
 impl<'a> MultiWriter<'a> {
-    pub fn new(writers: &'a mut [Option<BitrotWriterWrapper>], write_quorum: usize) -> Self {
-        Self::with_policy(writers, write_quorum, WriteProgressPolicy::default())
+    pub fn new(writers: &'a mut [Option<BitrotWriterWrapper>], write_policy: ShardWritePolicy) -> Self {
+        Self::with_policy(writers, write_policy, WriteProgressPolicy::default())
     }
 
-    pub fn with_policy(writers: &'a mut [Option<BitrotWriterWrapper>], write_quorum: usize, policy: WriteProgressPolicy) -> Self {
+    pub fn with_policy(
+        writers: &'a mut [Option<BitrotWriterWrapper>],
+        write_policy: ShardWritePolicy,
+        policy: WriteProgressPolicy,
+    ) -> Self {
         let length = writers.len();
         MultiWriter {
             writers,
             integrity: None,
-            write_quorum,
+            write_quorum: write_policy.quorum,
+            post_quorum: write_policy.post_quorum,
             errs: vec![None; length],
             policy,
             absolute_deadline: None,
@@ -506,9 +549,8 @@ impl<'a> MultiWriter<'a> {
                 // failed and its disk dropped, so a stalled peer cannot pin an
                 // otherwise-healthy write quorum (rustfs/backlog#1319). `budget`
                 // is recomputed per block, so it bounds a stall — not the total
-                // transfer time — and a slow-but-honest writer is never killed
-                // while quorum is still open. Once quorum has succeeded, a
-                // straggler only gets `WRITE_QUORUM_STRAGGLER_GRACE` more time.
+                // transfer time. Only ordinary object writes add the shorter
+                // post-quorum grace; multipart preserves this operation budget.
                 futures.push(async move {
                     match budget {
                         Some(budget) => match tokio::time::timeout(budget, Self::write_shard(writer_opt, err, shard)).await {
@@ -524,7 +566,7 @@ impl<'a> MultiWriter<'a> {
                     (index, succeeded)
                 });
             }
-            await_write_quorum(write_quorum, stall_timeout, slot_count, futures).await
+            await_write_quorum(write_quorum, stall_timeout, slot_count, self.post_quorum, futures).await
         };
         abandon_unfinished_writers(self.writers, &mut self.errs, &saw);
 
@@ -587,8 +629,8 @@ impl<'a> MultiWriter<'a> {
                 // here forever for a small object whose bytes were fully buffered
                 // (so `write` never blocked). Bound it with the same progress
                 // budget and drop the stalled writer before the quorum check.
-                // After quorum has shut down, do not keep waiting out the rest of
-                // that budget for a peer that has stopped answering.
+                // Ordinary object writes also apply their post-quorum grace;
+                // multipart waits for all candidates within this same budget.
                 futures.push(async move {
                     match budget {
                         Some(budget) => match tokio::time::timeout(budget, Self::shutdown_writer(writer_opt, err)).await {
@@ -604,7 +646,7 @@ impl<'a> MultiWriter<'a> {
                     (index, succeeded)
                 });
             }
-            await_write_quorum(write_quorum, stall_timeout, slot_count, futures).await
+            await_write_quorum(write_quorum, stall_timeout, slot_count, self.post_quorum, futures).await
         };
         abandon_unfinished_writers(self.writers, &mut self.errs, &saw);
 
@@ -639,7 +681,7 @@ impl Erasure {
         self: Arc<Self>,
         reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         part_number: usize,
         mode: IntegrityEncodeMode,
         protected: bool,
@@ -668,19 +710,19 @@ impl Erasure {
             }
             IntegrityEncodeMode::SingleBlock(hint) => {
                 let (reader, size) = self
-                    .encode_small_direct(reader, writers, quorum, true, hint, integrity.as_mut())
+                    .encode_small_direct(reader, writers, write_policy, true, hint, integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
             IntegrityEncodeMode::Streaming => {
                 let (reader, size) = self
-                    .encode_with_ingest_mode(reader, writers, quorum, use_bytesmut_ingest(), integrity.as_mut())
+                    .encode_with_ingest_mode(reader, writers, write_policy, use_bytesmut_ingest(), integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
             IntegrityEncodeMode::Batched => {
                 let (reader, size) = self
-                    .encode_batched_with_integrity(reader, writers, quorum, integrity.as_mut())
+                    .encode_batched_with_integrity(reader, writers, write_policy, integrity.as_mut())
                     .await?;
                 (reader, size, None)
             }
@@ -745,7 +787,7 @@ impl Erasure {
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         require_single_block: bool,
         size_hint: usize,
         integrity: Option<&mut IntegrityBuilder>,
@@ -780,7 +822,7 @@ impl Erasure {
         }
 
         let block = self.encode_data_owned_block(buf)?;
-        let mut mw = MultiWriter::new(writers, quorum);
+        let mut mw = MultiWriter::new(writers, write_policy);
         mw.integrity = integrity;
         mw.write_block(&block).await?;
         mw.shutdown().await?;
@@ -847,7 +889,7 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin + 'static,
     {
         let use_bytesmut_ingest = use_bytesmut_ingest();
-        self.encode_with_ingest_mode(reader, writers, quorum, use_bytesmut_ingest, None)
+        self.encode_with_ingest_mode(reader, writers, ShardWritePolicy::object(quorum), use_bytesmut_ingest, None)
             .await
     }
 
@@ -859,7 +901,7 @@ impl Erasure {
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         use_bytesmut_ingest: bool,
         integrity: Option<&mut IntegrityBuilder>,
     ) -> std::io::Result<(R, usize)>
@@ -961,7 +1003,7 @@ impl Erasure {
             Ok((reader, total))
         }));
 
-        let mut writers = MultiWriter::new(writers, quorum);
+        let mut writers = MultiWriter::new(writers, write_policy);
         writers.integrity = integrity;
 
         let mut write_err = None;
@@ -1017,14 +1059,15 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin + 'static,
     {
-        self.encode_batched_with_integrity(reader, writers, quorum, None).await
+        self.encode_batched_with_integrity(reader, writers, ShardWritePolicy::object(quorum), None)
+            .await
     }
 
     async fn encode_batched_with_integrity<R>(
         self: Arc<Self>,
         mut reader: R,
         writers: &mut [Option<BitrotWriterWrapper>],
-        quorum: usize,
+        write_policy: ShardWritePolicy,
         integrity: Option<&mut IntegrityBuilder>,
     ) -> std::io::Result<(R, usize)>
     where
@@ -1105,7 +1148,7 @@ impl Erasure {
             Ok((reader, total))
         }));
 
-        let mut writers = MultiWriter::new(writers, quorum);
+        let mut writers = MultiWriter::new(writers, write_policy);
         writers.integrity = integrity;
         let mut write_err = None;
 
@@ -1165,7 +1208,7 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin,
     {
         let size_hint = self.block_size;
-        self.encode_small_direct(reader, writers, quorum, false, size_hint, None)
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), false, size_hint, None)
             .await
     }
 
@@ -1182,7 +1225,7 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, false, size_hint, None)
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), false, size_hint, None)
             .await
     }
 
@@ -1199,7 +1242,8 @@ impl Erasure {
         R: AsyncRead + Send + Sync + Unpin,
     {
         let size_hint = self.block_size;
-        self.encode_small_direct(reader, writers, quorum, true, size_hint, None).await
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), true, size_hint, None)
+            .await
     }
 
     /// Size-aware single-block fast path. `size_hint` only controls the bounded
@@ -1215,7 +1259,8 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, true, size_hint, None).await
+        self.encode_small_direct(reader, writers, ShardWritePolicy::object(quorum), true, size_hint, None)
+            .await
     }
 }
 
@@ -1632,6 +1677,219 @@ mod tests {
         }
     }
 
+    #[derive(Clone, Copy, Debug)]
+    enum DelayedPhase {
+        Write,
+        Flush,
+        Shutdown,
+    }
+
+    struct PhaseDelayedWriter {
+        inner: DeferredCommitWriter,
+        phase: DelayedPhase,
+        delay: Duration,
+        timer: Option<Pin<Box<tokio::time::Sleep>>>,
+        ready: bool,
+    }
+
+    impl PhaseDelayedWriter {
+        fn poll_delay(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+            if self.ready {
+                return Poll::Ready(());
+            }
+            let timer = self.timer.get_or_insert_with(|| Box::pin(tokio::time::sleep(self.delay)));
+            match timer.as_mut().poll(cx) {
+                Poll::Pending => Poll::Pending,
+                Poll::Ready(()) => {
+                    self.ready = true;
+                    Poll::Ready(())
+                }
+            }
+        }
+    }
+
+    impl AsyncWrite for PhaseDelayedWriter {
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bytes: &[u8]) -> Poll<std::io::Result<usize>> {
+            if matches!(self.phase, DelayedPhase::Write) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_write(cx, bytes)
+        }
+
+        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            if matches!(self.phase, DelayedPhase::Flush) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_flush(cx)
+        }
+
+        fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            if matches!(self.phase, DelayedPhase::Shutdown) && self.poll_delay(cx).is_pending() {
+                return Poll::Pending;
+            }
+            Pin::new(&mut self.inner).poll_shutdown(cx)
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_keeps_slow_writers_in_every_encode_mode() {
+        const BLOCK: usize = 64;
+        for make_mode in [
+            || IntegrityEncodeMode::SingleBlock(BLOCK),
+            || IntegrityEncodeMode::Streaming,
+            || IntegrityEncodeMode::Batched,
+        ] {
+            for phase in [DelayedPhase::Write, DelayedPhase::Flush, DelayedPhase::Shutdown] {
+                for delay in [Duration::from_secs(2), Duration::from_secs(5)] {
+                    let committed: Vec<_> = (0..4).map(|_| Arc::new(Mutex::new(Vec::new()))).collect();
+                    let mut writers = committed
+                        .iter()
+                        .enumerate()
+                        .map(|(slot, data)| {
+                            if slot == 0 {
+                                Some(bitrot_writer(
+                                    PhaseDelayedWriter {
+                                        inner: DeferredCommitWriter::new(data.clone()),
+                                        phase,
+                                        delay,
+                                        timer: None,
+                                        ready: false,
+                                    },
+                                    BLOCK / 2,
+                                ))
+                            } else {
+                                Some(bitrot_writer(DeferredCommitWriter::new(data.clone()), BLOCK / 2))
+                            }
+                        })
+                        .collect::<Vec<_>>();
+                    let payload = vec![0x5a; BLOCK];
+                    let erasure = Arc::new(Erasure::new(2, 2, BLOCK));
+                    let started = tokio::time::Instant::now();
+                    erasure
+                        .encode_with_shard_integrity(
+                            Cursor::new(payload.clone()),
+                            &mut writers,
+                            ShardWritePolicy::multipart(3),
+                            1,
+                            make_mode(),
+                            false,
+                        )
+                        .await
+                        .expect("multipart must keep finite slow writers within the existing budget");
+                    assert!(started.elapsed() >= delay, "must await the slow phase: {phase:?}");
+                    assert!(writers.iter().all(Option::is_some), "all four completed shards must survive");
+                    for data in &committed {
+                        let bytes = data.lock().expect("committed bytes").clone();
+                        crate::erasure::coding::bitrot_verify(
+                            Cursor::new(bytes.clone()),
+                            bytes.len(),
+                            BLOCK / 2,
+                            HashAlgorithm::HighwayHash256S,
+                            BLOCK / 2,
+                        )
+                        .await
+                        .expect("each completed shard must retain valid bitrot framing");
+                    }
+                }
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_expires_blackholes_without_restarting_deadlines() {
+        for shutdown in [false, true] {
+            for blackholes in [1, 2] {
+                for absolute in [Duration::ZERO, Duration::from_secs(3)] {
+                    let mut writers = (0..4)
+                        .map(|slot| {
+                            if slot < blackholes {
+                                Some(bitrot_writer_plain(
+                                    if shutdown {
+                                        StallingWriter::stalls_on_shutdown()
+                                    } else {
+                                        StallingWriter::stalls_on_write(0)
+                                    },
+                                    64,
+                                ))
+                            } else {
+                                Some(bitrot_writer_plain(DeferredCommitWriter::new(Arc::new(Mutex::new(Vec::new()))), 64))
+                            }
+                        })
+                        .collect::<Vec<_>>();
+                    let started = tokio::time::Instant::now();
+                    let result = {
+                        let mut writer = MultiWriter::with_policy(
+                            &mut writers,
+                            ShardWritePolicy::multipart(3),
+                            WriteProgressPolicy::new(Duration::from_secs(7), absolute),
+                        );
+                        if shutdown {
+                            writer.write(four_shards()).await.expect("buffered writes should complete");
+                            tokio::time::sleep(Duration::from_secs(1)).await;
+                            writer.shutdown().await
+                        } else {
+                            writer.write(four_shards()).await
+                        }
+                    };
+                    assert_eq!(result.is_ok(), blackholes == 1, "only a confirmed quorum may succeed");
+                    let expected = if absolute.is_zero() {
+                        Duration::from_secs(7 + u64::from(shutdown))
+                    } else {
+                        absolute
+                    };
+                    assert_eq!(started.elapsed(), expected, "budget must not restart at quorum or shutdown");
+                    assert!(writers[..blackholes].iter().all(Option::is_none));
+                    assert!(writers[blackholes..].iter().all(Option::is_some));
+                }
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn multipart_budget_cancellation_drops_pending_writers() {
+        struct DropSignal(Option<oneshot::Sender<()>>);
+        impl Drop for DropSignal {
+            fn drop(&mut self) {
+                if let Some(tx) = self.0.take() {
+                    let _ = tx.send(());
+                }
+            }
+        }
+        impl AsyncWrite for DropSignal {
+            fn poll_write(self: Pin<&mut Self>, _: &mut Context<'_>, bytes: &[u8]) -> Poll<std::io::Result<usize>> {
+                Poll::Ready(Ok(bytes.len()))
+            }
+            fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+            fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Pending
+            }
+        }
+        let (tx, rx) = oneshot::channel();
+        let (entered_tx, entered) = oneshot::channel();
+        let task = tokio::spawn(async move {
+            let mut writers = vec![Some(bitrot_writer_plain(DropSignal(Some(tx)), 64))];
+            writers.extend(
+                (0..3).map(|_| Some(bitrot_writer_plain(DeferredCommitWriter::new(Arc::new(Mutex::new(Vec::new()))), 64))),
+            );
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::multipart(3));
+            writer.write(four_shards()).await.expect("buffered writes should succeed");
+            let _ = entered_tx.send(());
+            writer.shutdown().await
+        });
+        entered.await.expect("write must reach shutdown");
+        tokio::task::yield_now().await;
+        assert!(!task.is_finished());
+        task.abort();
+        assert!(
+            task.await
+                .expect_err("caller cancellation should cancel the task")
+                .is_cancelled()
+        );
+        rx.await.expect("pending writer must be dropped with its owning task");
+    }
+
     /// A writer that always makes forward progress but each `poll_write` first
     /// waits `delay` (on the runtime timer). Under a paused virtual clock this
     /// deterministically models a slow-but-honest peer: it must never be killed
@@ -1704,12 +1962,16 @@ mod tests {
         let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
 
         let encode = match pipeline {
-            EncodePipeline::Vec => {
-                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await })
-            }
-            EncodePipeline::BytesMut => {
-                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await })
-            }
+            EncodePipeline::Vec => tokio::spawn(async move {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }),
+            EncodePipeline::BytesMut => tokio::spawn(async move {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }),
             EncodePipeline::Batched => tokio::spawn(async move { erasure.encode_batched(reader, &mut writers, 1).await }),
         };
 
@@ -1767,8 +2029,16 @@ mod tests {
         ))];
 
         let result = match pipeline {
-            EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await,
-            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await,
+            EncodePipeline::Vec => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }
+            EncodePipeline::BytesMut => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await,
         };
 
@@ -1841,8 +2111,16 @@ mod tests {
         let mut writers = vec![Some(bitrot_writer(FailAfterReaderErrorWriter { reader_failed }, BLOCK_SIZE))];
 
         let result = match pipeline {
-            EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false, None).await,
-            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await,
+            EncodePipeline::Vec => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
+                    .await
+            }
+            EncodePipeline::BytesMut => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+                    .await
+            }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await,
         };
 
@@ -1879,7 +2157,7 @@ mod tests {
         let erasure_for_task = erasure.clone();
         let encode = tokio::spawn(async move {
             erasure_for_task
-                .encode_with_ingest_mode(reader, &mut writers, 1, false, None)
+                .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), false, None)
                 .await
         });
 
@@ -2048,8 +2326,16 @@ mod tests {
             let erasure = Arc::new(Erasure::new(2, 2, 64));
             let result = match path {
                 "direct" => erasure.encode_single_block_non_inline(reader, &mut writers, 2).await,
-                "vec" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, false, None).await,
-                "bytesmut" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, true, None).await,
+                "vec" => {
+                    erasure
+                        .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(2), false, None)
+                        .await
+                }
+                "bytesmut" => {
+                    erasure
+                        .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(2), true, None)
+                        .await
+                }
                 "batched" => erasure.encode_batched(reader, &mut writers, 2).await,
                 _ => unreachable!(),
             };
@@ -2094,7 +2380,7 @@ mod tests {
     async fn multi_writer_short_write_fails_before_shutdown() {
         let mut writers = vec![Some(bitrot_writer(ShortWriteWriter, 32))];
         let err = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             writer
                 .write(vec![Bytes::from_static(b"short-write payload")])
                 .await
@@ -2111,7 +2397,7 @@ mod tests {
     async fn multi_writer_reports_fallback_summary_when_only_offline_writers_remain() {
         let mut writers = vec![None, None];
         let (err, summary) = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             let err = writer
                 .write(vec![Bytes::from_static(b"offline-a"), Bytes::from_static(b"offline-b")])
                 .await
@@ -2131,7 +2417,7 @@ mod tests {
         assert!(summary.contains("required=1"));
 
         let (shutdown_err, summary) = {
-            let mut writer = MultiWriter::new(&mut writers, 1);
+            let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(1));
             let err = writer
                 .shutdown()
                 .await
@@ -2155,7 +2441,7 @@ mod tests {
     async fn multi_writer_reports_quorum_failure_when_quorum_exceeds_writer_count() {
         let committed = Arc::new(Mutex::new(Vec::new()));
         let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed), 32))];
-        let mut writer = MultiWriter::new(&mut writers, 2);
+        let mut writer = MultiWriter::new(&mut writers, ShardWritePolicy::object(2));
 
         let err = writer
             .write(vec![Bytes::from_static(b"quorum impossible")])
@@ -2244,7 +2530,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("one stalled writer must still satisfy a 3/4 write quorum without hanging");
@@ -2271,7 +2557,7 @@ mod tests {
         let started = tokio::time::Instant::now();
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("quorum must succeed without waiting out the black-hole stall");
@@ -2304,7 +2590,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards())
                 .await
                 .expect("a straggler inside the grace window must still satisfy the write");
@@ -2329,7 +2615,7 @@ mod tests {
         ];
 
         let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-        let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+        let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
         let err = mw
             .write(four_shards())
             .await
@@ -2358,7 +2644,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             mw.write(four_shards()).await.expect("all writers accept the buffered shard");
             mw.shutdown()
                 .await
@@ -2379,7 +2665,7 @@ mod tests {
         ];
 
         let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-        let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+        let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
         mw.write(four_shards()).await.expect("all writers accept the buffered shard");
         let err = mw
             .shutdown()
@@ -2408,7 +2694,7 @@ mod tests {
 
         {
             let policy = WriteProgressPolicy::new(Duration::from_secs(5), Duration::ZERO);
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             for _ in 0..8 {
                 mw.write(four_full_shards(64))
                     .await
@@ -2438,7 +2724,7 @@ mod tests {
             // stall (10s) never fires for an 8s-per-block writer, but the 15s
             // absolute cap does once its aggregate engagement crosses the cap.
             let policy = WriteProgressPolicy::new(Duration::from_secs(10), Duration::from_secs(15));
-            let mut mw = MultiWriter::with_policy(&mut writers, 3, policy);
+            let mut mw = MultiWriter::with_policy(&mut writers, ShardWritePolicy::object(3), policy);
             for _ in 0..4 {
                 mw.write(four_full_shards(64))
                     .await
@@ -2583,7 +2869,7 @@ mod tests {
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
         let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
         let (_reader, written) = erasure
-            .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, true, None)
+            .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), true, None)
             .await
             .expect("BytesMut ingest path should encode the streaming payload");
 
@@ -2612,7 +2898,9 @@ mod tests {
         let reader = tokio::io::BufReader::new(Cursor::new(vec![0x5a; block_size * 2]));
 
         rustfs_io_metrics::set_put_stage_metrics_enabled(true);
-        let result = erasure.encode_with_ingest_mode(reader, &mut writers, 1, true, None).await;
+        let result = erasure
+            .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(1), true, None)
+            .await;
         rustfs_io_metrics::set_put_stage_metrics_enabled(false);
 
         let (_reader, written) = result.expect("bytesmut streaming encode should complete");
@@ -3230,12 +3518,12 @@ mod tests {
         let (_reader, total) = match pipeline {
             EncodePipeline::Vec => {
                 erasure
-                    .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, false, None)
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), false, None)
                     .await
             }
             EncodePipeline::BytesMut => {
                 erasure
-                    .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, true, None)
+                    .encode_with_ingest_mode(reader, &mut writers, ShardWritePolicy::object(DATA_SHARDS), true, None)
                     .await
             }
             EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, DATA_SHARDS).await,

crates/ecstore/src/erasure/coding/heal.rs

@@ -205,7 +205,7 @@ impl super::Erasure {
         // the survivors. Reconstruction correctness is still guaranteed by the
         // read-side quorum check and parity cross-checks below, which are unaffected.
         let write_quorum = 1;
-        let mut writers = MultiWriter::new(writers, write_quorum);
+        let mut writers = MultiWriter::new(writers, super::encode::ShardWritePolicy::object(write_quorum));
         let read_timeout = get_object_disk_read_timeout();
         let shard_file_size = self.shard_file_size(total_length as i64) as usize;
 

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

@@ -112,6 +112,10 @@ use std::time::Duration;
 use tokio::io::AsyncReadExt;
 use tokio::task::JoinSet;
 
+#[cfg(test)]
+#[path = "multipart_quorum_tests.rs"]
+mod multipart_writer_test_hook;
+
 // The erasure-set width bounds fan-out. Await every opener so errors retain
 // their disk slots and every successful writer remains owned until quorum is checked.
 async fn create_part_writers(
@@ -124,6 +128,13 @@ async fn create_part_writers(
         let Some(disk) = disk else {
             return (None, Some(DiskError::DiskNotFound));
         };
+        #[cfg(test)]
+        if let Some(result) = multipart_writer_test_hook::create_writer(disk, path, length, shard_size).await {
+            return match result {
+                Ok(writer) => (Some(writer), None),
+                Err(error) => (None, Some(error)),
+            };
+        }
         match create_bitrot_writer_with_orig_volume(
             false,
             Some(disk),
@@ -2644,8 +2655,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 SmallWritePath::PipelineBatchedLarge => IntegrityEncodeMode::Batched,
                 SmallWritePath::Inline | SmallWritePath::Pipeline => IntegrityEncodeMode::Streaming,
             };
+            let shard_write_policy = crate::erasure::coding::encode::ShardWritePolicy::multipart(write_quorum);
+            #[cfg(test)]
+            let shard_write_policy = multipart_writer_test_hook::write_policy(&shuffle_disks, write_quorum)
+                .unwrap_or(shard_write_policy);
             let (reader, w_size, _, integrity) = Arc::clone(&erasure)
-                .encode_with_shard_integrity(stream, &mut writers, write_quorum, part_id, mode, protected_upload)
+                .encode_with_shard_integrity(
+                    stream,
+                    &mut writers,
+                    shard_write_policy,
+                    part_id,
+                    mode,
+                    protected_upload,
+                )
                 .await?;
 
             if let Some(stage_start) = encode_stage_start {

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

@@ -0,0 +1,303 @@
+//! Exercise participant preservation through real multipart staging and commit.
+
+use super::*;
+use crate::disk::FileWriter;
+use crate::erasure::coding::CustomWriter;
+use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
+use crate::storage_api_contracts::object::ObjectIO as _;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll, Waker};
+use tokio::io::AsyncWrite;
+
+#[derive(Clone, Copy)]
+enum Fault {
+    DelayedShutdown,
+    WriteFailure,
+}
+
+struct Round {
+    fault: Fault,
+    legacy_policy: bool,
+    failed_slot: usize,
+    completed: AtomicUsize,
+    delayed_observed: AtomicBool,
+    delayed_waker: Mutex<Option<Waker>>,
+    delayed_dropped: AtomicBool,
+}
+
+type RegisteredWriter = (Arc<Round>, usize);
+static WRITER_ROUNDS: OnceLock<Mutex<HashMap<String, RegisteredWriter>>> = OnceLock::new();
+
+struct WriterRound {
+    endpoints: Vec<String>,
+    round: Arc<Round>,
+}
+
+impl WriterRound {
+    fn install(disks: &[DiskStore], failed_slot: usize, fault: Fault, legacy_policy: bool) -> Self {
+        let round = Arc::new(Round {
+            fault,
+            legacy_policy,
+            failed_slot,
+            completed: AtomicUsize::new(0),
+            delayed_observed: AtomicBool::new(false),
+            delayed_waker: Mutex::new(None),
+            delayed_dropped: AtomicBool::new(false),
+        });
+        let endpoints = disks.iter().map(|disk| disk.endpoint().to_string()).collect::<Vec<_>>();
+        let mut registry = WRITER_ROUNDS
+            .get_or_init(Mutex::default)
+            .lock()
+            .expect("writer hook registry");
+        for (slot, endpoint) in endpoints.iter().enumerate() {
+            assert!(registry.insert(endpoint.clone(), (round.clone(), slot)).is_none());
+        }
+        Self { endpoints, round }
+    }
+}
+
+pub(super) fn write_policy(
+    disks: &[Option<DiskStore>],
+    quorum: usize,
+) -> Option<crate::erasure::coding::encode::ShardWritePolicy> {
+    let registry = WRITER_ROUNDS.get()?.lock().expect("writer hook registry");
+    disks.iter().flatten().find_map(|disk| {
+        registry.get(&disk.endpoint().to_string()).and_then(|(round, _)| {
+            round
+                .legacy_policy
+                .then(|| crate::erasure::coding::encode::ShardWritePolicy::object(quorum))
+        })
+    })
+}
+
+impl Drop for WriterRound {
+    fn drop(&mut self) {
+        let mut registry = WRITER_ROUNDS
+            .get()
+            .expect("installed writer registry")
+            .lock()
+            .expect("writer hook registry");
+        for endpoint in &self.endpoints {
+            registry.remove(endpoint);
+        }
+    }
+}
+
+pub(super) async fn create_writer(
+    disk: &DiskStore,
+    path: &str,
+    length: i64,
+    shard_size: usize,
+) -> Option<disk::error::Result<BitrotWriterWrapper>> {
+    let registered = WRITER_ROUNDS
+        .get()?
+        .lock()
+        .expect("writer hook registry")
+        .get(&disk.endpoint().to_string())
+        .cloned();
+    let (round, slot) = registered?;
+    let length = usize::try_from(length).expect("the fixture writes known-size parts");
+    let encoded_length = coding::bitrot_shard_file_size(length, shard_size, HashAlgorithm::HighwayHash256S);
+    let inner = match disk
+        .create_file(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, path, encoded_length as i64)
+        .await
+    {
+        Ok(inner) => inner,
+        Err(error) => return Some(Err(error)),
+    };
+    let writer = ControlledWriter {
+        inner,
+        round,
+        slot,
+        delay: None,
+        completed: false,
+    };
+    Some(Ok(BitrotWriterWrapper::new(
+        CustomWriter::new_tokio_writer(Box::new(writer)),
+        shard_size,
+        HashAlgorithm::HighwayHash256S,
+    )))
+}
+
+struct ControlledWriter {
+    inner: FileWriter,
+    round: Arc<Round>,
+    slot: usize,
+    delay: Option<Pin<Box<tokio::time::Sleep>>>,
+    completed: bool,
+}
+
+impl Drop for ControlledWriter {
+    fn drop(&mut self) {
+        if self.slot == self.round.failed_slot && !self.completed {
+            self.round.delayed_dropped.store(true, Ordering::Release);
+        }
+    }
+}
+
+impl AsyncWrite for ControlledWriter {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bytes: &[u8]) -> Poll<io::Result<usize>> {
+        if self.slot == self.round.failed_slot && matches!(self.round.fault, Fault::WriteFailure) {
+            return Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "injected shard write failure")));
+        }
+        Pin::new(&mut self.inner).poll_write(cx, bytes)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        if self.slot == self.round.failed_slot && matches!(self.round.fault, Fault::DelayedShutdown) {
+            self.round.delayed_observed.store(true, Ordering::Release);
+            // Delay only after three real writers finish. This ensures the old
+            // post-quorum grace expires regardless of filesystem scheduling.
+            {
+                let mut waker = self.round.delayed_waker.lock().expect("delayed writer waker");
+                *waker = Some(cx.waker().clone());
+                if self.round.completed.load(Ordering::Acquire) < 3 {
+                    return Poll::Pending;
+                }
+            }
+            let delay = self
+                .delay
+                .get_or_insert_with(|| Box::pin(tokio::time::sleep(Duration::from_secs(2))));
+            if delay.as_mut().poll(cx).is_pending() {
+                return Poll::Pending;
+            }
+        }
+        let result = Pin::new(&mut self.inner).poll_shutdown(cx);
+        if matches!(result, Poll::Ready(Ok(()))) && !self.completed {
+            self.completed = true;
+            self.round.completed.fetch_add(1, Ordering::AcqRel);
+            if let Some(waker) = self.round.delayed_waker.lock().expect("delayed writer waker").take() {
+                waker.wake();
+            }
+        }
+        result
+    }
+}
+
+async fn upload_two_parts(
+    set: &Arc<SetDisks>,
+    disks: &[DiskStore],
+    bucket: &str,
+    object: &str,
+    fault: Fault,
+    legacy_policy: bool,
+) -> (String, Vec<CompletePart>, Vec<u8>, FileInfo) {
+    for disk in disks {
+        disk.make_volume(bucket).await.expect("bucket volume");
+    }
+    let opts = ObjectOptions::default();
+    let upload = set
+        .new_multipart_upload(bucket, object, &opts)
+        .await
+        .expect("new multipart upload");
+    let mut parts = Vec::new();
+    let mut expected = Vec::new();
+    for (number, size, slot) in [(1, 5 * 1024 * 1024, 2), (2, 4096, 3)] {
+        let payload = vec![number as u8; size];
+        expected.extend_from_slice(&payload);
+        let hook = WriterRound::install(disks, slot, fault, legacy_policy);
+        let mut reader = PutObjReader::from_vec(payload);
+        let part = set
+            .put_object_part(bucket, object, &upload.upload_id, number, &mut reader, &opts)
+            .await
+            .expect("three completed shards must admit each part");
+        match fault {
+            Fault::DelayedShutdown if !legacy_policy => {
+                assert!(hook.round.delayed_observed.load(Ordering::Acquire));
+                assert!(!hook.round.delayed_dropped.load(Ordering::Acquire), "finite slow shard must finish");
+                assert_eq!(hook.round.completed.load(Ordering::Acquire), 4, "all real shard writers finished");
+            }
+            Fault::DelayedShutdown | Fault::WriteFailure => {
+                assert!(hook.round.delayed_dropped.load(Ordering::Acquire), "failed shard must be dropped");
+                assert_eq!(hook.round.completed.load(Ordering::Acquire), 3);
+            }
+        }
+        parts.push(CompletePart {
+            part_num: number,
+            etag: part.etag,
+            ..Default::default()
+        });
+    }
+    let (metadata, _) = set
+        .check_upload_id_exists(bucket, object, &upload.upload_id, true)
+        .await
+        .expect("upload metadata");
+    (upload.upload_id, parts, expected, metadata)
+}
+
+#[tokio::test]
+async fn multipart_budget_keeps_different_finite_shutdown_stragglers_and_completes() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-finite-shutdown";
+    let object = "object";
+    let (upload, parts, expected, metadata) = upload_two_parts(&set, &disks, bucket, object, Fault::DelayedShutdown, false).await;
+    let data_dir = metadata.data_dir.expect("parts-backed data directory");
+    let erasure =
+        coding::Erasure::try_new(metadata.erasure.data_blocks, metadata.erasure.parity_blocks, metadata.erasure.block_size)
+            .expect("upload erasure layout");
+    let stage = SetDisks::get_upload_id_dir(bucket, object, &upload);
+    for disk in &disks {
+        for (number, size) in [(1, 5 * 1024 * 1024), (2, 4096)] {
+            let path = format!("{stage}/{data_dir}/part.{number}");
+            let bytes = disk
+                .read_all(RUSTFS_META_MULTIPART_BUCKET, &path)
+                .await
+                .expect("every disk has every part");
+            let shard_len = erasure.shard_file_size(size);
+            let encoded_len =
+                coding::bitrot_shard_file_size(shard_len as usize, erasure.shard_size(), HashAlgorithm::HighwayHash256S);
+            assert_eq!(bytes.len(), encoded_len, "complete physical shard: {path}");
+        }
+    }
+    set.clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect("all parts have a common completed quorum");
+    let mut reader = set
+        .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+        .await
+        .expect("read committed object");
+    let mut actual = Vec::new();
+    reader.stream.read_to_end(&mut actual).await.expect("whole object readback");
+    assert_eq!(actual, expected);
+}
+
+#[tokio::test]
+async fn multipart_budget_distinct_failed_shards_do_not_bypass_complete_quorum() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-distinct-failures";
+    let object = "object";
+    let (upload, parts, _, _) = upload_two_parts(&set, &disks, bucket, object, Fault::WriteFailure, false).await;
+    let error = set
+        .clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect_err("two common complete disks cannot meet the write quorum of three");
+    assert!(error.is_quorum_error(), "expected complete quorum refusal, got {error:?}");
+    set.check_upload_id_exists(bucket, object, &upload, false)
+        .await
+        .expect("failed completion retains retryable upload state");
+    assert!(
+        set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(),
+        "failed completion must not publish an object"
+    );
+}
+
+#[tokio::test]
+async fn multipart_budget_old_grace_loses_common_part_quorum() {
+    let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+    let bucket = "multipart-old-grace";
+    let object = "object";
+    let (upload, parts, _, _) = upload_two_parts(&set, &disks, bucket, object, Fault::DelayedShutdown, true).await;
+    let error = set
+        .clone()
+        .complete_multipart_upload(bucket, object, &upload, parts, &ObjectOptions::default())
+        .await
+        .expect_err("the old per-part grace loses the common completion quorum");
+    assert!(error.is_quorum_error(), "expected complete quorum refusal, got {error:?}");
+}

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

@@ -4562,7 +4562,14 @@ impl SetDisks {
                 SmallWritePath::Pipeline => IntegrityEncodeMode::Streaming,
             };
             let (reader, w_size, inline_shards, integrity) = Arc::clone(&erasure)
-                .encode_with_shard_integrity(stream, &mut writers, write_quorum, 1, mode, protect_write)
+                .encode_with_shard_integrity(
+                    stream,
+                    &mut writers,
+                    coding::encode::ShardWritePolicy::object(write_quorum),
+                    1,
+                    mode,
+                    protect_write,
+                )
                 .await?;
             let encode_elapsed = encode_stage_start.map(|stage_start| stage_start.elapsed());
             let encode_ms = encode_elapsed.map(|elapsed| elapsed.as_millis() as u64).unwrap_or_default();

docs/operations/drive-timeout-tuning.md

@@ -24,6 +24,12 @@ may therefore remain undetected until real I/O reaches it. Missing slots,
 invalidated identities, unhealthy drives, and remote endpoints still get probed
 for recovery. Scanner, heal, and other background services have separate controls.
 
+## Multipart shard completion
+
+UploadPart waits for every still-pending shard operation to finish or reach its existing `RUSTFS_OBJECT_DISK_WRITE_STALL_TIMEOUT` budget (default 30 seconds), bounded further by the remaining `RUSTFS_OBJECT_DISK_WRITE_ABSOLUTE_CAP` when configured. Reaching write quorum does not start the one-second straggler cutoff used by ordinary PUT. This keeps finite slow shards from being discarded on different disks for different parts, which could leave too few disks carrying the entire object at CompleteMultipartUpload.
+
+This does not require every disk to succeed, count an unfinished writer as committed, or reduce the final object's write quorum. Failed and expired writers remain excluded. The tradeoff is that a black-hole peer can delay each part up to the existing operation budget rather than one second after quorum. The stall budget resets per shard-write operation; the optional absolute cap belongs to one part's encoding instance, not the entire multipart upload. Setting both budgets to zero retains the existing unbounded-wait behavior. Actual failures can still leave different parts missing on different disks, in which case completion must fail if a common complete write quorum is unavailable.
+
 ## Background: what these timeouts bound
 
 Every foreground drive operation carries a liveness budget so a hung disk fails