From 358bf6832d5287d4096050fd36f146687640599d Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 10 Sep 2026 12:03:55 +0800 Subject: [PATCH] fix(scanner): harden heal release gates (#7614) --- crates/scanner/src/lib.rs | 6 +- .../scanner/tests/scoped_ack_publication.rs | 19 ++- crates/scanner/src/scanner_io.rs | 80 +++++++-- crates/scanner/src/scanner_io/dirty_usage.rs | 98 ++++++++++- crates/scanner/src/scanner_io/io_cycle.rs | 2 +- crates/scanner/src/scanner_io/tests.rs | 156 +++++++++++++++++- .../scanner_io/tests/scoped_entry_fallback.rs | 31 ++++ rustfs/src/storage/storage_api.rs | 6 +- scripts/check_test_wiring.py | 28 +++- .../diagnose_scanner_enumeration_restart.py | 22 ++- ..._scanner_heal_checkpoint_crash_evidence.py | 1 + .../run_scanner_heal_maintenance_evidence.py | 26 ++- ...st_diagnose_scanner_enumeration_restart.py | 39 +++++ 13 files changed, 466 insertions(+), 48 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index e5461df9d..5d5e7dc51 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -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}; diff --git a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs index 2fa58fff2..472e595d7 100644 --- a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs +++ b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs @@ -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(), diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 17570af12..69603627a 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -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, + 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::>(); 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::(); + 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, ) -> 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}; diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index a0bd24554..2c0e7fe4e 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -30,6 +30,8 @@ pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock>> = LazyLock::new(|| StdRwLock::new(None)); +static DIRTY_USAGE_MUTATION_OBSERVER: LazyLock>> = + LazyLock::new(|| StdRwLock::new(None)); pub(super) static DIRTY_USAGE_PRODUCER_COVERAGE: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock = LazyLock::new(Notify::new); pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock = LazyLock::new(|| format!("{:032x}", rand::random::())); @@ -53,6 +55,8 @@ pub struct ScannerDirtyUsageBucket { } pub type ScannerDirtyUsageClearObserver = Arc) + Send + Sync + 'static>; +pub type ScannerDirtyUsageMutationObserver = + Arc; /// A non-durable optimization hint for a dirty bucket. /// @@ -83,6 +87,7 @@ pub(super) type DirtyUsageProducerIdentities = BTreeMap, +) -> Option { + 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) { if cleared.is_empty() { return; @@ -125,6 +139,30 @@ fn notify_dirty_usage_clear(cleared: Vec) { } } +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(bucket: &str, producers: I) where I: IntoIterator, { + let producers = producers.into_iter().collect::>(); 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 }, diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index caf526ed7..08a177984 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -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, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 5e0122628..de1438fe8 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -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::<&str>::new()); + assert!(!preflight.scanner_segment_reuse_activated); + assert_eq!( + preflight.fail_closed_blockers().collect::>(), + 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 { 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 844f9642f..d974bd923 100644 --- a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs +++ b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs @@ -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); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index fce77106d..5c8a095ec 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -955,7 +955,10 @@ pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperti pub(crate) async fn init_background_replication(store: Arc) { 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) { } }; 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); diff --git a/scripts/check_test_wiring.py b/scripts/check_test_wiring.py index d5919c0d7..8762d7291 100755 --- a/scripts/check_test_wiring.py +++ b/scripts/check_test_wiring.py @@ -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", diff --git a/scripts/diagnose_scanner_enumeration_restart.py b/scripts/diagnose_scanner_enumeration_restart.py index 37aa49e23..95d8b2a6e 100644 --- a/scripts/diagnose_scanner_enumeration_restart.py +++ b/scripts/diagnose_scanner_enumeration_restart.py @@ -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"] diff --git a/scripts/run_scanner_heal_checkpoint_crash_evidence.py b/scripts/run_scanner_heal_checkpoint_crash_evidence.py index f1662d9dc..ea9671199 100755 --- a/scripts/run_scanner_heal_checkpoint_crash_evidence.py +++ b/scripts/run_scanner_heal_checkpoint_crash_evidence.py @@ -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, }), diff --git a/scripts/run_scanner_heal_maintenance_evidence.py b/scripts/run_scanner_heal_maintenance_evidence.py index 5a7a237e8..1fc7a5b9c 100755 --- a/scripts/run_scanner_heal_maintenance_evidence.py +++ b/scripts/run_scanner_heal_maintenance_evidence.py @@ -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__) diff --git a/scripts/test_diagnose_scanner_enumeration_restart.py b/scripts/test_diagnose_scanner_enumeration_restart.py index 370ebf580..7ca76a368 100644 --- a/scripts/test_diagnose_scanner_enumeration_restart.py +++ b/scripts/test_diagnose_scanner_enumeration_restart.py @@ -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,