diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 872fbf3d5..f174f502f 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -24,9 +24,10 @@ use std::{ io::ErrorKind, pin::Pin, sync::{Arc, OnceLock}, + task::{Context, Poll}, time::Duration, }; -use tokio::io::AsyncRead; +use tokio::io::{AsyncRead, AsyncWrite}; use tokio::spawn; use tokio::sync::Mutex as TokioMutex; use tokio::time::timeout; @@ -102,6 +103,56 @@ fn duration_millis(duration: Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } +fn classify_listing_quorum_failure(errors: &[DiskError]) -> DiskError { + if errors.contains(&DiskError::Timeout) { + return DiskError::Timeout; + } + if let Some(first) = errors.first() + && !matches!(first, DiskError::Io(_)) + && errors.iter().all(|err| err == first) + { + return first.clone(); + } + DiskError::ErasureReadQuorum +} + +struct PublishedBytesWriter { + inner: W, + published: bool, +} + +impl PublishedBytesWriter { + fn new(inner: W) -> Self { + Self { inner, published: false } + } + + fn has_published(&self) -> bool { + self.published + } +} + +impl AsyncWrite for PublishedBytesWriter { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + match Pin::new(&mut self.inner).poll_write(cx, buf) { + Poll::Ready(Ok(written)) => { + if written != 0 && !self.published { + self.published = true; + } + Poll::Ready(Ok(written)) + } + other => other, + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + #[cfg(test)] #[derive(Clone)] pub(crate) enum TestReaderBehavior { @@ -223,12 +274,12 @@ async fn list_path_raw_inner( let (rd, wr) = tokio::io::duplex(64); readers.push(MetacacheReader::new(rd)); jobs.push(spawn(async move { + let mut wr = PublishedBytesWriter::new(wr); #[cfg(test)] let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() { match behavior { TestReaderBehavior::Eof => return Ok(()), TestReaderBehavior::Entries(entries) => { - let mut wr = wr; let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); out.write(&entries).await.expect("test entries should be written"); out.close().await.expect("test entries should close"); @@ -250,20 +301,17 @@ async fn list_path_raw_inner( } TestReaderBehavior::PrimaryErrorThenFallback(err) => Some(err), TestReaderBehavior::PartialThenTimeout(entries) => { - let mut wr = wr; let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); let err = DiskError::Timeout; - record_producer_error(&producer_errs_clone, disk_idx, &err); let _ = out.write(&entries).await; drop(out); - return Err(err); + Some(err) } } } else { None }; - let mut wr = wr; let wakl_opts = WalkDirOptions { bucket: opts_clone.bucket.clone(), base_dir: opts_clone.path.clone(), @@ -283,6 +331,10 @@ async fn list_path_raw_inner( let mut last_err = None; #[cfg(test)] if let Some(err) = test_primary_error { + if wr.has_published() { + record_producer_error(&producer_errs_clone, disk_idx, &err); + return Err(err); + } last_err = Some(err); need_fallback = true; } @@ -333,6 +385,10 @@ async fn list_path_raw_inner( "Metacache walk_dir failed" ); } + if wr.has_published() { + record_producer_error(&producer_errs_clone, disk_idx, &err); + return Err(err); + } last_err = Some(err); need_fallback = true; } @@ -371,12 +427,18 @@ async fn list_path_raw_inner( last_err = Some(err); continue; } - TestReaderBehavior::Stall - | TestReaderBehavior::IgnoreCancel - | TestReaderBehavior::PartialThenTimeout(_) => { + TestReaderBehavior::Stall | TestReaderBehavior::IgnoreCancel => { last_err = Some(DiskError::Timeout); continue; } + TestReaderBehavior::PartialThenTimeout(entries) => { + let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); + let err = DiskError::Timeout; + let _ = out.write(&entries).await; + drop(out); + record_producer_error(&producer_errs_clone, disk_idx, &err); + return Err(err); + } } } @@ -489,6 +551,10 @@ async fn list_path_raw_inner( "Metacache fallback walk_dir failed" ); } + if wr.has_published() { + record_producer_error(&producer_errs_clone, disk_idx, &err); + return Err(err); + } last_err = Some(err); } } @@ -693,15 +759,6 @@ async fn list_path_raw_inner( if let Some(finished_fn) = opts.finished.as_ref() { finished_fn(&errs).await; } - if errs.iter().flatten().any(|err| *err == DiskError::Timeout) { - return Err(DiskError::Timeout); - } - let mut err_iter = errs.iter().flatten(); - if let Some(err) = err_iter.next() - && err_iter.next().is_none() - { - return Err(err.clone()); - } let mut combined_err = Vec::new(); errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) { (Some(err), Some(disk)) => { @@ -723,7 +780,8 @@ async fn list_path_raw_inner( error = %combined_err.join(", "), "Metacache listing quorum failed" ); - return Err(DiskError::other(combined_err.join(", "))); + let failures = errs.iter().flatten().cloned().collect::>(); + return Err(classify_listing_quorum_failure(&failures)); } // Break if all at EOF or error. @@ -872,7 +930,7 @@ async fn list_path_raw_inner( } if job_errs.len() > max_disk_failures { - return Err(job_errs.remove(0)); + return Err(classify_listing_quorum_failure(&job_errs)); } // warn!("list_path_raw: done"); @@ -923,6 +981,44 @@ mod tests { assert_eq!(err, DiskError::ErasureReadQuorum); } + #[tokio::test] + async fn list_path_raw_returns_typed_quorum_error_for_multiple_drive_failures() { + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None], + min_disks: 2, + test_reader_behaviors: vec![ + TestReaderBehavior::ProducerError(DiskError::other("/sensitive/disk-a")), + TestReaderBehavior::ProducerError(DiskError::other("http://internal-node/disk-b")), + TestReaderBehavior::Eof, + ], + ..Default::default() + }, + ) + .await + .expect_err("multiple producer failures beyond tolerance should fail listing quorum"); + + assert_eq!(err, DiskError::ErasureReadQuorum); + assert!(!err.to_string().contains("sensitive")); + assert!(!err.to_string().contains("internal-node")); + } + + #[test] + fn listing_quorum_failure_never_returns_raw_io_details() { + let err = classify_listing_quorum_failure(&[DiskError::other("/sensitive/disk-a")]); + + assert_eq!(err, DiskError::ErasureReadQuorum); + assert!(!err.to_string().contains("sensitive")); + + let access_denied = classify_listing_quorum_failure(&[DiskError::FileAccessDenied, DiskError::FileAccessDenied]); + assert_eq!(access_denied, DiskError::FileAccessDenied); + + let mixed_timeout = + classify_listing_quorum_failure(&[DiskError::Timeout, DiskError::FileAccessDenied, DiskError::ErasureReadQuorum]); + assert_eq!(mixed_timeout, DiskError::Timeout); + } + #[test] fn missing_path_error_classification_excludes_actionable_failures() { assert!(is_missing_path_error(&DiskError::FileNotFound)); @@ -1067,6 +1163,27 @@ mod tests { assert_eq!(err, DiskError::Timeout); } + #[tokio::test] + async fn list_path_raw_prefers_timeout_for_mixed_errors_beyond_quorum_budget() { + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None], + min_disks: 2, + test_reader_behaviors: vec![ + TestReaderBehavior::ProducerError(DiskError::FileAccessDenied), + TestReaderBehavior::ProducerError(DiskError::Timeout), + TestReaderBehavior::Eof, + ], + ..Default::default() + }, + ) + .await + .expect_err("mixed failures beyond the tolerated budget should fail listing"); + + assert_eq!(err, DiskError::Timeout); + } + #[tokio::test] async fn list_path_raw_tolerates_stalled_reader_after_quorum_eof() { list_path_raw( @@ -1211,8 +1328,9 @@ mod tests { async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() { let seen = Arc::new(Mutex::new(Vec::new())); let seen_clone = seen.clone(); + let claim_tracker = FallbackClaimTracker::default(); - let err = list_path_raw( + let err = list_path_raw_with_claim_tracker( CancellationToken::new(), ListPathRawOptions { disks: vec![None], @@ -1223,6 +1341,7 @@ mod tests { cached: None, reusable: false, }])], + test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], agreed: Some(Box::new(move |entry: MetaCacheEntry| { let seen = seen_clone.clone(); Box::pin(async move { @@ -1231,12 +1350,59 @@ mod tests { })), ..Default::default() }, + claim_tracker.clone(), ) .await .expect_err("producer timeout after partial output must fail the listing"); assert_eq!(err, DiskError::Timeout); assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/object".to_string()]); + assert!( + claim_tracker.claimed_keys().await.is_empty(), + "fallback must not append a second metacache stream after partial output" + ); + } + + #[tokio::test] + async fn list_path_raw_does_not_try_second_fallback_after_partial_fallback_output() { + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + let claim_tracker = FallbackClaimTracker::default(); + + let err = list_path_raw_with_claim_tracker( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None], + min_disks: 1, + test_reader_behaviors: vec![TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound)], + test_fallback_reader_behaviors: vec![ + TestReaderBehavior::PartialThenTimeout(vec![MetaCacheEntry { + name: "bucket/partial-fallback".to_string(), + metadata: vec![1, 2, 3], + cached: None, + reusable: false, + }]), + TestReaderBehavior::Entries(vec![fallback_test_entry()]), + ], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + Box::pin(async move { + seen.lock().expect("seen mutex poisoned").push(entry.name); + }) + })), + ..Default::default() + }, + claim_tracker.clone(), + ) + .await + .expect_err("a fallback timeout after partial output must fail the listing"); + + assert_eq!(err, DiskError::Timeout); + assert_eq!( + seen.lock().expect("seen mutex poisoned").as_slice(), + &["bucket/partial-fallback".to_string()] + ); + assert_eq!(claim_tracker.claimed_keys().await.len(), 1, "a second fallback must not be claimed"); } #[tokio::test] diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 9d072652d..28885e9c5 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -69,6 +69,13 @@ fn signature_payload(url: &str, method: &Method, timestamp: i64) -> String { format!("{url}|{method}|{timestamp}") } +fn redacted_rpc_path(url: &str) -> String { + url.parse::() + .ok() + .map(|uri| uri.path().to_string()) + .unwrap_or_else(|| "".to_string()) +} + /// Generate HMAC-SHA256 signature for the given data fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64) -> String { let data = signature_payload(url, method, timestamp); @@ -146,12 +153,13 @@ pub fn verify_rpc_signature(url: &str, method: &Method, headers: &HeaderMap) -> let secret = get_shared_secret()?; if !verify_signature(&secret, url, method, timestamp, signature) { + let rpc_path = redacted_rpc_path(url); error!( - "verify_rpc_signature: Invalid signature: url {}, method {}, timestamp {}, signature_len {}", - url, - method, + rpc_path = %rpc_path, + method = %method, timestamp, - signature.len() + signature_len = signature.len(), + "verify_rpc_signature: Invalid signature" ); return Err(std::io::Error::other("Invalid signature")); @@ -386,10 +394,31 @@ mod tests { assert!(!verify_signature(secret, url, &method, timestamp, "invalid-signature")); } + #[test] + fn walk_dir_capability_is_covered_by_the_signature() { + let secret = "test-secret"; + let signed_url = concat!( + "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a&walk_dir_stream_completion=error-v1", + "&walk_dir_body_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ); + let downgraded_url = "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a"; + let tampered_body_digest = concat!( + "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a&walk_dir_stream_completion=error-v1", + "&walk_dir_body_sha256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ); + let method = Method::GET; + let timestamp = 1_640_995_200; + let signature = generate_signature(secret, signed_url, &method, timestamp); + + assert!(verify_signature(secret, signed_url, &method, timestamp, &signature)); + assert!(!verify_signature(secret, downgraded_url, &method, timestamp, &signature)); + assert!(!verify_signature(secret, tampered_body_digest, &method, timestamp, &signature)); + } + #[test] fn test_invalid_signature_log_contract_excludes_secrets() { ensure_test_rpc_secret(); - let url = "http://example.com/api/test"; + let url = "http://example.com/api/test?disk=/sensitive/path&token=private"; let method = Method::GET; let timestamp = OffsetDateTime::now_utc().unix_timestamp(); let secret = get_shared_secret().expect("test RPC secret should resolve"); @@ -417,6 +446,8 @@ mod tests { assert!(!captured.contains(&secret)); assert!(!captured.contains(&expected_signature)); assert!(!captured.contains(invalid_signature)); + assert!(!captured.contains("sensitive")); + assert!(!captured.contains("private")); } #[test] diff --git a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs index c1c788284..b7138b111 100644 --- a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs +++ b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs @@ -15,6 +15,9 @@ use crate::cluster::rpc::build_auth_headers; use crate::disk::error::{Error, Result}; use crate::disk::{FileReader, FileWriter}; +use crate::storage_api_contracts::internode::{ + WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, +}; use async_trait::async_trait; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use rustfs_config::{ @@ -22,6 +25,7 @@ use rustfs_config::{ KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS, }; use rustfs_rio::{HttpReader, HttpWriter}; +use sha2::{Digest, Sha256}; use std::sync::{Arc, OnceLock}; use std::time::Duration; @@ -180,7 +184,17 @@ fn build_put_file_stream_url(request: &WriteStreamRequest) -> String { } fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String { - format!("{}{}?disk={}", request.endpoint, WALK_DIR_PATH, urlencoding::encode(&request.disk)) + let body_sha256 = hex_simd::encode_to_string(Sha256::digest(&request.body), hex_simd::AsciiCase::Lower); + format!( + "{}{}?disk={}&{}={}&{}={}", + request.endpoint, + WALK_DIR_PATH, + urlencoding::encode(&request.disk), + WALK_DIR_STREAM_COMPLETION_QUERY, + WALK_DIR_STREAM_COMPLETION_V1, + WALK_DIR_BODY_SHA256_QUERY, + body_sha256 + ) } fn json_headers() -> HeaderMap { @@ -300,7 +314,11 @@ mod tests { assert_eq!( url, - "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0" + concat!( + "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0", + "&walk_dir_stream_completion=error-v1", + "&walk_dir_body_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ) ); } diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index b81e6c159..900d2e780 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -827,14 +827,13 @@ impl RemoteDisk { .expect("operation should succeed") .as_nanos() as i64; self.health.last_started.store(now, std::sync::atomic::Ordering::Relaxed); - self.health.increment_waiting(); + let _waiting_guard = self.health.waiting_guard(); if timeout_duration == Duration::ZERO { let operation_result = operation().await; if operation_result.is_ok() { self.health.log_success(); } - self.health.decrement_waiting(); self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action) .await; return operation_result; @@ -845,18 +844,16 @@ impl RemoteDisk { match result { Ok(operation_result) => { - // Log success and decrement waiting counter + // Log success; the waiting guard balances every exit path. if operation_result.is_ok() { self.health.log_success(); } - self.health.decrement_waiting(); self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action) .await; operation_result } Err(_) => { // Timeout occurred, mark disk as potentially faulty - self.health.decrement_waiting(); counter!( "rustfs_drive_op_timeout_total", "endpoint" => self.endpoint.to_string(), @@ -3314,6 +3311,33 @@ mod tests { .expect("operation should succeed") } + #[tokio::test] + async fn remote_disk_health_wrapper_balances_task_cancellation() { + let disk = Arc::new(new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await); + let task_disk = Arc::clone(&disk); + let task = tokio::spawn(async move { + task_disk + .execute_with_timeout_for_op( + "cancellation-test", + || async { std::future::pending::>().await }, + Duration::ZERO, + ) + .await + }); + + tokio::time::timeout(Duration::from_secs(1), async { + while disk.health.waiting_count() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("operation should enter remote disk health tracking"); + task.abort(); + let _ = task.await; + + assert_eq!(disk.health.waiting_count(), 0); + } + #[derive(Debug)] struct PartialThenErrorReader { cursor: Cursor>, diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 3e1c38511..8f524d3fb 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -19,7 +19,7 @@ pub mod local_snapshot; use crate::storage_api_contracts::{ bucket::{BucketOperations as _, BucketOptions}, - list::ListOperations as _, + list::{ListOperations as _, StorageListObjectVersionsInfo}, object::ObjectIO as _, }; use crate::{ @@ -27,8 +27,9 @@ use crate::{ config::com::read_config, disk::DiskAPI, error::{Error, classify_system_path_failure_reason}, + object_api::ObjectInfo, runtime::sources as runtime_sources, - store::ECStore, + store::{ECStore, list_objects::list_marker_key}, }; pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path}; use rustfs_data_usage::{ @@ -39,6 +40,7 @@ use rustfs_io_metrics::record_system_path_failure; use rustfs_utils::path::SLASH_SEPARATOR; use std::{ collections::{HashMap, HashSet, hash_map::Entry}, + future::Future, sync::{Arc, LazyLock, OnceLock}, time::{Duration, SystemTime}, }; @@ -53,6 +55,7 @@ const DATA_COMPRESSION_TOTAL_NAME: &str = ".compression.json"; const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin"; pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin"; const DATA_USAGE_CACHE_TTL_SECS: u64 = 30; +const LIVE_BUCKET_USAGE_MAX_ENTRIES: u64 = 1024; #[derive(Debug, Clone)] struct CachedBucketUsage { @@ -67,9 +70,11 @@ struct CachedBucketUsage { type UsageMemoryCache = Arc>>; type CacheUpdating = Arc>; +type LiveBucketUsageCache = moka::future::Cache; static USAGE_MEMORY_CACHE: OnceLock = OnceLock::new(); static USAGE_CACHE_UPDATING: OnceLock = OnceLock::new(); +static LIVE_BUCKET_USAGE_CACHE: OnceLock = OnceLock::new(); /// Deferred persist thresholds for compression totals: persist after this many /// operations recorded, but no more often than the min interval. @@ -110,6 +115,14 @@ fn cache_updating() -> &'static CacheUpdating { USAGE_CACHE_UPDATING.get_or_init(|| Arc::new(RwLock::new(false))) } +fn live_bucket_usage_cache() -> &'static LiveBucketUsageCache { + LIVE_BUCKET_USAGE_CACHE.get_or_init(|| { + moka::future::Cache::builder() + .max_capacity(LIVE_BUCKET_USAGE_MAX_ENTRIES) + .build() + }) +} + // Data usage storage paths lazy_static::lazy_static! { pub static ref DATA_USAGE_BUCKET: String = format!("{}{}{}", @@ -208,6 +221,7 @@ async fn clear_bucket_usage_memory(bucket: &str) { } pub async fn remove_bucket_usage_from_backend(store: Arc, bucket: &str) -> Result<(), Error> { + live_bucket_usage_cache().invalidate(bucket).await; clear_bucket_usage_memory(bucket).await; let data_usage_info = load_data_usage_from_backend(store.clone()).await?; @@ -442,79 +456,203 @@ pub async fn aggregate_local_snapshots(store: Arc) -> Result<(Vec, + // FileMeta caps versions per object, so replay detection remains bounded. + current_object_versions: HashSet>, + current_live_versions: u64, + objects_count: u64, + versions_count: u64, + total_size: u64, + delete_markers: u64, + size_histogram: SizeHistogram, + versions_histogram: VersionsHistogram, +} + +impl BucketUsageAccumulator { + fn record(&mut self, bucket: &str, object: &ObjectInfo) -> Result<(), Error> { + if object.is_dir { + return Ok(()); + } + + if self + .current_object_name + .as_deref() + .is_some_and(|current_name| object.name.as_str() != current_name) + { + self.finish_current_object(); + } + + record_version_listing_entry( + bucket, + &mut self.current_object_name, + &mut self.current_object_versions, + &object.name, + object.version_id.as_ref().map(|version_id| version_id.as_bytes()), + )?; + + if object.delete_marker { + self.delete_markers = self.delete_markers.saturating_add(1); + return Ok(()); + } + + let object_size = object.size.max(0) as u64; + self.current_live_versions = self.current_live_versions.saturating_add(1); + self.size_histogram.add(object_size); + self.total_size = self.total_size.saturating_add(object_size); + self.versions_count = self.versions_count.saturating_add(1); + Ok(()) + } + + fn finish_current_object(&mut self) { + if self.current_live_versions > 0 { + self.objects_count = self.objects_count.saturating_add(1); + self.versions_histogram.add(self.current_live_versions); + } + self.current_live_versions = 0; + } + + fn finish(mut self) -> BucketUsageInfo { + self.finish_current_object(); + BucketUsageInfo { + size: self.total_size, + objects_count: self.objects_count, + versions_count: self.versions_count, + delete_markers_count: self.delete_markers, + object_size_histogram: self.size_histogram.to_map(), + object_versions_histogram: self.versions_histogram.to_map(), + ..Default::default() + } + } +} + +type UsageVersionPage = StorageListObjectVersionsInfo; + pub async fn compute_bucket_usage(store: Arc, bucket_name: &str) -> Result { + let bucket = bucket_name.to_string(); + compute_bucket_usage_with_pages(bucket_name, move |marker, version_marker| { + let store = Arc::clone(&store); + let bucket = bucket.clone(); + async move { + store + .list_object_versions(&bucket, "", marker, version_marker, None, 1000) + .await + } + }) + .await +} + +async fn compute_bucket_usage_with_pages(bucket_name: &str, mut fetch_page: F) -> Result +where + F: FnMut(Option, Option) -> Fut, + Fut: Future>, +{ let mut marker: Option = None; let mut version_marker: Option = None; - let mut object_names: HashSet = HashSet::new(); - let mut object_versions: HashMap = HashMap::new(); - let mut versions_count: u64 = 0; - let mut total_size: u64 = 0; - let mut delete_markers: u64 = 0; - let mut size_histogram = SizeHistogram::default(); + let mut usage = BucketUsageAccumulator::default(); loop { - let result = store - .clone() - .list_object_versions( - bucket_name, - "", // prefix - marker.clone(), - version_marker.clone(), - None, // delimiter - 1000, // max_keys - ) - .await?; + let result = fetch_page(marker.clone(), version_marker.clone()).await?; + let page_entries = result.objects.len(); for object in result.objects.iter() { - if object.is_dir { - continue; - } - - if object.delete_marker { - delete_markers = delete_markers.saturating_add(1); - continue; - } - - let object_size = object.size.max(0) as u64; - object_names.insert(object.name.clone()); - *object_versions.entry(object.name.clone()).or_insert(0) += 1; - size_histogram.add(object_size); - total_size = total_size.saturating_add(object_size); - versions_count = versions_count.saturating_add(1); + usage.record(bucket_name, object)?; } if !result.is_truncated { break; } + ensure_truncated_version_page_has_entries(bucket_name, page_entries)?; - marker = result.next_marker.clone(); - version_marker = result.next_version_idmarker.clone(); - if marker.is_none() { - info!( - "Bucket {} version listing marked truncated but no marker returned; stopping early", - bucket_name - ); - break; + advance_version_listing_cursor( + bucket_name, + &mut marker, + &mut version_marker, + result.next_marker, + result.next_version_idmarker, + )?; + } + + Ok(usage.finish()) +} + +fn ensure_truncated_version_page_has_entries(bucket: &str, page_entries: usize) -> Result<(), Error> { + if page_entries == 0 { + return Err(Error::other(format!("bucket {bucket} version listing returned an empty truncated page"))); + } + Ok(()) +} + +fn record_version_listing_entry( + bucket: &str, + current_object_name: &mut Option, + current_object_versions: &mut HashSet>, + object_name: &str, + version_id: Option<&[u8; 16]>, +) -> Result<(), Error> { + match current_object_name.as_deref() { + Some(current_name) if object_name < current_name => { + return Err(Error::other(format!("bucket {bucket} version listing returned an out-of-order object"))); } + Some(current_name) if object_name > current_name => { + current_object_versions.clear(); + *current_object_name = Some(object_name.to_string()); + } + None => *current_object_name = Some(object_name.to_string()), + Some(_) => {} } - let objects_count = object_names.len() as u64; - let mut versions_histogram = VersionsHistogram::default(); - for version_count in object_versions.values() { - versions_histogram.add(*version_count); + if !current_object_versions.insert(version_id.copied()) { + return Err(Error::other(format!( + "bucket {bucket} version listing returned a repeated object version" + ))); } + Ok(()) +} - let usage = BucketUsageInfo { - size: total_size, - objects_count, - versions_count, - delete_markers_count: delete_markers, - object_size_histogram: size_histogram.to_map(), - object_versions_histogram: versions_histogram.to_map(), - ..Default::default() - }; +fn advance_version_listing_cursor( + bucket: &str, + marker: &mut Option, + version_marker: &mut Option, + next_marker: Option, + next_version_marker: Option, +) -> Result<(), Error> { + let next_marker = next_marker + .filter(|next_marker| !next_marker.is_empty()) + .ok_or_else(|| Error::other(format!("bucket {bucket} version listing was truncated without a key marker")))?; + let next_version_marker = next_version_marker + .filter(|next_version_marker| !next_version_marker.is_empty()) + .ok_or_else(|| Error::other(format!("bucket {bucket} version listing was truncated without a version marker")))?; + let current_key_marker = marker.as_deref().map(list_marker_key); + let next_key_marker = list_marker_key(&next_marker); + if current_key_marker == Some(next_key_marker) && version_marker.as_deref() == Some(next_version_marker.as_str()) { + return Err(Error::other(format!( + "bucket {bucket} version listing returned a repeated continuation marker" + ))); + } + if current_key_marker.is_some_and(|marker| next_key_marker < marker) { + return Err(Error::other(format!("bucket {bucket} version listing returned a regressing key marker"))); + } + *marker = Some(next_marker); + *version_marker = Some(next_version_marker); + Ok(()) +} - Ok(usage) +async fn coalesce_live_bucket_usage(bucket: String, init: F) -> Result +where + F: Future> + Send + 'static, +{ + let result = live_bucket_usage_cache().try_get_with(bucket.clone(), init).await; + live_bucket_usage_cache().invalidate(&bucket).await; + result.map_err(|err| Error::other(err.to_string())) +} + +fn apply_live_bucket_usage_to_response(data_usage_info: &mut DataUsageInfo, bucket: &str, usage: &BucketUsageInfo) { + data_usage_info.bucket_sizes.insert(bucket.to_string(), usage.size); + data_usage_info.buckets_usage.insert(bucket.to_string(), usage.clone()); + set_buckets_count_from_usage(data_usage_info); + data_usage_info.calculate_totals(); } pub async fn refresh_bucket_usage_from_object_layer( @@ -522,13 +660,12 @@ pub async fn refresh_bucket_usage_from_object_layer( data_usage_info: &mut DataUsageInfo, bucket: &str, ) -> Result { - let refresh_started_at = SystemTime::now(); - let usage = compute_bucket_usage(store, bucket).await?; - replace_bucket_usage_memory_from_authoritative(bucket, usage.clone(), refresh_started_at).await; - data_usage_info.bucket_sizes.insert(bucket.to_string(), usage.size); - data_usage_info.buckets_usage.insert(bucket.to_string(), usage.clone()); - set_buckets_count_from_usage(data_usage_info); - data_usage_info.calculate_totals(); + let bucket_name = bucket.to_string(); + let usage = + coalesce_live_bucket_usage(bucket_name.clone(), async move { compute_bucket_usage(store, &bucket_name).await }).await?; + // Request-time listings are not linearizable with writes on other nodes. + // Keep the live result response-local instead of promoting it into the quota cache. + apply_live_bucket_usage_to_response(data_usage_info, bucket, &usage); Ok(usage) } @@ -546,45 +683,26 @@ pub async fn refresh_versioned_bucket_usage_from_object_layer(store: Arc>(); + buckets.extend(listed_bucket_names.into_iter().filter(|bucket| !bucket.is_empty())); + let mut buckets = buckets.into_iter().collect::>(); + buckets.sort(); for bucket in buckets { let Ok(versioning) = BucketVersioningSys::get(&bucket).await else { continue; }; - if !versioning.enabled() && !versioning.suspended() { continue; } - if let Err(err) = refresh_bucket_usage_from_object_layer(store.clone(), data_usage_info, &bucket).await { debug!( bucket = %bucket, error = %err, "failed to refresh versioned bucket usage from object layer" ); - continue; } - changed = true; } - - if changed { - set_buckets_count_from_usage(data_usage_info); - data_usage_info.calculate_totals(); - } -} - -fn bucket_names_for_versioned_refresh( - data_usage_info: &DataUsageInfo, - listed_bucket_names: impl IntoIterator, -) -> Vec { - let mut buckets = data_usage_info.buckets_usage.keys().cloned().collect::>(); - buckets.extend(listed_bucket_names.into_iter().filter(|bucket| !bucket.is_empty())); - - let mut buckets = buckets.into_iter().collect::>(); - buckets.sort(); - buckets } async fn ensure_bucket_usage_cached(bucket: &str) { @@ -629,6 +747,7 @@ fn bucket_usage_counts_match(left: &BucketUsageInfo, right: &BucketUsageInfo) -> && left.delete_markers_count == right.delete_markers_count } +#[cfg(test)] async fn replace_bucket_usage_memory_from_authoritative(bucket: &str, usage: BucketUsageInfo, refresh_started_at: SystemTime) { let mut cache = memory_cache().write().await; if let Some(existing) = cache.get(bucket) @@ -1253,6 +1372,216 @@ mod tests { info } + #[tokio::test] + async fn compute_usage_preserves_same_object_across_1000_entry_page_boundary() { + let first_page = UsageVersionPage { + is_truncated: true, + next_marker: Some("object-a".to_string()), + next_version_idmarker: Some(uuid::Uuid::from_u128(1000).to_string()), + objects: (1..=1000_u128) + .map(|version| ObjectInfo { + name: "object-a".to_string(), + size: 1, + version_id: Some(uuid::Uuid::from_u128(version)), + ..Default::default() + }) + .collect(), + ..Default::default() + }; + let second_page = UsageVersionPage { + objects: vec![ + ObjectInfo { + name: "object-a".to_string(), + size: 1, + version_id: Some(uuid::Uuid::from_u128(1001)), + ..Default::default() + }, + ObjectInfo { + name: "object-a".to_string(), + version_id: Some(uuid::Uuid::from_u128(1002)), + delete_marker: true, + ..Default::default() + }, + ObjectInfo { + name: "object-b".to_string(), + size: 2, + version_id: Some(uuid::Uuid::from_u128(1003)), + ..Default::default() + }, + ], + ..Default::default() + }; + let pages = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([first_page, second_page]))); + let fetch_pages = Arc::clone(&pages); + + let usage = compute_bucket_usage_with_pages("bucket-a", move |marker, version_marker| { + let page = fetch_pages + .lock() + .expect("page queue lock should not be poisoned") + .pop_front() + .expect("pagination must not request an unexpected page"); + if page.is_truncated { + assert_eq!((marker, version_marker), (None, None)); + } else { + let expected_version_marker = uuid::Uuid::from_u128(1000).to_string(); + assert_eq!(marker.as_deref(), Some("object-a")); + assert_eq!(version_marker.as_deref(), Some(expected_version_marker.as_str())); + } + async move { Ok(page) } + }) + .await + .expect("two-page usage aggregation should succeed"); + + assert!(pages.lock().expect("page queue lock should not be poisoned").is_empty()); + assert_eq!(usage.objects_count, 2); + assert_eq!(usage.versions_count, 1002); + assert_eq!(usage.delete_markers_count, 1); + assert_eq!(usage.size, 1003); + assert_eq!(usage.object_versions_histogram.get("SINGLE_VERSION"), Some(&1)); + assert_eq!(usage.object_versions_histogram.get("BETWEEN_1000_AND_10000"), Some(&1)); + } + + #[tokio::test] + #[serial] + async fn live_bucket_usage_refreshes_are_coalesced_only_while_in_flight() { + const BUCKET: &str = "coalesced-live-usage-test"; + live_bucket_usage_cache().invalidate(BUCKET).await; + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut tasks = Vec::new(); + + for _ in 0..8 { + let calls = Arc::clone(&calls); + tasks.push(tokio::spawn(coalesce_live_bucket_usage(BUCKET.to_string(), async move { + calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(25)).await; + Ok(BucketUsageInfo { + objects_count: 7, + size: 42, + ..Default::default() + }) + }))); + } + + for task in tasks { + let usage = task + .await + .expect("coalesced refresh task should not panic") + .expect("coalesced refresh should succeed"); + assert_eq!((usage.objects_count, usage.size), (7, 42)); + } + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + + let calls_after_batch = Arc::clone(&calls); + coalesce_live_bucket_usage(BUCKET.to_string(), async move { + calls_after_batch.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(BucketUsageInfo::default()) + }) + .await + .expect("a later refresh should run after the in-flight entry is removed"); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + live_bucket_usage_cache().invalidate(BUCKET).await; + } + + #[tokio::test] + #[serial] + async fn live_usage_updates_response_without_replacing_quota_memory() { + clear_usage_memory_cache_for_test().await; + let persisted = data_usage_info_for_test("bucket-a", 100, 1_000, SystemTime::now()); + replace_bucket_usage_memory_from_info(&persisted).await; + let mut response = persisted.clone(); + + apply_live_bucket_usage_to_response(&mut response, "bucket-a", &BucketUsageInfo::default()); + + assert_eq!(response.buckets_usage.get("bucket-a").map(|usage| usage.objects_count), Some(0)); + assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(1_000)); + } + + #[test] + fn version_listing_cursor_rejects_incomplete_or_non_advancing_pages() { + let mut marker = Some("object-a".to_string()); + let mut version_marker = Some("version-a".to_string()); + + assert!( + advance_version_listing_cursor("bucket-a", &mut marker, &mut version_marker, None, Some("version-b".to_string()),) + .is_err() + ); + assert!( + advance_version_listing_cursor("bucket-a", &mut marker, &mut version_marker, Some("object-a".to_string()), None,) + .is_err() + ); + assert!( + advance_version_listing_cursor( + "bucket-a", + &mut marker, + &mut version_marker, + Some("object-a".to_string()), + Some("version-a".to_string()), + ) + .is_err() + ); + assert!(ensure_truncated_version_page_has_entries("bucket-a", 0).is_err()); + ensure_truncated_version_page_has_entries("bucket-a", 1).expect("a truncated page with an entry can advance"); + + let mut tagged_marker = Some("object-a[rustfs_cache:v2,id:old]".to_string()); + let mut tagged_version_marker = Some("version-a".to_string()); + assert!( + advance_version_listing_cursor( + "bucket-a", + &mut tagged_marker, + &mut tagged_version_marker, + Some("object-a[rustfs_cache:v2,id:new]".to_string()), + Some("version-a".to_string()), + ) + .is_err(), + "different cache tags for the same logical marker must not count as progress" + ); + } + + #[test] + fn version_listing_rejects_replayed_and_out_of_order_entries() { + let version_a = uuid::Uuid::from_u128(1); + let version_b = uuid::Uuid::from_u128(2); + let mut current_object = None; + let mut current_versions = HashSet::new(); + + record_version_listing_entry( + "bucket-a", + &mut current_object, + &mut current_versions, + "object-b", + Some(version_a.as_bytes()), + ) + .expect("first version should be accepted"); + assert!( + record_version_listing_entry( + "bucket-a", + &mut current_object, + &mut current_versions, + "object-b", + Some(version_a.as_bytes()), + ) + .is_err() + ); + record_version_listing_entry( + "bucket-a", + &mut current_object, + &mut current_versions, + "object-c", + Some(version_b.as_bytes()), + ) + .expect("a lexicographically later object should reset version history"); + assert!( + record_version_listing_entry( + "bucket-a", + &mut current_object, + &mut current_versions, + "object-a", + Some(version_a.as_bytes()), + ) + .is_err() + ); + } + fn aggregate_for_test( inputs: Vec<(DiskUsageStatus, Result, Error>)>, ) -> (Vec, DataUsageInfo) { @@ -1574,24 +1903,6 @@ mod tests { ); } - #[test] - fn versioned_refresh_bucket_names_include_live_buckets_without_usage_snapshot() { - let mut persisted = DataUsageInfo::default(); - persisted.buckets_usage.insert( - "bucket-a".to_string(), - BucketUsageInfo { - objects_count: 1, - versions_count: 1, - size: 10, - ..Default::default() - }, - ); - - let buckets = bucket_names_for_versioned_refresh(&persisted, vec!["bucket-b".to_string(), "bucket-a".to_string()]); - - assert_eq!(buckets, vec!["bucket-a".to_string(), "bucket-b".to_string()]); - } - #[tokio::test] #[serial] async fn authoritative_versioned_refresh_replaces_stale_dirty_memory() { diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 5a7b71d77..e2edaa7db 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -283,6 +283,16 @@ pub struct DiskHealthTracker { pub last_capacity_probe_unix_secs: AtomicI64, } +pub(crate) struct DiskHealthWaitingGuard<'a> { + health: &'a DiskHealthTracker, +} + +impl Drop for DiskHealthWaitingGuard<'_> { + fn drop(&mut self) { + self.health.decrement_waiting(); + } +} + impl DiskHealthTracker { /// Create a new disk health tracker pub fn new() -> Self { @@ -537,6 +547,11 @@ impl DiskHealthTracker { self.waiting.fetch_add(1, Ordering::Relaxed); } + pub(crate) fn waiting_guard(&self) -> DiskHealthWaitingGuard<'_> { + self.increment_waiting(); + DiskHealthWaitingGuard { health: self } + } + /// Decrement waiting operations counter pub fn decrement_waiting(&self) { self.waiting.fetch_sub(1, Ordering::Relaxed); @@ -1038,11 +1053,10 @@ impl LocalDiskWrapper { .unwrap() .as_nanos() as i64; self.health.last_started.store(now, Ordering::Relaxed); - self.health.increment_waiting(); + let _waiting_guard = self.health.waiting_guard(); if timeout_duration == Duration::ZERO { let result = operation().await; - self.health.decrement_waiting(); if result.is_ok() { self.health.record_operation_success(&self.endpoint(), "operation_success"); } @@ -1053,16 +1067,14 @@ impl LocalDiskWrapper { match result { Ok(operation_result) => { - // Log success and decrement waiting counter + // Log success; the waiting guard balances every exit path. if operation_result.is_ok() { self.health.record_operation_success(&self.endpoint(), "operation_success"); } - self.health.decrement_waiting(); operation_result } Err(_) => { - // Timeout occurred, mark disk as potentially faulty and decrement waiting counter - self.health.decrement_waiting(); + // Timeout occurred, mark disk as potentially faulty. if timeout_health_action == TimeoutHealthAction::MarkFailure && self.health.mark_failure(&self.endpoint(), "operation_timeout") { @@ -1468,6 +1480,43 @@ mod tests { struct PendingWriter; + #[test] + fn disk_health_waiting_guard_balances_cancellation() { + let health = DiskHealthTracker::new(); + { + let _guard = health.waiting_guard(); + assert_eq!(health.waiting_count(), 1); + } + assert_eq!(health.waiting_count(), 0); + } + + #[tokio::test] + async fn local_disk_health_wrapper_balances_task_cancellation() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let wrapper = Arc::new(LocalDiskWrapper::new(disk, false)); + let task_wrapper = Arc::clone(&wrapper); + let task = tokio::spawn(async move { + task_wrapper + .track_disk_health(|| async { std::future::pending::>().await }, Duration::ZERO) + .await + }); + + tokio::time::timeout(Duration::from_secs(1), async { + while wrapper.health.waiting_count() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("operation should enter disk health tracking"); + task.abort(); + let _ = task.await; + + assert_eq!(wrapper.health.waiting_count(), 0); + } + impl AsyncWrite for PendingWriter { fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { Poll::Pending diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 92ac8a399..ae2c09353 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -892,6 +892,25 @@ mod tests { assert_eq!(versions[1].version_id, Some(last_version)); } + #[test] + fn versions_after_marker_preserves_stale_marker_compatibility() { + let existing_version = + Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("existing version UUID should parse"); + let deleted_marker = Uuid::parse_str("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee").expect("delete marker UUID should parse"); + let file_infos = rustfs_filemeta::FileInfoVersions { + versions: vec![FileInfo { + version_id: Some(existing_version), + ..Default::default() + }], + ..Default::default() + }; + + let versions = versions_after_marker(&file_infos, VersionMarker::Version(deleted_marker)); + + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].version_id, Some(existing_version)); + } + #[tokio::test] async fn versions_listing_applies_version_marker_only_to_first_entry() { let metadata = rustfs_filemeta::test_data::create_real_xlmeta().expect("test metadata should be valid"); diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index b85e4a68b..cc3d77f03 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -21,6 +21,12 @@ pub(crate) mod heal { pub(crate) use rustfs_storage_api::HealOperations; } +pub(crate) mod internode { + pub(crate) use rustfs_storage_api::{ + WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, + }; +} + pub(crate) mod lifecycle { pub use rustfs_storage_api::{ExpirationOptions, TransitionedObject}; } diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 2cb87afc1..aa41d0b72 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -271,6 +271,7 @@ pub struct ListPathOptions { const MARKER_TAG_VERSION: &str = "v2"; const LEGACY_MARKER_TAG_VERSIONS: &[&str] = &["v1", MARKER_TAG_VERSION]; +const LIST_CACHE_MARKER_PREFIX: &str = "[rustfs_cache:"; const LIST_CURSOR_SOURCE_WALKER: &str = "walker"; const LIST_CURSOR_SOURCE_INDEX_KEY_ONLY: &str = "index_key_only"; const LIST_CURSOR_SOURCE_INDEX_VERIFIED_PAGE: &str = "index_verified_page"; @@ -1831,7 +1832,7 @@ impl ListContinuationV2 { fn encode_marker(&self, marker: &str) -> String { let mut marker_tag = String::with_capacity(marker.len() + 64); marker_tag.push_str(marker); - marker_tag.push_str("[rustfs_cache:"); + marker_tag.push_str(LIST_CACHE_MARKER_PREFIX); marker_tag.push_str(self.version); match &self.id { Some(id) => { @@ -1928,6 +1929,13 @@ impl ListContinuationV2 { } } +pub(crate) fn list_marker_key(marker: &str) -> &str { + marker + .rfind(LIST_CACHE_MARKER_PREFIX) + .filter(|start_idx| marker[*start_idx..].ends_with(']')) + .map_or(marker, |start_idx| &marker[..start_idx]) +} + fn normalize_list_quorum(value: &str) -> &'static str { let value = value.trim(); if value.eq_ignore_ascii_case("disk") { @@ -3085,7 +3093,7 @@ impl ListPathOptions { let Some(marker) = self.marker.clone() else { return; }; - let Some(start_idx) = marker.rfind("[rustfs_cache:") else { + let Some(start_idx) = marker.rfind(LIST_CACHE_MARKER_PREFIX) else { return; }; let Some(end_offset) = marker[start_idx..].rfind(']') else { diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index b8e3ef201..d25ad13eb 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -156,7 +156,7 @@ impl std::fmt::Display for InternodeHttpRequestContext { } } -#[derive(Debug, thiserror::Error)] +#[derive(thiserror::Error)] #[error("{kind}: {context}")] pub struct InternodeHttpError { kind: InternodeHttpErrorKind, @@ -165,6 +165,16 @@ pub struct InternodeHttpError { source: Option>, } +impl std::fmt::Debug for InternodeHttpError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("InternodeHttpError") + .field("kind", &self.kind) + .field("context", &self.context) + .field("source_present", &self.source.is_some()) + .finish() + } +} + impl InternodeHttpError { pub fn kind(&self) -> InternodeHttpErrorKind { self.kind @@ -605,11 +615,8 @@ async fn get_http_client(url: &str) -> io::Result { fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext { let target = reqwest::Url::parse(url) .ok() - .map(|parsed| match parsed.query() { - Some(query) => format!("{}?{query}", parsed.path()), - None => parsed.path().to_string(), - }) - .unwrap_or_else(|| url.to_string()); + .map(|parsed| parsed.path().to_string()) + .unwrap_or_else(|| "".to_string()); InternodeHttpRequestContext { method: method.to_string(), target, @@ -723,6 +730,12 @@ fn internode_reqwest_error(method: &Method, url: &str, operation: Option<&'stati InternodeHttpError::with_source(classified, context, err).into_io_error() } +fn internode_reqwest_body_error(method: &Method, url: &str, operation: Option<&'static str>, err: reqwest::Error) -> io::Error { + let context = internode_request_context(method, url, operation); + let classified = classify_transport_error(&err, err.is_timeout(), err.is_connect(), true); + InternodeHttpError::with_source(classified, context, err).into_io_error() +} + fn internode_classified_error( method: &Method, url: &str, @@ -823,8 +836,9 @@ impl HttpReader { let stream_error_method = method.clone(); let stream = resp.bytes_stream().map_err(move |e| { record_internode_error(track_internode_metrics, internode_operation); - record_internode_classified_error(track_internode_metrics, internode_operation, classify_reqwest_error(&e)); - internode_reqwest_error(&stream_error_method, &stream_error_url, internode_operation, e) + let classified = classify_transport_error(&e, e.is_timeout(), e.is_connect(), true); + record_internode_classified_error(track_internode_metrics, internode_operation, classified); + internode_reqwest_body_error(&stream_error_method, &stream_error_url, internode_operation, e) }); Ok(Self { @@ -1653,6 +1667,16 @@ mod tests { (StatusCode::OK, Body::from_stream(body_stream)) } + async fn get_failing_stream(State(state): State) -> impl IntoResponse { + state.get_count.fetch_add(1, Ordering::SeqCst); + let body_stream = + stream::once(async { Ok::(Bytes::from_static(b"partial")) }).chain(stream::once(async { + tokio::time::sleep(Duration::from_millis(25)).await; + Err(io::Error::other("stream failed")) + })); + (StatusCode::OK, Body::from_stream(body_stream)) + } + async fn reject_head(State(state): State) -> impl IntoResponse { state.head_count.fetch_add(1, Ordering::SeqCst); StatusCode::METHOD_NOT_ALLOWED @@ -1684,6 +1708,7 @@ mod tests { .route("/reject-put", get(get_stream).put(reject_put)) .route("/stall", get(get_stalling_stream)) .route("/delayed-first", get(get_delayed_first_chunk)) + .route("/fail-after-partial", get(get_failing_stream)) .with_state(state); let handle = tokio::spawn(async move { @@ -1918,6 +1943,37 @@ mod tests { assert!(source.context().target().contains("/stream")); } + #[tokio::test] + async fn http_reader_surfaces_body_error_after_partial_data() { + let state = TestState::default(); + let Some((base_url, handle)) = start_test_server(state).await else { + return; + }; + let url = base_url.replace("/stream", "/fail-after-partial"); + let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None) + .await + .expect("reader should accept the successful response headers"); + let mut partial = [0_u8; 7]; + + reader + .read_exact(&mut partial) + .await + .expect("partial response bytes should arrive before the terminal error"); + let err = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("terminal body errors must not become clean EOF"); + + assert_eq!(&partial, b"partial"); + let source = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("body error should retain internode classification"); + assert_eq!(source.kind(), InternodeHttpErrorKind::BodyStreamAborted); + + handle.abort(); + } + #[test] fn classify_http_status_marks_retryable_gateway_errors() { let unavailable = classify_http_status(reqwest::StatusCode::SERVICE_UNAVAILABLE); @@ -1961,7 +2017,36 @@ mod tests { InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE) ); assert_eq!(source.context().method(), "PUT"); - assert!(source.context().target().contains(PUT_FILE_STREAM_PATH)); + assert_eq!(source.context().target(), PUT_FILE_STREAM_PATH); + assert!(!err.to_string().contains("disk-a")); + } + + #[test] + fn internode_request_context_redacts_malformed_targets() { + let context = + internode_request_context(&Method::GET, "not a url?disk=/sensitive/path", Some(INTERNODE_OPERATION_READ_FILE_STREAM)); + + assert_eq!(context.target(), ""); + assert!(!context.to_string().contains("sensitive")); + } + + #[test] + fn internode_http_error_debug_redacts_source_details() { + let context = InternodeHttpRequestContext { + method: "GET".to_string(), + target: WALK_DIR_PATH.to_string(), + operation: Some(INTERNODE_OPERATION_WALK_DIR), + }; + let error = InternodeHttpError::with_source( + InternodeHttpErrorKind::BodyStreamAborted, + context, + io::Error::other("http://node/rustfs/rpc/walk_dir?disk=/sensitive/path&token=private"), + ); + let debug = format!("{error:?}"); + + assert!(debug.contains("source_present: true")); + assert!(!debug.contains("sensitive")); + assert!(!debug.contains("private")); } #[test] diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index deffaf1cf..e54d36d86 100644 --- a/crates/storage-api/src/lib.rs +++ b/crates/storage-api/src/lib.rs @@ -14,6 +14,10 @@ //! Storage API contracts for RustFS. +pub const WALK_DIR_STREAM_COMPLETION_QUERY: &str = "walk_dir_stream_completion"; +pub const WALK_DIR_STREAM_COMPLETION_V1: &str = "error-v1"; +pub const WALK_DIR_BODY_SHA256_QUERY: &str = "walk_dir_body_sha256"; + pub mod admin; pub mod bucket; pub mod capability; diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index bd6eebf08..4ebeda47d 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -12,6 +12,8 @@ for later deletion. ## Open Items +- `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability. + ## Review Checklist Before completing a PR that adds wrappers, re-exports, fallbacks, legacy action diff --git a/rustfs/src/admin/handlers/account_info.rs b/rustfs/src/admin/handlers/account_info.rs index 62adc7b54..d0d4e2a95 100644 --- a/rustfs/src/admin/handlers/account_info.rs +++ b/rustfs/src/admin/handlers/account_info.rs @@ -20,7 +20,7 @@ use crate::admin::storage_api::contract::admin::StorageAdminApi; use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions}; use crate::admin::storage_api::data_usage::{ apply_bucket_usage_memory_overlay, load_data_usage_from_backend, refresh_bucket_usage_from_object_layer, - refresh_versioned_bucket_usage_from_object_layer, replace_bucket_usage_memory_from_info, + replace_bucket_usage_memory_from_info, }; use crate::admin::storage_api::metadata_sys; use crate::auth::get_condition_values; @@ -258,7 +258,6 @@ impl Operation for AccountInfoHandler { let mut data_usage_info = load_data_usage_from_backend(store.clone()) .await .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?; - refresh_versioned_bucket_usage_from_object_layer(store.clone(), &mut data_usage_info).await; replace_bucket_usage_memory_from_info(&data_usage_info).await; apply_bucket_usage_memory_overlay(&mut data_usage_info).await; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index e30fa6219..0d7c8ebb3 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -476,14 +476,6 @@ pub(crate) mod data_usage { crate::storage::storage_api::ecstore_data_usage::load_data_usage_from_backend(store).await } - pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer( - store: Arc, - data_usage_info: &mut rustfs_data_usage::DataUsageInfo, - ) { - crate::storage::storage_api::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info) - .await; - } - pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) { crate::storage::storage_api::ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await; } diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 8549f4c4e..001cc5240 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -22,7 +22,7 @@ use super::storage_api::admin_usecase::contract::StorageAdminApi; use super::storage_api::admin_usecase::contract::bucket::{BucketOperations, BucketOptions}; use super::storage_api::admin_usecase::data_usage::{ apply_bucket_usage_memory_overlay, load_data_usage_from_backend, refresh_bucket_usage_from_object_layer, - refresh_versioned_bucket_usage_from_object_layer, replace_bucket_usage_memory_from_info, + replace_bucket_usage_memory_from_info, }; use super::storage_api::admin_usecase::{ECStore, EndpointServerPools}; use crate::app::runtime_sources::{ @@ -249,7 +249,6 @@ impl DefaultAdminUsecase { error!("load_data_usage_from_backend failed {:?}", e); Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed") })?; - refresh_versioned_bucket_usage_from_object_layer(store.clone(), &mut info).await; replace_bucket_usage_memory_from_info(&info).await; apply_bucket_usage_memory_overlay(&mut info).await; Self::refresh_live_bucket_usage_for_data_usage_info(store.clone(), &mut info).await; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index ea61c2ef6..7a2fdc376 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -70,14 +70,6 @@ pub(crate) mod data_usage { .await } - pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer( - store: Arc, - data_usage_info: &mut rustfs_data_usage::DataUsageInfo, - ) { - crate::storage::storage_api::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info) - .await; - } - pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) { crate::storage::storage_api::ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await; } diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 11a2dcbb2..dd0cfc887 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -14,12 +14,15 @@ use crate::server::RPC_PREFIX; use crate::storage::request_context::spawn_traced; +#[cfg(test)] +use crate::storage::storage_api::rpc_consumer::http_service::WALK_DIR_BODY_SHA256_QUERY; use crate::storage::storage_api::rpc_consumer::http_service::{ - DEFAULT_READ_BUFFER_SIZE, StorageDiskRpcExt as _, WalkDirOptions, find_local_disk_by_ref, verify_rpc_signature, + DEFAULT_READ_BUFFER_SIZE, StorageDiskRpcExt as _, WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, find_local_disk_by_ref, + verify_rpc_signature, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use bytes::{Bytes, BytesMut}; -use futures_util::TryStreamExt; +use futures_util::{Stream, StreamExt, TryStreamExt, stream}; use http::{HeaderMap, Method, Request, Response, StatusCode, Uri}; use http_body_util::{BodyExt, Limited}; use hyper::body::Incoming; @@ -33,11 +36,13 @@ use s3s::Body; use s3s::dto::StreamingBlob; use serde::de::DeserializeOwned; use serde_urlencoded::from_bytes; +use sha2::{Digest, Sha256}; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::Instant; use tokio::io::{self, AsyncWriteExt}; +use tokio::sync::oneshot; use tokio_util::io::ReaderStream; use tower::Service; use tracing::{error, warn}; @@ -238,6 +243,32 @@ struct ReadFileQuery { #[derive(Debug, Default, serde::Deserialize)] struct WalkDirQuery { disk: String, + walk_dir_stream_completion: Option, + walk_dir_body_sha256: Option, +} + +fn supports_walk_dir_stream_completion(query: &WalkDirQuery) -> bool { + query.walk_dir_stream_completion.as_deref() == Some(WALK_DIR_STREAM_COMPLETION_V1) +} + +fn verify_walk_dir_body_digest(query: &WalkDirQuery, body: &[u8]) -> bool { + if !supports_walk_dir_stream_completion(query) { + return true; + } + + let Some(expected) = query.walk_dir_body_sha256.as_deref() else { + return false; + }; + let actual = hex_simd::encode_to_string(Sha256::digest(body), hex_simd::AsciiCase::Lower); + expected == actual +} + +fn validate_walk_dir_completion_request(query: &WalkDirQuery, body: &[u8]) -> Option { + let propagate_completion_errors = supports_walk_dir_stream_completion(query); + if !verify_walk_dir_body_digest(query, body) { + return None; + } + Some(propagate_completion_errors) } #[derive(Debug, Default, serde::Deserialize)] @@ -450,7 +481,6 @@ async fn handle_walk_dir(req: Request) -> Response { Ok(query) => query, Err(response) => return *response, }; - let Some(disk) = find_local_disk_by_ref(&query.disk).await else { warn!( event = EVENT_RPC_REQUEST_REJECTED, @@ -485,6 +515,26 @@ async fn handle_walk_dir(req: Request) -> Response { return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, message); } }; + // RUSTFS_COMPAT_TODO(#4648): old clients retry terminal stream failures on an already-used writer. + // Remove after every supported peer version advertises walk-dir stream completion v1. + let propagate_completion_errors = match validate_walk_dir_completion_request(&query, &body) { + Some(propagate_completion_errors) => propagate_completion_errors, + None => { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK, + operation = INTERNODE_OPERATION_WALK_DIR, + result = "rejected", + status_code = StatusCode::FORBIDDEN.as_u16(), + rpc_path = WALK_DIR_PATH, + method = %Method::GET, + reason = "request_body_digest_mismatch", + "internode rpc request rejected" + ); + return response_with_status(StatusCode::FORBIDDEN, "invalid request body digest"); + } + }; let args: WalkDirOptions = match serde_json::from_slice(&body) { Ok(args) => args, @@ -518,9 +568,8 @@ async fn handle_walk_dir(req: Request) -> Response { let log_limit = args.limit; let log_disk_id = args.disk_id.clone(); let log_skip_total_timeout = args.skip_total_timeout; - let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); - spawn_traced(async move { - if let Err(e) = disk.walk_dir(args, &mut wd).await { + let body = walk_dir_response_body(propagate_completion_errors, move |mut writer| async move { + disk.walk_dir(args, &mut writer).await.map_err(|e| { warn!( event = EVENT_RPC_BACKGROUND_TASK_FAILED, component = LOG_COMPONENT_INTERNODE_RPC, @@ -540,13 +589,38 @@ async fn handle_walk_dir(req: Request) -> Response { error = %e, "internode rpc background task failed" ); - } + io::Error::other("remote walk_dir failed") + }) }); runtime_sources::current_internode_metrics() .record_incoming_request_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP); + + Response::builder() + .status(StatusCode::OK) + .body(body) + .expect("failed to build walk dir response") +} + +fn walk_dir_response_body(propagate_completion_errors: bool, producer: F) -> Body +where + F: FnOnce(tokio::io::DuplexStream) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, +{ + let (reader, writer) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); + let (mut completion_tx, completion_rx) = oneshot::channel(); + spawn_traced(async move { + tokio::select! { + biased; + result = producer(writer) => { + let _ = completion_tx.send(result); + } + _ = completion_tx.closed() => {} + } + }); + let metrics = runtime_sources::current_internode_metrics(); - let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { + let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { metrics.record_sent_bytes_for_operation_and_backend( INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, @@ -554,11 +628,32 @@ async fn handle_walk_dir(req: Request) -> Response { ); bytes }); + let stream = append_walk_dir_completion(stream, completion_rx, propagate_completion_errors); - Response::builder() - .status(StatusCode::OK) - .body(Body::from(StreamingBlob::wrap(stream))) - .expect("failed to build walk dir response") + Body::from(StreamingBlob::wrap(stream)) +} + +fn append_walk_dir_completion( + stream: S, + completion_rx: oneshot::Receiver>, + propagate_completion_errors: bool, +) -> impl Stream> +where + S: Stream>, +{ + stream.chain( + stream::once(async move { + match completion_rx.await { + Ok(Ok(())) => None, + Ok(Err(err)) if propagate_completion_errors => Some(Err(err)), + Err(err) if propagate_completion_errors => { + Some(Err(io::Error::other(format!("remote walk_dir task ended without a result: {err}")))) + } + Ok(Err(_)) | Err(_) => None, + } + }) + .filter_map(std::future::ready), + ) } async fn handle_put_file(req: Request) -> Response { @@ -734,20 +829,34 @@ fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std 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_body_size_mismatch, put_file_stage_error_message, read_file_body_stream, verify_internode_rpc_signature, - write_body_chunks_to_writer, + READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, append_walk_dir_completion, + internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, put_body_size_mismatch, + put_file_stage_error_message, read_file_body_stream, supports_walk_dir_stream_completion, + validate_walk_dir_completion_request, verify_internode_rpc_signature, verify_walk_dir_body_digest, + walk_dir_response_body, write_body_chunks_to_writer, }; use bytes::Bytes; use http::{HeaderMap, Method, StatusCode, Uri}; + use http_body_util::BodyExt; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, }; + use sha2::Digest as _; use tokio::io; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_stream::StreamExt; use tokio_stream::iter; + struct DropNotifier(Option>); + + impl Drop for DropNotifier { + fn drop(&mut self) { + if let Some(sender) = self.0.take() { + let _ = sender.send(()); + } + } + } + #[test] fn internode_rpc_path_matches_rpc_prefix() { assert!(is_internode_rpc_path("/rustfs/rpc/read_file_stream")); @@ -852,6 +961,130 @@ mod tests { assert_eq!(out, b"hello world"); } + #[tokio::test] + async fn walk_dir_body_surfaces_background_failure_after_data() { + let body = walk_dir_response_body(true, |mut writer| async move { + writer.write_all(b"partial walk data").await?; + Err(io::Error::other("remote walk_dir failed")) + }); + let err = BodyExt::collect(body) + .await + .expect_err("failed completion must fail body collection"); + + assert!(err.to_string().contains("remote walk_dir failed")); + } + + #[tokio::test] + async fn walk_dir_body_preserves_data_after_success() { + let body = walk_dir_response_body(true, |mut writer| async move { + writer.write_all(b"complete walk data").await?; + Ok(()) + }); + let bytes = BodyExt::collect(body) + .await + .expect("successful completion should preserve the body") + .to_bytes(); + + assert_eq!(bytes, Bytes::from_static(b"complete walk data")); + } + + #[tokio::test] + async fn walk_dir_completion_stream_surfaces_cancelled_producer() { + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); + drop(completion_tx); + let stream = iter([Ok::(Bytes::from_static(b"partial walk data"))]); + let body = s3s::Body::from(s3s::dto::StreamingBlob::wrap(append_walk_dir_completion(stream, completion_rx, true))); + + let err = BodyExt::collect(body) + .await + .expect_err("a cancelled producer must fail body collection"); + + assert!(err.to_string().contains("ended without a result")); + } + + #[tokio::test] + async fn dropping_walk_dir_body_cancels_blocked_producer() { + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel(); + let body = walk_dir_response_body(true, move |_writer| async move { + let _drop_notifier = DropNotifier(Some(dropped_tx)); + let _ = started_tx.send(()); + std::future::pending::>().await + }); + + started_rx.await.expect("walk producer should start"); + drop(body); + + tokio::time::timeout(std::time::Duration::from_secs(1), dropped_rx) + .await + .expect("dropping the response body should cancel the walk producer") + .expect("drop notifier should send a cancellation signal"); + } + + #[tokio::test] + async fn legacy_walk_dir_client_keeps_clean_eof_compatibility() { + let body = walk_dir_response_body(false, |mut writer| async move { + writer.write_all(b"legacy partial data").await?; + Err(io::Error::other("remote walk_dir failed")) + }); + + let bytes = BodyExt::collect(body) + .await + .expect("legacy clients must retain clean EOF until they advertise stream completion support") + .to_bytes(); + + assert_eq!(bytes, Bytes::from_static(b"legacy partial data")); + } + + #[tokio::test] + async fn legacy_walk_dir_client_keeps_clean_eof_when_producer_is_cancelled() { + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); + drop(completion_tx); + let stream = iter([Ok::(Bytes::from_static(b"legacy partial data"))]); + let body = s3s::Body::from(s3s::dto::StreamingBlob::wrap(append_walk_dir_completion(stream, completion_rx, false))); + + let bytes = BodyExt::collect(body) + .await + .expect("legacy clients must retain clean EOF after producer cancellation") + .to_bytes(); + + assert_eq!(bytes, Bytes::from_static(b"legacy partial data")); + } + + #[test] + fn walk_dir_completion_requires_the_exact_signed_query_capability() { + let legacy: WalkDirQuery = serde_urlencoded::from_str("disk=disk-a").expect("legacy query should parse"); + let unknown: WalkDirQuery = + serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v2").expect("unknown query should parse"); + let capable: WalkDirQuery = + serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v1").expect("capable query should parse"); + + assert!(!supports_walk_dir_stream_completion(&legacy)); + assert!(!supports_walk_dir_stream_completion(&unknown)); + assert!(supports_walk_dir_stream_completion(&capable)); + } + + #[test] + fn walk_dir_completion_requires_matching_signed_body_digest() { + let body = br#"{"bucket":"bucket-a"}"#; + let digest = hex_simd::encode_to_string(sha2::Sha256::digest(body), hex_simd::AsciiCase::Lower); + let capable: WalkDirQuery = serde_urlencoded::from_str(&format!( + "disk=disk-a&walk_dir_stream_completion=error-v1&{WALK_DIR_BODY_SHA256_QUERY}={digest}" + )) + .expect("capable query should parse"); + let missing: WalkDirQuery = + serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v1").expect("query should parse"); + let legacy: WalkDirQuery = serde_urlencoded::from_str("disk=disk-a").expect("legacy query should parse"); + + assert!(verify_walk_dir_body_digest(&capable, body)); + assert!(!verify_walk_dir_body_digest(&capable, b"tampered")); + assert!(!verify_walk_dir_body_digest(&missing, body)); + assert!(verify_walk_dir_body_digest(&legacy, b"legacy body")); + assert_eq!(validate_walk_dir_completion_request(&capable, body), Some(true)); + assert_eq!(validate_walk_dir_completion_request(&legacy, b"legacy body"), Some(false)); + assert_eq!(validate_walk_dir_completion_request(&capable, b"tampered"), None); + } + #[tokio::test] async fn read_file_body_stream_keeps_full_stream_when_length_is_zero() { let (reader, mut writer) = tokio::io::duplex(64); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index cc4befc45..788bf8615 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -188,6 +188,9 @@ pub(crate) mod rpc_consumer { pub(crate) mod http_service { pub(crate) const DEFAULT_READ_BUFFER_SIZE: usize = super::super::DEFAULT_READ_BUFFER_SIZE; + #[cfg(test)] + pub(crate) use super::super::storage_contracts::WALK_DIR_BODY_SHA256_QUERY; + pub(crate) use super::super::storage_contracts::WALK_DIR_STREAM_COMPLETION_V1; pub(crate) use super::super::{StorageDiskRpcExt, WalkDirOptions, find_local_disk_by_ref, verify_rpc_signature}; } @@ -361,8 +364,8 @@ pub(crate) mod ecstore_data_usage { apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend, record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory, - refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer, - remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend, + refresh_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, + store_compression_total_in_backend, }; }