From b41bbe2db459cb74b4cb500867f79c26df17bd29 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 17 Jul 2026 01:38:15 +0800 Subject: [PATCH] fix(ecstore): split rename_data signature from heal-convergence decision (#4926) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CompleteMultipartUpload enqueued a normal-priority heal whenever `rename_data` returned a `Some(versions)` signature. But the per-disk signature is produced for every object with <=10 versions, and a healthy quorum reduces to `Some` as well, so the `Option>` return value conflated two distinct facts — "a version signature exists" and "the committed replicas need heal". The result: nearly every healthy MPU completion self-enqueued a heal, while >10-version objects (signature `None`) did not — an algorithmic heal amplification on the healthy path (rustfs/backlog#1321). Replace the overloaded `Option>` second element of `SetDisks::rename_data` with an explicit `RenameConvergence` classification computed after the write-quorum gate: - AllSuccessIdentical — every attempted disk committed with an identical, known signature (no heal). - PartialCommit — write quorum met but a disk failed/offline; a committed replica is missing or stale (heal). - SignatureDivergent — all committed but signatures diverge, or mix signed (<=10-version) with unsigned (>10-version) disks (heal). - Unknown — all committed, no signature produced (>10 versions); latent divergence is left to the scanner backstop, not self-enqueued. `RenameConvergence::needs_heal()` is the single decision point. The version signature is now only comparison material; it no longer doubles as a heal flag. The old `select_rename_data_versions` / `reduce_common_versions` / `rename_data_versions_key` machinery that carried the conflation is removed. The heal submission in `complete_multipart_upload` moves off the ACK critical path into a detached task: it runs after the object lock is dropped and after the durable `rename_data` commit, survives cancellation of the completion future, and coalesces through the existing bounded / deduplicated / observable heal-channel admission (one submit per degraded completion, at most). A completion cancelled in the narrow window between the durable commit and reaching the enqueue is scanner-backstopped, as is the Unknown (>10-version) case. The PUT path (`object.rs`) binds the second element as `_` and is unchanged. The change is orthogonal to and composes with the #1312 commit fence on the same `rename_data` path (epoch rejection is a commit-gate failure surfaced through `Result::Err`, convergence is a post-commit signal); documented in docs/architecture/unified-object-generation.md. Tests: `classify_rename_convergence` white-box cases cover the full acceptance matrix (healthy 4/4 and 8/8, 3-same-1-divergent, failed/offline disk, no-common-quorum split, >10-version all-success and with-failure, mixed signed/unsigned) and fail on revert to the old "signature exists => heal" semantics. The decision function is tested directly rather than through the process-global heal channel, whose receiver is owned exclusively by the blackbox serial test (init_heal_channel is once per binary). Refs: https://github.com/rustfs/backlog/issues/1321 --- .../src/set_disk/core/io_primitives.rs | 157 ++++++++++++------ crates/ecstore/src/set_disk/mod.rs | 114 ++++++++++--- crates/ecstore/src/set_disk/ops/multipart.rs | 51 ++++-- .../architecture/unified-object-generation.md | 21 +++ 4 files changed, 257 insertions(+), 86 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 6cff5cda0..cc5bb432a 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -2524,11 +2524,57 @@ enum OrphanDirScan { Missing, } -fn rename_data_versions_key(versions: &[u8]) -> Option<[u8; 8]> { - let prefix = versions.get(..8)?; - let mut key = [0; 8]; - key.copy_from_slice(prefix); - Some(key) +/// Outcome of a *post-quorum* `rename_data` commit, classifying whether the +/// committed replicas converged so the caller can decide heal admission +/// WITHOUT conflating "a version signature exists" with "this write needs +/// heal" (rustfs/backlog#1321). +/// +/// `rename_data` reaches this classification only once write quorum is met — a +/// sub-quorum commit returns `Err` and never produces a convergence — so every +/// variant describes an already-durable, already-ACKable write. +/// +/// # Extension point for #1312 (commit fencing) +/// +/// #1312 layers a per-object fencing epoch onto the SAME `rename_data` path. +/// An epoch rejection is a *commit* failure surfaced through the existing +/// `Result::Err` channel (the write never lands), which is orthogonal to this +/// post-commit convergence signal (the write landed; do the replicas need +/// heal). The two therefore compose: fencing gates whether we reach a +/// convergence at all, and this enum classifies the convergence once reached. +/// A future fence-aware variant, if ever needed, is an additive change here. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(in crate::set_disk) enum RenameConvergence { + /// Every attempted disk committed and reported an identical, known version + /// signature. The committed replicas are already converged; no heal needed. + AllSuccessIdentical, + /// Write quorum was met but at least one attempted disk failed to commit + /// (error or offline). The committed set may be short a replica; the + /// laggard must be converged by heal. + PartialCommit, + /// Every attempted disk committed but their version signatures diverge (or + /// mix signed <=10-version disks with unsigned >10-version disks, itself a + /// version-count divergence). Heal must reconcile the committed replicas. + SignatureDivergent, + /// Every attempted disk committed but none produced a version signature + /// (all observed >10 versions, where the signature is deliberately + /// omitted). Divergence can neither be proven nor ruled out here, so any + /// latent divergence is left to the scanner backstop rather than enqueued + /// for heal. + Unknown, +} + +impl RenameConvergence { + /// Whether this commit outcome must enqueue an object heal to converge the + /// committed replicas. + /// + /// `Unknown` and `AllSuccessIdentical` return `false`: the former is + /// scanner-backstopped (see the variant docs), the latter is already + /// converged. This is the single decision point that replaced the old + /// `Option>::is_some()` heuristic, under which a healthy MPU + /// (identical signatures on every disk) always looked like it needed heal. + pub(in crate::set_disk) fn needs_heal(&self) -> bool { + matches!(self, Self::PartialCommit | Self::SignatureDivergent) + } } impl SetDisks { @@ -2557,7 +2603,7 @@ impl SetDisks { write_quorum: usize, ) -> disk::error::Result<( Vec>, - Option>, + RenameConvergence, Option, Vec>, Option, @@ -2711,7 +2757,7 @@ impl SetDisks { } let data_dir = Self::reduce_common_data_dir(&data_dirs, write_quorum); - let versions = Self::select_rename_data_versions(&disk_versions, &errs, write_quorum); + let convergence = Self::classify_rename_convergence(&disk_versions, &errs); let old_current_size = Self::reduce_common_old_current_size(&old_current_sizes, write_quorum); let online_disks = Self::eval_disks(disks, &errs); let cleanup_disks = if let Some(data_dir) = data_dir { @@ -2731,7 +2777,7 @@ impl SetDisks { vec![None; disks.len()] }; - Ok((online_disks, versions, data_dir, cleanup_disks, old_current_size)) + Ok((online_disks, convergence, data_dir, cleanup_disks, old_current_size)) } /// rustfs/backlog#1009: reduce the per-disk observations of the @@ -2762,61 +2808,62 @@ impl SetDisks { if max >= write_quorum { old_current_size } else { None } } - pub(in crate::set_disk) fn reduce_common_versions(disk_versions: &[Option>], write_quorum: usize) -> Option> { - let mut versions_count = HashMap::new(); - - for versions in disk_versions.iter().flatten() { - if let Some(key) = rename_data_versions_key(versions) { - *versions_count.entry(key).or_insert(0usize) += 1; - } - } - - let (common_versions, max_count) = versions_count - .into_iter() - .max_by_key(|(_, count)| *count) - .unwrap_or(([0; 8], 0)); - - if max_count < write_quorum { - return None; - } - - disk_versions - .iter() - .flatten() - .find(|versions| rename_data_versions_key(versions).is_some_and(|key| key == common_versions)) - .cloned() - } - - pub(in crate::set_disk) fn select_rename_data_versions( + /// Classify a *post-quorum* `rename_data` commit into an explicit + /// convergence outcome (rustfs/backlog#1321). This runs only after the + /// write-quorum gate has already passed, so every returned variant + /// describes an already-durable, already-ACKable write; the decision here + /// is purely "do the committed replicas need heal to converge". + /// + /// The per-disk version signature (`disk_versions[i]`) is used only as + /// comparison material — it is deliberately NOT overloaded to also mean + /// "needs heal". A `Some(bytes)` signature carries the concatenated + /// version-id bytes of a disk that observed <=10 versions; a `None` + /// signature is a disk that committed but deliberately produced no + /// signature (>10 versions). A `None` entry for a *failed* disk never + /// reaches this point as a signature because failures are handled first. + pub(in crate::set_disk) fn classify_rename_convergence( disk_versions: &[Option>], errs: &[Option], - write_quorum: usize, - ) -> Option> { - let mut versions = Self::reduce_common_versions(disk_versions, write_quorum); - for (dversions, err) in disk_versions.iter().zip(errs.iter()) { - if err.is_some() { - continue; - } - let Some(dversions) = dversions.as_ref().filter(|versions| !versions.is_empty()) else { + ) -> RenameConvergence { + // Any failed / offline disk that got past the write-quorum gate means a + // committed replica is missing or stale: converge it via heal, + // regardless of what the surviving disks' signatures say. + if errs.iter().any(|err| err.is_some()) { + return RenameConvergence::PartialCommit; + } + + // Every disk committed. Compare their reported version signatures. + let mut seen: Option<&Vec> = None; + let mut signed_count = 0usize; + let mut divergent = false; + for signature in disk_versions.iter() { + let Some(sig) = signature.as_ref() else { continue; }; - - match versions.as_ref() { - Some(current_versions) if dversions != current_versions => { - if dversions.len() > current_versions.len() { - versions = Some(dversions.clone()); - } - break; - } + signed_count += 1; + match seen { + None => seen = Some(sig), + Some(prev) if prev != sig => divergent = true, Some(_) => {} - None => { - versions = Some(dversions.clone()); - break; - } } } - versions + if divergent { + return RenameConvergence::SignatureDivergent; + } + if signed_count == 0 { + // No disk produced a signature (all observed >10 versions): a + // latent divergence cannot be proven or ruled out here, so it is + // left to the scanner backstop rather than enqueued for heal. + return RenameConvergence::Unknown; + } + if signed_count < disk_versions.len() { + // A mix of signed (<=10 versions) and unsigned (>10 versions) disks + // is itself a version-count divergence between committed replicas: + // reconcile it via heal. + return RenameConvergence::SignatureDivergent; + } + RenameConvergence::AllSuccessIdentical } /// Reclaim the old (now dereferenced) `object/` on the disks diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 7ae1c79fd..8bb71dcda 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -6288,33 +6288,107 @@ mod tests { signature } + // backlog#1321: `classify_rename_convergence` is the single decision that + // replaced the old `Option>::is_some()` heal gate. These are the + // revert-fails white-box cases behind the issue acceptance matrix — if the + // decision regressed to "a signature exists => needs heal", the healthy + // 4/4 and 8/8 cases below would flip from `AllSuccessIdentical` to a + // heal-worthy variant and fail. #[test] - fn test_reduce_common_versions_requires_write_quorum() { - let common = rename_versions_signature(1, 1); - let other = rename_versions_signature(2, 1); + fn test_classify_rename_convergence_healthy_identical_needs_no_heal() { + use crate::set_disk::core::io_primitives::RenameConvergence; + let sig = rename_versions_signature(1, 3); - let disk_versions = vec![Some(common.clone()), Some(common.clone()), Some(other)]; - let result = SetDisks::reduce_common_versions(&disk_versions, 2); - assert_eq!(result, Some(common)); + // Healthy 4/4: every disk committed with an identical, known signature. + let disk_versions = vec![Some(sig.clone()), Some(sig.clone()), Some(sig.clone()), Some(sig.clone())]; + let errs = vec![None, None, None, None]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::AllSuccessIdentical); + assert!(!convergence.needs_heal(), "a fully converged healthy MPU must not enqueue heal"); - let split_versions = vec![ - Some(rename_versions_signature(1, 1)), - Some(rename_versions_signature(2, 1)), - None, - ]; - let result = SetDisks::reduce_common_versions(&split_versions, 2); - assert_eq!(result, None); + // Healthy 8/8: same property at a wider set width. + let disk_versions = vec![Some(sig); 8]; + let errs = vec![None; 8]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::AllSuccessIdentical); + assert!(!convergence.needs_heal()); } #[test] - fn test_select_rename_data_versions_keeps_longer_success_disparity() { - let common = rename_versions_signature(1, 1); - let longer = rename_versions_signature(2, 2); - let disk_versions = vec![Some(common.clone()), Some(common), Some(longer.clone())]; - let errs = vec![None, None, None]; + fn test_classify_rename_convergence_three_agree_one_diverges_heals() { + use crate::set_disk::core::io_primitives::RenameConvergence; + let common = rename_versions_signature(1, 2); + let odd = rename_versions_signature(2, 2); - let result = SetDisks::select_rename_data_versions(&disk_versions, &errs, 2); - assert_eq!(result, Some(longer)); + // 3-same, 1-divergent, all committed: reconcile the odd replica. + let disk_versions = vec![Some(common.clone()), Some(common.clone()), Some(common), Some(odd)]; + let errs = vec![None, None, None, None]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::SignatureDivergent); + assert!(convergence.needs_heal()); + } + + #[test] + fn test_classify_rename_convergence_failed_or_offline_disk_heals() { + use crate::set_disk::core::io_primitives::RenameConvergence; + let sig = rename_versions_signature(1, 2); + + // 1 disk failed/offline while the rest committed identically: past the + // write-quorum gate this is a `PartialCommit` — a replica is missing. + let disk_versions = vec![Some(sig.clone()), Some(sig.clone()), Some(sig), None]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::PartialCommit); + assert!(convergence.needs_heal()); + } + + #[test] + fn test_classify_rename_convergence_no_common_quorum_heals() { + use crate::set_disk::core::io_primitives::RenameConvergence; + // No signature holds a majority (2/2 split), every disk committed: + // divergence with no common quorum still reconciles via heal. + let a = rename_versions_signature(1, 1); + let b = rename_versions_signature(2, 1); + let disk_versions = vec![Some(a.clone()), Some(a), Some(b.clone()), Some(b)]; + let errs = vec![None, None, None, None]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::SignatureDivergent); + assert!(convergence.needs_heal()); + } + + #[test] + fn test_classify_rename_convergence_over_ten_versions_by_success() { + use crate::set_disk::core::io_primitives::RenameConvergence; + // >10 versions: every disk deliberately omits the signature (`None`). + // All committed => `Unknown` => scanner-backstopped, no self-enqueue. + let disk_versions = vec![None, None, None, None]; + let errs = vec![None, None, None, None]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::Unknown); + assert!( + !convergence.needs_heal(), + ">10-version healthy commit relies on the scanner, not self-enqueue" + ); + + // Same >10-version shape but with a failed disk: a failure is + // conservative-heal regardless of signatures being unavailable. + let errs = vec![None, None, None, Some(DiskError::FileCorrupt)]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::PartialCommit); + assert!(convergence.needs_heal()); + } + + #[test] + fn test_classify_rename_convergence_mixed_signed_unsigned_heals() { + use crate::set_disk::core::io_primitives::RenameConvergence; + // A committed replica with <=10 versions (signed) alongside one with + // >10 versions (unsigned) is itself a version-count divergence. + let sig = rename_versions_signature(1, 2); + let disk_versions = vec![Some(sig.clone()), Some(sig.clone()), Some(sig), None]; + let errs = vec![None, None, None, None]; + let convergence = SetDisks::classify_rename_convergence(&disk_versions, &errs); + assert_eq!(convergence, RenameConvergence::SignatureDivergent); + assert!(convergence.needs_heal()); } #[test] diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 55be919a4..38704c3ff 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1433,7 +1433,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // The trailing `_` drops the rename_data old-size backfill // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. - let (online_disks, versions, op_old_dir, cleanup_disks, _) = Self::rename_data( + let (online_disks, convergence, op_old_dir, cleanup_disks, _) = Self::rename_data( &shuffle_disks, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, @@ -1485,17 +1485,46 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { drop(object_lock_guard); // drop object lock guard to release the lock - if let Some(versions) = versions { - let _ = - 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), - )) + // 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; + }); } let upload_id_path = upload_id_path.clone(); diff --git a/docs/architecture/unified-object-generation.md b/docs/architecture/unified-object-generation.md index 12fbbd028..e4e6535e5 100644 --- a/docs/architecture/unified-object-generation.md +++ b/docs/architecture/unified-object-generation.md @@ -97,6 +97,27 @@ orphan an acceptable residue accounted for by GC metrics — the white-box acceptance "no background disk write after release" must be rewritten accordingly. +### Post-commit convergence is orthogonal to the fence (#1321) + +The same `SetDisks::rename_data` path already returns a post-commit +convergence classification (`RenameConvergence`, rustfs/backlog#1321) that +tells the caller whether the *committed* replicas need heal to converge — +`AllSuccessIdentical` (no heal), `PartialCommit` (a replica failed/offline), +`SignatureDivergent` (committed replicas' version signatures differ), or +`Unknown` (no signature was produced, e.g. >10 versions — scanner-backstopped). +This replaced an earlier `Option>` heuristic under which any +version signature looked like "needs heal", so every healthy multipart +completion self-enqueued. + +Convergence is a *post-commit* signal (the write landed; do the replicas need +reconciliation), whereas the #1312 fence is a *commit* gate (a stale epoch is +rejected before the write lands, surfaced through the existing `Result::Err` +channel). They compose on the one `rename_data` path rather than competing: +the fence decides whether a convergence is produced at all, and +`RenameConvergence` classifies it once produced. A future fence-aware +convergence variant, if ever needed, is an additive change to that enum and +does not disturb the epoch comparison at the disk-write points above. + ## Transport and security contract Generation and all derived tokens (lease, reservation) cross node boundaries in