From 1c874137881d321426303665e9e8a1c10e7d6d44 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 9 Sep 2026 16:22:26 +0800 Subject: [PATCH] fix(scanner): mark committed replication and tier mutations (#7568) * fix(scanner): mark committed replication and tier mutations Bridge ECStore committed replication status writebacks and lifecycle tier expiration cleanup into scanner dirty-usage producer identity tracking. Only successful replication metadata updates mark Replication; validate-only, superseded, and retry paths remain unmarked. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * feat(scanner): replay durable dirty producer records Add a strict scanner dirty-usage producer replay schema that can restore typed producer coverage after restart without treating ordinary process-local mutations as durable evidence. Replay records validate schema, cache key format, generations, scopes, producer identities, duplicate buckets, byte limits, and entry limits before modifying dirty state. Keep segment activation fail-closed unless the restored bucket state came from durable replay; subsequent local mutations clear the durable authority bit for that bucket. Also fix the release scanner clippy lint in scoped cold-reuse proof creation. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/ecstore/src/api/mod.rs | 17 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 15 + .../src/bucket/lifecycle/runtime_boundary.rs | 5 + crates/ecstore/src/bucket/replication/mod.rs | 3 + .../replication/replication_resyncer.rs | 7 + .../bucket/replication/runtime_boundary.rs | 7 + crates/ecstore/src/runtime/sources.rs | 79 +++- crates/scanner/src/lib.rs | 8 +- crates/scanner/src/scanner_io.rs | 10 +- crates/scanner/src/scanner_io/dirty_usage.rs | 382 +++++++++++++++++- crates/scanner/src/scanner_io/io_cache.rs | 2 +- crates/scanner/src/segment_invalidation.rs | 3 +- rustfs/src/storage/storage_api.rs | 11 + 13 files changed, 510 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 08432c7e4..dcbab07f1 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -260,18 +260,19 @@ pub mod bucket { ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, - ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus, - VersionPurgeStatusType, XferStats, assign_site_replication_rule_priorities, commit_force_delete_intent, - complete_force_delete_intent, delete_replication_state_from_config, delete_replication_version_id, - get_global_replication_pool, get_global_replication_stats, get_proxy_targets, init_background_replication, + ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, ScannerDirtyUsageMutationObserver, + ScannerDirtyUsageMutationSource, TargetReplicationResyncStatus, VersionPurgeStatusType, XferStats, + assign_site_replication_rule_priorities, commit_force_delete_intent, complete_force_delete_intent, + delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, + get_global_replication_stats, get_proxy_targets, init_background_replication, invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arn_deployment_id, replication_target_arns, resync_start_conflict_id, - should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info, - should_use_existing_delete_replication_source, site_replication_rule_deployment_id, - unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, - version_purge_status_to_filemeta, + set_scanner_dirty_usage_mutation_observer, should_remove_replication_target, should_schedule_delete_replication, + should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, + site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 1d4cb1d56..93627d2ab 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -966,6 +966,11 @@ async fn cleanup_free_version_exact(api: Arc, oi: &ObjectInfo, cancel: if let Some(err) = first_error { return Err(err); } + runtime_sources::notify_scanner_dirty_usage_mutation( + &oi.bucket, + &oi.name, + runtime_sources::ScannerDirtyUsageMutationSource::TierExpiration, + ); Ok(true) } @@ -4749,6 +4754,11 @@ async fn expire_transitioned_object_with_lock_lost_signal( // Drop any cached restored-copy body so it does not sit resident // until TTL after the copy is expired (ODC-26). crate::object_api::notify_object_mutation(&oi.bucket, &oi.name).await; + runtime_sources::notify_scanner_dirty_usage_mutation( + &oi.bucket, + &oi.name, + runtime_sources::ScannerDirtyUsageMutationSource::TierExpiration, + ); //audit_log_lifecycle(*oi, ILMExpiry, tags, traceFn); Ok(dobj) } @@ -4784,6 +4794,11 @@ async fn expire_transitioned_object_with_lock_lost_signal( // The transitioned version is gone; evict any cached body for this object // so it does not linger until TTL (ODC-26). crate::object_api::notify_object_mutation(&oi.bucket, &oi.name).await; + runtime_sources::notify_scanner_dirty_usage_mutation( + &oi.bucket, + &oi.name, + runtime_sources::ScannerDirtyUsageMutationSource::TierExpiration, + ); //audit_log_lifecycle(oi, ILMExpiry, tags); diff --git a/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs b/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs index fcd886ca5..8e9057931 100644 --- a/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs +++ b/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs @@ -20,6 +20,7 @@ use tokio_util::sync::CancellationToken; use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}; use crate::runtime::sources; +pub(crate) use crate::runtime::sources::ScannerDirtyUsageMutationSource; use crate::services::tier::tier::TierConfigMgr; use crate::store::ECStore; @@ -54,3 +55,7 @@ pub(crate) fn deployment_id() -> Option { pub(crate) async fn bucket_lifecycle_config(bucket: &str) -> Option { sources::bucket_lifecycle_config(bucket).await } + +pub(crate) fn notify_scanner_dirty_usage_mutation(bucket: &str, object: &str, source: ScannerDirtyUsageMutationSource) { + sources::notify_scanner_dirty_usage_mutation(bucket, object, source); +} diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index bb9e002d0..031fc9d32 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -92,3 +92,6 @@ pub use replication_target_boundary::SsecPassthroughCapability; pub use replication_target_boundary::VersionIdentityCapability; pub use replication_target_boundary::{ObjectLockIntegrity, object_lock_put_integrity}; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; +pub use runtime_boundary::{ + ScannerDirtyUsageMutationObserver, ScannerDirtyUsageMutationSource, set_scanner_dirty_usage_mutation_observer, +}; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 1a8094465..bb768e974 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -3843,6 +3843,13 @@ async fn persist_replication_state_if_current( match storage.put_object_metadata(&roi.bucket, &roi.name, &write_opts).await { Ok(updated) => { *object_info = updated; + if mode == ReplicationStatusWritebackMode::Update { + runtime_sources::notify_scanner_dirty_usage_mutation( + &roi.bucket, + &roi.name, + runtime_sources::ScannerDirtyUsageMutationSource::Replication, + ); + } Ok(ReplicationStatePersistOutcome::Updated) } Err(Error::PreconditionFailed) => Ok(ReplicationStatePersistOutcome::Superseded), diff --git a/crates/ecstore/src/bucket/replication/runtime_boundary.rs b/crates/ecstore/src/bucket/replication/runtime_boundary.rs index d930c6119..36df30fc3 100644 --- a/crates/ecstore/src/bucket/replication/runtime_boundary.rs +++ b/crates/ecstore/src/bucket/replication/runtime_boundary.rs @@ -19,6 +19,9 @@ use super::replication_pool::DynReplicationPool; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::ReplicationObjectStore; use crate::runtime::sources; +pub use crate::runtime::sources::{ + ScannerDirtyUsageMutationObserver, ScannerDirtyUsageMutationSource, set_scanner_dirty_usage_mutation_observer, +}; pub(crate) fn object_store_handle() -> Option> { sources::object_store_handle() @@ -43,3 +46,7 @@ pub(crate) fn replication_runtime_initialized() -> bool { pub(crate) fn bucket_monitor() -> Option> { sources::bucket_monitor() } + +pub(crate) fn notify_scanner_dirty_usage_mutation(bucket: &str, object: &str, source: ScannerDirtyUsageMutationSource) { + sources::notify_scanner_dirty_usage_mutation(bucket, object, source); +} diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 5b9bada4c..54150d6ae 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -14,7 +14,7 @@ use std::{ collections::{HashMap, HashSet}, - sync::{Arc, OnceLock}, + sync::{Arc, LazyLock, OnceLock, RwLock as StdRwLock}, time::SystemTime, }; @@ -57,6 +57,13 @@ use uuid::Uuid; const TEST_RPC_SECRET: &str = "test-rpc-secret"; pub(crate) type WorkloadSnapshotProviderRef = Arc; +pub type ScannerDirtyUsageMutationObserver = Arc; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ScannerDirtyUsageMutationSource { + Replication, + TierExpiration, +} #[derive(Clone, Default)] pub(crate) struct LockRegistry { @@ -88,6 +95,8 @@ impl LockRegistry { } static WORKLOAD_ADMISSION_SNAPSHOT_PROVIDER: OnceLock = OnceLock::new(); +static SCANNER_DIRTY_USAGE_MUTATION_OBSERVER: LazyLock>> = + LazyLock::new(|| StdRwLock::new(None)); pub(crate) fn set_workload_admission_snapshot_provider( provider: WorkloadSnapshotProviderRef, @@ -99,6 +108,28 @@ pub(crate) fn workload_admission_snapshot_provider() -> Option, +) -> Option { + let mut slot = SCANNER_DIRTY_USAGE_MUTATION_OBSERVER + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + std::mem::replace(&mut *slot, observer) +} + +pub(crate) fn notify_scanner_dirty_usage_mutation(bucket: &str, object: &str, source: ScannerDirtyUsageMutationSource) { + if bucket.is_empty() || object.is_empty() { + return; + } + let observer = SCANNER_DIRTY_USAGE_MUTATION_OBSERVER + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + if let Some(observer) = observer { + observer(bucket, object, source); + } +} + pub(crate) fn record_erasure_write_quorum_failure(stage: &'static str, dominant_error: &'static str) { global_internode_metrics().record_erasure_write_quorum_failure(stage, dominant_error); } @@ -580,12 +611,16 @@ pub(crate) async fn init_tier_config_mgr(store: Arc) -> Result<()> { #[cfg(test)] mod tests { use super::{ - LockRegistry, clear_local_disk_id_map_for_test, local_disk_path_by_id, local_node_name, reconcile_local_disk_ids, - replace_local_disk_id, set_local_node_name, + LockRegistry, ScannerDirtyUsageMutationSource, clear_local_disk_id_map_for_test, local_disk_path_by_id, local_node_name, + notify_scanner_dirty_usage_mutation, reconcile_local_disk_ids, replace_local_disk_id, set_local_node_name, + set_scanner_dirty_usage_mutation_observer, }; use crate::disk::endpoint::Endpoint; use rustfs_lock::{LocalClient, LockClient}; - use std::{collections::HashMap, sync::Arc}; + use std::{ + collections::HashMap, + sync::{Arc, Mutex}, + }; use uuid::Uuid; fn url_endpoint(raw: &str) -> Endpoint { @@ -620,6 +655,42 @@ mod tests { assert!(Arc::ptr_eq(&clients[1], &client_b)); } + #[test] + #[serial_test::serial(scanner_dirty_usage_mutation_observer)] + fn scanner_dirty_usage_mutation_observer_filters_empty_identity_and_preserves_source() { + let observed = Arc::new(Mutex::new(Vec::new())); + let observed_clone = Arc::clone(&observed); + let previous = set_scanner_dirty_usage_mutation_observer(Some(Arc::new(move |bucket, object, source| { + observed_clone.lock().expect("observer lock should not be poisoned").push(( + bucket.to_string(), + object.to_string(), + source, + )); + }))); + + notify_scanner_dirty_usage_mutation("photos", "2026/image.jpg", ScannerDirtyUsageMutationSource::Replication); + notify_scanner_dirty_usage_mutation("", "2026/empty-bucket.jpg", ScannerDirtyUsageMutationSource::TierExpiration); + notify_scanner_dirty_usage_mutation("photos", "", ScannerDirtyUsageMutationSource::TierExpiration); + notify_scanner_dirty_usage_mutation("archive", "expired.bin", ScannerDirtyUsageMutationSource::TierExpiration); + set_scanner_dirty_usage_mutation_observer(previous); + + assert_eq!( + *observed.lock().expect("observer lock should not be poisoned"), + vec![ + ( + "photos".to_string(), + "2026/image.jpg".to_string(), + ScannerDirtyUsageMutationSource::Replication + ), + ( + "archive".to_string(), + "expired.bin".to_string(), + ScannerDirtyUsageMutationSource::TierExpiration + ), + ] + ); + } + #[tokio::test] #[serial_test::serial] async fn local_node_name_round_trips_through_common_runtime_helper() { diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 76ba60b97..5aa2e56c5 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -96,10 +96,12 @@ 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, + ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError, ScannerDurableDirtyUsageReplayRecord, + ScannerDurableDirtyUsageReplayScope, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, + clear_dirty_usage_bucket, encode_durable_dirty_usage_producer_replay_record, record_dirty_usage_bucket, 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, + record_dirty_usage_object_from_producer, record_scanner_maintenance_change, replay_durable_dirty_usage_producer_record, + 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 a95bc3a3d..2b64d2ce4 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -39,8 +39,6 @@ use s3s::dto::{ BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration, }; use sha2::{Digest as _, Sha256}; -#[cfg(test)] -use std::collections::BTreeSet; use std::collections::{BTreeMap, HashMap, HashSet}; use std::future::Future; use std::path::Path; @@ -1559,10 +1557,12 @@ 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, + ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayError, ScannerDurableDirtyUsageReplayRecord, + ScannerDurableDirtyUsageReplayScope, acknowledge_dirty_usage_generation, acknowledge_scoped_dirty_usage, + clear_dirty_usage_bucket, encode_durable_dirty_usage_producer_replay_record, record_dirty_usage_bucket, 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, + record_dirty_usage_object_from_producer, record_scanner_maintenance_change, replay_durable_dirty_usage_producer_record, + 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 4c4919ff5..ee6a51a88 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -13,6 +13,7 @@ // limitations under the License. /// process-wide dirty-usage invalidation state, its acknowledgment protocol, and snapshot helpers. use super::*; +use std::collections::BTreeSet; pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); @@ -31,6 +32,10 @@ pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock = LazyLock::new(No pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock = LazyLock::new(|| format!("{:032x}", rand::random::())); pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static SCANNER_MAINTENANCE_NOTIFY: LazyLock = LazyLock::new(Notify::new); +const SCANNER_DURABLE_DIRTY_USAGE_REPLAY_SCHEMA: u16 = 1; +const SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES: usize = 64 * 1024; +const SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES: usize = 1024; +const SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES: usize = 128; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct ScannerDirtyUsageState { @@ -64,6 +69,7 @@ pub(super) struct DirtyUsageProducerIdentityState { first_generation: u64, last_generation: u64, fully_identified: bool, + durable_replayed: bool, } pub(super) type DirtyUsageProducerIdentities = BTreeMap; @@ -115,6 +121,185 @@ pub enum ScannerDirtyUsageAckError { IncarnationUnavailable, } +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +pub enum ScannerDurableDirtyUsageReplayError { + #[error("scanner durable dirty usage replay record exceeds size limit")] + ByteLimit, + #[error("scanner durable dirty usage replay record cannot be decoded")] + InvalidJson, + #[error("scanner durable dirty usage replay schema is unsupported")] + UnsupportedSchema, + #[error("scanner durable dirty usage replay record is empty or over entry limit")] + EntryLimit, + #[error("scanner durable dirty usage replay record contains invalid generation")] + InvalidGeneration, + #[error("scanner durable dirty usage replay record contains an invalid bucket or scope")] + InvalidScope, + #[error("scanner durable dirty usage replay record contains an invalid producer identity")] + InvalidProducer, + #[error("scanner durable dirty usage replay record contains duplicate buckets")] + DuplicateBucket, + #[error("scanner durable dirty usage replay record belongs to a different cache key format")] + MixedVersion, +} + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScannerDurableDirtyUsageReplayRecord { + pub schema: u16, + pub cache_key_format: u16, + pub writer_epoch: String, + pub entries: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScannerDurableDirtyUsageReplayEntry { + pub bucket: String, + pub generation: u64, + pub scope: ScannerDurableDirtyUsageReplayScope, + pub producers: BTreeSet, +} + +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ScannerDurableDirtyUsageReplayScope { + WholeBucket, + TopLevelEntries { entries: BTreeSet }, +} + +pub fn encode_durable_dirty_usage_producer_replay_record( + entries: Vec, +) -> std::result::Result, ScannerDurableDirtyUsageReplayError> { + let record = ScannerDurableDirtyUsageReplayRecord { + schema: SCANNER_DURABLE_DIRTY_USAGE_REPLAY_SCHEMA, + cache_key_format: crate::DATA_USAGE_CACHE_KEY_FORMAT, + writer_epoch: scanner_activity_epoch().to_string(), + entries, + }; + validate_durable_dirty_usage_replay_record(&record)?; + serde_json::to_vec(&record).map_err(|_| ScannerDurableDirtyUsageReplayError::InvalidJson) +} + +pub fn replay_durable_dirty_usage_producer_record( + bytes: &[u8], +) -> std::result::Result { + if bytes.len() > SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES { + return Err(ScannerDurableDirtyUsageReplayError::ByteLimit); + } + let record: ScannerDurableDirtyUsageReplayRecord = + serde_json::from_slice(bytes).map_err(|_| ScannerDurableDirtyUsageReplayError::InvalidJson)?; + let replay = validate_durable_dirty_usage_replay_record(&record)?; + let max_generation = replay.iter().map(|entry| entry.generation).max().unwrap_or(0); + set_dirty_usage_generation_floor(max_generation)?; + let pending = { + let mut dirty_buckets = dirty_usage_buckets(); + let mut dirty_scopes = dirty_usage_bucket_scopes(); + let mut producer_identities = dirty_usage_producer_identities(); + let mut coverage = 0_u64; + for entry in replay { + dirty_buckets.insert(entry.bucket.clone(), entry.generation); + dirty_scopes.insert(entry.bucket.clone(), entry.scope); + coverage |= entry.coverage; + producer_identities.insert( + entry.bucket, + DirtyUsageProducerIdentityState { + first_generation: entry.generation, + last_generation: entry.generation, + fully_identified: true, + durable_replayed: true, + }, + ); + } + DIRTY_USAGE_PRODUCER_COVERAGE.fetch_or(coverage, Ordering::AcqRel); + dirty_buckets.len() + }; + global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending)); + DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); + Ok(scanner_dirty_usage_state()) +} + +#[derive(Debug)] +struct ValidatedDurableDirtyUsageReplayEntry { + bucket: String, + generation: u64, + scope: DirtyUsageBucketScope, + coverage: u64, +} + +fn validate_durable_dirty_usage_replay_record( + record: &ScannerDurableDirtyUsageReplayRecord, +) -> std::result::Result, ScannerDurableDirtyUsageReplayError> { + if record.schema != SCANNER_DURABLE_DIRTY_USAGE_REPLAY_SCHEMA { + return Err(ScannerDurableDirtyUsageReplayError::UnsupportedSchema); + } + if record.cache_key_format != crate::DATA_USAGE_CACHE_KEY_FORMAT { + return Err(ScannerDurableDirtyUsageReplayError::MixedVersion); + } + if record.writer_epoch.is_empty() || record.writer_epoch.contains('\0') { + return Err(ScannerDurableDirtyUsageReplayError::InvalidScope); + } + if record.entries.is_empty() || record.entries.len() > SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES { + return Err(ScannerDurableDirtyUsageReplayError::EntryLimit); + } + let mut buckets = HashSet::with_capacity(record.entries.len()); + let mut replay = Vec::with_capacity(record.entries.len()); + for entry in &record.entries { + if !valid_bucket_for_durable_dirty_usage_replay(&entry.bucket) { + return Err(ScannerDurableDirtyUsageReplayError::InvalidScope); + } + if !buckets.insert(entry.bucket.clone()) { + return Err(ScannerDurableDirtyUsageReplayError::DuplicateBucket); + } + if entry.generation == 0 || entry.generation == u64::MAX { + return Err(ScannerDurableDirtyUsageReplayError::InvalidGeneration); + } + let scope = validate_durable_dirty_usage_replay_scope(&entry.scope)?; + let mut coverage = 0_u64; + for producer in &entry.producers { + let Some(bit) = producer.production_coverage_bit() else { + return Err(ScannerDurableDirtyUsageReplayError::InvalidProducer); + }; + coverage |= bit; + } + if coverage == 0 { + return Err(ScannerDurableDirtyUsageReplayError::InvalidProducer); + } + replay.push(ValidatedDurableDirtyUsageReplayEntry { + bucket: entry.bucket.clone(), + generation: entry.generation, + scope, + coverage, + }); + } + Ok(replay) +} + +fn valid_bucket_for_durable_dirty_usage_replay(bucket: &str) -> bool { + !bucket.is_empty() && !bucket.contains(['/', '\\', '\0']) && bucket != "." && bucket != ".." +} + +fn validate_durable_dirty_usage_replay_scope( + scope: &ScannerDurableDirtyUsageReplayScope, +) -> std::result::Result { + match scope { + ScannerDurableDirtyUsageReplayScope::WholeBucket => Ok(DirtyUsageBucketScope::WholeBucket), + ScannerDurableDirtyUsageReplayScope::TopLevelEntries { entries } => { + if entries.is_empty() || entries.len() > SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES { + return Err(ScannerDurableDirtyUsageReplayError::InvalidScope); + } + let mut validated = HashSet::with_capacity(entries.len()); + for entry in entries { + if dirty_usage_top_level_entry(entry).as_deref() != Some(entry.as_str()) { + return Err(ScannerDurableDirtyUsageReplayError::InvalidScope); + } + validated.insert(entry.clone()); + } + Ok(DirtyUsageBucketScope::TopLevelEntries(validated)) + } + } +} + /// A scoped ACK requires storage-owned lifecycle and incarnation fences. /// Callers must only send ACKs backed by durable per-bucket publication. pub fn acknowledge_scoped_dirty_usage( @@ -302,6 +487,150 @@ mod scoped_dirty_usage_tests { clear_dirty_usage_buckets_for_tests(); assert!(dirty_usage_producer_identities_for_tests().is_empty()); } + + #[test] + #[serial] + fn durable_dirty_usage_replay_restores_producer_identity_authority() { + clear_dirty_usage_buckets_for_tests(); + let bytes = encode_durable_dirty_usage_producer_replay_record(vec![ScannerDurableDirtyUsageReplayEntry { + bucket: "photos".to_string(), + generation: 7, + scope: ScannerDurableDirtyUsageReplayScope::TopLevelEntries { + entries: BTreeSet::from(["2026".to_string(), "archive".to_string()]), + }, + producers: SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION.into_iter().collect(), + }]) + .expect("durable replay record should encode"); + + let state = replay_durable_dirty_usage_producer_record(&bytes).expect("durable replay should be accepted"); + assert!(state.pending); + assert!(state.generation >= 7); + assert_eq!( + dirty_usage_bucket_scopes_for_tests().get("photos"), + Some(&DirtyUsageBucketScope::TopLevelEntries(HashSet::from([ + "2026".to_string(), + "archive".to_string() + ]))) + ); + + let snapshot = snapshot_dirty_usage_buckets( + &[BucketInfo { + name: "photos".to_string(), + created: None, + deleted: None, + versioning: false, + object_locking: false, + }], + dirty_usage_generation(), + ); + let evidence = dirty_usage_producer_evidence(&snapshot); + assert!(evidence.producer_identity_coverage_complete); + assert!(evidence.durable_producer_identity); + assert!(evidence.restart_gap_absent); + assert_eq!(evidence.generation_start, 7); + assert_eq!(evidence.generation_end, 7); + + record_dirty_usage_object_from_producer("photos", "new/object", SegmentInvalidationProducerIdentity::PutObject); + let changed = snapshot_dirty_usage_buckets( + &[BucketInfo { + name: "photos".to_string(), + created: None, + deleted: None, + versioning: false, + object_locking: false, + }], + dirty_usage_generation(), + ); + let changed_evidence = dirty_usage_producer_evidence(&changed); + assert!(!changed_evidence.durable_producer_identity); + assert!(!changed_evidence.restart_gap_absent); + clear_dirty_usage_buckets_for_tests(); + } + + #[test] + #[serial] + fn durable_dirty_usage_replay_rejects_invalid_records_without_partial_state() { + let valid_entry = ScannerDurableDirtyUsageReplayEntry { + bucket: "photos".to_string(), + generation: 7, + scope: ScannerDurableDirtyUsageReplayScope::WholeBucket, + producers: BTreeSet::from([SegmentInvalidationProducerIdentity::PutObject]), + }; + let valid_record = || ScannerDurableDirtyUsageReplayRecord { + schema: SCANNER_DURABLE_DIRTY_USAGE_REPLAY_SCHEMA, + cache_key_format: crate::DATA_USAGE_CACHE_KEY_FORMAT, + writer_epoch: "writer".to_string(), + entries: vec![valid_entry.clone()], + }; + + let invalid_cases = [ + { + let mut record = valid_record(); + record.schema = SCANNER_DURABLE_DIRTY_USAGE_REPLAY_SCHEMA + 1; + ( + serde_json::to_vec(&record).expect("invalid schema record should encode"), + ScannerDurableDirtyUsageReplayError::UnsupportedSchema, + ) + }, + { + let mut record = valid_record(); + record.cache_key_format = crate::DATA_USAGE_CACHE_KEY_FORMAT + 1; + ( + serde_json::to_vec(&record).expect("mixed key format record should encode"), + ScannerDurableDirtyUsageReplayError::MixedVersion, + ) + }, + { + let mut record = valid_record(); + record.entries[0].producers = BTreeSet::from([SegmentInvalidationProducerIdentity::Unknown]); + ( + serde_json::to_vec(&record).expect("unknown producer record should encode"), + ScannerDurableDirtyUsageReplayError::InvalidProducer, + ) + }, + { + let mut record = valid_record(); + record.entries.push(valid_entry.clone()); + ( + serde_json::to_vec(&record).expect("duplicate bucket record should encode"), + ScannerDurableDirtyUsageReplayError::DuplicateBucket, + ) + }, + { + let mut record = valid_record(); + record.entries[0].scope = ScannerDurableDirtyUsageReplayScope::TopLevelEntries { + entries: BTreeSet::from(["bad/child".to_string()]), + }; + ( + serde_json::to_vec(&record).expect("invalid scope record should encode"), + ScannerDurableDirtyUsageReplayError::InvalidScope, + ) + }, + { + let mut record = valid_record(); + record.entries[0].generation = u64::MAX; + ( + serde_json::to_vec(&record).expect("invalid generation record should encode"), + ScannerDurableDirtyUsageReplayError::InvalidGeneration, + ) + }, + ]; + + for (bytes, expected_error) in invalid_cases { + clear_dirty_usage_buckets_for_tests(); + assert_eq!(replay_durable_dirty_usage_producer_record(&bytes), Err(expected_error)); + assert!(dirty_usage_buckets_for_tests().is_empty()); + assert!(dirty_usage_bucket_scopes_for_tests().is_empty()); + } + + clear_dirty_usage_buckets_for_tests(); + let oversized = vec![b' '; SCANNER_DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES + 1]; + assert_eq!( + replay_durable_dirty_usage_producer_record(&oversized), + Err(ScannerDurableDirtyUsageReplayError::ByteLimit) + ); + assert!(dirty_usage_buckets_for_tests().is_empty()); + } } pub(super) fn dirty_usage_buckets() -> MutexGuard<'static, DirtyUsageBuckets> { @@ -330,6 +659,18 @@ pub(super) fn advance_generation(generation: &AtomicU64) -> u64 { .map_or_else(|current| current, |previous| previous.saturating_add(1)) } +fn set_dirty_usage_generation_floor(floor: u64) -> std::result::Result<(), ScannerDurableDirtyUsageReplayError> { + if floor == 0 || floor == u64::MAX { + return Err(ScannerDurableDirtyUsageReplayError::InvalidGeneration); + } + DIRTY_USAGE_BUCKET_GENERATION + .try_update(Ordering::AcqRel, Ordering::Acquire, |current| { + (current != u64::MAX).then_some(current.max(floor)) + }) + .map(|_| ()) + .map_err(|_| ScannerDurableDirtyUsageReplayError::InvalidGeneration) +} + pub fn record_dirty_usage_bucket(bucket: &str) { if bucket.is_empty() { return; @@ -464,18 +805,23 @@ fn record_segment_invalidation_producer_identities_for_generation( fully_identified &= event_coverage != 0; DIRTY_USAGE_PRODUCER_COVERAGE.fetch_or(event_coverage, Ordering::AcqRel); - if let Some(state) = identities.get_mut(bucket) { - state.last_generation = state.last_generation.max(generation); - state.fully_identified &= fully_identified; - } else { - identities.insert( - bucket.to_string(), - DirtyUsageProducerIdentityState { - first_generation: generation, - last_generation: generation, - fully_identified, - }, - ); + match identities.get_mut(bucket) { + Some(state) => { + state.last_generation = state.last_generation.max(generation); + state.fully_identified &= fully_identified; + state.durable_replayed = false; + } + None => { + identities.insert( + bucket.to_string(), + DirtyUsageProducerIdentityState { + first_generation: generation, + last_generation: generation, + fully_identified, + durable_replayed: false, + }, + ); + } } } @@ -786,6 +1132,11 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di && DIRTY_USAGE_PRODUCER_COVERAGE.load(Ordering::Acquire) & crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK == crate::segment_invalidation::SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION_COVERAGE_MASK; + let durable_producer_identity = producer_identity_coverage_complete + && snapshot + .buckets + .keys() + .all(|bucket| producer_identities.get(bucket).is_some_and(|state| state.durable_replayed)); let generation_window_bound = producer_identity_coverage_complete && generation_start != u64::MAX && generation_end >= generation_start @@ -793,11 +1144,8 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di 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, + durable_producer_identity, + restart_gap_absent: durable_producer_identity, generation_window_bound, generation_start: if generation_window_bound { generation_start } else { 0 }, generation_end: if generation_window_bound { generation_end } else { 0 }, diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index d1a8ea028..d99e889ad 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -71,7 +71,7 @@ pub(super) fn prepare_scoped_set_scan( } let unselected_bucket_incarnations = unselected_bucket_incarnation_bindings(old_cache, all_buckets, selected_buckets, current_bucket_incarnations)?; - let cold_bucket_reuse_proof = (!unselected_bucket_incarnations.is_empty()).then(|| ScopedColdBucketReuseProof { + let cold_bucket_reuse_proof = (!unselected_bucket_incarnations.is_empty()).then_some(ScopedColdBucketReuseProof { baseline_scan_plan_digest, source: generation.source, bucket_incarnations: unselected_bucket_incarnations, diff --git a/crates/scanner/src/segment_invalidation.rs b/crates/scanner/src/segment_invalidation.rs index dabc14ca3..308f3e6f6 100644 --- a/crates/scanner/src/segment_invalidation.rs +++ b/crates/scanner/src/segment_invalidation.rs @@ -51,7 +51,8 @@ impl SegmentInvalidationProducer { ]; } -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SegmentInvalidationProducerIdentity { PutObject, DeleteObject, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 4dbc1b341..59ddade5e 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -952,6 +952,17 @@ pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperti } pub(crate) async fn init_background_replication(store: Arc) { + ecstore_bucket::replication::set_scanner_dirty_usage_mutation_observer(Some(Arc::new(|bucket, object, source| { + let producer = match source { + ecstore_bucket::replication::ScannerDirtyUsageMutationSource::Replication => { + rustfs_scanner::SegmentInvalidationProducerIdentity::Replication + } + ecstore_bucket::replication::ScannerDirtyUsageMutationSource::TierExpiration => { + rustfs_scanner::SegmentInvalidationProducerIdentity::TierExpiration + } + }; + rustfs_scanner::record_dirty_usage_object_from_producer(bucket, object, producer); + }))); ecstore_bucket::replication::init_background_replication(store).await; }