From 0b40d47a8a10da75bc010ff6912cbf092371e194 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 10 Sep 2026 12:38:23 +0800 Subject: [PATCH] fix(scanner): bind scoped ack confirmation to generation (#7619) Require lost scoped dirty-usage ACK reconciliation to observe a clean peer activity generation that covers the requested ACK generation. Apply the same guard in the scanner aggregation path so a stale clean activity snapshot cannot discharge pending maintenance after an uncertain ACK response. Refs rustfs/backlog#2427 Refs rustfs/backlog#2281 Refs rustfs/backlog#2240 Co-authored-by: zhi22915 --- .../src/cluster/rpc/peer_rest_client.rs | 51 ++++++++++++---- crates/scanner/src/scanner/activity.rs | 19 +++++- crates/scanner/src/scanner/tests.rs | 60 +++++++++++++++++++ 3 files changed, 117 insertions(+), 13 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index ab854bbc0..ea3e361a6 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -303,8 +303,16 @@ fn scanner_scoped_dirty_usage_ack_response_matches( && cleared_within_request } -fn scanner_scoped_dirty_usage_ack_reconciled(activity: &ScannerPeerActivity, expected_instance_id: &str) -> bool { - activity.instance_id == expected_instance_id && activity.dirty_usage_pending == Some(false) +fn scanner_scoped_dirty_usage_ack_reconciled( + activity: &ScannerPeerActivity, + expected_instance_id: &str, + expected_generation: u64, +) -> bool { + activity.instance_id == expected_instance_id + && activity.dirty_usage_pending == Some(false) + && activity + .dirty_usage_generation + .is_some_and(|generation| generation >= expected_generation) } fn scanner_instance_id_is_valid(instance_id: &str) -> bool { @@ -2290,6 +2298,7 @@ impl PeerRestClient { entries: Vec, ) -> Result { use rustfs_protos::scoped_dirty_usage::*; + let expected_generation = entries.iter().map(|entry| entry.generation).max().unwrap_or(0); let payloads = scanner_scoped_dirty_usage_ack_payloads(owner_id, instance_id.clone(), false, entries)?; let ack_attempt = async { let mut client = super::client::scanner_control_time_out_client( @@ -2337,7 +2346,9 @@ impl PeerRestClient { .await; } match self.scanner_scoped_dirty_usage_activity_confirmation().await { - Ok(activity) if scanner_scoped_dirty_usage_ack_reconciled(&activity, &instance_id) => Ok(activity), + Ok(activity) if scanner_scoped_dirty_usage_ack_reconciled(&activity, &instance_id, expected_generation) => { + Ok(activity) + } _ => Err(err), } } @@ -3176,30 +3187,48 @@ mod tests { #[test] fn scanner_scoped_dirty_usage_ack_reconciliation_requires_same_clean_instance() { - let activity = |instance_id: &str, pending| ScannerPeerActivity { + let activity = |instance_id: &str, generation, pending| ScannerPeerActivity { instance_id: instance_id.to_string(), namespace_generation: 1, maintenance_generation: 1, protocol_version: SCANNER_ACTIVITY_PROTOCOL_VERSION, topology_digest: Some([1; 32]), data_movement_active: Some(false), - dirty_usage_generation: Some(9), + dirty_usage_generation: generation, dirty_usage_pending: pending, movement_generation: Some(1), publication_blocked: Some(false), }; assert!(scanner_scoped_dirty_usage_ack_reconciled( - &activity("0123456789abcdef0123456789abcdef", Some(false)), - "0123456789abcdef0123456789abcdef" + &activity("0123456789abcdef0123456789abcdef", Some(9), Some(false)), + "0123456789abcdef0123456789abcdef", + 9 + )); + assert!(scanner_scoped_dirty_usage_ack_reconciled( + &activity("0123456789abcdef0123456789abcdef", Some(10), Some(false)), + "0123456789abcdef0123456789abcdef", + 9 )); assert!(!scanner_scoped_dirty_usage_ack_reconciled( - &activity("0123456789abcdef0123456789abcdef", Some(true)), - "0123456789abcdef0123456789abcdef" + &activity("0123456789abcdef0123456789abcdef", Some(8), Some(false)), + "0123456789abcdef0123456789abcdef", + 9 )); assert!(!scanner_scoped_dirty_usage_ack_reconciled( - &activity("fedcba9876543210fedcba9876543210", Some(false)), - "0123456789abcdef0123456789abcdef" + &activity("0123456789abcdef0123456789abcdef", None, Some(false)), + "0123456789abcdef0123456789abcdef", + 9 + )); + assert!(!scanner_scoped_dirty_usage_ack_reconciled( + &activity("0123456789abcdef0123456789abcdef", Some(9), Some(true)), + "0123456789abcdef0123456789abcdef", + 9 + )); + assert!(!scanner_scoped_dirty_usage_ack_reconciled( + &activity("fedcba9876543210fedcba9876543210", Some(9), Some(false)), + "0123456789abcdef0123456789abcdef", + 9 )); } diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 0b5c38ab5..954515ae3 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -105,8 +105,14 @@ pub(super) fn remote_dirty_usage_acknowledgement_loss_reconciled( 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) + let Some(expected_generation) = acknowledgement.expected_dirty_usage_generation() else { + 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 && generation >= expected_generation && !pending + }, + ) }) } @@ -509,6 +515,15 @@ pub(crate) enum ScannerDirtyUsageAcknowledgementKind { }, } +impl ScannerDirtyUsageAcknowledgement { + fn expected_dirty_usage_generation(&self) -> Option { + match &self.kind { + ScannerDirtyUsageAcknowledgementKind::Generation(generation) => Some(*generation), + ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } => entries.iter().map(|entry| entry.generation).max(), + } + } +} + impl From for crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement { fn from(acknowledgement: ScannerDirtyUsageAcknowledgement) -> Self { match acknowledgement.kind { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 5c9dae5be..d9da0b619 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -7680,6 +7680,25 @@ async fn scanner_cycle_confirms_lost_remote_ack_from_activity_snapshot() { "a new peer instance cannot confirm whether the old ACK reached durable dirty state" ); + let mut stale_activity = scanner_node_activity("epoch-a", 7, 3); + stale_activity.dirty_usage_generation = 4; + let stale_clean_activity = BTreeMap::from([("node-2".to_string(), stale_activity)]); + let stale_clean = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + std::slice::from_ref(&acknowledgement), + std::future::ready(Err::(std::io::Error::other( + "response lost before newer generation was observed", + ))), + || async { Ok(stale_clean_activity) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, stale_clean), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "a clean peer snapshot from before the acknowledged generation cannot prove the 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; @@ -7753,6 +7772,47 @@ async fn scanner_cycle_confirms_lost_scoped_ack_only_after_same_instance_clean_a "a restarted peer cannot prove the scoped ACK reached the old scanner instance" ); + let mut stale_activity = scanner_node_activity("epoch-a", 7, 3); + stale_activity.dirty_usage_generation = 4; + let stale_clean_activity = BTreeMap::from([("node-2".to_string(), stale_activity)]); + let stale_clean = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + std::slice::from_ref(&acknowledgement), + std::future::ready(Err::(std::io::Error::other( + "scoped ACK transport failed before the requested generation was observed", + ))), + || async { Ok(stale_clean_activity) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, stale_clean), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "a clean peer snapshot from before the scoped ACK generation cannot prove the ACK reached durable dirty state" + ); + + let empty_scoped_ack = ScannerDirtyUsageAcknowledgement { + host: "node-2".to_string(), + instance_id: "epoch-a".to_string(), + kind: ScannerDirtyUsageAcknowledgementKind::Scoped { + owner_id: Uuid::from_u128(0x11111111111111111111111111111111).to_string(), + entries: Vec::new(), + }, + }; + let empty_scoped_clean = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + &[empty_scoped_ack], + std::future::ready(Err::(std::io::Error::other("empty scoped ACK failed before peer delivery"))), + || 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, empty_scoped_clean), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "an empty scoped ACK has no durable generation to reconcile after response loss" + ); + let mut written_activity = scanner_node_activity("epoch-a", 7, 3); written_activity.dirty_usage_generation = 6; written_activity.dirty_usage_pending = true;