Update scoped retry preflight candidate

This commit is contained in:
overtrue
2026-10-28 05:19:06 +08:00
parent 342c12c312
commit deb483ba82
4 changed files with 19 additions and 15 deletions
+2 -2
View File
@@ -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.
+3 -3
View File
@@ -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 {
@@ -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<crate::storage_api::ScannerPublicationLeaseFailureInfo<'a>> = if cycle == 607 {
let info: for<'a> fn(&'a String) -> Option<ScannerPublicationLeaseFailureInfo<'a>> = 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(),
+6 -3
View File
@@ -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 {