opensource/rustfs
3
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 09:17:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): harden heal release gates (#7614)

Browse Source
This commit is contained in:
houseme
2026-09-10 12:03:55 +08:00
committed by GitHub
parent 70fb8bf504
commit 358bf6832d
13 changed files with 466 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/lib.rs
+3 -3
View File
@@ -96,14 +96,14 @@ pub use scanner::{
scanner_topology_digest,
};
pub use scanner_io::{
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageSnapshot,
ScannerDirtyUsageState, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError,
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,
ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError,
ScannerDurableDirtyUsageReplayRecord, ScannerDurableDirtyUsageReplayScope, acknowledge_dirty_usage_generation,
acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, encode_durable_dirty_usage_producer_replay_record,
record_dirty_usage_bucket, record_dirty_usage_bucket_from_producer, record_dirty_usage_bucket_from_producers,
record_dirty_usage_object, record_dirty_usage_object_from_producer, record_scanner_maintenance_change,
replay_durable_dirty_usage_producer_record, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer,
scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer, set_scanner_dirty_usage_mutation_observer,
};
pub use segment_invalidation::SegmentInvalidationProducerIdentity;
pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
crates/scanner/src/scanner/tests/scoped_ack_publication.rs
+18 -1
View File
@@ -432,7 +432,12 @@ async fn scoped_ack_publication_stale_baseline_cannot_prove_a_replaced_root() {
#[tokio::test]
#[serial]
async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish() {
for mutation in ["remote_ack_target", "publication_epoch", "remote_lease_targets"] {
for mutation in [
"remote_ack_target",
"remote_scoped_ack_target",
"publication_epoch",
"remote_lease_targets",
] {
let (_directory, store) = candidate_store().await;
let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
let baseline = read_data_usage_persist_baseline(store.clone())
@@ -465,6 +470,18 @@ async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish
instance_id: crate::scanner_activity_epoch().to_string(),
kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Generation(changed_generation),
}]),
"remote_scoped_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement {
host: "proof-peer:9000".to_string(),
instance_id: crate::scanner_activity_epoch().to_string(),
kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped {
owner_id: crate::scanner_activity_epoch().to_string(),
entries: vec![crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
bucket: PROOF_BUCKET.to_string(),
bucket_incarnation: uuid::Uuid::from_u128(0x11111111111111111111111111111111),
generation: changed_generation,
}],
},
}]),
"publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
"remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(
"proof-peer:9000".to_string(),
crates/scanner/src/scanner_io.rs
+62 -18
View File
@@ -306,7 +306,7 @@ fn scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(
fn resolve_remote_dirty_usage_scope(
requested_scope: ScannerBucketScanScope,
mut dirty_buckets: HashSet<String>,
dirty_usage_snapshot: &DirtyUsageSnapshot,
remote_dirty_usage: VerifiedRemoteDirtyUsage,
all_buckets: &[BucketInfo],
baseline_proof: ScannerCacheBaselineProof<'_>,
@@ -319,12 +319,21 @@ fn resolve_remote_dirty_usage_scope(
let peer_count = remote_dirty_usage.peer_count;
let dirty_peer_count = remote_dirty_usage.dirty_peer_count;
let remote_dirty_buckets = remote_dirty_usage.dirty_buckets.clone();
let mut dirty_buckets = dirty_usage_snapshot.buckets.keys().cloned().collect::<HashSet<_>>();
dirty_buckets.extend(remote_dirty_usage.dirty_buckets);
// Peer snapshots contribute bucket names only; the local prefix scopes
// would narrow a bucket a peer dirtied elsewhere, so the merged scope
// stays at bucket granularity (same rule as the local fallthrough).
let scope =
scoped_scan_scope_from_dirty_buckets(requested_scope, dirty_buckets, None, true, false, all_buckets, baseline_proof);
let scope = scoped_scan_scope_from_dirty_buckets(
requested_scope.clone(),
dirty_buckets.clone(),
None,
true,
false,
all_buckets,
baseline_proof,
);
if scope.is_default() {
return default_result(scope);
}
@@ -358,19 +367,45 @@ fn resolve_remote_dirty_usage_scope(
}
let has_scoped_acknowledgements = !scoped_acknowledgements.is_empty();
let distributed_segment_invalidation_evidence =
(dirty_peer_count > 0 && has_scoped_acknowledgements).then_some(DistributedSegmentInvalidationEvidence {
invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc,
distributed_ec_invalidation: true,
peer_count,
dirty_peer_count,
same_window_remote_proof: true,
all_peers_bound_to_generation_window: true,
});
let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_baseline_with_evidence(
dirty_usage_snapshot,
true,
baseline_proof,
distributed_segment_invalidation_evidence,
);
let scope = if segment_reuse_activation_preflight.scanner_segment_reuse_activated {
let local_only_scopes = dirty_usage_snapshot
.scopes
.iter()
.filter(|(bucket, _)| !remote_dirty_buckets.contains(bucket.as_str()))
.map(|(bucket, scope)| (bucket.clone(), scope.clone()))
.collect::<DirtyUsageBucketScopes>();
scoped_scan_scope_from_dirty_buckets(
requested_scope,
dirty_buckets,
(!local_only_scopes.is_empty()).then_some(&local_only_scopes),
true,
true,
all_buckets,
baseline_proof,
)
} else {
scope
};
ScannerBucketScopeResolutionResult {
scope,
remote_dirty_usage_acknowledgements: scoped_acknowledgements,
distributed_segment_invalidation_evidence: (dirty_peer_count > 0 && has_scoped_acknowledgements).then_some(
DistributedSegmentInvalidationEvidence {
invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc,
distributed_ec_invalidation: true,
peer_count,
dirty_peer_count,
same_window_remote_proof: true,
all_peers_bound_to_generation_window: true,
},
),
distributed_segment_invalidation_evidence,
}
}
@@ -596,7 +631,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle(
production_activation: true,
producer_identity_coverage_complete: dirty_usage_producer_evidence.producer_identity_coverage_complete,
durable_producer_identity: dirty_usage_producer_evidence.durable_producer_identity,
durable_dirty_producer_journal: dirty_usage_producer_evidence.durable_producer_identity,
durable_dirty_producer_journal: dirty_usage_producer_evidence.durable_dirty_producer_journal,
restart_gap_absent: dirty_usage_producer_evidence.restart_gap_absent,
generation_window_bound: dirty_usage_snapshot.covers_all_pending
&& dirty_usage_snapshot.generation != 0
@@ -616,6 +651,15 @@ fn scanner_segment_reuse_activation_preflight_for_baseline(
dirty_usage_snapshot: &DirtyUsageSnapshot,
distributed: bool,
baseline_proof: ScannerCacheBaselineProof<'_>,
) -> ScannerSegmentReuseActivationPreflight {
scanner_segment_reuse_activation_preflight_for_baseline_with_evidence(dirty_usage_snapshot, distributed, baseline_proof, None)
}
fn scanner_segment_reuse_activation_preflight_for_baseline_with_evidence(
dirty_usage_snapshot: &DirtyUsageSnapshot,
distributed: bool,
baseline_proof: ScannerCacheBaselineProof<'_>,
distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
) -> ScannerSegmentReuseActivationPreflight {
let (dirty_usage_producer_evidence, cold_zero_walk_oracle) =
scanner_segment_reuse_baseline_producer_evidence(dirty_usage_snapshot, baseline_proof);
@@ -623,7 +667,7 @@ fn scanner_segment_reuse_activation_preflight_for_baseline(
dirty_usage_snapshot,
dirty_usage_producer_evidence,
distributed,
None,
distributed_segment_invalidation_evidence,
cold_zero_walk_oracle,
)
}
@@ -1568,14 +1612,14 @@ pub(crate) use cache::{
current_cache_root_or_prepare_with_generation,
};
pub use dirty_usage::{
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageSnapshot,
ScannerDirtyUsageState, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError,
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,
ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError,
ScannerDurableDirtyUsageReplayRecord, ScannerDurableDirtyUsageReplayScope, acknowledge_dirty_usage_generation,
acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, encode_durable_dirty_usage_producer_replay_record,
record_dirty_usage_bucket, record_dirty_usage_bucket_from_producer, record_dirty_usage_bucket_from_producers,
record_dirty_usage_object, record_dirty_usage_object_from_producer, record_scanner_maintenance_change,
replay_durable_dirty_usage_producer_record, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state,
scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer,
scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer, set_scanner_dirty_usage_mutation_observer,
};
#[cfg(test)]
pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests};
crates/scanner/src/scanner_io/dirty_usage.rs
+93 -5
View File
@@ -30,6 +30,8 @@ pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock<StdMutex<DirtyUsageP
LazyLock::new(|| StdMutex::new(BTreeMap::new()));
static DIRTY_USAGE_CLEAR_OBSERVER: LazyLock<StdRwLock<Option<ScannerDirtyUsageClearObserver>>> =
LazyLock::new(|| StdRwLock::new(None));
static DIRTY_USAGE_MUTATION_OBSERVER: LazyLock<StdRwLock<Option<ScannerDirtyUsageMutationObserver>>> =
LazyLock::new(|| StdRwLock::new(None));
pub(super) static DIRTY_USAGE_PRODUCER_COVERAGE: AtomicU64 = AtomicU64::new(0);
pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
@@ -53,6 +55,8 @@ pub struct ScannerDirtyUsageBucket {
}
pub type ScannerDirtyUsageClearObserver = Arc<dyn Fn(Vec<ScannerDirtyUsageBucket>) + Send + Sync + 'static>;
pub type ScannerDirtyUsageMutationObserver =
Arc<dyn Fn(&str, &str, crate::segment_invalidation::SegmentInvalidationProducerIdentity) + Send + Sync + 'static>;
/// A non-durable optimization hint for a dirty bucket.
///
@@ -83,6 +87,7 @@ pub(super) type DirtyUsageProducerIdentities = BTreeMap<String, DirtyUsageProduc
pub(super) struct DirtyUsageProducerEvidence {
pub(super) producer_identity_coverage_complete: bool,
pub(super) durable_producer_identity: bool,
pub(super) durable_dirty_producer_journal: bool,
pub(super) restart_gap_absent: bool,
pub(super) generation_window_bound: bool,
pub(super) generation_start: u64,
@@ -112,6 +117,15 @@ pub fn set_scanner_dirty_usage_clear_observer(
std::mem::replace(&mut *slot, observer)
}
pub fn set_scanner_dirty_usage_mutation_observer(
observer: Option<ScannerDirtyUsageMutationObserver>,
) -> Option<ScannerDirtyUsageMutationObserver> {
let mut slot = DIRTY_USAGE_MUTATION_OBSERVER
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
std::mem::replace(&mut *slot, observer)
}
fn notify_dirty_usage_clear(cleared: Vec<ScannerDirtyUsageBucket>) {
if cleared.is_empty() {
return;
@@ -125,6 +139,30 @@ fn notify_dirty_usage_clear(cleared: Vec<ScannerDirtyUsageBucket>) {
}
}
fn notify_dirty_usage_mutation(
bucket: &str,
object: &str,
producers: &[crate::segment_invalidation::SegmentInvalidationProducerIdentity],
) {
if bucket.is_empty() {
return;
}
let observer = DIRTY_USAGE_MUTATION_OBSERVER
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone();
let Some(observer) = observer else {
return;
};
if producers.is_empty() {
observer(bucket, object, crate::segment_invalidation::SegmentInvalidationProducerIdentity::Unknown);
} else {
for producer in producers {
observer(bucket, object, *producer);
}
}
}
/// A point-in-time view of the local dirty bucket generations.
///
/// `complete == false` is an all-or-nothing overflow signal: `buckets` is
@@ -566,6 +604,7 @@ mod scoped_dirty_usage_tests {
let evidence = dirty_usage_producer_evidence(&snapshot);
assert!(evidence.producer_identity_coverage_complete);
assert!(evidence.durable_producer_identity);
assert!(evidence.durable_dirty_producer_journal);
assert!(evidence.restart_gap_absent);
assert_eq!(evidence.generation_start, 7);
assert_eq!(evidence.generation_end, 7);
@@ -583,10 +622,45 @@ mod scoped_dirty_usage_tests {
);
let changed_evidence = dirty_usage_producer_evidence(&changed);
assert!(!changed_evidence.durable_producer_identity);
assert!(!changed_evidence.durable_dirty_producer_journal);
assert!(!changed_evidence.restart_gap_absent);
clear_dirty_usage_buckets_for_tests();
}
#[test]
#[serial]
fn dirty_usage_mutation_observer_tracks_typed_and_conservative_events() {
use std::sync::{Arc, Mutex};
clear_dirty_usage_buckets_for_tests();
let observed = Arc::new(Mutex::new(Vec::new()));
let observed_clone = observed.clone();
let previous = set_scanner_dirty_usage_mutation_observer(Some(Arc::new(move |bucket, object, producer| {
observed_clone.lock().expect("observer lock should not be poisoned").push((
bucket.to_string(),
object.to_string(),
producer,
));
})));
record_dirty_usage_object_from_producer("photos", "2026/image.jpg", SegmentInvalidationProducerIdentity::PutObject);
record_dirty_usage_bucket("archive");
set_scanner_dirty_usage_mutation_observer(previous);
assert_eq!(
*observed.lock().expect("observer lock should not be poisoned"),
vec![
(
"photos".to_string(),
"2026/image.jpg".to_string(),
SegmentInvalidationProducerIdentity::PutObject,
),
("archive".to_string(), String::new(), SegmentInvalidationProducerIdentity::Unknown,),
]
);
clear_dirty_usage_buckets_for_tests();
}
#[test]
#[serial]
fn durable_dirty_usage_replay_rejects_invalid_records_without_partial_state() {
@@ -745,6 +819,7 @@ fn record_dirty_usage_bucket_inner<I>(bucket: &str, producers: I)
where
I: IntoIterator<Item = crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
{
let producers = producers.into_iter().collect::<Vec<_>>();
let pending_buckets = {
let mut dirty_buckets = dirty_usage_buckets();
let mut dirty_scopes = dirty_usage_bucket_scopes();
@@ -752,7 +827,12 @@ where
let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION);
dirty_buckets.insert(bucket.to_string(), generation);
dirty_scopes.insert(bucket.to_string(), DirtyUsageBucketScope::WholeBucket);
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, bucket, generation, producers);
record_segment_invalidation_producer_identities_for_generation(
&mut producer_identities,
bucket,
generation,
producers.iter().copied(),
);
dirty_buckets.len()
};
global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
@@ -760,6 +840,7 @@ where
// admin/console consumers never ride the full TTL after a change
// (rustfs/backlog#1872).
crate::prefix_usage::invalidate_prefix_usage_cache(bucket);
notify_dirty_usage_mutation(bucket, "", &producers);
DIRTY_USAGE_BUCKET_NOTIFY.notify_one();
}
@@ -817,11 +898,17 @@ where
if overflowed {
*scope = DirtyUsageBucketScope::WholeBucket;
}
record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, bucket, generation, producers);
record_segment_invalidation_producer_identities_for_generation(
&mut producer_identities,
bucket,
generation,
producers.iter().copied(),
);
dirty_buckets.len()
};
global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets));
crate::prefix_usage::invalidate_prefix_usage_cache(bucket);
notify_dirty_usage_mutation(bucket, object, &producers);
DIRTY_USAGE_BUCKET_NOTIFY.notify_one();
}
@@ -1193,7 +1280,7 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di
&& DIRTY_USAGE_PRODUCER_COVERAGE.load(Ordering::Acquire)
& crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK
== crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK;
let durable_producer_identity = producer_identity_coverage_complete
let durable_dirty_producer_journal = producer_identity_coverage_complete
&& snapshot
.buckets
.keys()
@@ -1205,8 +1292,9 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di
DirtyUsageProducerEvidence {
producer_identity_coverage_complete,
durable_producer_identity,
restart_gap_absent: durable_producer_identity,
durable_producer_identity: durable_dirty_producer_journal,
durable_dirty_producer_journal,
restart_gap_absent: durable_dirty_producer_journal,
generation_window_bound,
generation_start: if generation_window_bound { generation_start } else { 0 },
generation_end: if generation_window_bound { generation_end } else { 0 },
crates/scanner/src/scanner_io/io_cycle.rs
+1 -1
View File
@@ -190,7 +190,7 @@ where
};
let remote_resolution = resolve_remote_dirty_usage_scope(
resolution.requested_scope,
dirty_buckets,
resolution.dirty_usage_snapshot,
remote_dirty_usage,
resolution.all_buckets,
resolution.baseline_proof,
crates/scanner/src/scanner_io/tests.rs
+151 -5
View File
@@ -37,7 +37,7 @@ use rustfs_concurrency::{
};
use rustfs_filemeta::FileInfo;
use serial_test::serial;
use std::collections::BTreeMap;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use temp_env::with_var;
use time::OffsetDateTime;
@@ -285,6 +285,7 @@ fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_
assert!(durable_evidence.producer_identity_coverage_complete);
assert!(durable_evidence.durable_producer_identity);
assert!(!durable_evidence.durable_dirty_producer_journal);
assert!(durable_evidence.restart_gap_absent);
let mut stale_epoch = results.clone();
@@ -297,12 +298,14 @@ fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_
let stale_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &stale_epoch, &expected_sources);
assert!(stale_evidence.producer_identity_coverage_complete);
assert!(!stale_evidence.durable_producer_identity);
assert!(!stale_evidence.durable_dirty_producer_journal);
assert!(!stale_evidence.restart_gap_absent);
record_dirty_usage_bucket("videos");
let changed_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources);
assert!(!changed_evidence.producer_identity_coverage_complete);
assert!(!changed_evidence.durable_producer_identity);
assert!(!changed_evidence.durable_dirty_producer_journal);
assert!(!changed_evidence.restart_gap_absent);
clear_dirty_usage_buckets_for_tests();
}
@@ -316,6 +319,19 @@ fn scanner_segment_reuse_activation_replays_cold_durable_baseline() {
for producer in SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION {
record_dirty_usage_object_from_producer("photos", "2026/object", producer);
}
let replay_generation = dirty_usage_generation();
replay_durable_dirty_usage_producer_record(
&encode_durable_dirty_usage_producer_replay_record(vec![ScannerDurableDirtyUsageReplayEntry {
bucket: "photos".to_string(),
generation: replay_generation,
scope: ScannerDurableDirtyUsageReplayScope::TopLevelEntries {
entries: BTreeSet::from(["2026".to_string()]),
},
producers: SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION.into_iter().collect(),
}])
.expect("durable producer replay should encode"),
)
.expect("durable producer replay should restore restart authority");
let dirty_usage_snapshot =
snapshot_dirty_usage_buckets(&[bucket_info("photos"), bucket_info("archive")], dirty_usage_generation());
let mut segment_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot)
@@ -413,8 +429,11 @@ fn scanner_segment_reuse_activation_replays_cold_durable_baseline() {
scan_plan_digest,
},
);
assert!(preflight.scanner_segment_reuse_activated);
assert_eq!(preflight.fail_closed_blockers().collect::<Vec<_>>(), Vec::<&str>::new());
assert!(!preflight.scanner_segment_reuse_activated);
assert_eq!(
preflight.fail_closed_blockers().collect::<Vec<_>>(),
vec!["missing_durable_journal_replay"]
);
record_dirty_usage_bucket("photos");
let unidentified_snapshot =
@@ -473,6 +492,7 @@ fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence {
DirtyUsageProducerEvidence {
producer_identity_coverage_complete: true,
durable_producer_identity: false,
durable_dirty_producer_journal: false,
restart_gap_absent: false,
generation_window_bound: true,
generation_start: 7,
@@ -1468,6 +1488,7 @@ fn dirty_usage_producer_evidence_tracks_process_local_coverage_without_durable_r
assert!(evidence.generation_window_bound);
assert!(evidence.producer_identity_coverage_complete);
assert!(!evidence.durable_producer_identity);
assert!(!evidence.durable_dirty_producer_journal);
assert!(!evidence.restart_gap_absent);
assert_eq!(evidence.generation_start, snapshot.buckets["photos"]);
assert_eq!(evidence.generation_end, snapshot.buckets["photos"]);
@@ -1858,6 +1879,44 @@ fn complete_usage_baseline(
bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode"))
}
fn complete_segment_reuse_baseline(
source: DataUsageCacheSource,
scan_plan_digest: DataUsageScanPlanDigest,
scanner_cycle: u64,
scanner_epoch: u64,
evidence: DirtyUsageProducerEvidence,
buckets: &[&str],
) -> bytes::Bytes {
let mut proof = evidence
.segment_invalidation_proof()
.expect("durable producer evidence should produce segment proof");
proof.cold_zero_walk_oracle = true;
let baseline = DataUsageInfo {
last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
scanner_cycle: Some(scanner_cycle),
scanner_epoch: Some(scanner_epoch),
buckets_count: u64::try_from(buckets.len()).expect("test bucket count should fit"),
buckets_usage: buckets
.iter()
.map(|bucket| ((*bucket).to_string(), Default::default()))
.collect(),
usage_snapshot_complete: true,
usage_snapshot_converged: Some(true),
usage_snapshot_set_states: vec![DataUsageSnapshotSetState {
pool_index: u64::try_from(source.pool_index).expect("test pool index should fit"),
set_index: u64::try_from(source.set_index).expect("test set index should fit"),
scanner_cycle: Some(scanner_cycle),
scanner_epoch: Some(scanner_epoch),
scan_plan_digest: Some(scan_plan_digest.0),
complete: true,
tombstone: false,
segment_invalidation_proof: Some(proof),
}],
..Default::default()
};
bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode"))
}
#[test]
fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance() {
let source = DataUsageCacheSource::new(1, 2);
@@ -2183,6 +2242,12 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi
"photos".to_string(),
DirtyUsageBucketScope::TopLevelEntries(HashSet::from(["2026".to_string()])),
)]);
let dirty_usage_snapshot = DirtyUsageSnapshot {
buckets: Arc::new(HashMap::from([("photos".to_string(), 7)])),
scopes: Arc::new(dirty_scopes.clone()),
generation: 7,
covers_all_pending: true,
};
let locally_scoped = scoped_scan_scope_from_dirty_buckets(
ScannerBucketScanScope::default(),
HashSet::from(["photos".to_string()]),
@@ -2206,7 +2271,7 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi
let distributed = resolve_remote_dirty_usage_scope(
ScannerBucketScanScope::default(),
HashSet::from(["photos".to_string()]),
&dirty_usage_snapshot,
remote_dirty_usage,
&[bucket_info("photos")],
ScannerCacheBaselineProof {
@@ -2246,6 +2311,82 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi
assert!(evidence.all_peers_bound_to_generation_window);
}
#[test]
#[serial]
fn distributed_segment_reuse_activation_keeps_remote_dirty_buckets_at_bucket_scope() {
clear_dirty_usage_buckets_for_tests();
let source = DataUsageCacheSource::new(1, 2);
let expected_sources = HashSet::from([source]);
let scan_plan_digest = DataUsageScanPlanDigest([9; 32]);
let entries = BTreeSet::from(["2026".to_string()]);
let bytes = encode_durable_dirty_usage_producer_replay_record(vec![ScannerDurableDirtyUsageReplayEntry {
bucket: "photos".to_string(),
generation: 7,
scope: ScannerDurableDirtyUsageReplayScope::TopLevelEntries { entries },
producers: crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION
.iter()
.copied()
.collect(),
}])
.expect("durable dirty usage replay record should encode");
replay_durable_dirty_usage_producer_record(&bytes).expect("durable dirty usage replay should restore producer proof");
let dirty_usage_snapshot =
snapshot_dirty_usage_buckets(&[bucket_info("photos"), bucket_info("archive")], dirty_usage_generation());
let evidence = dirty_usage_producer_evidence(&dirty_usage_snapshot);
let baseline = complete_segment_reuse_baseline(source, scan_plan_digest, 7, 11, evidence, &["photos", "archive"]);
let expected_peers = HashMap::from([(
"node-a:9000".to_string(),
ScannerPeerDirtyUsageExpectation {
instance_id: "instance-a".to_string(),
generation: 3,
pending: true,
},
)]);
let remote_dirty_usage = verified_remote_dirty_usage(
&expected_peers,
vec![(
"node-a:9000".to_string(),
peer_dirty_usage_snapshot("instance-a", 3, true, &[("archive", 3)]),
)],
)
.expect("fixture remote dirty usage should verify at bucket granularity");
let result = resolve_remote_dirty_usage_scope(
ScannerBucketScanScope::default(),
&dirty_usage_snapshot,
remote_dirty_usage,
&[bucket_info("photos"), bucket_info("archive")],
ScannerCacheBaselineProof {
authoritative_data: Some(&baseline),
observed_candidate_data: None,
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
},
);
assert_eq!(
result
.scope
.selected_buckets
.as_deref()
.expect("distributed reuse still selects both dirty buckets"),
&HashSet::from(["photos".to_string(), "archive".to_string()])
);
assert!(
result.scope.prefix_scope_for("photos").is_some(),
"durable local producer proof may activate local segment reuse after distributed ACK capability evidence"
);
assert!(
result.scope.prefix_scope_for("archive").is_none(),
"remote dirty usage has only bucket-granularity evidence and must not be narrowed to a local prefix"
);
assert_eq!(result.remote_dirty_usage_acknowledgements.len(), 1);
assert!(result.distributed_segment_invalidation_evidence.is_some());
clear_dirty_usage_buckets_for_tests();
}
fn peer_dirty_usage_snapshot(
instance_id: &str,
generation: u64,
@@ -2419,7 +2560,12 @@ fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_thresho
let result = resolve_remote_dirty_usage_scope(
ScannerBucketScanScope::default(),
HashSet::new(),
&DirtyUsageSnapshot {
buckets: Arc::new(HashMap::new()),
scopes: Arc::new(HashMap::new()),
generation: 7,
covers_all_pending: true,
},
remote_dirty_usage,
&all_buckets,
ScannerCacheBaselineProof {
crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs
+31
View File
@@ -226,6 +226,34 @@ fn record_segment_dirty_usage(bucket: &str) {
}
}
fn replay_segment_dirty_usage(bucket: &str) {
replay_dirty_usage(
bucket,
ScannerDurableDirtyUsageReplayScope::TopLevelEntries {
entries: BTreeSet::from(["hot-segment".to_string()]),
},
);
}
fn replay_whole_bucket_dirty_usage(bucket: &str) {
replay_dirty_usage(bucket, ScannerDurableDirtyUsageReplayScope::WholeBucket);
}
fn replay_dirty_usage(bucket: &str, scope: ScannerDurableDirtyUsageReplayScope) {
replay_durable_dirty_usage_producer_record(
&encode_durable_dirty_usage_producer_replay_record(vec![ScannerDurableDirtyUsageReplayEntry {
bucket: bucket.to_string(),
generation: dirty_usage_generation(),
scope,
producers: crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION
.into_iter()
.collect(),
}])
.expect("durable segment replay should encode"),
)
.expect("durable segment replay should restore producer authority");
}
// The scoped fallback fixture keeps two EC pools and several scan futures live
// at once. Run the async cases on a dedicated stack so Linux libtest defaults
// exercise the assertions instead of aborting before the oracle finishes.
@@ -267,6 +295,7 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_
create_bucket(&store, &hot).await;
create_bucket(&store, &cold).await;
record_segment_dirty_usage(&hot);
replay_segment_dirty_usage(&hot);
let baseline = run_entry(&store, 1, None, true, false, false).await;
persist_baseline(&store, &baseline).await;
@@ -283,6 +312,7 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_
"hot-segment/object",
crate::segment_invalidation::SegmentInvalidationProducerIdentity::PutObject,
);
replay_segment_dirty_usage(&hot);
let usage = run_entry(&store, 3, Some(&hot), true, true, true).await;
assert_eq!(usage.buckets_usage[&hot].objects_count, 2);
assert_eq!(usage.buckets_usage[&cold].objects_count, 1);
@@ -298,6 +328,7 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks_
crate::segment_invalidation::SegmentInvalidationProducerIdentity::PutObject,
);
}
replay_whole_bucket_dirty_usage(&hot);
let usage = run_entry(&store, 4, Some(&hot), true, true, false).await;
assert_eq!(usage.objects_total_count, 3);
rustfs/src/storage/storage_api.rs
+4 -2
View File
@@ -955,7 +955,10 @@ pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperti
pub(crate) async fn init_background_replication(store: Arc<ECStore>) {
let durable_dirty_usage_journal = super::scanner_dirty_journal::start_durable_dirty_usage_journal(store.clone()).await;
let journal_writer = durable_dirty_usage_journal.clone();
let mutation_journal = durable_dirty_usage_journal.clone();
rustfs_scanner::set_scanner_dirty_usage_mutation_observer(Some(Arc::new(move |bucket, object, producer| {
mutation_journal.record_committed_mutation(bucket, object, producer);
})));
ecstore_bucket::replication::set_scanner_dirty_usage_mutation_observer(Some(Arc::new(move |bucket, object, source| {
let producer = match source {
ecstore_bucket::replication::ScannerDirtyUsageMutationSource::Replication => {
@@ -966,7 +969,6 @@ pub(crate) async fn init_background_replication(store: Arc<ECStore>) {
}
};
rustfs_scanner::record_dirty_usage_object_from_producer(bucket, object, producer);
journal_writer.record_committed_mutation(bucket, object, producer);
})));
rustfs_scanner::set_scanner_dirty_usage_clear_observer(Some(Arc::new(move |cleared| {
durable_dirty_usage_journal.clear_confirmed_buckets(cleared);
scripts/check_test_wiring.py
+21 -7
View File
@@ -428,6 +428,7 @@ SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES = {
"exact-generation",
"scanner-instance",
"participating-peer-set",
"cleared-count-bound",
),
"participating_peer_capability_snapshot": (
"supports-scoped-ack",
@@ -452,6 +453,7 @@ SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS = {
"scanner_instance_observed",
"participating_peer_set_observed",
"whole_cycle_fallback_observed",
"cleared_count_bound_observed",
),
"participating_peer_capability_snapshot": (
"capability_probe_observed",
@@ -1827,6 +1829,7 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
"raw_entry_budget",
"max_raw_entries_per_round",
"max_objects_processed_per_round",
"bounded_work_quantum_observed",
"durable_checkpoint_committed",
"no_unbounded_tail",
))
@@ -2255,6 +2258,8 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic
f"{gate}.{field}.max_objects_processed_per_round", 1, 4096)
require(max_raw <= budget, f"{gate}.{field} raw entries exceed fixed budget")
require(max_objects <= budget, f"{gate}.{field} processed objects exceed fixed budget")
release_bundle_bool_true(evidence.get("bounded_work_quantum_observed"),
f"{gate}.{field}.bounded_work_quantum_observed")
release_bundle_bool_true(evidence.get("durable_checkpoint_committed"),
f"{gate}.{field}.durable_checkpoint_committed")
release_bundle_bool_true(evidence.get("no_unbounded_tail"), f"{gate}.{field}.no_unbounded_tail")
@@ -2772,10 +2777,10 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
require(evidence.get("stale_journals_after_gc") == 0,
f"{gate}.{field} requires zero stale journals after GC")
if field == "segment_activation_preflight":
require(evidence.get("production_activation") is False,
f"{gate}.{field} must keep production activation disabled")
require(evidence.get("scanner_segment_reuse_activated") is False,
f"{gate}.{field} must prove the runtime activation gate is disabled")
require(evidence.get("production_activation") is True,
f"{gate}.{field} must prove production activation is enabled")
require(evidence.get("scanner_segment_reuse_activated") is True,
f"{gate}.{field} must prove the runtime activation gate is enabled")
evidence_exact_strings(evidence.get("proof_inputs"),
SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS,
f"{gate}.{field}.proof_inputs")
@@ -3319,6 +3324,7 @@ def write_scanner_heal_release_bundle_fixture(root: Path, directory: Path) -> Pa
"raw_entry_budget": 8,
"max_raw_entries_per_round": 8,
"max_objects_processed_per_round": 8,
"bounded_work_quantum_observed": True,
"durable_checkpoint_committed": True,
"no_unbounded_tail": True,
})
@@ -3739,6 +3745,7 @@ class SelfTests(unittest.TestCase):
"raw_entry_budget": 8,
"max_raw_entries_per_round": 8,
"max_objects_processed_per_round": 8,
"bounded_work_quantum_observed": True,
"durable_checkpoint_committed": True,
"no_unbounded_tail": True,
})
@@ -3944,8 +3951,8 @@ class SelfTests(unittest.TestCase):
},
]
if field == "segment_activation_preflight":
evidence["production_activation"] = False
evidence["scanner_segment_reuse_activated"] = False
evidence["production_activation"] = True
evidence["scanner_segment_reuse_activated"] = True
evidence["proof_inputs"] = list(SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS)
evidence["fail_closed_checks"] = list(SCANNER_HEAL_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS)
if field == "cold_segment_reuse_measurement":
@@ -4447,7 +4454,14 @@ class SelfTests(unittest.TestCase):
"zero stale journals",
),
("same-window-fields", "G14", "same_window_field_evidence", lambda item: item.update({"same_window_fields": ["ec8_4_evidence", "multi_set_evidence"]}), "JSON artifact same_window_fields mismatch"),
("activation-enabled", "G11", "segment_activation_preflight", lambda item: item.update({"production_activation": True}), "production activation disabled"),
("activation-disabled", "G11", "segment_activation_preflight", lambda item: item.update({"production_activation": False}), "production activation is enabled"),
(
"activation-runtime-disabled",
"G11",
"segment_activation_preflight",
lambda item: item.update({"scanner_segment_reuse_activated": False}),
"runtime activation gate is enabled",
),
(
"activation-missing-fail-closed",
"G11",
scripts/diagnose_scanner_enumeration_restart.py
+20 -2
View File
@@ -78,10 +78,27 @@ def replays_raw_window(previous, current):
or previous["raw_first_entry"] is None or current["raw_first_entry"] is None
or previous["raw_last_entry"] is None or current["raw_last_entry"] is None):
return False
raw_page_index_progressed = (
previous["raw_page_index_parent"] == current["raw_page_index_parent"]
and (
current["raw_page_index_committed_entries"] > previous["raw_page_index_committed_entries"]
or (current["raw_page_index_complete"] and not previous["raw_page_index_complete"])
)
)
return (previous["raw_first_entry"] == current["raw_first_entry"]
and previous["raw_last_entry"] == current["raw_last_entry"]
and previous["objects_retained"] == current["objects_before"]
and current["objects_retained"] == previous["objects_retained"])
and current["objects_retained"] == previous["objects_retained"]
and not raw_page_index_progressed)
def fully_retained(report, objects):
return all(report[key] == objects for key in
("objects_retained", "versions_retained", "bytes_retained"))
def final_complete_recheck_after_full_retention(previous, current, objects):
return fully_retained(previous, objects) and converged(current, objects)
def validate_recoverable_quantum(reports, *, objects, budget, require_converged):
@@ -101,7 +118,8 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
raise ValueError("durable retained coverage did not survive process restart")
if report["objects_retained"] < previous["objects_retained"]:
raise ValueError("durable retained coverage regressed across restart")
if replays_raw_window(previous, report):
if (not final_complete_recheck_after_full_retention(previous, report, objects)
and replays_raw_window(previous, report)):
raise ValueError("raw enumeration window replayed without durable coverage")
if (report["raw_page_index_parent"] == previous["raw_page_index_parent"]
and report["raw_page_index_committed_entries"] < previous["raw_page_index_committed_entries"]
scripts/run_scanner_heal_checkpoint_crash_evidence.py
+1
View File
@@ -251,6 +251,7 @@ def build_descriptor(args: argparse.Namespace) -> Path:
"raw_entry_budget": summary["raw_entry_budget"],
"max_raw_entries_per_round": summary["max_raw_entries_per_round"],
"max_objects_processed_per_round": summary["max_objects_processed_per_round"],
"bounded_work_quantum_observed": True,
"durable_checkpoint_committed": True,
"no_unbounded_tail": True,
}),
scripts/run_scanner_heal_maintenance_evidence.py
+22 -4
View File
@@ -146,13 +146,16 @@ def build_descriptor(args: argparse.Namespace) -> Path:
"gates": gates,
})
for gate in ("G11", "G13"):
subprocess.check_call([
check = subprocess.run([
sys.executable,
str(ROOT / "scripts/check_test_wiring.py"),
"--check-scanner-heal-release-bundle-gate",
str(descriptor),
gate,
], cwd=ROOT)
], cwd=ROOT, capture_output=True, text=True)
if check.returncode != 0:
details = "\n".join(part for part in (check.stdout.strip(), check.stderr.strip()) if part)
raise ValueError(details or f"{gate} release bundle gate check failed")
return descriptor
@@ -188,8 +191,8 @@ def write_self_test_proof(path: Path, source_revision: str) -> None:
"unknown_producer_excluded": True,
},
"segment_activation_preflight": {
"production_activation": False,
"scanner_segment_reuse_activated": False,
"production_activation": True,
"scanner_segment_reuse_activated": True,
"proof_inputs": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS),
"fail_closed_checks": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS),
},
@@ -243,6 +246,21 @@ def run_self_test() -> None:
else:
raise ValueError("self-test accepted synthetic proof")
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
proof = root / "maintenance-proof.json"
write_self_test_proof(proof, source_revision)
payload = wiring.read_json(proof)
payload["segment_activation_preflight"]["scanner_segment_reuse_activated"] = False
wiring.write_json(proof, payload)
try:
build_descriptor(parse_args(["--proof-json", str(proof), "--out-dir", str(root / "out")]))
except ValueError as err:
wiring.require("runtime activation gate is enabled" in str(err), "wrong self-test failure for inactive segment reuse")
else:
raise ValueError("self-test accepted inactive segment reuse proof")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
scripts/test_diagnose_scanner_enumeration_restart.py
+39
View File
@@ -4,6 +4,8 @@ import unittest
from diagnose_scanner_enumeration_restart import (
converged,
final_complete_recheck_after_full_retention,
fully_retained,
replays_raw_window,
validate_recoverable_quantum,
validate_report,
@@ -29,6 +31,7 @@ class ReportTests(unittest.TestCase):
report = self.report()
self.validate(report)
self.assertTrue(converged(report, 4))
self.assertTrue(fully_retained(report, 4))
def test_incomplete_or_inexact_coverage_cannot_pass(self):
for key, value in (("snapshot_complete", False), ("objects_retained", 3),
@@ -132,6 +135,9 @@ class ReportTests(unittest.TestCase):
previous["versions_retained"] = 0
previous["bytes_retained"] = 0
previous["objects_processed"] = 0
previous["raw_page_index_committed_entries"] = 1
previous["raw_page_index_indexed_entries"] = 1
previous["raw_page_index_complete"] = False
previous["snapshot_complete"] = False
previous["outcome"] = "cancelled_without_cache"
current = dict(previous, round=1, pid=124, objects_before=0)
@@ -140,6 +146,10 @@ class ReportTests(unittest.TestCase):
advanced = dict(current, objects_retained=1)
self.assertFalse(replays_raw_window(previous, advanced))
indexed = dict(current, raw_page_index_committed_entries=2,
raw_page_index_indexed_entries=2)
self.assertFalse(replays_raw_window(previous, indexed))
def test_recoverable_quantum_rejects_replayed_raw_window(self):
previous = self.report()
previous.update(objects_retained=0, versions_retained=0, bytes_retained=0,
@@ -149,6 +159,35 @@ class ReportTests(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "raw enumeration window replayed"):
validate_recoverable_quantum([previous, current], objects=4, budget=16, require_converged=False)
def test_recoverable_quantum_allows_final_complete_round_after_full_retention(self):
previous = self.report()
previous.update(raw_entries=8, raw_page_index_committed_entries=4,
raw_page_index_indexed_entries=4, objects_before=2,
objects_processed=2, objects_retained=4,
versions_retained=4, bytes_retained=4,
snapshot_complete=False, outcome="partial")
current = dict(previous, round=1, pid=124, objects_before=4,
snapshot_complete=True, outcome="complete")
self.assertTrue(replays_raw_window(previous, current))
self.assertTrue(final_complete_recheck_after_full_retention(previous, current, 4))
validate_recoverable_quantum([previous, current], objects=4, budget=16, require_converged=True)
def test_recoverable_quantum_rejects_partial_replay_after_full_retention(self):
previous = self.report()
previous.update(raw_entries=8, raw_page_index_committed_entries=4,
raw_page_index_indexed_entries=4, objects_before=2,
objects_processed=2, objects_retained=4,
versions_retained=4, bytes_retained=4,
snapshot_complete=False, outcome="partial")
current = dict(previous, round=1, pid=124, objects_before=4,
snapshot_complete=False, outcome="partial")
self.assertTrue(replays_raw_window(previous, current))
self.assertFalse(final_complete_recheck_after_full_retention(previous, current, 4))
with self.assertRaisesRegex(ValueError, "raw enumeration window replayed"):
validate_recoverable_quantum([previous, current], objects=4, budget=16, require_converged=False)
def test_recoverable_quantum_requires_three_stage_progress_and_convergence(self):
first = self.report()
first.update(round=0, pid=123, raw_entries=2, raw_page_index_committed_entries=2,