gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 21:25:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): tag dirty usage producer identities (#7454)

Browse Source 
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 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 10:51:13 +08:00
committed by GitHub
parent be5d14c985
commit 590adad5ae
7 changed files with 111 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/lib.rs
+4 -2
View File
@@ -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;
crates/scanner/src/scanner_io.rs
+4 -3
View File
@@ -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};
crates/scanner/src/scanner_io/dirty_usage.rs
+82
View File
@@ -22,6 +22,11 @@ pub(super) static DIRTY_USAGE_BUCKETS: LazyLock<StdMutex<DirtyUsageBuckets>> = L
// matching scope.
pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock<StdMutex<DirtyUsageBucketScopes>> =
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<BTreeSet<crate::segment_invalidation::SegmentInvalidationProducerIdentity>>,
> = LazyLock::new(|| StdMutex::new(BTreeSet::new()));
pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
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<crate::segment_invalidation::SegmentInvalidationProducerIdentity>> {
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<crate::segment_invalidation::SegmentInvalidationProducerIdentity> {
dirty_usage_producer_identities().clone()
}
fn dirty_usage_top_level_entry(object: &str) -> Option<String> {
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)]
rustfs/src/app/object/copy.rs
+5 -1
View File
@@ -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))
}
});
rustfs/src/app/object/delete.rs
+6 -1
View File
@@ -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
}
}
rustfs/src/app/object/internal_put.rs
+5 -1
View File
@@ -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)
}
});
rustfs/src/app/object/put.rs
+5 -1
View File
@@ -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);