fix(storage): queue multipart parts for foreground write permits (#7337)

Multipart parts shared the 250 ms direct-PutObject wait on the foreground
write permit pool, so SDK-default concurrency (many parts per upload in
flight at once) was rejected wholesale with SlowDown at stock settings.
Keep the pool that bounds in-flight bodies, but let parts wait in a
bounded queue with their own timeout before body ingest, report the
queue depth in the ForegroundWrite admission snapshot, and document the
foreground write admission environment variables.

.github/workflows/docker.yml

@@ -121,6 +121,22 @@ 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"
@@ -184,8 +200,8 @@ jobs:
                 if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]] || [[ "$version" == *"rc"* ]]; then
                   build_type="prerelease"
                   is_prerelease=true
-                  # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
-                  if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
+                  # Pre-GA policy: prereleases update latest until the first stable tag exists.
+                  if prerelease_moves_latest; then
                     create_latest=true
                     echo "🧪 Building Docker image for prerelease: $version (creating latest tag)"
                   else
@@ -243,8 +259,8 @@ jobs:
               v*alpha*|v*beta*|v*rc*|*alpha*|*beta*|*rc*)
                 build_type="prerelease"
                 is_prerelease=true
-                # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
-                if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
+                # Pre-GA policy: prereleases update latest until the first stable tag exists.
+                if prerelease_moves_latest; then
                   create_latest=true
                   echo "🧪 Building with prerelease version: $input_version (creating latest tag)"
                 else
@@ -394,11 +410,13 @@ jobs:
           TAG_BASE="${VERSION}${VARIANT_SUFFIX}"
           TAGS="${{ env.REGISTRY_DOCKERHUB }}:$TAG_BASE,${{ env.REGISTRY_GHCR }}:$TAG_BASE,${{ env.REGISTRY_QUAY }}:$TAG_BASE"
 
-          # Add channel tags for prereleases and latest for stable
+          # Add latest when requested (stable releases, and prereleases before GA)
           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}"
-          elif [[ "$BUILD_TYPE" == "prerelease" ]]; then
+          fi
+
+          # Always add the channel tag for prereleases, independent of latest
+          if [[ "$BUILD_TYPE" == "prerelease" ]]; then
             # Prerelease channel tags (alpha, beta, rc)
             if [[ "$VERSION" == *"alpha"* ]]; then
               CHANNEL="alpha"
@@ -555,7 +573,7 @@ jobs:
             "prerelease")
               echo "🧪 Prerelease Docker image has been built with ${VERSION} tags"
               echo "⚠️  This is a prerelease image - use with caution"
-              # Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
+              # Prereleases move latest until the first stable tag exists (pre-GA policy).
               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.1"
+version = "2.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
+checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
 dependencies = [
  "adler32",
  "crc32fast",
@@ -10943,9 +10943,9 @@ dependencies = [
 
 [[package]]
 name = "rustfs-uring"
-version = "0.2.1"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
+checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
 dependencies = [
  "io-uring",
  "libc",
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
  "once_cell",
  "rustix",
  "windows-sys 0.59.0",

crates/config/README.md

@@ -130,6 +130,37 @@ 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,13 +365,36 @@ 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 write waits for a permit.
+/// Time in milliseconds an automatic foreground direct PutObject 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.1"
+rustfs-uring = "0.2.2"
 
 [target.'cfg(windows)'.dependencies]
 winapi-util.workspace = true

crates/ecstore/src/disk/local.rs

@@ -98,35 +98,6 @@ const INLINE_METADATA_ROLLBACK_DIR_XOR: u128 = 0x7275737466735f696e6c696e655f726
 const DELETE_MARKER_ROLLBACK_FILE: &str = "xl.meta.delete-marker.rollback";
 pub(crate) const DELETE_DATA_DIR_MARKER_PREFIX: &str = "delete-data.";
 pub(crate) const RESERVED_DELETE_DATA_DIR_MARKER_PREFIX: &str = "reserve-delete-data.";
-/// Largest directory read the delete-residue probe issues before it must
-/// fall back to a complete read. Residue holds one or two data dirs, so an
-/// under-filled batch settles the common case without materializing large
-/// child sets; a full batch cannot prove no listable child hides behind it.
-const DELETE_RESIDUE_PROBE_LIMIT: i32 = 8;
-
-/// A `part.N` file with a positive part number, the shape erasure data takes
-/// inside a version data dir.
-pub(crate) fn metadata_less_part_file(entry: &str) -> bool {
-    entry
-        .strip_prefix("part.")
-        .is_some_and(|part_number| part_number.parse::<usize>().is_ok_and(|part_number| part_number > 0))
-}
-
-fn is_delete_transaction_marker(entry: &str, prefix: &str) -> bool {
-    entry
-        .strip_prefix(prefix)
-        .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok_and(|uuid| !uuid.is_nil()))
-}
-
-/// Whether a `list_dir` entry inside a UUID data dir is erasure data or a
-/// delete-transaction marker. Anything else (a subdirectory, an `xl.meta`, an
-/// unknown file) means the directory is not plain delete residue.
-fn is_metadata_less_data_dir_entry(entry: &str) -> bool {
-    !entry.ends_with(SLASH_SEPARATOR)
-        && (metadata_less_part_file(entry)
-            || is_delete_transaction_marker(entry, DELETE_DATA_DIR_MARKER_PREFIX)
-            || is_delete_transaction_marker(entry, RESERVED_DELETE_DATA_DIR_MARKER_PREFIX))
-}
 const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
 const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_COUNT";
 const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS";
@@ -7673,18 +7644,15 @@ impl LocalDisk {
                         {
                             meta.name.push_str(SLASH_SEPARATOR);
                             // Conservative listings verify physical prefixes. Never-versioned
-                            // buckets use the bounded fast path, which only has to rule out
-                            // the data dirs a deleted version leaves behind; an empty listing
-                            // of such a prefix then reclaims committed residue.
-                            let listable = if opts.recursive || opts.incl_deleted {
-                                true
-                            } else if opts.skip_hidden_prefix_check {
-                                !self.directory_is_delete_residue(&opts.bucket, &meta.name, stall).await?
-                            } else {
-                                self.directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall)
+                            // buckets use the bounded fast path and reclaim residue after an
+                            // exact recursive listing proves that prefix empty.
+                            if opts.recursive
+                                || opts.incl_deleted
+                                || opts.skip_hidden_prefix_check
+                                || self
+                                    .directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall)
                                     .await?
-                            };
-                            if listable {
+                            {
                                 schedule_dir(&mut dir_stack, meta.name, false, None, true);
                             }
                         }
@@ -7808,74 +7776,6 @@ impl LocalDisk {
         Ok(false)
     }
 
-    /// Whether the metadata-less directory `dir_name` holds nothing but the
-    /// data dirs of deleted versions: it is itself a non-nil UUID directory of
-    /// `part.N` files and delete-transaction markers, or every child is one.
-    /// That is what an interrupted or deferred version delete leaves behind
-    /// once the `xl.meta` is gone, and it must not surface as a prefix. Real
-    /// object children are directories carrying their own `xl.meta`, so the
-    /// first non-UUID child, stray file, or subdirectory inside a UUID child
-    /// proves the directory is a genuine prefix. Reads are bounded: a
-    /// directory that vanishes mid-probe holds nothing listable.
-    async fn directory_is_delete_residue(&self, bucket: &str, dir_name: &str, stall: Option<Duration>) -> Result<bool> {
-        let dir_name = dir_name.trim_end_matches(SLASH_SEPARATOR);
-        let Some(entries) = self.read_dir_for_residue_probe(bucket, dir_name, stall).await? else {
-            return Ok(false);
-        };
-        if entries.is_empty() {
-            return Ok(false);
-        }
-
-        let is_data_dir = dir_name
-            .rsplit(SLASH_SEPARATOR)
-            .next()
-            .is_some_and(|name| Uuid::parse_str(name).is_ok_and(|uuid| !uuid.is_nil()));
-        if is_data_dir && entries.iter().all(|entry| is_metadata_less_data_dir_entry(entry)) {
-            return Ok(true);
-        }
-
-        for entry in entries {
-            let Some(child) = entry.strip_suffix(SLASH_SEPARATOR) else {
-                return Ok(false);
-            };
-            if !Uuid::parse_str(child).is_ok_and(|uuid| !uuid.is_nil()) {
-                return Ok(false);
-            }
-
-            let child_path = path_join_buf(&[dir_name, child]);
-            let Some(child_entries) = self.read_dir_for_residue_probe(bucket, &child_path, stall).await? else {
-                continue;
-            };
-            if !child_entries.iter().all(|entry| is_metadata_less_data_dir_entry(entry)) {
-                return Ok(false);
-            }
-        }
-
-        Ok(true)
-    }
-
-    /// Read `dir` with a bounded batch first and a complete read only when the
-    /// batch was full. `None` when the directory does not exist any more.
-    async fn read_dir_for_residue_probe(&self, bucket: &str, dir: &str, stall: Option<Duration>) -> Result<Option<Vec<String>>> {
-        for count in [DELETE_RESIDUE_PROBE_LIMIT, -1] {
-            let entries = match with_walk_stall_timeout(stall, self.list_dir("", bucket, dir, count)).await {
-                Ok(entries) => entries,
-                Err(err) => {
-                    if err == DiskError::VolumeNotFound || err == Error::FileNotFound {
-                        return Ok(None);
-                    }
-
-                    return Err(err);
-                }
-            };
-            if count < 0 || entries.len() < count as usize {
-                return Ok(Some(entries));
-            }
-        }
-
-        Ok(None)
-    }
-
     /// Whether anything under `dir_name` would appear in a listing. With
     /// `incl_deleted`, any `xl.meta` counts (versioned listings surface
     /// delete-marker-only objects too); otherwise the metadata must hold a
@@ -17855,133 +17755,6 @@ mod test {
         assert_eq!(fast_path_probes, 0);
     }
 
-    #[tokio::test]
-    async fn test_scan_dir_nonrecursive_fast_path_hides_delete_residue() {
-        use rustfs_filemeta::MetacacheReader;
-        use tempfile::tempdir;
-
-        let dir = tempdir().expect("tempdir should be created");
-        let bucket = "test-bucket";
-        let bucket_dir = dir.path().join(bucket);
-
-        async fn write_object(object_dir: &Path, object_name: &str) {
-            fs::create_dir_all(object_dir)
-                .await
-                .expect("object directory should be created");
-            let mut metadata = FileMeta::default();
-            let mut file_info = FileInfo::new(object_name, 1, 1);
-            file_info.mod_time = Some(OffsetDateTime::now_utc());
-            metadata.add_version(file_info).expect("metadata should be valid");
-            fs::write(
-                object_dir.join(STORAGE_FORMAT_FILE),
-                metadata.marshal_msg().expect("metadata should encode"),
-            )
-            .await
-            .expect("object metadata should be written");
-        }
-
-        // A deleted version whose data dir survived: part files only.
-        let residue = bucket_dir
-            .join("residue/2026/object.parquet")
-            .join(Uuid::new_v4().to_string());
-        fs::create_dir_all(&residue).await.expect("residue should be created");
-        fs::write(residue.join("part.1"), b"stale")
-            .await
-            .expect("stale part should be written");
-
-        // The same shape after a committed delete transaction.
-        let committed = bucket_dir.join("committed/object").join(Uuid::new_v4().to_string());
-        fs::create_dir_all(&committed)
-            .await
-            .expect("committed residue should be created");
-        fs::write(committed.join("part.1"), b"stale")
-            .await
-            .expect("stale part should be written");
-        fs::write(committed.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4())), [])
-            .await
-            .expect("delete marker should be written");
-
-        // A user prefix made of UUID-named directories holding real objects.
-        let upload = Uuid::new_v4().to_string();
-        write_object(&bucket_dir.join("uploads").join(&upload).join("file"), &format!("uploads/{upload}/file")).await;
-
-        // An object whose key is itself a UUID.
-        let named = Uuid::new_v4().to_string();
-        write_object(&bucket_dir.join("named").join(&named), &format!("named/{named}")).await;
-
-        // Residue next to a live child object.
-        let mixed_residue = bucket_dir.join("mixed").join(Uuid::new_v4().to_string());
-        fs::create_dir_all(&mixed_residue)
-            .await
-            .expect("mixed residue should be created");
-        fs::write(mixed_residue.join("part.1"), b"stale")
-            .await
-            .expect("stale part should be written");
-        write_object(&bucket_dir.join("mixed/child"), "mixed/child").await;
-
-        let endpoint =
-            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be UTF-8")).expect("endpoint should parse");
-        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should initialize");
-
-        async fn scan_names(disk: &LocalDisk, bucket: &str, current: &str) -> Vec<String> {
-            let (reader, mut writer) = tokio::io::duplex(64 * 1024);
-            let mut output = MetacacheWriter::new(&mut writer);
-            let opts = WalkDirOptions {
-                bucket: bucket.to_string(),
-                base_dir: current.to_string(),
-                skip_hidden_prefix_check: true,
-                ..Default::default()
-            };
-            let mut objects_returned = 0;
-            disk.scan_dir(
-                current.to_string(),
-                "".to_string(),
-                &opts,
-                &mut output,
-                &mut objects_returned,
-                false,
-                None,
-            )
-            .await
-            .expect("scan_dir should succeed");
-            output.close().await.expect("metacache writer should close");
-            drop(output);
-            drop(writer);
-
-            let mut names = MetacacheReader::new(reader)
-                .read_all()
-                .await
-                .expect("scan output should decode")
-                .into_iter()
-                .map(|entry| entry.name)
-                .collect::<Vec<_>>();
-            names.sort();
-            names
-        }
-
-        // Directories whose only content is a deleted version's data dir are
-        // not prefixes; their ancestors stay ordinary directories until an
-        // empty listing reclaims them.
-        assert_eq!(scan_names(&disk, bucket, "residue/2026/").await, Vec::<String>::new());
-        assert_eq!(scan_names(&disk, bucket, "committed/").await, Vec::<String>::new());
-
-        // UUID-named directories holding real objects, an object keyed by a
-        // UUID, and residue beside a live child all remain visible.
-        assert_eq!(scan_names(&disk, bucket, "uploads/").await, vec![format!("uploads/{upload}/")]);
-        assert_eq!(scan_names(&disk, bucket, "named/").await, vec![format!("named/{named}")]);
-        assert_eq!(scan_names(&disk, bucket, "mixed/").await, vec!["mixed/child".to_owned()]);
-        assert_eq!(
-            scan_names(&disk, bucket, "").await,
-            vec![
-                "committed/".to_owned(),
-                "mixed/".to_owned(),
-                "named/".to_owned(),
-                "residue/".to_owned(),
-                "uploads/".to_owned(),
-            ]
-        );
-    }
-
     #[tokio::test]
     async fn test_scan_dir_nonrecursive_skips_dirs_with_only_hidden_delete_markers() {
         use rustfs_filemeta::MetacacheReader;

crates/ecstore/src/set_disk/ops/heal.rs

@@ -24,7 +24,7 @@ use super::super::{
 };
 use crate::disk::DataDirDeleteStatus;
 use crate::disk::DiskAPI;
-use crate::disk::local::{DELETE_DATA_DIR_MARKER_PREFIX, metadata_less_part_file};
+use crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX;
 use crate::io_support::bitrot::object_mmap_read_enabled;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
@@ -301,6 +301,12 @@ struct MetadataLessDataDirCleanup {
     touched_disks: Vec<bool>,
 }
 
+fn metadata_less_part_file(entry: &str) -> bool {
+    entry
+        .strip_prefix("part.")
+        .is_some_and(|part_number| part_number.parse::<usize>().is_ok_and(|part_number| part_number > 0))
+}
+
 #[cfg(test)]
 struct DanglingCheckPartsFailure {
     key: DanglingCheckPartsFailureKey,

crates/ecstore/src/store/list_objects.rs

@@ -316,15 +316,10 @@ async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
         .unwrap_or(false)
 }
 
-/// Whether an empty listing of `prefix` is the proof that lets the caller
-/// reclaim delete residue under it. Any first page that scanned the whole
-/// prefix without finding an object or a sub-prefix qualifies, with or without
-/// a delimiter: that is the shape clients issue when they stat, browse, or
-/// recursively remove a phantom folder. The purge itself re-verifies every
-/// directory on every disk before deleting anything.
 fn should_purge_empty_directory_listing(
     prefix: &str,
     marker: Option<&str>,
+    delimiter: Option<&str>,
     max_keys: i32,
     incl_deleted: bool,
     result: &ListObjectsInfo,
@@ -332,7 +327,8 @@ fn should_purge_empty_directory_listing(
     !prefix.is_empty()
         && prefix.ends_with(SLASH_SEPARATOR)
         && marker.is_none()
-        && max_keys > 0
+        && delimiter.is_none_or(str::is_empty)
+        && max_keys == 1
         && !incl_deleted
         && !result.is_truncated
         && result.objects.is_empty()
@@ -3851,10 +3847,16 @@ impl ECStore {
                 .list_objects_from_opt_in_key_only_provider(&opts, mode, max_keys, incl_deleted)
                 .await?
         {
-            if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
-                && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-                    .await
-                    .unwrap_or(false)
+            if should_purge_empty_directory_listing(
+                prefix,
+                opts.marker.as_deref(),
+                delimiter.as_deref(),
+                max_keys,
+                incl_deleted,
+                &result,
+            ) && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
+                .await
+                .unwrap_or(false)
             {
                 self.purge_orphan_dir_object(bucket, prefix).await;
             }
@@ -3941,10 +3943,16 @@ impl ECStore {
             objects,
             prefixes,
         };
-        if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
-            && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-                .await
-                .unwrap_or(false)
+        if should_purge_empty_directory_listing(
+            prefix,
+            opts.marker.as_deref(),
+            delimiter.as_deref(),
+            max_keys,
+            incl_deleted,
+            &result,
+        ) && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
+            .await
+            .unwrap_or(false)
         {
             self.purge_orphan_dir_object(bucket, prefix).await;
         }
@@ -8835,28 +8843,26 @@ mod test {
     }
 
     #[test]
-    fn empty_directory_listing_purge_requires_complete_first_page_of_prefix() {
+    fn empty_directory_listing_purge_requires_complete_exact_recursive_request() {
         let empty = ListObjectsInfo::default();
-        assert!(should_purge_empty_directory_listing("ghost/", None, 1, false, &empty));
-        assert!(should_purge_empty_directory_listing("ghost/", None, 1000, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost", None, 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 0, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, true, &empty));
+        assert!(should_purge_empty_directory_listing("ghost/", None, None, 1, false, &empty));
+        assert!(should_purge_empty_directory_listing("ghost/", None, Some(""), 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost", None, None, 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), None, 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, Some("/"), 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 0, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 2, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, true, &empty));
 
         let mut live = ListObjectsInfo::default();
         live.objects.push(ObjectInfo::default());
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &live));
-
-        let mut prefixed = ListObjectsInfo::default();
-        prefixed.prefixes.push("ghost/child/".to_owned());
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &prefixed));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &live));
 
         let truncated = ListObjectsInfo {
             is_truncated: true,
             ..Default::default()
         };
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &truncated));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &truncated));
     }
 
     #[tokio::test]
@@ -8910,59 +8916,6 @@ mod test {
         }
     }
 
-    #[tokio::test]
-    async fn empty_delimiter_listing_hides_and_purges_committed_delete_residue() {
-        use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
-        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
-
-        let (dirs, store) = isolated_store_over_temp_disks().await;
-        let bucket = "listing-purge-delimiter-bucket";
-        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-        store
-            .make_bucket(bucket, &MakeBucketOptions::default())
-            .await
-            .expect("bucket should be created with authoritative metadata");
-        let data_dir = uuid::Uuid::new_v4();
-        let transaction = uuid::Uuid::new_v4();
-        for dir in &dirs {
-            let residue = dir
-                .path()
-                .join(bucket)
-                .join("metrics")
-                .join("2026")
-                .join("object.parquet")
-                .join(data_dir.to_string());
-            tokio::fs::create_dir_all(&residue)
-                .await
-                .expect("committed delete residue should be created");
-            tokio::fs::write(residue.join("part.1"), b"stale")
-                .await
-                .expect("stale part should be written");
-            tokio::fs::write(
-                residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
-                [],
-            )
-            .await
-            .expect("committed delete marker should be written");
-        }
-
-        // The object directory holds only a deleted version's data dir, so a
-        // console-style browse of its parent must not show it as a folder.
-        let result = store
-            .clone()
-            .list_objects_generic(bucket, "metrics/2026/", None, Some("/".to_owned()), 1000, false)
-            .await
-            .expect("delimiter listing should succeed");
-        assert!(result.objects.is_empty());
-        assert!(result.prefixes.is_empty(), "delete residue must not surface as a prefix");
-        for dir in &dirs {
-            assert!(
-                !dir.path().join(bucket).join("metrics").join("2026").exists(),
-                "the empty delimiter listing should reclaim the committed delete residue under it"
-            );
-        }
-    }
-
     #[test]
     fn list_objects_index_provider_state_uses_lifecycle_active_generation() {
         let provider = ListObjectsIndexProviderState::walker_key_only();

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 / `None` / 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 / 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 |
 | `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,12 +29,17 @@ 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.
@@ -149,6 +154,28 @@ 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 {
@@ -157,6 +184,7 @@ impl ForegroundWriteAdmissionGate {
             semaphore: Arc::new(Semaphore::new(limit)),
             limit,
             wait_timeout,
+            pending: Arc::new(AtomicUsize::new(0)),
         }
     }
 
@@ -164,6 +192,35 @@ 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() {
@@ -194,6 +251,8 @@ enum ForegroundWriteAdmissionPolicy {
         gate: ForegroundWriteAdmissionGate,
         put_object_min_size_bytes: usize,
         multipart_part_min_size_bytes: usize,
+        multipart_wait_timeout: Duration,
+        multipart_max_pending: usize,
     },
 }
 
@@ -252,11 +311,24 @@ 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,
         }
     }
 
@@ -275,11 +347,25 @@ 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
@@ -298,35 +384,45 @@ impl ForegroundWriteAdmissionPolicy {
                 gate,
                 put_object_min_size_bytes,
                 multipart_part_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) {
+                multipart_wait_timeout,
+                multipart_max_pending,
+            } => match kind {
+                ForegroundWriteAdmissionKind::PutObject if should_gate_foreground_write(size, *put_object_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, 0, None),
-            Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), legacy_limit, None),
+            Self::Disabled => put_admission_snapshot(0, None, 0, None),
+            Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), None, legacy_limit, None),
             Self::Strict(gate) => {
-                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"))
+                put_admission_snapshot(gate.active(), None, gate.limit, Some("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, limit: usize, hard_gate_reason: Option<&'static str>) -> WorkloadAdmissionSnapshot {
+fn put_admission_snapshot(
+    active: usize,
+    queued: Option<usize>,
+    limit: usize,
+    hard_gate_reason: Option<&'static str>,
+) -> WorkloadAdmissionSnapshot {
     let state = if limit == 0 {
         AdmissionState::Disabled
     } else if active >= limit {
@@ -336,7 +432,7 @@ fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<
     };
 
     let admission =
-        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));
+        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), queued, Some(limit));
 
     match state {
         AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
@@ -347,6 +443,13 @@ fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<
     }
 }
 
+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;
@@ -464,6 +567,24 @@ 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,
@@ -1108,7 +1229,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,
+        SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit, derive_multipart_admission_max_pending,
     };
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
@@ -1524,6 +1645,110 @@ 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);