diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 1966538a0..42d2dd6b0 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -406,7 +406,7 @@ pub mod notification { pub use crate::services::notification_sys::{ CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys, - new_global_notification_sys, start_remote_version_state_fleet_probe, + new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe, }; } diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index b69c60152..19660dd84 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -1962,6 +1962,12 @@ where .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))? } +/// Classify transport-only activity failures without treating an answered +/// peer's application error as an outage. +pub fn scanner_peer_transport_error_message_is_retryable(error: &str) -> bool { + crate::cluster::rpc::client::message_has_network_needle(error) +} + fn scanner_activity_should_retry(first_error: Option<&Error>, timed_out: bool) -> bool { timed_out || first_error.is_some_and(PeerRestClient::is_network_like_error) } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 788426b64..a1a832d9a 100644 --- a/crates/scanner/src/scanner.rs +++ b/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) diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 4d0018c62..e31f6c285 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/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 { + Ready(T), + Rejected(E), + Cancelled, +} + +pub(crate) async fn await_scanner_publication_proof( + ctx: &CancellationToken, + cycle: u64, + stage: &'static str, + mut proof: F, + retryable: Retryable, +) -> ScannerPublicationProofWait +where + F: FnMut() -> Fut, + Fut: Future>, + 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( + ctx: &CancellationToken, + cycle: u64, + stage: &'static str, + probe: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + 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 } diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 1aae284ee..a5877802c 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/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(); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index e1abbf474..6b8eaeb6b 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -574,10 +574,6 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec> 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", } diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 4278fdad8..4c406185c 100644 --- a/crates/scanner/src/storage_api.rs +++ b/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)]