gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-10 14:16:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): bind scoped ack confirmation to generation (#7619)

Browse Source 
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 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-10 12:38:23 +08:00
committed by GitHub
parent e292637ee0
commit 0b40d47a8a
3 changed files with 117 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cluster/rpc/peer_rest_client.rs
+40 -11
View File
@@ -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<ScannerScopedDirtyUsageAckEntry>,
) -> Result<ScannerPeerActivity> {
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
));
}
crates/scanner/src/scanner/activity.rs
+17 -2
View File
@@ -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<u64> {
match &self.kind {
ScannerDirtyUsageAcknowledgementKind::Generation(generation) => Some(*generation),
ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } => entries.iter().map(|entry| entry.generation).max(),
}
}
}
impl From<ScannerDirtyUsageAcknowledgement> for crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement {
fn from(acknowledgement: ScannerDirtyUsageAcknowledgement) -> Self {
match acknowledgement.kind {
crates/scanner/src/scanner/tests.rs
+60
View File
@@ -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::<bool, _>(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::<bool, _>(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::<bool, _>(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;