diff --git a/crates/ecstore/src/ecstore_validation_blackbox.rs b/crates/ecstore/src/ecstore_validation_blackbox.rs index 23fd5c39c..30e309e4f 100644 --- a/crates/ecstore/src/ecstore_validation_blackbox.rs +++ b/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; } diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 43a8c1b2b..6c8dda73d 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/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