fix(scanner): bound publication proof retries on main (#6870)

* fix(scanner): retain completed publication candidates

* fix(scanner): export publication activity helper

* test(ecstore): retain activity snapshot across retries

* fix(scanner): rebase publication proof retry onto main

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(scanner): resolve publication proof retry conflicts

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: heihutu <heihutu@gmail.com>

crates/scanner/src/scanner.rs

@@ -63,7 +63,6 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, info, instrument, warn};
 
-use crate::storage_api::owner::SCANNER_PUBLICATION_LEASE_TTL_MS;
 use crate::storage_api::scan::{
     BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
     SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
@@ -1205,10 +1204,6 @@ fn data_usage_persist_timeout() -> Duration {
     DataUsageCache::persistence_timeout()
 }
 
-fn scanner_publication_lease_budget_allows_persistence(timeout: Duration) -> bool {
-    timeout < Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS)
-}
-
 #[cfg(not(test))]
 const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30);
 #[cfg(test)]
@@ -1493,21 +1488,24 @@ async fn run_data_scanner_cycle_with_budget(
     let mut remote_publication_leases = None;
     let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() {
         None
-    } else if !scanner_publication_lease_budget_allows_persistence(usage_persist_timeout) {
-        // The lease is intentionally fixed-duration and has no renewal path.
-        // Refuse a persistence budget that could outlive it instead of
-        // allowing the peer to admit movement while a local PUT is in flight.
-        Some(ScannerCycleDeferReason::PublicationLeaseBudgetExceeded)
     } else if let Some(notification_system) = storeapi.notification_system() {
-        match notification_system
-            .acquire_scanner_publication_leases(remote_publication_lease_targets.clone())
-            .await
-        {
-            Ok(grants) => {
+        let publication_proof_ctx = cycle_budget.token();
+        let lease_result = await_scanner_publication_proof(
+            &publication_proof_ctx,
+            cycle_info.current,
+            "lease_acquire",
+            || notification_system.acquire_scanner_publication_leases(remote_publication_lease_targets.clone()),
+            |err| scanner_publication_lease_error_is_retryable(&err.to_string()),
+        )
+        .await;
+        match lease_result {
+            ScannerPublicationProofWait::Ready(grants) => {
                 remote_publication_leases = Some((notification_system, grants));
                 None
             }
-            Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+            ScannerPublicationProofWait::Rejected(_) | ScannerPublicationProofWait::Cancelled => {
+                Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+            }
         }
     } else {
         Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)

crates/scanner/src/scanner/activity.rs

@@ -108,6 +108,149 @@ impl ScannerRetryBackoff {
     }
 }
 
+const SCANNER_PUBLICATION_PROOF_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30);
+
+pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) -> Duration {
+    let exponent = consecutive_failures.saturating_sub(1).min(31);
+    let multiplier = 1u32.checked_shl(exponent).unwrap_or(u32::MAX);
+    SCANNER_RETRY_BASE_INTERVAL
+        .saturating_mul(multiplier)
+        .min(SCANNER_PUBLICATION_PROOF_RETRY_MAX_INTERVAL)
+}
+
+pub(crate) fn scanner_publication_activity_error_is_retryable(error: &str) -> bool {
+    crate::storage_api::scanner_peer_transport_error_message_is_retryable(error)
+}
+
+pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool {
+    scanner_publication_activity_error_is_retryable(error)
+        || error.ends_with("scanner publication lease capacity is exhausted")
+        || error.ends_with("scanner publication lease response arrived after its safety window")
+}
+
+pub(crate) enum ScannerPublicationProofWait<T, E> {
+    Ready(T),
+    Rejected(E),
+    Cancelled,
+}
+
+pub(crate) async fn await_scanner_publication_proof<T, E, F, Fut, Retryable>(
+    ctx: &CancellationToken,
+    cycle: u64,
+    stage: &'static str,
+    mut proof: F,
+    retryable: Retryable,
+) -> ScannerPublicationProofWait<T, E>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, E>>,
+    E: std::fmt::Display,
+    Retryable: Fn(&E) -> bool,
+{
+    let started_at = Instant::now();
+    let mut consecutive_failures = 0u32;
+
+    loop {
+        if ctx.is_cancelled() {
+            return ScannerPublicationProofWait::Cancelled;
+        }
+
+        match proof().await {
+            Ok(value) => {
+                if consecutive_failures > 0 {
+                    info!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_recovered",
+                        cycle,
+                        stage,
+                        attempts = consecutive_failures.saturating_add(1),
+                        pending_duration = ?started_at.elapsed(),
+                        "Scanner publication proof recovered"
+                    );
+                }
+                return ScannerPublicationProofWait::Ready(value);
+            }
+            Err(err) if !retryable(&err) => {
+                warn!(
+                    target: "rustfs::scanner",
+                    event = EVENT_SCANNER_CYCLE_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_RUNTIME,
+                    state = "publication_proof_rejected",
+                    cycle,
+                    stage,
+                    error = %err,
+                    "Scanner publication proof failed with a non-retryable cluster state"
+                );
+                return ScannerPublicationProofWait::Rejected(err);
+            }
+            Err(err) => {
+                consecutive_failures = consecutive_failures.saturating_add(1);
+                let retry_delay = scanner_publication_proof_retry_delay(consecutive_failures);
+                if consecutive_failures == 1 || consecutive_failures.is_multiple_of(20) {
+                    warn!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_pending",
+                        cycle,
+                        stage,
+                        attempt = consecutive_failures,
+                        retry_delay = ?retry_delay,
+                        error = %err,
+                        "Scanner retained a completed scan while publication proof is unavailable"
+                    );
+                } else {
+                    debug!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_CYCLE_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        state = "publication_proof_retry",
+                        cycle,
+                        stage,
+                        attempt = consecutive_failures,
+                        retry_delay = ?retry_delay,
+                        error = %err,
+                        "Scanner publication activity proof retry scheduled"
+                    );
+                }
+
+                tokio::select! {
+                    _ = ctx.cancelled() => return ScannerPublicationProofWait::Cancelled,
+                    _ = tokio::time::sleep(retry_delay) => {}
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+pub(crate) async fn await_scanner_publication_activity<F, Fut>(
+    ctx: &CancellationToken,
+    cycle: u64,
+    stage: &'static str,
+    probe: F,
+) -> Result<ScannerActivitySnapshot, String>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<ScannerActivitySnapshot, String>>,
+{
+    match await_scanner_publication_proof(ctx, cycle, stage, probe, |err: &String| {
+        scanner_publication_activity_error_is_retryable(err)
+    })
+    .await
+    {
+        ScannerPublicationProofWait::Ready(snapshot) => Ok(snapshot),
+        ScannerPublicationProofWait::Rejected(err) => Err(err),
+        ScannerPublicationProofWait::Cancelled => Err(format!("scanner publication activity proof was cancelled during {stage}")),
+    }
+}
+
 impl Default for ScannerCleanIdleBackoff {
     fn default() -> Self {
         Self { interval_multiplier: 1 }

crates/scanner/src/scanner/tests.rs

@@ -4583,7 +4583,6 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
     for reason in [
         ScannerCycleDeferReason::DataMovement,
         ScannerCycleDeferReason::ActivityBaselineUnavailable,
-        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded,
         ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded,
         ScannerCycleDeferReason::PublicationLeaseReleaseFailed,
     ] {
@@ -4755,29 +4754,6 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() {
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
-#[test]
-fn scanner_publication_lease_budget_has_a_strict_ttl_boundary() {
-    let ttl = Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS);
-
-    assert!(scanner_publication_lease_budget_allows_persistence(
-        ttl.saturating_sub(Duration::from_millis(1))
-    ));
-    assert!(!scanner_publication_lease_budget_allows_persistence(ttl));
-    assert!(!scanner_publication_lease_budget_allows_persistence(ttl + Duration::from_millis(1)));
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded.as_str(),
-        "publication_lease_budget_exceeded"
-    );
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(),
-        "publication_lease_deadline_exceeded"
-    );
-    assert_eq!(
-        ScannerCycleDeferReason::PublicationLeaseReleaseFailed.as_str(),
-        "publication_lease_release_failed"
-    );
-}
-
 #[tokio::test]
 async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
     let ctx = CancellationToken::new();
@@ -5032,6 +5008,140 @@ fn superseded_retry_backoff_grows_from_the_default_cycle() {
     }
 }
 
+#[test]
+fn publication_proof_retry_backoff_reaches_its_short_cap() {
+    for (failures, expected) in [(1, 5), (2, 10), (3, 20), (4, 30), (20, 30)] {
+        assert_eq!(scanner_publication_proof_retry_delay(failures), Duration::from_secs(expected));
+    }
+}
+
+#[test]
+fn publication_proof_retry_classifies_availability_without_masking_protocol_errors() {
+    for error in [
+        "peer node3 is temporarily offline",
+        "scanner activity peer node3 timed out after 5s",
+        "transport error: connection refused",
+    ] {
+        assert!(scanner_publication_activity_error_is_retryable(error), "{error}");
+    }
+
+    for error in [
+        "scanner activity peer node3 uses protocol 6, expected 7",
+        "scanner activity peer node3 has a different storage topology",
+        "scanner activity peer node3 omitted its movement generation",
+        "duplicate scanner activity peer: node3",
+        "scanner activity peer[2] is unreachable",
+        "scanner publication lease peer node3 is unavailable",
+    ] {
+        assert!(!scanner_publication_activity_error_is_retryable(error), "{error}");
+    }
+}
+
+#[test]
+fn publication_lease_retry_preserves_only_recoverable_candidates() {
+    for error in [
+        "scanner publication lease acquisition failed: scanner publication lease capacity is exhausted",
+        "scanner publication lease acquisition failed: scanner publication lease response arrived after its safety window",
+        "scanner publication lease acquisition failed: peer node3 is temporarily offline",
+    ] {
+        assert!(scanner_publication_lease_error_is_retryable(error), "{error}");
+    }
+
+    for error in [
+        "scanner publication lease acquisition failed: scanner publication lease generation is stale",
+        "scanner publication lease acquisition failed: peer returned a different scanner publication lease session",
+        "scanner publication lease acquisition failed: scanner publication lease is blocked by data movement",
+        "scanner publication lease acquisition failed: peer returned an invalid scanner publication lease proof",
+    ] {
+        assert!(!scanner_publication_lease_error_is_retryable(error), "{error}");
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_retains_candidate_until_activity_recovers() {
+    let ctx = CancellationToken::new();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+    let started_at = Instant::now();
+
+    let snapshot = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        let attempt = probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async move {
+            if attempt == 0 {
+                Err("peer temporarily offline".to_string())
+            } else {
+                Ok(ScannerActivitySnapshot::new())
+            }
+        }
+    })
+    .await
+    .expect("a retained publication candidate should survive one transient probe failure");
+
+    assert!(snapshot.is_empty());
+    assert_eq!(attempts.load(Ordering::SeqCst), 2);
+    assert_eq!(started_at.elapsed(), Duration::from_secs(5));
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_does_not_retry_a_protocol_mismatch() {
+    let ctx = CancellationToken::new();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Err("scanner activity peer node3 uses protocol 6, expected 7".to_string()) }
+    })
+    .await
+    .expect_err("a protocol mismatch must not be hidden behind availability retries");
+
+    assert!(err.contains("uses protocol"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 1);
+}
+
+#[tokio::test]
+async fn publication_proof_stops_waiting_when_the_cycle_is_cancelled() {
+    let ctx = CancellationToken::new();
+    ctx.cancel();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Ok(ScannerActivitySnapshot::new()) }
+    })
+    .await
+    .expect_err("a cancelled cycle must release its retained publication candidate");
+
+    assert!(err.contains("cancelled"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_proof_releases_candidate_when_cancelled_during_backoff() {
+    let ctx = CancellationToken::new();
+    let cancel_ctx = ctx.clone();
+    let attempts = Arc::new(AtomicUsize::new(0));
+    let probe_attempts = attempts.clone();
+    let started_at = Instant::now();
+
+    let cancel = tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        cancel_ctx.cancel();
+    });
+    let err = await_scanner_publication_activity(&ctx, 17, "postscan", move || {
+        probe_attempts.fetch_add(1, Ordering::SeqCst);
+        async { Err("peer temporarily offline".to_string()) }
+    })
+    .await
+    .expect_err("cycle cancellation must release a candidate waiting to retry publication proof");
+    cancel.await.expect("cancellation task should complete");
+
+    assert!(err.contains("cancelled"));
+    assert_eq!(attempts.load(Ordering::SeqCst), 1);
+    assert_eq!(started_at.elapsed(), Duration::from_secs(1));
+}
+
 #[tokio::test(start_paused = true)]
 async fn corrupt_cycle_state_backoff_uses_virtual_clock() {
     let mut backoff = ScannerRetryBackoff::default();

crates/scanner/src/scanner_io.rs

@@ -574,10 +574,6 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec<Arc<Disk>>
 pub(crate) enum ScannerCycleDeferReason {
     ActivityBaselineUnavailable,
     DataMovement,
-    /// The configured persistence budget cannot fit within the fixed remote
-    /// publication-lease TTL. This is a deterministic configuration/contract
-    /// mismatch, not evidence that a peer activity probe failed.
-    PublicationLeaseBudgetExceeded,
     /// A granted lease's absolute deadline cannot cover the persistence
     /// operation. This can occur even when the configured budget fits the
     /// nominal TTL because lease acquisition consumed part of the window.
@@ -592,7 +588,6 @@ impl ScannerCycleDeferReason {
         match self {
             Self::ActivityBaselineUnavailable => "activity_baseline_unavailable",
             Self::DataMovement => "data_movement",
-            Self::PublicationLeaseBudgetExceeded => "publication_lease_budget_exceeded",
             Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded",
             Self::PublicationLeaseReleaseFailed => "publication_lease_release_failed",
         }

crates/scanner/src/storage_api.rs

@@ -92,6 +92,7 @@ pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_
 pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
 };
+pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable;
 pub(crate) use rustfs_ecstore::api::object::SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::rebalance::{
@@ -106,9 +107,9 @@ pub(crate) use rustfs_ecstore::api::runtime::{
     setup_is_erasure_sd as ecstore_is_erasure_sd,
 };
 pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
+pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
-pub(crate) use rustfs_ecstore::api::storage::{ECStore as EcstoreStore, SCANNER_PUBLICATION_LEASE_TTL_MS};
 use rustfs_storage_api as storage_contracts;
 
 pub(crate) mod owner {
@@ -125,15 +126,14 @@ pub(crate) mod owner {
         EcstoreNsScannerOpenRequest, EcstoreObjectLockConfiguration, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
         EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
         EcstoreVersioningApi, EcstoreVersioningConfiguration, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
-        SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerReplicationHealObject, ScannerReplicationHealResult,
-        ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle,
-        ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config,
-        ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
-        ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd,
-        ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
-        ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config,
-        ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config,
-        ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
+        ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
+        ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr,
+        ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config,
+        ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure,
+        ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
+        ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
+        ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
+        ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
     };
 
     #[cfg(test)]