fix(ecstore): heal partial ordinary puts (#5783)

crates/ecstore/src/ecstore_validation_blackbox.rs

@@ -220,6 +220,126 @@ async fn blackbox_get_restores_body_and_enqueues_repair_after_one_corrupt_shard(
     let mut heal_rx = rustfs_common::heal_channel::init_heal_channel()
         .expect("this must be the only ecstore test that owns the heal channel receiver");
 
+    // Ordinary PUTs use the same admission channel as read repair. A single
+    // rename target failure still satisfies write quorum, so the committed
+    // version must be queued for convergence without delaying the PUT ACK.
+    let (_put_dirs, put_set) = make_local_set_disks(4, 2).await;
+    let put_bucket = "bb-put-partial-convergence";
+    let put_object = "object.bin";
+    put_set
+        .make_bucket(put_bucket, &MakeBucketOptions::default())
+        .await
+        .expect("PUT bucket should be created");
+    let offline_disk = {
+        let mut disks = put_set.disks.write().await;
+        disks[0].take()
+    };
+    let mut put_reader = PutObjReader::from_vec(vec![0x42; BLOCK_SIZE_V2 + 1024]);
+    let committed = put_set
+        .put_object(
+            put_bucket,
+            put_object,
+            &mut put_reader,
+            &ObjectOptions {
+                no_lock: true,
+                versioned: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("partial ordinary PUT should succeed at write quorum");
+    let committed_version = committed
+        .version_id
+        .expect("versioned PUT should return a version id")
+        .to_string();
+
+    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 == put_bucket && request.object_prefix.as_deref() == Some(put_object) =>
+                {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                    break request;
+                }
+                HealChannelCommand::Start { response_tx, .. } => {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("partial ordinary PUT should enqueue convergence heal");
+    assert_eq!(request.object_version_id.as_deref(), Some(committed_version.as_str()));
+    assert_eq!(request.pool_index, Some(0));
+    assert_eq!(request.set_index, Some(0));
+
+    let duplicate_request = tokio::time::timeout(std::time::Duration::from_millis(100), async {
+        loop {
+            match heal_rx.recv().await.expect("heal channel should stay open") {
+                HealChannelCommand::Start { request, response_tx }
+                    if request.bucket == put_bucket && request.object_prefix.as_deref() == Some(put_object) =>
+                {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                    break Some(request);
+                }
+                HealChannelCommand::Start { response_tx, .. } => {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .ok()
+    .flatten();
+    assert!(duplicate_request.is_none(), "partial ordinary PUT must enqueue exactly one heal request");
+
+    {
+        let mut disks = put_set.disks.write().await;
+        disks[0] = offline_disk;
+    }
+
+    let healthy_bucket = "bb-put-healthy-convergence";
+    put_set
+        .make_bucket(healthy_bucket, &MakeBucketOptions::default())
+        .await
+        .expect("healthy PUT bucket should be created");
+    let mut healthy_reader = PutObjReader::from_vec(b"healthy".to_vec());
+    put_set
+        .put_object(
+            healthy_bucket,
+            put_object,
+            &mut healthy_reader,
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("healthy ordinary PUT should succeed");
+    let healthy_request = tokio::time::timeout(std::time::Duration::from_millis(100), async {
+        loop {
+            match heal_rx.recv().await.expect("heal channel should stay open") {
+                HealChannelCommand::Start { request, response_tx }
+                    if request.bucket == healthy_bucket && request.object_prefix.as_deref() == Some(put_object) =>
+                {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                    break Some(request);
+                }
+                HealChannelCommand::Start { response_tx, .. } => {
+                    let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .ok()
+    .flatten();
+    assert!(healthy_request.is_none(), "fully converged ordinary PUT must not enqueue heal");
+
     // Keep data-blocks-first reader setup explicit for this deterministic
     // repair assertion (see ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP in
     // set_disk/core/io_primitives.rs): if a caller opts back into all-shards,

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

@@ -1445,7 +1445,7 @@ impl SetDisks {
             }
 
             let rename_stage_start = Instant::now();
-            let (online_disks, _, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
+            let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
                 &shuffle_disks,
                 RUSTFS_META_TMP_BUCKET,
                 tmp_dir.as_str(),
@@ -1455,6 +1455,23 @@ impl SetDisks {
                 write_quorum,
             )
             .await?;
+            // Do this before any post-commit await so request cancellation cannot
+            // bypass best-effort admission. A process crash before admission
+            // remains subject to the existing scanner reconciliation path.
+            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.map(|version_id| version_id.to_string());
+                tokio::spawn(async move {
+                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                });
+            }
             let rename_stage_ms = rename_stage_start.elapsed().as_millis() as u64;
             rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", rename_stage_ms as f64);
             if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
@@ -4368,17 +4385,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
     #[tracing::instrument(skip(self))]
     async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> {
-        if let Err(e) =
-            rustfs_common::heal_channel::send_heal_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),
-            ))
-            .await
-        {
+        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 = (!version_id.is_empty()).then(|| version_id.to_string());
+        if let Err(e) = rustfs_common::heal_channel::send_heal_request(request).await {
             warn!(
                 bucket,
                 object,