diff --git a/crates/config/README.md b/crates/config/README.md index 2313a41ec..b5db390c6 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -172,6 +172,14 @@ Drive timeout profile preset: - Then `RUSTFS_DRIVE_MAX_TIMEOUT_DURATION` legacy fallback. - Then the profile-derived default (`default` or `high_latency`). +## Admin peer probe timeout + +- `RUSTFS_ADMIN_PEER_PROBE_TIMEOUT_SECS` + - total per-peer budget for the `server_info`/`storage_info` admin probe round; `server_info` may reconnect once and `storage_info` remains a single attempt. + - default is `10` seconds, preserving the previous two-attempt worst-case budget. + - values must be positive; `0` or an invalid value falls back to the default, and values above `60` are clamped to `60`. + - the setting is read by the aggregating node only; it does not change the internode RPC wire contract. Any retry shares one round deadline rather than receiving a fresh timeout. + ## Startup filesystem boundary policy - `RUSTFS_UNSUPPORTED_FS_POLICY` controls startup behavior when RustFS detects local endpoint filesystems that are outside the supported production boundary. diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index 4a7910632..9fb78ebe2 100644 --- a/crates/config/src/constants/internode.rs +++ b/crates/config/src/constants/internode.rs @@ -39,6 +39,15 @@ pub const DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS: u64 = 20; pub const ENV_INTERNODE_RPC_TIMEOUT_SECS: &str = "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS"; pub const DEFAULT_INTERNODE_RPC_TIMEOUT_SECS: u64 = 30; +/// Total budget for one admin peer probe round, including any reconnect retry. +/// +/// This is intentionally separate from the transport-level RPC timeout: admin +/// probes may retry once, but the retry must consume the same round budget. +pub const ENV_ADMIN_PEER_PROBE_TIMEOUT_SECS: &str = "RUSTFS_ADMIN_PEER_PROBE_TIMEOUT_SECS"; +pub const DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS: u64 = 10; +pub const MAX_ADMIN_PEER_PROBE_TIMEOUT_SECS: u64 = 60; +const _: () = assert!(DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS <= MAX_ADMIN_PEER_PROBE_TIMEOUT_SECS); + // ── Client-side internode gRPC channel tuning (P0) ── // These mirror the server-side HTTP/2 transport tuning in `rustfs/src/server/http.rs` // on the *client* `tonic` `Endpoint` used for internode control-plane RPCs. Prior to @@ -312,6 +321,7 @@ mod tests { assert_eq!(DEFAULT_INTERNODE_HTTP2_KEEPALIVE_INTERVAL_SECS, 5); assert_eq!(DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS, 20); assert_eq!(DEFAULT_INTERNODE_RPC_TIMEOUT_SECS, 30); + assert_eq!(DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS, 10); assert_eq!(DEFAULT_INTERNODE_HTTP_TUNING_PROFILE, "legacy"); } @@ -412,6 +422,7 @@ mod tests { "RUSTFS_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS" ); assert_eq!(ENV_INTERNODE_RPC_TIMEOUT_SECS, "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS"); + assert_eq!(ENV_ADMIN_PEER_PROBE_TIMEOUT_SECS, "RUSTFS_ADMIN_PEER_PROBE_TIMEOUT_SECS"); assert_eq!(ENV_INTERNODE_HTTP_TUNING_PROFILE, "RUSTFS_INTERNODE_HTTP_TUNING_PROFILE"); assert_eq!(ENV_INTERNODE_HTTP_POOL_MAX_IDLE_PER_HOST, "RUSTFS_INTERNODE_HTTP_POOL_MAX_IDLE_PER_HOST"); assert_eq!(ENV_INTERNODE_HTTP_POOL_IDLE_TIMEOUT_SECS, "RUSTFS_INTERNODE_HTTP_POOL_IDLE_TIMEOUT_SECS"); diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 9d57c4647..c6a15a6b9 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -68,7 +68,10 @@ use std::{ }, time::SystemTime, }; -use tokio::{net::TcpStream, time::Duration}; +use tokio::{ + net::TcpStream, + time::{Duration, timeout}, +}; use tonic::Request; use tonic::service::interceptor::InterceptedService; use tracing::{debug, info, warn}; @@ -874,6 +877,16 @@ impl PeerRestClient { self.offline.store(false, Ordering::Release); } + /// Prepare a retry without allowing connection-cache cleanup to extend the + /// caller's probe deadline. The offline gate is cleared even when eviction + /// times out so a cancelled cleanup cannot strand the peer in fast-fail + /// mode; a later request can perform a fresh eviction if needed. + pub async fn prepare_retry_with_timeout(&self, timeout_duration: Duration) -> bool { + let evicted = timeout(timeout_duration, self.evict_connection()).await.is_ok(); + self.offline.store(false, Ordering::Release); + evicted + } + /// Whether this failure means the peer is unreachable, so it should be /// gated offline and its connection evicted. /// diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 04b84a18d..d1955a397 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -72,6 +72,29 @@ const LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4; /// service must not advertise this version until the conditional writer from /// rustfs/backlog#684 is available. const LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION: u32 = 5; + +fn resolve_admin_peer_probe_timeout_secs(configured: Option) -> u64 { + configured + .filter(|seconds| *seconds > 0) + .unwrap_or(rustfs_config::DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS) + .min(rustfs_config::MAX_ADMIN_PEER_PROBE_TIMEOUT_SECS) +} + +fn admin_peer_probe_timeout() -> Duration { + let configured = rustfs_utils::get_env_opt_u64_with_aliases(rustfs_config::ENV_ADMIN_PEER_PROBE_TIMEOUT_SECS, &[]); + let seconds = resolve_admin_peer_probe_timeout_secs(configured); + Duration::from_secs(seconds) +} + +fn remaining_admin_peer_probe_timeout(deadline: Instant) -> Option { + remaining_admin_peer_probe_timeout_at(deadline, Instant::now()) +} + +fn remaining_admin_peer_probe_timeout_at(deadline: Instant, now: Instant) -> Option { + let remaining = deadline.saturating_duration_since(now); + (!remaining.is_zero()).then_some(remaining) +} + type CrossPoolFencePolicyResult = Result>; fn cross_pool_fence_policy_results( @@ -1538,7 +1561,7 @@ impl NotificationSys { { let mut futures = Vec::with_capacity(self.peer_clients.len()); let endpoints = runtime_sources::endpoint_pools().unwrap_or_else(|| Vec::new().into()); - let peer_timeout = Duration::from_secs(5); + let peer_timeout = admin_peer_probe_timeout(); for (idx, client) in self.peer_clients.iter().enumerate() { let endpoints = endpoints.clone(); @@ -1546,7 +1569,9 @@ impl NotificationSys { futures.push(async move { if let Some(client) = client { let host = client.host.to_string(); - match timeout(peer_timeout, client.local_storage_info()).await { + let deadline = Instant::now() + peer_timeout; + let probe_timeout = remaining_admin_peer_probe_timeout(deadline).unwrap_or_default(); + match timeout(probe_timeout, client.local_storage_info()).await { Ok(Ok(mut info)) => { normalize_and_cache_peer_storage_info(cache, &host, &mut info); Some(info) @@ -1557,7 +1582,6 @@ impl NotificationSys { } Err(_) => { warn!("peer {} storage_info timed out after {:?}", host, peer_timeout); - client.evict_connection().await; handle_peer_failure(cache, &host, &endpoints) } } @@ -1583,7 +1607,7 @@ impl NotificationSys { pub async fn server_info(&self) -> Vec { let mut futures = Vec::with_capacity(self.peer_clients.len()); let endpoints = runtime_sources::endpoint_pools().unwrap_or_else(|| Vec::new().into()); - let peer_timeout = Duration::from_secs(5); + let peer_timeout = admin_peer_probe_timeout(); for (idx, client) in self.peer_clients.iter().enumerate() { let host = self @@ -1600,12 +1624,23 @@ impl NotificationSys { }; }; + let deadline = Instant::now() + peer_timeout; + let Some(first_timeout) = remaining_admin_peer_probe_timeout(deadline) else { + let health = peer_disk_health_with_deadline(&host, deadline).await; + return PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + }; + }; + // First attempt. A single evicted or half-open internode channel // is enough to fail one probe and, before retrying, would drop - // the member to unknown/offline for this whole snapshot. So on any - // first-attempt failure we evict the channel and re-dial once - // before falling back (rustfs/backlog#1049, P1-B). - match timeout(peer_timeout, client.server_info()).await { + // the member to unknown/offline for this whole snapshot. On a + // quick failure we evict the channel and re-dial once before + // falling back (rustfs/backlog#1049, P1-B). A slow attempt + // consumes the round budget and therefore does not trigger a + // second full wait or an asynchronous eviction side effect. + match timeout(first_timeout, client.server_info()).await { Ok(Ok(info)) => { return PeerServerInfoProbe { host, result: Ok(info) }; } @@ -1619,14 +1654,37 @@ impl NotificationSys { // `evict_connection` would leave that gate up and the retry would // fast-fail with "temporarily offline" instead of reconnecting // (rustfs/backlog#1049 P1-B). - client.prepare_retry().await; + let Some(retry_budget) = remaining_admin_peer_probe_timeout(deadline) else { + let health = peer_disk_health_with_deadline(&host, deadline).await; + return PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + }; + }; + // Bound connection-cache cleanup too. The helper clears the offline gate even + // when eviction itself times out, so cancellation cannot strand this peer in + // fast-fail mode. + if !client.prepare_retry_with_timeout(retry_budget).await { + let health = peer_disk_health_with_deadline(&host, deadline).await; + return PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + }; + } // Second and final attempt on the fresh channel. - match timeout(peer_timeout, client.server_info()).await { + let Some(retry_timeout) = remaining_admin_peer_probe_timeout(deadline) else { + let health = peer_disk_health_with_deadline(&host, deadline).await; + return PeerServerInfoProbe { + host, + result: Err(PeerServerInfoProbeFailure::Rpc { health }), + }; + }; + match timeout(retry_timeout, client.server_info()).await { Ok(Ok(info)) => PeerServerInfoProbe { host, result: Ok(info) }, Ok(Err(err)) => { warn!("peer {host} server_info failed after retry: {err}"); - let health = peer_disk_health(&host).await; + let health = peer_disk_health_with_deadline(&host, deadline).await; PeerServerInfoProbe { host, result: Err(PeerServerInfoProbeFailure::Rpc { health }), @@ -1634,8 +1692,7 @@ impl NotificationSys { } Err(_) => { warn!("peer {host} server_info timed out after retry ({peer_timeout:?})"); - client.evict_connection().await; - let health = peer_disk_health(&host).await; + let health = peer_disk_health_with_deadline(&host, deadline).await; PeerServerInfoProbe { host, result: Err(PeerServerInfoProbeFailure::Rpc { health }), @@ -3023,6 +3080,11 @@ async fn peer_disk_health(host: &str) -> Option { } } +async fn peer_disk_health_with_deadline(host: &str, deadline: Instant) -> Option { + let remaining = remaining_admin_peer_probe_timeout(deadline)?; + timeout(remaining, peer_disk_health(host)).await.ok().flatten() +} + /// Handle a peer failure for server_info: return cached data if available, or /// classify the member as `unknown` / `degraded` / `offline` depending on how /// many consecutive probes have failed and whether the peer's drives are still @@ -4017,6 +4079,37 @@ mod tests { } } + #[test] + fn admin_peer_probe_timeout_rejects_zero_and_caps_large_values() { + assert_eq!( + resolve_admin_peer_probe_timeout_secs(None), + rustfs_config::DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS + ); + assert_eq!( + resolve_admin_peer_probe_timeout_secs(Some(0)), + rustfs_config::DEFAULT_ADMIN_PEER_PROBE_TIMEOUT_SECS + ); + assert_eq!( + resolve_admin_peer_probe_timeout_secs(Some(rustfs_config::MAX_ADMIN_PEER_PROBE_TIMEOUT_SECS + 1)), + rustfs_config::MAX_ADMIN_PEER_PROBE_TIMEOUT_SECS + ); + assert_eq!(resolve_admin_peer_probe_timeout_secs(Some(7)), 7); + } + + #[tokio::test] + async fn admin_peer_probe_health_fallback_respects_expired_deadline() { + let deadline = Instant::now(); + assert!(peer_disk_health_with_deadline("peer-1", deadline).await.is_none()); + } + + #[test] + fn admin_peer_probe_deadline_is_shared_across_attempts() { + let start = Instant::now(); + let deadline = start + Duration::from_secs(10); + assert!(remaining_admin_peer_probe_timeout_at(deadline, start + Duration::from_secs(6)).is_some()); + assert!(remaining_admin_peer_probe_timeout_at(deadline, start + Duration::from_secs(10)).is_none()); + } + #[tokio::test] async fn call_peer_with_timeout_returns_value_when_fast() { let result = call_peer_with_timeout( diff --git a/docs/operations/admin-peer-probe-timeout.md b/docs/operations/admin-peer-probe-timeout.md new file mode 100644 index 000000000..b7b0e0d8b --- /dev/null +++ b/docs/operations/admin-peer-probe-timeout.md @@ -0,0 +1,27 @@ +# Admin peer probe timeout + +RustFS admin server information and storage information aggregate read-only +state from remote peers. A peer may answer the RPC while its local disk +diagnostic is still recovering after a restart or outage, so these probes use a +bounded per-peer round budget. + +## Configuration + +| Environment variable | Default | Accepted range | Behavior | +| --- | ---: | ---: | --- | +| `RUSTFS_ADMIN_PEER_PROBE_TIMEOUT_SECS` | `10` seconds | `1..=60` seconds | Total budget for one peer probe round; `server_info` may reconnect once, while `storage_info` remains a single attempt. | + +`0` and invalid values fall back to the default. Values above `60` are clamped +to `60`. The timeout is read by the node aggregating the admin response; it is +not a wire or mixed-version protocol setting. + +Any retry shares the same per-peer deadline. A fast transport failure can still +trigger the existing reconnect retry, but a slow first attempt consumes the +remaining budget and cannot add another full timeout. Configure this value +with margin below any external health-check deadline (for example, a +keepalived script timeout); the default preserves the previous two-attempt +worst-case budget and may need to be lowered for a tighter watchdog. + +This setting does not change `RUSTFS_INTERNODE_RPC_TIMEOUT_SECS` or the drive +health policy. A disk probe timeout can still update drive health according to +`RUSTFS_DRIVE_TIMEOUT_HEALTH_ACTION`.