diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index ad308a645..b73365df8 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::{BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::future::Future; use std::path::Path; use std::pin::Pin; @@ -204,6 +204,7 @@ pub(crate) struct DistributedSegmentInvalidationEvidence { #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct ScannerSegmentReuseActivationProof { pub(crate) production_activation: bool, + pub(crate) producer_identity_coverage_complete: bool, pub(crate) durable_producer_identity: bool, pub(crate) restart_gap_absent: bool, pub(crate) generation_window_bound: bool, @@ -495,6 +496,7 @@ fn scanner_segment_reuse_activation_preflight_from_proof( ScannerSegmentReuseActivationPreflight { production_activation: proof.production_activation, scanner_segment_reuse_activated: proof.production_activation + && proof.producer_identity_coverage_complete && proof.durable_producer_identity && proof.restart_gap_absent && proof.generation_window_bound @@ -504,7 +506,8 @@ fn scanner_segment_reuse_activation_preflight_from_proof( proof_inputs: &SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS, fail_closed_checks: &SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS, fail_closed_blockers: [ - (!proof.durable_producer_identity).then_some("missing_producer_identity"), + (!proof.producer_identity_coverage_complete || !proof.durable_producer_identity) + .then_some("missing_producer_identity"), (!proof.restart_gap_absent).then_some("restart_gap"), (!proof.generation_window_bound).then_some("generation_gap"), (!proof.overflow_absent).then_some("overflow"), @@ -516,17 +519,20 @@ fn scanner_segment_reuse_activation_preflight_from_proof( fn scanner_segment_reuse_activation_preflight_for_cycle( dirty_usage_snapshot: &DirtyUsageSnapshot, + dirty_usage_producer_evidence: DirtyUsageProducerEvidence, distributed: bool, distributed_segment_invalidation_evidence: Option, cold_zero_walk_oracle: bool, ) -> ScannerSegmentReuseActivationPreflight { scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof { production_activation: false, - durable_producer_identity: false, - restart_gap_absent: false, + producer_identity_coverage_complete: dirty_usage_producer_evidence.producer_identity_coverage_complete, + durable_producer_identity: dirty_usage_producer_evidence.durable_producer_identity, + restart_gap_absent: dirty_usage_producer_evidence.restart_gap_absent, generation_window_bound: dirty_usage_snapshot.covers_all_pending && dirty_usage_snapshot.generation != 0 - && dirty_usage_snapshot.generation != u64::MAX, + && dirty_usage_snapshot.generation != u64::MAX + && dirty_usage_producer_evidence.generation_window_bound, overflow_absent: dirty_usage_snapshot.covers_all_pending, cold_zero_walk_oracle, distributed_peer_invalidation: !distributed || distributed_segment_invalidation_evidence.is_some(), diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index fa94208cb..1c2af05e9 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -16,17 +16,16 @@ use super::*; pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); -// Lock order when both dirty maps are needed is `DIRTY_USAGE_BUCKETS` followed -// by `DIRTY_USAGE_BUCKET_SCOPES`. Both are held only for synchronous map -// updates, so no scanner task can observe a bucket generation without its -// matching scope. +// Lock order when dirty usage state is updated is `DIRTY_USAGE_BUCKETS`, +// `DIRTY_USAGE_BUCKET_SCOPES`, then `DIRTY_USAGE_PRODUCER_IDENTITIES`. All +// guards are held only for synchronous map updates, so no scanner task can +// observe a bucket generation without its matching scope and producer evidence. pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); // Non-authoritative process-local producer coverage. Any future segment reuse // activation must bind this to the exact generation window and durable proof. -pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock< - StdMutex>, -> = LazyLock::new(|| StdMutex::new(BTreeSet::new())); +pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock> = + LazyLock::new(|| StdMutex::new(BTreeMap::new())); 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::())); pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0); @@ -59,6 +58,23 @@ pub(super) type DirtyUsageBucketScopes = HashMap; const MAX_DIRTY_USAGE_TOP_LEVEL_ENTRIES_PER_BUCKET: usize = 128; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct DirtyUsageProducerIdentityState { + first_generation: u64, + last_generation: u64, +} + +pub(super) type DirtyUsageProducerIdentities = + BTreeMap; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct DirtyUsageProducerEvidence { + pub(super) producer_identity_coverage_complete: bool, + pub(super) durable_producer_identity: bool, + pub(super) restart_gap_absent: bool, + pub(super) generation_window_bound: bool, +} + /// A point-in-time view of the local dirty bucket generations. /// /// `complete == false` is an all-or-nothing overflow signal: `buckets` is @@ -264,8 +280,7 @@ fn dirty_usage_bucket_scopes() -> MutexGuard<'static, DirtyUsageBucketScopes> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } -fn dirty_usage_producer_identities() --> MutexGuard<'static, BTreeSet> { +fn dirty_usage_producer_identities() -> MutexGuard<'static, DirtyUsageProducerIdentities> { DIRTY_USAGE_PRODUCER_IDENTITIES .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) @@ -286,7 +301,7 @@ pub fn record_dirty_usage_bucket(bucket: &str) { return; } - record_dirty_usage_bucket_inner(bucket); + record_dirty_usage_bucket_inner(bucket, std::iter::empty()); } pub fn record_dirty_usage_bucket_from_producer( @@ -297,8 +312,7 @@ pub fn record_dirty_usage_bucket_from_producer( return; } - record_segment_invalidation_producer_identity(producer); - record_dirty_usage_bucket_inner(bucket); + record_dirty_usage_bucket_inner(bucket, [producer]); } pub fn record_dirty_usage_bucket_from_producers(bucket: &str, producers: I) @@ -309,17 +323,21 @@ where return; } - record_segment_invalidation_producer_identities(producers); - record_dirty_usage_bucket_inner(bucket); + record_dirty_usage_bucket_inner(bucket, producers); } -fn record_dirty_usage_bucket_inner(bucket: &str) { +fn record_dirty_usage_bucket_inner(bucket: &str, producers: I) +where + I: IntoIterator, +{ let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); + let mut producer_identities = dirty_usage_producer_identities(); 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, generation, producers); dirty_buckets.len() }; global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets)); @@ -337,7 +355,7 @@ fn record_dirty_usage_bucket_inner(bucket: &str) { /// local: after restart or any unverified distributed path the scanner falls /// back to its ordinary bucket scan. pub fn record_dirty_usage_object(bucket: &str, object: &str) { - record_dirty_usage_object_inner(bucket, object); + record_dirty_usage_object_inner(bucket, object, std::iter::empty()); } pub fn record_dirty_usage_object_from_producer( @@ -349,13 +367,16 @@ pub fn record_dirty_usage_object_from_producer( return; } - record_segment_invalidation_producer_identity(producer); - record_dirty_usage_object_inner(bucket, object); + record_dirty_usage_object_inner(bucket, object, [producer]); } -fn record_dirty_usage_object_inner(bucket: &str, object: &str) { +fn record_dirty_usage_object_inner(bucket: &str, object: &str, producers: I) +where + I: IntoIterator, +{ + let producers = producers.into_iter().collect::>(); let Some(top_level_entry) = dirty_usage_top_level_entry(object) else { - record_dirty_usage_bucket(bucket); + record_dirty_usage_bucket_inner(bucket, producers); return; }; if bucket.is_empty() { @@ -365,6 +386,7 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) { let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); + let mut producer_identities = dirty_usage_producer_identities(); let generation = advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); dirty_buckets.insert(bucket.to_string(), generation); let scope = dirty_scopes @@ -380,6 +402,7 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) { if overflowed { *scope = DirtyUsageBucketScope::WholeBucket; } + record_segment_invalidation_producer_identities_for_generation(&mut producer_identities, generation, producers); dirty_buckets.len() }; global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets)); @@ -387,25 +410,29 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) { DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); } -fn record_segment_invalidation_producer_identity(producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity) { - record_segment_invalidation_producer_identities([producer]); -} - -fn record_segment_invalidation_producer_identities(producers: I) -where +fn record_segment_invalidation_producer_identities_for_generation( + identities: &mut DirtyUsageProducerIdentities, + generation: u64, + producers: I, +) where I: IntoIterator, { - let mut identities = dirty_usage_producer_identities(); for producer in producers { if producer.producer().is_some() { - identities.insert(producer); + identities + .entry(producer) + .and_modify(|state| state.last_generation = state.last_generation.max(generation)) + .or_insert(DirtyUsageProducerIdentityState { + first_generation: generation, + last_generation: generation, + }); } } } #[cfg(test)] fn dirty_usage_producer_identities_for_tests() -> BTreeSet { - dirty_usage_producer_identities().clone() + dirty_usage_producer_identities().keys().copied().collect() } fn dirty_usage_top_level_entry(object: &str) -> Option { @@ -680,6 +707,29 @@ pub(super) fn dirty_usage_snapshot_status(snapshot: &DirtyUsageSnapshot) -> Dirt } } +pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> DirtyUsageProducerEvidence { + let generation_window_bound = dirty_usage_snapshot_status(snapshot) == DirtyUsageSnapshotStatus::Current + && snapshot.generation != 0 + && snapshot.generation != u64::MAX; + let identities = dirty_usage_producer_identities() + .iter() + .filter(|(_, state)| state.first_generation <= snapshot.generation) + .map(|(identity, _)| *identity) + .collect::>(); + let producer_identity_coverage_complete = + generation_window_bound && crate::segment_invalidation::complete_segment_invalidation_producers(identities).is_ok(); + + DirtyUsageProducerEvidence { + producer_identity_coverage_complete, + // The current producer journal is still process-local. Keep the + // durable/restart gates closed until the mutation evidence is persisted + // and replayable across scanner restarts. + durable_producer_identity: false, + restart_gap_absent: false, + generation_window_bound, + } +} + #[cfg(test)] pub(super) fn dirty_usage_bucket_count() -> usize { dirty_usage_buckets().len() diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index d598fff27..6bb6af130 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -467,8 +467,13 @@ where } else { Vec::new() }; - let segment_reuse_activation_preflight = - scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, distributed, None, false); + let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle( + &dirty_usage_snapshot, + dirty_usage_producer_evidence(&dirty_usage_snapshot), + distributed, + None, + false, + ); return Ok(ScannerCycleResult::new(status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) @@ -708,6 +713,7 @@ where ); let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle( &dirty_usage_snapshot, + dirty_usage_producer_evidence(&dirty_usage_snapshot), distributed, distributed_segment_invalidation_evidence, cold_zero_walk_oracle, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 076a1b1e6..fcbe1abd7 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -104,6 +104,7 @@ fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() { fn scanner_segment_reuse_activation_requires_every_preflight_proof() { let complete_proof = ScannerSegmentReuseActivationProof { production_activation: true, + producer_identity_coverage_complete: true, durable_producer_identity: true, restart_gap_absent: true, generation_window_bound: true, @@ -125,9 +126,13 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() { assert_eq!(preflight.fail_closed_blockers().collect::>(), Vec::<&str>::new()); let mut missing_identity = complete_proof; - missing_identity.durable_producer_identity = false; + missing_identity.producer_identity_coverage_complete = false; assert_segment_reuse_activation_blocked_by(missing_identity, "missing_producer_identity"); + let mut non_durable_identity = complete_proof; + non_durable_identity.durable_producer_identity = false; + assert_segment_reuse_activation_blocked_by(non_durable_identity, "missing_producer_identity"); + let mut restart_gap = complete_proof; restart_gap.restart_gap_absent = false; assert_segment_reuse_activation_blocked_by(restart_gap, "restart_gap"); @@ -166,8 +171,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit all_peers_bound_to_generation_window: true, }; - let preflight = - scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, Some(distributed_evidence), true); + let preflight = scanner_segment_reuse_activation_preflight_for_cycle( + &dirty_usage_snapshot, + complete_process_local_producer_evidence(), + true, + Some(distributed_evidence), + true, + ); assert!(!preflight.production_activation); assert!(!preflight.scanner_segment_reuse_activated); @@ -186,7 +196,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs( covers_all_pending: false, }; - let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, None, false); + let preflight = scanner_segment_reuse_activation_preflight_for_cycle( + &dirty_usage_snapshot, + DirtyUsageProducerEvidence::default(), + true, + None, + false, + ); assert!(!preflight.production_activation); assert!(!preflight.scanner_segment_reuse_activated); @@ -205,7 +221,13 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke covers_all_pending: true, }; - let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, false, None, true); + let preflight = scanner_segment_reuse_activation_preflight_for_cycle( + &dirty_usage_snapshot, + complete_process_local_producer_evidence(), + false, + None, + true, + ); assert!(!preflight.production_activation); assert!(!preflight.scanner_segment_reuse_activated); @@ -219,6 +241,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke fn scanner_cycle_result_returns_segment_reuse_activation_preflight() { let proof = ScannerSegmentReuseActivationProof { production_activation: true, + producer_identity_coverage_complete: true, durable_producer_identity: true, restart_gap_absent: true, generation_window_bound: true, @@ -241,6 +264,15 @@ fn assert_segment_reuse_activation_blocked_by(proof: ScannerSegmentReuseActivati assert_eq!(preflight.fail_closed_blockers().collect::>(), vec![blocker]); } +fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence { + DirtyUsageProducerEvidence { + producer_identity_coverage_complete: true, + durable_producer_identity: false, + restart_gap_absent: false, + generation_window_bound: true, + } +} + async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc) { init_ecstore_config_for_scanner_tests(); let temp_dir = tempfile::tempdir().expect("multi-pool scanner test directory should be created"); @@ -1195,6 +1227,29 @@ fn dirty_usage_snapshot_detects_uncovered_generation() { clear_dirty_usage_buckets_for_tests(); } +#[test] +#[serial] +fn dirty_usage_producer_evidence_tracks_process_local_coverage_without_durable_restart_authority() { + use crate::segment_invalidation::SegmentInvalidationProducerIdentity; + + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket_from_producers("photos", SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION); + let snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation()); + + let evidence = dirty_usage_producer_evidence(&snapshot); + + assert!(evidence.generation_window_bound); + assert!(evidence.producer_identity_coverage_complete); + assert!(!evidence.durable_producer_identity); + assert!(!evidence.restart_gap_absent); + + record_dirty_usage_bucket_from_producer("videos", SegmentInvalidationProducerIdentity::PutObject); + let stale_evidence = dirty_usage_producer_evidence(&snapshot); + assert!(!stale_evidence.generation_window_bound); + assert!(!stale_evidence.producer_identity_coverage_complete); + clear_dirty_usage_buckets_for_tests(); +} + #[test] fn generation_saturates_instead_of_wrapping() { let generation = AtomicU64::new(u64::MAX - 1);