fix(startup): survive slow multi-node cold starts (#4357)

* fix(server): make startup readiness wait configurable and raise default (#4264)

The startup runtime-readiness wait was a hardcoded 30s constant with no env
override. On slow multi-node cold starts (Docker/K8s/Synology NAS) this window
is shorter than the internal startup budgets it depends on — the endpoint
DNS-retry window (~90s) and the format-load retry loop (~100s worst case) — so
readiness times out and the node exits with
`startup readiness timed out after 30s: storage_ready=false, lock_quorum_ready=false`
before storage/lock quorum can converge, feeding the restart storm reported in
the issue.

- Add `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` (default 120s), documented in
  rustfs-config health constants.
- Resolve the wait at runtime via `startup_runtime_readiness_max_wait()`; a
  value of `0` falls back to the default instead of timing out instantly.
- Repoint `STARTUP_RUNTIME_READINESS_MAX_WAIT` at the shared config default so
  there is a single source of truth, and cover the getter with unit tests.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): stop peer/disk background monitors on graceful shutdown (#4264)

Long-lived peer health/recovery and remote-disk monitors are detached
`tokio::spawn` tasks that each hold a `tracing::Span` via `.instrument(..)` for
their whole lifetime. Nothing cancelled them at shutdown, so on the normal
return path the Tokio runtime was dropped while they were still alive and their
`Span`s were dropped during worker-thread thread-local-storage (TLS)
destruction. At that point `tracing-subscriber`'s fmt `on_close` 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 (SIGILL / exit 132) — the crash
reported on Synology in issue #4264, amplified by the restart storm.

- Add `cluster::rpc::background_monitor` with a process-global shutdown token,
  `spawn_background_monitor()` (races the monitor future against that token so
  its span drops while the runtime is alive), and public
  `shutdown_background_monitors()`.
- Route every span-holding peer_s3 / peer_rest / remote_disk monitor spawn
  through `spawn_background_monitor` instead of `tokio::spawn(..).instrument()`.
- Expose `rustfs_ecstore::shutdown_background_monitors()` and call it from the
  graceful shutdown sequence (right after `ctx.cancel()`, before runtime
  teardown) via the `storage_api` compatibility boundary.

Existing recovery-probe span-context tests still pass, confirming log
correlation is preserved.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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
     }