From 638da7a2e3df3145232af9d9efe54e75bfd6cb4b Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:27:53 +0800 Subject: [PATCH] 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 --- crates/scanner/src/scanner.rs | 3 + crates/scanner/src/scanner/activity.rs | 33 +++++++++- crates/scanner/src/scanner/tests.rs | 91 +++++++++++++++++++++++++- 3 files changed, 124 insertions(+), 3 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index b8c16f994..d96f90353 100644 --- a/crates/scanner/src/scanner.rs +++ b/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 { diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 582649147..0b5c38ab5 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/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( +pub(super) async fn remote_dirty_usage_acknowledgement_pending( cycle: u64, acknowledgement_count: usize, + acknowledgements: &[ScannerDirtyUsageAcknowledgement], acknowledgement: F, + confirm_after_error: C, ) -> bool where F: Future>, E: std::fmt::Display, + C: FnOnce() -> CF, + CF: Future>, { 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, +) -> 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, diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 5ee1840a0..331c2c107 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/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::(true))).await; + let acknowledgements = Vec::new(); + let pending = remote_dirty_usage_acknowledgement_pending( + 7, + 1, + &acknowledgements, + std::future::ready(Ok::(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::(false))).await; + let cleared = remote_dirty_usage_acknowledgement_pending( + 7, + 1, + &acknowledgements, + std::future::ready(Ok::(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::(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::(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::(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::(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::(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() {