diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index c42055f6f..e5461df9d 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -96,13 +96,14 @@ pub use scanner::{ scanner_topology_digest, }; pub use scanner_io::{ - ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, - 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, replay_durable_dirty_usage_producer_record, - scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, + ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageSnapshot, + ScannerDirtyUsageState, 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, + replay_durable_dirty_usage_producer_record, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, + scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer, }; 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 2b64d2ce4..fc52ba892 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -1556,13 +1556,14 @@ pub(crate) use cache::{ current_cache_root_or_prepare_with_generation, }; pub use dirty_usage::{ - ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, - 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, replay_durable_dirty_usage_producer_record, - scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation, + ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageSnapshot, + ScannerDirtyUsageState, 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, + replay_durable_dirty_usage_producer_record, scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, + scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer, }; #[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 ee6a51a88..a0bd24554 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -14,6 +14,7 @@ /// process-wide dirty-usage invalidation state, its acknowledgment protocol, and snapshot helpers. use super::*; use std::collections::BTreeSet; +use std::sync::RwLock as StdRwLock; pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); @@ -27,6 +28,8 @@ pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock> = LazyLock::new(|| StdMutex::new(BTreeMap::new())); +static DIRTY_USAGE_CLEAR_OBSERVER: LazyLock>> = + LazyLock::new(|| StdRwLock::new(None)); pub(super) static DIRTY_USAGE_PRODUCER_COVERAGE: AtomicU64 = AtomicU64::new(0); 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::())); @@ -49,6 +52,8 @@ pub struct ScannerDirtyUsageBucket { pub generation: u64, } +pub type ScannerDirtyUsageClearObserver = Arc) + Send + Sync + 'static>; + /// A non-durable optimization hint for a dirty bucket. /// /// A whole-bucket marker always wins over narrow path hints. The scanner never @@ -98,6 +103,28 @@ impl DirtyUsageProducerEvidence { } } +pub fn set_scanner_dirty_usage_clear_observer( + observer: Option, +) -> Option { + let mut slot = DIRTY_USAGE_CLEAR_OBSERVER + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + std::mem::replace(&mut *slot, observer) +} + +fn notify_dirty_usage_clear(cleared: Vec) { + if cleared.is_empty() { + return; + } + let observer = DIRTY_USAGE_CLEAR_OBSERVER + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + if let Some(observer) = observer { + observer(cleared); + } +} + /// A point-in-time view of the local dirty bucket generations. /// /// `complete == false` is an all-or-nothing overflow signal: `buckets` is @@ -309,7 +336,7 @@ pub fn acknowledge_scoped_dirty_usage( ) -> std::result::Result { // Lock order: sorted bucket lifecycle/metadata fences (caller), then dirty map. // No await or storage operation occurs while the dirty map is locked. - let (cleared, pending) = { + let (cleared, pending, cleared_buckets) = { let mut dirty = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); let mut producer_identities = dirty_usage_producer_identities(); @@ -322,6 +349,18 @@ pub fn acknowledge_scoped_dirty_usage( .map_err(|_| ScannerDirtyUsageAckError::IncarnationUnavailable) }) .collect::, _>>()?; + let cleared_buckets = if probe_only { + Vec::new() + } else { + checked + .iter() + .filter(|(bucket, generation)| dirty.get(*bucket) == Some(generation)) + .map(|(bucket, generation)| ScannerDirtyUsageBucket { + bucket: (*bucket).to_string(), + generation: *generation, + }) + .collect::>() + }; let cleared = apply_scoped_dirty_usage_ack( instance_id, scanner_activity_epoch(), @@ -335,8 +374,9 @@ pub fn acknowledge_scoped_dirty_usage( producer_identities.retain(|bucket, _| dirty.contains_key(bucket)); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); } - (cleared, dirty.len()) + (cleared, dirty.len(), cleared_buckets) }; + notify_dirty_usage_clear(cleared_buckets); if !probe_only { global_metrics().record_scanner_dirty_usage_cycle_clear(usize_to_u64_saturated(cleared), usize_to_u64_saturated(pending)); } @@ -911,7 +951,7 @@ pub fn acknowledge_dirty_usage_generation( return Err(ScannerDirtyUsageAckError::ProcessChanged); } - let (cleared_buckets, pending_buckets) = { + let (cleared_buckets, pending_buckets, cleared) = { let mut dirty_buckets = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); let mut producer_identities = dirty_usage_producer_identities(); @@ -921,6 +961,14 @@ pub fn acknowledge_dirty_usage_generation( } let before = dirty_buckets.len(); + let cleared = dirty_buckets + .iter() + .filter(|(_, dirty_generation)| **dirty_generation <= generation) + .map(|(bucket, dirty_generation)| ScannerDirtyUsageBucket { + bucket: bucket.clone(), + generation: *dirty_generation, + }) + .collect::>(); dirty_buckets.retain(|_, dirty_generation| *dirty_generation > generation); dirty_scopes.retain(|bucket, _| dirty_buckets.contains_key(bucket)); producer_identities.retain(|bucket, _| dirty_buckets.contains_key(bucket)); @@ -928,8 +976,9 @@ pub fn acknowledge_dirty_usage_generation( if cleared_buckets > 0 { advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); } - (cleared_buckets, dirty_buckets.len()) + (cleared_buckets, dirty_buckets.len(), cleared) }; + notify_dirty_usage_clear(cleared); global_metrics() .record_scanner_dirty_usage_cycle_clear(usize_to_u64_saturated(cleared_buckets), usize_to_u64_saturated(pending_buckets)); Ok(()) @@ -944,17 +993,23 @@ pub fn clear_dirty_usage_bucket(bucket: &str) { return; } - let pending_buckets = { + let (pending_buckets, cleared) = { let mut dirty_buckets = dirty_usage_buckets(); let mut dirty_scopes = dirty_usage_bucket_scopes(); let mut producer_identities = dirty_usage_producer_identities(); - dirty_buckets.remove(bucket); + let cleared = dirty_buckets.remove(bucket).map(|generation| ScannerDirtyUsageBucket { + bucket: bucket.to_string(), + generation, + }); dirty_scopes.remove(bucket); producer_identities.remove(bucket); DIRTY_USAGE_PRODUCER_COVERAGE.store(0, Ordering::Release); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); - dirty_buckets.len() + (dirty_buckets.len(), cleared) }; + if let Some(cleared) = cleared { + notify_dirty_usage_clear(vec![cleared]); + } global_metrics().record_scanner_dirty_usage_clear(usize_to_u64_saturated(pending_buckets)); } @@ -1010,24 +1065,30 @@ pub(crate) async fn dirty_usage_bucket_notified() { } pub(super) fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) { - let (cleared_buckets, pending_buckets) = { + let (cleared_buckets, pending_buckets, cleared) = { 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 cleared_buckets = 0usize; + let mut cleared = Vec::new(); for (bucket, generation) in snapshot { if dirty_buckets.get(bucket).is_some_and(|current| current == generation) { dirty_buckets.remove(bucket); dirty_scopes.remove(bucket); cleared_buckets += 1; + cleared.push(ScannerDirtyUsageBucket { + bucket: bucket.clone(), + generation: *generation, + }); } } if cleared_buckets > 0 { producer_identities.retain(|bucket, _| dirty_buckets.contains_key(bucket)); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); } - (cleared_buckets, dirty_buckets.len()) + (cleared_buckets, dirty_buckets.len(), cleared) }; + notify_dirty_usage_clear(cleared); global_metrics() .record_scanner_dirty_usage_cycle_clear(usize_to_u64_saturated(cleared_buckets), usize_to_u64_saturated(pending_buckets)); } diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index f1fe52979..af98b5bb3 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -38,6 +38,7 @@ mod minio_generated_read_test; #[cfg(test)] mod multi_factor_scheduler_integration_test; pub(crate) mod runtime_sources; +pub(crate) mod scanner_dirty_journal; #[cfg(test)] mod sse_test; pub(crate) mod storage_api; diff --git a/rustfs/src/storage/scanner_dirty_journal.rs b/rustfs/src/storage/scanner_dirty_journal.rs new file mode 100644 index 000000000..5ab3a1662 --- /dev/null +++ b/rustfs/src/storage/scanner_dirty_journal.rs @@ -0,0 +1,597 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::{ECStore, Error, ecstore_config}; +use rustfs_scanner::{ + ScannerDirtyUsageBucket, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayRecord, + ScannerDurableDirtyUsageReplayScope, SegmentInvalidationProducerIdentity, +}; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; +use tracing::{debug, warn}; + +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_SCANNER: &str = "scanner"; +const EVENT_SCANNER_DIRTY_USAGE_JOURNAL: &str = "scanner_dirty_usage_journal"; +const DURABLE_DIRTY_USAGE_REPLAY_OBJECT: &str = "scanner/durable-dirty-producer-replay.json"; +const DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES: usize = 64 * 1024; +const DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES: usize = 1024; +const DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES: usize = 128; +const DURABLE_DIRTY_USAGE_JOURNAL_CHANNEL_DEPTH: usize = 256; +const DURABLE_DIRTY_USAGE_JOURNAL_RETRY_INTERVAL: Duration = Duration::from_secs(5); + +#[derive(Clone)] +pub(crate) struct DurableDirtyUsageJournal { + sender: mpsc::Sender, +} + +impl DurableDirtyUsageJournal { + pub(crate) fn record_committed_mutation(&self, bucket: &str, object: &str, producer: SegmentInvalidationProducerIdentity) { + let snapshot = rustfs_scanner::scanner_dirty_usage_snapshot(DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES); + let generation = snapshot + .buckets + .into_iter() + .find(|entry| entry.bucket == bucket) + .map(|entry| entry.generation) + .unwrap_or(0); + self.dispatch(JournalCommand::RecordMutation { + bucket: bucket.to_string(), + object: object.to_string(), + generation, + producer, + }); + } + + pub(crate) fn clear_confirmed_buckets(&self, cleared: Vec) { + self.dispatch(JournalCommand::ClearBuckets { cleared }); + } + + fn dispatch(&self, command: JournalCommand) { + match self.sender.try_send(command) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(command)) => { + let sender = self.sender.clone(); + tokio::spawn(async move { + let _ = sender.send(command).await; + }); + } + Err(mpsc::error::TrySendError::Closed(_)) => {} + } + } +} + +#[derive(Debug)] +enum JournalCommand { + RecordMutation { + bucket: String, + object: String, + generation: u64, + producer: SegmentInvalidationProducerIdentity, + }, + ClearBuckets { + cleared: Vec, + }, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct JournalState { + buckets: BTreeMap, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct JournalBucketState { + generation: u64, + scope: JournalScope, + producers: BTreeSet, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum JournalScope { + WholeBucket, + TopLevelEntries(BTreeSet), +} + +impl JournalState { + fn record_mutation(&mut self, bucket: String, object: &str, generation: u64, producer: SegmentInvalidationProducerIdentity) { + if bucket.is_empty() || generation == 0 || generation == u64::MAX || !durable_dirty_usage_producer_is_supported(producer) + { + self.buckets.clear(); + return; + } + + let event_scope = durable_dirty_usage_journal_scope(object); + self.buckets + .entry(bucket) + .and_modify(|state| { + state.generation = state.generation.max(generation); + merge_journal_scope(&mut state.scope, event_scope.clone()); + state.producers.insert(producer); + }) + .or_insert_with(|| JournalBucketState { + generation, + scope: event_scope, + producers: BTreeSet::from([producer]), + }); + } + + fn clear_buckets(&mut self, cleared: &[ScannerDirtyUsageBucket]) { + for entry in cleared { + if self + .buckets + .get(&entry.bucket) + .is_some_and(|state| state.generation <= entry.generation) + { + self.buckets.remove(&entry.bucket); + } + } + } + + fn replay_entries(&self) -> Option> { + if self.buckets.is_empty() || self.buckets.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES { + return None; + } + let mut entries = Vec::with_capacity(self.buckets.len()); + for (bucket, state) in &self.buckets { + if state.producers.is_empty() { + return None; + } + let scope = match &state.scope { + JournalScope::WholeBucket => ScannerDurableDirtyUsageReplayScope::WholeBucket, + JournalScope::TopLevelEntries(entries) => { + if entries.is_empty() || entries.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES { + return None; + } + ScannerDurableDirtyUsageReplayScope::TopLevelEntries { + entries: entries.clone(), + } + } + }; + entries.push(ScannerDurableDirtyUsageReplayEntry { + bucket: bucket.clone(), + generation: state.generation, + scope, + producers: state.producers.clone(), + }); + } + Some(entries) + } + + fn from_replay_record(record: ScannerDurableDirtyUsageReplayRecord) -> Option { + if record.entries.is_empty() || record.entries.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_ENTRIES { + return None; + } + let mut state = JournalState::default(); + for entry in record.entries { + if entry.bucket.is_empty() + || entry.bucket.contains(['/', '\\', '\0']) + || entry.bucket == "." + || entry.bucket == ".." + || entry.generation == 0 + || entry.generation == u64::MAX + || entry.producers.is_empty() + || entry + .producers + .iter() + .any(|producer| !durable_dirty_usage_producer_is_supported(*producer)) + { + return None; + } + let scope = match entry.scope { + ScannerDurableDirtyUsageReplayScope::WholeBucket => JournalScope::WholeBucket, + ScannerDurableDirtyUsageReplayScope::TopLevelEntries { entries } => { + if entries.is_empty() || entries.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES { + return None; + } + if entries + .iter() + .any(|entry| durable_dirty_usage_top_level_entry(entry).as_deref() != Some(entry.as_str())) + { + return None; + } + JournalScope::TopLevelEntries(entries) + } + }; + if state + .buckets + .insert( + entry.bucket, + JournalBucketState { + generation: entry.generation, + scope, + producers: entry.producers, + }, + ) + .is_some() + { + return None; + } + } + Some(state) + } +} + +pub(crate) async fn start_durable_dirty_usage_journal(store: Arc) -> DurableDirtyUsageJournal { + let initial_state = replay_durable_dirty_usage_journal(store.clone()).await; + let (sender, receiver) = mpsc::channel(DURABLE_DIRTY_USAGE_JOURNAL_CHANNEL_DEPTH); + tokio::spawn(run_durable_dirty_usage_journal(store, receiver, initial_state)); + DurableDirtyUsageJournal { sender } +} + +async fn replay_durable_dirty_usage_journal(store: Arc) -> JournalState { + let bytes = match super::read_config(store, DURABLE_DIRTY_USAGE_REPLAY_OBJECT).await { + Ok(bytes) if !bytes.is_empty() => bytes, + Ok(_) | Err(Error::ConfigNotFound) => return JournalState::default(), + Err(err) => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "read_failed", + error = ?err, + "Durable scanner dirty usage journal could not be read" + ); + return JournalState::default(); + } + }; + if bytes.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "oversized", + "Durable scanner dirty usage journal exceeds replay size limit" + ); + return JournalState::default(); + } + match rustfs_scanner::replay_durable_dirty_usage_producer_record(&bytes) { + Ok(state) => { + debug!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "replayed", + generation = state.generation, + pending = state.pending, + "Durable scanner dirty usage journal replayed" + ); + match serde_json::from_slice::(&bytes) + .ok() + .and_then(JournalState::from_replay_record) + { + Some(state) => state, + None => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "hydrate_rejected", + "Durable scanner dirty usage journal could not hydrate writer state" + ); + JournalState::default() + } + } + } + Err(err) => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "replay_rejected", + error = ?err, + "Durable scanner dirty usage journal was rejected" + ); + JournalState::default() + } + } +} + +async fn run_durable_dirty_usage_journal( + store: Arc, + mut receiver: mpsc::Receiver, + initial_state: JournalState, +) { + let mut state = initial_state; + let mut dirty = false; + let mut retry = tokio::time::interval(DURABLE_DIRTY_USAGE_JOURNAL_RETRY_INTERVAL); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + tokio::select! { + command = receiver.recv() => { + let Some(command) = command else { + break; + }; + apply_journal_command(&mut state, command); + while let Ok(command) = receiver.try_recv() { + apply_journal_command(&mut state, command); + } + dirty = true; + } + _ = retry.tick(), if dirty => {} + } + + if dirty && flush_durable_dirty_usage_journal(store.clone(), &state).await { + dirty = false; + } + } +} + +fn apply_journal_command(state: &mut JournalState, command: JournalCommand) { + match command { + JournalCommand::RecordMutation { + bucket, + object, + generation, + producer, + } => state.record_mutation(bucket, &object, generation, producer), + JournalCommand::ClearBuckets { cleared } => state.clear_buckets(&cleared), + } +} + +async fn flush_durable_dirty_usage_journal(store: Arc, state: &JournalState) -> bool { + let Some(entries) = state.replay_entries() else { + return delete_durable_dirty_usage_journal(store).await; + }; + let bytes = match rustfs_scanner::encode_durable_dirty_usage_producer_replay_record(entries) { + Ok(bytes) => bytes, + Err(err) => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "encode_rejected", + error = ?err, + "Durable scanner dirty usage journal could not be encoded" + ); + return delete_durable_dirty_usage_journal(store).await; + } + }; + match ecstore_config::com::save_config(store, DURABLE_DIRTY_USAGE_REPLAY_OBJECT, bytes).await { + Ok(()) => true, + Err(err) => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "write_failed", + error = ?err, + "Durable scanner dirty usage journal write failed" + ); + false + } + } +} + +async fn delete_durable_dirty_usage_journal(store: Arc) -> bool { + match ecstore_config::com::delete_config(store, DURABLE_DIRTY_USAGE_REPLAY_OBJECT).await { + Ok(()) | Err(Error::ConfigNotFound) => true, + Err(err) => { + warn!( + event = EVENT_SCANNER_DIRTY_USAGE_JOURNAL, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SCANNER, + state = "delete_failed", + error = ?err, + "Durable scanner dirty usage journal delete failed" + ); + false + } + } +} + +fn durable_dirty_usage_journal_scope(object: &str) -> JournalScope { + match durable_dirty_usage_top_level_entry(object) { + Some(entry) => JournalScope::TopLevelEntries(BTreeSet::from([entry])), + None => JournalScope::WholeBucket, + } +} + +fn merge_journal_scope(current: &mut JournalScope, incoming: JournalScope) { + let overflowed = match (&mut *current, incoming) { + (JournalScope::WholeBucket, _) | (_, JournalScope::WholeBucket) => { + *current = JournalScope::WholeBucket; + false + } + (JournalScope::TopLevelEntries(entries), JournalScope::TopLevelEntries(incoming)) => { + entries.extend(incoming); + entries.len() > DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES + } + }; + if overflowed { + *current = JournalScope::WholeBucket; + } +} + +fn durable_dirty_usage_producer_is_supported(producer: SegmentInvalidationProducerIdentity) -> bool { + matches!( + producer, + SegmentInvalidationProducerIdentity::PutObject + | SegmentInvalidationProducerIdentity::DeleteObject + | SegmentInvalidationProducerIdentity::DeleteMarker + | SegmentInvalidationProducerIdentity::CompleteMultipartUpload + | SegmentInvalidationProducerIdentity::AbortMultipartUpload + | SegmentInvalidationProducerIdentity::ObjectMetadata + | SegmentInvalidationProducerIdentity::BucketMetadata + | SegmentInvalidationProducerIdentity::Replication + | SegmentInvalidationProducerIdentity::TierTransition + | SegmentInvalidationProducerIdentity::TierExpiration + | SegmentInvalidationProducerIdentity::DirectoryObject + ) +} + +fn durable_dirty_usage_top_level_entry(object: &str) -> Option { + let (top_level_entry, _) = object.split_once('/').unwrap_or((object, "")); + (!top_level_entry.is_empty() + && top_level_entry != "." + && top_level_entry != ".." + && !object.starts_with('/') + && !top_level_entry.contains(['\\', '\0'])) + .then(|| top_level_entry.to_string()) +} + +#[cfg(test)] +mod tests { + use super::{DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES, JournalState}; + use rustfs_scanner::{ + ScannerDirtyUsageBucket, ScannerDurableDirtyUsageReplayEntry, ScannerDurableDirtyUsageReplayRecord, + ScannerDurableDirtyUsageReplayScope, SegmentInvalidationProducerIdentity, + }; + use std::collections::BTreeSet; + + #[test] + fn journal_state_merges_scopes_and_clears_only_confirmed_generations() { + let mut state = JournalState::default(); + state.record_mutation("photos".to_string(), "2026/object-a", 7, SegmentInvalidationProducerIdentity::Replication); + state.record_mutation( + "photos".to_string(), + "archive/object-b", + 9, + SegmentInvalidationProducerIdentity::TierExpiration, + ); + + let entries = state.replay_entries().expect("journal should encode a bounded bucket"); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].generation, 9); + assert_eq!( + entries[0].producers, + BTreeSet::from([ + SegmentInvalidationProducerIdentity::Replication, + SegmentInvalidationProducerIdentity::TierExpiration + ]) + ); + assert_eq!( + entries[0].scope, + ScannerDurableDirtyUsageReplayScope::TopLevelEntries { + entries: BTreeSet::from(["2026".to_string(), "archive".to_string()]) + } + ); + + state.clear_buckets(&[ScannerDirtyUsageBucket { + bucket: "photos".to_string(), + generation: 7, + }]); + assert!(state.buckets.contains_key("photos")); + state.clear_buckets(&[ScannerDirtyUsageBucket { + bucket: "photos".to_string(), + generation: 9, + }]); + assert!(state.buckets.is_empty()); + } + + #[test] + fn journal_state_expands_to_whole_bucket_for_ambiguous_or_overflowing_scopes() { + let mut ambiguous = JournalState::default(); + ambiguous.record_mutation("photos".to_string(), "../bad", 7, SegmentInvalidationProducerIdentity::Replication); + assert_eq!( + ambiguous.replay_entries().expect("ambiguous object should still encode")[0].scope, + ScannerDurableDirtyUsageReplayScope::WholeBucket + ); + + let mut overflow = JournalState::default(); + for index in 0..=DURABLE_DIRTY_USAGE_REPLAY_MAX_TOP_LEVEL_ENTRIES { + overflow.record_mutation( + "photos".to_string(), + &format!("prefix-{index}/object"), + u64::try_from(index + 1).expect("test index fits in u64"), + SegmentInvalidationProducerIdentity::Replication, + ); + } + assert_eq!( + overflow.replay_entries().expect("overflow should encode as whole bucket")[0].scope, + ScannerDurableDirtyUsageReplayScope::WholeBucket + ); + } + + #[test] + fn journal_state_hydrates_replayed_records_before_later_flushes() { + let record = ScannerDurableDirtyUsageReplayRecord { + schema: 1, + cache_key_format: 1, + writer_epoch: "writer".to_string(), + entries: vec![ScannerDurableDirtyUsageReplayEntry { + bucket: "photos".to_string(), + generation: 7, + scope: ScannerDurableDirtyUsageReplayScope::TopLevelEntries { + entries: BTreeSet::from(["2026".to_string()]), + }, + producers: BTreeSet::from([SegmentInvalidationProducerIdentity::Replication]), + }], + }; + let mut state = JournalState::from_replay_record(record).expect("valid replay record should hydrate"); + state.record_mutation( + "videos".to_string(), + "clips/object", + 8, + SegmentInvalidationProducerIdentity::TierExpiration, + ); + + let entries = state.replay_entries().expect("hydrated state should remain encodable"); + assert_eq!(entries.len(), 2); + assert!(entries.iter().any(|entry| entry.bucket == "photos" && entry.generation == 7)); + assert!(entries.iter().any(|entry| entry.bucket == "videos" && entry.generation == 8)); + } + + #[test] + fn journal_state_hydration_preserves_replayed_bucket_scope_and_producers() { + let record = ScannerDurableDirtyUsageReplayRecord { + schema: 1, + cache_key_format: 1, + writer_epoch: "writer".to_string(), + entries: vec![ScannerDurableDirtyUsageReplayEntry { + bucket: "photos".to_string(), + generation: 7, + scope: ScannerDurableDirtyUsageReplayScope::WholeBucket, + producers: BTreeSet::from([SegmentInvalidationProducerIdentity::Replication]), + }], + }; + let mut state = JournalState::from_replay_record(record).expect("valid replay record should hydrate"); + state.record_mutation( + "photos".to_string(), + "2026/object", + 8, + SegmentInvalidationProducerIdentity::TierExpiration, + ); + + let entries = state.replay_entries().expect("hydrated state should remain encodable"); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].bucket, "photos"); + assert_eq!(entries[0].generation, 8); + assert_eq!(entries[0].scope, ScannerDurableDirtyUsageReplayScope::WholeBucket); + assert_eq!( + entries[0].producers, + BTreeSet::from([ + SegmentInvalidationProducerIdentity::Replication, + SegmentInvalidationProducerIdentity::TierExpiration + ]) + ); + } + + #[test] + fn journal_state_drops_invalid_generation_or_non_production_identity_fail_closed() { + let mut state = JournalState::default(); + state.record_mutation("photos".to_string(), "2026/object", 7, SegmentInvalidationProducerIdentity::Replication); + state.record_mutation( + "videos".to_string(), + "clip/object", + u64::MAX, + SegmentInvalidationProducerIdentity::Replication, + ); + assert!(state.buckets.is_empty()); + + state.record_mutation("photos".to_string(), "2026/object", 7, SegmentInvalidationProducerIdentity::Unknown); + assert!(state.buckets.is_empty()); + } +} diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 4449c0dd7..fce77106d 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -954,7 +954,9 @@ 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 durable_dirty_usage_journal = super::scanner_dirty_journal::start_durable_dirty_usage_journal(store.clone()).await; + let journal_writer = durable_dirty_usage_journal.clone(); + ecstore_bucket::replication::set_scanner_dirty_usage_mutation_observer(Some(Arc::new(move |bucket, object, source| { let producer = match source { ecstore_bucket::replication::ScannerDirtyUsageMutationSource::Replication => { rustfs_scanner::SegmentInvalidationProducerIdentity::Replication @@ -964,6 +966,10 @@ pub(crate) async fn init_background_replication(store: Arc) { } }; rustfs_scanner::record_dirty_usage_object_from_producer(bucket, object, producer); + journal_writer.record_committed_mutation(bucket, object, producer); + }))); + rustfs_scanner::set_scanner_dirty_usage_clear_observer(Some(Arc::new(move |cleared| { + durable_dirty_usage_journal.clear_confirmed_buckets(cleared); }))); ecstore_bucket::replication::init_background_replication(store).await; }