From 3df39ef4d7183c98533f6e5375e9f09a4cc86597 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 6 Sep 2026 10:51:46 +0800 Subject: [PATCH] fix(scanner): separate write retries from movement backlog (#7249) --- crates/scanner/src/scanner.rs | 24 +++++++++-- crates/scanner/src/scanner/tests.rs | 50 +++++++++++++++++------ crates/scanner/src/scanner/usage_store.rs | 38 +++++++++-------- crates/scanner/src/scanner_io/tests.rs | 12 ++++++ 4 files changed, 90 insertions(+), 34 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index e57a6cc1a..b9de1c073 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -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(storeapi: &S) -> Option +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, diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index dbbbdca62..37d45aa89 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -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; diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index af7c616ee..08a3c9996 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -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 + Send, + Fut: Future> + 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 + Send, + Fut: Future> + 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 + Send, + Fut: Future> + 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", diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 19e0fc457..9521fc2fa 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -124,6 +124,16 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc) { (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");