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; }