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() {