From 354e49de4cb4344a77b43fccc4d2e3710c9497c8 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 23:39:05 +0800 Subject: [PATCH] feat(scanner): implement V2 evidence preflights (#7523) * feat(scanner): wire dirty usage producer identities Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(scanner): add segment activation preflight proof Keep scanner segment reuse behind a structured activation preflight so release evidence can prove the production gate remains disabled until every producer, generation, overflow, cold-oracle, and distributed invalidation check is satisfied. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * feat(scanner): expose distributed invalidation evidence Record an explicit distributed segment invalidation evidence summary when remote dirty usage snapshots are bound to the current activity window and the authenticated scoped ACK capability probe succeeds. Reject peer dirty usage snapshots that contradict the peer activity pending bit so scoped ACKs fail closed instead of clearing an unadvertised remote mutation. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/scanner/src/lib.rs | 6 +- crates/scanner/src/scanner_io.rs | 127 ++++++++++++++++++- crates/scanner/src/scanner_io/dirty_usage.rs | 37 +++++- crates/scanner/src/scanner_io/io_cycle.rs | 3 + crates/scanner/src/scanner_io/tests.rs | 118 ++++++++++++++++- crates/scanner/src/segment_invalidation.rs | 24 +++- rustfs/src/app/bucket_usecase.rs | 20 ++- rustfs/src/app/multipart_usecase.rs | 12 +- rustfs/src/app/object/delete.rs | 19 ++- rustfs/src/app/object/extract.rs | 11 +- rustfs/src/app/object/internal_put.rs | 6 +- rustfs/src/app/object/restore.rs | 12 +- rustfs/src/storage/ecfs.rs | 29 ++++- 13 files changed, 393 insertions(+), 31 deletions(-) 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