From 38c03d9d5db4f16aed5894e2cff26082052a3dfa Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 16:22:13 +0800 Subject: [PATCH 1/8] chore(deps): refresh scanner heal batch dependency baseline Regenerate compatible lockfile selections before the next implementation batch. Cargo upgrade leaves direct requirements unchanged. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- Cargo.lock | 47 ++++++++++++++++++++++++++++------------------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8206b1951..1daa30d5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1679,7 +1679,7 @@ version = "0.10.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" dependencies = [ - "generic-array 0.14.7", + "generic-array 0.14.9", ] [[package]] @@ -1698,7 +1698,7 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93" dependencies = [ - "generic-array 0.14.7", + "generic-array 0.14.9", ] [[package]] @@ -2110,7 +2110,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common 0.1.7", + "crypto-common 0.1.6", "inout 0.1.4", ] @@ -2580,7 +2580,7 @@ version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" dependencies = [ - "generic-array 0.14.7", + "generic-array 0.14.9", "rand_core 0.6.4", "subtle", "zeroize", @@ -2605,11 +2605,11 @@ dependencies = [ [[package]] name = "crypto-common" -version = "0.1.7" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" dependencies = [ - "generic-array 0.14.7", + "generic-array 0.14.9", "typenum", ] @@ -3901,7 +3901,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", "const-oid 0.9.6", - "crypto-common 0.1.7", + "crypto-common 0.1.6", "subtle", ] @@ -4166,7 +4166,7 @@ dependencies = [ "crypto-bigint 0.5.5", "digest 0.10.7", "ff 0.13.1", - "generic-array 0.14.7", + "generic-array 0.14.9", "group 0.13.0", "hkdf 0.12.4", "pem-rfc7468 0.7.0", @@ -4499,7 +4499,7 @@ checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a" dependencies = [ "io-lifetimes 2.0.4", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -4622,9 +4622,9 @@ dependencies = [ [[package]] name = "generic-array" -version = "0.14.7" +version = "0.14.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" dependencies = [ "typenum", "version_check", @@ -4637,7 +4637,7 @@ version = "1.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e" dependencies = [ - "generic-array 0.14.7", + "generic-array 0.14.9", "rustversion", "typenum", ] @@ -5660,7 +5660,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ "block-padding 0.3.3", - "generic-array 0.14.7", + "generic-array 0.14.9", ] [[package]] @@ -5693,7 +5693,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20fd6de4ccfcc187e38bc21cfa543cb5a302cb86a8b114eb7f0bf0dc9f8ac00f" dependencies = [ "io-lifetimes 3.0.1", - "windows-sys 0.52.0", + "windows-sys 0.60.2", ] [[package]] @@ -7115,7 +7115,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.5.0", @@ -11399,7 +11399,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" dependencies = [ "base16ct 0.2.0", "der 0.7.10", - "generic-array 0.14.7", + "generic-array 0.14.9", "pkcs8 0.10.2", "subtle", "zeroize", @@ -12399,7 +12399,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -13651,6 +13651,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" @@ -13823,7 +13832,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d" dependencies = [ "bitflags 2.13.1", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] From 652ebb12c64d091f8b6bbc24c79211094c2e68e1 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 16:28:01 +0800 Subject: [PATCH 2/8] fix(ecstore): remove duplicate local rename implementation Keep the canonical commit module after concurrent storage changes merged. The control-write and rollback changes are already present there. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/ecstore/src/disk/local.rs | 997 ------------------------------- 1 file changed, 997 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index bf2239a9f..9cd683578 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -9862,1004 +9862,7 @@ fn should_read_legacy_inline_part(fi: &FileInfo, storage_class_config: &crate::c storage_class_config.should_inline(shard_size, fi.erasure.data_blocks, versioned) } -/// Proof produced only when the local rename returns at an existing access -/// preflight, before metadata, backups, or object data can be published. -#[derive(Debug)] -pub(in crate::disk) struct LocalRenamePreflightRejection(()); - impl LocalDisk { - #[tracing::instrument(name = "rename_data", level = "trace", skip_all)] - async fn rename_data_inner( - &self, - src_volume: &str, - src_path: &str, - fi: FileInfo, - dst_volume: &str, - dst_path: &str, - preflight_rejection: &mut Option, - ) -> Result { - crate::hp_guard!("LocalDisk::rename_data"); - let mut fi = fi; - // A non-force DeleteBucket must not remove a directory while a local - // object commit is publishing into it. The peer's empty scan remains - // optimistic; this lease establishes the local commit/delete order and - // remains owned by any blocking syscall that outlives async cancellation. - let destination_object_path = self.io_get_object_path(dst_volume, dst_path)?; - let quota_fence_token = - match rustfs_utils::http::metadata_compat::get_consistent_str(&fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX) { - Some(value) => { - let token = Uuid::parse_str(value).map_err(|_| DiskError::FileCorrupt)?; - Some(SnapshotLeaseToken::from_slice(token.as_bytes())?) - } - None if rustfs_utils::http::metadata_compat::contains_key_str( - &fi.metadata, - QUOTA_MUTATION_FENCE_METADATA_SUFFIX, - ) => - { - return Err(DiskError::FileCorrupt); - } - None => None, - }; - rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX); - let quota_fence_claim = match quota_fence_token { - Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?), - None => None, - }; - let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await; - if let Some(claim) = quota_fence_claim { - mutation_lease.attach_external_guard(claim); - } - if fi.is_legacy_indexed_delete_marker() { - fi.erasure.index = 0; - } - fi.validate_for_metadata_read()?; - // Snapshot the destination part paths before `fi` is consumed below. These - // are the descriptors a reader may hold for the version this call is about - // to replace (backlog#1145); readers build the identical string in - // `io_primitives`. An inline-data version has no parts and yields none. - let invalidate_part_paths: Vec = { - let data_dir = fi.data_dir.unwrap_or_default(); - fi.parts - .iter() - .map(|part| format!("{dst_path}/{data_dir}/part.{}", part.number)) - .collect() - }; - let src_volume_dir = self.io_get_bucket_path(src_volume)?; - if !skip_access_checks(src_volume) - && let Err(e) = super::fs::access_std(&src_volume_dir) - { - info!( - event = EVENT_DISK_LOCAL_ACCESS_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = ?src_volume_dir, - operation = "rename_data_src_access", - error = %e, - "Disk local access check failed" - ); - *preflight_rejection = Some(LocalRenamePreflightRejection(())); - return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); - } - - let dst_volume_dir = self.io_get_bucket_path(dst_volume)?; - if !skip_access_checks(dst_volume) - && let Err(e) = super::fs::access_std(&dst_volume_dir) - { - info!( - event = EVENT_DISK_LOCAL_ACCESS_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = ?dst_volume_dir, - operation = "rename_data_dst_access", - error = %e, - "Disk local access check failed" - ); - *preflight_rejection = Some(LocalRenamePreflightRejection(())); - return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); - } - - // xl.meta path - let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?; - let dst_file_path = self.io_get_object_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; - - // data_dir path - let has_data_dir_path = { - let has_data_dir = { - if !fi.is_remote() { - fi.data_dir - .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str())) - } else { - None - } - }; - - if let Some(data_dir) = has_data_dir { - let src_data_path = self.io_get_object_path( - src_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(), - )?; - let dst_data_path = self.io_get_object_path( - dst_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(), - )?; - - Some((src_data_path, dst_data_path)) - } else { - None - } - }; - - check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; - check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; - - let no_inline = fi.data.is_none() && fi.size > 0; - // Captured before `fi` is consumed by add_version; gates the stale - // destination purge below. - let fi_healing = fi.is_healing(); - - // Resolved once for the whole commit so a concurrent configuration - // change can never leave a single rename_data half-synced. The tier is - // keyed on the destination volume: user data staged in scratch - // namespaces follows the configured tier, while commits into - // system-critical namespaces (IAM, config, bucket metadata) stay - // pinned to strict. - let durability = effective_durability(dst_volume); - - let src_file_parent = src_file_path - .parent() - .ok_or_else(|| DiskError::other("missing staged metadata parent"))?; - let dst_file_parent = dst_file_path - .parent() - .ok_or_else(|| DiskError::other("missing object metadata parent"))?; - if !no_inline { - fs::create_dir_all(src_file_parent).await.map_err(to_file_error)?; - } - // Acquire the common trees before reading destination metadata. On - // Windows this pins the object directory identity across metadata - // preparation, data publication, rollback backup, and final commit. - let rename_commit_guard = lock_rename_commit_directories( - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?; - - if no_inline { - // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta - let mut xlmeta = FileMeta::new(); - // An existing dst xl.meta that fails to parse leaves `xlmeta` empty - // and gets overwritten by the commit below (pre-existing behavior); - // track that so the old-size observation reports unknown instead of - // a false `Absent` (rustfs/backlog#1009). - let mut dst_meta_unparsable = false; - if let Some(dst_buf) = has_dst_buf.as_ref() { - if FileMeta::is_xl2_v1_format(dst_buf) - && let Ok(nmeta) = FileMeta::load(dst_buf) - { - xlmeta = nmeta - } else { - dst_meta_unparsable = true; - } - } - - let old_current_size = if dst_meta_unparsable { - None - } else { - observe_old_current_size(has_dst_buf.is_some(), &xlmeta) - }; - - let mut skip_parent = dst_volume_dir.clone(); - if has_dst_buf.as_ref().is_some() - && let Some(parent) = dst_file_path.parent() - { - skip_parent = parent.to_path_buf(); - } - - let version_id = fi.version_id.unwrap_or_default(); - let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); - let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); - let rollback_data_dir = has_old_data_dir.or_else(|| { - if old_version_exists && has_dst_buf.is_some() { - Some(inline_metadata_rollback_dir(version_id, &xlmeta)) - } else { - None - } - }); - if let Some(old_data_dir) = has_old_data_dir.as_ref() { - let _ = xlmeta.data.remove_two(version_id, *old_data_dir); - } - xlmeta.add_version(fi)?; - let version_signature = rename_data_versions_signature(&xlmeta); - let new_dst_buf = xlmeta.marshal_msg()?; - - // This tmp xl.meta is renamed onto dst_file_path at the commit - // point below, so only its contents must be durable before the - // rename (SyncMode::FileOnly); the dst parent directory is fsynced - // after the commit rename, and a crash before the rename means the - // PUT was never acknowledged. A metadata commit: relaxed tiers - // leave it to the page cache. - let tmp_meta_sync = if durability.syncs_commit_metadata() { - SyncMode::FileOnly - } else { - SyncMode::None - }; - // The tmp xl.meta write and the shard-file fdatasync are independent - // (disjoint paths) and both only need to be durable before the commit - // renames below, so run them concurrently to drop a blocking - // round-trip from the PUT commit critical path (rustfs/backlog#922 - // step 2). The "contents durable -> rename -> dst dir fsync" ordering - // is unchanged — both futures complete before any rename — which the - // rename_data crash-consistency harness (backlog#935) exercises. - // - // Shard durability: once rename_data succeeds the write is - // acknowledged, so data must not live only in the page cache. - // Multipart parts were already synced during rename_part, so their - // fdatasync here is a cheap no-op. A missing source dir is left for the - // rename below to report through the existing rollback path. Payload - // durability is kept by both strict and relaxed. - let tmp_meta_write = { - let src_file_path = src_file_path.clone(); - let dst_file_path = dst_file_path.clone(); - let rename_commit_guard = rename_commit_guard.clone(); - let mutation_lease = mutation_lease.clone(); - async move { - os::run_blocking_namespace_operation(mutation_lease, move || { - #[cfg(test)] - run_owned_file_write_before_open(&src_file_path); - let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard( - &src_file_path, - &dst_file_path, - &rename_commit_guard, - )?; - prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?; - Ok(prepared_metadata_source) - }) - .await - .map_err(to_file_error) - .map_err(DiskError::from) - } - }; - let shard_sync = async { - if durability.syncs_data_shards() - && let Some((src_data_path, _)) = has_data_dir_path.as_ref() - && let Err(err) = os::sync_dir_files_with_limiter(src_data_path, self.file_sync_permits.clone()).await - && err.kind() != ErrorKind::NotFound - { - return Err::<(), DiskError>(to_file_error(err).into()); - } - Ok(()) - }; - let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync); - // Surface a tmp-meta failure first (its prior serial position), then a - // shard-sync failure; either aborts before any rename, exactly as the - // sequential version did. - let prepared_metadata_source = tmp_meta_res?; - shard_sync_res?; - let rename_commit_guard = remove_dst_base_before_commit( - dst_path, - rename_commit_guard, - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - if should_remove_staged_meta_before_commit(dst_path) { - drop(prepared_metadata_source); - std::fs::remove_file(&src_file_path).map_err(to_file_error)?; - return Err(DiskError::FileNotFound); - } - - // Heal reuses the version's data_dir, so for in-place corruption - // the destination dir still exists — and rename(2) cannot replace - // a non-empty directory (EEXIST on XFS, ENOTEMPTY on ext4). Purge - // it first, healing commits only; fresh PUTs mint a new data_dir - // and never collide. Best effort: a real failure surfaces in the - // rename below. - if fi_healing - && let Some((_, dst_data_path)) = has_data_dir_path.as_ref() - && let Err(err) = self.move_to_trash(dst_data_path, true, false).await - { - warn!( - event = EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - dst_path = ?dst_data_path, - error = ?err, - "Healing commit could not purge the stale destination data dir" - ); - } - if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() - && let Err(err) = os::rename_all_with_commit_guard( - src_data_path, - dst_data_path, - &skip_parent, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "rename_all_data_path_failed", - src_path = ?src_data_path, - dst_path = ?dst_data_path, - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - #[cfg(test)] - if has_data_dir_path.is_some() { - run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); - } - - // Crash-consistency injection: hard power loss after the data dir - // is in place but before xl.meta commits. No cleanup — the harness - // reopens the disk and asserts the object still reads as the old - // version (the staged data dir is a harmless orphan for GC). - if crash_inject::should_crash_at(CrashPoint::RenameAfterDataRename, dst_path) { - return Err(DiskError::Unexpected); - } - - if should_fail_before_old_metadata_backup(dst_path) { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "test_fail_before_old_metadata_backup", - "Disk local rename flow failed before metadata commit" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(DiskError::Unexpected); - } - - // The rollback backup stays where it is written (no rename) and is - // the sole restore source for a later undo_write, so under strict - // it keeps SyncMode::FileAndDir: contents and directory entry both - // durable. It is part of the metadata commit machinery, so relaxed - // tiers leave it to the page cache like the xl.meta it mirrors. - let backup_sync = if durability.syncs_commit_metadata() { - SyncMode::FileAndDir - } else { - SyncMode::None - }; - if let (Some(old_data_dir), Some(dst_buf)) = (rollback_data_dir, has_dst_buf.as_ref()) { - let backup_parent = dst_file_parent.join(old_data_dir.to_string()); - #[cfg(not(windows))] - if let Err(err) = os::make_dir_all(&backup_parent, &skip_parent).await { - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - let backup_path_guard = match rename_commit_guard.create_destination_directory_for_path_access(&backup_parent) { - Ok(guard) => guard, - Err(err) => { - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(DiskError::from(to_file_error(err))); - } - }; - let backup_path = backup_parent.join(STORAGE_FORMAT_FILE_BACKUP); - if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) { - #[cfg(windows)] - drop(backup_path_guard); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - let backup_bytes = dst_buf.clone(); - // Keep the volume, commit-tree, and exact destination-path - // guards in this task until the backup write and durability - // sync finish. A detached spawn_blocking writer could survive - // cancellation and later truncate a newer transaction's - // deterministic rollback backup. - let write_result = os::run_blocking_namespace_operation(mutation_lease.clone(), move || { - #[cfg(test)] - run_owned_file_write_before_open(&backup_path); - backup_path_guard.write_file_for_path_access( - &backup_path, - backup_bytes.as_ref(), - backup_sync != SyncMode::None, - backup_sync == SyncMode::FileAndDir, - ) - }) - .await - .map_err(to_file_error) - .map_err(DiskError::from); - if let Err(err) = write_result { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "write_old_metadata_backup_failed", - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - } - - // Crash-consistency injection: hard power loss after the rollback - // backup is durable but before the xl.meta commit rename. No - // cleanup — the harness asserts the object still reads as the old - // version, since the destination xl.meta is untouched here. - if crash_inject::should_crash_at(CrashPoint::RenameAfterBackupBeforeMetaCommit, dst_path) { - return Err(DiskError::Unexpected); - } - - if let Err(err) = os::rename_all_with_prepared_source( - prepared_metadata_source, - &src_file_path, - &dst_file_path, - &skip_parent, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "rename_all_metadata_failed", - src_path = ?src_file_path, - dst_path = ?dst_file_path, - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - - let committed_new_data_path = has_data_dir_path.as_ref().map(|(_, dst_data_path)| dst_data_path.as_path()); - if should_fail_after_metadata_commit(dst_path) { - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - return Err(DiskError::Unexpected); - } - - // Crash-consistency injection: hard power loss immediately after the - // xl.meta commit rename but before the durability fsync. Unlike the - // graceful failpoint above, no rollback runs — the commit rename is - // already on disk, so the harness asserts the object reads back as - // the new version. - if crash_inject::should_crash_at(CrashPoint::RenameAfterMetaCommit, dst_path) { - return Err(DiskError::Unexpected); - } - - // Persist the directory entries for both the data dir and xl.meta renames; - // without this the commit itself can vanish on power loss. Relaxed tiers - // accept that window (documented in docs/operations/durability-modes.md). - if durability.syncs_commit_metadata() - && let Some(parent) = dst_file_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dst_dir_group_commit(parent).await { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - // The commit rename changed the dst part inodes before this fsync - // failed and rolled them back; drop any fd cached during that - // window so readers re-open the restored inode (rustfs/backlog#1177). - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(to_file_error(err).into()); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - } - - // First PUT of an object creates its directory (and any missing prefix - // dirs) via reliable_mkdir_all, which never fsyncs the parent chain. The - // commit fsync above persists the object dir's *contents*, not its own - // entry in the bucket/prefix dir, so on power loss after ack the whole - // object dir could vanish (rustfs/backlog#922 step 4). For a new object - // (no prior xl.meta) fsync the ancestor chain from the object dir's - // parent up to and including the bucket so those new directory entries - // are durable. Overwrites already have a durable object dir. The - // starts_with guard bounds the walk to the bucket subtree. Relaxed/none - // accept the wider window, like the commit fsync above. - if has_dst_buf.is_none() && durability.syncs_commit_metadata() { - let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); - while let Some(dir) = ancestor { - if !dir.starts_with(&dst_volume_dir) { - break; - } - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dir(dir).await { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - // Same post-commit rollback window as above — drop cached - // dst part fds so readers re-open the restored inode - // (rustfs/backlog#1177). - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(to_file_error(err).into()); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - if dir == dst_volume_dir.as_path() { - break; - } - ancestor = dir.parent(); - } - } - - // Publication and every rollback-capable durability step are now - // complete. Do not retain the Windows object identity guard while - // cleaning staging paths or invalidating cached descriptors. - #[cfg(windows)] - drop(rename_commit_guard); - - if let Some(src_file_path_parent) = src_file_path.parent() { - if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { - let _ = std::fs::remove_dir(src_file_path_parent); - } else { - let _ = self - .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) - .await; - } - } - - // Heal reuses a version's `data_dir` and lands the rebuilt shard on - // the SAME `//part.N` path. Without this, a cached - // descriptor would keep serving the pre-heal inode, defeating the heal - // and eroding read quorum (backlog#1145). - // - // The exact keys are derivable here, and this runs on every write, so - // use them rather than registering a predicate the read path would then - // have to evaluate. Readers build the same string - // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every - // part of the version now at `dst_path` — any part path absent from it - // no longer exists for readers to ask for. - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - - Ok(RenameDataResp { - old_data_dir: has_old_data_dir, - rollback_data_dir, - cleanup_data_dir: has_old_data_dir, - sign: version_signature, - old_current_size, - }) - } else { - // Inline metadata preparation is blocking. The transaction lease is - // moved into that work so a timeout can release the async waiter without - // allowing a retry to reuse the deterministic staging path too early. - let src = src_file_path.clone(); - let dst = dst_file_path.clone(); - let cleanup_path = if src_volume == super::RUSTFS_META_MULTIPART_BUCKET { - src_file_path.parent().map(|p| p.to_path_buf()) - } else { - None - }; - let dst_path_for_failpoint = dst_path.to_string(); - #[cfg(windows)] - let source_parent = src_file_parent.to_path_buf(); - let rename_commit_guard_for_preparation = rename_commit_guard.clone(); - let sync = durability.syncs_commit_metadata(); - #[cfg(test)] - run_inline_before_file_sync_admission(dst_path); - let mut file_sync_admission = if sync { - Some( - os::acquire_file_sync_admission(self.file_sync_permits.clone()) - .await - .map_err(to_file_error) - .map_err(DiskError::from)?, - ) - } else { - None - }; - let prepare_inline_metadata = move || { - let mut prepared_metadata_source = - os::create_prepared_rename_source_with_commit_guard(&src, &dst, &rename_commit_guard_for_preparation)?; - #[cfg(windows)] - let source_metadata_guard = - rename_commit_guard_for_preparation.lock_source_directory_for_path_access(&source_parent)?; - let mut xlmeta = FileMeta::new(); - // Same as the non-inline branch: an unparsable existing dst - // xl.meta must surface as unknown, not `Absent` - // (rustfs/backlog#1009). - let mut dst_meta_unparsable = false; - if let Some(ref buf) = has_dst_buf { - if FileMeta::is_xl2_v1_format(buf) - && let Ok(nmeta) = FileMeta::load(buf) - { - xlmeta = nmeta - } else { - dst_meta_unparsable = true; - } - } - - let old_current_size = if dst_meta_unparsable { - None - } else { - observe_old_current_size(has_dst_buf.is_some(), &xlmeta) - }; - - let version_id = fi.version_id.unwrap_or_default(); - let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); - let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); - let rollback_data_dir = old_data_dir.or_else(|| { - if old_version_exists && has_dst_buf.is_some() { - Some(inline_metadata_rollback_dir(version_id, &xlmeta)) - } else { - None - } - }); - let mut staged_rollback_path = None; - if let Some(d) = old_data_dir.as_ref() { - let _ = xlmeta.data.remove_two(version_id, *d); - } - xlmeta.add_version(fi)?; - let version_signature = rename_data_versions_signature(&xlmeta); - let new_buf = xlmeta.marshal_msg()?; - // Write the staged xl.meta. Inline objects carry their data inside - // xl.meta, so this is the durable preparation for the metadata commit: - // relaxed tiers do no per-object fsync here at all (aligned - // with MinIO's default), trading a documented power-loss - // window for latency. - prepared_metadata_source.write_all(&new_buf, sync)?; - run_inline_preparation_before_backup(&dst_path_for_failpoint); - if let Some(ref old_metadata) = has_dst_buf - && (rollback_data_dir.is_some() || sync || cfg!(test)) - { - #[cfg(windows)] - let backup_path = { - let backup_path = src - .parent() - .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"))? - .join(STORAGE_FORMAT_FILE_BACKUP); - source_metadata_guard.write_file_for_path_access(&backup_path, old_metadata, sync, false)?; - backup_path - }; - #[cfg(not(windows))] - let backup_path = create_local_inline_rollback_backup(&dst, &src, old_metadata)?; - #[cfg(not(windows))] - if sync { - std::fs::File::open(&backup_path)?.sync_data()?; - } - staged_rollback_path = Some(backup_path); - } - - Ok::<_, std::io::Error>(( - rollback_data_dir, - old_data_dir, - version_signature, - old_current_size, - staged_rollback_path, - has_dst_buf.is_none(), - prepared_metadata_source, - )) - }; - let inline_preparation = if let Some(admission) = file_sync_admission.as_ref() { - os::run_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, prepare_inline_metadata).await - } else { - os::run_blocking_namespace_operation(mutation_lease.clone(), prepare_inline_metadata).await - } - .map_err(to_file_error) - .map_err(DiskError::from); - - let ( - rollback_data_dir, - cleanup_data_dir, - version_signature, - old_current_size, - mut local_rollback_path, - destination_was_absent, - prepared_metadata_source, - ) = match inline_preparation { - Ok(prepared) => prepared, - Err(err) => { - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(err); - } - }; - - let rename_commit_guard = remove_dst_base_before_commit( - dst_path, - rename_commit_guard, - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - - if should_remove_staged_meta_before_commit(dst_path) { - drop(prepared_metadata_source); - let remove_result = std::fs::remove_file(&src_file_path); - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - remove_result.map_err(to_file_error)?; - return Err(DiskError::FileNotFound); - } - - if let (Some(rollback_data_dir), Some(staged_backup)) = (rollback_data_dir, local_rollback_path.as_deref()) { - let Some(dst_parent) = dst_file_path.parent() else { - return Err(DiskError::other("missing object metadata parent")); - }; - let backup_path = dst_parent - .join(rollback_data_dir.to_string()) - .join(STORAGE_FORMAT_FILE_BACKUP); - // rename_all acquires the backup path's namespace lease. Do not - // hold a disk admission while acquiring another namespace lock. - drop(file_sync_admission.take()); - if let Err(err) = rename_all(staged_backup, &backup_path, &dst_volume_dir, &self.publication_root).await { - let _ = remove_file_if_exists(staged_backup); - return Err(err); - } - #[cfg(test)] - run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); - if sync { - file_sync_admission = Some( - os::acquire_file_sync_admission(self.file_sync_permits.clone()) - .await - .map_err(to_file_error) - .map_err(DiskError::from)?, - ); - } - if let Some(admission) = file_sync_admission.as_ref() - && let Some(backup_parent) = backup_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, - fsync_started, - ); - return Err(DiskError::from(to_file_error(err))); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, - fsync_started, - ); - } - local_rollback_path = None; - } - - let commit_result = if should_fail_commit_rename(dst_path) { - Err(DiskError::other("test fail during metadata commit rename")) - } else { - os::rename_all_with_prepared_source( - prepared_metadata_source, - &src_file_path, - &dst_file_path, - &dst_volume_dir, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - }; - if let Err(err) = commit_result { - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(err); - } - - let post_commit = async { - if should_fail_after_metadata_commit(dst_path) { - rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; - return Err(std::io::Error::other("test fail after metadata commit")); - } - - // Persist the commit rename's directory entry across power loss. - if let Some(admission) = file_sync_admission.as_ref() - && let Some(dst_parent) = dst_file_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission) - .await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; - return Err(err); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - } - - // Same power-loss gap as the non-inline path (rustfs/backlog#922 - // step 4): a first PUT creates the object dir (and any missing - // prefix dirs) whose entry in the bucket/prefix dir reliable_mkdir_all - // never fsynced. The fsync above persists the object dir's contents, - // not its own entry, so for a new inline object fsync the ancestor - // chain up to and including the bucket. Overwrites already have a - // durable object dir; the starts_with guard bounds the walk. - if let Some(admission) = file_sync_admission.as_ref() - && destination_was_absent - { - let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); - while let Some(ancestor_dir) = ancestor { - if !ancestor_dir.starts_with(&dst_volume_dir) { - break; - } - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - rollback_inline_metadata_commit_std( - &dst_file_path, - rollback_data_dir, - local_rollback_path.as_deref(), - )?; - return Err(err); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - if ancestor_dir == dst_volume_dir.as_path() { - break; - } - ancestor = ancestor_dir.parent(); - } - } - - Ok::<(), std::io::Error>(()) - } - .await; - - // The disk admission protects the durability chain, not staging - // cleanup or cache invalidation after that chain has completed. - drop(file_sync_admission.take()); - - // A post-commit rollback (for example, a commit-metadata fsync - // failure under strict durability) restores the old metadata; drop any - // descriptors cached during the committed window before propagating the - // error (rustfs/backlog#1177). Inline objects carry data in xl.meta, so - // this is mostly defensive and keeps both commit branches consistent. - if let Err(err) = post_commit { - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(DiskError::from(err)); - } - - // The commit no longer has a rollback path. Release the Windows - // object identity guard before best-effort staging cleanup. - #[cfg(windows)] - drop(rename_commit_guard); - - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - - // Cleanup - if let Some(ref cleanup) = cleanup_path { - let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await; - } else if let Some(parent) = src_file_path.parent() { - let _ = std::fs::remove_dir(parent); - } - - // Heal reuses a version's `data_dir` and lands the rebuilt shard on - // the SAME `//part.N` path. Without this, a cached - // descriptor would keep serving the pre-heal inode, defeating the heal - // and eroding read quorum (backlog#1145). - // - // The exact keys are derivable here, and this runs on every write, so - // use them rather than registering a predicate the read path would then - // have to evaluate. Readers build the same string - // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every - // part of the version now at `dst_path` — any part path absent from it - // no longer exists for readers to ask for. - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - - Ok(RenameDataResp { - old_data_dir: cleanup_data_dir, - rollback_data_dir, - cleanup_data_dir, - sign: version_signature, - old_current_size, - }) - } - } - - pub(in crate::disk) async fn rename_data_observed( - &self, - src_volume: &str, - src_path: &str, - fi: &FileInfo, - dst_volume: &str, - dst_path: &str, - ) -> super::RenameDataObservation { - let mut preflight_rejection = None; - let result = self - .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut preflight_rejection) - .await; - super::RenameDataObservation { - result, - preflight_rejection, - } - } - pub(crate) async fn rename_data_borrowed( &self, src_volume: &str, From 53efaa2b8ff91e48f8612e425943c9d01bc21b99 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 19:10:54 +0800 Subject: [PATCH 3/8] chore(deps): refresh profiling dependencies for the next batch Update hotpath and its macro crate to the compatible patch release before the next dependency-ready implementation tasks. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- Cargo.lock | 111 +++++++++-------------------------------------------- Cargo.toml | 2 +- 2 files changed, 20 insertions(+), 93 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1daa30d5f..9e929d439 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2220,7 +2220,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "958c5d6ecf1f214b4c2bbbbf6ab9523a864bd136dcf71a7e8904799acfe1ad47" dependencies = [ "unicode-segmentation", - "unicode-width 0.2.2", + "unicode-width", ] [[package]] @@ -3927,16 +3927,6 @@ dependencies = [ "dirs-sys", ] -[[package]] -name = "dirs-next" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1" -dependencies = [ - "cfg-if", - "dirs-sys-next", -] - [[package]] name = "dirs-sys" version = "0.5.0" @@ -3945,19 +3935,8 @@ checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab" dependencies = [ "libc", "option-ext", - "redox_users 0.5.2", - "windows-sys 0.61.2", -] - -[[package]] -name = "dirs-sys-next" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d" -dependencies = [ - "libc", - "redox_users 0.4.6", - "winapi", + "redox_users", + "windows-sys 0.59.0", ] [[package]] @@ -4295,7 +4274,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5286,9 +5265,9 @@ dependencies = [ [[package]] name = "hotpath" -version = "0.25.0" +version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ec7782e005cabd5eaf350febde384cd799faa3a0e624587aa8759c240e0b592" +checksum = "366c5db5a8e4643bdde121a1d6f35cfa035e7850b8b8ea94c6960a139d44e93a" dependencies = [ "arc-swap", "async-channel", @@ -5297,6 +5276,7 @@ dependencies = [ "crossbeam-channel", "flate2", "futures-channel", + "futures-core", "futures-util", "hdrhistogram", "hotpath-macros", @@ -5305,9 +5285,8 @@ dependencies = [ "object 0.36.7", "parking_lot", "pin-project-lite", - "prettytable-rs", "quanta", - "regex", + "regex-lite", "reqwest", "reqwest-middleware", "rustc-demangle", @@ -5319,9 +5298,9 @@ dependencies = [ [[package]] name = "hotpath-macros" -version = "0.25.0" +version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "929b2285d2cd21b2733a7fb6ebc843bb4f83dbd1db0122f5f9ebb9567b1e2613" +checksum = "846bde0d9600d98434e1aac376977d7718bfe3d2f5312a041b7c59a6a466c51a" dependencies = [ "proc-macro2", "quote", @@ -5693,7 +5672,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20fd6de4ccfcc187e38bc21cfa543cb5a302cb86a8b114eb7f0bf0dc9f8ac00f" dependencies = [ "io-lifetimes 3.0.1", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -5750,17 +5729,6 @@ dependencies = [ "serde", ] -[[package]] -name = "is-terminal" -version = "0.4.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" -dependencies = [ - "hermit-abi", - "libc", - "windows-sys 0.61.2", -] - [[package]] name = "is_debug" version = "1.1.0" @@ -6955,7 +6923,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7115,7 +7083,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.22.1", + "base64 0.21.7", "chrono", "getrandom 0.2.17", "http 1.5.0", @@ -8186,19 +8154,6 @@ dependencies = [ "syn 3.0.5", ] -[[package]] -name = "prettytable-rs" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a" -dependencies = [ - "encode_unicode", - "is-terminal", - "lazy_static", - "term", - "unicode-width 0.1.14", -] - [[package]] name = "primefield" version = "0.14.0" @@ -8706,7 +8661,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -8980,17 +8935,6 @@ dependencies = [ "bitflags 2.13.1", ] -[[package]] -name = "redox_users" -version = "0.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" -dependencies = [ - "getrandom 0.2.17", - "libredox", - "thiserror 1.0.69", -] - [[package]] name = "redox_users" version = "0.5.2" @@ -11059,7 +11003,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -11142,7 +11086,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -12399,21 +12343,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.61.2", -] - -[[package]] -name = "term" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f" -dependencies = [ - "dirs-next", - "rustversion", - "winapi", + "windows-sys 0.59.0", ] [[package]] @@ -13182,12 +13115,6 @@ version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" -[[package]] -name = "unicode-width" -version = "0.1.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" - [[package]] name = "unicode-width" version = "0.2.2" @@ -13521,7 +13448,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a7ecf889b..b1903badf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -371,7 +371,7 @@ dav-server = "0.11.0" # Performance Analysis and Memory Profiling rustfs-mimalloc = { version = "0.5.3" } -hotpath = { version = "0.25.0", default-features = false } +hotpath = { version = "0.25.1", default-features = false } # Snapshot testing for output format regression detection insta = { version = "1.48" } From bdbdca07c86c957b14e6b3c83b5452757eacea62 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 19:50:46 +0800 Subject: [PATCH 4/8] fix(deps): preserve supported hotpath focus expressions Keep the profiler runtime before its regex-lite compatibility regression. Track the opt-in validation required to remove this constraint in backlog. Refs rustfs/backlog#2302. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- Cargo.lock | 107 ++++++++++++++++++++++++++++++++++++++++++++--------- Cargo.toml | 3 +- 2 files changed, 92 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9e929d439..2d25d1530 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2220,7 +2220,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "958c5d6ecf1f214b4c2bbbbf6ab9523a864bd136dcf71a7e8904799acfe1ad47" dependencies = [ "unicode-segmentation", - "unicode-width", + "unicode-width 0.2.2", ] [[package]] @@ -3927,6 +3927,16 @@ dependencies = [ "dirs-sys", ] +[[package]] +name = "dirs-next" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1" +dependencies = [ + "cfg-if", + "dirs-sys-next", +] + [[package]] name = "dirs-sys" version = "0.5.0" @@ -3935,8 +3945,19 @@ checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab" dependencies = [ "libc", "option-ext", - "redox_users", - "windows-sys 0.59.0", + "redox_users 0.5.2", + "windows-sys 0.61.2", +] + +[[package]] +name = "dirs-sys-next" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d" +dependencies = [ + "libc", + "redox_users 0.4.6", + "winapi", ] [[package]] @@ -4274,7 +4295,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5265,9 +5286,9 @@ dependencies = [ [[package]] name = "hotpath" -version = "0.25.1" +version = "0.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "366c5db5a8e4643bdde121a1d6f35cfa035e7850b8b8ea94c6960a139d44e93a" +checksum = "2ec7782e005cabd5eaf350febde384cd799faa3a0e624587aa8759c240e0b592" dependencies = [ "arc-swap", "async-channel", @@ -5276,7 +5297,6 @@ dependencies = [ "crossbeam-channel", "flate2", "futures-channel", - "futures-core", "futures-util", "hdrhistogram", "hotpath-macros", @@ -5285,8 +5305,9 @@ dependencies = [ "object 0.36.7", "parking_lot", "pin-project-lite", + "prettytable-rs", "quanta", - "regex-lite", + "regex", "reqwest", "reqwest-middleware", "rustc-demangle", @@ -5672,7 +5693,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20fd6de4ccfcc187e38bc21cfa543cb5a302cb86a8b114eb7f0bf0dc9f8ac00f" dependencies = [ "io-lifetimes 3.0.1", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -5729,6 +5750,17 @@ dependencies = [ "serde", ] +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "is_debug" version = "1.1.0" @@ -6923,7 +6955,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7083,7 +7115,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.5.0", @@ -8154,6 +8186,19 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "prettytable-rs" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a" +dependencies = [ + "encode_unicode", + "is-terminal", + "lazy_static", + "term", + "unicode-width 0.1.14", +] + [[package]] name = "primefield" version = "0.14.0" @@ -8661,7 +8706,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8935,6 +8980,17 @@ dependencies = [ "bitflags 2.13.1", ] +[[package]] +name = "redox_users" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" +dependencies = [ + "getrandom 0.2.17", + "libredox", + "thiserror 1.0.69", +] + [[package]] name = "redox_users" version = "0.5.2" @@ -11003,7 +11059,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -11086,7 +11142,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -12343,10 +12399,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.61.2", +] + +[[package]] +name = "term" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f" +dependencies = [ + "dirs-next", + "rustversion", + "winapi", ] [[package]] @@ -13115,6 +13182,12 @@ version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" +[[package]] +name = "unicode-width" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" + [[package]] name = "unicode-width" version = "0.2.2" @@ -13448,7 +13521,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index b1903badf..9d0008be8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -371,7 +371,8 @@ dav-server = "0.11.0" # Performance Analysis and Memory Profiling rustfs-mimalloc = { version = "0.5.3" } -hotpath = { version = "0.25.1", default-features = false } +# Preserve Unicode focus filters until rustfs/backlog#2302 is resolved. +hotpath = { version = "=0.25.0", default-features = false } # Snapshot testing for output format regression detection insta = { version = "1.48" } From cbd3ff9ad73ac71af41f1bd75a7843c0636e7aa7 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 16:52:32 +0800 Subject: [PATCH 5/8] feat(heal): record bounded canonical object and task outcomes Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/channel.rs | 6 + crates/heal/src/heal/manager.rs | 6 + crates/heal/src/heal/manager/queue.rs | 3 + crates/heal/src/heal/manager/scheduler.rs | 1 + crates/heal/src/heal/manager/tests.rs | 11 + crates/heal/src/heal/mod.rs | 1 + crates/heal/src/heal/outcome.rs | 300 +++++++++++++++++++++ crates/heal/src/heal/task.rs | 110 ++++++++ crates/heal/src/heal/task/heal_bucket.rs | 59 +++- crates/heal/src/heal/task/heal_metadata.rs | 4 +- crates/heal/src/heal/task/tests.rs | 299 +++++++++++++++++++- 11 files changed, 795 insertions(+), 5 deletions(-) create mode 100644 crates/heal/src/heal/outcome.rs diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 33ac33c38..580b01246 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -447,6 +447,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "running".to_string(), None, @@ -463,6 +464,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "running".to_string(), Some(format!("heal task retrying after recoverable failure, attempt {retry_attempt}: {error}")), @@ -479,6 +481,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "finished".to_string(), None, @@ -495,6 +498,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "stopped".to_string(), Some("heal task cancelled".to_string()), @@ -511,6 +515,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "stopped".to_string(), Some("heal task timed out".to_string()), @@ -527,6 +532,7 @@ impl HealChannelProcessor { progress, next_seq, min_seq, + .. }) => ( "stopped".to_string(), Some(error), diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 971590fdb..c90a18729 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::heal::{ + outcome::HealTaskOutcome, progress::{HealProgress, HealStatistics}, resume::{ReplacementPhase, ResumeGc, ResumeManager, ResumeState, ResumeUtils}, storage::HealStorageAPI, @@ -185,6 +186,7 @@ fn record_displaced_terminal( request: &HealRequest, ) -> Arc { let terminal = Arc::new(CompletedHealStatus { + outcome: None, progress: None, retained_bytes: std::sync::OnceLock::new(), heal_type: request.heal_type.clone(), @@ -268,6 +270,7 @@ async fn publish_completed_heal( #[derive(Debug, Clone)] pub struct HealTaskReport { + pub outcome: Option>, pub status: HealTaskStatus, pub result_items: Vec, pub result_items_truncated: bool, @@ -285,6 +288,7 @@ async fn active_task_report(task: &HealTask, since: Option) -> HealTaskRepo let window = task.get_result_items_since(since).await; HealTaskReport { status: task.get_status().await, + outcome: Some(Arc::new(task.get_outcome().await)), result_items: window.items, // The legacy flag stays set once anything was evicted; a lagging // incremental cursor additionally marks this response truncated so @@ -298,6 +302,7 @@ async fn active_task_report(task: &HealTask, since: Option) -> HealTaskRepo fn empty_task_report(status: HealTaskStatus) -> HealTaskReport { HealTaskReport { + outcome: None, status, result_items: Vec::new(), result_items_truncated: false, @@ -325,6 +330,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option) -> }; HealTaskReport { status: completed.status.clone(), + outcome: completed.outcome.clone(), result_items, result_items_truncated: completed.result_items_truncated || lagged, progress: completed.progress.clone(), diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index aceabe42a..a47cd43e2 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -83,6 +83,7 @@ pub(super) struct CompletedHealStatus { pub(super) heal_type: HealType, pub(super) status: HealTaskStatus, pub(super) progress: Option, + pub(super) outcome: Option>, pub(super) retained_bytes: std::sync::OnceLock, pub(super) result_items_truncated: bool, pub(super) completed_at: SystemTime, @@ -105,6 +106,7 @@ impl CompletedHealStatus { fn measure_retained_bytes(&self) -> usize { let mut bytes = size_of::(); let mut add = |amount: usize| bytes = bytes.saturating_add(amount); + add(self.outcome.as_ref().map_or(0, |outcome| outcome.retained_bytes())); match &self.heal_type { HealType::Cluster => {} HealType::Bucket { bucket } => add(bucket.capacity()), @@ -209,6 +211,7 @@ impl CompletedHealStatus { heal_type: task.heal_type.clone(), status, progress: Some(task.get_progress().await), + outcome: Some(Arc::new(task.get_outcome().await)), retained_bytes: std::sync::OnceLock::new(), result_items_truncated: task.result_items_truncated(), completed_at: SystemTime::now(), diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index feb0c1231..704a7cbeb 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -298,6 +298,7 @@ impl HealManager { if cancelled_completion { completed_status = HealTaskStatus::Cancelled; completed_status_entry.status = HealTaskStatus::Cancelled; + completed_status_entry.outcome = Some(Arc::new(task.get_outcome().await)); } let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. }); let successful_completion = matches!(completed_status, HealTaskStatus::Completed); diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index aa1c1293f..cb5a52c1b 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -103,6 +103,7 @@ struct MockStorage; fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus { CompletedHealStatus { + outcome: None, heal_type: HealType::Cluster, status: HealTaskStatus::Completed, progress: Some(HealProgress { @@ -325,6 +326,10 @@ async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() { for token in [&task_id, &alias] { let report = manager.get_task_report(token).await.expect("cancelled token retained"); assert_eq!(report.status, HealTaskStatus::Cancelled); + assert_eq!( + report.outcome.as_ref().expect("cancelled outcome retained").execution, + crate::heal::outcome::HealExecutionOutcome::Aborted(crate::heal::outcome::HealAbortReason::Cancelled) + ); assert_eq!(report.progress.expect("frozen progress").objects_scanned, 1); } assert!(!manager.retrying_heals.lock().await.contains_key(&task_id)); @@ -391,6 +396,7 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han .expect("scheduler archives terminal"); assert!(!manager.active_heals.lock().await.contains_key(&task_id)); let expected = task.get_progress().await; + let expected_outcome = task.get_outcome().await; for token in [&task_id, &alias] { assert_eq!(manager.get_task_progress(token).await.expect("terminal progress query"), expected); let report = manager @@ -398,6 +404,7 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han .await .expect("terminal token remains queryable at handoff"); assert_eq!(report.progress.as_ref(), Some(&expected)); + assert_eq!(report.outcome.as_deref(), Some(&expected_outcome)); assert!(report.result_items.is_empty()); match outcome { "success" => assert_eq!(report.status, HealTaskStatus::Completed), @@ -1976,6 +1983,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) -> task_id, Arc::new(CompletedHealStatus { progress: None, + outcome: None, retained_bytes: std::sync::OnceLock::new(), heal_type: request.heal_type, status: HealTaskStatus::Retrying { @@ -2700,6 +2708,7 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() { task_id.clone(), Arc::new(CompletedHealStatus { progress: None, + outcome: None, retained_bytes: std::sync::OnceLock::new(), heal_type: request.heal_type.clone(), status: HealTaskStatus::Retrying { @@ -2737,6 +2746,7 @@ async fn test_get_task_status_reads_recent_completed_status() { "completed-token".to_string(), Arc::new(CompletedHealStatus { progress: None, + outcome: None, retained_bytes: std::sync::OnceLock::new(), heal_type: HealType::Bucket { bucket: "bucket".to_string(), @@ -2768,6 +2778,7 @@ async fn test_get_task_report_for_path_reads_completed_items() { "completed-token".to_string(), Arc::new(CompletedHealStatus { progress: None, + outcome: None, retained_bytes: std::sync::OnceLock::new(), heal_type: HealType::Object { bucket: "bucket".to_string(), diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index c918bea49..5f17c8cd8 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -16,6 +16,7 @@ pub mod channel; pub mod erasure_healer; pub mod manager; pub mod mrf_queue; +pub mod outcome; pub mod progress; pub(crate) mod replacement_readiness; pub mod resume; diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs new file mode 100644 index 000000000..36eab0225 --- /dev/null +++ b/crates/heal/src/heal/outcome.rs @@ -0,0 +1,300 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Execution results are separate from repair responsibility. A legacy +//! successful storage call supplies no authoritative repair receipt. + +use std::{collections::VecDeque, time::SystemTime}; +use uuid::Uuid; + +const MAX_OUTCOME_ITEMS: usize = 128; +const MAX_OUTCOME_BYTES: usize = 64 * 1024; +const MAX_OUTCOME_DETAIL_BYTES: usize = 1024; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HealObjectKind { + Object, + Metadata, + Decode, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HealObjectIdentity { + pub kind: HealObjectKind, + pub bucket: String, + pub object: String, + /// The requested version; None remains unresolved, never an absence proof. + pub version_id: Option, + pub bucket_incarnation_id: Option, + pub pool_index: Option, + pub set_index: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HealDeferredReason { + DanglingDeleteGrace, + TransientUsageCache, + TransientExistenceCheck, + Deadline, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HealFailureClass { + Recoverable, + RetryExhausted, + Permanent, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HealObjectDisposition { + /// The legacy storage response does not prove the requested check or commit. + Unknown, + Repaired, + VerifiedHealthy, + AuthoritativelyAbsent, + Deferred { + reason: HealDeferredReason, + retry_not_before: Option, + }, + Failed(HealFailureClass), + Cancelled, + DryRunObserved, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HealObjectOutcome { + pub identity: HealObjectIdentity, + pub disposition: HealObjectDisposition, + pub detail: Option, +} + +impl HealObjectOutcome { + fn retained_bytes(&self) -> usize { + size_of::() + .saturating_add(self.identity.bucket.capacity()) + .saturating_add(self.identity.object.capacity()) + .saturating_add(self.identity.version_id.as_ref().map_or(0, String::capacity)) + .saturating_add(self.detail.as_ref().map_or(0, String::capacity)) + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum HealTraversalCoverage { + #[default] + Unknown, + Partial, + Complete, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HealAbortReason { + Cancelled, + Deadline, + Untraversable, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum HealExecutionOutcome { + #[default] + Pending, + Running, + Completed, + CompletedWithErrors, + Aborted(HealAbortReason), +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct HealOutcomeCounters { + pub processed: u64, + pub healed: u64, + pub unchanged: u64, + /// Deferred, cancelled, dry-run and unverified results remain unresolved. + pub skipped: u64, + pub failed: u64, + pub unknown: u64, + pub attempt_failures: u64, + pub overflowed: bool, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct HealTaskOutcome { + pub execution: HealExecutionOutcome, + pub coverage: HealTraversalCoverage, + pub counters: HealOutcomeCounters, + /// A bounded diagnostic window, not a complete responsibility ledger. + pub objects: VecDeque, + pub objects_truncated: bool, + retained_object_bytes: usize, + untraversable: bool, +} + +impl HealTaskOutcome { + pub(crate) fn start(&mut self) { + self.execution = HealExecutionOutcome::Running; + self.coverage = HealTraversalCoverage::Partial; + } + + pub(crate) fn attempt_failed(&mut self) { + self.counters.overflowed |= !super::progress::increment_counter(&mut self.counters.attempt_failures); + } + + pub(crate) fn mark_untraversable(&mut self) { + self.untraversable = true; + self.coverage = HealTraversalCoverage::Partial; + } + + pub(crate) fn finish(&mut self, abort: Option) { + let abort = abort.or(self.untraversable.then_some(HealAbortReason::Untraversable)); + self.execution = match abort { + Some(reason) => HealExecutionOutcome::Aborted(reason), + None if self.counters.failed > 0 => HealExecutionOutcome::CompletedWithErrors, + None => HealExecutionOutcome::Completed, + }; + self.coverage = if abort.is_none() && !self.counters.overflowed { + HealTraversalCoverage::Complete + } else { + HealTraversalCoverage::Partial + }; + } + + pub(crate) fn record(&mut self, mut item: HealObjectOutcome) { + use super::progress::increment_counter; + let counters = &mut self.counters; + counters.overflowed |= !increment_counter(&mut counters.processed); + let counter = match item.disposition { + HealObjectDisposition::Repaired => &mut counters.healed, + HealObjectDisposition::VerifiedHealthy | HealObjectDisposition::AuthoritativelyAbsent => &mut counters.unchanged, + HealObjectDisposition::Failed(_) => &mut counters.failed, + HealObjectDisposition::Unknown => { + counters.overflowed |= !increment_counter(&mut counters.unknown); + &mut counters.skipped + } + _ => &mut counters.skipped, + }; + counters.overflowed |= !increment_counter(counter); + if let Some(detail) = &mut item.detail { + let mut end = detail.len().min(MAX_OUTCOME_DETAIL_BYTES); + while !detail.is_char_boundary(end) { + end -= 1; + } + self.objects_truncated |= end < detail.len(); + detail.truncate(end); + detail.shrink_to_fit(); + } + let bytes = item.retained_bytes(); + if bytes > MAX_OUTCOME_BYTES { + self.objects_truncated = true; + return; + } + while self.objects.len() >= MAX_OUTCOME_ITEMS || self.retained_object_bytes.saturating_add(bytes) > MAX_OUTCOME_BYTES { + let Some(oldest) = self.objects.pop_front() else { break }; + self.retained_object_bytes = self.retained_object_bytes.saturating_sub(oldest.retained_bytes()); + self.objects_truncated = true; + } + self.retained_object_bytes = self.retained_object_bytes.saturating_add(bytes); + self.objects.push_back(item); + } + + pub(crate) fn retained_bytes(&self) -> usize { + size_of::() + .saturating_add(self.retained_object_bytes) + .saturating_add(self.objects.capacity().saturating_mul(size_of::())) + } +} + +#[cfg(test)] +mod canonical_outcome_tests { + use super::*; + + fn item(disposition: HealObjectDisposition) -> HealObjectOutcome { + HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + bucket_incarnation_id: None, + pool_index: None, + set_index: None, + }, + disposition, + detail: None, + } + } + + #[test] + fn canonical_outcome_categories_have_one_terminal_count() { + let mut outcome = HealTaskOutcome::default(); + for disposition in [ + HealObjectDisposition::Unknown, + HealObjectDisposition::Repaired, + HealObjectDisposition::VerifiedHealthy, + HealObjectDisposition::AuthoritativelyAbsent, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + retry_not_before: None, + }, + HealObjectDisposition::Failed(HealFailureClass::Permanent), + HealObjectDisposition::Cancelled, + HealObjectDisposition::DryRunObserved, + ] { + outcome.record(item(disposition)); + } + let c = &outcome.counters; + assert_eq!((c.processed, c.healed, c.unchanged, c.skipped, c.failed, c.unknown), (8, 1, 2, 4, 1, 1)); + assert_eq!(c.processed, c.healed + c.unchanged + c.skipped + c.failed); + } + + #[test] + fn canonical_outcome_window_count_bytes_and_oversize_keep_total_counts() { + let mut outcome = HealTaskOutcome::default(); + for _ in 0..MAX_OUTCOME_ITEMS { + outcome.record(item(HealObjectDisposition::Unknown)); + } + assert_eq!(outcome.objects.len(), MAX_OUTCOME_ITEMS); + assert!(!outcome.objects_truncated); + outcome.record(item(HealObjectDisposition::Unknown)); + assert_eq!(outcome.objects.len(), MAX_OUTCOME_ITEMS); + assert!(outcome.objects_truncated); + let mut oversized = item(HealObjectDisposition::Failed(HealFailureClass::Permanent)); + oversized.identity.object = "x".repeat(MAX_OUTCOME_BYTES); + outcome.record(oversized); + assert_eq!(outcome.counters.processed, u64::try_from(MAX_OUTCOME_ITEMS + 2).expect("bounded count")); + assert_eq!(outcome.counters.failed, 1); + assert!(outcome.retained_object_bytes <= MAX_OUTCOME_BYTES); + for _ in 0..MAX_OUTCOME_ITEMS { + let mut failed = item(HealObjectDisposition::Failed(HealFailureClass::Permanent)); + failed.detail = Some("\u{4fee}".repeat(MAX_OUTCOME_DETAIL_BYTES)); + outcome.record(failed); + } + assert!(outcome.retained_object_bytes <= MAX_OUTCOME_BYTES); + assert!(outcome.objects.iter().all(|item| { + item.detail + .as_ref() + .is_none_or(|detail| detail.len() <= MAX_OUTCOME_DETAIL_BYTES) + })); + assert!(outcome.objects.len() < MAX_OUTCOME_ITEMS); + } + + #[test] + fn canonical_outcome_counter_overflow_cannot_claim_complete_coverage() { + let mut outcome = HealTaskOutcome::default(); + outcome.counters.processed = u64::MAX; + outcome.record(item(HealObjectDisposition::Unknown)); + outcome.finish(None); + assert!(outcome.counters.overflowed); + assert_eq!(outcome.counters.processed, u64::MAX); + assert_eq!(outcome.coverage, HealTraversalCoverage::Partial); + } +} diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index c4c1d902b..1d72bfb25 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -15,6 +15,10 @@ use crate::heal::{ DiskError, EcstoreError, ErasureSetHealer, HealDiskExt as _, erasure_healer::target_outcomes_complete, + outcome::{ + HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectDisposition, HealObjectIdentity, HealObjectKind, + HealObjectOutcome, HealTaskOutcome, + }, progress::HealProgress, resume::{ CheckpointManager, ReplacementPhase, ReplacementTargetIdentity, ResumeManager, replacement_target_identities_match, @@ -394,6 +398,7 @@ pub struct HealTask { pub status: Arc>, /// Progress tracking pub progress: Arc>, + outcome: Arc>, /// Result items collected from storage heal calls, each stamped with a /// monotonically increasing sequence number for incremental consumption /// (the client passes the last seen seq back and receives only newer @@ -460,6 +465,7 @@ impl HealTask { result_items_truncated: Arc::new(AtomicBool::new(false)), batch_failure: Arc::new(RwLock::new(None)), batch_failure_recorded: Arc::new(AtomicBool::new(false)), + outcome: Arc::new(RwLock::new(HealTaskOutcome::default())), created_at: request.created_at, enqueued_at: request.enqueued_at, started_at: Arc::new(RwLock::new(None)), @@ -507,6 +513,66 @@ impl HealTask { self.heal_type.kind_label() } + pub async fn get_outcome(&self) -> HealTaskOutcome { + self.outcome.read().await.clone() + } + + fn outcome_identity( + &self, + bucket: &str, + object: &str, + version_id: Option<&str>, + pool_index: Option, + set_index: Option, + ) -> HealObjectIdentity { + HealObjectIdentity { + kind: match self.heal_type { + HealType::Metadata { .. } => HealObjectKind::Metadata, + HealType::ECDecode { .. } => HealObjectKind::Decode, + _ => HealObjectKind::Object, + }, + bucket: bucket.to_owned(), + object: object.to_owned(), + version_id: version_id.map(ToOwned::to_owned), + bucket_incarnation_id: None, + pool_index, + set_index, + } + } + + fn single_object_identity(&self) -> Option { + let (bucket, object, version) = match &self.heal_type { + HealType::Object { + bucket, + object, + version_id, + } + | HealType::ECDecode { + bucket, + object, + version_id, + } => (bucket, object, version_id.as_deref()), + HealType::Metadata { bucket, object } => (bucket, object, None), + _ => return None, + }; + Some(self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index)) + } + + async fn record_deferred_object(&self, reason: HealDeferredReason) { + if let Some(identity) = self.single_object_identity() { + let mut outcome = self.outcome.write().await; + outcome.attempt_failed(); + outcome.record(HealObjectOutcome { + identity, + disposition: HealObjectDisposition::Deferred { + reason, + retry_not_before: None, + }, + detail: None, + }); + } + } + pub(crate) fn has_batch_failure(&self) -> bool { self.batch_failure_recorded.load(Ordering::Acquire) } @@ -634,6 +700,7 @@ impl HealTask { } async fn skip_due_to_transient_object_exists(&self, bucket: &str, object: &str, err: &Error) -> Result<()> { + self.record_deferred_object(HealDeferredReason::TransientExistenceCheck).await; warn!( target: "rustfs::heal::task", event = EVENT_HEAL_OBJECT_RESULT, @@ -733,6 +800,8 @@ impl HealTask { return false; } + self.record_deferred_object(HealDeferredReason::TransientUsageCache).await; + warn!( target: "rustfs::heal::task", event = EVENT_HEAL_OBJECT_RESULT, @@ -755,6 +824,8 @@ impl HealTask { return false; } + self.record_deferred_object(HealDeferredReason::DanglingDeleteGrace).await; + warn!( target: "rustfs::heal::task", event = EVENT_HEAL_OBJECT_RESULT, @@ -801,6 +872,7 @@ impl HealTask { #[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))] #[hotpath::measure] pub async fn execute(&self) -> Result<()> { + self.outcome.write().await.start(); // update status and timestamps atomically to avoid race conditions let now = SystemTime::now(); let start_instant = Instant::now(); @@ -860,6 +932,43 @@ impl HealTask { HealType::ErasureSet { buckets, set_disk_id } => self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await, }; + { + let mut outcome = self.outcome.write().await; + if outcome.counters.processed == 0 + && let Some(identity) = self.single_object_identity() + { + let disposition = match &result { + Ok(()) if self.options.dry_run => HealObjectDisposition::DryRunObserved, + Ok(()) => HealObjectDisposition::Unknown, + Err(Error::TaskCancelled) => HealObjectDisposition::Cancelled, + Err(Error::TaskTimeout) => HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + }, + Err(error) => { + outcome.attempt_failed(); + HealObjectDisposition::Failed(if error.is_recoverable_heal() { + HealFailureClass::Recoverable + } else { + HealFailureClass::Permanent + }) + } + }; + outcome.record(HealObjectOutcome { + identity, + disposition, + detail: result.as_ref().err().map(ToString::to_string), + }); + } + let abort = match &result { + Err(Error::TaskCancelled) => Some(HealAbortReason::Cancelled), + Err(Error::TaskTimeout) => Some(HealAbortReason::Deadline), + Err(_) if !self.has_batch_failure() && !self.heal_type.is_per_object() => Some(HealAbortReason::Untraversable), + _ => None, + }; + outcome.finish(abort); + } + // update completed time and status { let mut completed_at = self.completed_at.write().await; @@ -944,6 +1053,7 @@ impl HealTask { pub async fn cancel(&self) -> Result<()> { self.cancel_token.cancel(); + self.outcome.write().await.finish(Some(HealAbortReason::Cancelled)); let mut status = self.status.write().await; *status = HealTaskStatus::Cancelled; debug!( diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index d44df549e..241aaba5a 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -214,6 +214,7 @@ impl HealTask { continue; } failed = failed.saturating_add(1); + self.outcome.write().await.mark_untraversable(); if err.is_recoverable_heal() { retryable = retryable.saturating_add(1); } else { @@ -260,6 +261,7 @@ impl HealTask { #[hotpath::measure] async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> { + let previous_progress = self.get_progress().await; let mut scanned = 0u64; let mut healed = 0u64; let mut failed = 0u64; @@ -338,6 +340,14 @@ impl HealTask { self.check_control_flags().await?; let mut telemetry_unknown = false; let object = item.name.as_str(); + let identity = + self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set); + let mut disposition = if heal_opts.dry_run { + HealObjectDisposition::DryRunObserved + } else { + HealObjectDisposition::Unknown + }; + let mut detail = None; { let mut progress = self.progress.write().await; progress.set_current_object(Some(format!("{bucket}/{object}"))); @@ -380,7 +390,31 @@ impl HealTask { }; if let Some(err) = error { + match err { + Error::TaskCancelled | Error::TaskTimeout => { + let disposition = if matches!(err, Error::TaskCancelled) { + HealObjectDisposition::Cancelled + } else { + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + } + }; + self.outcome.write().await.record(HealObjectOutcome { + identity, + disposition, + detail: None, + }); + return Err(err); + } + _ => self.outcome.write().await.attempt_failed(), + } + detail = Some(err.to_string()); if Self::is_dangling_delete_grace_error(&err) { + disposition = HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + retry_not_before: None, + }; telemetry_unknown |= !increment_counter(&mut skipped); warn!( target: "rustfs::heal::task", @@ -395,6 +429,10 @@ impl HealTask { "Heal bucket object dangling cleanup deferred by grace window" ); } else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) { + disposition = HealObjectDisposition::Deferred { + reason: HealDeferredReason::TransientUsageCache, + retry_not_before: None, + }; telemetry_unknown |= !increment_counter(&mut skipped); warn!( target: "rustfs::heal::task", @@ -425,6 +463,11 @@ impl HealTask { ); retry.push(item); } else { + disposition = HealObjectDisposition::Failed(if err.is_recoverable_heal() { + HealFailureClass::RetryExhausted + } else { + HealFailureClass::Permanent + }); telemetry_unknown |= !increment_counter(&mut failed); if err.is_recoverable_heal() { retryable_failed = retryable_failed.saturating_add(1); @@ -459,8 +502,20 @@ impl HealTask { continue; } + self.outcome.write().await.record(HealObjectOutcome { + identity, + disposition, + detail, + }); + let mut progress = self.progress.write().await; - progress.update_object_progress(scanned, healed, failed, skipped, bytes); + progress.update_object_progress( + previous_progress.objects_scanned.saturating_add(scanned), + previous_progress.objects_healed.saturating_add(healed), + previous_progress.objects_failed.saturating_add(failed), + previous_progress.skipped_objects.saturating_add(skipped), + previous_progress.bytes_processed.saturating_add(bytes), + ); if telemetry_unknown { progress.mark_unknown(); } @@ -475,7 +530,7 @@ impl HealTask { continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; if continuation_token.is_none() { - // Truncated but no continuation token: end of listing. + self.outcome.write().await.mark_untraversable(); break; } } diff --git a/crates/heal/src/heal/task/heal_metadata.rs b/crates/heal/src/heal/task/heal_metadata.rs index fe2dc98ac..fad5a403e 100644 --- a/crates/heal/src/heal/task/heal_metadata.rs +++ b/crates/heal/src/heal/task/heal_metadata.rs @@ -261,8 +261,8 @@ impl HealTask { update_parity: true, no_lock: self.options.no_lock, read_repair: false, - pool: None, - set: None, + pool: self.options.pool_index, + set: self.options.set_index, }; let heal_result = self diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 1734d0c7a..9693f0ef0 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -14,6 +14,294 @@ use super::super::{DiskOption, DiskStore, Endpoint, new_disk}; use super::*; + +mod canonical_outcome { + use super::*; + use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage}; + + fn bucket_task(storage: Arc) -> HealTask { + HealTask::from_request( + HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage, + ) + } + + #[tokio::test] + async fn listing_failure_preserves_processed_objects_and_partial_coverage() { + let storage = Arc::new(MockStorage { + fail_second_listing_page: true, + ..Default::default() + }); + let task = bucket_task(storage); + task.execute().await.expect_err("second page cannot be traversed"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable)); + assert_eq!(outcome.coverage, HealTraversalCoverage::Partial); + assert_eq!(outcome.counters.processed, 1); + assert_eq!(outcome.objects[0].identity.object, "object-a"); + assert_eq!(task.get_progress().await.objects_scanned, 1); + } + + #[tokio::test] + async fn cluster_preserves_cumulative_progress_across_buckets() { + let storage = Arc::new(MockStorage { + list_each_bucket: true, + listed_buckets: Mutex::new(Some(vec!["bucket-a".to_string(), "bucket-b".to_string()])), + ..Default::default() + }); + let task = HealTask::from_request( + HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage, + ); + task.execute().await.expect("both buckets complete"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 4); + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + let progress = task.get_progress().await; + assert_eq!((progress.objects_scanned, progress.objects_healed), (4, 4)); + assert_eq!( + outcome + .objects + .iter() + .filter(|item| item.identity.bucket == "bucket-b") + .count(), + 2 + ); + } + + #[tokio::test(start_paused = true)] + async fn exhausted_object_does_not_abort_other_objects_or_erase_counts() { + let storage = Arc::new(MockStorage::default()); + storage.heal_object_outcomes.lock().expect("outcomes").insert( + "object-a".to_string(), + (0..4).map(|_| MockHealObjectOutcome::RetryableReadQuorum).collect(), + ); + let task = bucket_task(storage.clone()); + task.execute().await.expect_err("legacy adapter retains batch failure"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::CompletedWithErrors); + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + assert_eq!((outcome.counters.processed, outcome.counters.failed, outcome.counters.unknown), (2, 1, 1)); + assert_eq!(outcome.counters.attempt_failures, 4); + let failed = outcome + .objects + .iter() + .find(|item| item.identity.object == "object-a") + .expect("failed object"); + assert_eq!(failed.disposition, HealObjectDisposition::Failed(HealFailureClass::RetryExhausted)); + let calls = storage.heal_object_calls.lock().expect("calls"); + assert_eq!(calls.iter().filter(|object| object.as_str() == "object-b").count(), 1); + let progress = task.get_progress().await; + assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 1, 1)); + } + + #[tokio::test(start_paused = true)] + async fn retry_success_counts_one_terminal_outcome() { + let storage = Arc::new(MockStorage::default()); + storage + .heal_object_outcomes + .lock() + .expect("outcomes") + .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::RetryableReadQuorum])); + let task = bucket_task(storage); + task.execute().await.expect("retry should recover"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Completed); + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.counters.failed, 0); + assert_eq!(outcome.counters.attempt_failures, 1); + assert_eq!( + outcome + .objects + .iter() + .filter(|item| item.identity.object == "object-a") + .count(), + 1 + ); + assert_eq!( + outcome.counters.processed, + outcome.counters.healed + outcome.counters.unchanged + outcome.counters.skipped + outcome.counters.failed + ); + } + + #[tokio::test] + async fn mixed_grace_and_legacy_success_keep_distinct_dispositions() { + let storage = Arc::new(MockStorage::default()); + storage + .heal_object_outcomes + .lock() + .expect("outcomes") + .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::DanglingGraceDeferred])); + let task = bucket_task(storage); + task.execute().await.expect("grace permits traversal completion"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.counters.healed, 0, "legacy result is not a repair receipt"); + assert!(matches!( + outcome.objects[0].disposition, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + .. + } + )); + assert_eq!(outcome.objects[1].disposition, HealObjectDisposition::Unknown); + assert!( + outcome + .objects + .iter() + .all(|item| item.identity.bucket_incarnation_id.is_none()) + ); + assert_eq!( + task.get_progress().await.objects_healed, + 1, + "legacy display count remains distinct from proof" + ); + } + + #[tokio::test] + async fn grace_single_object_is_completed_but_deferred() { + let storage = Arc::new(MockStorage { + heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::DanglingGraceDeferred)), + ..Default::default() + }); + let task = HealTask::from_request(HealRequest::object("bucket-a".to_string(), "recent.txt".to_string(), None), storage); + task.execute().await.expect("grace is deferred"); + let outcome = task.get_outcome().await; + assert_eq!(task.get_status().await, HealTaskStatus::Completed); + assert_eq!(outcome.counters.processed, 1); + assert!(matches!( + outcome.objects[0].disposition, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + .. + } + )); + assert_eq!(outcome.counters.attempt_failures, 1); + } + + #[tokio::test] + async fn dry_run_and_transient_existence_do_not_prove_repair() { + for transient in [false, true] { + let storage = Arc::new(MockStorage::default()); + if transient { + storage + .object_exists_by_name + .lock() + .expect("existence fixture") + .insert("object".to_string(), MockObjectExists::TransientSkip("retry later")); + } + let mut request = HealRequest::object("bucket-a".to_string(), "object".to_string(), None); + request.options.dry_run = !transient; + let task = HealTask::from_request(request, storage); + task.execute().await.expect("observation may complete"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.healed, 0); + if transient { + assert!(matches!( + outcome.objects[0].disposition, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::TransientExistenceCheck, + .. + } + )); + } else { + assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::DryRunObserved); + } + } + } + + #[tokio::test] + async fn untraversable_bucket_does_not_claim_complete_cluster_coverage() { + let storage = Arc::new(MockStorage { + listed_buckets: Mutex::new(Some(vec!["bucket-a".to_string(), "bucket-b".to_string()])), + bucket_heal_errors: Mutex::new(HashMap::from([("bucket-a".to_string(), VecDeque::from(["metadata unavailable"]))])), + ..Default::default() + }); + let task = HealTask::from_request( + HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + task.execute().await.expect_err("structural bucket error"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable)); + assert_eq!(outcome.coverage, HealTraversalCoverage::Partial); + assert_eq!( + storage.bucket_heal_calls.lock().expect("bucket calls").as_slice(), + ["bucket-a", "bucket-b"] + ); + } + + #[tokio::test(start_paused = true)] + async fn cancellation_and_deadline_leave_partial_coverage() { + for cancel in [false, true] { + let storage = Arc::new(MockStorage { + block_heal_object: Mutex::new(true), + ..Default::default() + }); + let mut request = HealRequest::object("bucket-a".to_string(), "object".to_string(), None); + request.options.timeout = Some(Duration::from_secs(1)); + let task = HealTask::from_request(request, storage); + if cancel { + task.cancel().await.expect("cancel request"); + } + task.execute().await.expect_err("control interruption"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.coverage, HealTraversalCoverage::Partial); + assert_eq!( + outcome.execution, + HealExecutionOutcome::Aborted(if cancel { + HealAbortReason::Cancelled + } else { + HealAbortReason::Deadline + }) + ); + } + } + + #[tokio::test] + async fn decode_keeps_the_requested_pool_and_set() { + let storage = Arc::new(MockStorage::default()); + let mut request = HealRequest::ec_decode("bucket-a".to_string(), "object".to_string(), Some("version-a".to_string())); + request.options.pool_index = Some(2); + request.options.set_index = Some(3); + let task = HealTask::from_request(request, storage.clone()); + task.execute().await.expect("decode fixture"); + let options = storage.object_heal_opts.lock().expect("storage options"); + assert_eq!((options[0].pool, options[0].set), (Some(2), Some(3))); + let outcome = task.get_outcome().await; + let identity = &outcome.objects[0].identity; + assert_eq!((identity.pool_index, identity.set_index), (Some(2), Some(3))); + assert_eq!(identity.version_id.as_deref(), Some("version-a")); + assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown); + } +} use crate::heal::storage::{HealListItem, HealObjectInfo}; use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos}; @@ -582,6 +870,8 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() { #[derive(Default)] struct MockStorage { listed: Mutex, + list_each_bucket: bool, + fail_second_listing_page: bool, healed_objects: Mutex>, heal_object_calls: Mutex>, heal_object_version_ids: Mutex>>, @@ -995,12 +1285,19 @@ impl HealStorageAPI for MockStorage { _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { self.listed_prefixes.lock().unwrap().push(prefix.to_string()); + if self.fail_second_listing_page { + return if continuation_token.is_none() { + Ok((vec![heal_item("object-a")], Some("next-page".to_string()), true)) + } else { + Err(Error::other("listing unavailable")) + }; + } if *self.truncate_without_token.lock().unwrap() { return Ok((vec![heal_item("object-a")], None, true)); } let mut listed = self.listed.lock().unwrap(); - if continuation_token.is_none() && !*listed { + if continuation_token.is_none() && (!*listed || self.list_each_bucket) { *listed = true; let objects = if bucket == RUSTFS_META_BUCKET { vec![ From 71b19bd5229c414275e2cfc18d352460f90ebeb6 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 18:42:40 +0800 Subject: [PATCH 6/8] fix(heal): preserve cancellation and retry only failed listing pages Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/error.rs | 9 +++ crates/heal/src/heal/manager/tests.rs | 53 ++++++++++++++ crates/heal/src/heal/outcome.rs | 7 +- crates/heal/src/heal/task.rs | 22 ++++++ crates/heal/src/heal/task/heal_bucket.rs | 58 +++++++++++----- crates/heal/src/heal/task/tests.rs | 88 ++++++++++++++++++++++++ 6 files changed, 219 insertions(+), 18 deletions(-) diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index 336ba00ad..7fa8d7717 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -54,6 +54,15 @@ pub enum Error { #[error("Heal task execution failed: {message}")] TaskExecutionFailed { message: String }, + /// The current page already exhausted its local retry budget. Retrying + /// the enclosing bucket would replay pages whose results were counted. + #[error("Heal listing failed for bucket {bucket}: {source}")] + HealListingFailed { + bucket: String, + #[source] + source: Box, + }, + #[error("Invalid heal type: {heal_type}")] InvalidHealType { heal_type: String }, diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index cb5a52c1b..d8b3e3b16 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -288,6 +288,59 @@ pub(super) async fn pause_completed_retention_before_publish(task_id: &str, stat } } +#[tokio::test] +async fn canonical_outcome_cancel_wins_before_worker_finalizes_success() { + use crate::heal::outcome::{HealAbortReason, HealExecutionOutcome}; + use crate::heal::task::{OUTCOME_FINISH_TEST_HOOK, OutcomeFinishTestHook}; + let bucket = "canonical-outcome-cancel-before-finish"; + let manager = HealManager::new(Arc::new(MockStorage), None); + let request = HealRequest::object(bucket.to_string(), "object".to_string(), None); + let task_id = request.id.clone(); + let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None); + let alias = duplicate.id.clone(); + let retention_hook = Arc::new(CompletedRetentionHook::default()); + { + let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await; + hooks.insert(bucket.to_string(), retention_hook.clone()); + hooks.insert(task_id.clone(), retention_hook.clone()); + } + let finish_hook = Arc::new(OutcomeFinishTestHook { + task_id: task_id.clone(), + reached: Notify::new(), + release: Notify::new(), + }); + *OUTCOME_FINISH_TEST_HOOK.lock().await = Some(finish_hook.clone()); + manager.submit_heal_request(request).await.expect("admit original"); + manager.submit_heal_request(duplicate).await.expect("admit alias"); + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(5), retention_hook.started.notified()) + .await + .expect("storage started"); + retention_hook.execute.notify_one(); + tokio::time::timeout(Duration::from_secs(5), finish_hook.reached.notified()) + .await + .expect("storage returned before outcome finalization"); + manager.cancel_task(&alias).await.expect("cancel wins publication"); + finish_hook.release.notify_one(); + tokio::time::timeout(Duration::from_secs(5), retention_hook.handoff.notified()) + .await + .expect("scheduler completes cancelled handoff"); + for token in [&task_id, &alias] { + let report = manager.get_task_report(token).await.expect("cancelled token retained"); + assert_eq!(report.status, HealTaskStatus::Cancelled); + assert_eq!( + report.outcome.as_ref().expect("frozen outcome").execution, + HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) + ); + } + retention_hook.finish.notify_one(); + *OUTCOME_FINISH_TEST_HOOK.lock().await = None; + COMPLETED_RETENTION_HOOKS + .lock() + .await + .retain(|key, _| key != bucket && key != &task_id); +} + #[tokio::test] async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() { let bucket = "completed-retention-retry-cancel"; diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index 36eab0225..ada74376d 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -141,7 +141,9 @@ pub struct HealTaskOutcome { impl HealTaskOutcome { pub(crate) fn start(&mut self) { - self.execution = HealExecutionOutcome::Running; + if self.execution != HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) { + self.execution = HealExecutionOutcome::Running; + } self.coverage = HealTraversalCoverage::Partial; } @@ -155,6 +157,9 @@ impl HealTaskOutcome { } pub(crate) fn finish(&mut self, abort: Option) { + if self.execution == HealExecutionOutcome::Aborted(HealAbortReason::Cancelled) { + return; + } let abort = abort.or(self.untraversable.then_some(HealAbortReason::Untraversable)); self.execution = match abort { Some(reason) => HealExecutionOutcome::Aborted(reason), diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 1d72bfb25..a1b09c3b7 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -47,6 +47,26 @@ use uuid::Uuid; use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET}; +#[cfg(test)] +pub(crate) struct OutcomeFinishTestHook { + pub(crate) task_id: String, + pub(crate) reached: tokio::sync::Notify, + pub(crate) release: tokio::sync::Notify, +} + +#[cfg(test)] +pub(crate) static OUTCOME_FINISH_TEST_HOOK: std::sync::LazyLock>>> = + std::sync::LazyLock::new(|| tokio::sync::Mutex::new(None)); + +#[cfg(test)] +async fn pause_outcome_finish(task_id: &str) { + let hook = OUTCOME_FINISH_TEST_HOOK.lock().await.clone(); + if let Some(hook) = hook.filter(|hook| hook.task_id == task_id) { + hook.reached.notify_one(); + hook.release.notified().await; + } +} + const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_TASK: &str = "task"; const LOG_SUBSYSTEM_OBJECT: &str = "object"; @@ -932,6 +952,8 @@ impl HealTask { HealType::ErasureSet { buckets, set_disk_id } => self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await, }; + #[cfg(test)] + pause_outcome_finish(&self.id).await; { let mut outcome = self.outcome.write().await; if outcome.counters.processed == 0 diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index 241aaba5a..ae8c2891d 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -306,23 +306,47 @@ impl HealTask { let mut continuation_token: Option = None; loop { self.check_control_flags().await?; - let (objects, next_token, is_truncated) = if let Some(set_disk_id) = set_disk_id.as_deref() { - self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk( - set_disk_id, - bucket, - prefix, - continuation_token.as_deref(), - false, - )) - .await? - } else { - self.await_with_control(self.storage.list_objects_for_heal_page( - bucket, - prefix, - continuation_token.as_deref(), - false, - )) - .await? + let mut listing_attempt = 0; + let (objects, next_token, is_truncated) = loop { + let page = if let Some(set_disk_id) = set_disk_id.as_deref() { + self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk( + set_disk_id, + bucket, + prefix, + continuation_token.as_deref(), + false, + )) + .await + } else { + self.await_with_control(self.storage.list_objects_for_heal_page( + bucket, + prefix, + continuation_token.as_deref(), + false, + )) + .await + }; + match page { + Ok(page) => break page, + Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error), + Err(error) => { + self.outcome.write().await.attempt_failed(); + if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES { + listing_attempt += 1; + self.await_with_control(async { + tokio::time::sleep(self.bucket_object_retry_delay(listing_attempt)).await; + Ok(()) + }) + .await?; + continue; + } + self.outcome.write().await.mark_untraversable(); + return Err(Error::HealListingFailed { + bucket: bucket.to_string(), + source: Box::new(error), + }); + } + } }; let mut pending = objects; diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 9693f0ef0..3d354a8c6 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -36,6 +36,70 @@ mod canonical_outcome { ) } + #[tokio::test(start_paused = true)] + async fn cluster_retries_only_the_failed_listing_page() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(1)), + ..Default::default() + }); + let task = HealTask::from_request( + HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + task.execute().await.expect("second-page retry succeeds"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Completed); + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.counters.attempt_failures, 1); + assert_eq!(task.get_progress().await.objects_scanned, 2); + assert_eq!( + storage.heal_object_calls.lock().expect("object calls").as_slice(), + ["object-a", "object-b"] + ); + assert_eq!( + storage.listing_tokens.lock().expect("listing tokens").as_slice(), + [None, Some("second".to_string()), Some("second".to_string())] + ); + } + + #[tokio::test(start_paused = true)] + async fn exhausted_listing_page_cannot_restart_the_bucket() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(4)), + ..Default::default() + }); + let task = HealTask::from_request( + HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + task.execute().await.expect_err("listing page budget exhausted"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Untraversable)); + assert_eq!(outcome.coverage, HealTraversalCoverage::Partial); + assert_eq!(outcome.counters.processed, 1); + assert_eq!(outcome.counters.attempt_failures, 4); + assert_eq!(task.get_progress().await.objects_scanned, 1); + assert_eq!(storage.heal_object_calls.lock().expect("object calls").as_slice(), ["object-a"]); + assert_eq!(storage.bucket_heal_calls.lock().expect("bucket calls").as_slice(), ["bucket-a"]); + } + #[tokio::test] async fn listing_failure_preserves_processed_objects_and_partial_coverage() { let storage = Arc::new(MockStorage { @@ -872,6 +936,8 @@ struct MockStorage { listed: Mutex, list_each_bucket: bool, fail_second_listing_page: bool, + recoverable_second_page_failures: Mutex>, + listing_tokens: Mutex>>, healed_objects: Mutex>, heal_object_calls: Mutex>, heal_object_version_ids: Mutex>>, @@ -1285,6 +1351,28 @@ impl HealStorageAPI for MockStorage { _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { self.listed_prefixes.lock().unwrap().push(prefix.to_string()); + self.listing_tokens + .lock() + .expect("listing tokens") + .push(continuation_token.map(ToOwned::to_owned)); + if let Some(remaining) = self + .recoverable_second_page_failures + .lock() + .expect("listing failures") + .as_mut() + { + if continuation_token.is_none() { + return Ok((vec![heal_item("object-a")], Some("second".to_string()), true)); + } + if *remaining > 0 { + *remaining -= 1; + return Err(Error::Storage(EcstoreError::InsufficientReadQuorum( + bucket.to_string(), + "page".to_string(), + ))); + } + return Ok((vec![heal_item("object-b")], None, false)); + } if self.fail_second_listing_page { return if continuation_token.is_none() { Ok((vec![heal_item("object-a")], Some("next-page".to_string()), true)) From 09947213fe0bf1adde7fdbd2697554ed539e20e1 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 6 Sep 2026 02:08:30 +0800 Subject: [PATCH 7/8] fix(heal): preserve compatible listing EOF outcomes Keep truncated heal listings without continuation tokens as complete compatibility EOFs and assert the canonical task outcome. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/task/heal_bucket.rs | 2 +- crates/heal/src/heal/task/tests.rs | 10 +++++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index ae8c2891d..c6225c844 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -554,7 +554,7 @@ impl HealTask { continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; if continuation_token.is_none() { - self.outcome.write().await.mark_untraversable(); + // Truncated without a continuation token is a compatibility EOF. break; } } diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 3d354a8c6..ba842ab8b 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -1778,6 +1778,8 @@ async fn test_recursive_bucket_heal_skips_object_dir_candidates() { #[tokio::test] async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() { + use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage}; + // A version listing can report the final page as truncated with no // continuation token. That is treated as end-of-listing (not an error), // so the returned page is healed and the pass terminates cleanly instead @@ -1799,10 +1801,16 @@ async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() { ); let task = HealTask::from_request(request, storage.clone()); - task.heal_bucket("bucket-a") + task.execute() .await .expect("truncated-without-token must terminate cleanly, not loop or error"); + assert_eq!(task.get_status().await, HealTaskStatus::Completed); + let outcome = task.get_outcome().await; + assert_eq!(outcome.execution, HealExecutionOutcome::Completed); + assert_eq!(outcome.coverage, HealTraversalCoverage::Complete); + assert_eq!(outcome.counters.processed, 1); + assert_eq!( storage.healed_objects.lock().unwrap().as_slice(), ["object-a".to_string()], From 0449a9d544bcf948e5d1fea539fa66fdc3a3e213 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 6 Sep 2026 03:03:04 +0800 Subject: [PATCH 8/8] fix(heal): drop test locks before awaits Limit synchronous mock mutex guards to pre-await scopes in canonical outcome tests. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/task/tests.rs | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index ba842ab8b..8cabf1214 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -171,8 +171,11 @@ mod canonical_outcome { .find(|item| item.identity.object == "object-a") .expect("failed object"); assert_eq!(failed.disposition, HealObjectDisposition::Failed(HealFailureClass::RetryExhausted)); - let calls = storage.heal_object_calls.lock().expect("calls"); - assert_eq!(calls.iter().filter(|object| object.as_str() == "object-b").count(), 1); + let object_b_calls = { + let calls = storage.heal_object_calls.lock().expect("calls"); + calls.iter().filter(|object| object.as_str() == "object-b").count() + }; + assert_eq!(object_b_calls, 1); let progress = task.get_progress().await; assert_eq!((progress.objects_scanned, progress.objects_healed, progress.objects_failed), (2, 1, 1)); } @@ -357,8 +360,11 @@ mod canonical_outcome { request.options.set_index = Some(3); let task = HealTask::from_request(request, storage.clone()); task.execute().await.expect("decode fixture"); - let options = storage.object_heal_opts.lock().expect("storage options"); - assert_eq!((options[0].pool, options[0].set), (Some(2), Some(3))); + let pool_and_set = { + let options = storage.object_heal_opts.lock().expect("storage options"); + (options[0].pool, options[0].set) + }; + assert_eq!(pool_and_set, (Some(2), Some(3))); let outcome = task.get_outcome().await; let identity = &outcome.objects[0].identity; assert_eq!((identity.pool_index, identity.set_index), (Some(2), Some(3)));