fix(admin): probe set drives concurrently for storage info

The admin storage walk probed a set's drives one after another, each
bounded by the disk_info timeout, so a few drives still recovering
after a power cut pushed the local snapshot past the peer probe budget
and healthy peers rendered as unknown. Probe all drives at once so the
walk costs one timeout at most, and add a test-only probe delay hook to
pin that bound.

.github/workflows/docker.yml

@@ -121,22 +121,6 @@ jobs:
           create_latest=false
           source_ref="$GITHUB_SHA"
 
-          # Pre-GA policy: until the first stable (vX.Y.Z) tag exists, every
-          # prerelease (alpha/beta/rc) also moves `latest`, so users pulling
-          # `latest` get the newest test build. Once a stable tag is published
-          # this returns false and `latest` follows stable releases only.
-          prerelease_moves_latest() {
-            local stable_tags
-            stable_tags=$(git ls-remote --tags --refs origin 2>/dev/null \
-              | awk '{print $2}' \
-              | grep -E '^refs/tags/v?[0-9]+\.[0-9]+\.[0-9]+$' || true)
-            if [[ -z "$stable_tags" ]]; then
-              return 0
-            fi
-            echo "ℹ️  Stable release tag(s) already exist; prereleases no longer update latest"
-            return 1
-          }
-
           if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
             # Triggered by build workflow completion
             echo "🔗 Triggered by build workflow completion"
@@ -200,8 +184,8 @@ jobs:
                 if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]] || [[ "$version" == *"rc"* ]]; then
                   build_type="prerelease"
                   is_prerelease=true
-                  # Pre-GA policy: prereleases update latest until the first stable tag exists.
-                  if prerelease_moves_latest; then
+                  # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
+                  if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
                     create_latest=true
                     echo "🧪 Building Docker image for prerelease: $version (creating latest tag)"
                   else
@@ -259,8 +243,8 @@ jobs:
               v*alpha*|v*beta*|v*rc*|*alpha*|*beta*|*rc*)
                 build_type="prerelease"
                 is_prerelease=true
-                # Pre-GA policy: prereleases update latest until the first stable tag exists.
-                if prerelease_moves_latest; then
+                # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
+                if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
                   create_latest=true
                   echo "🧪 Building with prerelease version: $input_version (creating latest tag)"
                 else
@@ -410,13 +394,11 @@ jobs:
           TAG_BASE="${VERSION}${VARIANT_SUFFIX}"
           TAGS="${{ env.REGISTRY_DOCKERHUB }}:$TAG_BASE,${{ env.REGISTRY_GHCR }}:$TAG_BASE,${{ env.REGISTRY_QUAY }}:$TAG_BASE"
 
-          # Add latest when requested (stable releases, and prereleases before GA)
+          # Add channel tags for prereleases and latest for stable
           if [[ "$CREATE_LATEST" == "true" ]]; then
+            # Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
             TAGS="$TAGS,${{ env.REGISTRY_DOCKERHUB }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_GHCR }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_QUAY }}:latest${VARIANT_SUFFIX}"
-          fi
-
-          # Always add the channel tag for prereleases, independent of latest
-          if [[ "$BUILD_TYPE" == "prerelease" ]]; then
+          elif [[ "$BUILD_TYPE" == "prerelease" ]]; then
             # Prerelease channel tags (alpha, beta, rc)
             if [[ "$VERSION" == *"alpha"* ]]; then
               CHANNEL="alpha"
@@ -573,7 +555,7 @@ jobs:
             "prerelease")
               echo "🧪 Prerelease Docker image has been built with ${VERSION} tags"
               echo "⚠️  This is a prerelease image - use with caution"
-              # Prereleases move latest until the first stable tag exists (pre-GA policy).
+              # Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
               if [[ "$CREATE_LATEST" == "true" ]]; then
                 echo "🏷️  Latest tag has been created for prerelease: $VERSION"
               else

Cargo.lock

@@ -6139,9 +6139,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
 
 [[package]]
 name = "libflate"
-version = "2.3.2"
+version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
+checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
 dependencies = [
  "adler32",
  "crc32fast",
@@ -10943,9 +10943,9 @@ dependencies = [
 
 [[package]]
 name = "rustfs-uring"
-version = "0.2.2"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
+checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
 dependencies = [
  "io-uring",
  "libc",
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix",
  "windows-sys 0.59.0",

crates/config/README.md

@@ -130,37 +130,6 @@ Scanner cycle budget controls:
   - timeout returns S3 `SlowDown`, so clients should use normal SDK retry handling.
   - this is not a fdatasync or group-commit switch. Track fdatasync batching separately with `rustfs_s3_put_object_rename_fdatasync_batch_files`.
 
-## Foreground write admission environment variables
-
-Large direct `PutObject` requests and multipart `UploadPart` requests share one
-per-process permit pool that bounds how many bodies are ingested and written
-concurrently. Small direct PUTs stay on the legacy path.
-
-- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE`
-  - enables the default-on pool; `false` keeps only the soft request counter.
-  - default is `true`.
-- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT`
-  - permits in the pool; `0` derives half of `RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS`, clamped to `32`.
-  - default is `0` (32 permits at stock settings).
-- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES`
-  - smallest direct `PutObject` that takes a permit; unknown-size requests always do.
-  - default is `33554432` (32 MiB).
-- `RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS`
-  - how long a direct `PutObject` waits for a permit before returning S3 `SlowDown`.
-  - default is `250`.
-- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES`
-  - smallest `UploadPart` that takes a permit; `0` gates every part.
-  - default is `0`.
-- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS`
-  - how long an `UploadPart` waits in the bounded queue for a permit before returning S3 `SlowDown`; `0` rejects immediately when the pool is full.
-  - default is `30000`. Parts wait before body ingest, so SDK-default clients that send every part of an upload concurrently drain through the pool instead of failing.
-- `RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING`
-  - maximum `UploadPart` requests waiting for a permit at once; parts beyond it return `SlowDown` without waiting.
-  - default is `0`, which derives 16 times the permit limit (512 at stock settings).
-- `RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE`, `RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT`, `RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS`
-  - experimental strict gate that applies to every foreground write regardless of size and replaces the pool above when enabled.
-  - default is disabled; enabling it with limit `0` disables foreground write admission entirely.
-
 ## Remote tier timeout environment variables
 
 - `RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS`

crates/config/src/constants/object.rs

@@ -365,36 +365,13 @@ pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: &str =
     "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES";
 pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: usize = 0;
 
-/// Time in milliseconds an automatic foreground direct PutObject waits for a permit.
+/// Time in milliseconds an automatic foreground write waits for a permit.
 ///
 /// A short wait smooths transient bursts while still returning S3
 /// `SlowDown`/503 before body ingest when the node is already saturated.
 pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
 pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 250;
 
-/// Time in milliseconds a multipart UploadPart waits for a foreground write permit.
-///
-/// SDK-default multipart clients send every part of an upload concurrently, so
-/// a single node routinely sees several times more parts in flight than the
-/// permit pool allows. Those parts have not ingested a body yet, so queueing
-/// them costs a connection rather than memory or internode streams; the pool
-/// still bounds the number of parts being written. The wait is long enough for
-/// an ordinary queue to drain on modest hardware, and a part that cannot get a
-/// permit within it fails with S3 `SlowDown`/503 for the client to retry.
-/// `0` rejects immediately when the pool is full.
-pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str =
-    "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
-pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 30_000;
-
-/// Maximum multipart UploadPart requests waiting for a foreground write permit per process.
-///
-/// Parts beyond this queue depth are rejected with S3 `SlowDown`/503 without
-/// waiting, so a genuinely saturated node still fails fast instead of holding
-/// an unbounded set of connections open for the whole wait timeout.
-/// `0` derives the depth from the permit limit.
-pub const ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING: &str = "RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING";
-pub const DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING: usize = 0;
-
 const _: () = assert!(DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE);
 
 /// Environment variable for minimum GetObject timeout in seconds.

crates/ecstore/Cargo.toml

@@ -226,7 +226,7 @@ metrics = { workspace = true }
 # crates.io. The guard scripts/check_no_tokio_io_uring.sh allows an explicit
 # io-uring integration; only the tokio "io-uring" runtime feature is banned.
 [target.'cfg(target_os = "linux")'.dependencies]
-rustfs-uring = "0.2.2"
+rustfs-uring = "0.2.1"
 
 [target.'cfg(windows)'.dependencies]
 winapi-util.workspace = true

crates/ecstore/src/disk/disk_store.rs

@@ -195,6 +195,13 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile {
     DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default)
 }
 
+#[cfg(test)]
+tokio::task_local! {
+    /// Artificial `disk_info` latency for tests that pin how the admin storage
+    /// walk composes per-drive probe time.
+    pub(crate) static DISK_INFO_PROBE_DELAY_FOR_TEST: Duration;
+}
+
 fn get_drive_timeout_profile() -> DriveTimeoutProfile {
     #[cfg(test)]
     {
@@ -2036,6 +2043,10 @@ impl DiskAPI for LocalDiskWrapper {
             .track_disk_health_with_op_and_timeout_action(
                 "disk_info",
                 || async {
+                    #[cfg(test)]
+                    if let Ok(delay) = DISK_INFO_PROBE_DELAY_FOR_TEST.try_with(|delay| *delay) {
+                        tokio::time::sleep(delay).await;
+                    }
                     let result = self.disk.disk_info(opts).await?;
 
                     if let Some(current_disk_id) = *self.disk_id.read().await

crates/ecstore/src/set_disk/mod.rs

@@ -6463,113 +6463,114 @@ pub fn should_heal_object_on_disk(
     (false, false, None)
 }
 
+/// Probe every drive of the set at once. Each live probe is bounded by the
+/// drive `disk_info` timeout, and the admin peer probe budget only covers one
+/// such timeout; a sequential walk over several stalled drives after a power
+/// cut would exceed it and make healthy peers render as unknown (#6488).
 async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<rustfs_madmin::Disk> {
-    let mut ret = Vec::new();
+    join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
+}
 
-    for (i, pool) in disks.iter().enumerate() {
-        if let Some(disk) = pool {
-            let runtime_state = disk.runtime_state();
-            let offline_duration_seconds = disk.offline_duration_secs();
-            let capacity_snapshot = disk.last_capacity_snapshot();
-            let cached_disk_id = disk.cached_disk_id().await;
-            if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
-                match disk
-                    .disk_info(&DiskInfoOptions {
-                        metrics: true,
-                        ..Default::default()
-                    })
-                    .await
-                {
-                    Ok(res) => {
-                        disk.record_capacity_probe(res.total, res.used, res.free);
-                        ret.push(rustfs_madmin::Disk {
-                            endpoint: eps[i].to_string(),
-                            local: eps[i].is_local,
-                            pool_index: eps[i].pool_idx,
-                            set_index: eps[i].set_idx,
-                            disk_index: eps[i].disk_idx,
-                            state: "ok".to_owned(),
+async fn disk_admin_info(disk: Option<&DiskStore>, ep: &Endpoint) -> rustfs_madmin::Disk {
+    let Some(disk) = disk else {
+        return rustfs_madmin::Disk {
+            endpoint: ep.to_string(),
+            drive_path: ep.get_file_path(),
+            local: ep.is_local,
+            pool_index: ep.pool_idx,
+            set_index: ep.set_idx,
+            disk_index: ep.disk_idx,
+            runtime_state: None,
+            offline_duration_seconds: None,
+            state: DiskError::DiskNotFound.to_string(),
+            capacity_observation_source: Some("missing".to_owned()),
+            capacity_observation_age_seconds: Some(0),
+            ..Default::default()
+        };
+    };
 
-                            root_disk: res.root_disk,
-                            drive_path: res.mount_path.clone(),
-                            healing: res.healing,
-                            scanning: res.scanning,
-                            runtime_state: Some(runtime_state.as_str().to_string()),
-                            offline_duration_seconds,
-                            capacity_observation_source: Some("live_probe".to_owned()),
-                            capacity_observation_age_seconds: Some(0),
-
-                            uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
-                            major: res.major as u32,
-                            minor: res.minor as u32,
-                            model: None,
-                            total_space: res.total,
-                            used_space: res.used,
-                            available_space: res.free,
-                            physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
-                            utilization: utilization_percent(res.total, res.used),
-                            used_inodes: res.used_inodes,
-                            free_inodes: res.free_inodes,
-                            metrics: Some(res.metrics),
-                            ..Default::default()
-                        });
-                    }
-                    Err(err) => {
-                        let mut disk_info = rustfs_madmin::Disk {
-                            state: err.to_string(),
-                            endpoint: eps[i].to_string(),
-                            drive_path: eps[i].get_file_path(),
-                            local: eps[i].is_local,
-                            pool_index: eps[i].pool_idx,
-                            set_index: eps[i].set_idx,
-                            disk_index: eps[i].disk_idx,
-                            runtime_state: Some(runtime_state.as_str().to_string()),
-                            offline_duration_seconds,
-                            metrics: disk.metrics_snapshot(),
-                            uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
-                            ..Default::default()
-                        };
-                        if let Some((total, used, free, _)) = capacity_snapshot {
-                            disk_info.total_space = total;
-                            disk_info.used_space = used;
-                            disk_info.available_space = free;
-                            disk_info.utilization = utilization_percent(total, used);
-                            disk_info.capacity_observation_source = Some("snapshot".to_owned());
-                            disk_info.capacity_observation_age_seconds = capacity_snapshot
-                                .map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
-                        } else {
-                            disk_info.capacity_observation_source = Some("missing".to_owned());
-                            disk_info.capacity_observation_age_seconds = Some(0);
-                        }
-                        ret.push(disk_info);
-                    }
-                }
-            } else {
-                let mut disk_info =
-                    build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds, capacity_snapshot);
-                disk_info.metrics = disk.metrics_snapshot();
-                disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
-                ret.push(disk_info);
-            }
-        } else {
-            ret.push(rustfs_madmin::Disk {
-                endpoint: eps[i].to_string(),
-                drive_path: eps[i].get_file_path(),
-                local: eps[i].is_local,
-                pool_index: eps[i].pool_idx,
-                set_index: eps[i].set_idx,
-                disk_index: eps[i].disk_idx,
-                runtime_state: None,
-                offline_duration_seconds: None,
-                state: DiskError::DiskNotFound.to_string(),
-                capacity_observation_source: Some("missing".to_owned()),
-                capacity_observation_age_seconds: Some(0),
-                ..Default::default()
-            })
-        }
+    let runtime_state = disk.runtime_state();
+    let offline_duration_seconds = disk.offline_duration_secs();
+    let capacity_snapshot = disk.last_capacity_snapshot();
+    let cached_disk_id = disk.cached_disk_id().await;
+    if !(runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect) {
+        let mut disk_info = build_runtime_snapshot_disk(ep, runtime_state, offline_duration_seconds, capacity_snapshot);
+        disk_info.metrics = disk.metrics_snapshot();
+        disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
+        return disk_info;
     }
 
-    ret
+    match disk
+        .disk_info(&DiskInfoOptions {
+            metrics: true,
+            ..Default::default()
+        })
+        .await
+    {
+        Ok(res) => {
+            disk.record_capacity_probe(res.total, res.used, res.free);
+            rustfs_madmin::Disk {
+                endpoint: ep.to_string(),
+                local: ep.is_local,
+                pool_index: ep.pool_idx,
+                set_index: ep.set_idx,
+                disk_index: ep.disk_idx,
+                state: "ok".to_owned(),
+
+                root_disk: res.root_disk,
+                drive_path: res.mount_path.clone(),
+                healing: res.healing,
+                scanning: res.scanning,
+                runtime_state: Some(runtime_state.as_str().to_string()),
+                offline_duration_seconds,
+                capacity_observation_source: Some("live_probe".to_owned()),
+                capacity_observation_age_seconds: Some(0),
+
+                uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
+                major: res.major as u32,
+                minor: res.minor as u32,
+                model: None,
+                total_space: res.total,
+                used_space: res.used,
+                available_space: res.free,
+                physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
+                utilization: utilization_percent(res.total, res.used),
+                used_inodes: res.used_inodes,
+                free_inodes: res.free_inodes,
+                metrics: Some(res.metrics),
+                ..Default::default()
+            }
+        }
+        Err(err) => {
+            let mut disk_info = rustfs_madmin::Disk {
+                state: err.to_string(),
+                endpoint: ep.to_string(),
+                drive_path: ep.get_file_path(),
+                local: ep.is_local,
+                pool_index: ep.pool_idx,
+                set_index: ep.set_idx,
+                disk_index: ep.disk_idx,
+                runtime_state: Some(runtime_state.as_str().to_string()),
+                offline_duration_seconds,
+                metrics: disk.metrics_snapshot(),
+                uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
+                ..Default::default()
+            };
+            if let Some((total, used, free, _)) = capacity_snapshot {
+                disk_info.total_space = total;
+                disk_info.used_space = used;
+                disk_info.available_space = free;
+                disk_info.utilization = utilization_percent(total, used);
+                disk_info.capacity_observation_source = Some("snapshot".to_owned());
+                disk_info.capacity_observation_age_seconds =
+                    capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
+            } else {
+                disk_info.capacity_observation_source = Some("missing".to_owned());
+                disk_info.capacity_observation_age_seconds = Some(0);
+            }
+            disk_info
+        }
+    }
 }
 
 fn build_runtime_snapshot_disk(
@@ -10691,6 +10692,41 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn test_get_disks_info_probes_drives_concurrently() {
+        use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST;
+
+        let format = FormatV3::new(1, 4);
+        let mut temp_dirs = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+        for disk_idx in 0..4 {
+            let (dir, endpoint, disk) = make_formatted_local_disk_for_info_test(disk_idx, &format).await;
+            temp_dirs.push(dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+
+        let probe_delay = std::time::Duration::from_secs(2);
+        let started = tokio::time::Instant::now();
+        let info = DISK_INFO_PROBE_DELAY_FOR_TEST
+            .scope(probe_delay, get_disks_info(&disks, &endpoints))
+            .await;
+        let elapsed = started.elapsed();
+
+        assert_eq!(info.len(), 4);
+        assert!(info.iter().all(|disk| disk.state == "ok"), "every drive should still report a live probe");
+        assert_eq!(
+            info.iter().map(|disk| disk.disk_index).collect::<Vec<_>>(),
+            endpoints.iter().map(|ep| ep.disk_idx).collect::<Vec<_>>(),
+            "concurrent probes must keep endpoint order"
+        );
+        assert!(
+            elapsed < probe_delay * 2,
+            "four stalled drives must cost one probe delay, not four; took {elapsed:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() {
         let (endpoint, disk) = make_remote_disk_for_info_test(0).await;

docs/architecture/workload-admission-contracts.md

@@ -12,7 +12,7 @@
 | Class | Provider (`impl WorkloadAdmissionSnapshotProvider`) | `active` / `queued` / `limit` source | Reports `Unknown` when |
 |---|---|---|---|
 | `ForegroundRead` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | disk-read permits in use / `None` (the semaphore exposes no waiter count) / configured max concurrent disk reads | the storage registry has no entry |
-| `ForegroundWrite` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | foreground-write permits in use or legacy active-write counter / multipart parts waiting in the bounded admission queue (`None` for the strict and legacy policies) / configured or derived write-admission limit | the storage registry has no entry |
+| `ForegroundWrite` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | foreground-write permits in use or legacy active-write counter / `None` / configured or derived write-admission limit | the storage registry has no entry |
 | `Metadata` | `RustFsWorkloadAdmissionSnapshotProvider` in `rustfs/src/workload_admission.rs` | `Open` once the bucket metadata runtime handle exists; no counts | bucket metadata runtime not initialized |
 | `Scanner` | same | scanner active work-unit counter / none / configured set-scan limit when nonzero | scanner runtime not initialized |
 | `Repair` | same | heal active tasks / heal queue length / `None` (limits live behind the async heal manager state) | heal manager not initialized |

rustfs/src/storage/concurrency/manager.rs

@@ -29,17 +29,12 @@ use rustfs_io_core::BytesPool;
 use rustfs_io_core::io_profile::{AccessPattern, IoPatternDetector, StorageMedia, detect_storage_media};
 use rustfs_io_metrics::bandwidth::{BandwidthMonitor, BandwidthSnapshot};
 use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics};
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock, Mutex};
 use std::time::Duration;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::debug;
 
 const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32;
-// A queued multipart part holds a connection but no body, so the queue can be
-// several times deeper than the permit pool. Sixteen uploads sending sixteen
-// parts each through one node fits inside the derived depth of 32 * 16.
-const DERIVED_MULTIPART_ADMISSION_MAX_PENDING_FACTOR: usize = 16;
 // Framed S2 alone can retain one encoded and one decoded block of roughly
 // 4 MiB each, while other codecs have their own larger windows. Four keeps
 // useful request parallelism without scaling codec memory and CPU with clients.
@@ -154,28 +149,6 @@ struct ForegroundWriteAdmissionGate {
     semaphore: Arc<Semaphore>,
     limit: usize,
     wait_timeout: Duration,
-    /// Requests currently waiting in the bounded multipart queue.
-    pending: Arc<AtomicUsize>,
-}
-
-/// Reservation of one slot in the bounded multipart wait queue; released on
-/// drop so a cancelled or timed-out waiter never leaks queue depth.
-struct PendingSlot(Arc<AtomicUsize>);
-
-impl PendingSlot {
-    fn reserve(pending: &Arc<AtomicUsize>, max_pending: usize) -> Option<Self> {
-        if pending.fetch_add(1, Ordering::AcqRel) >= max_pending {
-            pending.fetch_sub(1, Ordering::AcqRel);
-            return None;
-        }
-        Some(Self(pending.clone()))
-    }
-}
-
-impl Drop for PendingSlot {
-    fn drop(&mut self) {
-        self.0.fetch_sub(1, Ordering::AcqRel);
-    }
 }
 
 impl ForegroundWriteAdmissionGate {
@@ -184,7 +157,6 @@ impl ForegroundWriteAdmissionGate {
             semaphore: Arc::new(Semaphore::new(limit)),
             limit,
             wait_timeout,
-            pending: Arc::new(AtomicUsize::new(0)),
         }
     }
 
@@ -192,35 +164,6 @@ impl ForegroundWriteAdmissionGate {
         self.limit.saturating_sub(self.semaphore.available_permits())
     }
 
-    fn pending(&self) -> usize {
-        self.pending.load(Ordering::Acquire)
-    }
-
-    /// Admit through the same permit pool as [`Self::admit`], but let the
-    /// request wait in a bounded queue for `wait_timeout` instead of failing
-    /// on the gate's own short wait. A full queue rejects immediately.
-    async fn admit_queued(
-        &self,
-        wait_timeout: Duration,
-        max_pending: usize,
-    ) -> Result<ForegroundWriteAdmission, tokio::sync::AcquireError> {
-        match self.semaphore.clone().try_acquire_owned() {
-            Ok(permit) => return Ok(ForegroundWriteAdmission::Admitted(permit)),
-            Err(tokio::sync::TryAcquireError::Closed) => return Ok(ForegroundWriteAdmission::Rejected),
-            Err(tokio::sync::TryAcquireError::NoPermits) => {}
-        }
-        if wait_timeout.is_zero() {
-            return Ok(ForegroundWriteAdmission::Rejected);
-        }
-        let Some(_slot) = PendingSlot::reserve(&self.pending, max_pending) else {
-            return Ok(ForegroundWriteAdmission::Rejected);
-        };
-        match tokio::time::timeout(wait_timeout, self.semaphore.clone().acquire_owned()).await {
-            Ok(permit) => Ok(ForegroundWriteAdmission::Admitted(permit?)),
-            Err(_) => Ok(ForegroundWriteAdmission::Rejected),
-        }
-    }
-
     async fn admit(&self) -> Result<ForegroundWriteAdmission, tokio::sync::AcquireError> {
         if self.wait_timeout.is_zero() {
             return Ok(match self.semaphore.clone().try_acquire_owned() {
@@ -251,8 +194,6 @@ enum ForegroundWriteAdmissionPolicy {
         gate: ForegroundWriteAdmissionGate,
         put_object_min_size_bytes: usize,
         multipart_part_min_size_bytes: usize,
-        multipart_wait_timeout: Duration,
-        multipart_max_pending: usize,
     },
 }
 
@@ -311,24 +252,11 @@ impl ForegroundWriteAdmissionPolicy {
             rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
             rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
         ));
-        let multipart_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
-            rustfs_config::ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-            rustfs_config::DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-        ));
-        let multipart_max_pending = derive_multipart_admission_max_pending(
-            rustfs_utils::get_env_usize(
-                rustfs_config::ENV_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING,
-                rustfs_config::DEFAULT_PUT_MULTIPART_FOREGROUND_ADMISSION_MAX_PENDING,
-            ),
-            large_limit,
-        );
 
         Self::Large {
             gate: ForegroundWriteAdmissionGate::new(large_limit, wait_timeout),
             put_object_min_size_bytes,
             multipart_part_min_size_bytes,
-            multipart_wait_timeout,
-            multipart_max_pending,
         }
     }
 
@@ -347,25 +275,11 @@ impl ForegroundWriteAdmissionPolicy {
 
     #[cfg(test)]
     fn large_for_test(enabled: bool, limit: usize, min_size_bytes: usize, wait_timeout: Duration) -> Self {
-        Self::large_with_multipart_queue_for_test(enabled, limit, min_size_bytes, wait_timeout, wait_timeout, 0)
-    }
-
-    #[cfg(test)]
-    fn large_with_multipart_queue_for_test(
-        enabled: bool,
-        limit: usize,
-        min_size_bytes: usize,
-        wait_timeout: Duration,
-        multipart_wait_timeout: Duration,
-        multipart_max_pending: usize,
-    ) -> Self {
         if enabled && limit > 0 {
             Self::Large {
                 gate: ForegroundWriteAdmissionGate::new(limit, wait_timeout),
                 put_object_min_size_bytes: min_size_bytes,
                 multipart_part_min_size_bytes: 0,
-                multipart_wait_timeout,
-                multipart_max_pending: derive_multipart_admission_max_pending(multipart_max_pending, limit),
             }
         } else {
             Self::LegacyCounterOnly
@@ -384,45 +298,35 @@ impl ForegroundWriteAdmissionPolicy {
                 gate,
                 put_object_min_size_bytes,
                 multipart_part_min_size_bytes,
-                multipart_wait_timeout,
-                multipart_max_pending,
-            } => match kind {
-                ForegroundWriteAdmissionKind::PutObject if should_gate_foreground_write(size, *put_object_min_size_bytes) => {
+            } => {
+                let min_size_bytes = match kind {
+                    ForegroundWriteAdmissionKind::PutObject => *put_object_min_size_bytes,
+                    ForegroundWriteAdmissionKind::MultipartPart => *multipart_part_min_size_bytes,
+                };
+                if should_gate_foreground_write(size, min_size_bytes) {
                     gate.admit().await
+                } else {
+                    Ok(ForegroundWriteAdmission::Disabled)
                 }
-                ForegroundWriteAdmissionKind::MultipartPart
-                    if should_gate_foreground_write(size, *multipart_part_min_size_bytes) =>
-                {
-                    gate.admit_queued(*multipart_wait_timeout, *multipart_max_pending).await
-                }
-                _ => Ok(ForegroundWriteAdmission::Disabled),
-            },
+            }
         }
     }
 
     fn snapshot(&self, legacy_limit: usize) -> WorkloadAdmissionSnapshot {
         match self {
-            Self::Disabled => put_admission_snapshot(0, None, 0, None),
-            Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), None, legacy_limit, None),
+            Self::Disabled => put_admission_snapshot(0, 0, None),
+            Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), legacy_limit, None),
             Self::Strict(gate) => {
-                put_admission_snapshot(gate.active(), None, gate.limit, Some("foreground write admission permits exhausted"))
+                put_admission_snapshot(gate.active(), gate.limit, Some("foreground write admission permits exhausted"))
+            }
+            Self::Large { gate, .. } => {
+                put_admission_snapshot(gate.active(), gate.limit, Some("large foreground write admission permits exhausted"))
             }
-            Self::Large { gate, .. } => put_admission_snapshot(
-                gate.active(),
-                Some(gate.pending()),
-                gate.limit,
-                Some("large foreground write admission permits exhausted"),
-            ),
         }
     }
 }
 
-fn put_admission_snapshot(
-    active: usize,
-    queued: Option<usize>,
-    limit: usize,
-    hard_gate_reason: Option<&'static str>,
-) -> WorkloadAdmissionSnapshot {
+fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<&'static str>) -> WorkloadAdmissionSnapshot {
     let state = if limit == 0 {
         AdmissionState::Disabled
     } else if active >= limit {
@@ -432,7 +336,7 @@ fn put_admission_snapshot(
     };
 
     let admission =
-        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), queued, Some(limit));
+        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));
 
     match state {
         AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
@@ -443,13 +347,6 @@ fn put_admission_snapshot(
     }
 }
 
-fn derive_multipart_admission_max_pending(configured_max_pending: usize, limit: usize) -> usize {
-    if configured_max_pending > 0 {
-        return configured_max_pending;
-    }
-    limit.saturating_mul(DERIVED_MULTIPART_ADMISSION_MAX_PENDING_FACTOR)
-}
-
 fn derive_large_put_admission_limit(configured_limit: usize, max_disk_reads: usize) -> usize {
     if configured_limit > 0 {
         return configured_limit;
@@ -567,24 +464,6 @@ impl ConcurrencyManager {
         manager
     }
 
-    #[cfg(test)]
-    pub(crate) fn with_multipart_admission_queue_for_test(
-        limit: usize,
-        multipart_wait_timeout: Duration,
-        multipart_max_pending: usize,
-    ) -> Self {
-        let mut manager = Self::new();
-        manager.foreground_write_admission_policy = ForegroundWriteAdmissionPolicy::large_with_multipart_queue_for_test(
-            true,
-            limit,
-            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
-            Duration::ZERO,
-            multipart_wait_timeout,
-            multipart_max_pending,
-        );
-        manager
-    }
-
     #[cfg(test)]
     pub(crate) fn with_large_put_admission_for_test(
         enabled: bool,
@@ -1229,7 +1108,7 @@ mod integration_tests {
     use super::super::request_guard::GetObjectGuard;
     use super::{
         ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_ARCHIVE_DECODER_LIMIT, SNOWBALL_MEMBER_COMMIT_LIMIT,
-        SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit, derive_multipart_admission_max_pending,
+        SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit,
     };
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
@@ -1645,110 +1524,6 @@ mod integration_tests {
         drop((first, second));
     }
 
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_concurrency_manager_multipart_part_waits_for_released_permit() {
-        let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 0);
-        let held = manager
-            .admit_multipart_part(8 * 1024 * 1024)
-            .await
-            .expect("first multipart part admission should acquire");
-        assert!(matches!(held, ForegroundWriteAdmission::Admitted(_)));
-
-        let waiter_manager = manager.clone();
-        let waiter = tokio::spawn(async move { waiter_manager.admit_multipart_part(8 * 1024 * 1024).await });
-        tokio::task::yield_now().await;
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(1));
-
-        tokio::time::advance(Duration::from_secs(5)).await;
-        drop(held);
-
-        let admission = waiter
-            .await
-            .expect("multipart admission waiter task must not panic")
-            .expect("multipart admission gate must stay open");
-        assert!(matches!(admission, ForegroundWriteAdmission::Admitted(_)));
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(0));
-    }
-
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_concurrency_manager_multipart_part_rejects_after_queue_wait_timeout() {
-        let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 0);
-        let held = manager
-            .admit_multipart_part(1024)
-            .await
-            .expect("first multipart part admission should acquire");
-
-        let waiter_manager = manager.clone();
-        let waiter = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await });
-        tokio::task::yield_now().await;
-        tokio::time::advance(Duration::from_secs(30)).await;
-
-        let admission = waiter
-            .await
-            .expect("multipart admission waiter task must not panic")
-            .expect("multipart admission gate must stay open");
-        assert!(matches!(admission, ForegroundWriteAdmission::Rejected));
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(0));
-        drop(held);
-    }
-
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_concurrency_manager_multipart_part_rejects_immediately_when_queue_is_full() {
-        let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 1);
-        let held = manager
-            .admit_multipart_part(1024)
-            .await
-            .expect("first multipart part admission should acquire");
-
-        let waiter_manager = manager.clone();
-        let queued = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await });
-        tokio::task::yield_now().await;
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(1));
-
-        let overflow = manager
-            .admit_multipart_part(1024)
-            .await
-            .expect("full multipart queue should reject, not close");
-        assert!(matches!(overflow, ForegroundWriteAdmission::Rejected));
-
-        drop(held);
-        let admission = queued
-            .await
-            .expect("queued multipart part task must not panic")
-            .expect("multipart admission gate must stay open");
-        assert!(matches!(admission, ForegroundWriteAdmission::Admitted(_)));
-    }
-
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_concurrency_manager_cancelled_multipart_waiter_releases_queue_slot() {
-        let manager = ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 1);
-        let held = manager
-            .admit_multipart_part(1024)
-            .await
-            .expect("first multipart part admission should acquire");
-
-        let waiter_manager = manager.clone();
-        let queued = tokio::spawn(async move { waiter_manager.admit_multipart_part(1024).await });
-        tokio::task::yield_now().await;
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(1));
-        queued.abort();
-        let _ = queued.await;
-
-        assert_eq!(manager.put_object_admission_snapshot().queued, Some(0));
-        drop(held);
-    }
-
-    #[test]
-    fn test_concurrency_manager_derives_multipart_admission_max_pending_from_limit() {
-        assert_eq!(derive_multipart_admission_max_pending(7, 32), 7);
-        assert_eq!(derive_multipart_admission_max_pending(0, 32), 512);
-        assert_eq!(derive_multipart_admission_max_pending(0, 1), 16);
-    }
-
     #[test]
     fn test_concurrency_manager_derives_large_put_admission_limit_from_scheduler_cap() {
         assert_eq!(derive_large_put_admission_limit(7, 64), 7);