fix(heal): address diagnostic lint and test isolation

This commit is contained in:
Chris An
2026-11-01 23:17:18 +08:00
parent 82556f0a22
commit 51abbf321b
4 changed files with 62 additions and 20 deletions
+3 -3
View File
@@ -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<HealListItem>, Option<String>, bool);
struct DeferredObject {
name: String,
version_id: Option<String>,
@@ -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<Box<dyn std::future::Future<Output = Result<(Vec<HealListItem>, Option<String>, bool)>> + Send>>,
> = None;
let mut listing_inflight: Option<futures::future::BoxFuture<'static, Result<HealListingPage>>> = None;
let scope_result: Result<()> = async {
loop {
self.check_control_flags().await?;
+2 -1
View File
@@ -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)]
+26 -10
View File
@@ -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<PathBuf>,
/// 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<ECStore>,
/// The single-pool, single-set topology the store was built from.
///
@@ -155,6 +155,7 @@ pub struct TestECStoreEnvBuilder {
prefix: String,
base_dir: Option<PathBuf>,
init_bucket_metadata: bool,
instance_context: Option<Arc<InstanceContext>>,
}
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<InstanceContext>) -> 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
+31 -6
View File
@@ -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};