From 222beaba009ba190dd229f089575de23299da7b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sun, 18 Oct 2026 00:25:02 +0800 Subject: [PATCH] fix(scanner): certify dirty usage per bucket and complete cycles with pending buckets (#8542) * fix(scanner): treat replication status writebacks as status-class hits A replication status writeback (persist_replication_state_if_current in Update mode, including the heals the scanner queues during its own walk) marked the bucket dirty on the namespace path: it advanced the dirty-usage generation, woke the scanner, reached the durable dirty journal and entered the activity snapshot. Any replicated bucket therefore made every scanner cycle Superseded, the authoritative usage baseline was never certified and usage freshness stayed stale for as long as replication ran (#8375). The writeback is now status class: record_status_class_mutation records only the process-local producer coverage bit and nothing else. There is no rescan queue for it, because a deployment with an active replication rule already walks every bucket on every cycle (ScannerMaintenanceFeatures::needs_regular_cycle) and that walk refreshes replication_stats. The routing lives in a named, unit-tested function in rustfs/src/storage/storage_api.rs; tier-expiration cleanup removes data and stays on the namespace path. No persisted format, protocol, admin API or configuration changes. This is phase 1 of docs/architecture/scanner-per-bucket-convergence.md, which this change adds together with its index entry. Refs #8375, #8494, rustfs/backlog#2816. * fix(scanner): refresh replication statistics during maintenance Refresh compacted prefixes and completed bucket caches during replication maintenance. Bind set execution reuse to the observed replication statistics so a same-cycle retry cannot restore older counters. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(scanner): avoid expanding the legacy S3 DTO surface Decode the existing replication configuration type from a JSON fixture instead of adding direct S3 DTO references to a new file. Preserve the compacted-prefix and same-cycle assertions and the footprint ratchet. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * feat(scanner): attribute activity guardrail vetoes per field The post-walk cluster activity proof compared the whole activity snapshot, so any namespace or dirty-usage generation movement vetoed publication. Veto only on the fields that change which heap the walk described (peer set, instance id, protocol version, topology, maintenance generation, data movement, movement generation, publication blocked) and name the field and host in the log and in a new usage-freshness counter. Generation movement is a shade signal consumed per bucket. A remote peer whose dirty-usage state moved keeps the whole-cycle outcome until scoped acknowledgements can name its buckets. * feat(data-usage): carry pending-bucket certification in DataUsageInfo Add the additive fields a per-bucket certified scanner snapshot needs: the sorted list and count of buckets whose dirty-usage generation moved during the walk, and an explicit observational marker for snapshots whose authoritative publication was vetoed. All three default when absent and are skipped when empty, so older readers see the existing snapshot shape. * feat(scanner): certify dirty usage per bucket and complete cycles with pending buckets A structurally complete walk that the guardrail did not veto is now `Complete` even when the dirty-usage generation moved during the walk. Certification keeps the compare-and-remove semantics of clear_dirty_usage_buckets: buckets whose generation still equals the start snapshot are cleared, buckets written during the walk stay in the dirty map and are published as the pending set, so usage_snapshot_converged becomes `pending == 0`. The authoritative object is written for such cycles; the observed route is selected by the explicit observational mark that vetoed cycles carry instead of by the convergence flag, and the root publication proof is read for every authoritative save. The post-lease activity recheck compares the guardrail fields, with the peer dirty-usage fallback, instead of the full snapshot digest. * feat(scanner): accept unconverged complete publications as the cache baseline complete_scanner_cache_baseline_plan_digest required the authoritative snapshot to be converged, so a site whose cycles always left a bucket pending could never scope a rescan and every cycle was a full walk. The authoritative object is now a structural baseline whenever it carries a convergence marker and no observational mark; bucket-cache reuse is then governed by certification, because pending buckets stay in the dirty map and are selected for the rescan while certified buckets reuse their set cache. Observed candidates keep requiring the unconverged marker, and segment reuse evidence keeps requiring a converged baseline. Add the single-disk cycle test: a bucket written during the walk is the only pending bucket, the others are certified, the publication is authoritative and unconverged, and the next cycle is scoped to that bucket; a held PUT tail still supersedes onto the observed route. * feat(admin): expose scanner guardrail vetoes in usage freshness Carry the attributed guardrail veto counter, timestamp and @ reason from the scanner metrics report into the madmin scanner status so an operator can tell a rebalance or a PUT tail from a broken probe. The fields are additive with serde defaults and merge like the deferral fields. * docs(scanner): record per-bucket certification in the design and changelog Amend the convergence design where the implementation deviates: pending buckets carry their walked value, maintenance_generation stays a veto, guardrail localisation and the remaining phase-2 items ship separately, last_cycle_result keeps its success label, and the admin surface gains the guardrail veto fields. Update the publication contract for the post-lease guardrail recheck and the explicit observational routing mark. * refactor(scanner): drop the unreachable failed-bucket input from the pending set A failed bucket never reaches a Complete cycle, so the pending set only needs the start snapshot. Also remove a redundant clone in the data-usage round-trip test. * test(scanner): name the activity field mutation table type * test(scanner): fill the post-walk gate in merged replication fixtures The replication maintenance fixture merged from main builds a ScannerCycleRequest literal and predates the test-only post_walk_gate field; give it the default so the test target compiles on the rebased branch. * ci(nextest): run the admin target-repair fixtures on the ecstore base stack repair_transaction_rejects_a_bucket_recreated_after_target_validation overflowed the 2 MiB libtest default twice in a row on Linux CI for this branch. Measured by sweeping RUST_MIN_STACK with the ci profile on macOS arm64: main (65f30bd5c) passes at 2032 KiB and overflows at 2016 KiB, this branch passes at 2040 KiB and overflows at 2032 KiB; on both trees the overflow is in the test's delete_bucket step on the real store, and the futures on that path are small (the peak is nested poll depth, not one large future). The module was given this stack in #7272 and removed in #7280 to check that boxing (#7277) sufficed; with the default now exhausted on main itself, give the fixtures the same 4 MiB base stack as the other real-storage rustfs fixtures, in both profiles. * test(scanner): fill the post-walk gate in the walkers-per-disk fixture The merge from main brought the walkers_per_disk cycle fixture (#8539), whose ScannerCycleRequest initializer predates the test-only post_walk_gate field this branch adds; the scanner test target did not compile. Fill the default so the merged head builds. --------- Co-authored-by: Hauser Co-authored-by: heihutu Co-authored-by: zhi22915 --- .config/nextest.toml | 7 +- CHANGELOG.md | 1 + crates/data-usage/src/data_usage.rs | 75 ++++++ .../ecstore/src/services/metrics_realtime.rs | 9 + crates/madmin/src/metrics.rs | 32 +++ crates/scanner-metrics/src/metrics.rs | 33 +++ crates/scanner/src/scanner.rs | 39 ++- crates/scanner/src/scanner/activity.rs | 105 +++++++- crates/scanner/src/scanner/tests.rs | 232 +++++++++++++++-- .../scanner/tests/scoped_ack_publication.rs | 3 + crates/scanner/src/scanner/usage_store.rs | 6 +- crates/scanner/src/scanner_io.rs | 146 ++++++++++- crates/scanner/src/scanner_io/cache.rs | 8 +- crates/scanner/src/scanner_io/dirty_usage.rs | 24 +- crates/scanner/src/scanner_io/io_cycle.rs | 43 ++- .../src/scanner_io/publish_gate_tests.rs | 15 +- crates/scanner/src/scanner_io/tests.rs | 176 +++++++++---- .../tests/per_bucket_certification.rs | 245 ++++++++++++++++++ .../scanner_io/tests/scoped_entry_fallback.rs | 10 +- .../src/scanner_io/tests/service_cohort.rs | 1 + .../scanner-per-bucket-convergence.md | 50 +++- .../architecture/scanner-usage-publication.md | 17 +- 22 files changed, 1158 insertions(+), 119 deletions(-) create mode 100644 crates/scanner/src/scanner_io/tests/per_bucket_certification.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 0bfabbe4a..92674305d 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -96,8 +96,11 @@ filter = 'package(rustfs-ecstore) & test(/^(bucket::lifecycle::bucket_lifecycle_ setup = 'ecstore-large-stack' # These cross-crate real-storage fixtures need the same bounded stack as ECStore tests. +# The admin target-repair fixtures compose the full bucket delete and recreate +# path on a real store; measured with RUST_MIN_STACK on 2026-10-15 they need +# 2016-2040 KiB of the 2 MiB libtest default and overflowed on Linux CI. [[profile.default.scripts]] -filter = '(package(rustfs) & test(/^(admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))' +filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))' setup = 'ecstore-base-stack' [[profile.default.scripts]] @@ -360,7 +363,7 @@ setup = 'ecstore-large-stack' # These cross-crate real-storage fixtures need the same bounded stack as ECStore tests. [[profile.ci.scripts]] -filter = '(package(rustfs) & test(/^(admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))' +filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))' setup = 'ecstore-base-stack' [[profile.ci.scripts]] diff --git a/CHANGELOG.md b/CHANGELOG.md index ca2187ecd..02eaf4e62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - **Replication status writebacks no longer invalidate scanner cycles**: every replication status writeback (`persist_replication_state_if_current` in `Update` mode, including the heals the scanner queues during its own walk) marked the bucket dirty on the namespace path, advanced the dirty-usage generation and woke the scanner, so a site-replication resync or any steadily replicated bucket ended every cycle `superseded`, the authoritative usage baseline was never certified, and the console reported usage freshness as stale for as long as replication ran (#8375). Those writebacks are now status class: they do not touch the dirty-usage generation, the activity snapshot, the dirty-usage wake or the durable dirty journal, and only the process-local producer coverage is recorded. They need no rescan queue, because a deployment with an active replication rule already walks every bucket on every cycle and that walk refreshes `replication_stats`. Tier-expiration cleanup, which removes data, stays on the namespace path. No persisted format, protocol, admin API or configuration changes. Design: `docs/architecture/scanner-per-bucket-convergence.md`. +- **A bucket written during a scanner walk no longer supersedes the whole cycle**: the post-walk cluster activity proof required the entire activity snapshot and the dirty-usage generation to be unchanged, so a single-node deployment with one bucket written every few seconds never reached a `Complete` cycle, `usage_snapshot_converged` never became true and the authoritative usage baseline was never certified (#8494). Dirty usage is now certified per bucket: a structurally complete walk that the guardrail did not veto is `Complete`, buckets whose dirty generation still matches the start snapshot are cleared, and buckets written during the walk stay dirty and are published as the pending set, so `usage_snapshot_converged` means "no bucket pending" and the next cycle is scoped to the pending buckets. Such a cycle writes the authoritative usage object; the observed object is now selected by an explicit observational mark that vetoed cycles carry. The guardrail itself vetoes only on the fields that change which namespace the walk described (peer set, process instance, protocol version, topology, maintenance generation, data movement, movement generation, publication blocked) and reports the field and host in the scanner log and in the new `usage_freshness` fields `guardrail_veto_total`, `last_guardrail_veto_unix_secs` and `last_guardrail_veto_reason`. In a distributed deployment a remote peer whose dirty usage moved during the walk still supersedes the cycle until scoped acknowledgements can name its buckets. `.usage.v2.json` gains the additive, default-when-absent fields `usage_snapshot_pending_buckets`, `usage_snapshot_pending_bucket_count` and `usage_snapshot_observational`; older readers ignore them and treat the snapshot as an authoritative but unconverged publication. Quota is unchanged: an uncertified bucket keeps its last usage value as the floor and live writes do not advance it. No protocol or configuration changes. Design: `docs/architecture/scanner-per-bucket-convergence.md`. - **SSE-C TLS requirement now covers copy-source keys**: with `RUSTFS_SSE_C_REQUIRE_TLS=true`, a `CopyObject` or `UploadPartCopy` that carried only the `x-amz-copy-source-server-side-encryption-customer-*` headers (an SSE-C source copied to a destination that is not SSE-C) was accepted on a plaintext transport and was not counted in `rustfs_ssec_plaintext_requests_total`, although those headers carry the source object's customer key. They now count as SSE-C headers: such requests are counted, and refused with `400 InvalidRequest` when the switch is on. Requests over TLS are unaffected. - **SSE-KMS encryption context could replace the location entry**: data keys are wrapped under an encryption-context entry, keyed by the bucket name, that ties them to the object's location. A client `x-amz-server-side-encryption-context` entry under the same key replaced it, and because the client context is stored with the object, the replacement also applied on every read. SSE-KMS writes now refuse such a context with `400 InvalidArgument`; other keys, including the names of other buckets, are unaffected. Objects stored with such an entry by an earlier release remain readable. See `docs/operations/kms-backend-security.md`. - **Two admin handlers reported a missing policy or user as `500 InternalError`**: `GET /rustfs/admin/v3/info-canned-policy` and `DELETE /rustfs/admin/v3/remove-user` wrapped every IAM error as an internal fault instead of classifying it, so a lookup for something that does not exist answered `500 InternalError` where the rest of the admin API answers `404 NoSuchResource` — `GET /rustfs/admin/v3/user-info` already did, and `iam_error_to_s3_error` already mapped `NoSuchPolicy`, `NoSuchUser`, `NoSuchServiceAccount` and `NoSuchTempAccount` to `NoSuchResource`. A client could not tell "does not exist" from a real server failure, and a controller that checks existence before creating (a Kubernetes operator reconciling policies or users) could never create the object: the missing-policy lookup failed with a `500` and it retried forever. Both handlers now map their IAM errors through the shared `iam_error_to_s3_error` helper, so a missing policy or user is `404 NoSuchResource` and the `error` is preserved as the S3 error's source; the three IAM calls in the removal path (temporary-user lookup, service-account lookup, delete) are all reclassified, and every other IAM error stays `500 InternalError`. Fixes #8117. diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index 3b0e81699..fd9681064 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -569,6 +569,25 @@ pub struct DataUsageInfo { /// which last-known-good generation contributed to the view. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub usage_snapshot_set_states: Vec, + /// Buckets whose dirty-usage generation moved while this complete + /// snapshot was being walked. Their `buckets_usage` values are the walk + /// result, which the cycle could not certify as current; the next cycle + /// rescans them. Sorted. Empty on a converged snapshot and on an + /// observational one. A bucket created during the walk may be listed + /// here without a `buckets_usage` entry. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub usage_snapshot_pending_buckets: Vec, + /// Length of `usage_snapshot_pending_buckets`, carried for readers that + /// only need the count. The scanner writes both from one set. + #[serde(default, skip_serializing_if = "u64_is_zero")] + pub usage_snapshot_pending_bucket_count: u64, + /// Set by the scanner on a snapshot whose authoritative publication was + /// vetoed or could not be verified. It is routed to the observed object + /// and never replaces the authoritative snapshot. Absent on snapshots + /// written before per-bucket certification, which keep + /// `usage_snapshot_converged == Some(false)` as their observed marker. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub usage_snapshot_observational: bool, /// An observational view may contain only the sets that completed this /// cycle (or retained a compatible last-known-good cache). #[serde(default)] @@ -584,6 +603,10 @@ pub struct DataUsageInfo { pub disk_usage_status: Vec, } +fn u64_is_zero(value: &u64) -> bool { + *value == 0 +} + /// Stable identity fields changed by both coordinated scanner publication and /// backward-compatible bucket namespace cleanup. #[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -3011,6 +3034,58 @@ mod tests { assert_eq!(current.usage_snapshot_converged, Some(false)); } + #[test] + fn pending_bucket_certification_fields_round_trip_and_stay_additive() { + let certified = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + usage_snapshot_pending_buckets: vec!["hot-a".to_string(), "hot-b".to_string()], + usage_snapshot_pending_bucket_count: 2, + ..Default::default() + }; + let encoded = rmp_serde::to_vec_named(&certified).expect("encode certified snapshot"); + let decoded: DataUsageInfo = rmp_serde::from_slice(&encoded).expect("decode certified snapshot"); + assert_eq!(decoded, certified); + assert!(!decoded.usage_snapshot_observational); + let legacy: LegacyUsageReader = rmp_serde::from_slice(&encoded).expect("legacy reader should ignore pending fields"); + assert_eq!(legacy.buckets_count, 0); + + let observed = DataUsageInfo { + usage_snapshot_observational: true, + ..certified + }; + let encoded = rmp_serde::to_vec_named(&observed).expect("encode observed snapshot"); + let decoded: DataUsageInfo = rmp_serde::from_slice(&encoded).expect("decode observed snapshot"); + assert!(decoded.usage_snapshot_observational); + + // Older snapshots carry none of the fields and decode to the defaults; + // a converged snapshot writes none of them (named encoding carries + // field names, so a skipped field leaves no trace in the bytes). + let converged = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(true), + ..Default::default() + }; + let encoded = rmp_serde::to_vec_named(&converged).expect("encode converged snapshot"); + for key in [ + "usage_snapshot_pending_buckets", + "usage_snapshot_pending_bucket_count", + "usage_snapshot_observational", + ] { + assert!( + !encoded.windows(key.len()).any(|window| window == key.as_bytes()), + "{key} must be skipped when empty" + ); + } + let decoded: DataUsageInfo = rmp_serde::from_slice(&encoded).expect("decode converged snapshot"); + assert!(decoded.usage_snapshot_pending_buckets.is_empty()); + assert_eq!(decoded.usage_snapshot_pending_bucket_count, 0); + assert!(!decoded.usage_snapshot_observational); + assert!(decoded.is_complete_bucket_usage_snapshot()); + } + #[test] fn convergence_marker_defaults_to_unknown_for_older_snapshots() { let encoded = rmp_serde::to_vec_named(&DataUsageInfo { diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index 41d2843c3..532f8f53c 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -231,6 +231,9 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe deferred_total: metrics.usage_freshness.deferred_total, last_deferred_unix_secs: metrics.usage_freshness.last_deferred_unix_secs, last_deferred_reason: metrics.usage_freshness.last_deferred_reason, + guardrail_veto_total: metrics.usage_freshness.guardrail_veto_total, + last_guardrail_veto_unix_secs: metrics.usage_freshness.last_guardrail_veto_unix_secs, + last_guardrail_veto_reason: metrics.usage_freshness.last_guardrail_veto_reason, }, maintenance_control: MadminScannerMaintenanceControlSnapshot { primary_control: metrics.maintenance_control.primary_control, @@ -886,10 +889,16 @@ mod test { deferred_total: 15, last_deferred_unix_secs: 16, last_deferred_reason: "data_movement".to_string(), + guardrail_veto_total: 4, + last_guardrail_veto_unix_secs: 17, + last_guardrail_veto_reason: "publication_blocked@".to_string(), }, ..Default::default() }); + assert_eq!(scanner.usage_freshness.guardrail_veto_total, 4); + assert_eq!(scanner.usage_freshness.last_guardrail_veto_unix_secs, 17); + assert_eq!(scanner.usage_freshness.last_guardrail_veto_reason, "publication_blocked@"); assert_eq!(scanner.usage_freshness.dirty_pending_buckets, 3); assert_eq!(scanner.usage_freshness.last_dirty_mark_unix_secs, 10); assert_eq!(scanner.usage_freshness.last_dirty_clear_unix_secs, 11); diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index 1753bde2b..3ca3daca9 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -318,6 +318,13 @@ pub struct ScannerUsageFreshnessSnapshot { pub last_deferred_unix_secs: u64, #[serde(rename = "last_deferred_reason", default)] pub last_deferred_reason: String, + /// Cluster-activity guardrail vetoes, attributed as `@`. + #[serde(rename = "guardrail_veto_total", default)] + pub guardrail_veto_total: u64, + #[serde(rename = "last_guardrail_veto_unix_secs", default)] + pub last_guardrail_veto_unix_secs: u64, + #[serde(rename = "last_guardrail_veto_reason", default)] + pub last_guardrail_veto_reason: String, } impl ScannerUsageFreshnessSnapshot { @@ -343,6 +350,11 @@ impl ScannerUsageFreshnessSnapshot { self.last_publication_reason = other.last_publication_reason.clone(); } self.deferred_total = self.deferred_total.saturating_add(other.deferred_total); + self.guardrail_veto_total = self.guardrail_veto_total.saturating_add(other.guardrail_veto_total); + if other.last_guardrail_veto_unix_secs > self.last_guardrail_veto_unix_secs { + self.last_guardrail_veto_unix_secs = other.last_guardrail_veto_unix_secs; + self.last_guardrail_veto_reason = other.last_guardrail_veto_reason.clone(); + } if other_deferred_state_at > self_deferred_state_at { self.deferred_pending = other.deferred_pending; } else if other_deferred_state_at == self_deferred_state_at { @@ -1719,6 +1731,26 @@ mod tests { .expect("legacy usage freshness should decode"); assert!(!legacy.deferred_pending); assert_eq!(legacy.deferred_total, 0); + assert_eq!(legacy.guardrail_veto_total, 0); + assert_eq!(legacy.last_guardrail_veto_unix_secs, 0); + assert!(legacy.last_guardrail_veto_reason.is_empty()); + + let mut vetoes = ScannerUsageFreshnessSnapshot::default(); + vetoes.merge(&ScannerUsageFreshnessSnapshot { + guardrail_veto_total: 2, + last_guardrail_veto_unix_secs: 30, + last_guardrail_veto_reason: "publication_blocked@".to_string(), + ..Default::default() + }); + vetoes.merge(&ScannerUsageFreshnessSnapshot { + guardrail_veto_total: 1, + last_guardrail_veto_unix_secs: 20, + last_guardrail_veto_reason: "movement_generation@node-2".to_string(), + ..Default::default() + }); + assert_eq!(vetoes.guardrail_veto_total, 3); + assert_eq!(vetoes.last_guardrail_veto_unix_secs, 30); + assert_eq!(vetoes.last_guardrail_veto_reason, "publication_blocked@"); let mut merged = ScannerUsageFreshnessSnapshot::default(); merged.merge(&ScannerUsageFreshnessSnapshot { diff --git a/crates/scanner-metrics/src/metrics.rs b/crates/scanner-metrics/src/metrics.rs index 8467dbd13..5044b4ac9 100644 --- a/crates/scanner-metrics/src/metrics.rs +++ b/crates/scanner-metrics/src/metrics.rs @@ -1032,6 +1032,9 @@ pub struct Metrics { scanner_usage_last_publication_unix_secs: AtomicU64, scanner_usage_last_publication_state: Mutex, scanner_usage_last_publication_reason: Mutex, + scanner_guardrail_veto_total: AtomicU64, + scanner_guardrail_last_veto_unix_secs: AtomicU64, + scanner_guardrail_last_veto_reason: Mutex, scanner_usage_last_save_result: AtomicU8, scanner_usage_deferred_pending: AtomicBool, scanner_usage_deferred_total: AtomicU64, @@ -1359,6 +1362,14 @@ pub struct ScannerUsageFreshnessSnapshot { pub last_deferred_unix_secs: u64, #[serde(default)] pub last_deferred_reason: String, + /// Guardrail vetoes attributed as `@`; a phase-2 peer + /// dirty-usage fallback is reported through the same fields. + #[serde(default)] + pub guardrail_veto_total: u64, + #[serde(default)] + pub last_guardrail_veto_unix_secs: u64, + #[serde(default)] + pub last_guardrail_veto_reason: String, } #[derive(Clone, Debug, Default, Serialize, Deserialize)] @@ -2100,6 +2111,9 @@ impl Metrics { scanner_usage_last_publication_unix_secs: AtomicU64::new(0), scanner_usage_last_publication_state: Mutex::new(String::new()), scanner_usage_last_publication_reason: Mutex::new(String::new()), + scanner_guardrail_veto_total: AtomicU64::new(0), + scanner_guardrail_last_veto_unix_secs: AtomicU64::new(0), + scanner_guardrail_last_veto_reason: Mutex::new(String::new()), scanner_usage_last_save_result: AtomicU8::new(ScannerUsageSaveResult::Unknown as u8), scanner_usage_deferred_pending: AtomicBool::new(false), scanner_usage_deferred_total: AtomicU64::new(0), @@ -2500,6 +2514,19 @@ impl Metrics { *last_reason = reason; } + /// Record an attributed cluster-activity guardrail veto (`@`) + /// so an operator can tell a rebalance or a PUT tail from a broken probe. + pub fn record_scanner_guardrail_veto(&self, reason: impl Into) { + self.scanner_guardrail_veto_total.fetch_add(1, Ordering::Relaxed); + self.scanner_guardrail_last_veto_unix_secs + .store(unix_now_secs(), Ordering::Relaxed); + let mut last_reason = match self.scanner_guardrail_last_veto_reason.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + *last_reason = reason.into(); + } + pub fn record_scanner_usage_durable_success(&self) { self.record_scanner_usage_publication("success", ""); self.scanner_usage_last_durable_success_unix_secs @@ -3593,6 +3620,12 @@ impl Metrics { Ok(reason) => reason.clone(), Err(poisoned) => poisoned.into_inner().clone(), }, + guardrail_veto_total: self.scanner_guardrail_veto_total.load(Ordering::Relaxed), + last_guardrail_veto_unix_secs: self.scanner_guardrail_last_veto_unix_secs.load(Ordering::Relaxed), + last_guardrail_veto_reason: match self.scanner_guardrail_last_veto_reason.lock() { + Ok(reason) => reason.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + }, }; m.throttle_idle_mode_enabled = self.scanner_throttle_idle_mode_enabled.load(Ordering::Relaxed); m.throttle_sleep_factor = self.scanner_throttle_sleep_factor_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0; diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 907440999..30769a174 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1904,6 +1904,8 @@ where service_cohort: scheduling.service_cohort, #[cfg(test)] resolved_scope_observer: None, + #[cfg(test)] + post_walk_gate: None, }, ) .await; @@ -2033,7 +2035,10 @@ where && let Ok(result) = &scan_result && result.status == ScannerCycleStatus::Complete { - scanner_post_lease_activity_defer_reason(result.activity_digest(), probe_scanner_activity(storeapi.as_ref(), true).await) + scanner_post_lease_activity_defer_reason( + result.activity_baseline(), + probe_scanner_activity(storeapi.as_ref(), true).await, + ) } else { None }; @@ -2376,6 +2381,7 @@ where } else { "incomplete_coverage" }; + let pending_dirty_usage_buckets = scan_cycle_result.pending_dirty_usage_buckets(); let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) = finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result); let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() { @@ -2587,6 +2593,7 @@ where state = "completed", duration = ?now.elapsed(), cycles_total = cycle_info.cycle_completed.len(), + pending_dirty_buckets = pending_dirty_usage_buckets, "Scanner cycle completed" ); @@ -3675,14 +3682,22 @@ impl ScannerCyclePublicationPlan { } } +/// Re-prove the guardrail after every peer granted a publication lease. The +/// comparison uses the guardrail fields (`scanner_activity_guardrail_veto`), +/// not the full digest: local generation movement in this window is the next +/// cycle's work and its buckets stay in the dirty map. A peer whose dirty +/// usage moved keeps deferring in this phase, as it does during the walk. fn scanner_post_lease_activity_defer_reason( - expected_digest: Option<[u8; 32]>, + baseline: Option<&ScannerActivitySnapshot>, activity: Result, ) -> Option { match activity { Ok(snapshot) if scanner_activity_allows_usage_publication(&snapshot) - && expected_digest == Some(scanner_activity_snapshot_digest(&snapshot)) => + && baseline.is_some_and(|baseline| { + scanner_activity_guardrail_veto(baseline, &snapshot).is_none() + && scanner_peer_dirty_usage_moved(baseline, &snapshot).is_none() + }) => { None } @@ -3690,8 +3705,13 @@ fn scanner_post_lease_activity_defer_reason( let blocker = match &observed { Err(_) => "probe_failed", Ok(snapshot) if !scanner_activity_allows_usage_publication(snapshot) => "publication_blocked", - Ok(_) if expected_digest.is_none() => "baseline_missing", - Ok(_) => "activity_changed", + Ok(_) if baseline.is_none() => "baseline_missing", + Ok(snapshot) => baseline + .and_then(|baseline| { + scanner_activity_guardrail_veto(baseline, snapshot) + .or_else(|| scanner_peer_dirty_usage_moved(baseline, snapshot)) + }) + .map_or("activity_changed", |veto| veto.field), }; debug!( target: "rustfs::scanner", @@ -3980,10 +4000,11 @@ use usage_store::*; pub(crate) use activity::scanner_node_activity_for_tests; pub use activity::scanner_topology_digest; pub(crate) use activity::{ - ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity, - scanner_activity_allows_usage_publication, scanner_activity_dirty_usage_state_for_host, - scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_activity_structural_digest, - scanner_dirty_usage_acknowledgements, + LOCAL_SCANNER_ACTIVITY_NODE, ScannerActivityGuardrailVeto, ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, + ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity, scanner_activity_allows_usage_publication, + scanner_activity_dirty_usage_state_for_host, scanner_activity_guardrail_veto, scanner_activity_publication_lease_targets, + scanner_activity_snapshot_digest, scanner_activity_structural_digest, scanner_dirty_usage_acknowledgements, + scanner_peer_dirty_usage_moved, }; pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance}; pub use backlog::{ diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index efa930a6e..c309a878d 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -468,7 +468,7 @@ pub(super) struct ScannerMovementWaitContext<'a, G, F> { pub(super) recovery_wake: Option<&'a Notify>, } -pub(super) const LOCAL_SCANNER_ACTIVITY_NODE: &str = ""; +pub(crate) const LOCAL_SCANNER_ACTIVITY_NODE: &str = ""; #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct ScannerNodeActivity { @@ -1061,6 +1061,109 @@ pub(crate) fn scanner_activity_structural_digest(snapshot: &ScannerActivitySnaps hasher.finalize().into() } +/// An attributed veto of the cluster activity guardrail: the named field on +/// the named host moved between the pre-walk and post-walk probes. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ScannerActivityGuardrailVeto { + pub(crate) host: String, + pub(crate) field: &'static str, +} + +impl ScannerActivityGuardrailVeto { + pub(crate) fn new(host: &str, field: &'static str) -> Self { + Self { + host: host.to_string(), + field, + } + } + + /// Operator-facing label, `@`. + pub(crate) fn reason(&self) -> String { + format!("{}@{}", self.field, self.host) + } +} + +/// `ScannerNodeActivity` fields whose movement means the heap the probe +/// describes is not the heap that was walked. Each is a guardrail veto on its +/// own; `peer_set` covers a host that appeared or disappeared. +#[cfg(test)] +pub(crate) const SCANNER_ACTIVITY_GUARDRAIL_VETO_FIELDS: [&str; 8] = [ + "peer_set", + "instance_id", + "protocol_version", + "topology_digest", + "maintenance_generation", + "data_movement_active", + "movement_generation", + "publication_blocked", +]; + +/// `ScannerNodeActivity` fields that only shade buckets: they are consumed +/// per bucket by dirty-usage certification and never veto on their own. +#[cfg(test)] +pub(crate) const SCANNER_ACTIVITY_SHADE_FIELDS: [&str; 3] = + ["namespace_generation", "dirty_usage_generation", "dirty_usage_pending"]; + +/// Compare the guardrail fields of two activity probes and name the first +/// field that moved. `maintenance_generation` is a veto because no cycle today +/// distinguishes whether it published maintenance results; it is also part of +/// `scanner_activity_structural_digest`. +pub(crate) fn scanner_activity_guardrail_veto( + before: &ScannerActivitySnapshot, + after: &ScannerActivitySnapshot, +) -> Option { + for (host, before_activity) in before { + let Some(after_activity) = after.get(host) else { + return Some(ScannerActivityGuardrailVeto::new(host, "peer_set")); + }; + let moved = if before_activity.instance_id != after_activity.instance_id { + "instance_id" + } else if before_activity.protocol_version != after_activity.protocol_version { + "protocol_version" + } else if before_activity.topology_digest != after_activity.topology_digest { + "topology_digest" + } else if before_activity.maintenance_generation != after_activity.maintenance_generation { + "maintenance_generation" + } else if before_activity.data_movement_active != after_activity.data_movement_active { + "data_movement_active" + } else if before_activity.movement_generation != after_activity.movement_generation { + "movement_generation" + } else if before_activity.publication_blocked != after_activity.publication_blocked { + "publication_blocked" + } else { + continue; + }; + return Some(ScannerActivityGuardrailVeto::new(host, moved)); + } + after + .keys() + .find(|host| !before.contains_key(*host)) + .map(|host| ScannerActivityGuardrailVeto::new(host, "peer_set")) +} + +/// Name the first remote peer whose dirty-usage state moved between the two +/// probes. The local node's dirty state is certified per bucket, but a peer's +/// moved buckets cannot be named without a post-walk scoped snapshot, so this +/// phase keeps the whole-cycle fallback for peers; phase 5 lifts it. +pub(crate) fn scanner_peer_dirty_usage_moved( + before: &ScannerActivitySnapshot, + after: &ScannerActivitySnapshot, +) -> Option { + before + .iter() + .filter(|(host, _)| host.as_str() != LOCAL_SCANNER_ACTIVITY_NODE) + .find_map(|(host, before_activity)| { + let after_activity = after.get(host)?; + if before_activity.dirty_usage_generation != after_activity.dirty_usage_generation { + Some(ScannerActivityGuardrailVeto::new(host, "dirty_usage_generation")) + } else if before_activity.dirty_usage_pending != after_activity.dirty_usage_pending { + Some(ScannerActivityGuardrailVeto::new(host, "dirty_usage_pending")) + } else { + None + } + }) +} + pub(crate) fn scanner_activity_allows_usage_publication(snapshot: &ScannerActivitySnapshot) -> bool { !snapshot.is_empty() && snapshot.values().all(|activity| { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index a3cb48e49..41adad216 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -6606,6 +6606,7 @@ async fn test_observational_usage_defers_when_authoritative_baseline_is_missing( let (sender, receiver) = mpsc::channel(1); let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); observation.usage_snapshot_converged = Some(false); + observation.usage_snapshot_observational = true; sender.send(observation).await.expect("observation should enqueue"); drop(sender); @@ -6654,6 +6655,7 @@ async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_iden let (sender, receiver) = mpsc::channel(1); let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); observation.usage_snapshot_converged = Some(false); + observation.usage_snapshot_observational = true; sender.send(observation).await.expect("observation should enqueue"); drop(sender); @@ -6695,6 +6697,7 @@ async fn test_observational_usage_uses_bootstrap_pending_primary_as_baseline() { let (sender, receiver) = mpsc::channel(1); let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1); observation.usage_snapshot_converged = Some(false); + observation.usage_snapshot_observational = true; sender.send(observation).await.expect("observation should enqueue"); drop(sender); @@ -7577,6 +7580,7 @@ async fn test_store_data_usage_in_backend_preserves_superseded_status() { scanner_epoch: Some(7), scanner_cycle: Some(11), usage_snapshot_converged: Some(false), + usage_snapshot_observational: true, ..complete_usage_with_bucket_count(None, 1) }) .await @@ -7604,6 +7608,56 @@ async fn test_store_data_usage_in_backend_preserves_superseded_status() { ); } +#[tokio::test] +async fn test_store_data_usage_in_backend_publishes_pending_bucket_certification_as_authoritative() { + let store = Arc::new(MemoryConfigStore::default()); + let authoritative_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + let observed_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()); + let authoritative = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(10), + usage_snapshot_converged: Some(true), + ..complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 1) + }; + store.objects.lock().await.insert( + authoritative_key.clone(), + serde_json::to_vec(&authoritative).expect("authoritative snapshot should encode"), + ); + store.revisions.lock().await.insert(authoritative_key.clone(), 1); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(7), + scanner_cycle: Some(11), + usage_snapshot_converged: Some(false), + usage_snapshot_pending_buckets: vec!["bucket-0".to_string()], + usage_snapshot_pending_bucket_count: 1, + ..complete_usage_with_bucket_count(None, 1) + }) + .await + .expect("certified usage snapshot should enqueue"); + drop(sender); + + let outcome = store_data_usage_in_backend_with_outcome(CancellationToken::new(), store.clone(), receiver).await; + assert_eq!(outcome, DataUsagePersistOutcome::Saved); + let objects = store.objects.lock().await; + assert!( + !objects.contains_key(&observed_key), + "pending buckets do not make a snapshot observational" + ); + let saved = serde_json::from_slice::(objects.get(&authoritative_key).expect("authoritative snapshot")) + .expect("persisted usage snapshot should decode"); + assert_eq!(saved.scanner_cycle, Some(11)); + assert_eq!(saved.usage_snapshot_converged, Some(false)); + assert_eq!(saved.usage_snapshot_pending_buckets, vec!["bucket-0".to_string()]); + assert_eq!(saved.usage_snapshot_pending_bucket_count, 1); + assert!(!saved.usage_snapshot_observational); + assert_eq!(saved.usage_snapshot_authoritative_baseline, None); + assert!(data_usage_info_has_persisted_baseline_identity(&saved)); +} + #[tokio::test] #[serial] async fn test_store_data_usage_in_backend_removes_observed_after_authoritative_save() { @@ -7627,6 +7681,7 @@ async fn test_store_data_usage_in_backend_removes_observed_after_authoritative_s scanner_epoch: Some(7), scanner_cycle: Some(11), usage_snapshot_converged: Some(false), + usage_snapshot_observational: true, ..complete_usage_with_bucket_count(None, 1) }) .await @@ -10009,24 +10064,47 @@ fn scoped_scan_remote_dirty_coverage_invalidates_local_bucket_current() { } #[test] -fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acquisition() { - let before = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]); - let expected_digest = Some(scanner_activity_snapshot_digest(&before)); - assert_eq!(scanner_post_lease_activity_defer_reason(expected_digest, Ok(before.clone())), None); +fn post_lease_activity_proof_rejects_a_peer_whose_dirty_usage_moved_before_lease_acquisition() { + let before = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("local", 5, 2)), + ("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3)), + ]); + assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(before.clone())), None); + // Local generation movement in the lease window is the next cycle's + // work: its buckets stay in the dirty map and the walk remains certified. + let mut local_shaded = before.clone(); + let local = local_shaded + .get_mut(LOCAL_SCANNER_ACTIVITY_NODE) + .expect("local node should be present"); + local.namespace_generation += 1; + local.dirty_usage_generation += 1; + local.dirty_usage_pending = true; + assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(local_shaded)), None); + + // A peer PUT tail that drained before the lease leaves the movement + // generation accepted by the lease unchanged, but the peer's dirty state + // moved and this phase cannot name its buckets. let mut after = before.clone(); - after - .get_mut("node-2") - .expect("writer should be present") - .namespace_generation += 1; + let writer = after.get_mut("node-2").expect("writer should be present"); + writer.namespace_generation += 1; + writer.dirty_usage_generation += 1; assert_eq!( before["node-2"].movement_generation, after["node-2"].movement_generation, "the existing movement-only lease remains valid after a PUT tail drains" ); assert!(scanner_activity_allows_usage_publication(&after)); - let reason = scanner_post_lease_activity_defer_reason(expected_digest, Ok(after)); + let reason = scanner_post_lease_activity_defer_reason(Some(&before), Ok(after)); assert_eq!(reason, Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)); + // A guardrail field moving in the window still defers. + let mut moved = before.clone(); + moved.get_mut("node-2").expect("peer should be present").movement_generation += 1; + assert_eq!( + scanner_post_lease_activity_defer_reason(Some(&before), Ok(moved)), + Some(ScannerCycleDeferReason::ActivityBaselineUnavailable) + ); + let result = ScannerCycleResult::new(ScannerCycleStatus::Complete, None).with_remote_dirty_usage_acknowledgements(vec![ ScannerDirtyUsageAcknowledgement { host: "node-2".to_string(), @@ -10109,15 +10187,13 @@ fn remote_lease_validation_failure_preserves_movement_defer_for_remote_fence_los #[test] fn post_lease_activity_proof_requires_a_complete_matching_baseline() { let before = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]); - let digest = scanner_activity_snapshot_digest(&before); let mut blocked = before.clone(); blocked.get_mut("node-2").expect("peer should be present").publication_blocked = true; - let blocked_digest = scanner_activity_snapshot_digest(&blocked); for (expected, observed) in [ - (None, Ok(before)), - (Some(digest), Err("peer is unavailable".to_string())), - (Some(digest), Ok(BTreeMap::new())), - (Some(blocked_digest), Ok(blocked)), + (None, Ok(before.clone())), + (Some(&before), Err("peer is unavailable".to_string())), + (Some(&before), Ok(BTreeMap::new())), + (Some(&blocked), Ok(blocked.clone())), ] { assert_eq!( scanner_post_lease_activity_defer_reason(expected, observed), @@ -10350,6 +10426,132 @@ fn remote_restart_is_distinct_from_deferred_cluster_activity() { assert!(scanner_activity_observed_work(ScannerActivityObservation::RemoteRestarted)); } +type ActivityFieldMutations = Vec<(&'static str, Box)>; + +#[test] +fn guardrail_veto_is_attributed_to_the_field_and_host_that_moved() { + let before = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("local", 5, 2)), + ("node-2".to_string(), scanner_node_activity("remote", 7, 3)), + ]); + assert_eq!(scanner_activity_guardrail_veto(&before, &before), None); + + let cases: ActivityFieldMutations = vec![ + ("instance_id", Box::new(|activity| activity.instance_id = "restarted".to_string())), + ("protocol_version", Box::new(|activity| activity.protocol_version += 1)), + ("topology_digest", Box::new(|activity| activity.topology_digest = [4; 32])), + ("maintenance_generation", Box::new(|activity| activity.maintenance_generation += 1)), + ("data_movement_active", Box::new(|activity| activity.data_movement_active = true)), + ("movement_generation", Box::new(|activity| activity.movement_generation += 1)), + ("publication_blocked", Box::new(|activity| activity.publication_blocked = true)), + ]; + for host in [LOCAL_SCANNER_ACTIVITY_NODE, "node-2"] { + for (field, mutate) in &cases { + let mut after = before.clone(); + mutate(after.get_mut(host).expect("host should be present")); + let veto = scanner_activity_guardrail_veto(&before, &after).unwrap_or_else(|| panic!("{field} on {host} must veto")); + assert_eq!(veto.field, *field, "{host}"); + assert_eq!(veto.host, host); + assert_eq!(veto.reason(), format!("{field}@{host}")); + assert!(SCANNER_ACTIVITY_GUARDRAIL_VETO_FIELDS.contains(field)); + } + } + + let mut peer_lost = before.clone(); + peer_lost.remove("node-2"); + assert_eq!( + scanner_activity_guardrail_veto(&before, &peer_lost), + Some(ScannerActivityGuardrailVeto::new("node-2", "peer_set")) + ); + let mut peer_added = before.clone(); + peer_added.insert("node-3".to_string(), scanner_node_activity("third", 1, 1)); + assert_eq!( + scanner_activity_guardrail_veto(&before, &peer_added), + Some(ScannerActivityGuardrailVeto::new("node-3", "peer_set")) + ); +} + +#[test] +fn generation_movement_shades_buckets_without_a_guardrail_veto() { + let before = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("local", 5, 2)), + ("node-2".to_string(), scanner_node_activity("remote", 7, 3)), + ]); + let shade: ActivityFieldMutations = vec![ + ("namespace_generation", Box::new(|activity| activity.namespace_generation += 1)), + ("dirty_usage_generation", Box::new(|activity| activity.dirty_usage_generation += 1)), + ("dirty_usage_pending", Box::new(|activity| activity.dirty_usage_pending = true)), + ]; + for host in [LOCAL_SCANNER_ACTIVITY_NODE, "node-2"] { + for (field, mutate) in &shade { + let mut after = before.clone(); + mutate(after.get_mut(host).expect("host should be present")); + assert_ne!(after, before); + assert_eq!(scanner_activity_guardrail_veto(&before, &after), None, "{field} on {host} must not veto"); + assert!(SCANNER_ACTIVITY_SHADE_FIELDS.contains(field)); + assert_eq!( + scanner_activity_structural_digest(&before), + scanner_activity_structural_digest(&after), + "{field} must stay outside the structural digest" + ); + } + } + + // Everything shaded at once on every host is still not a veto. + let mut all_shaded = before.clone(); + for activity in all_shaded.values_mut() { + activity.namespace_generation += 3; + activity.dirty_usage_generation += 3; + activity.dirty_usage_pending = true; + } + assert_eq!(scanner_activity_guardrail_veto(&before, &all_shaded), None); +} + +#[test] +fn peer_dirty_usage_movement_is_reported_for_remote_hosts_only() { + let before = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("local", 5, 2)), + ("node-2".to_string(), scanner_node_activity("remote", 7, 3)), + ]); + assert_eq!(scanner_peer_dirty_usage_moved(&before, &before), None); + + let mut local_moved = before.clone(); + let local = local_moved + .get_mut(LOCAL_SCANNER_ACTIVITY_NODE) + .expect("local activity should exist"); + local.dirty_usage_generation += 1; + local.dirty_usage_pending = true; + local.namespace_generation += 1; + assert_eq!(scanner_peer_dirty_usage_moved(&before, &local_moved), None); + + let mut remote_namespace_only = before.clone(); + remote_namespace_only + .get_mut("node-2") + .expect("remote activity should exist") + .namespace_generation += 1; + assert_eq!(scanner_peer_dirty_usage_moved(&before, &remote_namespace_only), None); + + let mut remote_generation = before.clone(); + remote_generation + .get_mut("node-2") + .expect("remote activity should exist") + .dirty_usage_generation += 1; + assert_eq!( + scanner_peer_dirty_usage_moved(&before, &remote_generation), + Some(ScannerActivityGuardrailVeto::new("node-2", "dirty_usage_generation")) + ); + + let mut remote_pending = before.clone(); + remote_pending + .get_mut("node-2") + .expect("remote activity should exist") + .dirty_usage_pending = true; + assert_eq!( + scanner_peer_dirty_usage_moved(&before, &remote_pending), + Some(ScannerActivityGuardrailVeto::new("node-2", "dirty_usage_pending")) + ); +} + #[test] fn local_maintenance_wakeup_releases_a_remote_maintenance_block() { let blocked = scanner_activity_backoff_blocked_after_wake(false, ScannerCycleWakeReason::ClusterMaintenance); diff --git a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs index 472e595d7..0acac8514 100644 --- a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs +++ b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs @@ -48,6 +48,7 @@ async fn complete_candidate(store: &Arc, cycle: u64) -> (crate::scanner requires_full_scan: true, service_cohort: None, resolved_scope_observer: None, + post_walk_gate: None, }, ) .await @@ -242,6 +243,7 @@ async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_pro // Seed a complete but unconverged observation from real scanner coverage; // the production writer attaches its authoritative baseline identity. observed_candidate.usage_snapshot_converged = Some(false); + observed_candidate.usage_snapshot_observational = true; let observation = publish_candidate(&store, &prior_scan, observed_candidate, None).await; let (outcome, proof) = observation.into_parts(); assert_eq!(outcome, DataUsagePersistOutcome::Saved); @@ -269,6 +271,7 @@ async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_pro requires_full_scan: false, service_cohort: None, resolved_scope_observer: Some(observer), + post_walk_gate: None, }, ) .await diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 21a8a7cad..828999202 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -577,7 +577,10 @@ where outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement); break 'updates; } - let observational = data_usage_info.usage_snapshot_converged == Some(false); + // Routing keys on the explicit producer mark, not on convergence: a + // `Complete` cycle with pending buckets is authoritative but not + // converged, while a vetoed cycle is observational. + let observational = data_usage_info.usage_snapshot_observational; let target_path = if observational { DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str() } else { @@ -1143,7 +1146,6 @@ where } } if !observational - && data_usage_info.usage_snapshot_converged == Some(true) && matches!(outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable) && (outcome == DataUsagePersistOutcome::AlreadyDurable || write_confirmed) && let (Some(expected), Some(epoch)) = (ack_expectation.as_ref(), ack_epoch) diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 2ba147f23..75ce2a07b 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -39,7 +39,7 @@ use s3s::dto::{ BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration, }; use sha2::{Digest as _, Sha256}; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::future::Future; use std::path::Path; use std::pin::Pin; @@ -409,14 +409,26 @@ fn resolve_remote_dirty_usage_scope( } } +/// `expected_converged` is `None` for the authoritative object: a `Complete` +/// cycle with pending buckets publishes there unconverged and is still a +/// structural baseline, because the buckets it could not certify stay in the +/// dirty map and are rescanned rather than reused. The convergence marker +/// must still be present; a snapshot without one was not written by a +/// certifying scanner. fn complete_scanner_cache_snapshot_plan_digest( snapshot: &DataUsageInfo, proof: ScannerCacheBaselineProof<'_>, - expected_converged: bool, + expected_converged: Option, ) -> Option { + // The observational mark disqualifies only the authoritative slot: an + // observed candidate carries it by construction. + let marks_accepted = match expected_converged { + Some(expected) => snapshot.usage_snapshot_converged == Some(expected), + None => snapshot.usage_snapshot_converged.is_some() && !snapshot.usage_snapshot_observational, + }; if !snapshot.is_complete_bucket_usage_snapshot() || snapshot.usage_snapshot_partial - || snapshot.usage_snapshot_converged != Some(expected_converged) + || !marks_accepted || snapshot.scanner_epoch != Some(proof.leader_epoch) || snapshot.usage_snapshot_set_states.len() != proof.expected_sources.len() { @@ -455,7 +467,7 @@ fn complete_scanner_cache_snapshot_plan_digest( fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option { let authoritative = serde_json::from_slice::(proof.authoritative_data?).ok()?; - if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) { + if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, None) { return Some(validated_digest); } @@ -464,7 +476,7 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof< // remains observational: this proof grants bucket-scope reuse only and // never changes authoritative usage publication or dirty acknowledgement. let authoritative_has_identity = (crate::scanner::data_usage_info_has_persisted_baseline_identity(&authoritative) - && authoritative.usage_snapshot_converged != Some(false)) + && !authoritative.usage_snapshot_observational) || crate::scanner::data_usage_info_is_bootstrap_pending(&authoritative); if !authoritative_has_identity { return None; @@ -474,7 +486,7 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof< return None; } - complete_scanner_cache_snapshot_plan_digest(&observed, proof, false) + complete_scanner_cache_snapshot_plan_digest(&observed, proof, Some(false)) } fn scanner_segment_reuse_baseline_producer_evidence( @@ -492,7 +504,9 @@ fn scanner_segment_reuse_baseline_producer_evidence( let Ok(authoritative) = serde_json::from_slice::(authoritative_data) else { return (evidence, false); }; - if complete_scanner_cache_snapshot_plan_digest(&authoritative, baseline_proof, true).is_none() { + // Segment (prefix) reuse keeps requiring a converged baseline: its + // producer evidence is proven against a dirty snapshot that was current. + if complete_scanner_cache_snapshot_plan_digest(&authoritative, baseline_proof, Some(true)).is_none() { return (evidence, false); } let Some(cold_zero_walk_oracle) = @@ -1014,9 +1028,13 @@ fn classify_nsscanner_cycle( return ScannerCycleStatus::Incomplete; } - match (activity_status, dirty_usage_status) { - (ScannerCycleActivityStatus::Unchanged, DirtyUsageSnapshotStatus::Current) => ScannerCycleStatus::Complete, - _ => ScannerCycleStatus::Superseded, + // Termination is snapshot-at-the-beginning: a structurally complete walk + // that the guardrail did not veto is `Complete`. Buckets whose dirty + // generation moved during the walk stay pending for the next cycle and + // are reported in the publication; they no longer supersede the cycle. + match activity_status { + ScannerCycleActivityStatus::Unchanged => ScannerCycleStatus::Complete, + ScannerCycleActivityStatus::Changed | ScannerCycleActivityStatus::Unverified => ScannerCycleStatus::Superseded, } } @@ -1028,24 +1046,52 @@ fn should_publish_observational_snapshot(status: ScannerCycleStatus) -> bool { matches!(status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)) } +/// Pending buckets of a cycle, sorted so the publication is deterministic. +pub(crate) type ScannerCyclePendingBuckets = BTreeSet; + +/// Stamp the certification outcome of a cycle on its usage snapshot. A +/// `Complete` cycle publishes an authoritative snapshot that is converged +/// only when no bucket is pending; any other publishable status is an +/// observational snapshot routed to the observed object. The marks are part +/// of the bytes the publication expectation digests, so every writer applies +/// exactly this function. +fn apply_scanner_cycle_publication_marks( + data_usage_info: &mut DataUsageInfo, + status: ScannerCycleStatus, + pending_buckets: &ScannerCyclePendingBuckets, +) { + let complete = status == ScannerCycleStatus::Complete; + data_usage_info.usage_snapshot_observational = !complete; + data_usage_info.usage_snapshot_converged = Some(complete && pending_buckets.is_empty()); + if complete { + data_usage_info.usage_snapshot_pending_buckets = pending_buckets.iter().cloned().collect(); + data_usage_info.usage_snapshot_pending_bucket_count = u64::try_from(pending_buckets.len()).unwrap_or(u64::MAX); + } else { + data_usage_info.usage_snapshot_pending_buckets.clear(); + data_usage_info.usage_snapshot_pending_bucket_count = 0; + } +} + fn prepare_usage_snapshot_for_publication( status: ScannerCycleStatus, + pending_buckets: &ScannerCyclePendingBuckets, mut data_usage_info: DataUsageInfo, ) -> Option { if !should_publish_usage_snapshot(status) { return None; } - data_usage_info.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete); + apply_scanner_cycle_publication_marks(&mut data_usage_info, status, pending_buckets); Some(data_usage_info) } async fn publish_usage_snapshot( updates: &mpsc::Sender, status: ScannerCycleStatus, + pending_buckets: &ScannerCyclePendingBuckets, data_usage_info: DataUsageInfo, ) -> Result { - let Some(data_usage_info) = prepare_usage_snapshot_for_publication(status, data_usage_info) else { + let Some(data_usage_info) = prepare_usage_snapshot_for_publication(status, pending_buckets, data_usage_info) else { return Ok(false); }; send_data_usage_update(updates, data_usage_info).await?; @@ -1059,10 +1105,17 @@ async fn publish_observational_snapshot( data_usage_info.usage_snapshot_complete = false; data_usage_info.usage_snapshot_partial = true; data_usage_info.usage_snapshot_converged = Some(false); + data_usage_info.usage_snapshot_observational = true; + data_usage_info.usage_snapshot_pending_buckets.clear(); + data_usage_info.usage_snapshot_pending_bucket_count = 0; send_data_usage_update(updates, data_usage_info).await?; Ok(true) } +/// Post-walk verdict of the cluster activity guardrail. `Changed` is an +/// attributed veto (or, for peers, the phase-2 dirty-usage fallback); local +/// namespace and dirty-usage generation movement is not a veto and is +/// certified per bucket instead. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum ScannerCycleActivityStatus { Unchanged, @@ -1083,10 +1136,47 @@ where let publication_blocked = store.scanner_data_usage_publication_blocked().await; match crate::scanner::probe_scanner_activity(store, distributed).await { Ok(after) => { - let status = if !publication_blocked && after == *before { - ScannerCycleActivityStatus::Unchanged + let veto = if publication_blocked { + Some(crate::scanner::ScannerActivityGuardrailVeto::new( + crate::scanner::LOCAL_SCANNER_ACTIVITY_NODE, + "publication_blocked", + )) } else { + crate::scanner::scanner_activity_guardrail_veto(before, &after) + }; + let status = if let Some(veto) = veto { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "cycle_guardrail_veto", + stage = "post_scan_probe", + field = veto.field, + host = %veto.host, + "Scanner cycle publication vetoed by the cluster activity guardrail" + ); + global_metrics().record_scanner_guardrail_veto(veto.reason()); ScannerCycleActivityStatus::Changed + } else if let Some(moved) = crate::scanner::scanner_peer_dirty_usage_moved(before, &after) { + // Per-bucket certification covers the local dirty map only; + // a peer whose dirty state moved keeps today's whole-cycle + // outcome until scoped acknowledgements can name its buckets. + tracing::info!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "cycle_peer_dirty_usage_moved", + stage = "post_scan_probe", + field = moved.field, + host = %moved.host, + "Scanner cycle superseded because a peer's dirty usage moved during the walk" + ); + global_metrics().record_scanner_guardrail_veto(moved.reason()); + ScannerCycleActivityStatus::Changed + } else { + ScannerCycleActivityStatus::Unchanged }; (status, crate::scanner::scanner_activity_publication_lease_targets(&after)) } @@ -1476,6 +1566,8 @@ pub(crate) struct ScannerCycleResult { pub(crate) status: ScannerCycleStatus, publication_epoch: Option, activity_digest: Option<[u8; 32]>, + activity_baseline: Option>, + pending_dirty_usage_buckets: usize, usage_update_emitted: bool, partial_bucket_progress: bool, dirty_usage_clear: Option, @@ -1495,6 +1587,8 @@ impl ScannerCycleResult { status, publication_epoch: None, activity_digest: None, + activity_baseline: None, + pending_dirty_usage_buckets: 0, usage_update_emitted: false, partial_bucket_progress: false, dirty_usage_clear, @@ -1525,10 +1619,34 @@ impl ScannerCycleResult { self } + #[cfg(test)] pub(crate) fn activity_digest(&self) -> Option<[u8; 32]> { self.activity_digest } + fn with_activity_baseline(mut self, baseline: Arc) -> Self { + self.publication_expectation = None; + self.activity_baseline = Some(baseline); + self + } + + /// The pre-walk activity probe the post-lease guardrail recheck compares + /// against. + pub(crate) fn activity_baseline(&self) -> Option<&crate::scanner::ScannerActivitySnapshot> { + self.activity_baseline.as_deref() + } + + fn with_pending_dirty_usage_buckets(mut self, pending: usize) -> Self { + self.publication_expectation = None; + self.pending_dirty_usage_buckets = pending; + self + } + + /// Buckets a `Complete` cycle left pending for the next cycle. + pub(crate) fn pending_dirty_usage_buckets(&self) -> usize { + self.pending_dirty_usage_buckets + } + pub(crate) fn with_usage_update_emitted(mut self, emitted: bool) -> Self { self.publication_expectation = None; self.usage_update_emitted = emitted; diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 8549ec077..21d2eb286 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -578,8 +578,12 @@ pub(super) fn empty_namespace_usage_candidate( } impl ValidatedUsageCandidate { - pub(super) fn prepare(mut self, status: ScannerCycleStatus) -> (DataUsageInfo, Option) { - self.data.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete); + pub(super) fn prepare( + mut self, + status: ScannerCycleStatus, + pending_buckets: &ScannerCyclePendingBuckets, + ) -> (DataUsageInfo, Option) { + apply_scanner_cycle_publication_marks(&mut self.data, status, pending_buckets); let expectation = if status == ScannerCycleStatus::Complete { struct DigestWriter(Sha256); impl std::io::Write for DigestWriter { diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index b4681e7b7..5e549d6c5 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -1325,21 +1325,41 @@ pub(super) fn dirty_usage_buckets_excluding_failed( .collect() } +/// Plan the per-bucket certification of a walk. The returned snapshot is +/// applied by `clear_dirty_usage_buckets`, which removes only the buckets +/// whose generation still equals the start snapshot: a bucket written during +/// the walk keeps its newer generation and stays pending. The dirty-usage +/// generation therefore does not need to be unchanged; only the guardrail +/// must have passed. pub(super) fn should_clear_dirty_usage_snapshot( result_ok: bool, completed_all_sets: bool, budget_elapsed: bool, - activity_and_generation_current: bool, + guardrail_passed: bool, dirty_buckets: &DirtyUsageBuckets, failed_buckets: &HashSet, ) -> Option { - if result_ok && completed_all_sets && !budget_elapsed && activity_and_generation_current { + if result_ok && completed_all_sets && !budget_elapsed && guardrail_passed { return Some(dirty_usage_buckets_excluding_failed(dirty_buckets, failed_buckets)); } None } +/// Buckets the walk cannot certify against its start snapshot: every dirty +/// bucket whose generation differs from the snapshot's, including buckets +/// dirtied after the snapshot was taken. Only a `Complete` cycle reads this, +/// and a failed bucket never reaches `Complete`, so the failed set that +/// `dirty_usage_buckets_excluding_failed` removes is empty here. +pub(super) fn pending_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) -> BTreeSet { + let dirty_buckets = dirty_usage_buckets(); + dirty_buckets + .iter() + .filter(|(bucket, generation)| snapshot.get(bucket.as_str()) != Some(*generation)) + .map(|(bucket, _)| bucket.clone()) + .collect() +} + pub(super) async fn record_failed_dirty_bucket(failed_buckets: &Arc>>, bucket: &str) { failed_buckets.lock().await.insert(bucket.to_string()); } diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 753b959fa..58e3150e1 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -77,6 +77,8 @@ where service_cohort: None, #[cfg(test)] resolved_scope_observer: None, + #[cfg(test)] + post_walk_gate: None, }; nsscanner_with_storage_status_scoped(store, request).await } @@ -96,6 +98,11 @@ pub(crate) struct ScannerCycleRequest { pub(crate) service_cohort: Option>>, #[cfg(test)] pub(crate) resolved_scope_observer: Option>, + /// Holds the post-walk guardrail probe until the test releases it, so a + /// write issued after `resolved_scope_observer` fired is ordered before + /// certification regardless of how fast the walk finishes. + #[cfg(test)] + pub(crate) post_walk_gate: Option>, } pub(super) struct ScannerBucketScopeResolution<'a> { @@ -287,6 +294,8 @@ where service_cohort, #[cfg(test)] resolved_scope_observer, + #[cfg(test)] + post_walk_gate, } = request; let child_token = ctx.child_token(); let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch); @@ -325,7 +334,7 @@ where let distributed = store.setup_is_dist_erasure().await; let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(store, distributed).await) { - ScannerActivityPreflight::Ready(snapshot) => snapshot, + ScannerActivityPreflight::Ready(snapshot) => Arc::new(snapshot), ScannerActivityPreflight::ActivityBaselineUnavailable(err) => { warn!( target: "rustfs::scanner::io", @@ -442,6 +451,11 @@ where dirty_usage_status, activity_status, ); + let pending_buckets = if status == ScannerCycleStatus::Complete { + pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets) + } else { + ScannerCyclePendingBuckets::new() + }; let Some(candidate) = empty_namespace_usage_candidate( &all_buckets, &expected_sources, @@ -456,11 +470,11 @@ where ) else { return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch)); }; - let (empty_usage, publication_expectation) = candidate.prepare(status); + let (empty_usage, publication_expectation) = candidate.prepare(status, &pending_buckets); let usage_update_emitted = if should_publish_observational_snapshot(status) { publish_observational_snapshot(&updates, empty_usage).await? } else { - publish_usage_snapshot(&updates, status, empty_usage).await? + publish_usage_snapshot(&updates, status, &pending_buckets, empty_usage).await? }; if !usage_update_emitted { return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch)); @@ -484,6 +498,8 @@ where return Ok(ScannerCycleResult::new(status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) + .with_activity_baseline(activity_before) + .with_pending_dirty_usage_buckets(pending_buckets.len()) .with_usage_update_emitted(usage_update_emitted) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) @@ -698,6 +714,10 @@ where record_set_scan_concurrency_limit(0); record_set_scans_queued(0); record_set_scans_active(0); + #[cfg(test)] + if let Some(gate) = post_walk_gate { + let _ = gate.await; + } let first_err = first_err_mutex.lock().await.take(); let has_set_failure = first_err.is_some(); @@ -745,7 +765,6 @@ where !namespace_not_found_buckets.is_empty(), ); let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot); - let dirty_usage_current = dirty_usage_status == DirtyUsageSnapshotStatus::Current; let (activity_status, remote_publication_lease_targets) = scanner_cycle_activity_status(store, distributed, &activity_before).await; let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::>(); @@ -790,14 +809,22 @@ where dirty_usage_status, activity_status, ); + // Certification is per bucket: the pending set is read after the + // guardrail verdict so a bucket written during the walk is reported in + // the same publication that certifies the others. + let pending_buckets = if cycle_status == ScannerCycleStatus::Complete { + pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets) + } else { + ScannerCyclePendingBuckets::new() + }; let mut publication_expectation = None; let usage_update_emitted = if let Some(candidate) = completed_usage { - let (data_usage_info, expectation) = candidate.prepare(cycle_status); + let (data_usage_info, expectation) = candidate.prepare(cycle_status, &pending_buckets); publication_expectation = expectation; if should_publish_observational_snapshot(cycle_status) { publish_observational_snapshot(&updates, data_usage_info).await? } else { - publish_usage_snapshot(&updates, cycle_status, data_usage_info).await? + publish_usage_snapshot(&updates, cycle_status, &pending_buckets, data_usage_info).await? } } else if !ctx.is_cancelled() && let Some((data_usage_info, _)) = observational_usage @@ -810,7 +837,7 @@ where result.is_ok(), structurally_complete_snapshot, budget_elapsed, - activity_status == ScannerCycleActivityStatus::Unchanged && dirty_usage_current, + activity_status == ScannerCycleActivityStatus::Unchanged, &dirty_usage_snapshot.buckets, &failed_buckets, ); @@ -829,6 +856,8 @@ where Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) + .with_activity_baseline(activity_before) + .with_pending_dirty_usage_buckets(pending_buckets.len()) .with_usage_update_emitted(usage_update_emitted) .with_partial_bucket_progress(partial_bucket_progress) .with_remote_publication_lease_targets(remote_publication_lease_targets) diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index e00e994d6..c44a16a33 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/crates/scanner/src/scanner_io/publish_gate_tests.rs @@ -934,12 +934,14 @@ fn completed_data_usage_info_publishes_confirmed_empty_namespace() { } #[test] -fn complete_usage_candidate_with_changed_generation_is_superseded() { +fn complete_usage_candidate_with_changed_generation_is_certified_per_bucket() { let all_buckets = vec!["bucket".to_string()]; let completed_set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); let completed_usage = completed_data_usage_info_for_test(&[completed_set], &all_buckets, false, false); assert!(completed_usage.is_some()); + // A moved dirty generation shades buckets for the next cycle; only a + // guardrail veto supersedes a structurally complete candidate. assert_eq!( classify_nsscanner_cycle( completed_usage.is_some(), @@ -949,6 +951,17 @@ fn complete_usage_candidate_with_changed_generation_is_superseded() { DirtyUsageSnapshotStatus::Changed, ScannerCycleActivityStatus::Unchanged, ), + ScannerCycleStatus::Complete + ); + assert_eq!( + classify_nsscanner_cycle( + completed_usage.is_some(), + false, + false, + ScannerBucketScanStatus::Complete, + DirtyUsageSnapshotStatus::Changed, + ScannerCycleActivityStatus::Changed, + ), ScannerCycleStatus::Superseded ); } diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index c2ba1a13d..018156a4f 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -44,6 +44,7 @@ use temp_env::with_var; use time::OffsetDateTime; use uuid::Uuid; +mod per_bucket_certification; mod scoped_entry_fallback; mod service_cohort; mod usage_expiration; @@ -1318,6 +1319,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work() observed_usage_candidate: None, requires_full_scan, resolved_scope_observer: Some(observer), + post_walk_gate: None, service_cohort: None, }, ), @@ -1408,6 +1410,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( observed_usage_candidate: None, requires_full_scan: false, resolved_scope_observer: None, + post_walk_gate: None, service_cohort: None, }, ), @@ -1461,6 +1464,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( observed_usage_candidate: None, requires_full_scan, resolved_scope_observer: None, + post_walk_gate: None, service_cohort: None, }, ), @@ -1504,17 +1508,20 @@ async fn data_usage_publish_rejects_a_second_terminal_update_without_blocking() ), ] { let (updates, mut receiver) = mpsc::channel(1); + let pending = ScannerCyclePendingBuckets::new(); assert!( - publish_usage_snapshot(&updates, status, data_usage_info) + publish_usage_snapshot(&updates, status, &pending, data_usage_info) .await .expect("first terminal update should be accepted") ); - let err = - tokio::time::timeout(Duration::from_secs(1), publish_usage_snapshot(&updates, status, DataUsageInfo::default())) - .await - .expect("a full terminal update must fail without waiting") - .expect_err("a second terminal update must be rejected"); + let err = tokio::time::timeout( + Duration::from_secs(1), + publish_usage_snapshot(&updates, status, &pending, DataUsageInfo::default()), + ) + .await + .expect("a full terminal update must fail without waiting") + .expect_err("a second terminal update must be rejected"); assert!(err.to_string().contains("already queued")); assert!(receiver.try_recv().is_ok(), "the first terminal update must remain owned by the receiver"); assert!(receiver.try_recv().is_err(), "the rejected second update must not enter the channel"); @@ -2391,7 +2398,7 @@ fn complete_segment_reuse_baseline( } #[test] -fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance() { +fn scoped_scan_requires_a_complete_authoritative_baseline_with_exact_set_provenance() { let source = DataUsageCacheSource::new(1, 2); let expected_sources = HashSet::from([source]); let scan_plan_digest = DataUsageScanPlanDigest([9; 32]); @@ -2409,19 +2416,41 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance( Some(scan_plan_digest) ); - let mut incomplete = serde_json::from_slice::(&baseline).expect("test baseline should decode"); - incomplete.usage_snapshot_converged = Some(false); - let incomplete = bytes::Bytes::from(serde_json::to_vec(&incomplete).expect("test baseline should encode")); + // A complete cycle that left buckets pending publishes unconverged on + // the authoritative object and is still the structural baseline: its + // pending buckets stay dirty and are rescanned, the rest reuse. + let mut pending = serde_json::from_slice::(&baseline).expect("test baseline should decode"); + pending.usage_snapshot_converged = Some(false); + pending.usage_snapshot_pending_buckets = vec!["bucket-a".to_string()]; + pending.usage_snapshot_pending_bucket_count = 1; + let pending = bytes::Bytes::from(serde_json::to_vec(&pending).expect("test baseline should encode")); assert_eq!( complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { - authoritative_data: Some(&incomplete), + authoritative_data: Some(&pending), observed_candidate_data: None, expected_sources: &expected_sources, leader_epoch: 11, want_cycle: 8, scan_plan_digest, }), - None + Some(scan_plan_digest) + ); + + let mut observational = serde_json::from_slice::(&baseline).expect("test baseline should decode"); + observational.usage_snapshot_converged = Some(false); + observational.usage_snapshot_observational = true; + let observational = bytes::Bytes::from(serde_json::to_vec(&observational).expect("test baseline should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { + authoritative_data: Some(&observational), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }), + None, + "an observational snapshot in the authoritative slot is not a baseline" ); let mut wrong_provenance = serde_json::from_slice::(&baseline).expect("test baseline should decode"); @@ -2528,10 +2557,18 @@ fn scoped_scan_accepts_only_a_complete_observation_tied_to_the_authoritative_bas observation_of_nonconverged_authoritative.usage_snapshot_set_states[0].scanner_cycle = Some(8); observation_of_nonconverged_authoritative.usage_snapshot_authoritative_baseline = Some(nonconverged_authoritative.snapshot_identity()); + observation_of_nonconverged_authoritative.usage_snapshot_observational = true; + let mut observational_in_authoritative_slot = nonconverged_authoritative.clone(); + observational_in_authoritative_slot.usage_snapshot_observational = true; let nonconverged_authoritative = bytes::Bytes::from(serde_json::to_vec(&nonconverged_authoritative).expect("nonconverged baseline should encode")); + let observational_in_authoritative_slot = bytes::Bytes::from( + serde_json::to_vec(&observational_in_authoritative_slot).expect("observational baseline should encode"), + ); let observation_of_nonconverged_authoritative = bytes::Bytes::from(serde_json::to_vec(&observation_of_nonconverged_authoritative).expect("observation should encode")); + // The unconverged authoritative snapshot is itself the baseline, so the + // observation chained on it is not consulted. assert_eq!( complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { authoritative_data: Some(&nonconverged_authoritative), @@ -2541,6 +2578,19 @@ fn scoped_scan_accepts_only_a_complete_observation_tied_to_the_authoritative_bas want_cycle: 9, scan_plan_digest, }), + Some(scan_plan_digest) + ); + // An observational snapshot that somehow sits in the authoritative slot + // is neither a baseline nor an anchor for a chained observation. + assert_eq!( + complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { + authoritative_data: Some(&observational_in_authoritative_slot), + observed_candidate_data: Some(&observation_of_nonconverged_authoritative), + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 9, + scan_plan_digest, + }), None ); } @@ -3583,6 +3633,8 @@ fn scanner_cycle_status_requires_a_clean_complete_snapshot() { ), ScannerCycleStatus::Complete ); + // Dirty-usage movement during the walk shades buckets; it is certified + // per bucket and no longer supersedes a walk the guardrail accepted. assert_eq!( classify_nsscanner_cycle( true, @@ -3592,19 +3644,22 @@ fn scanner_cycle_status_requires_a_clean_complete_snapshot() { DirtyUsageSnapshotStatus::Changed, ScannerCycleActivityStatus::Unchanged, ), - ScannerCycleStatus::Superseded - ); - assert_eq!( - classify_nsscanner_cycle( - true, - false, - false, - ScannerBucketScanStatus::Complete, - DirtyUsageSnapshotStatus::Current, - ScannerCycleActivityStatus::Changed, - ), - ScannerCycleStatus::Superseded + ScannerCycleStatus::Complete ); + for dirty_usage_status in [DirtyUsageSnapshotStatus::Current, DirtyUsageSnapshotStatus::Changed] { + assert_eq!( + classify_nsscanner_cycle( + true, + false, + false, + ScannerBucketScanStatus::Complete, + dirty_usage_status, + ScannerCycleActivityStatus::Changed, + ), + ScannerCycleStatus::Superseded, + "a guardrail veto supersedes regardless of dirty state ({dirty_usage_status:?})" + ); + } assert_eq!( classify_nsscanner_cycle( true, @@ -3677,11 +3732,26 @@ fn checkpoint_fixture_superseded_is_distinct_from_partial_and_cancel() { cancelled, bucket, DirtyUsageSnapshotStatus::Changed, - ScannerCycleActivityStatus::Unchanged + ScannerCycleActivityStatus::Changed ), expected, ); } + // Budget and cancellation stay `Incomplete` even when the guardrail + // passed and only dirty usage moved. + for (budget, cancelled) in [(true, false), (false, true)] { + assert_eq!( + classify_nsscanner_cycle( + true, + budget, + cancelled, + ScannerBucketScanStatus::Complete, + DirtyUsageSnapshotStatus::Changed, + ScannerCycleActivityStatus::Unchanged + ), + ScannerCycleStatus::Incomplete, + ); + } } #[test] @@ -3714,20 +3784,28 @@ fn unverified_activity_defers_partial_and_floor_cycles() { #[tokio::test] async fn structurally_complete_superseded_cycles_publish_without_claiming_convergence() { - let (updates, mut receiver) = mpsc::channel(2); + let (updates, mut receiver) = mpsc::channel(3); + let no_pending = ScannerCyclePendingBuckets::new(); + let pending = ScannerCyclePendingBuckets::from(["hot-b".to_string(), "hot-a".to_string()]); assert!( - publish_usage_snapshot(&updates, ScannerCycleStatus::Complete, DataUsageInfo::default()) + publish_usage_snapshot(&updates, ScannerCycleStatus::Complete, &no_pending, DataUsageInfo::default()) .await .expect("complete snapshot publication should succeed") ); assert!( - publish_usage_snapshot(&updates, ScannerCycleStatus::Superseded, DataUsageInfo::default()) + publish_usage_snapshot(&updates, ScannerCycleStatus::Complete, &pending, DataUsageInfo::default()) + .await + .expect("complete snapshot with pending buckets should publish") + ); + // A vetoed cycle never carries a pending list even when one is supplied. + assert!( + publish_usage_snapshot(&updates, ScannerCycleStatus::Superseded, &pending, DataUsageInfo::default()) .await .expect("superseded snapshot publication should succeed") ); assert!( - !publish_usage_snapshot(&updates, ScannerCycleStatus::Incomplete, DataUsageInfo::default()) + !publish_usage_snapshot(&updates, ScannerCycleStatus::Incomplete, &no_pending, DataUsageInfo::default()) .await .expect("incomplete snapshot suppression should succeed") ); @@ -3735,28 +3813,33 @@ async fn structurally_complete_superseded_cycles_publish_without_claiming_conver !publish_usage_snapshot( &updates, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), + &no_pending, DataUsageInfo::default(), ) .await .expect("unverified activity suppression should succeed") ); - assert_eq!( - receiver - .recv() - .await - .expect("complete update should be sent") - .usage_snapshot_converged, - Some(true) - ); - assert_eq!( - receiver - .recv() - .await - .expect("superseded update should be sent") - .usage_snapshot_converged, - Some(false) - ); + let converged = receiver.recv().await.expect("complete update should be sent"); + assert_eq!(converged.usage_snapshot_converged, Some(true)); + assert!(!converged.usage_snapshot_observational); + assert!(converged.usage_snapshot_pending_buckets.is_empty()); + assert_eq!(converged.usage_snapshot_pending_bucket_count, 0); + + let certified = receiver + .recv() + .await + .expect("complete update with pending buckets should be sent"); + assert_eq!(certified.usage_snapshot_converged, Some(false)); + assert!(!certified.usage_snapshot_observational, "pending buckets keep the authoritative route"); + assert_eq!(certified.usage_snapshot_pending_buckets, vec!["hot-a".to_string(), "hot-b".to_string()]); + assert_eq!(certified.usage_snapshot_pending_bucket_count, 2); + + let superseded = receiver.recv().await.expect("superseded update should be sent"); + assert_eq!(superseded.usage_snapshot_converged, Some(false)); + assert!(superseded.usage_snapshot_observational); + assert!(superseded.usage_snapshot_pending_buckets.is_empty()); + assert_eq!(superseded.usage_snapshot_pending_bucket_count, 0); } #[tokio::test] @@ -4526,6 +4609,7 @@ async fn replication_maintenance_refreshes_compacted_stats_across_cycles_and_ret requires_full_scan: true, service_cohort: None, resolved_scope_observer: None, + post_walk_gate: None, }, ) .await @@ -4632,6 +4716,7 @@ async fn replication_maintenance_refreshes_compacted_stats_across_cycles_and_ret requires_full_scan: true, service_cohort: None, resolved_scope_observer: None, + post_walk_gate: None, }, ) .await @@ -4773,6 +4858,7 @@ async fn run_full_cycle(store: &Arc, cycle: u64) -> ScannerCycleStatus requires_full_scan: true, resolved_scope_observer: None, service_cohort: None, + post_walk_gate: None, }, ), ) diff --git a/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs b/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs new file mode 100644 index 000000000..3b504c2e6 --- /dev/null +++ b/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs @@ -0,0 +1,245 @@ +// Copyright 2026 RustFS Team +// SPDX-License-Identifier: Apache-2.0 + +//! Per-bucket certification on the single-disk fixture: a bucket written +//! during the walk is reported pending while the other buckets are certified, +//! the publication stays authoritative, and the next cycle is scoped to the +//! pending bucket. A guardrail veto still supersedes the whole cycle. + +use super::*; + +const BUCKETS: [&str; 3] = ["cold-bucket", "hot-bucket", "idle-bucket"]; + +async fn put_fixture_object(store: &Arc, bucket: &str, object: &str) { + let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]); + store + .put_object(bucket, object, &mut reader, &ScannerObjectOptions::default()) + .await + .expect("fixture object should persist"); +} + +struct CycleHooks { + resolved_scope_observer: Option>, + post_walk_gate: Option>, +} + +async fn run_cycle( + store: &Arc, + cycle: u64, + baseline: Option, + requires_full_scan: bool, + hooks: CycleHooks, +) -> (ScannerCycleResult, DataUsageInfo) { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let (updates, mut receiver) = mpsc::channel(1); + let result = tokio::time::timeout( + Duration::from_secs(60), + nsscanner_with_storage_status_scoped( + store.as_ref(), + ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle: cycle, + leader_epoch: 11, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: baseline, + observed_usage_candidate: None, + requires_full_scan, + service_cohort: None, + resolved_scope_observer: hooks.resolved_scope_observer, + post_walk_gate: hooks.post_walk_gate, + }, + ), + ) + .await + .expect("cycle should finish within the test deadline") + .expect("cycle should succeed"); + let snapshot = receiver.recv().await.expect("cycle should publish a snapshot"); + assert!(receiver.recv().await.is_none(), "a cycle publishes one terminal snapshot"); + (result, snapshot) +} + +fn baseline_bytes(snapshot: &DataUsageInfo) -> Bytes { + Bytes::from(serde_json::to_vec(snapshot).expect("snapshot should encode")) +} + +#[tokio::test] +#[serial] +async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_cycle() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in BUCKETS { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + put_fixture_object(&store, bucket, "initial").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + + // Cycle 1: a quiet full walk certifies every bucket. + let (result, first) = run_cycle( + &store, + 1, + None, + true, + CycleHooks { + resolved_scope_observer: None, + post_walk_gate: None, + }, + ) + .await; + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert_eq!(first.usage_snapshot_converged, Some(true)); + assert!(!first.usage_snapshot_observational); + assert!(first.usage_snapshot_pending_buckets.is_empty()); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty(), "a converged cycle clears the dirty map"); + + // Cycle 2: cold and idle are dirty at the start snapshot; hot is written + // after the scope resolved and before the post-walk proof runs. + for bucket in ["cold-bucket", "idle-bucket"] { + put_fixture_object(&store, bucket, "second").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + let (observer, resolved_scope) = tokio::sync::oneshot::channel(); + let (release, gate) = tokio::sync::oneshot::channel(); + let writer_store = store.clone(); + let writer = tokio::spawn(async move { + let scope = resolved_scope.await.expect("the cycle should report its resolved scope"); + put_fixture_object(&writer_store, "hot-bucket", "during-walk").await; + record_dirty_usage_bucket("hot-bucket"); + wait_for_namespace_commit_tails(writer_store.as_ref()).await; + release.send(()).expect("the cycle should still be waiting on the gate"); + scope + }); + let (result, second) = run_cycle( + &store, + 2, + Some(baseline_bytes(&first)), + false, + CycleHooks { + resolved_scope_observer: Some(observer), + post_walk_gate: Some(gate), + }, + ) + .await; + let scope = writer.await.expect("writer task should finish"); + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["cold-bucket".to_string(), "idle-bucket".to_string()])), + "the converged baseline scopes cycle 2 to the buckets dirty at its start" + ); + + assert_eq!( + result.status, + ScannerCycleStatus::Complete, + "a write during the walk no longer supersedes" + ); + assert_eq!(result.pending_dirty_usage_buckets(), 1); + assert!(!second.usage_snapshot_observational, "pending buckets publish on the authoritative route"); + assert_eq!(second.usage_snapshot_converged, Some(false)); + assert_eq!(second.usage_snapshot_pending_buckets, vec!["hot-bucket".to_string()]); + assert_eq!(second.usage_snapshot_pending_bucket_count, 1); + assert!(second.is_complete_bucket_usage_snapshot()); + assert_eq!(second.buckets_usage["cold-bucket"].objects_count, 2); + assert_eq!(second.buckets_usage["idle-bucket"].objects_count, 2); + assert_eq!( + second.buckets_usage["hot-bucket"].objects_count, 1, + "the pending bucket carries its last walked value" + ); + assert!(result.has_dirty_usage_to_acknowledge()); + assert!(result.clear_verified_usage().is_empty()); + let dirty = dirty_usage_buckets_for_tests(); + assert_eq!( + dirty.keys().cloned().collect::>(), + vec!["hot-bucket".to_string()], + "certification clears the unchanged buckets and keeps the written one" + ); + + // Cycle 3: the unconverged authoritative snapshot is still a structural + // baseline, so the walk is scoped to the pending bucket. + let (observer, resolved_scope) = tokio::sync::oneshot::channel(); + let (result, third) = run_cycle( + &store, + 3, + Some(baseline_bytes(&second)), + false, + CycleHooks { + resolved_scope_observer: Some(observer), + post_walk_gate: None, + }, + ) + .await; + let scope = resolved_scope.await.expect("the cycle should report its resolved scope"); + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["hot-bucket".to_string()])), + "the next cycle walks only the pending bucket" + ); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert_eq!(third.usage_snapshot_converged, Some(true)); + assert!(third.usage_snapshot_pending_buckets.is_empty()); + assert_eq!(third.buckets_usage["hot-bucket"].objects_count, 2); + assert_eq!(third.buckets_usage["cold-bucket"].objects_count, 2); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in BUCKETS { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + put_fixture_object(&store, bucket, "initial").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + let vetoes_before = global_metrics().report().await.usage_freshness.guardrail_veto_total; + + // A PUT tail held across the walk flips the local publication barrier, + // which is a guardrail veto regardless of the dirty map. + let pending_commit = ecstore_hold_namespace_commit(store.as_ref()); + assert!(store.scanner_data_usage_publication_blocked().await); + let (result, snapshot) = run_cycle( + &store, + 1, + None, + true, + CycleHooks { + resolved_scope_observer: None, + post_walk_gate: None, + }, + ) + .await; + drop(pending_commit); + + assert_eq!(result.status, ScannerCycleStatus::Superseded); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert!(snapshot.usage_snapshot_observational, "a vetoed cycle routes to the observed object"); + assert_eq!(snapshot.usage_snapshot_converged, Some(false)); + assert!(snapshot.usage_snapshot_pending_buckets.is_empty()); + assert!(snapshot.is_complete_bucket_usage_snapshot()); + assert!(!result.has_dirty_usage_to_acknowledge(), "nothing is certified under a veto"); + assert_eq!(dirty_usage_buckets_for_tests().len(), BUCKETS.len(), "the dirty map is untouched"); + let freshness = global_metrics().report().await.usage_freshness; + assert_eq!(freshness.guardrail_veto_total, vetoes_before + 1); + assert_eq!( + freshness.last_guardrail_veto_reason, + format!("publication_blocked@{}", crate::scanner::LOCAL_SCANNER_ACTIVITY_NODE) + ); + clear_dirty_usage_buckets_for_tests(); +} diff --git a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs index d974bd923..f31ae0f0d 100644 --- a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs +++ b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs @@ -138,6 +138,7 @@ async fn run_entry( requires_full_scan: false, service_cohort: None, resolved_scope_observer: Some(observer), + post_walk_gate: None, }, ), ) @@ -366,7 +367,7 @@ async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker_ let baseline = run_entry(&store, 1, None, true, false, false).await; for (index, kind) in [ "malformed", - "unconverged", + "observational", "missing-set", "wrong-source", "mixed-plan", @@ -378,7 +379,12 @@ async fn scoped_entry_fallback_rejects_invalid_persisted_baseline_at_the_walker_ let mut candidate = baseline.clone(); candidate.usage_snapshot_converged = Some(true); match kind { - "unconverged" => candidate.usage_snapshot_converged = Some(false), + // An unconverged authoritative snapshot is a valid baseline since + // per-bucket certification; only the observational mark rejects. + "observational" => { + candidate.usage_snapshot_converged = Some(false); + candidate.usage_snapshot_observational = true; + } "missing-set" => { candidate.usage_snapshot_set_states.pop(); } diff --git a/crates/scanner/src/scanner_io/tests/service_cohort.rs b/crates/scanner/src/scanner_io/tests/service_cohort.rs index deb1da547..1deb5d6d1 100644 --- a/crates/scanner/src/scanner_io/tests/service_cohort.rs +++ b/crates/scanner/src/scanner_io/tests/service_cohort.rs @@ -73,6 +73,7 @@ async fn run_cohort_cycle( requires_full_scan: false, service_cohort: Some(cohort), resolved_scope_observer: None, + post_walk_gate: None, }, ), ) diff --git a/docs/architecture/scanner-per-bucket-convergence.md b/docs/architecture/scanner-per-bucket-convergence.md index 063fe043d..f1950981a 100644 --- a/docs/architecture/scanner-per-bucket-convergence.md +++ b/docs/architecture/scanner-per-bucket-convergence.md @@ -180,13 +180,23 @@ equality test to: `usage_snapshot_converged` becomes `pending_dirty_buckets == 0`. A `Complete` cycle with pending buckets publishes an authoritative snapshot: certified -buckets are current, pending buckets carry their last certified values in -`buckets_usage`. `DataUsageInfo` in `crates/data-usage/src/data_usage.rs` -gains an additive, map-encoded list of pending bucket names beside -`usage_snapshot_set_states`, so readers can tell a certified value from a -carried one without a schema break. This replaces the behaviour where a -superseded result is published only as an observed snapshot and never -becomes a baseline. +buckets are current, pending buckets carry the value the walk produced, +which is at least as recent as the previous certified value and may miss +the write that shaded the bucket; the next cycle rescans them. `DataUsageInfo` +in `crates/data-usage/src/data_usage.rs` gains additive fields beside +`usage_snapshot_set_states`: the sorted list `usage_snapshot_pending_buckets`, +its length `usage_snapshot_pending_bucket_count`, and the routing mark +`usage_snapshot_observational` that a vetoed cycle sets so the observed +object is selected by an explicit mark rather than by the convergence flag. +Readers can tell a certified value from an uncertified one without a schema +break. This replaces the behaviour where a superseded result is published +only as an observed snapshot and never becomes a baseline. + +The pending set is read after the guardrail verdict and before publication. +Certification at finalize still compares each bucket's generation against the +start snapshot, so a write that lands between publication and finalize keeps +its bucket dirty even though the published list does not name it; it is the +next cycle's work, exactly as a write after a converged publication is today. ### Guardrail @@ -197,7 +207,7 @@ walk. Field roles: |---|---| | `instance_id`, `protocol_version`, `topology_digest` | veto: a peer restarted, changed version, or the topology changed | | `data_movement_active`, `movement_generation`, `publication_blocked` | veto: rebalance, decommission, or a pending PUT tail can make the walk describe a different generation | -| `maintenance_generation` | veto only when the cycle published maintenance results; otherwise a shade signal | +| `maintenance_generation` | veto: no cycle today records whether it published maintenance results, and the field is already part of the structural digest; it becomes a shade signal only once that distinction exists | | `namespace_generation`, `dirty_usage_generation`, `dirty_usage_pending` | shade signal: consumed per bucket; never a veto on their own | `scanner_activity_structural_digest` already separates the structural part @@ -217,6 +227,13 @@ The guardrail is bounded in three ways: movement or topology veto, which describes the whole heap, defers the cycle. `ActivityBaselineUnavailable` therefore becomes a per-peer condition, not a cycle outcome, unless no peer answered at all. + Phase 2 ships attribution but not localisation: per-bucket certification + covers the local dirty map, and a remote peer whose `dirty_usage_generation` + or `dirty_usage_pending` moved during the walk still supersedes the whole + cycle (`scanner_peer_dirty_usage_moved`), because without a post-walk scoped + snapshot the coordinator cannot name the buckets that peer dirtied and would + otherwise claim convergence over them. The same rule is applied again after + remote leases are granted. Phase 5 lifts it with scoped acknowledgements. - **Retried on a bound.** A deferred cycle retries on the existing superseded backoff schedule with a cap, and `usage_freshness` reports the field and peer that caused the last deferral so the operator can tell a @@ -335,7 +352,7 @@ require a scanner reset. | Artifact | Encoding and tolerance today | Rule in this design | Rollback behaviour | |---|---|---|---| -| `.usage.v2.json` (`DataUsageInfo`) | JSON; every optional field is `serde(default)` and skipped when empty; unknown fields are ignored | The pending-bucket list is an additive optional field. `usage_snapshot_converged` keeps its meaning. No existing field changes type or semantics. | An older reader ignores the pending list and sees a complete snapshot with `usage_snapshot_converged = false`, which it already handles as an authoritative but unconverged publication. | +| `.usage.v2.json` (`DataUsageInfo`) | JSON; every optional field is `serde(default)` and skipped when empty; unknown fields are ignored | The pending-bucket list, its count, and the observational routing mark are additive optional fields. `usage_snapshot_converged` keeps its meaning. No existing field changes type or semantics. | An older reader ignores the pending list and sees a complete snapshot with `usage_snapshot_converged = false`. Its admin and quota readers accept that as an authoritative publication; its scanner does not use it as a scoped-scan baseline and runs full walks until it publishes a converged snapshot of its own, which is today's behaviour, not a reset. | | `.usage-cache.bin` (`DataUsageCache`) | MessagePack; `DataUsageScanIdentity`, `DataUsageScanProgress`, and `DataUsageRawEnumerationCursor` deny unknown fields; the map-encoded cache metadata tolerates unknown keys | No field is added to a struct that denies unknown fields. Bucket color and start generation are process state in the dirty map, not cache fields. If a per-bucket value must persist, it goes into the map-encoded metadata as `scan_execution_digest` did. The old-generation sweep reuses the existing checkpoint, resume, and cursor fields unchanged. | An older reader ignores any added map key and treats the cache under its existing identity rules. | | `.bloomcycle.bin` and recovery markers | Versioned by `schema_version`; a mismatch is quarantined | No schema bump. Pending-bucket state is derived from the dirty map and the published snapshot, not stored in cycle state. | Unchanged. | | `scanner/durable-dirty-producer-replay/.json` | Strict schema; entries deny unknown fields | The barrier-hit class is not persisted. Replayed entries are treated as `Namespace` class, which can only cause an extra scoped rescan, never a missed one. | Unchanged. | @@ -416,11 +433,15 @@ condition before merging. ## Surfaces - `usage_freshness` in the admin scanner status reports - `pending_dirty_buckets` (`Namespace` class only), plus the pacer target, + `dirty_pending_buckets` (`Namespace` class only; after certification the + dirty map holds exactly the pending set), the attributed guardrail veto + (`guardrail_veto_total`, `last_guardrail_veto_unix_secs`, + `last_guardrail_veto_reason` as `@`), plus the pacer target, achieved rate, and whether a ceiling is binding. The console marks usage stale only for that count or for a guardrail veto. -- `last_cycle_result` reports `complete` for a cycle with pending buckets; - `superseded` is reserved for guardrail vetoes and should be rare. +- `last_cycle_result` reports `success`, its existing label for a completed + cycle, for a cycle with pending buckets; `superseded` is reserved for + guardrail vetoes and the phase-2 peer fallback and should be rare. - The superseded retry schedule applies only to guardrail vetoes and to explicit full sweeps. A `Complete` cycle with pending buckets uses the ordinary dirty-usage wake and the next cycle is scoped to the pending set. @@ -443,7 +464,10 @@ modes. non-converged `Complete` publication as the structural baseline, order grey buckets by waiting age under a per-cycle quantum, give each prefix inside a bucket its own quantum, and attribute and localise guardrail - vetoes. Oracle: a single-node deployment with one bucket written every + vetoes. The first change of this phase ships the termination rule: + certification, `Complete` with pending buckets, the baseline bootstrap + and guardrail attribution. Bounded remark, aging order, the per-prefix + quantum and guardrail localisation follow as their own changes. Oracle: a single-node deployment with one bucket written every few seconds reaches `Complete` on the first full walk, reports exactly that bucket pending, and the next cycle walks only that bucket; a deployment with more buckets than the quantum walks every bucket within diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index f712cbbbc..80232185b 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/docs/architecture/scanner-usage-publication.md @@ -29,10 +29,15 @@ rename work ends, including when the request caller is cancelled. Scan admission remains movement-only so sustained PUTs do not stop namespace walks and scanner-driven lifecycle discovery. The post-walk local publication check and remote publication leases reject pending fanouts. Begin/end namespace generations -invalidate scans and cached plans across the fanout; after acquiring remote -leases, the coordinator rechecks the full activity digest before publishing an +invalidate cached plans across the fanout; after acquiring remote leases, the +coordinator rechecks the guardrail fields of the activity snapshot (the +structural digest and the movement fields, plus the peers' dirty-usage state +while per-bucket certification is local only) before publishing an authoritative aggregate. This catches a tail that finishes between the scan's -last probe and lease acquisition. +last probe and lease acquisition. Local namespace and dirty-usage generation +movement in that window is not a veto: its buckets stay in the dirty map and +are the next cycle's work, as described in +[scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md). This adds no namespace or movement lock. An already-verified older snapshot may still precede a newly started write. Sustained or stalled PUT tails can delay @@ -160,7 +165,11 @@ requires a compatibility window and a dedicated cleanup entry. Observed snapshots are a diagnostic and availability layer. They may be served only when the snapshot explicitly reports that it is partial or observational, and only to consumers that do not make hard quota, durability, or deletion -decisions from it. Admin usage views may expose this state with completeness +decisions from it. The scanner routes a snapshot to the observed object by the +explicit `usage_snapshot_observational` mark that a vetoed or unverifiable cycle +sets, not by `usage_snapshot_converged`: a complete cycle that certified every +bucket but the ones written during its walk publishes an unconverged +authoritative snapshot that names those buckets as pending. Admin usage views may expose this state with completeness flags so operators can see progress while the authoritative publication is blocked. Quota enforcement must not use an observed snapshot as the current usage authority.