From 590adad5aee8cb745006406578204d7db4eaba95 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 10:51:13 +0800 Subject: [PATCH] fix(scanner): tag dirty usage producer identities (#7454) Record production-facing segment invalidation producer identities when existing object-level dirty usage hooks observe PUT, CopyObject, DeleteObject/DeleteMarker, and CompleteMultipartUpload mutations. Keep the data non-authoritative and process-local so segment reuse activation still requires durable generation-window proof. Co-authored-by: zhi22915 --- crates/scanner/src/lib.rs | 6 +- crates/scanner/src/scanner_io.rs | 7 +- crates/scanner/src/scanner_io/dirty_usage.rs | 82 ++++++++++++++++++++ rustfs/src/app/object/copy.rs | 6 +- rustfs/src/app/object/delete.rs | 7 +- rustfs/src/app/object/internal_put.rs | 6 +- rustfs/src/app/object/put.rs | 6 +- 7 files changed, 111 insertions(+), 9 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index fff6245bd..0a9c2c2f0 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -97,9 +97,11 @@ 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_object, 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_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}; use std::sync::atomic::{AtomicU64, Ordering}; pub use storage_api::ScannerReplicationConfig as ReplicationConfig; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index a9f4e8eab..9bde466ce 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::{HashMap, HashSet}; +use std::collections::{BTreeSet, HashMap, HashSet}; use std::future::Future; use std::path::Path; use std::pin::Pin; @@ -1224,8 +1224,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_object, 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_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 6457f9585..4c9fdc837 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -22,6 +22,11 @@ pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = L // matching scope. 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_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); @@ -153,6 +158,7 @@ fn apply_scoped_dirty_usage_ack( #[cfg(test)] mod scoped_dirty_usage_tests { use super::*; + use crate::segment_invalidation::SegmentInvalidationProducerIdentity; #[test] fn scoped_dirty_usage_preserves_uncovered_newer_and_replayed_generations() { @@ -213,6 +219,30 @@ mod scoped_dirty_usage_tests { assert_eq!(scopes, original_scopes); } } + + #[test] + fn dirty_usage_tracks_known_segment_producer_identities_without_authorizing_unknown_sources() { + clear_dirty_usage_buckets_for_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); + + assert_eq!( + dirty_usage_producer_identities_for_tests(), + BTreeSet::from([ + SegmentInvalidationProducerIdentity::PutObject, + SegmentInvalidationProducerIdentity::DeleteObject + ]) + ); + assert_eq!( + dirty_usage_bucket_scopes_for_tests().get("photos"), + Some(&DirtyUsageBucketScope::WholeBucket), + "an unknown producer keeps the bucket dirty but must not count as producer coverage" + ); + + clear_dirty_usage_buckets_for_tests(); + assert!(dirty_usage_producer_identities_for_tests().is_empty()); + } } pub(super) fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> { @@ -225,6 +255,13 @@ fn dirty_usage_bucket_scopes() -> MutexGuard<'static, DirtyUsageBucketScopes> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn dirty_usage_producer_identities() +-> MutexGuard<'static, BTreeSet> { + DIRTY_USAGE_PRODUCER_IDENTITIES + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + pub(super) fn usize_to_u64_saturated(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } @@ -240,6 +277,22 @@ pub fn record_dirty_usage_bucket(bucket: &str) { return; } + record_dirty_usage_bucket_inner(bucket); +} + +pub fn record_dirty_usage_bucket_from_producer( + bucket: &str, + producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity, +) { + if bucket.is_empty() { + return; + } + + record_segment_invalidation_producer_identity(producer); + record_dirty_usage_bucket_inner(bucket); +} + +fn record_dirty_usage_bucket_inner(bucket: &str) { let pending_buckets = { let mut dirty_buckets = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); @@ -263,6 +316,23 @@ pub fn record_dirty_usage_bucket(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); +} + +pub fn record_dirty_usage_object_from_producer( + bucket: &str, + object: &str, + producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity, +) { + if bucket.is_empty() { + return; + } + + record_segment_invalidation_producer_identity(producer); + record_dirty_usage_object_inner(bucket, object); +} + +fn record_dirty_usage_object_inner(bucket: &str, object: &str) { let Some(top_level_entry) = dirty_usage_top_level_entry(object) else { record_dirty_usage_bucket(bucket); return; @@ -296,6 +366,17 @@ pub fn record_dirty_usage_object(bucket: &str, object: &str) { DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); } +fn record_segment_invalidation_producer_identity(producer: crate::segment_invalidation::SegmentInvalidationProducerIdentity) { + if producer.producer().is_some() { + dirty_usage_producer_identities().insert(producer); + } +} + +#[cfg(test)] +fn dirty_usage_producer_identities_for_tests() -> BTreeSet { + dirty_usage_producer_identities().clone() +} + fn dirty_usage_top_level_entry(object: &str) -> Option { let (top_level_entry, _) = object.split_once('/').unwrap_or((object, "")); (!top_level_entry.is_empty() @@ -577,6 +658,7 @@ pub(super) fn dirty_usage_bucket_count() -> usize { pub(crate) fn clear_dirty_usage_buckets_for_tests() { dirty_usage_buckets().clear(); dirty_usage_bucket_scopes().clear(); + dirty_usage_producer_identities().clear(); } #[cfg(test)] diff --git a/rustfs/src/app/object/copy.rs b/rustfs/src/app/object/copy.rs index 76192ca9d..10aa663ce 100644 --- a/rustfs/src/app/object/copy.rs +++ b/rustfs/src/app/object/copy.rs @@ -814,7 +814,11 @@ impl DefaultObjectUsecase { } } - rustfs_scanner::record_dirty_usage_object(&bucket, &key); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject, + ); Ok::<_, S3Error>((oi, dest_versioned)) } }); diff --git a/rustfs/src/app/object/delete.rs b/rustfs/src/app/object/delete.rs index eb19768ae..5b19409f1 100644 --- a/rustfs/src/app/object/delete.rs +++ b/rustfs/src/app/object/delete.rs @@ -1175,7 +1175,12 @@ impl DefaultObjectUsecase { let manager = get_capacity_manager(); manager.record_write_operation().await; let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_object(&bucket, &key); + let producer = if delete_marker && version_id_clone.is_none() { + rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteMarker + } else { + rustfs_scanner::SegmentInvalidationProducerIdentity::DeleteObject + }; + rustfs_scanner::record_dirty_usage_object_from_producer(&bucket, &key, producer); result } } diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs index 7f24c6e5c..99a7f6039 100644 --- a/rustfs/src/app/object/internal_put.rs +++ b/rustfs/src/app/object/internal_put.rs @@ -689,7 +689,11 @@ impl DefaultObjectUsecase { schedule_object_replication(obj_info.clone(), store, completion_replication_decision).await; } - rustfs_scanner::record_dirty_usage_object(&bucket, &key); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::CompleteMultipartUpload, + ); Ok::<_, ApiError>(obj_info) } }); diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index f7672369a..e4fd9a228 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -2054,7 +2054,11 @@ impl DefaultObjectUsecase { schedule_object_replication(obj_info.clone(), store, dsc).await; } - rustfs_scanner::record_dirty_usage_object(&bucket, &key); + rustfs_scanner::record_dirty_usage_object_from_producer( + &bucket, + &key, + rustfs_scanner::SegmentInvalidationProducerIdentity::PutObject, + ); rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now);