fix(ecstore): bind multipart convergence heal versions (#5786)

crates/ecstore/src/ecstore_validation_blackbox.rs

@@ -7,6 +7,7 @@ use crate::io_support::rio::HashReader;
 use crate::object_api::{BLOCK_SIZE_V2, ObjectLockConfigSnapshot, ObjectOptions, PutObjReader};
 use crate::set_disk::SetDisks;
 use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+use crate::storage_api_contracts::multipart::{CompletePart, MultipartOperations as _};
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
 use crate::storage_api_contracts::range::HTTPRangeSpec;
 use crate::store::init_format::save_format_file;
@@ -206,8 +207,28 @@ async fn blackbox_get_restores_body_after_one_shard_file_is_removed() {
 #[tokio::test]
 // Serialized: forces the reader-setup strategy through a process-global env var.
 #[serial_test::serial]
-async fn blackbox_get_restores_body_and_enqueues_repair_after_one_corrupt_shard() {
-    use rustfs_common::heal_channel::{HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealRequestSource};
+async fn blackbox_heal_requests_preserve_repair_scope() {
+    use rustfs_common::heal_channel::{
+        HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealRequestSource,
+    };
+
+    async fn receive_matching_heal(rx: &mut HealChannelReceiver, bucket: &str, object: &str) -> HealChannelRequest {
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
+            loop {
+                match rx.recv().await.expect("heal channel should stay open") {
+                    HealChannelCommand::Start { request, response_tx }
+                        if request.bucket == bucket && request.object_prefix.as_deref() == Some(object) =>
+                    {
+                        let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                        break request;
+                    }
+                    _ => continue,
+                }
+            }
+        })
+        .await
+        .expect("matching heal request should be submitted")
+    }
 
     // Own the process-global heal channel so the read path's repair submission
     // becomes observable. init_heal_channel() succeeds exactly once per test
@@ -396,19 +417,7 @@ async fn blackbox_get_restores_body_and_enqueues_repair_after_one_corrupt_shard(
 
         assert_eq!(restored, payload);
 
-        let request = tokio::time::timeout(std::time::Duration::from_secs(30), async {
-            loop {
-                match heal_rx.recv().await.expect("heal channel should stay open") {
-                    HealChannelCommand::Start { request, response_tx } if request.bucket == bucket => {
-                        let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
-                        break request;
-                    }
-                    _ => continue,
-                }
-            }
-        })
-        .await
-        .expect("corrupt-shard GET should enqueue a read-repair heal request");
+        let request = receive_matching_heal(&mut heal_rx, bucket, object).await;
 
         assert_eq!(request.source, HealRequestSource::ReadRepair);
         assert_eq!(request.object_prefix.as_deref(), Some(object));
@@ -417,6 +426,142 @@ async fn blackbox_get_restores_body_and_enqueues_repair_after_one_corrupt_shard(
         assert_eq!(request.set_index, Some(0));
         assert_eq!(request.priority, HealChannelPriority::Low);
         assert_eq!(request.recreate_missing, Some(true));
+
+        let mpu_bucket = "bb-mpu-convergence-heal";
+        let partial_object = "partial.bin";
+        let suspended_object = "suspended.bin";
+        let payload = vec![0x5a; 1 << 20];
+        let mpu_opts = ObjectOptions {
+            no_lock: true,
+            versioned: true,
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(mpu_bucket, &MakeBucketOptions::default())
+            .await
+            .expect("multipart bucket should be created");
+
+        let stage_upload = async |object: &str| {
+            let upload = set_disks
+                .new_multipart_upload(mpu_bucket, object, &mpu_opts)
+                .await
+                .expect("multipart upload should be created");
+            let mut reader = PutObjReader::new(
+                HashReader::from_stream(
+                    Cursor::new(payload.clone()),
+                    payload.len() as i64,
+                    payload.len() as i64,
+                    None,
+                    None,
+                    false,
+                )
+                .expect("multipart reader should be constructed"),
+            );
+            let part = set_disks
+                .put_object_part(mpu_bucket, object, &upload.upload_id, 1, &mut reader, &mpu_opts)
+                .await
+                .expect("multipart part should be written");
+            (
+                upload.upload_id,
+                vec![CompletePart {
+                    part_num: part.part_num,
+                    etag: part.etag,
+                    ..Default::default()
+                }],
+            )
+        };
+
+        let (partial_upload_id, partial_parts) = stage_upload(partial_object).await;
+        let offline_disk = {
+            let mut disks = set_disks.disks.write().await;
+            disks[3].take().expect("fourth disk should be online before completion")
+        };
+        crate::crash_inject::arm(crate::crash_inject::CrashPoint::MultipartAfterCommitBeforePartsCleanup, partial_object);
+        let completed = set_disks
+            .clone()
+            .complete_multipart_upload(mpu_bucket, partial_object, &partial_upload_id, partial_parts, &mpu_opts)
+            .await;
+        assert!(
+            matches!(completed, Err(Error::Unexpected)),
+            "partial multipart completion should reach the post-commit crash point, got {completed:?}"
+        );
+        crate::crash_inject::disarm(crate::crash_inject::CrashPoint::MultipartAfterCommitBeforePartsCleanup, partial_object);
+
+        let request = receive_matching_heal(&mut heal_rx, mpu_bucket, partial_object).await;
+
+        let completed_version_id = request
+            .object_version_id
+            .clone()
+            .expect("versioned multipart convergence heal must bind a version id");
+        assert_eq!(request.pool_index, Some(0));
+        assert_eq!(request.set_index, Some(0));
+        assert_eq!(request.priority, HealChannelPriority::Normal);
+
+        let duplicate = tokio::time::timeout(std::time::Duration::from_millis(250), async {
+            loop {
+                match heal_rx.recv().await.expect("heal channel should stay open") {
+                    HealChannelCommand::Start { request, .. }
+                        if request.bucket == mpu_bucket && request.object_prefix.as_deref() == Some(partial_object) =>
+                    {
+                        break request;
+                    }
+                    _ => continue,
+                }
+            }
+        })
+        .await;
+        assert!(duplicate.is_err(), "partial multipart completion must enqueue exactly one heal request");
+
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[3] = Some(offline_disk);
+        }
+        let committed = set_disks
+            .get_object_info(
+                mpu_bucket,
+                partial_object,
+                &ObjectOptions {
+                    no_lock: true,
+                    versioned: true,
+                    version_id: Some(completed_version_id.clone()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("heal-bound multipart version should be committed and addressable");
+        assert_eq!(committed.version_id.map(|version_id| version_id.to_string()), Some(completed_version_id));
+
+        let (suspended_upload_id, suspended_parts) = stage_upload(suspended_object).await;
+        let offline_disk = {
+            let mut disks = set_disks.disks.write().await;
+            disks[3]
+                .take()
+                .expect("fourth disk should be online before suspended completion")
+        };
+        let suspended_opts = ObjectOptions {
+            no_lock: true,
+            version_suspended: true,
+            ..Default::default()
+        };
+        let suspended = set_disks
+            .clone()
+            .complete_multipart_upload(mpu_bucket, suspended_object, &suspended_upload_id, suspended_parts, &suspended_opts)
+            .await
+            .expect("suspended multipart completion should succeed at write quorum");
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[3] = Some(offline_disk);
+        }
+        assert!(
+            suspended.version_id.is_some_and(|version_id| version_id.is_nil()),
+            "suspended multipart completion should publish the null version"
+        );
+
+        let request = receive_matching_heal(&mut heal_rx, mpu_bucket, suspended_object).await;
+
+        let null_version_id = uuid::Uuid::nil().to_string();
+        assert_eq!(request.object_version_id.as_deref(), Some(null_version_id.as_str()));
     })
     .await;
 }

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

@@ -2055,6 +2055,26 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         )
         .await?;
 
+        // Detach admission before any post-commit await: client cancellation
+        // must not couple durable convergence repair to cleanup work.
+        if convergence.needs_heal() {
+            let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                bucket.to_string(),
+                Some(object.to_string()),
+                false,
+                Some(HealChannelPriority::Normal),
+                Some(self.pool_index),
+                Some(self.set_index),
+            );
+            request.object_version_id = fi
+                .version_id
+                .or_else(|| opts.version_suspended.then(Uuid::nil))
+                .map(|version_id| version_id.to_string());
+            tokio::spawn(async move {
+                let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+            });
+        }
+
         // Crash-consistency injection: hard power loss after the authoritative
         // rename_data commit succeeded but before the stale part.N.meta cleanup.
         // The new version is durably committed and visible, so a crash here must
@@ -2128,48 +2148,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         drop(object_lock_guard); // drop object lock guard to release the lock
 
-        // backlog#1321: enqueue heal only when the committed replicas actually
-        // need to converge — a partial commit (some disk failed/offline) or a
-        // signature divergence between committed replicas. A fully healthy MPU
-        // (identical signatures on every disk) is `AllSuccessIdentical` and
-        // submits nothing, which is the fix: the old `Option::is_some()` gate
-        // treated the mere existence of a version signature as "needs heal", so
-        // every healthy <=10-version completion self-enqueued.
-        //
-        // The submit is detached (`tokio::spawn`) so it stays off the ACK
-        // critical path AND survives cancellation of the completion future: the
-        // write is already durable and ACK-worthy, so the heal admission must
-        // not ride the client's request lifetime. The admission itself is
-        // bounded / deduplicated / observable (`send_heal_request` ->
-        // `HealAdmissionResult`), so this emits at most one submit per
-        // completion and coalesces with any in-flight heal for the same object.
-        //
-        // Scanner backstop (backlog#1321 patch): a `PartialCommit` whose
-        // completion is cancelled in the narrow window after the durable commit
-        // but before this spawn runs is not lost — the divergence it would have
-        // healed is exactly what the background scanner reconciles. `Unknown`
-        // (>10 versions, no signature produced) likewise relies on the scanner
-        // rather than self-enqueuing.
-        if convergence.needs_heal() {
-            let bucket = bucket.to_string();
-            let object = object.to_string();
-            let pool_index = self.pool_index;
-            let set_index = self.set_index;
-            tokio::spawn(async move {
-                let _ = rustfs_common::heal_channel::send_heal_request(
-                    rustfs_common::heal_channel::create_heal_request_with_options(
-                        bucket,
-                        Some(object),
-                        false,
-                        Some(HealChannelPriority::Normal),
-                        Some(pool_index),
-                        Some(set_index),
-                    ),
-                )
-                .await;
-            });
-        }
-
         for (i, op_disk) in online_disks.iter().enumerate() {
             if let Some(disk) = op_disk
                 && disk.is_online().await