gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): persist dirty producer journal (#7580)

Browse Source 
Persist scanner dirty producer replay records from committed replication and tier mutations, hydrate journal state on restart, and clear durable records only after acknowledged dirty generations.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 18:12:06 +08:00
committed by GitHub
parent ca876930b3
commit cf41b1d3cf
6 changed files with 691 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/lib.rs
+8 -7
View File
@@ -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};
crates/scanner/src/scanner_io.rs
+8 -7
View File
@@ -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};
crates/scanner/src/scanner_io/dirty_usage.rs
+70 -9
View File
@@ -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<StdMutex<DirtyUsageBuckets>> = LazyLock::new(|| StdMutex::new(HashMap::new()));
@@ -27,6 +28,8 @@ pub(super) static DIRTY_USAGE_BUCKET_SCOPES: LazyLock<StdMutex<DirtyUsageBucketS
// per-bucket suffix to an earlier durable proof from the same process epoch.
pub(super) static DIRTY_USAGE_PRODUCER_IDENTITIES: LazyLock<StdMutex<DirtyUsageProducerIdentities>> =
LazyLock::new(|| StdMutex::new(BTreeMap::new()));
static DIRTY_USAGE_CLEAR_OBSERVER: LazyLock<StdRwLock<Option<ScannerDirtyUsageClearObserver>>> =
LazyLock::new(|| StdRwLock::new(None));
pub(super) static DIRTY_USAGE_PRODUCER_COVERAGE: AtomicU64 = AtomicU64::new(0);
pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
@@ -49,6 +52,8 @@ pub struct ScannerDirtyUsageBucket {
pub generation: u64,
}
pub type ScannerDirtyUsageClearObserver = Arc<dyn Fn(Vec<ScannerDirtyUsageBucket>) + 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<ScannerDirtyUsageClearObserver>,
) -> Option<ScannerDirtyUsageClearObserver> {
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<ScannerDirtyUsageBucket>) {
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<u64, ScannerDirtyUsageAckError> {
// 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::<std::result::Result<Vec<_>, _>>()?;
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::<Vec<_>>()
};
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::<Vec<_>>();
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));
}
rustfs/src/storage/mod.rs
+1
View File
@@ -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;
rustfs/src/storage/scanner_dirty_journal.rs
+597
View File
@@ -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<JournalCommand>,
}
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<ScannerDirtyUsageBucket>) {
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<ScannerDirtyUsageBucket>,
},
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
struct JournalState {
buckets: BTreeMap<String, JournalBucketState>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct JournalBucketState {
generation: u64,
scope: JournalScope,
producers: BTreeSet<SegmentInvalidationProducerIdentity>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
enum JournalScope {
WholeBucket,
TopLevelEntries(BTreeSet<String>),
}
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<Vec<ScannerDurableDirtyUsageReplayEntry>> {
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<Self> {
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<ECStore>) -> 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<ECStore>) -> 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::<ScannerDurableDirtyUsageReplayRecord>(&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<ECStore>,
mut receiver: mpsc::Receiver<JournalCommand>,
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<ECStore>, 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<ECStore>) -> 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<String> {
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());
}
}
rustfs/src/storage/storage_api.rs
+7 -1
View File
@@ -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>) {
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<ECStore>) {
}
};
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;
}