diff --git a/Cargo.lock b/Cargo.lock index 548869cfd..acd9bee4e 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index 2caf2a84f..63e23d156 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/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>>, +} + +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::>().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::>().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() { diff --git a/crates/rio/Cargo.toml b/crates/rio/Cargo.toml index 19a0057de..7971ad78f 100644 --- a/crates/rio/Cargo.toml +++ b/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 diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index a5378c205..6c52c48e1 100644 --- a/crates/rio/src/http_reader.rs +++ b/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>> = 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 = 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 { 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 { 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 { 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()) diff --git a/rustfs/src/capacity/service.rs b/rustfs/src/capacity/service.rs index d008d42bd..16eda0cbe 100644 --- a/rustfs/src/capacity/service.rs +++ b/rustfs/src/capacity/service.rs @@ -33,8 +33,14 @@ pub fn capacity_disk_ref(endpoint: impl Into, drive_path: impl Into Vec { + // 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"); + } +} diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 1820a3eb0..11a2dcbb2 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -623,6 +623,16 @@ async fn handle_put_file(req: Request) -> Response { 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!(