fix(get): bound GET disk-read admission with a hard cap instead of unbounded permit bypass (#4935)

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.

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<GetObjectIoPlanning> {
         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();