fix(scanner): confirm lost dirty usage acknowledgements (#7357)

When the remote dirty-usage ACK response is lost, re-probe scanner activity once and accept the ACK only if every target host still reports the same scanner instance with no dirty usage pending. Duplicate targets, restarted peers, unverified activity, and concurrent writes remain pending.

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

crates/scanner/src/scanner.rs

@@ -2255,11 +2255,14 @@ where
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
         let acknowledgement_count = remote_dirty_usage_acknowledgements.len();
+        let acknowledgement_proof = remote_dirty_usage_acknowledgements.clone();
         let acknowledgements = remote_dirty_usage_acknowledgements.into_iter().map(Into::into).collect();
         remote_dirty_usage_acknowledgement_pending(
             cycle_info.current,
             acknowledgement_count,
+            &acknowledgement_proof,
             notification_system.acknowledge_scanner_dirty_usage(acknowledgements),
+            || probe_scanner_activity(storeapi.as_ref(), true),
         )
         .await
     } else {

crates/scanner/src/scanner/activity.rs

@@ -15,6 +15,7 @@
 use super::*;
 use crate::storage_api::ScannerStorage;
 use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION;
+use std::collections::HashSet;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum ScannerCycleWakeReason {
@@ -51,18 +52,25 @@ pub(crate) fn scanner_cycle_outcome_with_pending_maintenance(
     }
 }
 
-pub(super) async fn remote_dirty_usage_acknowledgement_pending<F, E>(
+pub(super) async fn remote_dirty_usage_acknowledgement_pending<F, E, C, CF>(
     cycle: u64,
     acknowledgement_count: usize,
+    acknowledgements: &[ScannerDirtyUsageAcknowledgement],
     acknowledgement: F,
+    confirm_after_error: C,
 ) -> bool
 where
     F: Future<Output = Result<bool, E>>,
     E: std::fmt::Display,
+    C: FnOnce() -> CF,
+    CF: Future<Output = Result<ScannerActivitySnapshot, String>>,
 {
     match acknowledgement.await {
         Ok(dirty_usage_pending) => dirty_usage_pending,
         Err(err) => {
+            if remote_dirty_usage_acknowledgement_loss_reconciled(acknowledgements, confirm_after_error().await) {
+                return false;
+            }
             warn!(
                 target: "rustfs::scanner",
                 event = EVENT_SCANNER_PERSIST_STATE,
@@ -79,6 +87,29 @@ where
     }
 }
 
+pub(super) fn remote_dirty_usage_acknowledgement_loss_reconciled(
+    acknowledgements: &[ScannerDirtyUsageAcknowledgement],
+    activity_after_error: Result<ScannerActivitySnapshot, String>,
+) -> bool {
+    if acknowledgements.is_empty() {
+        return false;
+    }
+    let Ok(activity_after_error) = activity_after_error else {
+        return false;
+    };
+    if !scanner_activity_allows_usage_publication(&activity_after_error) {
+        return false;
+    }
+    let mut acknowledged_hosts = HashSet::with_capacity(acknowledgements.len());
+    acknowledgements.iter().all(|acknowledgement| {
+        if !acknowledged_hosts.insert(acknowledgement.host.as_str()) {
+            return false;
+        }
+        scanner_activity_dirty_usage_state_for_host(&activity_after_error, &acknowledgement.host)
+            .is_some_and(|(instance_id, _generation, pending)| instance_id == acknowledgement.instance_id && !pending)
+    })
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) struct ScannerCleanIdleBackoff {
     pub(super) interval_multiplier: u32,

crates/scanner/src/scanner/tests.rs

@@ -7498,13 +7498,28 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
 
 #[tokio::test]
 async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
-    let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(true))).await;
+    let acknowledgements = Vec::new();
+    let pending = remote_dirty_usage_acknowledgement_pending(
+        7,
+        1,
+        &acknowledgements,
+        std::future::ready(Ok::<bool, std::io::Error>(true)),
+        || async { Ok(BTreeMap::new()) },
+    )
+    .await;
     assert_eq!(
         scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending),
         ScannerCycleOutcome::CompletedWithPendingMaintenance
     );
 
-    let cleared = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(false))).await;
+    let cleared = remote_dirty_usage_acknowledgement_pending(
+        7,
+        1,
+        &acknowledgements,
+        std::future::ready(Ok::<bool, std::io::Error>(false)),
+        || async { Ok(BTreeMap::new()) },
+    )
+    .await;
     assert_eq!(
         scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, cleared),
         ScannerCycleOutcome::Completed
@@ -7513,7 +7528,9 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     let failed = remote_dirty_usage_acknowledgement_pending(
         7,
         1,
+        &acknowledgements,
         std::future::ready(Err::<bool, _>(std::io::Error::other("injected acknowledgement failure"))),
+        || async { Err("confirmation probe failed".to_string()) },
     )
     .await;
     assert_eq!(
@@ -7522,6 +7539,76 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     );
 }
 
+#[tokio::test]
+async fn scanner_cycle_confirms_lost_remote_ack_from_activity_snapshot() {
+    let acknowledgement = ScannerDirtyUsageAcknowledgement {
+        host: "node-2".to_string(),
+        instance_id: "epoch-a".to_string(),
+        kind: ScannerDirtyUsageAcknowledgementKind::Generation(5),
+    };
+    let cleared_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let response_lost = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        std::slice::from_ref(&acknowledgement),
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost after peer ack"))),
+        || async { Ok(cleared_activity) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, response_lost),
+        ScannerCycleOutcome::Completed,
+        "a same-instance activity confirmation with no dirty work closes the uncertain ACK"
+    );
+
+    let duplicate_acknowledgements = vec![acknowledgement.clone(), acknowledgement.clone()];
+    let duplicate_target = remote_dirty_usage_acknowledgement_pending(
+        8,
+        duplicate_acknowledgements.len(),
+        &duplicate_acknowledgements,
+        std::future::ready(Err::<bool, _>(std::io::Error::other("duplicate target rejected before peer ack"))),
+        || async { Ok(BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))])) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, duplicate_target),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "a request rejected before peer delivery cannot be recovered by a clean activity snapshot"
+    );
+
+    let restarted_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-b", 7, 3))]);
+    let peer_restarted = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        std::slice::from_ref(&acknowledgement),
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost before restart was observed"))),
+        || async { Ok(restarted_activity) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, peer_restarted),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "a new peer instance cannot confirm whether the old ACK reached durable dirty state"
+    );
+
+    let mut written_activity = scanner_node_activity("epoch-a", 7, 3);
+    written_activity.dirty_usage_generation = 6;
+    written_activity.dirty_usage_pending = true;
+    let concurrent_write = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        &[acknowledgement],
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost before concurrent write"))),
+        || async { Ok(BTreeMap::from([("node-2".to_string(), written_activity)])) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, concurrent_write),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "new dirty usage on the same peer must keep maintenance pending"
+    );
+}
+
 #[test]
 #[serial]
 fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() {