diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index dfce7dbf3..003960bac 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -15,6 +15,7 @@ use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout}; use crate::disk::error::DiskError; use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions}; +use futures::future::join_all; use metrics::counter; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof}; use std::{ @@ -655,6 +656,7 @@ async fn list_path_raw_inner( errs.push(None); } let mut pending_entries: Vec> = vec![None; readers.len()]; + let mut peek_outcomes: Vec> = std::iter::repeat_with(|| None).take(readers.len()).collect(); loop { let mut current = MetaCacheEntry::default(); @@ -676,6 +678,21 @@ async fn list_path_raw_inner( let mut has_err = 0; let mut agree = 0; + // Start every missing head read in the same round so one stalled + // disk cannot multiply the wait budget by the erasure-set width. + // Outcomes are still consumed below in stable disk-index order. + let concurrent_peeks = readers.iter_mut().enumerate().filter_map(|(i, reader)| { + if errs[i].is_some() || pending_entries[i].is_some() { + return None; + } + + let cancel = &revjob_rx; + Some(async move { (i, peek_with_timeout(cancel, reader, peek_timeout).await) }) + }); + for (i, outcome) in join_all(concurrent_peeks).await { + peek_outcomes[i] = Some(outcome); + } + for (i, r) in readers.iter_mut().enumerate() { if errs[i].is_some() { has_err += 1; @@ -685,7 +702,10 @@ async fn list_path_raw_inner( let entry = if let Some(entry) = pending_entries[i].take() { entry } else { - match peek_with_timeout(&revjob_rx, r, peek_timeout).await { + let Some(outcome) = peek_outcomes[i].take() else { + return Err(DiskError::Unexpected); + }; + match outcome { PeekOutcome::Ready(res) => { if let Some(entry) = res { // info!("read entry disk: {}, name: {}", i, entry.name); @@ -1295,6 +1315,36 @@ mod tests { assert_eq!(err, DiskError::Timeout); } + #[tokio::test(start_paused = true)] + async fn list_path_raw_bounds_multiple_stalled_readers_by_one_peek_deadline() { + let peek_timeout = Duration::from_millis(20); + let started = tokio::time::Instant::now(); + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None], + min_disks: 1, + test_reader_behaviors: vec![ + TestReaderBehavior::Stall, + TestReaderBehavior::Stall, + TestReaderBehavior::Stall, + TestReaderBehavior::Stall, + ], + peek_timeout: Some(peek_timeout), + ..Default::default() + }, + ) + .await + .expect_err("all stalled readers should fail the listing"); + + assert_eq!(err, DiskError::Timeout); + assert_eq!( + started.elapsed(), + peek_timeout, + "reader deadlines must overlap instead of accumulating once per disk" + ); + } + #[tokio::test] async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() { let entry = MetaCacheEntry { diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index dd8bfa17f..d00c7ed49 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -40,7 +40,7 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::LazyLock; -use tracing::{Span, error, info, warn}; +use tracing::{error, info, warn}; const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; const LOG_SUBSYSTEM_AUDIT_TARGET: &str = "audit_target"; @@ -278,8 +278,6 @@ pub struct AuditTargetConfig {} #[async_trait::async_trait] impl Operation for AuditTargetConfig { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); let (target_type, target_name) = extract_target_params(¶ms)?; authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; @@ -345,8 +343,6 @@ pub struct ListAuditTargets {} #[async_trait::async_trait] impl Operation for ListAuditTargets { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?; let mut runtime_statuses = HashMap::new(); @@ -370,8 +366,6 @@ pub struct RemoveAuditTarget {} #[async_trait::async_trait] impl Operation for RemoveAuditTarget { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); let (target_type, target_name) = extract_target_params(¶ms)?; authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; @@ -838,6 +832,13 @@ mod tests { extract_block_between_markers(src, "impl Operation for ListAuditTargets", "pub struct RemoveAuditTarget"); let delete_block = extract_block_between_markers(src, "impl Operation for RemoveAuditTarget", "#[cfg(test)]"); + for block in [put_block, list_block, delete_block] { + assert!( + !block.contains(".enter()"), + "async audit handlers must rely on request-future instrumentation instead of holding span guards across awaits" + ); + } + assert!( put_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "audit target writes should require SetBucketTargetAction" diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index f4a1ba2d1..a820e1fc0 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -43,7 +43,7 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::LazyLock; -use tracing::{Span, error, info, warn}; +use tracing::{error, info, warn}; const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; @@ -333,8 +333,6 @@ pub struct NotificationTarget {} #[async_trait::async_trait] impl Operation for NotificationTarget { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); let (target_type, target_name) = extract_target_params(¶ms)?; let context = app_context_from_req(&req); @@ -401,8 +399,6 @@ pub struct ListNotificationTargets {} #[async_trait::async_trait] impl Operation for ListNotificationTargets { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?; refresh_persisted_module_switches_from_store().await.map_err(|err| { warn!( @@ -439,8 +435,6 @@ pub struct ListTargetsArns {} #[async_trait::async_trait] impl Operation for ListTargetsArns { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?; if let Some(reason) = notification_target_operation_block_reason( "querying notification target ARNs for bucket associations from the console", @@ -485,8 +479,6 @@ pub struct RemoveNotificationTarget {} #[async_trait::async_trait] impl Operation for RemoveNotificationTarget { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { - let span = Span::current(); - let _enter = span.enter(); let (target_type, target_name) = extract_target_params(¶ms)?; let context = app_context_from_req(&req); @@ -1006,6 +998,13 @@ mod tests { extract_block_between_markers(src, "impl Operation for ListTargetsArns", "pub struct RemoveNotificationTarget"); let delete_block = extract_block_between_markers(src, "impl Operation for RemoveNotificationTarget", "fn extract_param"); + for block in [put_block, list_block, arns_block, delete_block] { + assert!( + !block.contains(".enter()"), + "async notification handlers must rely on request-future instrumentation instead of holding span guards across awaits" + ); + } + assert!( put_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "notification target writes should require SetBucketTargetAction"