test(scanner): parameterize heal evidence oracles

Allow the scanner/heal release evidence e2e helper to bind the active case identity and oracle file through an explicit descriptor instead of hard-coding the background target restart artifact. Extend the checker self-test so a single run can finish multiple registry-declared oracles while release gates remain pending until every required lane is complete.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

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/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -54,6 +54,23 @@ mod tests {
         test_binary: EvidenceBuild,
     }
 
+    #[derive(Clone, Copy)]
+    struct ScannerHealEvidenceCase {
+        id: &'static str,
+        oracle: &'static str,
+    }
+
+    const BACKGROUND_TARGET_RESTART_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase {
+        id: "background-target-restart",
+        oracle: "background-target-restart.json",
+    };
+
+    struct RestartEvidenceContext {
+        directory: PathBuf,
+        run: RestartEvidenceRun,
+        case: ScannerHealEvidenceCase,
+    }
+
     fn file_sha256(path: &Path) -> Result<String, Box<dyn Error + Send + Sync>> {
         let mut file = std::fs::File::open(path)?;
         let mut digest = Sha256::new();
@@ -68,10 +85,22 @@ mod tests {
         Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
     }
 
-    fn restart_evidence_run(binary: &Path) -> Result<Option<(PathBuf, RestartEvidenceRun)>, Box<dyn Error + Send + Sync>> {
+    fn restart_evidence_run(
+        binary: &Path,
+        case: ScannerHealEvidenceCase,
+    ) -> Result<Option<RestartEvidenceContext>, Box<dyn Error + Send + Sync>> {
         let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
             return Ok(None);
         };
+        if case.id.is_empty()
+            || case.oracle.is_empty()
+            || !case.oracle.ends_with(".json")
+            || case.oracle.contains('/')
+            || case.oracle.contains('\\')
+            || case.oracle.contains("..")
+        {
+            return Err("invalid scanner/heal evidence case".into());
+        }
         let directory = PathBuf::from(directory);
         let receipt = directory.join("run.json");
         if receipt.metadata()?.len() > 1024 * 1024 {
@@ -91,10 +120,10 @@ mod tests {
             run.test_binary.sha256,
             "test executable must match the run receipt"
         );
-        if directory.join("background-target-restart.json").exists() {
+        if directory.join(case.oracle).exists() {
             return Err("scanner/heal oracle already exists; create a new execution receipt".into());
         }
-        Ok(Some((directory, run)))
+        Ok(Some(RestartEvidenceContext { directory, run, case }))
     }
 
     fn compiled_test_identity() -> serde_json::Value {
@@ -964,7 +993,7 @@ mod tests {
     async fn run_cluster_root_heal_interruption(scenario: InterruptionScenario) -> Result<(), Box<dyn Error + Send + Sync>> {
         let server_binary = rustfs_binary_path();
         let evidence_run = if scenario == InterruptionScenario::BackgroundTargetRestart {
-            restart_evidence_run(&server_binary)?
+            restart_evidence_run(&server_binary, BACKGROUND_TARGET_RESTART_EVIDENCE)?
         } else {
             None
         };
@@ -1613,15 +1642,20 @@ mod tests {
             return Err(format!("heal data rebuilt but task did not finish successfully: {task_status}").into());
         }
 
-        if let Some((directory, run)) = evidence_run {
+        if let Some(evidence_context) = evidence_run {
             let restarted_pid = cluster.nodes[1].process.as_ref().ok_or("restarted target is absent")?.id();
             assert_ne!(target_pid, restarted_pid, "target must be a new process");
-            assert_eq!(file_sha256(&server_binary)?, run.binary.sha256, "server build changed during restart");
+            assert_eq!(
+                file_sha256(&server_binary)?,
+                evidence_context.run.binary.sha256,
+                "server build changed during restart"
+            );
             let evidence = serde_json::json!({
-                "schema": 1, "case": "background-target-restart", "evidence": "process-restart",
-                "run_id": run.run_id, "source_revision": run.source_revision,
+                "schema": 1, "case": evidence_context.case.id, "evidence": "process-restart",
+                "run_id": evidence_context.run.run_id, "source_revision": evidence_context.run.source_revision,
                 "test_build": compiled_test_identity(),
-                "binary_sha256": run.binary.sha256, "test_binary_sha256": run.test_binary.sha256,
+                "binary_sha256": evidence_context.run.binary.sha256,
+                "test_binary_sha256": evidence_context.run.test_binary.sha256,
                 "topology": {"nodes": cluster.nodes.len(), "drives_per_node": cluster.nodes[0].data_dirs.len()},
                 "pid_before": target_pid, "pid_after": restarted_pid,
                 "objects": evidence_objects, "node_listings": node_listings,
@@ -1633,7 +1667,7 @@ mod tests {
             let mut output = std::fs::OpenOptions::new()
                 .write(true)
                 .create_new(true)
-                .open(directory.join("background-target-restart.json"))?;
+                .open(evidence_context.directory.join(evidence_context.case.oracle))?;
             output.write_all(&data)?;
             output.sync_all()?;
         }

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);

scripts/check_test_wiring.py

@@ -1309,6 +1309,29 @@ class SelfTests(unittest.TestCase):
             self.assertIn("alternate-target-restart.json", read_json(run_dir / "execution.json")["artifacts"])
             self.assertEqual(check_scanner_heal_evidence(root, run_dir, "alternate-target-restart"), [])
 
+    def test_scanner_heal_release_consumes_multiple_registry_oracles(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            alternate = dict(registry["cases"]["background-target-restart"])
+            alternate["oracle"] = "alternate-target-restart.json"
+            registry["cases"]["alternate-target-restart"] = alternate
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            oracle = read_json(run_dir / "background-target-restart.json")
+            oracle["case"] = "alternate-target-restart"
+            write_json(run_dir / "alternate-target-restart.json", oracle)
+            (run_dir / "execution.json").unlink()
+
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            artifacts = read_json(run_dir / "execution.json")["artifacts"]
+            self.assertIn("background-target-restart.json", artifacts)
+            self.assertIn("alternate-target-restart.json", artifacts)
+            self.assertEqual(check_scanner_heal_evidence(root, run_dir, "alternate-target-restart"), [])
+            errors = check_scanner_heal_evidence(root, run_dir, "release")
+            self.assertEqual(len(errors), 21)
+            self.assertTrue(all(error.startswith("pending ") for error in errors))
+
     def test_scanner_heal_rejects_broken_execution_and_artifacts(self) -> None:
         for fault in ("exit", "missing", "zero", "skipped", "failed", "retry", "filtered", "ignored", "stale",
                       "hash", "binary", "synthetic", "wrong-run", "same-pid", "body", "parts", "listing", "topology"):