diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 0a9c2c2f0..76ba60b97 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -97,9 +97,9 @@ pub use scanner::{ pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket, - record_dirty_usage_bucket_from_producer, record_dirty_usage_object, record_dirty_usage_object_from_producer, - record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, - scanner_maintenance_generation, + 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, scanner_activity_epoch, + scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, }; pub use segment_invalidation::SegmentInvalidationProducerIdentity; pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER}; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 9bde466ce..007ae23d7 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -98,6 +98,24 @@ const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued"; const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active"; const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued"; +pub(crate) const SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS: [&str; 7] = [ + "source", + "bucket_incarnation", + "key_format", + "baseline_scan_plan_digest", + "process_epoch", + "generation_window", + "producer_identities", +]; +pub(crate) const SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS: [&str; 6] = [ + "missing_producer_identity", + "restart_gap", + "generation_gap", + "overflow", + "missing_cold_zero_walk_oracle", + "distributed_without_peer_invalidation", +]; + pub type DirtyUsageBuckets = HashMap; #[derive(Clone, Debug)] @@ -163,11 +181,51 @@ struct ScannerPeerDirtyUsageExpectation { struct VerifiedRemoteDirtyUsage { dirty_buckets: HashSet, acknowledgements: Vec, + peer_count: usize, + dirty_peer_count: usize, } struct ScannerBucketScopeResolutionResult { scope: ScannerBucketScanScope, remote_dirty_usage_acknowledgements: Vec, + distributed_segment_invalidation_evidence: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct DistributedSegmentInvalidationEvidence { + pub(crate) invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain, + pub(crate) distributed_ec_invalidation: bool, + pub(crate) peer_count: usize, + pub(crate) dirty_peer_count: usize, + pub(crate) same_window_remote_proof: bool, + pub(crate) all_peers_bound_to_generation_window: bool, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct ScannerSegmentReuseActivationProof { + pub(crate) production_activation: bool, + pub(crate) durable_producer_identity: bool, + pub(crate) restart_gap_absent: bool, + pub(crate) generation_window_bound: bool, + pub(crate) overflow_absent: bool, + pub(crate) cold_zero_walk_oracle: bool, + pub(crate) distributed_peer_invalidation: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct ScannerSegmentReuseActivationPreflight { + pub(crate) production_activation: bool, + pub(crate) scanner_segment_reuse_activated: bool, + pub(crate) proof_inputs: &'static [&'static str], + pub(crate) fail_closed_checks: &'static [&'static str], + pub(crate) fail_closed_blockers: [Option<&'static str>; 6], +} + +impl ScannerSegmentReuseActivationPreflight { + #[cfg(test)] + pub(crate) fn fail_closed_blockers(&self) -> impl Iterator + '_ { + self.fail_closed_blockers.iter().filter_map(|blocker| *blocker) + } } fn verified_remote_dirty_usage( @@ -191,6 +249,7 @@ fn verified_remote_dirty_usage( || !snapshot.complete || snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).unwrap_or(u64::MAX) || (expected.pending && snapshot.pending_bucket_count == 0) + || (!expected.pending && snapshot.pending_bucket_count != 0) { return None; } @@ -216,9 +275,13 @@ fn verified_remote_dirty_usage( } } + let peer_count = received_peers.len(); + let dirty_peer_count = acknowledgements.len(); (received_peers.len() == expected_peers.len()).then_some(VerifiedRemoteDirtyUsage { dirty_buckets, acknowledgements, + peer_count, + dirty_peer_count, }) } @@ -244,8 +307,11 @@ fn resolve_remote_dirty_usage_scope( let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult { scope, remote_dirty_usage_acknowledgements: Vec::new(), + distributed_segment_invalidation_evidence: None, }; + let peer_count = remote_dirty_usage.peer_count; + let dirty_peer_count = remote_dirty_usage.dirty_peer_count; 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 @@ -283,10 +349,21 @@ fn resolve_remote_dirty_usage_scope( if scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) { return default_result(ScannerBucketScanScope::default()); } + let has_scoped_acknowledgements = !scoped_acknowledgements.is_empty(); 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, + }, + ), } } @@ -406,10 +483,39 @@ fn scoped_scan_scope_from_dirty_buckets( ScannerBucketScanScope::from_dirty_buckets(selected_buckets, selected_bucket_prefixes, baseline_scan_plan_digest) } -fn scanner_segment_reuse_activated() -> bool { +fn scanner_segment_reuse_activation_preflight() -> ScannerSegmentReuseActivationPreflight { // Production segment reuse stays disabled until a durable mutation-stream // proof satisfies the segment invalidation contract. - false + scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof::default()) +} + +fn scanner_segment_reuse_activation_preflight_from_proof( + proof: ScannerSegmentReuseActivationProof, +) -> ScannerSegmentReuseActivationPreflight { + ScannerSegmentReuseActivationPreflight { + production_activation: proof.production_activation, + scanner_segment_reuse_activated: proof.production_activation + && proof.durable_producer_identity + && proof.restart_gap_absent + && proof.generation_window_bound + && proof.overflow_absent + && proof.cold_zero_walk_oracle + && proof.distributed_peer_invalidation, + 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.restart_gap_absent).then_some("restart_gap"), + (!proof.generation_window_bound).then_some("generation_gap"), + (!proof.overflow_absent).then_some("overflow"), + (!proof.cold_zero_walk_oracle).then_some("missing_cold_zero_walk_oracle"), + (!proof.distributed_peer_invalidation).then_some("distributed_without_peer_invalidation"), + ], + } +} + +fn scanner_segment_reuse_activated() -> bool { + scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated } pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool { @@ -1056,6 +1162,7 @@ pub(crate) struct ScannerCycleResult { observational_snapshot_published: bool, dirty_usage_clear: Option, remote_dirty_usage_acknowledgements: Vec, + distributed_segment_invalidation_evidence: Option, remote_publication_lease_targets: Vec<(String, String, u64)>, failed_dirty_usage: bool, pending_maintenance_work: bool, @@ -1072,6 +1179,7 @@ impl ScannerCycleResult { observational_snapshot_published: false, dirty_usage_clear, remote_dirty_usage_acknowledgements: Vec::new(), + distributed_segment_invalidation_evidence: None, remote_publication_lease_targets: Vec::new(), failed_dirty_usage: false, pending_maintenance_work: false, @@ -1137,6 +1245,15 @@ impl ScannerCycleResult { self } + fn with_distributed_segment_invalidation_evidence( + mut self, + evidence: Option, + ) -> Self { + self.publication_expectation = None; + self.distributed_segment_invalidation_evidence = evidence; + self + } + pub(crate) fn with_remote_publication_lease_targets(mut self, targets: Vec<(String, String, u64)>) -> Self { self.publication_expectation = None; self.remote_publication_lease_targets = targets; @@ -1224,9 +1341,9 @@ pub(crate) use cache::{ pub use dirty_usage::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, clear_dirty_usage_bucket, record_dirty_usage_bucket, - record_dirty_usage_bucket_from_producer, record_dirty_usage_object, record_dirty_usage_object_from_producer, - record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, - scanner_maintenance_generation, + 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, scanner_activity_epoch, + scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, }; #[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 4c9fdc837..fa94208cb 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -226,12 +226,21 @@ mod scoped_dirty_usage_tests { record_dirty_usage_object_from_producer("photos", "hot/object", SegmentInvalidationProducerIdentity::PutObject); record_dirty_usage_object_from_producer("photos", "archive/object", SegmentInvalidationProducerIdentity::DeleteObject); record_dirty_usage_bucket_from_producer("photos", SegmentInvalidationProducerIdentity::Unknown); + record_dirty_usage_bucket_from_producers( + "photos", + [ + SegmentInvalidationProducerIdentity::DeleteMarker, + SegmentInvalidationProducerIdentity::AbortMultipartUpload, + ], + ); assert_eq!( dirty_usage_producer_identities_for_tests(), BTreeSet::from([ SegmentInvalidationProducerIdentity::PutObject, - SegmentInvalidationProducerIdentity::DeleteObject + SegmentInvalidationProducerIdentity::DeleteObject, + SegmentInvalidationProducerIdentity::DeleteMarker, + SegmentInvalidationProducerIdentity::AbortMultipartUpload ]) ); assert_eq!( @@ -292,6 +301,18 @@ pub fn record_dirty_usage_bucket_from_producer( record_dirty_usage_bucket_inner(bucket); } +pub fn record_dirty_usage_bucket_from_producers(bucket: &str, producers: I) +where + I: IntoIterator, +{ + if bucket.is_empty() { + return; + } + + record_segment_invalidation_producer_identities(producers); + record_dirty_usage_bucket_inner(bucket); +} + fn record_dirty_usage_bucket_inner(bucket: &str) { let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); @@ -367,8 +388,18 @@ fn record_dirty_usage_object_inner(bucket: &str, object: &str) { } fn record_segment_invalidation_producer_identity(producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity) { - if producer.producer().is_some() { - dirty_usage_producer_identities().insert(producer); + record_segment_invalidation_producer_identities([producer]); +} + +fn record_segment_invalidation_producer_identities(producers: I) +where + I: IntoIterator, +{ + let mut identities = dirty_usage_producer_identities(); + for producer in producers { + if producer.producer().is_some() { + identities.insert(producer); + } } } diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 8535468ed..544354147 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -122,6 +122,7 @@ where let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult { scope, remote_dirty_usage_acknowledgements: Vec::new(), + distributed_segment_invalidation_evidence: None, }; if resolution.requires_full_scan { return default_result(ScannerBucketScanScope::default()); @@ -408,6 +409,7 @@ where ) .await; let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements; + let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence; let scan_scope = scope_resolution.scope; #[cfg(test)] if let Some(observer) = resolved_scope_observer { @@ -783,6 +785,7 @@ where .with_observational_snapshot_published(observational_snapshot_published) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) + .with_distributed_segment_invalidation_evidence(distributed_segment_invalidation_evidence) .with_failed_dirty_usage(!failed_buckets.is_empty()) .with_pending_maintenance_work(pending_maintenance_work) .with_required_cycle_floor(required_cycle_floor) diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 01ee44627..e4caede9d 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -85,6 +85,78 @@ fn scanner_activity_preflight_defers_a_temporarily_offline_peer() { } } +#[test] +fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() { + let preflight = scanner_segment_reuse_activation_preflight(); + + assert!(!preflight.production_activation); + assert!(!preflight.scanner_segment_reuse_activated); + assert!(!scanner_segment_reuse_activated()); + assert_eq!(preflight.proof_inputs, SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS); + assert_eq!(preflight.fail_closed_checks, SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS); + assert_eq!( + preflight.fail_closed_blockers().collect::>(), + SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS + ); +} + +#[test] +fn scanner_segment_reuse_activation_requires_every_preflight_proof() { + let complete_proof = ScannerSegmentReuseActivationProof { + production_activation: true, + durable_producer_identity: true, + restart_gap_absent: true, + generation_window_bound: true, + overflow_absent: true, + cold_zero_walk_oracle: true, + distributed_peer_invalidation: true, + }; + + let mut production_disabled = complete_proof; + production_disabled.production_activation = false; + let preflight = scanner_segment_reuse_activation_preflight_from_proof(production_disabled); + assert!(!preflight.production_activation); + assert!(!preflight.scanner_segment_reuse_activated); + assert_eq!(preflight.fail_closed_blockers().collect::>(), Vec::<&str>::new()); + + let preflight = scanner_segment_reuse_activation_preflight_from_proof(complete_proof); + assert!(preflight.production_activation); + assert!(preflight.scanner_segment_reuse_activated); + assert_eq!(preflight.fail_closed_blockers().collect::>(), Vec::<&str>::new()); + + let mut missing_identity = complete_proof; + missing_identity.durable_producer_identity = false; + assert_segment_reuse_activation_blocked_by(missing_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"); + + let mut generation_gap = complete_proof; + generation_gap.generation_window_bound = false; + assert_segment_reuse_activation_blocked_by(generation_gap, "generation_gap"); + + let mut overflow = complete_proof; + overflow.overflow_absent = false; + assert_segment_reuse_activation_blocked_by(overflow, "overflow"); + + let mut missing_cold_oracle = complete_proof; + missing_cold_oracle.cold_zero_walk_oracle = false; + assert_segment_reuse_activation_blocked_by(missing_cold_oracle, "missing_cold_zero_walk_oracle"); + + let mut missing_distributed_invalidation = complete_proof; + missing_distributed_invalidation.distributed_peer_invalidation = false; + assert_segment_reuse_activation_blocked_by(missing_distributed_invalidation, "distributed_without_peer_invalidation"); +} + +fn assert_segment_reuse_activation_blocked_by(proof: ScannerSegmentReuseActivationProof, blocker: &'static str) { + let preflight = scanner_segment_reuse_activation_preflight_from_proof(proof); + + assert!(preflight.production_activation); + assert!(!preflight.scanner_segment_reuse_activated); + assert_eq!(preflight.fail_closed_blockers().collect::>(), vec![blocker]); +} + 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"); @@ -1781,6 +1853,18 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi "peer dirty state is not a distributed segment invalidation proof" ); assert_eq!(distributed.remote_dirty_usage_acknowledgements.len(), 1); + let evidence = distributed + .distributed_segment_invalidation_evidence + .expect("same-window peer snapshot and scoped ACK capability form distributed evidence"); + assert_eq!(evidence.peer_count, 1); + assert_eq!(evidence.dirty_peer_count, 1); + assert_eq!( + evidence.invalidation_domain, + crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc + ); + assert!(evidence.distributed_ec_invalidation); + assert!(evidence.same_window_remote_proof); + assert!(evidence.all_peers_bound_to_generation_window); } fn peer_dirty_usage_snapshot( @@ -1827,7 +1911,7 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots() ScannerPeerDirtyUsageExpectation { instance_id: "instance-b".to_string(), generation: 3, - pending: false, + pending: true, }, ), ]); @@ -1874,10 +1958,36 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots() }, }, ], + peer_count: 2, + dirty_peer_count: 2, }) ); } +#[test] +fn verified_remote_dirty_usage_rejects_peer_snapshot_that_contradicts_activity_pending_state() { + let expected_peers = HashMap::from([( + "node-a:9000".to_string(), + ScannerPeerDirtyUsageExpectation { + instance_id: "instance-a".to_string(), + generation: 7, + pending: false, + }, + )]); + + assert!( + verified_remote_dirty_usage( + &expected_peers, + vec![( + "node-a:9000".to_string(), + peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]), + )], + ) + .is_none(), + "a clean activity window cannot authorize a dirty peer snapshot or scoped ACK" + ); +} + #[test] fn scanner_scoped_dirty_usage_ack_cost_threshold_is_single_protocol_batch() { let acknowledgement = |entry_count: usize| crate::scanner::ScannerDirtyUsageAcknowledgement { @@ -1951,6 +2061,7 @@ fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_thresho result.remote_dirty_usage_acknowledgements.is_empty(), "full-scan fallback must not send a scoped ACK that peers would reject or split" ); + assert!(result.distributed_segment_invalidation_evidence.is_none()); } #[test] @@ -2092,6 +2203,11 @@ async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is assert_eq!(result.scope.selected_buckets.as_deref(), expected_buckets.as_ref()); assert_eq!(result.remote_dirty_usage_acknowledgements.len(), expected_ack_count); + assert_eq!( + result.distributed_segment_invalidation_evidence.is_some(), + capability, + "distributed evidence requires an authenticated scoped ACK capability probe" + ); } } diff --git a/crates/scanner/src/segment_invalidation.rs b/crates/scanner/src/segment_invalidation.rs index d1136809b..d5a2c96ca 100644 --- a/crates/scanner/src/segment_invalidation.rs +++ b/crates/scanner/src/segment_invalidation.rs @@ -57,6 +57,9 @@ pub enum SegmentInvalidationProducerIdentity { DeleteObject, DeleteMarker, CompleteMultipartUpload, + AbortMultipartUpload, + ObjectMetadata, + BucketMetadata, Replication, TierTransition, TierExpiration, @@ -66,11 +69,14 @@ pub enum SegmentInvalidationProducerIdentity { } impl SegmentInvalidationProducerIdentity { - pub const REQUIRED_PRODUCTION: [Self; 8] = [ + pub const REQUIRED_PRODUCTION: [Self; 11] = [ Self::PutObject, Self::DeleteObject, Self::DeleteMarker, Self::CompleteMultipartUpload, + Self::AbortMultipartUpload, + Self::ObjectMetadata, + Self::BucketMetadata, Self::Replication, Self::TierTransition, Self::TierExpiration, @@ -82,7 +88,9 @@ impl SegmentInvalidationProducerIdentity { Self::PutObject => Some(SegmentInvalidationProducer::Put), Self::DeleteObject => Some(SegmentInvalidationProducer::Delete), Self::DeleteMarker => Some(SegmentInvalidationProducer::DeleteMarker), - Self::CompleteMultipartUpload => Some(SegmentInvalidationProducer::Multipart), + Self::CompleteMultipartUpload | Self::AbortMultipartUpload => Some(SegmentInvalidationProducer::Multipart), + Self::ObjectMetadata => Some(SegmentInvalidationProducer::Put), + Self::BucketMetadata => Some(SegmentInvalidationProducer::DirectoryObject), Self::Replication => Some(SegmentInvalidationProducer::Replication), Self::TierTransition | Self::TierExpiration => Some(SegmentInvalidationProducer::Tier), Self::DirectoryObject => Some(SegmentInvalidationProducer::DirectoryObject), @@ -402,8 +410,12 @@ mod tests { SegmentInvalidationProducerIdentity::DeleteObject, SegmentInvalidationProducerIdentity::DeleteMarker, SegmentInvalidationProducerIdentity::CompleteMultipartUpload, + SegmentInvalidationProducerIdentity::AbortMultipartUpload, + SegmentInvalidationProducerIdentity::ObjectMetadata, + SegmentInvalidationProducerIdentity::BucketMetadata, SegmentInvalidationProducerIdentity::Replication, SegmentInvalidationProducerIdentity::TierTransition, + SegmentInvalidationProducerIdentity::TierExpiration, SegmentInvalidationProducerIdentity::DirectoryObject, SegmentInvalidationProducerIdentity::Unknown, ]), @@ -415,8 +427,12 @@ mod tests { SegmentInvalidationProducerIdentity::DeleteObject, SegmentInvalidationProducerIdentity::DeleteMarker, SegmentInvalidationProducerIdentity::CompleteMultipartUpload, + SegmentInvalidationProducerIdentity::AbortMultipartUpload, + SegmentInvalidationProducerIdentity::ObjectMetadata, + SegmentInvalidationProducerIdentity::BucketMetadata, SegmentInvalidationProducerIdentity::Replication, SegmentInvalidationProducerIdentity::TierTransition, + SegmentInvalidationProducerIdentity::TierExpiration, SegmentInvalidationProducerIdentity::DirectoryObject, SegmentInvalidationProducerIdentity::TestFixture, ]), @@ -427,6 +443,7 @@ mod tests { SegmentInvalidationProducerIdentity::PutObject, SegmentInvalidationProducerIdentity::DeleteObject, SegmentInvalidationProducerIdentity::DeleteMarker, + SegmentInvalidationProducerIdentity::CompleteMultipartUpload, SegmentInvalidationProducerIdentity::Replication, SegmentInvalidationProducerIdentity::TierTransition, SegmentInvalidationProducerIdentity::DirectoryObject, @@ -439,6 +456,9 @@ mod tests { SegmentInvalidationProducerIdentity::DeleteObject, SegmentInvalidationProducerIdentity::DeleteMarker, SegmentInvalidationProducerIdentity::CompleteMultipartUpload, + SegmentInvalidationProducerIdentity::AbortMultipartUpload, + SegmentInvalidationProducerIdentity::ObjectMetadata, + SegmentInvalidationProducerIdentity::BucketMetadata, SegmentInvalidationProducerIdentity::Replication, SegmentInvalidationProducerIdentity::TierTransition, SegmentInvalidationProducerIdentity::DirectoryObject, diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index d77531e1f..f9c4879e3 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -1389,7 +1389,10 @@ impl DefaultBucketUsecase { counter!("rustfs_create_bucket_total").increment(1); let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata, + ); result } @@ -1781,7 +1784,10 @@ impl DefaultBucketUsecase { warn!(bucket = %bucket, error = ?err, "site replication bucket tagging delete hook failed"); } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata, + ); Ok(S3Response::new(DeleteBucketTaggingOutput {})) } @@ -2698,7 +2704,10 @@ impl DefaultBucketUsecase { warn!(bucket = %bucket, error = ?err, "site replication bucket tagging hook failed"); } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata, + ); Ok(S3Response::new(PutBucketTaggingOutput::default())) } @@ -2733,7 +2742,10 @@ impl DefaultBucketUsecase { warn!(bucket = %bucket, error = ?err, "site replication bucket versioning hook failed"); } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata, + ); Ok(S3Response::new(PutBucketVersioningOutput {})) } diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 00811cdad..0d442a48a 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -532,7 +532,11 @@ impl DefaultMultipartUsecase { .await { Ok(_) => { - rustfs_scanner::record_dirty_usage_object(&bucket, &key); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::AbortMultipartUpload, + ); Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() })) } Err(err) => { @@ -802,7 +806,11 @@ impl DefaultMultipartUsecase { schedule_object_replication(obj_info.clone(), store, completion_replication_decision).await; } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::CompleteMultipartUpload, + ); Ok::<_, S3Error>(obj_info) } }); diff --git a/rustfs/src/app/object/delete.rs b/rustfs/src/app/object/delete.rs index 5b19409f1..22460dd98 100644 --- a/rustfs/src/app/object/delete.rs +++ b/rustfs/src/app/object/delete.rs @@ -797,6 +797,18 @@ impl DefaultObjectUsecase { let resp_elements = build_event_resp_elements(&S3Response::new(DeleteObjectsOutput::default()), &request_context.request_id); let deleted_any = delete_results.iter().any(|result| result.delete_object.is_some()); + let delete_producers = delete_results + .iter() + .filter_map(|result| { + result.delete_object.as_ref().map(|deleted_object| { + if deleted_object.delete_marker && result.requested_version_id.is_none() { + rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteMarker + } else { + rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteObject + } + }) + }) + .collect::>(); let notify_bucket = bucket.clone(); spawn_background_with_context(Some(request_context), async move { let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Notify); @@ -838,7 +850,7 @@ impl DefaultObjectUsecase { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); if deleted_any { - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producers(&bucket, delete_producers); } // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead) let manager = get_capacity_manager(); @@ -1101,7 +1113,10 @@ impl DefaultObjectUsecase { let manager = get_capacity_manager(); manager.record_write_operation().await; let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteObject, + ); return result; } diff --git a/rustfs/src/app/object/extract.rs b/rustfs/src/app/object/extract.rs index aaa6323e8..ae79af6af 100644 --- a/rustfs/src/app/object/extract.rs +++ b/rustfs/src/app/object/extract.rs @@ -627,6 +627,7 @@ struct ExtractPreparedMember { write_plan: WritePlan, opts: ObjectOptions, replication: ReplicateDecision, + producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity, staging_permit: OwnedSemaphorePermit, member_permit: OwnedSemaphorePermit, } @@ -765,6 +766,7 @@ where write_plan, opts, replication, + producer_identity, staging_permit, member_permit, } = member; @@ -793,7 +795,7 @@ where // the scanner before its post-store awaits, then retains the lifecycle slot // through quota, cache, replication, and event construction. if !context.wrote_any_entry.swap(true, Ordering::AcqRel) { - rustfs_scanner::record_dirty_usage_bucket(&context.bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer(&context.bucket, producer_identity); } let success = complete_extract_member_post_commit(context, key, opts, replication, obj_info, backfilled_old_current_size).await; @@ -2603,6 +2605,7 @@ impl DefaultObjectUsecase { write_plan, opts, replication, + producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity::DirectoryObject, staging_permit, member_permit, }, @@ -2627,6 +2630,7 @@ impl DefaultObjectUsecase { write_plan, opts, replication, + producer_identity: rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject, staging_permit, member_permit, }, @@ -2656,6 +2660,11 @@ impl DefaultObjectUsecase { write_plan, opts, replication, + producer_identity: if is_dir { + rustfs_scanner::SegmentInvalidationProducerIdentity::DirectoryObject + } else { + rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject + }, staging_permit, member_permit, }); diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs index 99a7f6039..e932a9cd0 100644 --- a/rustfs/src/app/object/internal_put.rs +++ b/rustfs/src/app/object/internal_put.rs @@ -733,7 +733,11 @@ impl DefaultObjectUsecase { ) .await .map_err(ApiError::from)?; - rustfs_scanner::record_dirty_usage_bucket(bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + bucket, + key, + rustfs_scanner::SegmentInvalidationProducerIdentity::AbortMultipartUpload, + ); Ok(()) } } diff --git a/rustfs/src/app/object/restore.rs b/rustfs/src/app/object/restore.rs index 506826281..f941e7e3d 100644 --- a/rustfs/src/app/object/restore.rs +++ b/rustfs/src/app/object/restore.rs @@ -420,7 +420,11 @@ impl DefaultObjectUsecase { ) .await .map_err(ApiError::from)?; - rustfs_scanner::record_dirty_usage_object(&bucket, &object); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &object, + rustfs_scanner::SegmentInvalidationProducerIdentity::TierTransition, + ); #[cfg(test)] maybe_pause_after_restore_status_commit(&bucket, &object).await; drop(superseded_worker_guard.take()); @@ -494,7 +498,11 @@ impl DefaultObjectUsecase { err.to_string() ); } else { - rustfs_scanner::record_dirty_usage_object(&bucket_clone, &object_clone); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket_clone, + &object_clone, + rustfs_scanner::SegmentInvalidationProducerIdentity::TierTransition, + ); debug!(bucket = %bucket_clone, object = %object_clone, "Transitioned object restored"); } }); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index e62cd6d82..93cbc5ee9 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -730,7 +730,11 @@ impl S3 for FS { let result = Ok(S3Response::new(DeleteObjectTaggingOutput { version_id })); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &object, + rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata, + ); let duration = start_time.elapsed(); histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete").record(duration.as_secs_f64()); result @@ -1629,7 +1633,11 @@ impl S3 for FS { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata, + ); result } @@ -1733,7 +1741,10 @@ impl S3 for FS { ); } - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_bucket_from_producer( + &bucket, + rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata, + ); Ok(S3Response::new(PutObjectLockConfigurationOutput::default())) } @@ -1849,7 +1860,11 @@ impl S3 for FS { let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata, + ); result } @@ -1959,7 +1974,11 @@ impl S3 for FS { version_id: req.input.version_id.clone(), })); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &object, + rustfs_scanner::SegmentInvalidationProducerIdentity::ObjectMetadata, + ); let duration = start_time.elapsed(); histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put").record(duration.as_secs_f64()); result