From 77bd43fc22a51475062688aeb25ee4d20c87403f Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 26 Jul 2026 16:06:57 +0800 Subject: [PATCH] fix(ecstore): never cache fabricated bucket metadata as authoritative BucketMetadataSys::get_config lazily fabricated a default BucketMetadata (object-lock off) for any bucket whose .metadata.bin was ConfigNotFound and cached it in the map that the map-only, fail-closed metadata_sys::get() serves. The object-lock batch-delete gate (object_lock_delete_check_required, backlog#929 / #4297) treats that map as authoritative, so a metadata miss became a cached "no lock" answer: a versioning peek could poison the cache and let delete_objects skip the per-object retention/legal-hold stat. The same fabrication raced make_bucket (lost update overwriting freshly persisted lock-enabled metadata) and let the 15-minute refresh loop replace good cached metadata on a transient quorum dip. Production changes: - get_config caches only metadata actually read from disk; misses are recorded in a bounded negative cache (30s TTL, 10k entries, invalidated by set()) so repeated lookups for metadata-less names cost no extra namespace-lock + erasure-set fanout (reachable pre-auth via CORS preflight and per-key in DeleteObjects). - concurrent_load never lets a fabricated default REPLACE an existing map entry; startup insert-if-vacant behavior for legacy buckets is preserved. - delete_objects and new_ns_lock resolve dist-erasure, versioning, and the object-lock gate from the set's own instance context (backlog#1052) instead of the ambient facade, so a second in-process instance (or, in tests, another test's transient DistErasure window) cannot reroute locking onto an empty dist locker list or answer with the wrong instance's bucket state. Test-isolation changes (the bug that surfaced all of the above: the delete_objects lock-gating test failed deterministically when sharing a process with the lifecycle env tests): - The MinIO-migration test builds on an isolated InstanceContext instead of registering soon-deleted disks in the shared bootstrap registry. - The cached lifecycle env re-registers its disks on every use, surviving other serial tests' reset_local_disk_test_state. - Hermetic SetDisks helpers gain isolated-context variants pinned to plain erasure; tier-free non-serial test modules use them, guard-based SetupTypeGuard tests stay on the bootstrap context. - Three deterministic pin tests (nextest-safe) cover the caching contract, the delete gate resolution source, and the ns-lock resolution source. Verification: - cargo test -p rustfs-ecstore --lib -- --exact <4-test combo from the report> (previously failing, now green) - cargo test -p rustfs-ecstore --lib: 3169 passed / 0 failed across repeated runs; cargo fmt --check and cargo clippy --lib --tests clean - Adversarial validation (high-risk tier, all seven roles) run per AGENTS.md; all findings fixed or rebutted with evidence --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 19 ++ crates/ecstore/src/bucket/metadata.rs | 23 +- crates/ecstore/src/bucket/metadata_sys.rs | 197 ++++++++++++++++-- crates/ecstore/src/bucket/migration.rs | 18 +- crates/ecstore/src/bucket/quota/checker.rs | 30 ++- crates/ecstore/src/bucket/versioning_sys.rs | 17 ++ .../src/ecstore_validation_blackbox.rs | 16 +- crates/ecstore/src/set_disk/mod.rs | 52 ++++- crates/ecstore/src/set_disk/ops/locking.rs | 7 +- crates/ecstore/src/set_disk/ops/multipart.rs | 7 +- crates/ecstore/src/set_disk/ops/object.rs | 129 +++++++++++- crates/ecstore/src/set_disk/read.rs | 9 +- 12 files changed, 485 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index dba337333..d3ad0c0e0 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -8366,8 +8366,27 @@ mod tests { static STALE_MULTIPART_TEST_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); + /// Re-register the cached environment's disks into its (shared bootstrap) + /// context registry. Other `#[serial]` tests reset or reshape that + /// registry (`reset_local_disk_test_state`, their own `init_local_disks`), + /// and the peer-sys bucket operations of this env resolve local disks + /// through it at call time — without this repair, a lifecycle test that + /// runs after such a test fails bucket creation on write quorum. + async fn reregister_env_local_disks(ecstore: &Arc) { + use crate::disk::DiskAPI as _; + + let map = ecstore.ctx.local_disk_map(); + let mut guard = map.write().await; + for disks in ecstore.disk_map.values() { + for disk in disks.iter().flatten() { + guard.insert(disk.endpoint().to_string(), Some(disk.clone())); + } + } + } + async fn setup_test_env() -> (Vec, Arc) { if let Some((paths, ecstore)) = STALE_MULTIPART_TEST_ENV.get() { + reregister_env_local_disks(ecstore).await; return (paths.clone(), ecstore.clone()); } diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index b0ca13060..eddce55fc 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -649,7 +649,7 @@ impl BucketMetadata { Ok(()) } - fn default_timestamps(&mut self) { + pub(crate) fn default_timestamps(&mut self) { if self.policy_config_updated_at == OffsetDateTime::UNIX_EPOCH { self.policy_config_updated_at = self.created } @@ -1093,16 +1093,25 @@ pub async fn load_bucket_metadata(api: Arc, bucket: &str) -> Result, bucket: &str, parse: bool) -> Result { - let mut bm = match read_bucket_metadata(api.clone(), bucket).await { - Ok(res) => res, + Ok(load_bucket_metadata_parse_with_presence(api, bucket, parse).await?.0) +} + +/// The returned `bool` reports whether the metadata was actually read from +/// persisted storage; `false` means no metadata exists for this bucket on this +/// store and the returned value is a fabricated in-memory default. +pub(crate) async fn load_bucket_metadata_parse_with_presence( + api: Arc, + bucket: &str, + parse: bool, +) -> Result<(BucketMetadata, bool)> { + let (mut bm, persisted) = match read_bucket_metadata(api.clone(), bucket).await { + Ok(res) => (res, true), Err(err) => { if err != Error::ConfigNotFound { return Err(err); } - // info!("bucketmeta {} not found with err {:?}, start to init ", bucket, &err); - - BucketMetadata::new(bucket) + (BucketMetadata::new(bucket), false) } }; @@ -1112,7 +1121,7 @@ pub async fn load_bucket_metadata_parse(api: Arc, bucket: &str, parse: bm.parse_all_configs()?; } - Ok(bm) + Ok((bm, persisted)) } async fn read_bucket_metadata(api: Arc, bucket: &str) -> Result { diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index b4610d415..f937e0eb2 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -16,7 +16,7 @@ use super::metadata::{BUCKET_TARGETS_FILE, BucketMetadata, load_bucket_metadata} use super::quota::BucketQuota; use super::target::BucketTargets; use crate::bucket::bucket_target_sys::BucketTargetSys; -use crate::bucket::metadata::load_bucket_metadata_parse; +use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_parse_with_presence}; use crate::bucket::utils::is_meta_bucketname; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result, is_err_bucket_not_found}; @@ -190,7 +190,7 @@ pub async fn get(bucket: &str) -> Result> { // instance cell is not initialized yet (early startup) they fall back to the // ambient default — the single-instance legacy behavior. -fn bucket_metadata_sys_of(ctx: &crate::runtime::instance::InstanceContext) -> Result>> { +pub(crate) fn bucket_metadata_sys_of(ctx: &crate::runtime::instance::InstanceContext) -> Result>> { if let Some(sys) = ctx.bucket_metadata_sys() { return Ok(sys); } @@ -422,9 +422,23 @@ pub async fn list_bucket_targets(bucket: &str) -> Result { bucket_meta_sys.get_bucket_targets_config(bucket).await } +/// Bound and lifetime of the negative cache for buckets with no persisted +/// metadata. Entries are invalidated the moment real metadata is cached, so +/// the TTL only bounds staleness for out-of-band creations whose reload +/// notification was lost; the capacity bounds memory under bogus-name floods. +const ABSENT_BUCKET_METADATA_TTL: Duration = Duration::from_secs(30); +const ABSENT_BUCKET_METADATA_MAX_ENTRIES: u64 = 10_000; + #[derive(Debug)] pub struct BucketMetadataSys { metadata_map: RwLock>>, + /// Buckets recently observed to have no persisted metadata. Serving the + /// fabricated default from here (instead of re-reading disk) keeps the + /// per-request cost of repeated lookups for such names bounded — without + /// this, every request naming a nonexistent bucket pays a namespace-lock + /// acquisition plus a full erasure-set metadata fanout (reachable + /// pre-auth via CORS preflight, and per-key in DeleteObjects). + absent_metadata: moka::future::Cache, api: Arc, initialized: RwLock, } @@ -433,6 +447,10 @@ impl BucketMetadataSys { pub fn new(api: Arc) -> Self { Self { metadata_map: RwLock::new(HashMap::new()), + absent_metadata: moka::future::Cache::builder() + .max_capacity(ABSENT_BUCKET_METADATA_MAX_ENTRIES) + .time_to_live(ABSENT_BUCKET_METADATA_TTL) + .build(), api, initialized: RwLock::new(false), } @@ -487,7 +505,7 @@ impl BucketMetadataSys { }, ) .await; - load_bucket_metadata(self.api.clone(), bucket.as_str()).await + load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await }); } @@ -495,9 +513,24 @@ impl BucketMetadataSys { for (idx, res) in results.into_iter().enumerate() { match res { - Ok(res) => { + Ok((bm, persisted)) => { if let Some(bucket) = buckets.get(idx) { - self.set(bucket.clone(), Arc::new(res)).await; + if persisted { + self.set(bucket.clone(), Arc::new(bm)).await; + } else { + // A fabricated default (no persisted metadata + // readable right now) must never REPLACE an + // existing entry: the periodic refresh would + // otherwise downgrade a lock-enabled bucket to an + // authoritative "no lock" default on a transient + // ConfigNotFound, disabling the object-lock + // delete gate and wiping its target/durability + // sync state. Insert-if-vacant keeps the startup + // behavior for legacy buckets without a metadata + // file, atomically under the map write lock. + let mut map = self.metadata_map.write().await; + map.entry(bucket.clone()).or_insert_with(|| Arc::new(bm)); + } } } Err(e) => { @@ -528,6 +561,8 @@ impl BucketMetadataSys { let mut map = self.metadata_map.write().await; map.insert(bucket.clone(), bm.clone()); drop(map); + // Real metadata supersedes any recorded absence immediately. + self.absent_metadata.invalidate(&bucket).await; sync_bucket_target_sys(&bucket, &bm).await; sync_bucket_durability(&bucket, &bm); } @@ -630,13 +665,22 @@ impl BucketMetadataSys { pub async fn get_config(&self, bucket: &str) -> Result<(Arc, bool)> { let has_bm = { let map = self.metadata_map.read().await; - map.get(&bucket.to_string()).cloned() + map.get(bucket).cloned() }; if let Some(bm) = has_bm { Ok((bm, false)) } else { - let bm = match load_bucket_metadata(self.api.clone(), bucket).await { + // A recent lookup already established there is no persisted + // metadata: serve the fabricated default without another + // namespace-lock + erasure-set fanout. + if self.absent_metadata.get(bucket).await.is_some() { + let mut bm = BucketMetadata::new(bucket); + bm.default_timestamps(); + return Ok((Arc::new(bm), true)); + } + + let (bm, persisted) = match load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true).await { Ok(res) => res, Err(err) => { return if *self.initialized.read().await { @@ -647,13 +691,27 @@ impl BucketMetadataSys { } }; - let mut map = self.metadata_map.write().await; - let bm = Arc::new(bm); - map.insert(bucket.to_string(), bm.clone()); - drop(map); - sync_bucket_target_sys(bucket, &bm).await; - sync_bucket_durability(bucket, &bm); + + // This lazy path caches only metadata that actually exists on + // this store. A fabricated default must not enter the map: + // `get()` is map-only and fail-closed — the object-lock delete + // gate (`object_lock_delete_check_required`) skips its per-object + // protection stat exactly when the map serves metadata saying the + // bucket has no Object Lock, so caching a fabricated default here + // would turn a metadata miss into an authoritative "no lock" + // answer. (Startup `concurrent_load` still caches fabricated + // defaults for buckets listed on disk — legacy buckets without a + // metadata file — but never lets one replace an existing entry.) + if persisted { + let mut map = self.metadata_map.write().await; + map.insert(bucket.to_string(), bm.clone()); + drop(map); + sync_bucket_target_sys(bucket, &bm).await; + sync_bucket_durability(bucket, &bm); + } else { + self.absent_metadata.insert(bucket.to_string(), ()).await; + } Ok((bm, true)) } @@ -873,13 +931,126 @@ impl BucketMetadataSys { } } +/// Test-only fixture shared with sibling modules (e.g. the quota checker +/// tests): a 4-disk `ECStore` on an isolated instance context, so tests +/// exercising the metadata system never touch ambient process state. +#[cfg(test)] +pub(crate) mod test_support { + use super::*; + use crate::disk::endpoint::Endpoint; + use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; + use crate::runtime::instance::InstanceContext; + use crate::store::init_local_disks_with_instance_ctx; + + pub(crate) async fn isolated_store_over_temp_disks() -> (Vec, Arc) { + let mut dirs = Vec::with_capacity(4); + let mut endpoints = Vec::with_capacity(4); + for disk_idx in 0..4 { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let mut endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + dirs.push(dir); + endpoints.push(endpoint); + } + let endpoint_pools = EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "metadata-sys-cache-test".to_string(), + platform: "test".to_string(), + }]); + let instance_ctx = Arc::new(InstanceContext::new()); + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) + .await + .expect("local disks should initialize"); + let ecstore = ECStore::new_with_instance_ctx( + "127.0.0.1:0".parse().expect("test address"), + endpoint_pools, + CancellationToken::new(), + instance_ctx, + ) + .await + .expect("ECStore should initialize"); + (dirs, ecstore) + } +} + #[cfg(test)] mod tests { + use super::test_support::isolated_store_over_temp_disks; use super::*; use crate::bucket::target::{BucketTarget, BucketTargetType, Credentials}; use serial_test::serial; use tokio::time::timeout; + /// Pins the fail-closed caching contract of the lazy `get_config` path + /// and the refresh no-replace rule: fabricated defaults are returned but + /// never served by the map-only `get()`, persisted metadata is cached on + /// lazy load (superseding a recorded absence), and a refresh-load miss + /// never replaces an existing entry. + #[tokio::test] + async fn get_config_never_caches_fabricated_defaults_as_authoritative() { + let (_dirs, ecstore) = isolated_store_over_temp_disks().await; + let sys = BucketMetadataSys::new(ecstore); + + // (a) Miss: the fabricated default is returned but not cached. + let (bm, _) = sys + .get_config("absent-bucket") + .await + .expect("fabricated default should be returned"); + assert!(bm.object_lock_config_xml.is_empty()); + assert!( + sys.get("absent-bucket").await.is_err(), + "a fabricated default must never be served by the map-only get()" + ); + + // The repeat lookup is served from the negative cache, same answer. + let (bm, _) = sys + .get_config("absent-bucket") + .await + .expect("negative-cached default should be returned"); + assert!(bm.object_lock_config_xml.is_empty()); + assert!(sys.get("absent-bucket").await.is_err()); + + // (b) Persisting real metadata supersedes the recorded absence, and a + // lazy reload after a map wipe re-caches it. + let mut persisted = BucketMetadata::new("absent-bucket"); + persisted.policy_config_json = b"persisted-marker".to_vec(); + sys.persist_and_set(persisted).await.expect("metadata should persist"); + sys.metadata_map.write().await.clear(); + let _ = sys + .get_config("absent-bucket") + .await + .expect("persisted metadata should lazily reload"); + let cached = sys + .get("absent-bucket") + .await + .expect("lazily loaded persisted metadata must be cached"); + assert_eq!(cached.policy_config_json, b"persisted-marker".to_vec()); + + // (c) A refresh-load miss (no persisted metadata readable) must not + // replace an existing entry. + let mut kept = BucketMetadata::new("kept-bucket"); + kept.policy_config_json = b"kept-marker".to_vec(); + sys.set("kept-bucket".to_string(), Arc::new(kept)).await; + let mut failed = HashSet::new(); + let refresh_targets = vec!["kept-bucket".to_string()]; + sys.concurrent_load(&refresh_targets, &mut failed).await; + let kept = sys + .get("kept-bucket") + .await + .expect("existing entry must survive a refresh miss"); + assert_eq!( + kept.policy_config_json, + b"kept-marker".to_vec(), + "a fabricated refresh default must not replace real metadata" + ); + } + fn target(bucket: &str, id: &str) -> BucketTarget { BucketTarget { source_bucket: bucket.to_string(), diff --git a/crates/ecstore/src/bucket/migration.rs b/crates/ecstore/src/bucket/migration.rs index ead8dee73..ae775a014 100644 --- a/crates/ecstore/src/bucket/migration.rs +++ b/crates/ecstore/src/bucket/migration.rs @@ -538,10 +538,12 @@ mod tests { use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::object_api::{ObjectOptions, PutObjReader}; + use crate::runtime::instance::InstanceContext; use crate::storage_api_contracts::bucket::{BucketOperations, BucketOptions, MakeBucketOptions}; use crate::storage_api_contracts::object::{ObjectIO, ObjectOperations}; - use crate::store::{ECStore, init_local_disks}; + use crate::store::{ECStore, init_local_disks_with_instance_ctx}; use rustfs_utils::path::SLASH_SEPARATOR; + use std::sync::Arc; use tokio::fs; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -570,10 +572,20 @@ mod tests { cmd_line: "minio-migrate-test".to_string(), platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), }]); - init_local_disks(endpoint_pools.clone()).await.unwrap(); - let ecstore = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, CancellationToken::new()) + // Isolated instance context: this test deletes its disks at the end, + // and dead entries in the shared registry break other cached envs. + let instance_ctx = Arc::new(InstanceContext::new()); + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) .await .unwrap(); + let ecstore = ECStore::new_with_instance_ctx( + "127.0.0.1:0".parse().unwrap(), + endpoint_pools, + CancellationToken::new(), + instance_ctx, + ) + .await + .unwrap(); let existing: Vec = ecstore .list_bucket(&BucketOptions { no_metadata: true, diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index 63b2e9587..0ba137d54 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -118,11 +118,17 @@ impl QuotaChecker { } pub async fn get_quota_config(&self, bucket: &str) -> Result { - let meta = self + // `get_config`, not the map-only `get()`: a bucket with no persisted + // metadata must resolve to the fabricated default (no quota + // configured) so the admission check passes and the request reaches + // the NoSuchBucket answer — a map-only miss would fail every such + // PUT closed with 503 before the 404 could be produced. Real read + // faults still surface as errors and keep the fail-closed behavior. + let (meta, _) = self .metadata_sys .read() .await - .get(bucket) + .get_config(bucket) .await .map_err(QuotaError::StorageError)?; @@ -178,6 +184,26 @@ impl QuotaChecker { #[cfg(test)] mod tests { use super::*; + use crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks; + + /// Regression (PR #5307 / s3-tests `test_100_continue_error_retry`): a + /// bucket with no persisted metadata has no quota, so the admission check + /// must pass and let the request reach its NoSuchBucket answer. With the + /// map-only `get()` this failed closed as a retryable 503 on every PUT to + /// a nonexistent bucket. + #[tokio::test] + async fn quota_check_allows_bucket_without_persisted_metadata() { + let (_dirs, ecstore) = isolated_store_over_temp_disks().await; + let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore))); + let checker = QuotaChecker::new(sys); + + let result = checker + .check_quota("no-such-bucket", QuotaOperation::PutObject, 1024) + .await + .expect("a bucket with no persisted metadata has no quota and must not fail the check"); + assert!(result.allowed); + assert_eq!(result.quota_limit, None); + } #[tokio::test] async fn test_quota_check_no_limit() { diff --git a/crates/ecstore/src/bucket/versioning_sys.rs b/crates/ecstore/src/bucket/versioning_sys.rs index b418152c1..685e95af9 100644 --- a/crates/ecstore/src/bucket/versioning_sys.rs +++ b/crates/ecstore/src/bucket/versioning_sys.rs @@ -85,4 +85,21 @@ impl BucketVersioningSys { Ok(cfg) } + + /// Instance-scoped variant of [`Self::get`] (backlog#1052): resolves the + /// caller's own instance context so a second in-process store never + /// answers with the first instance's versioning state; falls back to the + /// ambient system when the instance cell is not initialized. + pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result { + if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) { + return Ok(VersioningConfiguration::default()); + } + + let bucket_meta_sys_lock = crate::bucket::metadata_sys::bucket_metadata_sys_of(ctx)?; + let bucket_meta_sys = bucket_meta_sys_lock.read().await; + + let (cfg, _) = bucket_meta_sys.get_versioning_config(bucket).await?; + + Ok(cfg) + } } diff --git a/crates/ecstore/src/ecstore_validation_blackbox.rs b/crates/ecstore/src/ecstore_validation_blackbox.rs index 4a566431e..b36fdde42 100644 --- a/crates/ecstore/src/ecstore_validation_blackbox.rs +++ b/crates/ecstore/src/ecstore_validation_blackbox.rs @@ -21,6 +21,19 @@ use tokio::sync::RwLock; /// Returns the backing [`tempfile::TempDir`]s alongside the set so callers keep /// them alive for the test's duration and the directories are removed on drop. pub(crate) async fn make_local_set_disks(drive_count: usize, parity_count: usize) -> (Vec, Arc) { + make_local_set_disks_with_ctx(drive_count, parity_count, crate::runtime::instance::bootstrap_ctx()).await +} + +/// Like [`make_local_set_disks`], but binds the set to an explicit instance +/// context. Tests whose assertions depend on ambient runtime state that other +/// tests mutate (e.g. `SetupTypeGuard` flipping the shared setup type to +/// DistErasure from `#[serial]` tests in a different serial group) pass an +/// isolated context here so that state cannot leak into their code path. +pub(crate) async fn make_local_set_disks_with_ctx( + drive_count: usize, + parity_count: usize, + ctx: Arc, +) -> (Vec, Arc) { let format = FormatV3::new(1, drive_count); let mut dirs = Vec::with_capacity(drive_count); let mut endpoints = Vec::with_capacity(drive_count); @@ -55,7 +68,7 @@ pub(crate) async fn make_local_set_disks(drive_count: usize, parity_count: usize disks.push(Some(disk)); } - let set_disks = SetDisks::new( + let set_disks = SetDisks::new_with_instance_ctx( "ecstore-validation-blackbox".to_string(), Arc::new(RwLock::new(disks)), drive_count, @@ -65,6 +78,7 @@ pub(crate) async fn make_local_set_disks(drive_count: usize, parity_count: usize endpoints, format, Vec::new(), + ctx, ) .await; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4c7308a81..d0f8ccc85 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4596,12 +4596,19 @@ mod tests { } async fn make_test_set_disks(lockers: Vec>) -> Arc { + make_test_set_disks_with_ctx(lockers, bootstrap_ctx()).await + } + + async fn make_test_set_disks_with_ctx( + lockers: Vec>, + instance_ctx: Arc, + ) -> Arc { let endpoints = vec![ Endpoint::try_from("http://127.0.0.1:9000/data").expect("first endpoint should parse"), Endpoint::try_from("http://127.0.0.1:9001/data").expect("second endpoint should parse"), ]; - SetDisks::new( + SetDisks::new_with_instance_ctx( "test-owner".to_string(), Arc::new(RwLock::new(vec![None, None])), 2, @@ -4611,10 +4618,53 @@ mod tests { endpoints, FormatV3::new(1, 2), lockers, + instance_ctx, ) .await } + /// Pins the dist-erasure resolution SOURCE for `new_ns_lock` (adversarial + /// review): the lock strategy must come from the set's own instance + /// context, never the ambient facade — otherwise another in-process + /// instance (or a concurrent test's ambient DistErasure window) reroutes + /// this set's locking onto the wrong strategy. + #[tokio::test(flavor = "multi_thread")] + async fn new_ns_lock_resolves_dist_from_set_instance_context() { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let locker: Arc = Arc::new(LocalClient::with_manager(manager)); + + let dist_ctx = Arc::new(InstanceContext::new()); + dist_ctx.update_erasure_type(SetupType::DistErasure).await; + let dist_set = make_test_set_disks_with_ctx(vec![locker.clone()], dist_ctx).await; + let dist_guard = dist_set + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created") + .get_read_lock(Duration::from_millis(500)) + .await + .expect("dist read lock should succeed with one healthy locker"); + assert!( + matches!(dist_guard, NamespaceLockGuard::Standard(_)), + "a DistErasure instance context must select the distributed lock strategy" + ); + drop(dist_guard); + + let local_ctx = Arc::new(InstanceContext::new()); + local_ctx.update_erasure_type(SetupType::Erasure).await; + let local_set = make_test_set_disks_with_ctx(vec![locker], local_ctx).await; + let local_guard = local_set + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created") + .get_read_lock(Duration::from_millis(500)) + .await + .expect("local read lock should succeed"); + assert!( + matches!(local_guard, NamespaceLockGuard::Fast(_)), + "a plain-erasure instance context must select the local lock strategy" + ); + } + struct SetupTypeGuard { previous: SetupType, } diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index a1baf24d0..a9787ba13 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -30,7 +30,12 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { #[tracing::instrument(skip(self))] async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { - let set_lock = if runtime_sources::setup_is_dist_erasure().await { + // Resolved from this set's own instance context (backlog#1052), not the + // ambient facade: the facade tracks whichever context is currently + // published process-wide, so a second instance (or, in tests, another + // test's transient DistErasure window) would push this set's namespace + // locking onto its own — possibly empty — dist locker list. + let set_lock = if self.ctx.is_dist_erasure().await { // Calculate quorum based on lockers count (majority) let lockers_count = self.lockers.len(); let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 }; diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 37d4d6209..a3228e079 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1898,8 +1898,13 @@ mod tests { use crate::disk::DiskAPI as _; use crate::disk::{endpoint::Endpoint, format::FormatV3}; use crate::layout::endpoints::SetupType; + // No-locker helpers resolve to the isolated-context variants (see + // `hermetic_set_disks_isolated`); the guard-based tests build through + // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context + // so their own ambient SetupTypeGuard flip remains visible to them. use crate::set_disk::ops::object::hermetic_set_disks_support::{ - hermetic_set_disks, hermetic_set_disks_for_pool_with_default_parity, hermetic_set_disks_with_lockers, + hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity, + hermetic_set_disks_isolated as hermetic_set_disks, hermetic_set_disks_with_lockers, }; use crate::storage_api_contracts::namespace::NamespaceLocking as _; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 3b15a94a1..757af70b4 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -2671,7 +2671,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let mut _local_batch_guards: Vec = Vec::with_capacity(batch.requests.len()); let mut locked_objects = HashSet::new(); - let dist_erasure = runtime_sources::setup_is_dist_erasure().await; + // Instance-scoped, not the ambient facade (backlog#1052) — see + // new_ns_lock. The same applies to the versioning and bucket-metadata + // lookups below: resolving them through the first published + // instance's context would let a second in-process store delete with + // the wrong versioning semantics or skip the object-lock gate. + let dist_erasure = self.ctx.is_dist_erasure().await; let mut dist_batch_lock_ids = vec![Vec::new(); self.lockers.len()]; if opts.no_lock { @@ -2716,7 +2721,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } } - let ver_cfg = BucketVersioningSys::get(bucket).await.unwrap_or_default(); + let ver_cfg = BucketVersioningSys::get_in(&self.ctx, bucket).await.unwrap_or_default(); // backlog#929 (HP-8): the per-object stat below exists solely to feed // check_object_lock_delete (#4297). Resolve the bucket lock @@ -2724,7 +2729,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // for buckets without Object Lock; unknown metadata fails closed and // keeps the stat, so the #4297 protection is preserved verbatim for // every object-lock-enabled bucket. - let object_lock_checks_required = object_lock_delete_check_required(metadata_sys::get(bucket).await.ok().as_deref()); + let object_lock_checks_required = + object_lock_delete_check_required(metadata_sys::get_in(&self.ctx, bucket).await.ok().as_deref()); let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new(); @@ -4098,11 +4104,57 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support { hermetic_set_disks_with_lockers(disk_count, pool_index, default_parity_count, Vec::new()).await } + /// Like [`hermetic_set_disks`], but binds the set to an isolated instance + /// context pinned to plain erasure. `#[serial]` tests elsewhere flip the + /// shared bootstrap context to DistErasure (`SetupTypeGuard`) while + /// non-serial hermetic tests run; on the shared context such a window + /// reroutes locking onto the empty dist locker list ("No lock clients + /// available") and the batch-delete gate onto the dist path. Only suitable + /// for tests that never touch context-resolved services registered on the + /// ambient context (tier config manager, expiry state, ...), because the + /// isolated context starts every one of those cells fresh. + pub(in crate::set_disk::ops) async fn hermetic_set_disks_isolated( + disk_count: usize, + ) -> (Vec, Vec, Arc) { + hermetic_set_disks_for_pool_with_default_parity_isolated(disk_count, 0, disk_count / 2).await + } + + /// Pool-parameterized variant of [`hermetic_set_disks_isolated`] with the + /// same isolation contract. + pub(in crate::set_disk::ops) async fn hermetic_set_disks_for_pool_with_default_parity_isolated( + disk_count: usize, + pool_index: usize, + default_parity_count: usize, + ) -> (Vec, Vec, Arc) { + let isolated_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + isolated_ctx + .update_erasure_type(crate::layout::endpoints::SetupType::Erasure) + .await; + hermetic_set_disks_with_lockers_and_ctx(disk_count, pool_index, default_parity_count, Vec::new(), isolated_ctx).await + } + pub(in crate::set_disk::ops) async fn hermetic_set_disks_with_lockers( disk_count: usize, pool_index: usize, default_parity_count: usize, lockers: Vec>, + ) -> (Vec, Vec, Arc) { + hermetic_set_disks_with_lockers_and_ctx( + disk_count, + pool_index, + default_parity_count, + lockers, + crate::runtime::instance::bootstrap_ctx(), + ) + .await + } + + pub(in crate::set_disk::ops) async fn hermetic_set_disks_with_lockers_and_ctx( + disk_count: usize, + pool_index: usize, + default_parity_count: usize, + lockers: Vec>, + instance_ctx: Arc, ) -> (Vec, Vec, Arc) { let format = FormatV3::new(1, disk_count); @@ -4119,7 +4171,7 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support { disks.push(Some(disk)); } - let set_disks = SetDisks::new( + let set_disks = SetDisks::new_with_instance_ctx( "hermetic-ops-test-owner".to_string(), Arc::new(RwLock::new(disks)), disk_count, @@ -4129,6 +4181,7 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support { endpoints, format, lockers, + instance_ctx, ) .await; @@ -4237,7 +4290,7 @@ mod get_object_downstream_close_accounting_tests { #[cfg(test)] mod metadata_mutation_generation_tests { - use super::hermetic_set_disks_support::hermetic_set_disks; + use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; @@ -6587,7 +6640,7 @@ mod transition_source_identity_matrix_tests { #[cfg(test)] mod heterogeneous_pool_put_tests { - use super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity; + use super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity; use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::{DiskAPI as _, ReadOptions}; @@ -6647,7 +6700,7 @@ mod put_object_tmp_cleanup_tests { //! response path), while a failed PUT must still clean its tmp shards //! inline before returning. - use super::hermetic_set_disks_support::hermetic_set_disks; + use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; use std::time::Duration; @@ -6749,7 +6802,7 @@ mod put_object_tags_early_stop_regression_tests { //! with early-stop enabled, the tag must land on EVERY online disk's xl.meta, //! not a read-quorum subset. - use super::hermetic_set_disks_support::hermetic_set_disks; + use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; use crate::disk::{DiskAPI as _, ReadOptions}; @@ -6811,7 +6864,8 @@ mod delete_objects_lock_gating_tests { //! locked-stat path and prove the #4297 delete protection is intact end to //! end, while per-key result mapping of mixed batches stays stable. - use super::hermetic_set_disks_support::hermetic_set_disks; + use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; + use super::hermetic_set_disks_support::hermetic_set_disks_with_lockers_and_ctx; use super::*; use crate::disk::DiskAPI as _; @@ -6943,6 +6997,63 @@ mod delete_objects_lock_gating_tests { .expect_err("plain object must be deleted"); } + /// Pins the dist-erasure resolution SOURCE for the batch-delete lock gate + /// (adversarial review): the decision must come from the set's own + /// instance context. With a DistErasure context and no dist lockers the + /// batch must fail closed on lock acquisition; ambient resolution + /// (non-dist in this process) would take local locks and let the delete + /// through. + #[tokio::test] + async fn delete_objects_dist_gate_uses_set_instance_context() { + let dist_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + dist_ctx + .update_erasure_type(crate::layout::endpoints::SetupType::DistErasure) + .await; + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, Vec::new(), dist_ctx).await; + let bucket = "dist-gate-ctx-bucket"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + // The put must bypass locking: with a DistErasure context and no + // lockers, every namespace lock acquisition fails closed by design. + let mut reader = PutObjReader::from_vec(vec![5u8; 256]); + set_disks + .put_object( + bucket, + "obj", + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("object should be written without locks"); + + let objects = vec![ObjectToDelete { + object_name: "obj".to_string(), + ..Default::default() + }]; + let (_deleted, errs) = set_disks.delete_objects(bucket, objects, ObjectOptions::default()).await; + + assert!( + errs[0].is_some(), + "the dist gate resolved from the set's own context must fail closed on an empty locker list" + ); + set_disks + .get_object_info( + bucket, + "obj", + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("object must survive the failed batch delete"); + } + #[tokio::test] async fn delete_objects_honors_no_lock_when_outer_write_lock_is_held() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index c55d4ed60..034f4b674 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2611,7 +2611,14 @@ mod metadata_cache_tests { #[tokio::test] #[serial(metadata_cache_publish_barrier)] async fn metadata_cache_production_fanout_cannot_publish_after_invalidation() { - let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(4, 2).await; + // Isolated context: an ambient DistErasure window (another test's + // SetupTypeGuard) would bypass metadata-cache publication entirely + // and time out the barrier below. + let isolated_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + isolated_ctx + .update_erasure_type(crate::layout::endpoints::SetupType::Erasure) + .await; + let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(4, 2, isolated_ctx).await; let bucket = "metadata-cache-production-fence"; let object = "object"; let disks = set.disks.read().await.clone();