gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): stabilize retired backlog survivor commits (#7585)

Browse Source 
Treat an authoritative scanner pause-backlog commit as stable when every member in that commit stores the matching stable and committed record. Retired source replicas can remain missing in the observed topology while survivor membership is already stable.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 18:40:43 +08:00
committed by GitHub
parent 15949180c9
commit d9ae1654b6
1 changed files with 44 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner/backlog.rs
+44 -5
View File
@@ -1100,7 +1100,23 @@ fn select_scanner_pause_backlog_replicas(replicas: Vec<ScannerPauseBacklogReplic
});
};
let stable_matches_ledger = matches!(&stable_consensus, Ok(Some(stable)) if stable == &selected);
let stable_matches_ledger = match &authoritative_commit {
Some(committed) => {
committed.ledger == selected
&& committed.replicas.iter().all(|id| {
replicas.iter().any(|replica| {
replica.id == *id
&& matches!(
&replica.state,
ScannerPauseBacklogReplicaState::Valid(record)
if record.stable.as_ref() == Some(&selected)
&& record.committed.as_ref() == Some(committed)
)
})
})
}
None => matches!(&stable_consensus, Ok(Some(stable)) if stable == &selected),
};
let healthy_replicas = replicas
.iter()
.filter(|replica| {
@@ -1901,12 +1917,35 @@ mod tests {
.await
.expect("fresh native disk selection");
assert_eq!(&loaded.ledger, expected, "membership repair preserves every ledger field");
assert!(loaded.durable && loaded.stable_matches_ledger);
assert!(
loaded.durable && loaded.stable_matches_ledger,
"unexpected loaded state: persistence_state={}, durable={}, stable_matches_ledger={}, healthy_replicas={}, stale_or_unavailable_replicas={}",
loaded.persistence_state,
loaded.durable,
loaded.stable_matches_ledger,
loaded.healthy_replicas,
loaded.stale_or_unavailable_replicas
);
assert_eq!(loaded.healthy_replicas, 4);
let committed = loaded.authoritative_commit.expect("complete current cohort proof");
assert_eq!(committed.replicas, scanner_pause_backlog_replica_ids(&loaded.replicas));
for set in store.scanner_pause_backlog_writable_set_disks().await {
let (bytes, _) = native_replica_bytes(&set).await;
let mut healthy_ids = loaded
.replicas
.iter()
.filter_map(|replica| {
matches!(
&replica.state,
ScannerPauseBacklogReplicaState::Valid(record)
if record.stable.as_ref() == Some(expected)
&& record.committed.as_ref() == Some(&committed)
)
.then_some(replica.id)
})
.collect::<Vec<_>>();
healthy_ids.sort_unstable();
assert_eq!(committed.replicas, healthy_ids);
for id in &committed.replicas {
let set = &store.pools[id.pool_index].disk_set[id.set_index];
let (bytes, _) = native_replica_bytes(set).await;
let ScannerPauseBacklogReplicaState::Valid(record) = decode_scanner_pause_backlog_ledger(&bytes) else {
panic!("native survivor record");
};