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

fix(scanner): separate write retries from movement backlog (#7249)

Browse Source
This commit is contained in:
cxymds
2026-09-06 10:51:46 +08:00
committed by GitHub
parent b0c73c1224
commit 3df39ef4d7
4 changed files with 90 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner.rs
+20 -4
View File
@@ -1888,9 +1888,9 @@ where
// A remote restart or movement flip invalidates
// the token proof; usage_store interprets this
// as a publication barrier and performs no PUT.
return true;
return Some(ScannerCycleDeferReason::DataMovement);
}
storeapi.scanner_data_usage_publication_blocked().await
scanner_local_publication_defer_reason(storeapi.as_ref()).await
}
},
)
@@ -3239,8 +3239,8 @@ where
{
match status {
ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => {
if storeapi.scanner_data_usage_publication_blocked().await {
return Some(ScannerCycleDeferReason::DataMovement);
if let Some(reason) = scanner_local_publication_defer_reason(storeapi).await {
return Some(reason);
}
if status == ScannerCycleStatus::Complete {
let distributed = storeapi.setup_is_dist_erasure().await;
@@ -3263,6 +3263,22 @@ where
}
}
async fn scanner_local_publication_defer_reason<S>(storeapi: &S) -> Option<ScannerCycleDeferReason>
where
S: ScannerStorage,
{
if !storeapi.scanner_data_usage_publication_blocked().await {
return None;
}
// Pending namespace commits invalidate this publication attempt, but only
// storage movement creates durable, rate-limited catch-up debt.
if storeapi.scanner_data_movement_pause_status().await.paused {
Some(ScannerCycleDeferReason::DataMovement)
} else {
Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
}
}
fn scanner_post_lease_activity_defer_reason(
expected_digest: Option<[u8; 32]>,
activity: Result<ScannerActivitySnapshot, String>,
crates/scanner/src/scanner/tests.rs
+38 -12
View File
@@ -266,6 +266,11 @@ async fn running_main_loop_catches_up_pause_cleared_after_startup_observe() {
}
let pause_status = store.scanner_data_movement_pause_status().await;
assert!(pause_status.paused);
assert_eq!(
scanner_local_publication_defer_reason(store.as_ref()).await,
Some(ScannerCycleDeferReason::DataMovement),
"an actual data-movement pause must retain durable catch-up tracking"
);
paused_probe.wait().await;
drop(paused_probe);
@@ -1171,6 +1176,9 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledging_usage() {
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
let (_temp_dir, store) = setup_scanner_cycle_store().await;
let mut pause_backlog = ScannerPauseBacklogController::claim(store.clone(), scanner_pause_backlog_now())
.await
.expect("scanner pause backlog should be available");
let bucket = format!("scanner-coordinator-pending-{}", Uuid::new_v4().simple());
store
.make_bucket(&bucket, &crate::storage_api::scan::MakeBucketOptions::default())
@@ -1195,6 +1203,13 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
.await
.expect("fixture usage baseline should be readable");
let pending = ecstore_hold_namespace_commit(store.as_ref());
assert_eq!(
scanner_local_publication_defer_reason(store.as_ref()).await,
Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
"an ordinary namespace commit must not be classified as data movement"
);
let pause_backlog_attempt = pause_backlog.begin_attempt(scanner_pause_backlog_now()).await;
assert_eq!(pause_backlog_attempt, ScannerPauseBacklogAttemptDecision::Untracked);
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
let mut cycle_info = CurrentCycle {
@@ -1209,7 +1224,15 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin
.await
.expect("the coordinator must finish its namespace walk while a PUT is pending");
assert_eq!(budget.progress().0, 1, "the coordinator must reach actual object traversal");
assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
assert_eq!(
outcome,
ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
);
finish_scanner_pause_backlog_cycle(&mut pause_backlog, &store, pause_backlog_attempt, outcome).await;
let pause_backlog_status = scanner_pause_backlog_status(store.clone()).await;
assert_eq!(pause_backlog_status.phase, ScannerPauseBacklogPhase::Idle);
assert!(!pause_backlog_status.pending_full_scan);
assert_eq!(pause_backlog_status.catch_up_attempts, 0);
assert_eq!(cycle_info.next, 1, "a rejected publication must not advance the cycle");
assert_eq!(revision, DataUsageCacheRevision::Missing);
assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
@@ -5826,7 +5849,7 @@ async fn test_usage_save_object_not_found_defers_only_with_a_fresh_route_barrier
let probe_calls = route_probe_calls.clone();
async move {
let call = probe_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
route_blocked && call > 1
(route_blocked && call > 1).then_some(ScannerCycleDeferReason::DataMovement)
}
},
)
@@ -5872,16 +5895,19 @@ async fn test_usage_save_route_barrier_prevents_missing_snapshot_creation() {
data: None,
revision: DataUsageCacheRevision::Missing,
}),
|| async { true },
|| async { Some(ScannerCycleDeferReason::ActivityBaselineUnavailable) },
)
.await;
assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
assert_eq!(
outcome,
DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
);
assert!(!store.objects.lock().await.contains_key(&target_key));
assert_eq!(
store.put_counts.lock().await.get(&target_key),
None,
"the final pool-state fence must run before the first PUT"
"the final publication fence must run before the first PUT"
);
}
}
@@ -5902,7 +5928,7 @@ async fn test_observational_usage_defers_when_authoritative_baseline_is_missing(
receiver,
None,
None,
|| async { false },
|| async { None },
)
.await;
@@ -5950,7 +5976,7 @@ async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_iden
receiver,
None,
None,
|| async { false },
|| async { None },
)
.await;
@@ -5991,7 +6017,7 @@ async fn test_observational_usage_uses_bootstrap_pending_primary_as_baseline() {
receiver,
None,
None,
|| async { false },
|| async { None },
)
.await;
@@ -6051,7 +6077,7 @@ async fn test_usage_route_barrier_precedes_durable_reconciliation() {
data: Some(Bytes::from(snapshot_data)),
revision: DataUsageCacheRevision::Etag("memory-1".to_string()),
}),
|| async { true },
|| async { Some(ScannerCycleDeferReason::DataMovement) },
)
.await;
@@ -6091,7 +6117,7 @@ async fn coordinator_does_not_put_after_remote_generation_flip() {
// Model the remote lease holder flipping its movement generation
// after the activity probe but before the coordinator's PUT.
route_store.publication_admission_blocked.store(true, Ordering::Release);
false
None
}
},
)
@@ -6129,7 +6155,7 @@ async fn coordinator_classifies_an_expired_publication_lease() {
revision: DataUsageCacheRevision::Missing,
}),
ScannerPublicationFence::new(None, Some(expired), None),
|| async { false },
|| async { None },
)
.await;
@@ -6208,7 +6234,7 @@ async fn test_deferred_usage_save_keeps_last_real_save_metric() {
data: None,
revision: DataUsageCacheRevision::Missing,
}),
|| async { true },
|| async { Some(ScannerCycleDeferReason::DataMovement) },
)
.await;
crates/scanner/src/scanner/usage_store.rs
+20 -18
View File
@@ -265,7 +265,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
receiver,
leader_epoch,
initial_baseline,
|| async { false },
|| async { None },
)
.await
}
@@ -280,7 +280,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
) -> DataUsagePersistOutcome
where
F: Fn() -> Fut + Send + Sync,
Fut: Future<Output = bool> + Send,
Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
{
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch(
ctx,
@@ -308,7 +308,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
) -> DataUsagePersistOutcome
where
F: Fn() -> Fut + Send + Sync,
Fut: Future<Output = bool> + Send,
Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
{
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
ctx,
@@ -336,7 +336,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
) -> DataUsagePersistOutcome
where
F: Fn() -> Fut + Send + Sync,
Fut: Future<Output = bool> + Send,
Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
{
let ScannerPublicationFence {
expected_publication_epoch,
@@ -374,18 +374,19 @@ where
} else {
DATA_USAGE_OBJ_NAME_PATH.as_str()
};
if route_probe().await {
if let Some(reason) = route_probe().await {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "publication_blocked_before_reconcile",
"Scanner data usage publication deferred by the pool-state fence"
reason = reason.as_str(),
path = %target_path,
"Scanner data usage publication deferred by the publication fence"
);
global_metrics().record_scanner_usage_deferred(ScannerCycleDeferReason::DataMovement.as_str());
outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
global_metrics().record_scanner_usage_deferred(reason.as_str());
outcome = DataUsagePersistOutcome::Deferred(reason);
break;
}
@@ -626,17 +627,18 @@ where
if ctx.is_cancelled() {
break 'updates;
}
if route_probe().await {
if let Some(reason) = route_probe().await {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "publication_blocked_before_save",
"Scanner data usage publication deferred by the final pool-state fence"
reason = reason.as_str(),
path = %target_path,
"Scanner data usage publication deferred by the final publication fence"
);
break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
break DataUsagePersistOutcome::Deferred(reason);
}
if remote_lease_expired(remote_lease_deadline) {
break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
@@ -722,19 +724,19 @@ where
);
}
Err(e @ EcstoreError::ObjectNotFound(_, _)) => {
let route_blocked = route_probe().await;
if route_blocked {
if let Some(reason) = route_probe().await {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "publication_deferred",
reason = reason.as_str(),
path = %target_path,
error = %e,
"Scanner data usage route is blocked by data movement; retrying later"
"Scanner data usage route remains blocked; retrying later"
);
break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
break DataUsagePersistOutcome::Deferred(reason);
}
error!(
target: "rustfs::scanner",
crates/scanner/src/scanner_io/tests.rs
+12
View File
@@ -124,6 +124,16 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
(temp_dir, store)
}
async fn wait_for_namespace_commit_tails(store: &ECStore) {
tokio::time::timeout(Duration::from_secs(30), async {
while store.scanner_data_usage_publication_blocked().await {
tokio::time::sleep(Duration::from_millis(1)).await;
}
})
.await
.expect("namespace commit tails should drain before the scanner fixture runs");
}
#[tokio::test]
#[serial]
async fn checkpoint_fixture_bucket_identity_uses_its_set_instance_owner() {
@@ -334,6 +344,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
.await
.expect("initial object should persist");
}
wait_for_namespace_commit_tails(store.as_ref()).await;
let mut baseline = None;
for (index, (scan_mode, requires_full_scan, explicit_scope)) in [
(HealScanMode::Normal, true, false),
@@ -352,6 +363,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
.put_object("cold-bucket", &format!("added-{index}"), &mut reader, &ScannerObjectOptions::default())
.await
.expect("cold bucket mutation should persist");
wait_for_namespace_commit_tails(store.as_ref()).await;
// Only the hot bucket is in the usage hint. The cold result must
// come from this cycle's storage walk, not its previous baseline.
record_dirty_usage_bucket("hot-bucket");