From 78469aa63b3f5a8aed7a21641ef6d4984bbc753d Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 17 Jul 2026 08:30:39 +0800 Subject: [PATCH] fix(get): bound GET disk-read admission with a hard cap instead of unbounded permit bypass (#4935) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs https://github.com/rustfs/backlog/issues/1317 Previously, when the primary disk-read permit pool stayed saturated past `RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT` (default 5s) — which a handful of slow clients can cause because a permit is held for the whole body transfer — the GET path set `disk_permit = None` and continued reading with no admission token at all. That made the disk-read concurrency limit unbounded under exactly the overload it is meant to protect against: any number of GETs could pile onto the disks simultaneously. This replaces the permit-less bypass with a bounded overflow lane and a hard cap: - A new bounded degraded semaphore (`RUSTFS_OBJECT_DISK_DEGRADED_READ_CAP`, default mirrors the primary cap) is consulted only after the primary wait times out. A GET takes one degraded permit without blocking, or is rejected with `SlowDown`/503 once that lane is also full. The total number of GETs performing disk-active reads is therefore hard-capped at `primary_cap + degraded_cap`, and no GET ever reads without holding an admission token. - Admission is centralized in `ConcurrencyManager::admit_disk_read`, returning `Primary`/`Degraded`/`Unbounded`/`Rejected`. `Degraded` and `Rejected` are counted in metrics (`rustfs.get_object.disk_permit.degraded.total`, `rustfs.get_object.disk_permit.hard_reject.total`), replacing the removed `rustfs.get_object.disk_permit.bypass.total`. - A primary cap of `0` (disk-read throttling disabled) is preserved as the only intentional permit-less path via `Unbounded`, so that degenerate-but-served configuration is not turned into all-503. The `primary_wait == 0` wait-forever opt-out is likewise unchanged. Healthy GETs are unaffected: concurrency at or below the primary cap is admitted immediately from the primary pool exactly as before, with no new latency or rejection. The rejection is surfaced before response headers are constructed, so it is a clean pre-header 503, never a post-header 504 masquerade. Degraded-lane GETs use the identical streaming reader and forward-progress stall timeout as primary GETs, so a slow but progressing large download is not killed. Owned permits (primary or degraded) are held by `DiskReadPermitReader` and released on body EOF or client drop/cancel, so tokens are always returned. Body stall timeout already resets on forward progress and post-header failures already surface as body errors, so no timeout-split changes were needed here. Scope: this PR implements the minimal safe correctness core (eliminate unbounded bypass + hard cap + SlowDown). The two-level weighted-fair + aging scheduler (small setup cap feeding a size-aware data-producer fair queue) and the strictly-bounded producer/client buffer with cancellation propagation from the issue plan remain follow-ups. The black-box slow-client soak is deferred to the unbuilt facility in https://github.com/rustfs/backlog/issues/1325 rather than faked. Tests (white-box, virtual clock via `start_paused`): degrade-then-hard-reject with token release; 100 concurrent GETs against primary=1/degraded=1 never exceed 2 simultaneous admissions with every request either admitted or explicitly rejected; disabled cap serves unbounded; zero-wait blocks on the primary lane. Reverting the hard cap makes the concurrency invariant fail. --- crates/config/src/constants/object.rs | 29 ++- rustfs/src/app/object_usecase.rs | 65 +++--- rustfs/src/app/storage_api.rs | 4 +- rustfs/src/storage/concurrency/manager.rs | 248 ++++++++++++++++++++++ rustfs/src/storage/concurrency/mod.rs | 2 +- rustfs/src/storage/storage_api.rs | 4 +- 6 files changed, 318 insertions(+), 34 deletions(-) diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index d4527fad6..28222c824 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -73,15 +73,36 @@ pub const DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS: usize = 64; /// - Example: `export RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT=5` pub const ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT"; -/// Maximum time a GET request waits for a disk read permit (seconds). +/// Maximum time a GET request waits for a primary disk read permit (seconds). /// /// Permits are held for the whole response body transfer, so slow clients can /// occupy all of them while the disks sit idle. Instead of stalling until the -/// request-level timeout fires, a GET that waits longer than this proceeds -/// without a permit (degraded pass-through) and the bypass is counted in -/// metrics/logs. Set to 0 to wait indefinitely (previous behavior). +/// request-level timeout fires, a GET that waits longer than this falls through +/// to a bounded degraded admission lane; if that lane is also full the request +/// is rejected with `SlowDown`/503 rather than proceeding without any permit. +/// Set to 0 to wait on the primary lane indefinitely (never degrade or reject). pub const DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: u64 = 5; +/// Environment variable for the bounded degraded disk-read admission lane size. +/// - Purpose: Cap how many GETs may proceed after the primary disk-read permit +/// pool is saturated, giving a hard upper bound on concurrent disk-active +/// reads (primary cap + degraded cap) instead of an unbounded pass-through. +/// - Unit: request count (usize). `0` means "mirror the primary cap", so the +/// absolute hard cap defaults to twice the primary disk-read cap. +/// - Example: `export RUSTFS_OBJECT_DISK_DEGRADED_READ_CAP=16` +pub const ENV_OBJECT_DISK_DEGRADED_READ_CAP: &str = "RUSTFS_OBJECT_DISK_DEGRADED_READ_CAP"; + +/// Size of the bounded degraded disk-read admission lane. +/// +/// When the primary disk-read permit pool is saturated and a GET exceeds +/// [`DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT`], it may take one permit from this +/// bounded overflow lane instead of reading without any admission token. The +/// total number of GETs performing disk-active reads is therefore hard-capped at +/// `primary_cap + degraded_cap`; beyond that a GET is rejected with `SlowDown`. +/// The default `0` mirrors the primary cap, so the hard cap is twice the primary +/// disk-read concurrency. +pub const DEFAULT_OBJECT_DISK_DEGRADED_READ_CAP: usize = 0; + /// Skip bitrot hash verification on GetObject reads. /// /// When enabled, GetObject reads skip the per-shard hash diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 4d44218d4..0e55fba5a 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -53,8 +53,8 @@ use super::storage_api::object_usecase::bucket::{ }; use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; use super::storage_api::object_usecase::concurrency::{ - self, ConcurrencyManager, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, - get_put_concurrency_aware_buffer_size, + self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, + get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; #[cfg(test)] use super::storage_api::object_usecase::contract::http::HTTPPreconditions; @@ -2871,29 +2871,44 @@ impl DefaultObjectUsecase { ) -> S3Result { let permit_wait_start = std::time::Instant::now(); let permit_wait_timeout = Self::disk_permit_wait_timeout(); - // Permits are held for the whole body transfer, so slow clients can - // pin all of them while disks are idle. Bound the wait and degrade to - // a permit-less read instead of stalling into the request timeout. - let disk_permit = if permit_wait_timeout.is_zero() { - Some( - manager - .acquire_owned_disk_read_permit() - .await - .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?, - ) - } else { - match tokio::time::timeout(permit_wait_timeout, manager.acquire_owned_disk_read_permit()).await { - Ok(permit) => Some(permit.map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?), - Err(_) => { - metrics::counter!("rustfs.get_object.disk_permit.bypass.total").increment(1); - warn!( - bucket = %bucket, - key = %key, - wait_ms = permit_wait_start.elapsed().as_millis() as u64, - "GetObject proceeding without disk read permit after bounded wait" - ); - None - } + // Permits are held for the whole body transfer, so slow clients can pin + // all of them while disks are idle. Bound the wait on the primary pool + // and, on timeout, admit from a bounded degraded overflow lane. Total + // concurrent disk-active GETs are hard-capped at + // `primary_cap + degraded_cap`; once that cap is reached we reject with + // `SlowDown` instead of reading without any admission token. Never + // proceed permit-less. + let disk_permit = match manager + .admit_disk_read(permit_wait_timeout) + .await + .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))? + { + DiskReadAdmission::Primary(permit) => Some(permit), + // Throttling disabled by config (primary cap 0): proceed without an + // admission token. Not a saturation bypass. + DiskReadAdmission::Unbounded => None, + DiskReadAdmission::Degraded(permit) => { + metrics::counter!("rustfs.get_object.disk_permit.degraded.total").increment(1); + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_start.elapsed().as_millis() as u64, + "GetObject admitted into bounded degraded disk-read lane after primary pool saturation" + ); + Some(permit) + } + DiskReadAdmission::Rejected => { + metrics::counter!("rustfs.get_object.disk_permit.hard_reject.total").increment(1); + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_start.elapsed().as_millis() as u64, + "GetObject rejected: disk-read hard concurrency cap reached" + ); + return Err(s3_error!( + SlowDown, + "disk read concurrency limit reached, please reduce your request rate" + )); } }; let permit_wait_duration = permit_wait_start.elapsed(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index b43b8442b..22e581340 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -825,8 +825,8 @@ pub(crate) mod bucket { pub(crate) mod concurrency { pub(crate) use crate::storage::storage_api::concurrency_consumer::{ - ConcurrencyManager, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, get_concurrency_aware_buffer_size, - get_concurrency_manager, get_put_concurrency_aware_buffer_size, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; } diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index aabe85268..3794e76d6 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -41,6 +41,12 @@ pub(crate) static CONCURRENCY_MANAGER: LazyLock = LazyLock:: pub struct ConcurrencyManager { /// Semaphore to limit concurrent disk reads disk_read_semaphore: Arc, + /// Bounded overflow lane for GETs that time out waiting on the primary + /// disk-read permit pool. Admitting from this lane instead of reading + /// without any permit gives a hard upper bound on concurrent disk-active + /// reads (`primary cap + degraded cap`); when it is also full a GET is + /// rejected with `SlowDown` rather than proceeding unbounded. + degraded_read_semaphore: Arc, /// I/O load metrics for adaptive strategy calculation io_metrics: Arc>, /// I/O priority queue for request scheduling @@ -87,6 +93,27 @@ impl std::fmt::Debug for ConcurrencyManager { .finish() } } +/// Outcome of [`ConcurrencyManager::admit_disk_read`]. +/// +/// `Primary`/`Degraded` both carry an owned permit that must be held for the +/// full body transfer; `Rejected` means the hard concurrency cap was reached and +/// the caller must fail the GET with `SlowDown`/503 instead of reading without a +/// permit. +#[derive(Debug)] +pub enum DiskReadAdmission { + /// Admitted from the primary disk-read permit pool. + Primary(tokio::sync::OwnedSemaphorePermit), + /// Admitted from the bounded degraded overflow lane after the primary pool + /// stayed saturated past the configured wait. + Degraded(tokio::sync::OwnedSemaphorePermit), + /// Disk-read throttling is disabled (primary cap configured to `0`): the GET + /// proceeds without an admission token. This is the only permit-less path + /// and is an explicit operator opt-out, not a saturation bypass. + Unbounded, + /// Hard concurrency cap reached; the caller must reject with `SlowDown`. + Rejected, +} + #[allow(dead_code)] impl ConcurrencyManager { /// Create a new concurrency manager with default settings @@ -99,6 +126,17 @@ impl ConcurrencyManager { let max_disk_reads = scheduler_config.max_concurrent_reads; + // Bounded degraded admission lane. A configured `0` mirrors the primary + // cap, so the absolute hard cap on concurrent disk-active reads defaults + // to twice the primary disk-read concurrency. + let degraded_read_cap = { + let configured = rustfs_utils::get_env_usize( + rustfs_config::ENV_OBJECT_DISK_DEGRADED_READ_CAP, + rustfs_config::DEFAULT_OBJECT_DISK_DEGRADED_READ_CAP, + ); + if configured == 0 { max_disk_reads } else { configured } + }; + // Detect storage media let storage_media = detect_storage_media(scheduler_config.storage_detection_enabled, &scheduler_config.storage_media_override); @@ -129,6 +167,7 @@ impl ConcurrencyManager { Self { disk_read_semaphore: Arc::new(Semaphore::new(max_disk_reads)), + degraded_read_semaphore: Arc::new(Semaphore::new(degraded_read_cap)), io_metrics: Arc::new(Mutex::new(IoLoadMetrics::new(scheduler_config.load_sample_window))), priority_queue: Arc::new(IoPriorityQueue::new(queue_config)), bytes_pool: Arc::new(BytesPool::new_tiered()), @@ -140,6 +179,20 @@ impl ConcurrencyManager { } } + /// Build a manager with explicit disk-read admission caps for tests. + /// + /// Overrides only the primary/degraded semaphores and the snapshot cap so + /// admission behavior can be exercised deterministically under a paused + /// virtual clock, without depending on process-wide environment variables. + #[cfg(test)] + pub(crate) fn with_disk_read_caps_for_test(primary: usize, degraded: usize) -> Self { + let mut manager = Self::new(); + manager.disk_read_semaphore = Arc::new(Semaphore::new(primary)); + manager.degraded_read_semaphore = Arc::new(Semaphore::new(degraded)); + manager.scheduler_config.max_concurrent_reads = primary; + manager + } + /// Track a GetObject request pub fn track_request() -> GetObjectGuard { GetObjectGuard::new() @@ -177,6 +230,54 @@ impl ConcurrencyManager { self.disk_read_semaphore.clone().acquire_owned().await } + /// Admit a GET disk read under a hard concurrency cap. + /// + /// The permit is held for the whole response body transfer, so this is the + /// single admission boundary that bounds concurrent disk-active reads: + /// + /// - `primary_wait == 0` waits on the primary permit pool indefinitely and + /// always yields a [`DiskReadAdmission::Primary`] permit (no degrade, no + /// reject); this preserves the "wait forever" opt-out. + /// - Otherwise it waits up to `primary_wait` for a primary permit. On + /// timeout it takes one permit from the bounded degraded overflow lane + /// without blocking. If that lane is also full the request is + /// [`DiskReadAdmission::Rejected`] — it must surface as `SlowDown`/503 + /// rather than reading without any admission token. + /// + /// Total GETs holding a permit therefore never exceed + /// `primary_cap + degraded_cap`. + pub async fn admit_disk_read(&self, primary_wait: Duration) -> Result { + // Primary cap of 0 means disk-read throttling is disabled (matches the + // `AdmissionState::Disabled` snapshot semantics). Serve without an + // admission token rather than rejecting every GET; this preserves the + // pre-existing "permits disabled" behavior and is the only permit-less + // path. + if self.scheduler_config.max_concurrent_reads == 0 { + return Ok(DiskReadAdmission::Unbounded); + } + + if primary_wait.is_zero() { + let permit = self.disk_read_semaphore.clone().acquire_owned().await?; + return Ok(DiskReadAdmission::Primary(permit)); + } + + match tokio::time::timeout(primary_wait, self.disk_read_semaphore.clone().acquire_owned()).await { + Ok(permit) => Ok(DiskReadAdmission::Primary(permit?)), + Err(_) => { + // Primary pool saturated within the wait window. Do not read + // without a permit; fall through to the bounded degraded lane, + // and reject once the hard cap is reached. + match self.degraded_read_semaphore.clone().try_acquire_owned() { + Ok(permit) => Ok(DiskReadAdmission::Degraded(permit)), + // NoPermits => hard cap reached; Closed never happens for a + // semaphore we never close. Either way, reject rather than + // bypass admission. + Err(_) => Ok(DiskReadAdmission::Rejected), + } + } + } + } + // ============================================ // Adaptive I/O Strategy Methods // ============================================ @@ -1035,4 +1136,151 @@ mod integration_tests { assert_eq!(small_file_strategy.priority, IoPriority::High); assert_eq!(large_file_strategy.priority, IoPriority::Low); } + + // ============================================ + // Bounded disk-read admission (backlog#1317) + // ============================================ + + use super::DiskReadAdmission; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + + /// Primary pool saturation degrades into the bounded lane, and the hard cap + /// rejects with `Rejected` instead of admitting a permit-less read. Dropping + /// a permit frees the token so the next admission succeeds again. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_degrades_then_hard_rejects() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(1, 1); + + // Primary permit granted immediately. + let primary = match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Primary(permit) => permit, + other => panic!("expected primary admission, got {other:?}"), + }; + + // Primary pool saturated: next admission waits out the primary timeout + // and falls through to the bounded degraded lane. + let degraded = match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Degraded(permit) => permit, + other => panic!("expected degraded admission, got {other:?}"), + }; + + // Both lanes full: hard cap reached, must reject rather than bypass. + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Rejected => {} + other => panic!("expected hard rejection, got {other:?}"), + } + + // Releasing the degraded permit re-opens exactly one degraded slot. + drop(degraded); + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Degraded(_permit) => {} + other => panic!("expected degraded admission after release, got {other:?}"), + } + + // Releasing the primary permit re-opens the primary lane immediately + // (no timeout wait), proving EOF/drop returns the token. + drop(primary); + match manager.admit_disk_read(Duration::from_millis(100)).await.unwrap() { + DiskReadAdmission::Primary(_permit) => {} + other => panic!("expected primary admission after release, got {other:?}"), + } + } + + /// With max primary = 1 and degraded = 1, 100 concurrent GETs never hold + /// more than `1 + 1` admission tokens simultaneously. Reverting the hard cap + /// (unbounded bypass) would let this exceed 2 and fail the test. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_hard_caps_concurrent_admissions() { + let manager = std::sync::Arc::new(ConcurrencyManager::with_disk_read_caps_for_test(1, 1)); + let in_flight = std::sync::Arc::new(AtomicUsize::new(0)); + let max_in_flight = std::sync::Arc::new(AtomicUsize::new(0)); + let admitted = std::sync::Arc::new(AtomicUsize::new(0)); + let rejected = std::sync::Arc::new(AtomicUsize::new(0)); + + let mut handles = Vec::with_capacity(100); + for _ in 0..100 { + let manager = manager.clone(); + let in_flight = in_flight.clone(); + let max_in_flight = max_in_flight.clone(); + let admitted = admitted.clone(); + let rejected = rejected.clone(); + handles.push(tokio::spawn(async move { + match manager.admit_disk_read(Duration::from_millis(50)).await.unwrap() { + DiskReadAdmission::Primary(permit) | DiskReadAdmission::Degraded(permit) => { + let current = in_flight.fetch_add(1, AtomicOrdering::SeqCst) + 1; + max_in_flight.fetch_max(current, AtomicOrdering::SeqCst); + admitted.fetch_add(1, AtomicOrdering::SeqCst); + // Hold the admission token across an await point, as the + // real body transfer does, then release it. + tokio::time::sleep(Duration::from_millis(10)).await; + in_flight.fetch_sub(1, AtomicOrdering::SeqCst); + drop(permit); + } + DiskReadAdmission::Rejected => { + rejected.fetch_add(1, AtomicOrdering::SeqCst); + } + DiskReadAdmission::Unbounded => { + unreachable!("throttling is enabled (primary cap = 1) in this test"); + } + } + })); + } + + for handle in handles { + handle.await.unwrap(); + } + + // The invariant the hard cap guarantees: never more than + // primary + degraded = 2 GETs admitted at once. + assert!( + max_in_flight.load(AtomicOrdering::SeqCst) <= 2, + "max concurrent admissions {} exceeded the hard cap of 2", + max_in_flight.load(AtomicOrdering::SeqCst) + ); + // Every request is accounted for: admitted or explicitly rejected, never + // a permit-less bypass. + assert_eq!(admitted.load(AtomicOrdering::SeqCst) + rejected.load(AtomicOrdering::SeqCst), 100); + assert!( + rejected.load(AtomicOrdering::SeqCst) > 0, + "expected some hard rejections under saturation" + ); + } + + /// A primary cap of 0 (throttling disabled) serves every GET without an + /// admission token instead of rejecting them, preserving the pre-existing + /// "disk read permits disabled" behavior. + #[tokio::test(start_paused = true)] + async fn test_admit_disk_read_disabled_cap_is_unbounded() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(0, 0); + for _ in 0..10 { + match manager.admit_disk_read(Duration::from_millis(50)).await.unwrap() { + DiskReadAdmission::Unbounded => {} + other => panic!("expected unbounded admission when throttling disabled, got {other:?}"), + } + } + } + + /// `primary_wait == 0` preserves the wait-forever opt-out: it always yields + /// a primary permit and never degrades or rejects. + #[tokio::test] + async fn test_admit_disk_read_zero_wait_waits_on_primary() { + let manager = ConcurrencyManager::with_disk_read_caps_for_test(1, 1); + let held = match manager.admit_disk_read(Duration::ZERO).await.unwrap() { + DiskReadAdmission::Primary(permit) => permit, + other => panic!("expected primary admission, got {other:?}"), + }; + + // A second zero-wait admission blocks on the primary pool rather than + // degrading; it completes only once the first permit is released. + let manager2 = manager.clone(); + let waiter = tokio::spawn(async move { manager2.admit_disk_read(Duration::ZERO).await.unwrap() }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished(), "zero-wait admission must block on the primary lane"); + + drop(held); + match waiter.await.unwrap() { + DiskReadAdmission::Primary(_permit) => {} + other => panic!("expected primary admission after release, got {other:?}"), + } + } } diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index ff8e7d57c..110e14cf6 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/rustfs/src/storage/concurrency/mod.rs @@ -54,7 +54,7 @@ pub use io_schedule::{ pub use request_guard::{GetObjectGuard, PutObjectGuard}; // Concurrency manager -pub use manager::ConcurrencyManager; +pub use manager::{ConcurrencyManager, DiskReadAdmission}; // ============================================ // New Module Re-exports (for gradual migration) diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index d573af5fd..d1afb3b68 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -114,8 +114,8 @@ pub(crate) mod access_consumer { pub(crate) mod concurrency_consumer { pub(crate) use super::super::concurrency::{ - ConcurrencyManager, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, get_concurrency_aware_buffer_size, - get_concurrency_manager, get_put_concurrency_aware_buffer_size, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; }