diff --git a/crates/ecstore/src/ecstore_validation_blackbox.rs b/crates/ecstore/src/ecstore_validation_blackbox.rs index 2531b39a8..23fd5c39c 100644 --- a/crates/ecstore/src/ecstore_validation_blackbox.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index cd669a875..ba264f4bb 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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,