diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index b54922797..1a83fe326 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -27,6 +27,18 @@ pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000; pub const ENV_HEALTH_CLUSTER_TIMEOUT_MS: &str = "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS"; pub const DEFAULT_HEALTH_CLUSTER_TIMEOUT_MS: u64 = 2000; +/// Maximum time to wait for local node runtime readiness (storage / IAM / lock +/// quorum) during startup before failing fast (seconds). +/// +/// On slow or multi-node cold starts — e.g. Docker/Kubernetes/NAS deployments +/// where peer DNS records and erasure-format quorum take time to converge — this +/// budget must be large enough to outlast the internal startup DNS-retry window +/// and the format-load retry loop. Increase it for slow NAS/edge clusters; lower +/// it for fast single-node setups that should fail fast. A value of `0` is +/// treated as the default rather than an instant timeout. +pub const ENV_STARTUP_READINESS_MAX_WAIT_SECS: &str = "RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS"; +pub const DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS: u64 = 120; + /// Enable minimal health payload mode for GET `/health*` responses. /// When enabled, only `status` and `ready` fields are returned. pub const ENV_HEALTH_MINIMAL_RESPONSE_ENABLE: &str = "RUSTFS_HEALTH_MINIMAL_RESPONSE_ENABLE"; diff --git a/crates/ecstore/src/cluster/rpc/background_monitor.rs b/crates/ecstore/src/cluster/rpc/background_monitor.rs new file mode 100644 index 000000000..897b6895e --- /dev/null +++ b/crates/ecstore/src/cluster/rpc/background_monitor.rs @@ -0,0 +1,133 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Lifecycle control for long-lived peer/disk background monitor tasks. +//! +//! Peer health and recovery monitors (see [`super::peer_s3_client`], +//! [`super::peer_rest_client`], [`super::remote_disk`]) are detached +//! `tokio::spawn` tasks that run for the whole life of the process and each hold +//! a `tracing::Span` via `.instrument(..)`. If such a task is still alive when +//! the Tokio runtime is dropped, its future — and the `Span` it holds — is +//! dropped during worker-thread thread-local-storage (TLS) destruction. At that +//! point the `tracing-subscriber` fmt layer's `on_close` handler can touch an +//! already-destroyed TLS slot and panic with +//! `cannot access a Thread Local Storage value during or after destruction`, +//! which escalates to a panic-during-panic abort (see issue #4264). +//! +//! To avoid this, all such monitors are spawned through +//! [`spawn_background_monitor`], which races the monitor future against a +//! process-global shutdown token. Calling [`shutdown_background_monitors`] +//! during graceful shutdown — before the runtime is torn down — resolves every +//! monitor promptly so its `Span` is dropped while the runtime and the tracing +//! subscriber are still alive. + +use std::future::Future; +use std::sync::OnceLock; + +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; +use tracing::{Instrument, Span}; + +/// Process-global shutdown signal for background monitor tasks. +fn global_monitor_shutdown_token() -> &'static CancellationToken { + static TOKEN: OnceLock = OnceLock::new(); + TOKEN.get_or_init(CancellationToken::new) +} + +/// Request shutdown of every background monitor task. +/// +/// Intended to be called once during graceful shutdown, *before* the Tokio +/// runtime is dropped. It is idempotent and safe to call even if no monitors +/// were ever spawned. +pub fn shutdown_background_monitors() { + global_monitor_shutdown_token().cancel(); +} + +/// Spawn a long-lived background monitor task that stops on graceful shutdown. +/// +/// `task` is raced against the global monitor shutdown token; when shutdown is +/// requested the future is dropped promptly (while the runtime is alive), +/// taking `span` with it. This is the only sanctioned way to spawn a monitor +/// that holds an instrumentation `Span` for its whole lifetime. +pub(crate) fn spawn_background_monitor(span: Span, task: F) -> JoinHandle<()> +where + F: Future + Send + 'static, +{ + spawn_background_monitor_on(global_monitor_shutdown_token().clone(), span, task) +} + +/// Core implementation parameterised over the shutdown token, so tests can drive +/// it with a local token without touching the process-global one. +fn spawn_background_monitor_on(shutdown: CancellationToken, span: Span, task: F) -> JoinHandle<()> +where + F: Future + Send + 'static, +{ + tokio::spawn( + async move { + tokio::select! { + _ = shutdown.cancelled() => {} + _ = task => {} + } + } + .instrument(span), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::Duration; + + #[tokio::test] + async fn spawn_background_monitor_on_stops_when_token_cancelled() { + let shutdown = CancellationToken::new(); + let ran_to_completion = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&ran_to_completion); + + // A monitor future that would otherwise never return. + let handle = spawn_background_monitor_on(shutdown.clone(), Span::none(), async move { + std::future::pending::<()>().await; + flag.store(true, Ordering::SeqCst); + }); + + shutdown.cancel(); + + // The wrapper resolves via the shutdown branch and the task completes, + // dropping the never-ending future (and its span) while the runtime is + // alive. + tokio::time::timeout(Duration::from_secs(5), handle) + .await + .expect("monitor task should stop after shutdown") + .expect("monitor task should not panic"); + assert!(!ran_to_completion.load(Ordering::SeqCst), "inner future must be cancelled, not completed"); + } + + #[tokio::test] + async fn spawn_background_monitor_on_completes_when_task_finishes() { + let shutdown = CancellationToken::new(); + let handle = spawn_background_monitor_on(shutdown, Span::none(), async {}); + tokio::time::timeout(Duration::from_secs(5), handle) + .await + .expect("finished monitor task should join") + .expect("monitor task should not panic"); + } + + // NOTE: intentionally no test cancels `global_monitor_shutdown_token()` — it + // is a process-global shared by every test in this binary that spawns a real + // monitor via `spawn_background_monitor`, so cancelling it here would race + // those tests. The cancellation mechanism is covered above via the + // token-parameterised `spawn_background_monitor_on`. +} diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index b7e860bcb..f6f5ec284 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(crate) mod background_monitor; pub(crate) mod client; pub(crate) mod context_propagation; pub(crate) mod http_auth; @@ -22,6 +23,8 @@ pub(crate) mod remote_disk; pub(crate) mod remote_locker; pub(crate) mod runtime_sources; +pub use background_monitor::shutdown_background_monitors; +pub(crate) use background_monitor::spawn_background_monitor; pub use client::{ TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth, }; diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 23227652e..7279e84b4 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -53,7 +53,7 @@ use tokio::{net::TcpStream, time::Duration}; use tonic::Request; use tonic::service::interceptor::InterceptedService; use tonic::transport::Channel; -use tracing::{Instrument, debug, warn}; +use tracing::{debug, warn}; pub const PEER_RESTSIGNAL: &str = "signal"; pub const PEER_RESTSUB_SYS: &str = "sub-sys"; @@ -184,31 +184,28 @@ impl PeerRestClient { let offline = Arc::clone(&self.offline); let recovery_running = Arc::clone(&self.recovery_running); let span = Self::recovery_monitor_span(&grid_host); - tokio::spawn( - async move { - let mut delay = get_drive_active_check_interval(); - let connect_timeout = get_drive_active_check_timeout(); + super::spawn_background_monitor(span, async move { + let mut delay = get_drive_active_check_interval(); + let connect_timeout = get_drive_active_check_timeout(); - for _ in 0..PEER_REST_RECOVERY_MAX_ATTEMPTS { - tokio::time::sleep(delay).await; - if Self::perform_connectivity_check(&grid_host, connect_timeout).await.is_ok() { - offline.store(false, Ordering::Release); - recovery_running.store(false, Ordering::Release); - return; - } - - delay = std::cmp::min(delay.saturating_mul(2), PEER_REST_RECOVERY_MAX_BACKOFF); + for _ in 0..PEER_REST_RECOVERY_MAX_ATTEMPTS { + tokio::time::sleep(delay).await; + if Self::perform_connectivity_check(&grid_host, connect_timeout).await.is_ok() { + offline.store(false, Ordering::Release); + recovery_running.store(false, Ordering::Release); + return; } - warn!( - grid_host = %grid_host, - attempts = PEER_REST_RECOVERY_MAX_ATTEMPTS, - "peer recovery monitor reached max attempts; will retry on next request" - ); - recovery_running.store(false, Ordering::Release); + delay = std::cmp::min(delay.saturating_mul(2), PEER_REST_RECOVERY_MAX_BACKOFF); } - .instrument(span), - ); + + warn!( + grid_host = %grid_host, + attempts = PEER_REST_RECOVERY_MAX_ATTEMPTS, + "peer recovery monitor reached max attempts; will retry on next request" + ); + recovery_running.store(false, Ordering::Release); + }); } #[cfg(test)] @@ -216,13 +213,10 @@ impl PeerRestClient { let (tx, rx) = tokio::sync::oneshot::channel(); let grid_host = self.grid_host.clone(); let span = Self::recovery_monitor_span(&grid_host); - tokio::spawn( - async move { - warn!(grid_host = %grid_host, "peer recovery monitor log probe"); - let _ = tx.send(()); - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + warn!(grid_host = %grid_host, "peer recovery monitor log probe"); + let _ = tx.send(()); + }); rx } diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 2640c60b3..ca6744c86 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -45,7 +45,7 @@ use tokio_util::sync::CancellationToken; use tonic::Request; use tonic::service::interceptor::InterceptedService; use tonic::transport::Channel; -use tracing::{Instrument, debug, info, warn}; +use tracing::{debug, info, warn}; type Client = Arc>; @@ -693,12 +693,9 @@ impl RemotePeerS3Client { let cancel_clone = cancel_token.clone(); let span = Self::recovery_monitor_span(&addr_clone); - tokio::spawn( - async move { - Self::monitor_remote_peer_recovery(addr_clone, health_clone, cancel_clone).await; - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + Self::monitor_remote_peer_recovery(addr_clone, health_clone, cancel_clone).await; + }); } } } @@ -802,12 +799,9 @@ impl RemotePeerS3Client { let cancel_token = self.cancel_token.clone(); let addr = self.addr.clone(); let span = Self::recovery_monitor_span(&addr); - tokio::spawn( - async move { - Self::monitor_remote_peer_recovery(addr, health, cancel_token).await; - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + Self::monitor_remote_peer_recovery(addr, health, cancel_token).await; + }); } } } diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index acd94726f..177f39f3d 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -68,7 +68,7 @@ use tokio::{ }; use tokio_util::sync::CancellationToken; use tonic::{Code, Request, service::interceptor::InterceptedService, transport::Channel}; -use tracing::{Instrument, debug, warn}; +use tracing::{debug, warn}; use uuid::Uuid; #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -393,12 +393,9 @@ impl RemoteDisk { let health = Arc::clone(&self.health); let cancel_token = self.cancel_token.clone(); let span = Self::recovery_monitor_span(&addr, &endpoint); - tokio::spawn( - async move { - Self::monitor_remote_disk_recovery(addr, endpoint, health, cancel_token).await; - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + Self::monitor_remote_disk_recovery(addr, endpoint, health, cancel_token).await; + }); } #[cfg(test)] @@ -407,21 +404,18 @@ impl RemoteDisk { let endpoint = self.endpoint.clone(); let addr = self.addr.clone(); let span = Self::recovery_monitor_span(&addr, &endpoint); - tokio::spawn( - async move { - warn!( - event = EVENT_REMOTE_DISK_HEALTH, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REMOTE_DISK, - endpoint = %endpoint, - addr, - state = "probe", - "remote disk recovery monitor log probe" - ); - let _ = tx.send(()); - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + warn!( + event = EVENT_REMOTE_DISK_HEALTH, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_DISK, + endpoint = %endpoint, + addr, + state = "probe", + "remote disk recovery monitor log probe" + ); + let _ = tx.send(()); + }); rx } @@ -478,12 +472,9 @@ impl RemoteDisk { let cancel_clone = cancel_token.clone(); let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone); - tokio::spawn( - async move { - Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; + }); } loop { @@ -544,12 +535,9 @@ impl RemoteDisk { let cancel_clone = cancel_token.clone(); let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone); - tokio::spawn( - async move { - Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; - } - .instrument(span), - ); + super::spawn_background_monitor(span, async move { + Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; + }); } } } diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index c395395a5..85b5637c3 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -48,6 +48,17 @@ pub fn set_workload_admission_snapshot_provider( runtime::sources::set_workload_admission_snapshot_provider(provider) } +/// Request shutdown of all long-lived peer/disk background monitor tasks. +/// +/// Call this during graceful shutdown, *before* the Tokio runtime is dropped, so +/// each monitor future (and the `tracing::Span` it holds) is dropped while the +/// runtime and tracing subscriber are still alive. This avoids the +/// thread-local-storage `on_close` panic that can otherwise abort the process +/// during worker-thread teardown (issue #4264). Idempotent and cheap. +pub fn shutdown_background_monitors() { + cluster::rpc::shutdown_background_monitors(); +} + #[cfg(test)] mod rio_tests { #[test] diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index b290a39e7..4634adad0 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -43,8 +43,34 @@ use tokio::sync::Mutex; use tower::{Layer, Service}; use tracing::{debug, info}; -pub const STARTUP_RUNTIME_READINESS_MAX_WAIT: Duration = Duration::from_secs(30); +/// Default upper bound for the startup runtime-readiness wait. +/// +/// This is the compile-time fallback used when +/// [`rustfs_config::ENV_STARTUP_READINESS_MAX_WAIT_SECS`] is unset or `0`. The +/// effective value is resolved at runtime by [`startup_runtime_readiness_max_wait`], +/// which lets slow multi-node cold starts (Docker/K8s/NAS) extend the budget past +/// the internal DNS-retry and format-load windows without a rebuild. +pub const STARTUP_RUNTIME_READINESS_MAX_WAIT: Duration = + Duration::from_secs(rustfs_config::DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS); pub const STARTUP_RUNTIME_READINESS_POLL_INTERVAL: Duration = Duration::from_secs(1); + +/// Resolve the effective startup runtime-readiness wait. +/// +/// Reads `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS`, falling back to +/// [`STARTUP_RUNTIME_READINESS_MAX_WAIT`] when unset. A configured value of `0` +/// is treated as "use the default" rather than an instant timeout, so an empty or +/// misconfigured env var can never make the server give up on readiness immediately. +fn startup_runtime_readiness_max_wait() -> Duration { + let secs = rustfs_utils::get_env_u64( + rustfs_config::ENV_STARTUP_READINESS_MAX_WAIT_SECS, + rustfs_config::DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS, + ); + if secs == 0 { + STARTUP_RUNTIME_READINESS_MAX_WAIT + } else { + Duration::from_secs(secs) + } +} const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready"; const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total"; @@ -227,7 +253,7 @@ pub async fn publish_ready_when_runtime_ready( state_manager: Option<&ServiceStateManager>, ) -> Result<(), std::io::Error> { wait_for_runtime_readiness_with( - STARTUP_RUNTIME_READINESS_MAX_WAIT, + startup_runtime_readiness_max_wait(), STARTUP_RUNTIME_READINESS_POLL_INTERVAL, collect_node_readiness, |dependency_readiness| { @@ -914,10 +940,32 @@ mod tests { #[test] fn startup_runtime_readiness_wait_constants_are_ordered() { assert!(STARTUP_RUNTIME_READINESS_MAX_WAIT > STARTUP_RUNTIME_READINESS_POLL_INTERVAL); - assert_eq!(STARTUP_RUNTIME_READINESS_MAX_WAIT.as_secs(), 30); + assert_eq!( + STARTUP_RUNTIME_READINESS_MAX_WAIT.as_secs(), + rustfs_config::DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS + ); assert_eq!(STARTUP_RUNTIME_READINESS_POLL_INTERVAL.as_secs(), 1); } + #[test] + #[serial] + fn startup_runtime_readiness_max_wait_reads_env_override() { + // An explicit override extends (or shortens) the budget. + with_var(rustfs_config::ENV_STARTUP_READINESS_MAX_WAIT_SECS, Some("240"), || { + assert_eq!(startup_runtime_readiness_max_wait(), Duration::from_secs(240)); + }); + + // Unset falls back to the compile-time default. + with_var(rustfs_config::ENV_STARTUP_READINESS_MAX_WAIT_SECS, None::<&str>, || { + assert_eq!(startup_runtime_readiness_max_wait(), STARTUP_RUNTIME_READINESS_MAX_WAIT); + }); + + // Zero is treated as "use default", never an instant timeout. + with_var(rustfs_config::ENV_STARTUP_READINESS_MAX_WAIT_SECS, Some("0"), || { + assert_eq!(startup_runtime_readiness_max_wait(), STARTUP_RUNTIME_READINESS_MAX_WAIT); + }); + } + fn peer_health_test_pools() -> EndpointServerPools { EndpointServerPools::from(vec![PoolEndpoints { legacy: false, diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index 051ce9c91..28197cfb4 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/rustfs/src/startup_shutdown.rs @@ -69,6 +69,12 @@ pub(crate) async fn run_startup_shutdown_sequence( ) { ctx.cancel(); + // Stop long-lived peer/disk background monitors while the runtime and tracing + // subscriber are still alive, so the `tracing::Span` each monitor holds is + // dropped here instead of during worker-thread TLS destruction at runtime + // teardown (which can panic in the fmt layer's `on_close`; see issue #4264). + crate::storage_api::startup::shutdown::shutdown_background_monitors(); + info!( target: "rustfs::main::handle_shutdown", event = EVENT_SHUTDOWN_SIGNAL_RECEIVED, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index a6304c66f..89430e120 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -782,6 +782,10 @@ pub(crate) fn shutdown_background_services() { ecstore_global::shutdown_background_services(); } +pub(crate) fn shutdown_background_monitors() { + rustfs_ecstore::shutdown_background_monitors(); +} + pub(crate) fn set_global_endpoints(endpoints: Vec) { ecstore_global::set_global_endpoints(endpoints); } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 7785efc4b..4f83c6611 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -219,7 +219,9 @@ pub(crate) mod startup { } pub(crate) mod shutdown { - pub(crate) use crate::storage::storage_api::{shutdown_background_services, store_compression_total_in_backend}; + pub(crate) use crate::storage::storage_api::{ + shutdown_background_monitors, shutdown_background_services, store_compression_total_in_backend, + }; } pub(crate) mod storage {