From deb483ba82e4b37bbf1b33090eec673d014ec762 Mon Sep 17 00:00:00 2001 From: overtrue Date: Wed, 28 Oct 2026 05:19:06 +0800 Subject: [PATCH] Update scoped retry preflight candidate --- crates/scanner/src/scanner.rs | 4 ++-- crates/scanner/src/scanner/activity.rs | 6 +++--- .../src/scanner/tests/cycle_budget_publication.rs | 15 ++++++++------- crates/scanner/src/storage_api.rs | 9 ++++++--- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 103c1eab7..e447da943 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -3947,11 +3947,11 @@ impl ScannerCyclePublicationPlan { || acquire(self.uses_observational_lease()), |err| { // The structured reason comes from local acquisition, not a peer report. - crate::storage_api::scanner_publication_lease_failure_info(err) + crate::storage_api::scan::scanner_publication_lease_failure_info(err) .is_some_and(|info| info.reason == Some("channel_acquisition_timed_out")) || scanner_publication_lease_error_is_retryable(&err.to_string()) }, - crate::storage_api::scanner_publication_lease_failure_info, + crate::storage_api::scan::scanner_publication_lease_failure_info, ); let (proof, timed_out) = if self == Self::ProgressOnly { // Object/directory budgets can expire with no runtime deadline. diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index bb3ffec29..8391c44d9 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -13,8 +13,8 @@ // limitations under the License. /// Cycle wake/backoff policy and scanner activity observation (probing, generations, topology digest). use super::*; -use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION; -use crate::storage_api::{ScannerPublicationLeaseFailureInfo, ScannerStorage}; +use crate::storage_api::ScannerStorage; +use crate::storage_api::scan::{SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, ScannerPublicationLeaseFailureInfo}; use std::collections::HashSet; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -239,7 +239,7 @@ pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) - } pub(crate) fn scanner_publication_activity_error_is_retryable(error: &str) -> bool { - crate::storage_api::scanner_peer_transport_error_message_is_retryable(error) + crate::storage_api::scan::scanner_peer_transport_error_message_is_retryable(error) } pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool { diff --git a/crates/scanner/src/scanner/tests/cycle_budget_publication.rs b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs index 769b57bae..67b3cce68 100644 --- a/crates/scanner/src/scanner/tests/cycle_budget_publication.rs +++ b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::storage_api::scan::ScannerPublicationLeaseFailureInfo; use rustfs_utils::path::path_join_buf; #[derive(Clone)] @@ -273,9 +274,9 @@ async fn exercise_interleaved_lease_waits() { // Verify event serialization independently of the ecstore tests that // exercise the local and remote error producers. for cycle in [606, 607, 608] { - let info: for<'a> fn(&'a String) -> Option> = if cycle == 607 { + let info: for<'a> fn(&'a String) -> Option> = if cycle == 607 { |_| { - Some(crate::storage_api::ScannerPublicationLeaseFailureInfo { + Some(ScannerPublicationLeaseFailureInfo { peer: None, kind: "peer_request_failed", reason: None, @@ -284,7 +285,7 @@ async fn exercise_interleaved_lease_waits() { } } else if cycle == 606 { |_| { - Some(crate::storage_api::ScannerPublicationLeaseFailureInfo { + Some(ScannerPublicationLeaseFailureInfo { peer: None, kind: "peer_request_failed", reason: Some("invalid_proof"), @@ -293,7 +294,7 @@ async fn exercise_interleaved_lease_waits() { } } else { |_| { - Some(crate::storage_api::ScannerPublicationLeaseFailureInfo { + Some(ScannerPublicationLeaseFailureInfo { peer: None, kind: "peer_request_failed", reason: Some("channel_acquisition_timed_out"), @@ -511,7 +512,7 @@ fn publication_lease_retries_local_channel_deadline_and_stops_on_cancel() { .build() .expect("test runtime"); runtime.block_on(async { - use rustfs_ecstore::api::rpc::PeerRestClient; + use crate::storage_api::scan::EcstorePeerRestClient; use rustfs_utils::net::XHost; async fn channel_deadline_error( @@ -525,7 +526,7 @@ fn publication_lease_retries_local_channel_deadline_and_stops_on_cancel() { Ok(_) => panic!("the stalled channel cannot grant a lease"), Err(error) => error, }; - let info = crate::storage_api::scanner_publication_lease_failure_info(&error) + let info = crate::storage_api::scan::scanner_publication_lease_failure_info(&error) .expect("local channel deadline must carry structured diagnostics"); assert_eq!(info.kind, "peer_request_failed"); assert_eq!(info.reason, Some("channel_acquisition_timed_out")); @@ -538,7 +539,7 @@ fn publication_lease_retries_local_channel_deadline_and_stops_on_cancel() { let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("local listener"); let address = listener.local_addr().expect("listener address"); let target = format!("https://{address}"); - let client = PeerRestClient::new( + let client = EcstorePeerRestClient::new( XHost { name: "127.0.0.1".to_owned(), port: address.port(), diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index b906e83cf..af14bd2ca 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -109,9 +109,6 @@ pub(crate) use rustfs_ecstore::api::layout::{ SetupType as EcstoreSetupType, }; pub(crate) use rustfs_ecstore::api::notification::NotificationSys as EcstoreNotificationSys; -pub(crate) use rustfs_ecstore::api::notification::{ - ScannerPublicationLeaseFailureInfo, scanner_peer_transport_error_message_is_retryable, scanner_publication_lease_failure_info, -}; pub(crate) use rustfs_ecstore::api::object::{ SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState, }; @@ -367,6 +364,12 @@ pub(crate) mod scan { SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE, SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES, }; + pub(crate) use rustfs_ecstore::api::notification::{ + ScannerPublicationLeaseFailureInfo, scanner_peer_transport_error_message_is_retryable, + scanner_publication_lease_failure_info, + }; + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::rpc::PeerRestClient as EcstorePeerRestClient; } pub(crate) mod scanner_io {