feat(scanner): reuse complete observed scan candidates (#7206)

crates/scanner/src/scanner.rs

@@ -1730,6 +1730,7 @@ where
     };
     let baseline_publication_epoch = baseline_publication_guard.epoch();
     let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await;
+    let observed_usage_candidate_result = read_config(storeapi.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await;
     drop(baseline_publication_guard);
     let usage_persist_baseline = match usage_persist_baseline_result {
         Ok(baseline) => baseline,
@@ -1750,6 +1751,24 @@ where
             return ScannerCycleOutcome::Failed;
         }
     };
+    let observed_usage_candidate = match observed_usage_candidate_result {
+        Ok(candidate) => Some(Bytes::from(candidate)),
+        Err(EcstoreError::ConfigNotFound) => None,
+        Err(err) => {
+            debug!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                cycle = cycle_info.current,
+                path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
+                state = "observed_candidate_load_failed",
+                error = %err,
+                "Scanner skipped an unavailable observed usage candidate for scoped refresh"
+            );
+            None
+        }
+    };
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
@@ -1764,6 +1783,7 @@ where
             scan_mode,
             scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
             persisted_usage_baseline: usage_persist_baseline.data.clone(),
+            observed_usage_candidate,
             requires_full_scan: scheduling.requires_full_scan,
             service_cohort: scheduling.service_cohort,
             #[cfg(test)]

crates/scanner/src/scanner_io.rs

@@ -27,7 +27,7 @@ use metrics::counter;
 use rand::seq::SliceRandom as _;
 #[cfg(test)]
 use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
-use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo};
+use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo, observed_data_usage_is_newer};
 use rustfs_filemeta::FileMeta;
 use rustfs_heal_contracts::heal_channel::HealScanMode;
 use rustfs_lock::{LockError, NamespaceLockGuard};
@@ -128,7 +128,8 @@ impl ScannerBucketScanScope {
 
 #[derive(Clone, Copy)]
 pub(super) struct ScannerCacheBaselineProof<'a> {
-    pub(super) data: Option<&'a Bytes>,
+    pub(super) authoritative_data: Option<&'a Bytes>,
+    pub(super) observed_candidate_data: Option<&'a Bytes>,
     pub(super) expected_sources: &'a HashSet<DataUsageCacheSource>,
     pub(super) leader_epoch: u64,
     pub(super) want_cycle: u64,
@@ -171,21 +172,23 @@ fn verified_remote_dirty_usage_buckets(
     (received_peers.len() == expected_peers.len()).then_some(dirty_buckets)
 }
 
-fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
+fn complete_scanner_cache_snapshot_plan_digest(
-    let data = proof.data?;
+    snapshot: &DataUsageInfo,
-    let baseline = serde_json::from_slice::<DataUsageInfo>(data).ok()?;
+    proof: ScannerCacheBaselineProof<'_>,
-    if !baseline.is_complete_bucket_usage_snapshot()
+    expected_converged: bool,
-        || baseline.usage_snapshot_partial
+) -> Option<DataUsageScanPlanDigest> {
-        || baseline.usage_snapshot_converged != Some(true)
+    if !snapshot.is_complete_bucket_usage_snapshot()
-        || baseline.scanner_epoch != Some(proof.leader_epoch)
+        || snapshot.usage_snapshot_partial
-        || baseline.usage_snapshot_set_states.len() != proof.expected_sources.len()
+        || snapshot.usage_snapshot_converged != Some(expected_converged)
+        || snapshot.scanner_epoch != Some(proof.leader_epoch)
+        || snapshot.usage_snapshot_set_states.len() != proof.expected_sources.len()
     {
         return None;
     }
 
     // Completed maintenance also covers ordinary usage. Keep its exact stored
     // proof for cache reuse, and reject mixtures of different set work proofs.
-    let baseline_plan_digest = DataUsageScanPlanDigest(baseline.usage_snapshot_set_states.first()?.scan_plan_digest?);
+    let baseline_plan_digest = DataUsageScanPlanDigest(snapshot.usage_snapshot_set_states.first()?.scan_plan_digest?);
     if ![
         proof.scan_plan_digest,
         scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Normal, true),
@@ -195,8 +198,8 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<
     {
         return None;
     }
-    let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len());
+    let mut states = HashSet::with_capacity(snapshot.usage_snapshot_set_states.len());
-    for state in &baseline.usage_snapshot_set_states {
+    for state in &snapshot.usage_snapshot_set_states {
         let source = DataUsageCacheSource::new(usize::try_from(state.pool_index).ok()?, usize::try_from(state.set_index).ok()?);
         if !proof.expected_sources.contains(&source)
             || !states.insert(source)
@@ -213,6 +216,30 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<
     (states == *proof.expected_sources).then_some(baseline_plan_digest)
 }
 
+fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
+    let authoritative = serde_json::from_slice::<DataUsageInfo>(proof.authoritative_data?).ok()?;
+    if complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true).is_some() {
+        return Some(proof.scan_plan_digest);
+    }
+
+    // A complete but superseded observation may reuse its per-set cache only
+    // when it was explicitly tied to the durable authoritative baseline. It
+    // remains observational: this proof grants bucket-scope reuse only and
+    // never changes authoritative usage publication or dirty acknowledgement.
+    let authoritative_has_identity = (crate::scanner::data_usage_info_has_persisted_baseline_identity(&authoritative)
+        && authoritative.usage_snapshot_converged != Some(false))
+        || crate::scanner::data_usage_info_is_bootstrap_pending(&authoritative);
+    if !authoritative_has_identity {
+        return None;
+    }
+    let observed = serde_json::from_slice::<DataUsageInfo>(proof.observed_candidate_data?).ok()?;
+    if !observed_data_usage_is_newer(&observed, &authoritative) {
+        return None;
+    }
+
+    complete_scanner_cache_snapshot_plan_digest(&observed, proof, false)
+}
+
 fn scoped_scan_scope_from_dirty_buckets(
     requested_scope: ScannerBucketScanScope,
     dirty_buckets: HashSet<String>,

crates/scanner/src/scanner_io/io_cycle.rs

@@ -72,6 +72,7 @@ where
         scan_mode,
         scan_scope: ScannerBucketScanScope::default(),
         persisted_usage_baseline: None,
+        observed_usage_candidate: None,
         requires_full_scan: true,
         service_cohort: None,
         #[cfg(test)]
@@ -89,6 +90,7 @@ pub(crate) struct ScannerCycleRequest {
     pub(crate) scan_mode: HealScanMode,
     pub(crate) scan_scope: ScannerBucketScanScope,
     pub(crate) persisted_usage_baseline: Option<Bytes>,
+    pub(crate) observed_usage_candidate: Option<Bytes>,
     /// Scheduled maintenance must visit clean buckets even with a valid dirty scope.
     pub(crate) requires_full_scan: bool,
     pub(crate) service_cohort: Option<Arc<StdMutex<ScannerServiceCohort>>>,
@@ -185,6 +187,7 @@ where
         scan_mode,
         scan_scope,
         persisted_usage_baseline,
+        observed_usage_candidate,
         requires_full_scan,
         service_cohort,
         #[cfg(test)]
@@ -297,7 +300,8 @@ where
         ScannerBucketScopeResolution {
             requested_scope: scan_scope,
             baseline_proof: ScannerCacheBaselineProof {
-                data: persisted_usage_baseline.as_ref(),
+                authoritative_data: persisted_usage_baseline.as_ref(),
+                observed_candidate_data: observed_usage_candidate.as_ref(),
                 expected_sources: &expected_sources,
                 leader_epoch,
                 want_cycle,

crates/scanner/src/scanner_io/tests.rs

@@ -397,6 +397,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
                     scan_mode,
                     scan_scope: requested_scope,
                     persisted_usage_baseline: baseline,
+                    observed_usage_candidate: None,
                     requires_full_scan,
                     resolved_scope_observer: Some(observer),
                     service_cohort: None,
@@ -470,6 +471,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                     scan_mode: HealScanMode::Normal,
                     scan_scope: ScannerBucketScanScope::default(),
                     persisted_usage_baseline: None,
+                    observed_usage_candidate: None,
                     requires_full_scan: false,
                     resolved_scope_observer: None,
                     service_cohort: None,
@@ -521,6 +523,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                     scan_mode,
                     scan_scope: ScannerBucketScanScope::default(),
                     persisted_usage_baseline: None,
+                    observed_usage_candidate: None,
                     requires_full_scan,
                     resolved_scope_observer: None,
                     service_cohort: None,
@@ -1331,7 +1334,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance(
 
     assert_eq!(
         complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
-            data: Some(&baseline),
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
             expected_sources: &expected_sources,
             leader_epoch: 11,
             want_cycle: 8,
@@ -1345,7 +1349,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance(
     let incomplete = bytes::Bytes::from(serde_json::to_vec(&incomplete).expect("test baseline should encode"));
     assert_eq!(
         complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
-            data: Some(&incomplete),
+            authoritative_data: Some(&incomplete),
+            observed_candidate_data: None,
             expected_sources: &expected_sources,
             leader_epoch: 11,
             want_cycle: 8,
@@ -1359,7 +1364,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance(
     let wrong_provenance = bytes::Bytes::from(serde_json::to_vec(&wrong_provenance).expect("test baseline should encode"));
     assert_eq!(
         complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
-            data: Some(&wrong_provenance),
+            authoritative_data: Some(&wrong_provenance),
+            observed_candidate_data: None,
             expected_sources: &expected_sources,
             leader_epoch: 11,
             want_cycle: 8,
@@ -1369,6 +1375,111 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance(
     );
 }
 
+#[test]
+fn scoped_scan_accepts_only_a_complete_observation_tied_to_the_authoritative_baseline() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([9; 32]);
+    let authoritative = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+    let authoritative_info =
+        serde_json::from_slice::<DataUsageInfo>(&authoritative).expect("authoritative baseline should decode");
+    let bootstrap_authoritative_info =
+        crate::scanner::scanner_usage_bootstrap_marker(SystemTime::UNIX_EPOCH + Duration::from_secs(9), Some(11));
+    let bootstrap_authoritative =
+        bytes::Bytes::from(serde_json::to_vec(&bootstrap_authoritative_info).expect("bootstrap baseline should encode"));
+    let mut observed_info = authoritative_info.clone();
+    observed_info.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(11));
+    observed_info.scanner_cycle = Some(8);
+    observed_info.usage_snapshot_converged = Some(false);
+    observed_info.usage_snapshot_authoritative_baseline = Some(bootstrap_authoritative_info.snapshot_identity());
+    observed_info.usage_snapshot_set_states[0].scanner_cycle = Some(8);
+    let observed = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("observation should encode"));
+
+    macro_rules! proof {
+        ($authoritative:expr, $candidate:expr) => {
+            ScannerCacheBaselineProof {
+                authoritative_data: Some($authoritative),
+                observed_candidate_data: $candidate,
+                expected_sources: &expected_sources,
+                leader_epoch: 11,
+                want_cycle: 9,
+                scan_plan_digest,
+            }
+        };
+    }
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&observed))),
+        Some(scan_plan_digest)
+    );
+
+    observed_info.usage_snapshot_authoritative_baseline = Some(DataUsageInfo::default().snapshot_identity());
+    let mismatched_baseline = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("observation should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&mismatched_baseline))),
+        None
+    );
+
+    observed_info = serde_json::from_slice(&observed).expect("observation should decode");
+    observed_info.usage_snapshot_partial = true;
+    let partial = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("partial observation should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&partial))),
+        None
+    );
+
+    observed_info = serde_json::from_slice(&observed).expect("observation should decode");
+    observed_info.usage_snapshot_converged = Some(true);
+    let converged = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("converged observation should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&converged))),
+        None
+    );
+
+    let mut legacy_authoritative = authoritative_info.clone();
+    legacy_authoritative.usage_snapshot_converged = None;
+    let mut stale_info = serde_json::from_slice::<DataUsageInfo>(&observed).expect("observation should decode");
+    stale_info.scanner_cycle = Some(7);
+    stale_info.usage_snapshot_set_states[0].scanner_cycle = Some(7);
+    stale_info.usage_snapshot_authoritative_baseline = Some(legacy_authoritative.snapshot_identity());
+    let legacy_authoritative =
+        bytes::Bytes::from(serde_json::to_vec(&legacy_authoritative).expect("legacy baseline should encode"));
+    let stale = bytes::Bytes::from(serde_json::to_vec(&stale_info).expect("stale observation should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&legacy_authoritative, Some(&stale))),
+        None
+    );
+
+    let malformed = bytes::Bytes::from_static(b"not data usage json");
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&malformed))),
+        None
+    );
+
+    let mut nonconverged_authoritative = authoritative_info;
+    nonconverged_authoritative.usage_snapshot_converged = Some(false);
+    let mut observation_of_nonconverged_authoritative = nonconverged_authoritative.clone();
+    observation_of_nonconverged_authoritative.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(12));
+    observation_of_nonconverged_authoritative.scanner_cycle = Some(8);
+    observation_of_nonconverged_authoritative.usage_snapshot_set_states[0].scanner_cycle = Some(8);
+    observation_of_nonconverged_authoritative.usage_snapshot_authoritative_baseline =
+        Some(nonconverged_authoritative.snapshot_identity());
+    let nonconverged_authoritative =
+        bytes::Bytes::from(serde_json::to_vec(&nonconverged_authoritative).expect("nonconverged baseline should encode"));
+    let observation_of_nonconverged_authoritative =
+        bytes::Bytes::from(serde_json::to_vec(&observation_of_nonconverged_authoritative).expect("observation should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            authoritative_data: Some(&nonconverged_authoritative),
+            observed_candidate_data: Some(&observation_of_nonconverged_authoritative),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 9,
+            scan_plan_digest,
+        }),
+        None
+    );
+}
+
 #[test]
 fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
     let source = DataUsageCacheSource::new(1, 2);
@@ -1382,7 +1493,8 @@ fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
         true,
         &[bucket_info("photos")],
         ScannerCacheBaselineProof {
-            data: Some(&baseline),
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
             expected_sources: &expected_sources,
             leader_epoch: 11,
             want_cycle: 8,
@@ -1420,7 +1532,8 @@ fn scoped_scan_baseline_work_proof_requires_uniform_known_set_identity() {
         let data = Bytes::from(serde_json::to_vec(&candidate).expect("candidate should encode"));
         assert_eq!(
             complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
-                data: Some(&data),
+                authoritative_data: Some(&data),
+                observed_candidate_data: None,
                 expected_sources: &sources,
                 leader_epoch: 11,
                 want_cycle: 8,

crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs

@@ -126,6 +126,7 @@ async fn run_entry(store: &Arc<ECStore>, cycle: u64, selected: Option<&str>, exp
                 scan_mode: HealScanMode::Normal,
                 scan_scope: ScannerBucketScanScope::default(),
                 persisted_usage_baseline: root_before.0.clone().map(Bytes::from),
+                observed_usage_candidate: None,
                 requires_full_scan: false,
                 service_cohort: None,
                 resolved_scope_observer: Some(observer),

crates/scanner/src/scanner_io/tests/service_cohort.rs

@@ -61,6 +61,7 @@ async fn run_cohort_cycle(
                 scan_mode: HealScanMode::Normal,
                 scan_scope: ScannerBucketScanScope::default(),
                 persisted_usage_baseline: None,
+                observed_usage_candidate: None,
                 requires_full_scan: false,
                 service_cohort: Some(cohort),
                 resolved_scope_observer: None,