fix(object-capacity,rio): capacity refresh safety + internode HTTP hardening (#4246)

* fix(object-capacity): stop cancelled/remote-disk refreshes from corrupting capacity

Two independent capacity-refresh bugs (backlog rustfs/backlog#805):

- refresh_or_join set the singleflight `running` flag then awaited the
  refresh future with no drop guard. When the admin request that became
  leader was cancelled (client disconnect) mid-await, `running` stayed true
  forever: joiners blocked indefinitely and the 120s scheduled refresh could
  never start again. catch_unwind covered panics but not cancellation. A
  RefreshLeaderGuard now resets the state and publishes an error on drop.

- capacity_disk_refs mapped the cluster-wide storage_info disk list without
  filtering non-local disks, so admin-triggered refreshes ran a local WalkDir
  over remote disks' drive_path. On multi-node clusters this double-counted
  local bytes (shared mount layout) or hit NotFound (per-node layouts), and
  poisoned the per-disk cache the scheduled local-only refresh depends on,
  making the cached total oscillate. Filter to local disks.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(rio): harden internode HTTP client build, cache, and PUT body integrity

Follow-ups from the internode HTTP review (backlog rustfs/backlog#805):

- handle_put_file accepted a truncated body as success: a HttpWriter dropped
  mid-stream closes the chunked body cleanly, indistinguishable from EOF, and
  the server never compared bytes copied against the declared size. Reject
  size mismatches on the create path (append/unknown-size writes send size<=0
  and are exempt).

- build_http_client used `.expect()` on ClientBuilder::build(), which runs
  lazily on the first request and on every TLS generation bump (cert
  rotation), so a build failure panicked a serving task. It now returns an
  io::Error; get_http_client falls back to the previous TLS generation when a
  rebuild fails instead of failing the request.

- CLIENT_CACHE was a tokio::Mutex taken on every stream open (data_shards
  times per GET). Replaced with arc_swap::ArcSwapOption for lock-free reads on
  the hot path; the generation-monotonic replacement guard is preserved via rcu.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Cargo.lock

@@ -9636,6 +9636,7 @@ name = "rustfs-rio"
 version = "1.0.0-beta.8"
 dependencies = [
  "aes-gcm",
+ "arc-swap",
  "axum",
  "base64 0.22.1",
  "bytes",

crates/object-capacity/src/capacity_manager.rs

@@ -50,6 +50,7 @@ const EVENT_CAPACITY_REFRESH_CACHE_UPDATED: &str = "capacity_refresh_cache_updat
 const EVENT_CAPACITY_REFRESH_WRITE_RECORDED: &str = "capacity_refresh_write_recorded";
 const EVENT_CAPACITY_REFRESH_DEBOUNCE_STATE: &str = "capacity_refresh_debounce_state";
 const EVENT_CAPACITY_REFRESH_PANIC: &str = "capacity_refresh_panic";
+const EVENT_CAPACITY_REFRESH_CANCELLED: &str = "capacity_refresh_cancelled";
 const EVENT_CAPACITY_REFRESH_RUNTIME_SUMMARY: &str = "capacity_refresh_runtime_summary";
 const EVENT_CAPACITY_REFRESH_INTERVAL_CLAMPED: &str = "capacity_refresh_interval_clamped";
 const EVENT_CAPACITY_REFRESH_SCHEDULED: &str = "capacity_refresh_scheduled";
@@ -545,6 +546,55 @@ impl Default for RefreshState {
     }
 }
 
+fn reset_cancelled_refresh_state(state: &mut RefreshState) {
+    state.running = false;
+    record_capacity_refresh_inflight(0);
+    let _ = state
+        .result_tx
+        .send(Some(Err("capacity refresh leader was cancelled".to_string())));
+}
+
+/// Resets the singleflight leader state if the leading future is dropped before the
+/// refresh cycle completes (e.g. the admin request that became leader is cancelled by
+/// a client disconnect). Without this, `running` stays `true` forever: joiners block
+/// indefinitely and no future refresh can start. `catch_unwind` covers panics but not
+/// cancellation, so the reset must live in `Drop`.
+struct RefreshLeaderGuard {
+    state: Option<Arc<Mutex<RefreshState>>>,
+}
+
+impl RefreshLeaderGuard {
+    fn disarm(&mut self) {
+        self.state = None;
+    }
+}
+
+impl Drop for RefreshLeaderGuard {
+    fn drop(&mut self) {
+        let Some(state) = self.state.take() else {
+            return;
+        };
+        warn!(
+            event = EVENT_CAPACITY_REFRESH_CANCELLED,
+            component = LOG_COMPONENT_CAPACITY,
+            subsystem = LOG_SUBSYSTEM_REFRESH,
+            result = "cancelled",
+            "capacity refresh leader dropped before completing; resetting refresh state"
+        );
+        if let Ok(mut guard) = state.try_lock() {
+            reset_cancelled_refresh_state(&mut guard);
+            return;
+        }
+        // The mutex is momentarily held by a joiner subscribing; finish the
+        // reset from a detached task since Drop cannot await.
+        if let Ok(handle) = tokio::runtime::Handle::try_current() {
+            handle.spawn(async move {
+                reset_cancelled_refresh_state(&mut *state.lock().await);
+            });
+        }
+    }
+}
+
 /// Hybrid capacity manager
 pub struct HybridCapacityManager {
     /// Capacity cache
@@ -843,6 +893,13 @@ impl HybridCapacityManager {
                 .unwrap_or_else(|| Err("capacity refresh completed without a result".to_string()));
         }
 
+        // From here on this future is the leader; if it is dropped at any await point
+        // below (request cancellation), the guard resets the singleflight state so
+        // joiners unblock and later refreshes are not wedged behind `running = true`.
+        let mut leader_guard = RefreshLeaderGuard {
+            state: Some(self.refresh_state.clone()),
+        };
+
         let refresh_start = Instant::now();
         let result = AssertUnwindSafe(refresh_fn()).catch_unwind().await.unwrap_or_else(|err| {
             warn!(
@@ -871,6 +928,7 @@ impl HybridCapacityManager {
 
         {
             let mut state = self.refresh_state.lock().await;
+            leader_guard.disarm();
             state.running = false;
             record_capacity_refresh_inflight(0);
             let _ = state.result_tx.send(Some(result.clone()));
@@ -1436,6 +1494,63 @@ mod tests {
         assert_eq!(cached.file_count, 8);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_refresh_or_join_recovers_after_leader_cancellation() {
+        let manager = Arc::new(HybridCapacityManager::from_env());
+
+        // Become the leader with a refresh that never completes, then drop the
+        // future mid-flight to simulate a cancelled admin request.
+        let mgr = manager.clone();
+        let mut leader = Box::pin(mgr.refresh_or_join(DataSource::Scheduled, || async {
+            futures::future::pending::<Result<CapacityUpdate, String>>().await
+        }));
+        assert!(futures::poll!(leader.as_mut()).is_pending());
+        drop(leader);
+        // Let a possibly-spawned reset task run.
+        tokio::task::yield_now().await;
+
+        // A joiner that subscribed to the cancelled cycle must unblock with an error
+        // (not hang), and a subsequent refresh must be able to become the new leader.
+        let refreshed = tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.refresh_or_join(DataSource::WriteTriggered, || async { Ok(CapacityUpdate::exact(1024, 4)) }),
+        )
+        .await
+        .expect("refresh after cancelled leader must not hang")
+        .expect("new leader refresh should succeed");
+        assert_eq!(refreshed.total_used, 1024);
+        assert!(!manager.refresh_in_progress().await);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_refresh_or_join_cancelled_leader_unblocks_joiner() {
+        let manager = Arc::new(HybridCapacityManager::from_env());
+
+        let mgr = manager.clone();
+        let mut leader = Box::pin(mgr.refresh_or_join(DataSource::Scheduled, || async {
+            futures::future::pending::<Result<CapacityUpdate, String>>().await
+        }));
+        assert!(futures::poll!(leader.as_mut()).is_pending());
+
+        // Subscribe a joiner while the leader is still alive.
+        let mgr2 = manager.clone();
+        let joiner = tokio::spawn(async move {
+            mgr2.refresh_or_join(DataSource::WriteTriggered, || async { Ok(CapacityUpdate::exact(2048, 8)) })
+                .await
+        });
+        tokio::time::sleep(Duration::from_millis(20)).await;
+
+        drop(leader);
+
+        let joined = tokio::time::timeout(Duration::from_secs(1), joiner)
+            .await
+            .expect("joiner must unblock after leader cancellation")
+            .expect("joiner task must not panic");
+        assert!(joined.is_err(), "joiner should observe the cancellation error, got {joined:?}");
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_spawn_refresh_if_needed_deduplicates_background_refresh() {

crates/rio/Cargo.toml

@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-rio/latest/rustfs_rio/"
 workspace = true
 
 [dependencies]
+arc-swap.workspace = true
 tokio = { workspace = true, features = ["full"] }
 rand = { workspace = true }
 http.workspace = true

crates/rio/src/http_reader.rs

@@ -33,7 +33,7 @@ use std::sync::LazyLock;
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::mpsc;
 use tokio::time::{self, Sleep};
 use tokio_util::io::StreamReader;
 use tokio_util::sync::PollSender;
@@ -221,7 +221,20 @@ struct CachedClients {
     no_proxy_client: Client,
 }
 
-static CLIENT_CACHE: LazyLock<Mutex<Option<CachedClients>>> = LazyLock::new(|| Mutex::new(None));
+impl CachedClients {
+    fn client_for(&self, disable_proxy: bool) -> Client {
+        if disable_proxy {
+            self.no_proxy_client.clone()
+        } else {
+            self.client.clone()
+        }
+    }
+}
+
+// Lock-free reads: this cache is hit once per stream open (data_shards times per
+// GET), so the hot path must not serialize on a mutex. Writes only happen on
+// outbound-TLS generation bumps (cert rotation), which are rare.
+static CLIENT_CACHE: arc_swap::ArcSwapOption<CachedClients> = arc_swap::ArcSwapOption::const_empty();
 
 const INTERNODE_HTTP_PROFILE_LEGACY: &str = "legacy";
 const INTERNODE_HTTP_PROFILE_BALANCED: &str = "balanced";
@@ -396,7 +409,7 @@ async fn build_http_client(
     disable_proxy: bool,
     tuning: InternodeHttpClientTuning,
     outbound_tls: &rustfs_tls_runtime::GlobalPublishedOutboundTlsState,
-) -> Client {
+) -> io::Result<Client> {
     let mut builder = Client::builder()
         .connect_timeout(std::time::Duration::from_secs(5))
         .tcp_keepalive(std::time::Duration::from_secs(10))
@@ -456,7 +469,11 @@ async fn build_http_client(
         }
     }
 
-    builder.build().expect("Failed to create global HTTP client")
+    // This runs lazily on the first internode request and again on every TLS
+    // generation bump, so a build failure must surface as an error, not a panic.
+    builder
+        .build()
+        .map_err(|err| Error::other(format!("failed to build internode HTTP client: {err}")))
 }
 
 fn should_bypass_proxy_for_url(url: &str) -> bool {
@@ -479,7 +496,7 @@ fn should_disable_proxy_for_url(url: &str, tuning: InternodeHttpClientTuning) ->
     }
 }
 
-async fn get_http_client(url: &str) -> Client {
+async fn get_http_client(url: &str) -> io::Result<Client> {
     let tuning = internode_http_client_tuning();
     // Reuse HTTP connection pools while honoring the configured internode proxy
     // policy. The legacy profile only bypasses loopback URLs to preserve defaults.
@@ -489,44 +506,55 @@ async fn get_http_client(url: &str) -> Client {
     // the full PEM + identity bytes when the TLS state hasn't changed.
     let generation = crate::http_runtime_sources::outbound_tls_generation();
 
-    let guard = CLIENT_CACHE.lock().await;
-    if let Some(cached) = guard.as_ref() {
+    let previous = CLIENT_CACHE.load_full();
+    if let Some(cached) = previous.as_ref() {
         if cached.generation == generation {
-            return if disable_proxy {
-                cached.no_proxy_client.clone()
-            } else {
-                cached.client.clone()
-            };
+            return Ok(cached.client_for(disable_proxy));
         }
         crate::http_runtime_sources::record_stale_outbound_tls_generation("rio_http_reader");
     }
-    drop(guard);
 
     // Cache miss or stale generation — load full outbound TLS state.
     let outbound_tls = crate::http_runtime_sources::outbound_tls_state().await;
 
-    let client = build_http_client(false, tuning, &outbound_tls).await;
-    let no_proxy_client = build_http_client(true, tuning, &outbound_tls).await;
-    let cached = CachedClients {
+    let built = match build_http_client(false, tuning, &outbound_tls).await {
+        Ok(client) => match build_http_client(true, tuning, &outbound_tls).await {
+            Ok(no_proxy_client) => Ok((client, no_proxy_client)),
+            Err(err) => Err(err),
+        },
+        Err(err) => Err(err),
+    };
+    let (client, no_proxy_client) = match built {
+        Ok(pair) => pair,
+        Err(err) => {
+            // Prefer serving with the previous TLS generation over failing the
+            // request outright; the stale-generation metric already fired above.
+            if let Some(cached) = previous {
+                warn!(
+                    error = %err,
+                    stale_generation = cached.generation,
+                    target_generation = generation,
+                    "failed to rebuild internode HTTP client; falling back to previous TLS generation"
+                );
+                return Ok(cached.client_for(disable_proxy));
+            }
+            return Err(err);
+        }
+    };
+    let cached = std::sync::Arc::new(CachedClients {
         generation,
         client,
         no_proxy_client,
-    };
+    });
 
-    let return_client = if disable_proxy {
-        cached.no_proxy_client.clone()
-    } else {
-        cached.client.clone()
-    };
-
-    let mut guard = CLIENT_CACHE.lock().await;
-    // Guard against races: only overwrite the cache if it is empty or
-    // contains an older generation, so a slower task cannot regress the
-    // TLS state after a faster task already cached a newer generation.
-    if guard.as_ref().is_none_or(|c| c.generation <= generation) {
-        *guard = Some(cached);
-    }
-    return_client
+    // Guard against races: only overwrite the cache if it is empty or contains an
+    // older generation, so a slower task cannot regress the TLS state after a
+    // faster task already cached a newer generation.
+    CLIENT_CACHE.rcu(|current| match current {
+        Some(existing) if existing.generation > generation => Some(existing.clone()),
+        _ => Some(cached.clone()),
+    });
+    Ok(cached.client_for(disable_proxy))
 }
 
 fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext {
@@ -651,7 +679,9 @@ impl HttpReader {
     ) -> io::Result<Self> {
         let track_internode_metrics = is_internode_rpc_url(&url);
         let internode_operation = internode_rpc_operation(&url);
-        let client = get_http_client(&url).await;
+        let client = get_http_client(&url).await.inspect_err(|_| {
+            record_internode_error(track_internode_metrics, internode_operation);
+        })?;
         let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone());
         if let Some(body) = body {
             request = request.body(body);
@@ -861,7 +891,14 @@ impl HttpWriter {
             //     "[HttpWriter::spawn] sending HTTP request: url={url_clone}, method={method_clone:?}, headers={headers_clone:?}"
             // );
 
-            let client = get_http_client(&url_clone).await;
+            let client = match get_http_client(&url_clone).await {
+                Ok(client) => client,
+                Err(err) => {
+                    record_internode_error(track_internode_metrics, internode_operation);
+                    let _ = err_tx.send(Error::new(err.kind(), err.to_string()));
+                    return Err(err);
+                }
+            };
             let request = client
                 .request(method_clone.clone(), url_clone.clone())
                 .headers(headers_clone.clone())

rustfs/src/capacity/service.rs

@@ -33,8 +33,14 @@ pub fn capacity_disk_ref(endpoint: impl Into<String>, drive_path: impl Into<Stri
 }
 
 fn capacity_disk_refs(disks: &[rustfs_madmin::Disk]) -> Vec<CapacityDiskRef> {
+    // Admin callers pass cluster-wide `storage_info` disks. The scan walks
+    // `drive_path` on the local filesystem, so remote disks must be excluded:
+    // scanning them either double-counts local bytes (same mount layout on every
+    // node) or fails with NotFound (per-node layouts), and both poison the
+    // per-disk cache that the scheduled local-only refresh relies on.
     disks
         .iter()
+        .filter(|disk| disk.local)
         .map(|disk| capacity_disk_ref(disk.endpoint.clone(), disk.drive_path.clone()))
         .collect()
 }
@@ -327,3 +333,30 @@ fn capacity_source_label(source: capacity_manager::DataSource) -> &'static str {
         capacity_manager::DataSource::Fallback => "fallback",
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn capacity_disk_refs_excludes_remote_disks() {
+        let disks = vec![
+            rustfs_madmin::Disk {
+                endpoint: "http://node1:9000/data/rustfs0".to_string(),
+                drive_path: "/data/rustfs0".to_string(),
+                local: true,
+                ..Default::default()
+            },
+            rustfs_madmin::Disk {
+                endpoint: "http://node2:9000/data/rustfs0".to_string(),
+                drive_path: "/data/rustfs0".to_string(),
+                local: false,
+                ..Default::default()
+            },
+        ];
+
+        let refs = capacity_disk_refs(&disks);
+        assert_eq!(refs.len(), 1);
+        assert_eq!(refs[0].endpoint, "http://node1:9000/data/rustfs0");
+    }
+}

rustfs/src/storage/rpc/http_service.rs

@@ -623,6 +623,16 @@ async fn handle_put_file(req: Request<Incoming>) -> Response<Body> {
         usize::try_from(copied).unwrap_or(usize::MAX),
     );
 
+    if put_body_size_mismatch(&query, copied) {
+        let err = std::io::Error::new(
+            std::io::ErrorKind::UnexpectedEof,
+            format!("body size mismatch: expected {} bytes, received {copied}", query.size),
+        );
+        let message = put_file_stage_error_message("verify_size", &query, &err);
+        log_internode_put_file_stage_failure!("verify_size", query, err);
+        return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message);
+    }
+
     if let Err(e) = file.flush().await {
         let message = put_file_stage_error_message("flush", &query, &e);
         log_internode_put_file_stage_failure!("flush", query, e);
@@ -705,6 +715,14 @@ fn internode_rpc_subsystem(operation: Option<&'static str>) -> &'static str {
     }
 }
 
+/// A writer that is dropped mid-stream (cancelled sender task) terminates the chunked
+/// body cleanly, indistinguishable from intentional EOF. When the client declared the
+/// exact size up front (create path; append and unknown-size writes send `size <= 0`),
+/// a byte-count mismatch means the body was truncated and must not be acknowledged.
+fn put_body_size_mismatch(query: &PutFileQuery, copied: u64) -> bool {
+    !query.append && query.size > 0 && copied != u64::try_from(query.size).unwrap_or(u64::MAX)
+}
+
 fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std::fmt::Display) -> String {
     format!(
         "{stage} file err {err} [disk={}, volume={}, path={}, append={}, size={}]",
@@ -717,7 +735,8 @@ mod tests {
     use super::{
         LOG_SUBSYSTEM_DIRECTORY_WALK, LOG_SUBSYSTEM_FILE_TRANSFER, LOG_SUBSYSTEM_ROUTING, PUT_FILE_STREAM_PATH, PutFileQuery,
         READ_FILE_STREAM_PATH, WALK_DIR_PATH, internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path,
-        put_file_stage_error_message, read_file_body_stream, verify_internode_rpc_signature, write_body_chunks_to_writer,
+        put_body_size_mismatch, put_file_stage_error_message, read_file_body_stream, verify_internode_rpc_signature,
+        write_body_chunks_to_writer,
     };
     use bytes::Bytes;
     use http::{HeaderMap, Method, StatusCode, Uri};
@@ -781,6 +800,26 @@ mod tests {
         assert!(msg.contains("size=1024"));
     }
 
+    #[test]
+    fn put_body_size_mismatch_rejects_truncated_create_only() {
+        let query = |append: bool, size: i64| PutFileQuery {
+            disk: "disk-a".to_string(),
+            volume: "bucket".to_string(),
+            path: "object/part.1".to_string(),
+            append,
+            size,
+        };
+
+        // Truncated (or over-long) body on the create path is rejected.
+        assert!(put_body_size_mismatch(&query(false, 1024), 512));
+        assert!(put_body_size_mismatch(&query(false, 1024), 2048));
+        assert!(!put_body_size_mismatch(&query(false, 1024), 1024));
+        // Append streams send size=0; unknown-size creates send size<=0 — never rejected.
+        assert!(!put_body_size_mismatch(&query(true, 0), 512));
+        assert!(!put_body_size_mismatch(&query(false, 0), 512));
+        assert!(!put_body_size_mismatch(&query(false, -1), 512));
+    }
+
     #[test]
     fn internode_rpc_subsystem_matches_known_operations() {
         assert_eq!(