diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 64ceb292e..c566bd0a9 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -725,7 +725,7 @@ impl PeerRestClient { /// never take it offline no matter what its message says. The substring /// fallback only covers failures that exist purely as text, such as the /// dial errors `get_client` wraps. - fn is_network_like_error(err: &Error) -> bool { + pub(crate) fn is_network_like_error(err: &Error) -> bool { if let Error::Io(io_err) = err && let Some(status) = embedded_tonic_status(io_err) { diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 0c78007c2..5158c82a7 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -1483,7 +1483,7 @@ impl NotificationSys { futures.push(async move { let client = client.ok_or_else(|| Error::other(format!("scanner activity peer[{idx}] is unreachable")))?; let host = client.grid_host.clone(); - scanner_activity_with_timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, &host, client.scanner_activity()) + scanner_activity_with_retry(&client, &host) .await .map(|activity| (host, activity)) }); @@ -1962,6 +1962,46 @@ where .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))? } +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) +} + +/// Retry one activity probe after a bounded reconnect when the first attempt +/// failed at the transport boundary. A peer that answered with an invalid or +/// incompatible activity response is not retried here: it must remain a hard +/// fail-closed result for the all-peer publication proof. +async fn scanner_activity_with_retry(client: &PeerRestClient, host: &str) -> Result { + let first = timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await; + let should_retry = match &first { + Ok(Ok(_)) => false, + Ok(Err(err)) => scanner_activity_should_retry(Some(err), false), + Err(_) => scanner_activity_should_retry(None, true), + }; + + match first { + Ok(Ok(activity)) => return Ok(activity), + Ok(Err(err)) if !should_retry => return Err(err), + Ok(Err(err)) => { + debug!(peer = host, error = %err, "scanner activity probe failed on first transport attempt; reconnecting"); + client.prepare_retry().await; + } + Err(_) => { + debug!(peer = host, timeout = ?SCANNER_ACTIVITY_PROBE_TIMEOUT, "scanner activity probe timed out on first attempt; reconnecting"); + client.prepare_retry().await; + } + } + + match timeout(SCANNER_ACTIVITY_PROBE_TIMEOUT, client.scanner_activity()).await { + Ok(result) => result, + Err(_) => { + client.evict_connection().await; + Err(Error::other(format!( + "scanner activity peer {host} timed out after retry ({SCANNER_ACTIVITY_PROBE_TIMEOUT:?})" + ))) + } + } +} + #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] async fn call_peer_with_timeout( timeout_dur: Duration, @@ -2882,6 +2922,20 @@ mod tests { assert!(err.to_string().contains("peer-1")); } + #[test] + fn scanner_activity_retry_only_reconnects_transport_failures() { + assert!(scanner_activity_should_retry(None, true)); + assert!(scanner_activity_should_retry(Some(&Error::other("connection refused")), false)); + assert!(!scanner_activity_should_retry( + Some(&Error::other("peer returned an invalid scanner activity response proof")), + false + )); + assert!(!scanner_activity_should_retry( + Some(&Error::from(tonic::Status::internal("peer rejected activity"))), + false + )); + } + #[tokio::test] async fn scanner_dirty_usage_acknowledgement_rejects_missing_and_duplicate_targets() { let sys = NotificationSys {