From 51abbf321bb797be5ec8a7263d294e3e4c92bd85 Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 1 Nov 2026 23:17:18 +0800 Subject: [PATCH] fix(heal): address diagnostic lint and test isolation --- crates/heal/src/heal/task/heal_bucket.rs | 6 ++-- crates/test-utils/src/ecstore_test_compat.rs | 3 +- crates/test-utils/src/lib.rs | 36 +++++++++++++------ rustfs/src/storage/rpc/node_service.rs | 37 ++++++++++++++++---- 4 files changed, 62 insertions(+), 20 deletions(-) diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index d7e780a04..ccf1c1e54 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -25,6 +25,8 @@ const MAX_DEFERRED_AGE: Duration = Duration::from_secs(30); // Keep this extra attempt inside the existing deferred age and backoff caps. const MAX_SLOWDOWN_OBJECT_RETRIES: u32 = MAX_BUCKET_OBJECT_HEAL_RETRIES + 1; +type HealListingPage = (Vec, Option, bool); + struct DeferredObject { name: String, version_id: Option, @@ -663,9 +665,7 @@ impl HealTask { let mut listing_finished = false; let mut listing_attempt = 0; let mut listing_due = tokio::time::Instant::now(); - let mut listing_inflight: Option< - std::pin::Pin, Option, bool)>> + Send>>, - > = None; + let mut listing_inflight: Option>> = None; let scope_result: Result<()> = async { loop { self.check_control_flags().await?; diff --git a/crates/test-utils/src/ecstore_test_compat.rs b/crates/test-utils/src/ecstore_test_compat.rs index 6ed8bfaa1..f184c1979 100644 --- a/crates/test-utils/src/ecstore_test_compat.rs +++ b/crates/test-utils/src/ecstore_test_compat.rs @@ -27,9 +27,10 @@ pub(crate) mod fixture { pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint; pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints}; pub(crate) use rustfs_ecstore::api::object::{PutObjReader, SelectObjectSnapshot}; + pub(crate) use rustfs_ecstore::api::runtime::InstanceContext; #[cfg(feature = "put-object-commit-barrier")] pub(crate) use rustfs_ecstore::api::set_disk as ecstore_set_disk; - pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks}; + pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks, init_local_disks_with_instance_ctx}; pub(crate) use rustfs_storage_api::{BucketOperations, BucketOptions, MakeBucketOptions, ObjectIO}; #[cfg(test)] diff --git a/crates/test-utils/src/lib.rs b/crates/test-utils/src/lib.rs index 354716dc0..187fb7ddb 100644 --- a/crates/test-utils/src/lib.rs +++ b/crates/test-utils/src/lib.rs @@ -35,8 +35,9 @@ use std::sync::{Arc, Once}; #[cfg(feature = "put-object-commit-barrier")] use ecstore_test_compat::fixture::ecstore_set_disk; use ecstore_test_compat::fixture::{ - BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _, - PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks, + BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, MakeBucketOptions, + ObjectIO as _, PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks, + init_local_disks_with_instance_ctx, }; use tokio_util::sync::CancellationToken; @@ -90,9 +91,8 @@ pub struct TestECStoreEnv { pub temp_root: PathBuf, /// The per-disk directories (`disk1`..`diskN`) under `temp_root`. pub disk_paths: Vec, - /// The store, bootstrapped exactly like the historical test setups: - /// `init_local_disks` + `ECStore::new` on `127.0.0.1:0` (random port keeps - /// nextest's process-per-test parallelism safe). + /// The store, using the historical bootstrap context by default or the + /// explicit instance supplied to the builder, on `127.0.0.1:0`. pub ecstore: Arc, /// The single-pool, single-set topology the store was built from. /// @@ -155,6 +155,7 @@ pub struct TestECStoreEnvBuilder { prefix: String, base_dir: Option, init_bucket_metadata: bool, + instance_context: Option>, } impl Default for TestECStoreEnvBuilder { @@ -164,6 +165,7 @@ impl Default for TestECStoreEnvBuilder { prefix: "rustfs_test_utils".to_string(), base_dir: None, init_bucket_metadata: true, + instance_context: None, } } } @@ -196,6 +198,12 @@ impl TestECStoreEnvBuilder { self } + /// Use an explicit instance instead of the process bootstrap context. + pub fn instance_context(mut self, context: Arc) -> Self { + self.instance_context = Some(context); + self + } + /// Build the environment. Panics on any bootstrap failure — this is test /// scaffolding, and a broken environment must fail the test loudly. pub async fn build(self) -> TestECStoreEnv { @@ -237,14 +245,22 @@ impl TestECStoreEnvBuilder { }; let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]); - init_local_disks(endpoint_pools.clone()).await.expect("init local disks"); - // Port 0 keeps ECStore-backed integration binaries parallel-safe under // nextest: no fixed peer port is ever shared between test processes. let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().expect("parse test addr"); - let ecstore = ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new()) - .await - .expect("build test ECStore"); + let ecstore = if let Some(context) = self.instance_context { + init_local_disks_with_instance_ctx(&context, endpoint_pools.clone()) + .await + .expect("init isolated local disks"); + ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), CancellationToken::new(), context) + .await + .expect("build isolated test ECStore") + } else { + init_local_disks(endpoint_pools.clone()).await.expect("init local disks"); + ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new()) + .await + .expect("build test ECStore") + }; // The production bootstrap only persists pool.bin from the elected // first cluster node. Test stores intentionally have no cluster diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 772c32ca0..3029ba4a4 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -7572,11 +7572,20 @@ mod tests { async fn target_rpc_fixture_with_bucket_metadata(initialize_bucket_metadata: bool) -> TargetRpcFixture { super::timeout(Duration::from_secs(90), async { let root = tempfile::tempdir().expect("target RPC root"); - let env = rustfs_test_utils::TestECStoreEnv::builder() + let instance = if initialize_bucket_metadata { + crate::storage::storage_api::new_instance_ctx() + } else { + crate::storage::storage_api::bootstrap_instance_ctx() + }; + let builder = rustfs_test_utils::TestECStoreEnv::builder() .base_dir(root.path()) - .init_bucket_metadata(initialize_bucket_metadata) - .build() - .await; + .init_bucket_metadata(initialize_bucket_metadata); + let builder = if initialize_bucket_metadata { + builder.instance_context(instance.clone()) + } else { + builder + }; + let env = builder.build().await; ObjectStore::new(env.ecstore.clone()) .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) .await @@ -7589,10 +7598,9 @@ mod tests { iam.clone(), Arc::new(KmsServiceManager::new()), )); - let instance = crate::storage::storage_api::bootstrap_instance_ctx(); assert!( super::BootstrapLocalTarget::new(instance.clone()).is_for_store(&env.ecstore), - "the standard builder must use this exact instance context" + "the fixture must use this exact instance context" ); super::timeout(Duration::from_secs(10), async { while env.ecstore.namespace_commits_pending_for_test() { @@ -8325,6 +8333,21 @@ mod tests { } #[tokio::test] + #[serial_test::serial] + async fn isolated_reservation_fixture_does_not_capture_gating_store() { + let fixture = target_rpc_fixture_with_bucket_metadata(true).await; + fixture.env.make_bucket("rpc-isolated-order", false).await; + let (_store, bucket) = tokio::time::timeout( + Duration::from_secs(20), + crate::app::gating_test_env::durable_quota_test_bucket("rpc-isolated-order", 1024), + ) + .await + .expect("gating fixture initializes after isolated RPC fixture"); + assert!(!bucket.is_empty()); + } + + #[tokio::test] + #[serial_test::serial] async fn lost_short_reservation_rpc_response_releases_the_bucket_fence() { use crate::storage::storage_api::contract::namespace::NamespaceLocking as _; @@ -8500,6 +8523,7 @@ mod tests { } #[tokio::test] + #[serial_test::serial] async fn short_metadata_reservation_expires_and_preserves_legacy_lifetime() { use crate::storage::storage_api::contract::namespace::NamespaceLocking as _; @@ -8600,6 +8624,7 @@ mod tests { } #[tokio::test] + #[serial_test::serial] async fn metadata_reservation_handlers_bind_and_publish_on_the_target_disk() { use crate::storage::storage_api::ecstore_disk::{DiskAPI, RUSTFS_META_TMP_BUCKET, ReadOptions};