test(scanner): cover scoped ack resolver fallback (#7332)

* test(scanner): cover scoped ack resolver fallback Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): reduce scoped resolver helper args Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-07 12:35:54 +00:00 · 2026-09-07 10:28:19 +08:00
parent 05efc584b0
commit a32360198c
5 changed files with 343 additions and 70 deletions
@@ -3573,6 +3573,8 @@ use leadership::*;
 pub(crate) use usage_store::RootPublicationProof;
 use usage_store::*;

+#[cfg(test)]
+pub(crate) use activity::scanner_node_activity_for_tests;
 pub use activity::scanner_topology_digest;
 pub(crate) use activity::{
    ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity,
@@ -441,6 +441,27 @@ pub(crate) struct ScannerNodeActivity {

 pub(crate) type ScannerActivitySnapshot = BTreeMap<String, ScannerNodeActivity>;

+#[cfg(test)]
+pub(crate) fn scanner_node_activity_for_tests(
+    instance_id: &str,
+    namespace_generation: u64,
+    dirty_usage_generation: u64,
+    dirty_usage_pending: bool,
+) -> ScannerNodeActivity {
+    ScannerNodeActivity {
+        instance_id: instance_id.to_string(),
+        namespace_generation,
+        maintenance_generation: 0,
+        protocol_version: SCANNER_ACTIVITY_PROTOCOL_VERSION,
+        topology_digest: [0; 32],
+        data_movement_active: false,
+        dirty_usage_generation,
+        dirty_usage_pending,
+        movement_generation: 0,
+        publication_blocked: false,
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct ScannerDirtyUsageAcknowledgement {
    pub(crate) host: String,
@@ -165,6 +165,11 @@ struct VerifiedRemoteDirtyUsage {
    acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
 }

+struct ScannerBucketScopeResolutionResult {
+    scope: ScannerBucketScanScope,
+    remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
+}
+
 fn verified_remote_dirty_usage(
    expected_peers: &HashMap<String, ScannerPeerDirtyUsageExpectation>,
    peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
@@ -229,6 +234,61 @@ fn scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(
    })
 }

+fn resolve_remote_dirty_usage_scope(
+    requested_scope: ScannerBucketScanScope,
+    mut dirty_buckets: HashSet<String>,
+    remote_dirty_usage: VerifiedRemoteDirtyUsage,
+    all_buckets: &[BucketInfo],
+    baseline_proof: ScannerCacheBaselineProof<'_>,
+) -> ScannerBucketScopeResolutionResult {
+    let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult {
+        scope,
+        remote_dirty_usage_acknowledgements: Vec::new(),
+    };
+
+    dirty_buckets.extend(remote_dirty_usage.dirty_buckets);
+    // Peer snapshots contribute bucket names only; the local prefix scopes
+    // would narrow a bucket a peer dirtied elsewhere, so the merged scope
+    // stays at bucket granularity (same rule as the local fallthrough).
+    let scope = scoped_scan_scope_from_dirty_buckets(requested_scope, dirty_buckets, None, true, all_buckets, baseline_proof);
+    if scope.is_default() {
+        return default_result(scope);
+    }
+    let Some(selected_buckets) = scope.selected_buckets.as_ref() else {
+        return default_result(scope);
+    };
+    let mut scoped_acknowledgements = Vec::with_capacity(remote_dirty_usage.acknowledgements.len());
+    for acknowledgement in remote_dirty_usage.acknowledgements {
+        let crate::scanner::ScannerDirtyUsageAcknowledgement {
+            host,
+            instance_id,
+            kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
+        } = acknowledgement
+        else {
+            return default_result(scope);
+        };
+        let entries = entries
+            .into_iter()
+            .filter(|entry| selected_buckets.contains(&entry.bucket))
+            .collect::<Vec<_>>();
+        if !entries.is_empty() {
+            scoped_acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement {
+                host,
+                instance_id,
+                kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
+            });
+        }
+    }
+    if scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) {
+        return default_result(ScannerBucketScanScope::default());
+    }
+
+    ScannerBucketScopeResolutionResult {
+        scope,
+        remote_dirty_usage_acknowledgements: scoped_acknowledgements,
+    }
+}
+
 fn complete_scanner_cache_snapshot_plan_digest(
    snapshot: &DataUsageInfo,
    proof: ScannerCacheBaselineProof<'_>,
@@ -98,18 +98,17 @@ pub(crate) struct ScannerCycleRequest {
    pub(crate) resolved_scope_observer: Option<tokio::sync::oneshot::Sender<ScannerBucketScanScope>>,
 }

-struct ScannerBucketScopeResolution<'a> {
-    requested_scope: ScannerBucketScanScope,
-    baseline_proof: ScannerCacheBaselineProof<'a>,
-    activity_before: &'a crate::scanner::ScannerActivitySnapshot,
-    dirty_usage_snapshot: &'a DirtyUsageSnapshot,
-    all_buckets: &'a [BucketInfo],
-    requires_full_scan: bool,
-}
-
-struct ScannerBucketScopeResolutionResult {
-    scope: ScannerBucketScanScope,
-    remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
+pub(super) struct ScannerBucketScopeResolution<'a> {
+    pub(super) requested_scope: ScannerBucketScanScope,
+    pub(super) baseline_proof: ScannerCacheBaselineProof<'a>,
+    pub(super) activity_before: &'a crate::scanner::ScannerActivitySnapshot,
+    pub(super) dirty_usage_snapshot: &'a DirtyUsageSnapshot,
+    pub(super) all_buckets: &'a [BucketInfo],
+    pub(super) requires_full_scan: bool,
+    #[cfg(test)]
+    pub(super) test_peer_snapshots: Option<Vec<(String, crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot)>>,
+    #[cfg(test)]
+    pub(super) test_scoped_dirty_usage_capability: Option<bool>,
 }

 async fn resolve_scanner_bucket_scan_scope<S>(
@@ -135,18 +134,35 @@ where
        return default_result(resolution.requested_scope);
    }

-    let mut dirty_buckets = resolution
+    let dirty_buckets = resolution
        .dirty_usage_snapshot
        .buckets
        .keys()
        .cloned()
        .collect::<HashSet<_>>();
    if distributed {
-        let Some(notification_system) = store.scanner_notification_system() else {
-            return default_result(resolution.requested_scope);
+        let notification_system = store.scanner_notification_system();
+        #[cfg(test)]
+        let peer_snapshots = if let Some(peer_snapshots) = resolution.test_peer_snapshots.clone() {
+            peer_snapshots
+        } else {
+            let Some(notification_system) = notification_system.as_ref() else {
+                return default_result(resolution.requested_scope);
+            };
+            let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
+                return default_result(resolution.requested_scope);
+            };
+            peer_snapshots
        };
-        let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
-            return default_result(resolution.requested_scope);
+        #[cfg(not(test))]
+        let peer_snapshots = {
+            let Some(notification_system) = notification_system.as_ref() else {
+                return default_result(resolution.requested_scope);
+            };
+            let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
+                return default_result(resolution.requested_scope);
+            };
+            peer_snapshots
        };
        let mut expected_peers = HashMap::new();
        for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before)
@@ -171,68 +187,57 @@ where
        let Some(remote_dirty_usage) = verified_remote_dirty_usage(&expected_peers, peer_snapshots) else {
            return default_result(resolution.requested_scope);
        };
-        dirty_buckets.extend(remote_dirty_usage.dirty_buckets);
-        // Peer snapshots contribute bucket names only; the local prefix scopes
-        // would narrow a bucket a peer dirtied elsewhere, so the merged scope
-        // stays at bucket granularity (same rule as the local fallthrough).
-        let scope = scoped_scan_scope_from_dirty_buckets(
+        let remote_resolution = resolve_remote_dirty_usage_scope(
            resolution.requested_scope,
            dirty_buckets,
-            None,
-            true,
+            remote_dirty_usage,
            resolution.all_buckets,
            resolution.baseline_proof,
        );
-        if scope.is_default() {
-            return default_result(scope);
-        }
-        let Some(selected_buckets) = scope.selected_buckets.as_ref() else {
-            return default_result(scope);
-        };
-        let mut scoped_acknowledgements = Vec::with_capacity(remote_dirty_usage.acknowledgements.len());
-        for acknowledgement in remote_dirty_usage.acknowledgements {
-            let crate::scanner::ScannerDirtyUsageAcknowledgement {
-                host,
-                instance_id,
-                kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
-            } = acknowledgement
-            else {
-                return default_result(scope);
+        if !remote_resolution.remote_dirty_usage_acknowledgements.is_empty() {
+            #[cfg(test)]
+            let capability_supported = if let Some(capability_supported) = resolution.test_scoped_dirty_usage_capability {
+                capability_supported
+            } else {
+                let Some(notification_system) = notification_system.as_ref() else {
+                    return default_result(ScannerBucketScanScope::default());
+                };
+                let capability_acknowledgements = remote_resolution
+                    .remote_dirty_usage_acknowledgements
+                    .clone()
+                    .into_iter()
+                    .map(Into::into)
+                    .collect::<Vec<crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement>>();
+                matches!(
+                    notification_system
+                        .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements)
+                        .await,
+                    Ok(true)
+                )
            };
-            let entries = entries
-                .into_iter()
-                .filter(|entry| selected_buckets.contains(&entry.bucket))
-                .collect::<Vec<_>>();
-            if !entries.is_empty() {
-                scoped_acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement {
-                    host,
-                    instance_id,
-                    kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
-                });
-            }
-        }
-        if super::scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) {
-            return default_result(ScannerBucketScanScope::default());
-        }
-        if !scoped_acknowledgements.is_empty() {
-            let capability_acknowledgements = scoped_acknowledgements
-                .clone()
-                .into_iter()
-                .map(Into::into)
-                .collect::<Vec<crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement>>();
-            if !matches!(
-                notification_system
-                    .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements)
-                    .await,
-                Ok(true)
-            ) {
+            #[cfg(not(test))]
+            let capability_supported = {
+                let Some(notification_system) = notification_system.as_ref() else {
+                    return default_result(ScannerBucketScanScope::default());
+                };
+                let capability_acknowledgements = remote_resolution
+                    .remote_dirty_usage_acknowledgements
+                    .clone()
+                    .into_iter()
+                    .map(Into::into)
+                    .collect::<Vec<crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement>>();
+                matches!(
+                    notification_system
+                        .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements)
+                        .await,
+                    Ok(true)
+                )
+            };
+            if !capability_supported {
                return default_result(ScannerBucketScanScope::default());
            }
        }
-        return ScannerBucketScopeResolutionResult {
-            scope,
-            remote_dirty_usage_acknowledgements: scoped_acknowledgements,
-        };
+        return remote_resolution;
    }

    default_result(scoped_scan_scope_from_dirty_buckets(
@@ -245,6 +250,18 @@ where
    ))
 }

+#[cfg(test)]
+pub(super) async fn resolve_scanner_bucket_scan_scope_for_tests<S>(
+    store: &S,
+    distributed: bool,
+    resolution: ScannerBucketScopeResolution<'_>,
+) -> ScannerBucketScopeResolutionResult
+where
+    S: ScannerStorage,
+{
+    resolve_scanner_bucket_scan_scope(store, distributed, resolution).await
+}
+
 pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
 where
    S: ScannerStorage,
@@ -382,6 +399,10 @@ where
            dirty_usage_snapshot: &dirty_usage_snapshot,
            all_buckets: &all_buckets,
            requires_full_scan: requires_full_scan || scan_mode == HealScanMode::Deep,
+            #[cfg(test)]
+            test_peer_snapshots: None,
+            #[cfg(test)]
+            test_scoped_dirty_usage_capability: None,
        },
    )
    .await;
@@ -37,6 +37,7 @@ use rustfs_concurrency::{
 };
 use rustfs_filemeta::FileInfo;
 use serial_test::serial;
+use std::collections::BTreeMap;
 use std::sync::Arc;
 use temp_env::with_var;
 use time::OffsetDateTime;
@@ -1770,6 +1771,56 @@ fn scanner_scoped_dirty_usage_ack_cost_threshold_is_single_protocol_batch() {
    )]));
 }

+#[test]
+fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_threshold() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([7; 32]);
+    let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+    let bucket_names = (0..=crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES)
+        .map(|index| format!("remote-{index:02}"))
+        .collect::<Vec<_>>();
+    let bucket_refs = bucket_names.iter().map(|bucket| (bucket.as_str(), 7)).collect::<Vec<_>>();
+    let all_buckets = bucket_names.iter().map(|bucket| bucket_info(bucket)).collect::<Vec<_>>();
+    let expected_peers = HashMap::from([(
+        "node-a:9000".to_string(),
+        ScannerPeerDirtyUsageExpectation {
+            instance_id: "instance-a".to_string(),
+            generation: 7,
+            pending: true,
+        },
+    )]);
+    let remote_dirty_usage = verified_remote_dirty_usage(
+        &expected_peers,
+        vec![("node-a:9000".to_string(), peer_dirty_usage_snapshot("instance-a", 7, true, &bucket_refs))],
+    )
+    .expect("fixture peer state should verify before the resolver cost gate");
+
+    let result = resolve_remote_dirty_usage_scope(
+        ScannerBucketScanScope::default(),
+        HashSet::new(),
+        remote_dirty_usage,
+        &all_buckets,
+        ScannerCacheBaselineProof {
+            authoritative_data: Some(&baseline),
+            observed_candidate_data: None,
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        },
+    );
+
+    assert!(
+        result.scope.is_default(),
+        "oversized scoped ACK batches must force the production resolver back to a full scan"
+    );
+    assert!(
+        result.remote_dirty_usage_acknowledgements.is_empty(),
+        "full-scan fallback must not send a scoped ACK that peers would reject or split"
+    );
+}
+
 #[test]
 fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() {
    let expected_peers = HashMap::from([(
@@ -1794,6 +1845,124 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state()
    }
 }

+#[tokio::test]
+#[serial]
+async fn distributed_scoped_scan_falls_back_when_remote_ack_exceeds_protocol_batch() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    clear_dirty_usage_buckets_for_tests();
+    record_dirty_usage_bucket("local-dirty");
+    let local_generation = dirty_usage_generation();
+    let remote_dirty_buckets = (0..=crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES)
+        .map(|index| (format!("remote-{index:02}"), 7))
+        .collect::<Vec<_>>();
+    let mut all_buckets = vec![bucket_info_with_created_time("local-dirty")];
+    all_buckets.extend(
+        remote_dirty_buckets
+            .iter()
+            .map(|(bucket, _)| bucket_info_with_created_time(bucket)),
+    );
+    let snapshot_buckets = remote_dirty_buckets
+        .iter()
+        .map(|(bucket, generation)| (bucket.as_str(), *generation))
+        .collect::<Vec<_>>();
+    let baseline_digest = DataUsageScanPlanDigest([8; 32]);
+    let baseline = complete_usage_baseline(DataUsageCacheSource::new(1, 2), baseline_digest, 7, 11);
+    let expected_sources = HashSet::from([DataUsageCacheSource::new(1, 2)]);
+    let dirty_usage_snapshot = snapshot_dirty_usage_buckets(&all_buckets, local_generation);
+    let activity_before = BTreeMap::from([(
+        "node-a:9000".to_string(),
+        crate::scanner::scanner_node_activity_for_tests("instance-a", 5, 7, true),
+    )]);
+
+    let result = super::io_cycle::resolve_scanner_bucket_scan_scope_for_tests(
+        store.as_ref(),
+        true,
+        super::io_cycle::ScannerBucketScopeResolution {
+            requested_scope: ScannerBucketScanScope::default(),
+            baseline_proof: ScannerCacheBaselineProof {
+                authoritative_data: Some(&baseline),
+                observed_candidate_data: None,
+                expected_sources: &expected_sources,
+                leader_epoch: 11,
+                want_cycle: 8,
+                scan_plan_digest: baseline_digest,
+            },
+            activity_before: &activity_before,
+            dirty_usage_snapshot: &dirty_usage_snapshot,
+            all_buckets: &all_buckets,
+            requires_full_scan: false,
+            test_peer_snapshots: Some(vec![(
+                "node-a:9000".to_string(),
+                peer_dirty_usage_snapshot("instance-a", 7, true, &snapshot_buckets),
+            )]),
+            test_scoped_dirty_usage_capability: Some(true),
+        },
+    )
+    .await;
+
+    assert!(
+        result.scope.is_default(),
+        "remote scoped acknowledgements above one protocol batch must force a full scan"
+    );
+    assert!(
+        result.remote_dirty_usage_acknowledgements.is_empty(),
+        "full-scan fallback must not send scoped remote acknowledgements"
+    );
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is_rejected() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([7; 32]);
+    let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+    let dirty_usage_snapshot = DirtyUsageSnapshot {
+        buckets: Arc::new(HashMap::new()),
+        scopes: Arc::new(HashMap::new()),
+        generation: 7,
+        covers_all_pending: true,
+    };
+    let activity_before = BTreeMap::from([(
+        "node-a:9000".to_string(),
+        crate::scanner::scanner_node_activity_for_tests("instance-a", 5, 7, true),
+    )]);
+
+    for (capability, expected_buckets, expected_ack_count) in
+        [(true, Some(HashSet::from(["photos".to_string()])), 1), (false, None, 0)]
+    {
+        let result = super::io_cycle::resolve_scanner_bucket_scan_scope_for_tests(
+            store.as_ref(),
+            true,
+            super::io_cycle::ScannerBucketScopeResolution {
+                requested_scope: ScannerBucketScanScope::default(),
+                baseline_proof: ScannerCacheBaselineProof {
+                    authoritative_data: Some(&baseline),
+                    observed_candidate_data: None,
+                    expected_sources: &expected_sources,
+                    leader_epoch: 11,
+                    want_cycle: 8,
+                    scan_plan_digest,
+                },
+                activity_before: &activity_before,
+                dirty_usage_snapshot: &dirty_usage_snapshot,
+                all_buckets: &[bucket_info_with_created_time("photos")],
+                requires_full_scan: false,
+                test_peer_snapshots: Some(vec![(
+                    "node-a:9000".to_string(),
+                    peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
+                )]),
+                test_scoped_dirty_usage_capability: Some(capability),
+            },
+        )
+        .await;
+
+        assert_eq!(result.scope.selected_buckets.as_deref(), expected_buckets.as_ref());
+        assert_eq!(result.remote_dirty_usage_acknowledgements.len(), expected_ack_count);
+    }
+}
+
 fn bucket_info_with_created_time(name: &str) -> BucketInfo {
    BucketInfo {
        created: Some(time::OffsetDateTime::UNIX_EPOCH),