fix(scanner): mark committed replication and tier mutations (#7568)

* fix(scanner): mark committed replication and tier mutations

Bridge ECStore committed replication status writebacks and lifecycle tier expiration cleanup into scanner dirty-usage producer identity tracking. Only successful replication metadata updates mark Replication; validate-only, superseded, and retry paths remain unmarked.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* feat(scanner): replay durable dirty producer records

Add a strict scanner dirty-usage producer replay schema that can restore typed producer coverage after restart without treating ordinary process-local mutations as durable evidence. Replay records validate schema, cache key format, generations, scopes, producer identities, duplicate buckets, byte limits, and entry limits before modifying dirty state.

Keep segment activation fail-closed unless the restored bucket state came from durable replay; subsequent local mutations clear the durable authority bit for that bucket. Also fix the release scanner clippy lint in scoped cold-reuse proof creation.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -966,6 +966,11 @@ async fn cleanup_free_version_exact(api: Arc<ECStore>, 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);
 

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<String> {
 pub(crate) async fn bucket_lifecycle_config(bucket: &str) -> Option<BucketLifecycleConfiguration> {
     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);
+}

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

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -3843,6 +3843,13 @@ async fn persist_replication_state_if_current<S: ReplicationStorage>(
     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),

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<Arc<ReplicationObjectStore>> {
     sources::object_store_handle()
@@ -43,3 +46,7 @@ pub(crate) fn replication_runtime_initialized() -> bool {
 pub(crate) fn bucket_monitor() -> Option<Arc<ReplicationBucketMonitor>> {
     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);
+}

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<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
+pub type ScannerDirtyUsageMutationObserver = Arc<dyn Fn(&str, &str, ScannerDirtyUsageMutationSource) + Send + Sync + 'static>;
+
+#[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<WorkloadSnapshotProviderRef> = OnceLock::new();
+static SCANNER_DIRTY_USAGE_MUTATION_OBSERVER: LazyLock<StdRwLock<Option<ScannerDirtyUsageMutationObserver>>> =
+    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<WorkloadSnapshotP
     WORKLOAD_ADMISSION_SNAPSHOT_PROVIDER.get().cloned()
 }
 
+pub fn set_scanner_dirty_usage_mutation_observer(
+    observer: Option<ScannerDirtyUsageMutationObserver>,
+) -> Option<ScannerDirtyUsageMutationObserver> {
+    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<ECStore>) -> 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() {

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

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

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<StdMutex<DirtyUsageBuckets>> = LazyLock::new(|| StdMutex::new(HashMap::new()));
@@ -31,6 +32,10 @@ pub(super) static DIRTY_USAGE_BUCKET_NOTIFY: LazyLock<Notify> = LazyLock::new(No
 pub(super) static SCANNER_ACTIVITY_EPOCH: LazyLock<String> = LazyLock::new(|| format!("{:032x}", rand::random::<u128>()));
 pub(super) static SCANNER_MAINTENANCE_GENERATION: AtomicU64 = AtomicU64::new(0);
 pub(super) static SCANNER_MAINTENANCE_NOTIFY: LazyLock<Notify> = 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<String, DirtyUsageProducerIdentityState>;
@@ -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<ScannerDurableDirtyUsageReplayEntry>,
+}
+
+#[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<crate::segment_invalidation::SegmentInvalidationProducerIdentity>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ScannerDurableDirtyUsageReplayScope {
+    WholeBucket,
+    TopLevelEntries { entries: BTreeSet<String> },
+}
+
+pub fn encode_durable_dirty_usage_producer_replay_record(
+    entries: Vec<ScannerDurableDirtyUsageReplayEntry>,
+) -> std::result::Result<Vec<u8>, 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<ScannerDirtyUsageState, ScannerDurableDirtyUsageReplayError> {
+    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<Vec<ValidatedDurableDirtyUsageReplayEntry>, 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<DirtyUsageBucketScope, ScannerDurableDirtyUsageReplayError> {
+    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<I>(
     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 },

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,

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,

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>) {
+    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;
 }