fix(ecstore): overlap metacache reader deadlines (#6098)

* fix(ecstore): overlap metacache reader deadlines

* fix(admin): avoid span guards across awaits

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

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<Option<MetaCacheEntry>> = vec![None; readers.len()];
+        let mut peek_outcomes: Vec<Option<PeekOutcome>> = 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 {

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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let span = Span::current();
-        let _enter = span.enter();
         let (target_type, target_name) = extract_target_params(&params)?;
 
         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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let span = Span::current();
-        let _enter = span.enter();
         let (target_type, target_name) = extract_target_params(&params)?;
 
         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"

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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let span = Span::current();
-        let _enter = span.enter();
         let (target_type, target_name) = extract_target_params(&params)?;
         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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let span = Span::current();
-        let _enter = span.enter();
         let (target_type, target_name) = extract_target_params(&params)?;
         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"