From 54c11ef28b6b97828d6e5645b41084378ab51cdb Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 6 Sep 2026 10:51:30 +0800 Subject: [PATCH] test(scanner): bound segment observation diagnostics (#7240) * test(scanner): bound segment observation diagnostics Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(scanner): observe committed fixture changes during walks Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(scanner): validate segment fixture metadata and off state Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 Co-authored-by: overtrue --- crates/ecstore/src/set_disk/ops/object.rs | 59 +++++ .../tests/checkpoint_fixture.rs | 2 + .../checkpoint_fixture/segment_observation.rs | 202 ++++++++++++++++++ docs/testing/scanner-checkpoint-fixture.md | 16 ++ 4 files changed, 279 insertions(+) create mode 100644 crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c57f0b36b..4fcb450bd 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -12672,6 +12672,65 @@ mod metadata_mutation_generation_tests { set_disks.invalidate_get_object_metadata_cache(bucket, object).await; } + #[tokio::test] + #[serial_test::serial(metadata_cache_invalidation_probe)] + async fn segment_observation_equal_size_mutations_retire_metadata_generation() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "segment-observation-bucket"; + let object = "hot/object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("create segment fixture bucket"); + } + let (before, old_key) = put_and_prime(&set_disks, bucket, object, b"before").await; + let probe = MetadataCacheInvalidationProbe::install(bucket, object); + let mut replacement = PutObjReader::from_vec(b"after!".to_vec()); + set_disks + .put_object(bucket, object, &mut replacement, &ObjectOptions::default()) + .await + .expect("commit same-length replacement with normal owner locking"); + assert_eq!(probe.count(), 2, "same-length PUT must retire its metadata generation"); + assert_retired(&set_disks, &old_key).await; + drop(probe); + let after = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("read replacement metadata"); + assert_eq!(before.size, after.size); + assert_ne!(before.etag, after.etag, "equal size is not equal content"); + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("read replacement body through the owner"); + let mut body = Vec::new(); + reader.stream.read_to_end(&mut body).await.expect("drain replacement body"); + assert_eq!(body, b"after!"); + drop(reader); + + let (before, old_key) = put_and_prime(&set_disks, bucket, object, b"after!").await; + let probe = MetadataCacheInvalidationProbe::install(bucket, object); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(HashMap::from([("x-amz-meta-segment".to_string(), "changed".to_string())])), + ..Default::default() + }, + ) + .await + .expect("commit metadata-only mutation with normal owner locking"); + assert_eq!(probe.count(), 4, "metadata-only mutation must retire both owner fences"); + assert_retired(&set_disks, &old_key).await; + let after = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("read committed metadata-only mutation"); + assert_eq!(before.size, after.size); + assert_eq!(before.etag, after.etag); + assert!(!before.user_defined.contains_key("x-amz-meta-segment")); + assert_eq!(after.user_defined.get("x-amz-meta-segment").map(String::as_str), Some("changed")); + } + #[tokio::test] #[serial_test::serial(metadata_cache_invalidation_probe)] async fn metadata_semantic_mutation_generation_matrix_retires_cached_snapshot() { diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs index eb10f02e6..7052e9de4 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs @@ -20,6 +20,8 @@ use crate::{DataUsageCacheSource, DataUsageScanPlanDigest}; use std::io::Cursor; use tokio::io::AsyncReadExt; +mod segment_observation; + const CACHE_NAME: &str = "bucket/checkpoint-fixture.bin"; const STATIC_OBJECTS: u64 = 24; const MAX_CACHE_BYTES: u64 = 1024 * 1024; diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs new file mode 100644 index 000000000..b54f36665 --- /dev/null +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs @@ -0,0 +1,202 @@ +//! Fixture-only range diagnostics. No result is supplied to a scan selector. + +use super::*; +use std::collections::BTreeSet; + +const MAX_SEGMENTS: usize = 4; +const MAX_SEGMENT_BYTES: usize = 128; +const MAX_WALK_SAMPLES: usize = 32; +const MAX_WALK_BYTES: usize = 1024; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ProposalError { + EntryLimit, + ByteLimit, + InvalidKey, +} + +// Keys come from successful fixture writes, not a production mutation stream. +fn fixture_proposal(keys: &[&str]) -> Result, ProposalError> { + let mut segments = BTreeSet::new(); + let mut bytes = 0; + for key in keys { + if key.is_empty() || key.contains(['\\', '\0']) || key.split('/').any(|part| matches!(part, "" | "." | "..")) { + return Err(ProposalError::InvalidKey); + } + let segment = key.split('/').next().expect("validated nonempty key"); + if segments.contains(segment) { + continue; + } + if segments.len() == MAX_SEGMENTS { + return Err(ProposalError::EntryLimit); + } + if segment.len() > MAX_SEGMENT_BYTES - bytes { + return Err(ProposalError::ByteLimit); + } + bytes += segment.len(); + segments.insert(segment.to_string()); + } + Ok(segments) +} + +#[test] +fn segment_observation_fixture_proposal_bounds() { + assert_eq!(fixture_proposal(&["hot/one", "hot/two"]), Ok(BTreeSet::from(["hot".to_string()]))); + assert_eq!(fixture_proposal(&["a", "b", "c", "d"]).expect("entry boundary").len(), MAX_SEGMENTS); + assert_eq!(fixture_proposal(&["a", "b", "c", "d", "e"]), Err(ProposalError::EntryLimit)); + let exact = "x".repeat(MAX_SEGMENT_BYTES); + assert!(fixture_proposal(&[&exact]).is_ok()); + assert_eq!(fixture_proposal(&[&exact, "y"]), Err(ProposalError::ByteLimit)); + let oversized = "x".repeat(MAX_SEGMENT_BYTES + 1); + assert_eq!(fixture_proposal(&[&oversized]), Err(ProposalError::ByteLimit)); + for key in ["", "/hot", "hot/../cold", "hot//one", "hot\\one", "hot/\0"] { + assert_eq!(fixture_proposal(&[key]), Err(ProposalError::InvalidKey)); + } +} + +fn cache_value(cache: &DataUsageCache) -> serde_json::Value { + let mut value = serde_json::to_value(cache).expect("serialize the entire cache"); + // Children are a HashSet: canonicalize only that unordered field, without + // discarding any cache fields or changing ordered histogram arrays. + for (path, entry) in &cache.cache { + value["cache"][path]["children"] = + serde_json::to_value(entry.children.iter().collect::>()).expect("canonical child set"); + } + value +} + +async fn walk_and_save(observe: bool) -> (Vec, serde_json::Value) { + let (mut scanner, root) = build_test_scanner().await; + let _guard = TestGuard { + temp_dir: Some(root.clone()), + }; + for prefix in ["hot", "cold", "other"] { + for leaf in ["one", "two"] { + let object = format!("{prefix}/{leaf}"); + let mut metadata = FileMeta::new(); + let mut info = FileInfo::new(&object, 4, 2); + info.volume = "bucket".to_string(); + info.name = object.clone(); + info.size = 1; + info.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid fixture timestamp")); + info.metadata.insert("etag".to_string(), "before".to_string()); + metadata.add_version(info).expect("construct segment fixture metadata"); + write_test_object_metadata_bytes(&root, "bucket", &object, &metadata.marshal_msg().expect("encode metadata")).await; + } + } + let changed_key = "hot/one"; + let changed_path = root.join("bucket").join(changed_key).join("xl.meta"); + let before = tokio::fs::read(&changed_path).await.expect("read initial hot metadata"); + let mut metadata = FileMeta::new(); + let mut info = FileInfo::new(changed_key, 4, 2); + info.volume = "bucket".to_string(); + info.name = changed_key.to_string(); + info.size = 1; + info.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid fixture timestamp")); + info.metadata.insert("etag".to_string(), "after!".to_string()); + metadata.add_version(info).expect("construct same-size hot mutation"); + write_test_object_metadata_bytes(&root, "bucket", changed_key, &metadata.marshal_msg().expect("encode hot mutation")).await; + let after = tokio::fs::read(&changed_path) + .await + .expect("read back committed fixture mutation"); + assert_eq!(before.len(), after.len(), "fixture rewrite must keep metadata byte length unchanged"); + assert_ne!(before, after, "a changed key requires an observable successful fixture write"); + scanner.old_cache.info.name = "bucket".to_string(); + scanner.new_cache.info.name = "bucket".to_string(); + scanner.update_cache.info.name = "bucket".to_string(); + let paths = Arc::new(Mutex::new(Vec::::new())); + let proposed_walked = Arc::new(Mutex::new(BTreeSet::::new())); + scanner.update_current_path = Arc::new({ + let paths = paths.clone(); + let proposed_walked = proposed_walked.clone(); + move |path: &str| { + let mut paths = paths.lock().expect("lock bounded actual-walk samples"); + assert!(paths.len() < MAX_WALK_SAMPLES, "fixture walk exceeded its entry budget"); + let bytes: usize = paths.iter().map(String::len).sum(); + assert!(path.len() <= MAX_WALK_BYTES - bytes, "fixture walk exceeded its byte budget"); + paths.push(path.to_string()); + if observe { + let proposed = fixture_proposal(&[changed_key]).expect("bounded successful fixture mutation"); + if let Some(segment) = path.strip_prefix("bucket/").and_then(|path| path.split('/').next()) + && proposed.contains(segment) + { + proposed_walked + .lock() + .expect("lock bounded observed segments") + .insert(segment.to_string()); + } + } + Box::pin(async {}) + } + }); + scanner + .scan_folder( + CancellationToken::new(), + CachedFolder { + name: "bucket".to_string(), + parent: None, + object_heal_prob_div: 1, + }, + &mut DataUsageEntry::default(), + ) + .await + .expect("actual folder walker must finish independently of diagnostics"); + let paths = paths.lock().expect("read walk samples").clone(); + assert!(!paths.is_empty()); + for prefix in ["hot", "cold", "other"] { + assert!( + paths.iter().any(|path| path == &format!("bucket/{prefix}")), + "all fixture segments must actually be walked" + ); + } + let store = FixtureStore::new(); + let revisions = DataUsageCache::default() + .load_with_revisions(store.clone(), CACHE_NAME) + .await + .expect("read empty fixture revisions"); + scanner + .new_cache + .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0) + .await + .expect("save actual walker output through the cache codec and revision gate"); + let loaded = store.strict_load().await; + assert_eq!(loaded.checked_flatten("bucket").expect("complete fixture tree").objects, 6); + assert_eq!( + cache_value(&loaded), + cache_value(&scanner.new_cache), + "codec round-trip must retain the entire cache, not just aggregate size" + ); + if observe { + let proposed = proposed_walked.lock().expect("read callback observations").clone(); + assert_eq!(proposed, BTreeSet::from(["hot".to_string()])); + let walked_segments: BTreeSet<_> = paths + .iter() + .filter_map(|path| path.strip_prefix("bucket/")) + .filter_map(|path| path.split('/').next()) + .collect(); + assert_eq!(walked_segments, BTreeSet::from(["cold", "hot", "other"])); + assert!(proposed.iter().all(|segment| walked_segments.contains(segment.as_str()))); + assert_eq!( + walked_segments.len() - proposed.len(), + 2, + "the two non-proposed segments must still be walked" + ); + eprintln!( + "segment fixture: proposed={proposed:?}, actual_segments={walked_segments:?}, actual_walk_callbacks={}, production_producer_coverage=unverified", + paths.len() + ); + } else { + assert!(proposed_walked.lock().expect("read disabled observations").is_empty()); + } + // Compare semantic values because map encoding order is not content identity. + (paths, cache_value(&loaded)) +} + +#[tokio::test] +#[serial] +async fn segment_observation_on_off_preserves_actual_walk_and_saved_cache() { + let off = walk_and_save(false).await; + let on = walk_and_save(true).await; + assert_eq!(off.0, on.0, "diagnostics must not change actual traversal order or coverage"); + assert_eq!(off.1, on.1, "diagnostics must not change the saved cache result"); +} diff --git a/docs/testing/scanner-checkpoint-fixture.md b/docs/testing/scanner-checkpoint-fixture.md index 9feb095cc..f9d437905 100644 --- a/docs/testing/scanner-checkpoint-fixture.md +++ b/docs/testing/scanner-checkpoint-fixture.md @@ -57,3 +57,19 @@ This fixture bounds object processing after directory enumeration. It does not p For every saved partial cache, the fixture also passes its progress through the production authenticated remote terminal-frame writer and stream consumer. A remote partial result must remain partial even when its progress reports visited objects. This covers the return-frame contract; it does not execute the remote RPC server, distributed locks, EC quorum persistence, mixed-version peers, process crashes, or fsync durability. The file backend models revision preconditions and persistence errors, not a concurrent object store. The synthetic namespace contains no customer data. Temporary files are removed with their owning fixture. Rolling back to a reader without the optional checkpoint metadata rebuilds partial coverage; it must not clear quota floors or complete authoritative snapshots. A passing fixture alone does not establish that the field report in [issue #7108](https://github.com/rustfs/rustfs/issues/7108) has been independently reproduced or fixed. A field diagnosis must separately identify the source capture, cycle and leader identity, and decoded bucket/set caches. + +## Segment Observation Diagnostics + +The nested `segment_observation` fixture compares diagnostic on/off runs of the real folder walker over six objects in `hot/`, `cold/`, and `other/`. Each run first rewrites `hot/one` with a different, equal-length ETag in real fixture metadata and reads it back to verify changed bytes at unchanged length. The successful fixture write supplies its known key to a diagnostic executed inside the real walker's path callback. Both runs save and reload the actual cache through the existing codec and revision-aware file backend. Assertions compare traversal order and the entire decoded cache, not encoded map order or aggregate size alone. Proposed top-level segments never reach a scanner selector or publication decision, and non-proposed segments must still be walked. The diagnostic retains at most four segments and 128 segment-name bytes; actual-walk samples are limited to 32 entries and 1,024 bytes. Exceeding sample limits fails the fixture rather than silently truncating its oracle. Saving a cache here is not an authoritative root publication. + +Entry/byte overflow and malformed keys reject the fixture proposal. Missing producers, process restarts, event gaps, and compacted child coverage remain **unverified production capabilities**, not simulated success cases in this fixture. Mainline bucket dirty generations and hashed metadata-cache invalidation stripes are not an exact, replayable object-key stream. The open [prefix reuse proposal #7208](https://github.com/rustfs/rustfs/pull/7208) is a separate candidate implementation; these tests neither import its hint map nor activate its skip path. + +The ECStore `segment_observation_equal_size_mutations_retire_metadata_generation` test uses the existing exact-key, test-only invalidation probe and actual owner operations. A same-length PUT must change the returned body and ETag while retiring the old generation; metadata-only PUT must change returned metadata and retire the old generation while size and ETag remain equal. Setup uses the existing full-fanout cache-priming helper; the observed mutations use normal owner locking. This is focused producer evidence, not an end-to-end connection between the owner probe and scanner range selection. The existing semantic mutation matrix covers additional owner entry points separately. + +```sh +cargo test -p rustfs-scanner --lib segment_observation -- --list +RUST_MIN_STACK=4194304 cargo test -p rustfs-scanner --lib segment_observation -- --nocapture +RUST_MIN_STACK=4194304 cargo test -p rustfs-ecstore --lib segment_observation_equal_size_mutations_retire_metadata_generation -- --nocapture +``` + +[W19](https://github.com/rustfs/backlog/issues/2272) remains open for trustworthy producer coverage, source/incarnation binding, and production shadow observations. No production stream, durable journal, runtime feature switch, scan skipping, or performance claim is introduced here. No restart/gap detection or restart-safe production coverage is established, and the revision-aware file backend does not prove EC publication durability.