fix(ecstore): enforce receiver-side background admission

crates/ecstore/src/api/mod.rs

@@ -442,7 +442,10 @@ pub mod disk {
         ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
         is_quota_mutation_fence_path, new_disk, validate_batch_read_version_item_count,
     };
-    pub use crate::disk::{PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan};
+    pub use crate::disk::{
+        PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan, RemoteBackgroundIo,
+        with_remote_background_io,
+    };
     pub use bytes::Bytes;
     pub use endpoint::Endpoint;
     pub use error::DiskError;
@@ -589,8 +592,8 @@ pub mod rpc {
         TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce,
         decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers, gen_tonic_replay_scope_headers,
         gen_tonic_signature_headers, gen_tonic_signature_interceptor, node_service_time_out_client,
-        node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest,
-        sign_ns_scanner_capability, sign_ns_scanner_capability_with_checkpoint_coverage,
+        node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, rpc_background_admission,
+        set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_ns_scanner_capability_with_checkpoint_coverage,
         sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
         tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
         verify_ns_scanner_capability, verify_ns_scanner_capability_with_checkpoint_coverage,
@@ -640,8 +643,9 @@ pub mod storage {
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
-        all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients,
-        prewarm_local_disk_id_map, prewarm_local_disk_id_map_with_instance_ctx,
+        all_local_disk_path, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc, init_local_disks,
+        init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
+        prewarm_local_disk_id_map_with_instance_ctx,
     };
     pub use crate::store::{HealObjectAbsenceProof, HealObjectStorageResult};
 }

crates/ecstore/src/cluster/rpc/client.rs

@@ -17,10 +17,10 @@ use crate::cluster::rpc::http_auth::RPC_REPLAY_SCOPE_VERSION_HEADER;
 use crate::cluster::rpc::http_auth::{
     AuthenticatedPeerReplayCapabilities, RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER,
     RPC_CONTENT_SHA256_HEADER, RPC_REPLAY_CACHE_CAPABILITY_HEADER, RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER,
-    RollingMutationBodyDigest, TIMESTAMP_HEADER, internode_rpc_body_digest_strict,
-    verify_tonic_peer_replay_capabilities_response,
+    RollingMutationBodyDigest, TIMESTAMP_HEADER, gen_tonic_replay_scope_headers_with_admission, internode_rpc_body_digest_strict,
+    rpc_background_admission, verify_tonic_peer_replay_capabilities_response,
 };
-use crate::cluster::rpc::{gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience};
+use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
 #[cfg(test)]
 use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers};
 use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
@@ -413,7 +413,16 @@ where
                     .get(RPC_CONTENT_SHA256_HEADER)
                     .and_then(|value| value.to_str().ok()),
             ) {
-                match gen_tonic_replay_scope_headers(audience, request.uri().path(), timestamp, content_sha256, boot_epoch) {
+                match rpc_background_admission(request.headers()).and_then(|background| {
+                    gen_tonic_replay_scope_headers_with_admission(
+                        audience,
+                        request.uri().path(),
+                        timestamp,
+                        content_sha256,
+                        boot_epoch,
+                        background,
+                    )
+                }) {
                     Ok(headers) => request.headers_mut().extend(headers),
                     Err(error) => debug!(error = %error, "could not attach replay-scoped RPC signature"),
                 }

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -61,6 +61,9 @@ use uuid::Uuid;
 type HmacSha256 = Hmac<Sha256>;
 
 const SIGNATURE_HEADER: &str = "x-rustfs-signature";
+pub const RPC_BACKGROUND_ADMISSION_HEADER: &str = "x-rustfs-rpc-background-admission";
+const RPC_BACKGROUND_ADMISSION_V1: &str = "v1";
+const RPC_BACKGROUND_ADMISSION_DOMAIN: &[u8] = b"rustfs-rpc-background-admission-v1\0";
 pub(crate) const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp";
 pub(crate) const RPC_AUTH_VERSION_HEADER: &str = "x-rustfs-rpc-auth-version";
 const RPC_SIGNATURE_V2_HEADER: &str = "x-rustfs-rpc-signature-v2";
@@ -77,6 +80,44 @@ pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_HEADER: &str = "x-rustfs-rpc-replay
 pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER: &str = "x-rustfs-rpc-replay-cache-capability-proof";
 const RPC_REPLAY_SCOPE_VERSION_V3: &str = "3";
 const RPC_RESPONSE_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-response-proof-v1\0";
+
+tokio::task_local! {
+    static RPC_BACKGROUND_REQUEST: bool;
+}
+
+pub(crate) fn current_rpc_background_request() -> bool {
+    RPC_BACKGROUND_REQUEST
+        .try_with(|background| *background)
+        .unwrap_or_else(|_| crate::disk::has_resource_io_scope())
+}
+
+pub(crate) async fn with_rpc_background_request<F: std::future::Future>(background: bool, operation: F) -> F::Output {
+    RPC_BACKGROUND_REQUEST.scope(background, operation).await
+}
+
+/// Parse the request class only; callers must verify the signature before using it.
+pub fn rpc_background_admission(headers: &HeaderMap) -> std::io::Result<bool> {
+    let mut values = headers.get_all(RPC_BACKGROUND_ADMISSION_HEADER).iter();
+    match (values.next(), values.next()) {
+        (None, None) => Ok(false),
+        (Some(value), None) if value == RPC_BACKGROUND_ADMISSION_V1 => Ok(true),
+        _ => Err(std::io::Error::other("Invalid RPC background admission version")),
+    }
+}
+
+fn bind_background_admission(mac: &mut HmacSha256, background: bool) {
+    if background {
+        // Every auth lane binds the class, so stripping the marker or falling
+        // back to an older signature cannot turn admitted work into foreground I/O.
+        mac.update(RPC_BACKGROUND_ADMISSION_DOMAIN);
+    }
+}
+
+fn insert_background_admission(headers: &mut HeaderMap, background: bool) {
+    if background {
+        headers.insert(RPC_BACKGROUND_ADMISSION_HEADER, HeaderValue::from_static(RPC_BACKGROUND_ADMISSION_V1));
+    }
+}
 const RPC_REPLAY_SCOPE_DOMAIN: &[u8] = b"rustfs-rpc-replay-scope-v3\0";
 const RPC_BOOT_EPOCH_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-boot-epoch-proof-v1\0";
 const RPC_REPLAY_CACHE_CAPABILITY_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-replay-cache-capability-proof-v1\0";
@@ -519,21 +560,23 @@ fn redacted_rpc_path(url: &str) -> String {
 }
 
 /// Generate HMAC-SHA256 signature for the given data
-fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64) -> String {
+fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64, background: bool) -> String {
     let data = signature_payload(url, method, timestamp);
     let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    bind_background_admission(&mut mac, background);
     mac.update(data.as_bytes());
     let result = mac.finalize();
     base64_simd::STANDARD.encode_to_string(result.into_bytes())
 }
 
-fn verify_signature(secret: &str, url: &str, method: &Method, timestamp: i64, signature: &str) -> bool {
+fn verify_signature(secret: &str, url: &str, method: &Method, timestamp: i64, signature: &str, background: bool) -> bool {
     let Ok(signature) = base64_simd::STANDARD.decode_to_vec(signature) else {
         return false;
     };
 
     let data = signature_payload(url, method, timestamp);
     let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    bind_background_admission(&mut mac, background);
     mac.update(data.as_bytes());
     mac.verify_slice(&signature).is_ok()
 }
@@ -771,6 +814,7 @@ pub fn verify_ns_scanner_capability_with_checkpoint_coverage(
 
 #[derive(Clone, Copy)]
 struct SignatureV2Scope<'a> {
+    background: bool,
     audience: &'a str,
     service: &'a str,
     rpc_method: &'a str,
@@ -780,6 +824,7 @@ struct SignatureV2Scope<'a> {
 }
 
 fn update_signature_v2(mac: &mut HmacSha256, scope: SignatureV2Scope<'_>) {
+    bind_background_admission(mac, scope.background);
     for part in [
         b"rustfs-rpc-auth-v2|".as_slice(),
         scope.audience.as_bytes(),
@@ -818,6 +863,7 @@ fn verify_signature_v2(secret: &str, scope: SignatureV2Scope<'_>, signature: &st
 
 #[derive(Clone, Copy)]
 struct ReplayScope<'a> {
+    background: bool,
     audience: &'a str,
     path: &'a str,
     timestamp: &'a str,
@@ -827,6 +873,7 @@ struct ReplayScope<'a> {
 }
 
 fn update_replay_scope(mac: &mut HmacSha256, scope: ReplayScope<'_>) {
+    bind_background_admission(mac, scope.background);
     mac.update(RPC_REPLAY_SCOPE_DOMAIN);
     for part in [
         scope.audience.as_bytes(),
@@ -970,6 +1017,24 @@ pub fn gen_tonic_replay_scope_headers(
     timestamp: &str,
     content_sha256: &str,
     boot_epoch: Uuid,
+) -> std::io::Result<HeaderMap> {
+    gen_tonic_replay_scope_headers_with_admission(
+        audience,
+        path,
+        timestamp,
+        content_sha256,
+        boot_epoch,
+        current_rpc_background_request(),
+    )
+}
+
+pub(crate) fn gen_tonic_replay_scope_headers_with_admission(
+    audience: &str,
+    path: &str,
+    timestamp: &str,
+    content_sha256: &str,
+    boot_epoch: Uuid,
+    background: bool,
 ) -> std::io::Result<HeaderMap> {
     if audience.is_empty() || !path.starts_with('/') || !valid_content_sha256(content_sha256) || boot_epoch.is_nil() {
         return Err(std::io::Error::other("Invalid replay-scoped RPC signing scope"));
@@ -983,6 +1048,7 @@ pub fn gen_tonic_replay_scope_headers(
     let signature = generate_replay_scope_signature(
         &get_shared_secret()?,
         ReplayScope {
+            background,
             audience,
             path,
             timestamp,
@@ -992,6 +1058,7 @@ pub fn gen_tonic_replay_scope_headers(
         },
     )?;
     let mut headers = HeaderMap::new();
+    insert_background_admission(&mut headers, background);
     headers.insert(RPC_REPLAY_SCOPE_VERSION_HEADER, HeaderValue::from_static(RPC_REPLAY_SCOPE_VERSION_V3));
     headers.insert(
         RPC_REPLAY_SCOPE_SIGNATURE_HEADER,
@@ -1226,10 +1293,12 @@ pub fn build_auth_headers(url: &str, method: &Method, headers: &mut HeaderMap) -
 pub fn gen_signature_headers(url: &str, method: &Method) -> std::io::Result<HeaderMap> {
     let secret = get_shared_secret()?;
     let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+    let background = current_rpc_background_request();
 
-    let signature = generate_signature(&secret, url, method, timestamp);
+    let signature = generate_signature(&secret, url, method, timestamp, background);
 
     let mut headers = HeaderMap::new();
+    insert_background_admission(&mut headers, background);
     headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).expect("Invalid header value"));
     headers.insert(
         TIMESTAMP_HEADER,
@@ -1264,10 +1333,12 @@ pub fn gen_tonic_signature_headers(
     let timestamp_header = timestamp.to_string();
     let body_nonce = (content_sha256 != UNSIGNED_PAYLOAD).then(|| Uuid::new_v4().to_string());
     let nonce = body_nonce.as_deref().unwrap_or(UNSIGNED_PAYLOAD_NONCE);
-    let legacy_signature = generate_signature(&secret, TONIC_RPC_PREFIX, &Method::GET, timestamp);
+    let background = current_rpc_background_request();
+    let legacy_signature = generate_signature(&secret, TONIC_RPC_PREFIX, &Method::GET, timestamp, background);
     let signature_v2 = generate_signature_v2(
         &secret,
         SignatureV2Scope {
+            background,
             audience,
             service,
             rpc_method,
@@ -1278,6 +1349,7 @@ pub fn gen_tonic_signature_headers(
     )?;
 
     let mut headers = HeaderMap::new();
+    insert_background_admission(&mut headers, background);
     headers.insert(SIGNATURE_HEADER, header_value(&legacy_signature, SIGNATURE_HEADER)?);
     headers.insert(TIMESTAMP_HEADER, header_value(&timestamp_header, TIMESTAMP_HEADER)?);
     headers.insert(RPC_AUTH_VERSION_HEADER, HeaderValue::from_static(RPC_AUTH_VERSION_V2));
@@ -1501,6 +1573,7 @@ fn verify_tonic_replay_scope_signature(audience: &str, path: &str, headers: &Hea
     if !verify_replay_scope_signature(
         &secret,
         ReplayScope {
+            background: rpc_background_admission(headers)?,
             audience,
             path,
             timestamp,
@@ -1625,6 +1698,9 @@ fn verify_tonic_rpc_signature_with_strictness(
     strict: bool,
 ) -> std::io::Result<()> {
     if !has_v2_auth_headers(headers) {
+        if rpc_background_admission(headers)? {
+            return Err(std::io::Error::other("RPC background admission requires v2 authentication"));
+        }
         // RUSTFS_COMPAT_TODO(heal-rpc-auth-v2): accept old peers during rolling upgrades. Remove after the minimum
         // supported RustFS peer version sends v2 authentication on every internode gRPC request.
         if strict {
@@ -1696,6 +1772,7 @@ fn verify_tonic_rpc_signature_with_strictness(
     if !verify_signature_v2(
         &secret,
         SignatureV2Scope {
+            background: rpc_background_admission(headers)?,
             audience,
             service,
             rpc_method,
@@ -1736,7 +1813,7 @@ pub fn verify_rpc_signature(url: &str, method: &Method, headers: &HeaderMap) ->
     // Verify signature with constant-time HMAC comparison.
     let secret = get_shared_secret()?;
 
-    if !verify_signature(&secret, url, method, timestamp, signature) {
+    if !verify_signature(&secret, url, method, timestamp, signature, rpc_background_admission(headers)?) {
         let rpc_path = redacted_rpc_path(url);
         error!(
             rpc_path = %rpc_path,
@@ -1828,6 +1905,98 @@ mod tests {
         runtime_sources::ensure_test_rpc_secret();
     }
 
+    #[tokio::test]
+    async fn background_admission_is_bound_in_http_and_both_tonic_signature_lanes() {
+        ensure_test_rpc_secret();
+        let secret = get_shared_secret().expect("RPC secret");
+        let path = "/node_service.NodeService/ReadAll";
+        for background in [false, true] {
+            let (http, tonic) = with_rpc_background_request(background, async {
+                (
+                    gen_signature_headers("/rustfs/rpc/read?disk=a", &Method::GET).expect("HTTP signature"),
+                    gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "ReadAll", None)
+                        .expect("gRPC signature"),
+                )
+            })
+            .await;
+            assert_eq!(rpc_background_admission(&http).expect("HTTP class"), background);
+            assert_eq!(rpc_background_admission(&tonic).expect("gRPC class"), background);
+            verify_rpc_signature("/rustfs/rpc/read?disk=a", &Method::GET, &http).expect("HTTP authentication");
+            verify_tonic_rpc_signature_with_strictness("node-a:9000", path, &tonic, true).expect("v2 authentication");
+
+            let timestamp = tonic[TIMESTAMP_HEADER].to_str().expect("timestamp");
+            let content = tonic[RPC_CONTENT_SHA256_HEADER].to_str().expect("content digest");
+            assert_eq!(
+                verify_signature_v2(
+                    &secret,
+                    SignatureV2Scope {
+                        background: false,
+                        audience: "node-a:9000",
+                        service: "node_service.NodeService",
+                        rpc_method: "ReadAll",
+                        timestamp,
+                        nonce: tonic[RPC_NONCE_HEADER].to_str().expect("nonce"),
+                        content_sha256: content,
+                    },
+                    tonic[RPC_SIGNATURE_V2_HEADER].to_str().expect("signature"),
+                ),
+                !background,
+                "an old peer cannot authenticate a background request as foreground"
+            );
+
+            let mut replay = tonic.clone();
+            // The channel can run outside the caller's task-local scope.
+            replay.extend(
+                gen_tonic_replay_scope_headers_with_admission(
+                    "node-a:9000",
+                    path,
+                    timestamp,
+                    content,
+                    tonic_rpc_boot_epoch(),
+                    background,
+                )
+                .expect("v3 signature"),
+            );
+            for mutation in ["strip-or-add", "unknown", "duplicate"] {
+                let alter = |headers: &HeaderMap| {
+                    let mut changed = headers.clone();
+                    match mutation {
+                        "strip-or-add" if background => {
+                            changed.remove(RPC_BACKGROUND_ADMISSION_HEADER);
+                        }
+                        "strip-or-add" => insert_background_admission(&mut changed, true),
+                        "unknown" => {
+                            changed.insert(RPC_BACKGROUND_ADMISSION_HEADER, HeaderValue::from_static("v2"));
+                        }
+                        _ => {
+                            insert_background_admission(&mut changed, true);
+                            changed.append(RPC_BACKGROUND_ADMISSION_HEADER, HeaderValue::from_static("v1"));
+                        }
+                    }
+                    changed
+                };
+                assert!(verify_rpc_signature("/rustfs/rpc/read?disk=a", &Method::GET, &alter(&http)).is_err());
+                assert!(verify_tonic_rpc_signature_with_strictness("node-a:9000", path, &alter(&tonic), true).is_err());
+                assert!(verify_tonic_replay_scope_signature("node-a:9000", path, &alter(&replay)).is_err());
+            }
+            verify_tonic_replay_scope_signature("node-a:9000", path, &replay).expect("v3 authentication");
+            assert!(
+                verify_tonic_replay_scope_signature("node-a:9000", path, &replay).is_err(),
+                "replay rejected"
+            );
+
+            if background {
+                let mut downgraded = HeaderMap::new();
+                for name in [SIGNATURE_HEADER, TIMESTAMP_HEADER, RPC_BACKGROUND_ADMISSION_HEADER] {
+                    downgraded.insert(name, tonic[name].clone());
+                }
+                assert!(verify_tonic_rpc_signature_with_strictness("node-a:9000", path, &downgraded, false).is_err());
+                downgraded.remove(RPC_BACKGROUND_ADMISSION_HEADER);
+                assert!(verify_tonic_rpc_signature_with_strictness("node-a:9000", path, &downgraded, false).is_err());
+            }
+        }
+    }
+
     #[test]
     fn namespace_scanner_capability_proof_binds_challenge_and_server_epoch() {
         let secret = "test-scanner-capability-secret";
@@ -1965,8 +2134,8 @@ mod tests {
         let method = Method::GET;
         let timestamp = 1640995200; // Fixed timestamp
 
-        let signature1 = generate_signature(secret, url, &method, timestamp);
-        let signature2 = generate_signature(secret, url, &method, timestamp);
+        let signature1 = generate_signature(secret, url, &method, timestamp, false);
+        let signature2 = generate_signature(secret, url, &method, timestamp, false);
 
         assert_eq!(signature1, signature2, "Same inputs should produce same signature");
         assert!(!signature1.is_empty(), "Signature should not be empty");
@@ -1979,10 +2148,10 @@ mod tests {
         let method = Method::GET;
         let timestamp = 1640995200;
 
-        let signature1 = generate_signature(secret, url, &method, timestamp);
-        let signature2 = generate_signature(secret, "http://different.com/api/test2", &method, timestamp);
-        let signature3 = generate_signature(secret, url, &Method::POST, timestamp);
-        let signature4 = generate_signature(secret, url, &method, timestamp + 1);
+        let signature1 = generate_signature(secret, url, &method, timestamp, false);
+        let signature2 = generate_signature(secret, "http://different.com/api/test2", &method, timestamp, false);
+        let signature3 = generate_signature(secret, url, &Method::POST, timestamp, false);
+        let signature4 = generate_signature(secret, url, &method, timestamp + 1, false);
 
         assert_ne!(signature1, signature2, "Different URLs should produce different signatures");
         assert_ne!(signature1, signature3, "Different methods should produce different signatures");
@@ -2088,14 +2257,14 @@ mod tests {
         let url = "http://example.com/api/test";
         let method = Method::GET;
         let timestamp = 1640995200;
-        let signature = generate_signature(secret, url, &method, timestamp);
+        let signature = generate_signature(secret, url, &method, timestamp, false);
         let mut tampered = base64_simd::STANDARD.decode_to_vec(&signature).unwrap();
         tampered[0] ^= 1;
         let tampered_signature = base64_simd::STANDARD.encode_to_string(tampered);
 
-        assert!(verify_signature(secret, url, &method, timestamp, &signature));
-        assert!(!verify_signature(secret, url, &method, timestamp, &tampered_signature));
-        assert!(!verify_signature(secret, url, &method, timestamp, "invalid-signature"));
+        assert!(verify_signature(secret, url, &method, timestamp, &signature, false));
+        assert!(!verify_signature(secret, url, &method, timestamp, &tampered_signature, false));
+        assert!(!verify_signature(secret, url, &method, timestamp, "invalid-signature", false));
     }
 
     #[test]
@@ -2112,11 +2281,11 @@ mod tests {
         );
         let method = Method::GET;
         let timestamp = 1_640_995_200;
-        let signature = generate_signature(secret, signed_url, &method, timestamp);
+        let signature = generate_signature(secret, signed_url, &method, timestamp, false);
 
-        assert!(verify_signature(secret, signed_url, &method, timestamp, &signature));
-        assert!(!verify_signature(secret, downgraded_url, &method, timestamp, &signature));
-        assert!(!verify_signature(secret, tampered_body_digest, &method, timestamp, &signature));
+        assert!(verify_signature(secret, signed_url, &method, timestamp, &signature, false));
+        assert!(!verify_signature(secret, downgraded_url, &method, timestamp, &signature, false));
+        assert!(!verify_signature(secret, tampered_body_digest, &method, timestamp, &signature, false));
     }
 
     #[test]
@@ -2126,7 +2295,7 @@ mod tests {
         let method = Method::GET;
         let timestamp = OffsetDateTime::now_utc().unix_timestamp();
         let secret = get_shared_secret().expect("test RPC secret should resolve");
-        let expected_signature = generate_signature(&secret, url, &method, timestamp);
+        let expected_signature = generate_signature(&secret, url, &method, timestamp, false);
         let invalid_signature = "invalid-signature";
         let logs = CapturedLogs::default();
         let subscriber = tracing_subscriber::fmt()
@@ -2164,7 +2333,7 @@ mod tests {
         // Set expired timestamp (older than SIGNATURE_VALID_DURATION)
         let expired_timestamp = OffsetDateTime::now_utc().unix_timestamp() - SIGNATURE_VALID_DURATION - 10;
         let secret = get_shared_secret().expect("test RPC secret should resolve");
-        let signature = generate_signature(&secret, url, &method, expired_timestamp);
+        let signature = generate_signature(&secret, url, &method, expired_timestamp, false);
 
         headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
         headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&expired_timestamp.to_string()).unwrap());
@@ -2186,7 +2355,7 @@ mod tests {
 
         let future_timestamp = OffsetDateTime::now_utc().unix_timestamp() + SIGNATURE_VALID_DURATION + 10;
         let secret = get_shared_secret().expect("test RPC secret should resolve");
-        let signature = generate_signature(&secret, url, &method, future_timestamp);
+        let signature = generate_signature(&secret, url, &method, future_timestamp, false);
 
         headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
         headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&future_timestamp.to_string()).unwrap());
@@ -2300,7 +2469,7 @@ mod tests {
         // Test timestamp just within valid duration
         let valid_timestamp = current_time - SIGNATURE_VALID_DURATION + 1;
 
-        let signature = generate_signature(&secret, url, &method, valid_timestamp);
+        let signature = generate_signature(&secret, url, &method, valid_timestamp, false);
 
         headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
         headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&valid_timestamp.to_string()).unwrap());
@@ -2311,7 +2480,7 @@ mod tests {
         // Test timestamp just outside valid duration
         let mut headers = HeaderMap::new();
         let invalid_timestamp = current_time - SIGNATURE_VALID_DURATION - 15;
-        let signature = generate_signature(&secret, url, &method, invalid_timestamp);
+        let signature = generate_signature(&secret, url, &method, invalid_timestamp, false);
 
         headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
         headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&invalid_timestamp.to_string()).unwrap());

crates/ecstore/src/cluster/rpc/internode_data_transport.rs

@@ -948,6 +948,90 @@ mod tests {
     use tokio::io::AsyncWriteExt;
     use tokio::sync::{Barrier, Notify};
 
+    #[tokio::test]
+    async fn admitted_remote_reads_bind_receiver_admission_to_the_http_signature() {
+        use crate::disk::with_background_resource_unit;
+        use rustfs_concurrency::adaptive::AdaptiveAdmission;
+        use std::num::NonZeroU16;
+
+        crate::runtime::sources::ensure_test_rpc_secret();
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("remote admission listener");
+        let address = listener.local_addr().expect("remote admission address");
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.expect("remote read connection");
+            let mut bytes = Vec::new();
+            while !bytes.windows(4).any(|window| window == b"\r\n\r\n") {
+                let mut chunk = [0; 1024];
+                let read = socket.read(&mut chunk).await.expect("remote read headers");
+                assert!(read > 0 && bytes.len() < 16 * 1024, "bounded complete request headers");
+                bytes.extend_from_slice(&chunk[..read]);
+            }
+            let request = std::str::from_utf8(&bytes).expect("HTTP request");
+            let mut lines = request.split("\r\n");
+            let target = lines
+                .next()
+                .expect("request line")
+                .split_whitespace()
+                .nth(1)
+                .expect("request target")
+                .to_owned();
+            let mut headers = HeaderMap::new();
+            for line in lines.take_while(|line| !line.is_empty()) {
+                let (name, value) = line.split_once(':').expect("header field");
+                headers.append(
+                    http::header::HeaderName::from_bytes(name.as_bytes()).expect("header name"),
+                    HeaderValue::from_str(value.trim()).expect("header value"),
+                );
+            }
+            socket
+                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 5\r\nconnection: close\r\n\r\nhello")
+                .await
+                .expect("remote read response");
+            (target, headers)
+        });
+        let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+        let resource = gate.try_acquire().expect("sender device admission");
+        let mut reader = tokio::time::timeout(
+            Duration::from_secs(5),
+            with_background_resource_unit(
+                resource,
+                TcpHttpInternodeDataTransport.open_read(ReadStreamRequest {
+                    endpoint: format!("http://{address}"),
+                    disk: "disk-a".to_owned(),
+                    volume: "bucket".to_owned(),
+                    path: "part.1".to_owned(),
+                    offset: 0,
+                    length: 5,
+                    stall_timeout: Some(Duration::from_secs(5)),
+                }),
+            ),
+        )
+        .await
+        .expect("bounded remote request")
+        .expect("remote reader");
+        let mut body = Vec::new();
+        reader.read_to_end(&mut body).await.expect("complete remote body");
+        assert_eq!(body, b"hello");
+        let (target, headers) = server.await.expect("remote request capture");
+        assert_eq!(gate.snapshot().active, 0, "remote body does not retain the sender's device");
+        assert_eq!(
+            headers
+                .get("x-rustfs-rpc-background-admission")
+                .and_then(|value| value.to_str().ok()),
+            Some("v1"),
+            "resource-controlled remote work must request receiver-side admission"
+        );
+        crate::cluster::rpc::verify_rpc_signature(&target, &Method::GET, &headers).expect("bound request signature");
+        let mut stripped = headers;
+        stripped.remove("x-rustfs-rpc-background-admission");
+        assert!(
+            crate::cluster::rpc::verify_rpc_signature(&target, &Method::GET, &stripped).is_err(),
+            "removing background admission must invalidate the signature"
+        );
+    }
+
     async fn wait_for_capability_flight_waiters(entry: &PutFileCapabilityCacheEntry, waiters: usize) {
         tokio::time::timeout(Duration::from_secs(5), async {
             loop {

crates/ecstore/src/cluster/rpc/mod.rs

@@ -33,16 +33,17 @@ pub use client::{
 // Re-exported through `api::rpc`; not every item is consumed inside this crate.
 #[allow(unused_imports)]
 pub use http_auth::{
-    TONIC_RPC_PREFIX, build_auth_headers, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, gen_signature_headers,
-    gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest,
-    set_tonic_mutation_body_digest, set_tonic_rolling_canonical_body_digest, set_tonic_rolling_mutation_body_digest,
-    sign_ns_scanner_capability, sign_ns_scanner_capability_with_checkpoint_coverage,
-    sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
-    tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_ns_scanner_capability,
-    verify_ns_scanner_capability_with_checkpoint_coverage, verify_ns_scanner_capability_with_tier_registry_generation,
-    verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response,
-    verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned,
-    verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
+    RPC_BACKGROUND_ADMISSION_HEADER, TONIC_RPC_PREFIX, build_auth_headers, build_put_file_auth_trailer,
+    check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
+    normalize_tonic_rpc_audience, rpc_background_admission, set_tonic_canonical_body_digest, set_tonic_mutation_body_digest,
+    set_tonic_rolling_canonical_body_digest, set_tonic_rolling_mutation_body_digest, sign_ns_scanner_capability,
+    sign_ns_scanner_capability_with_checkpoint_coverage, sign_ns_scanner_capability_with_tier_registry_generation,
+    sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
+    tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_ns_scanner_capability_with_checkpoint_coverage,
+    verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer, verify_put_file_capability,
+    verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
+    verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned, verify_tonic_rpc_response_proof,
+    verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
 };
 #[cfg(test)]
 pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -17,6 +17,7 @@ use crate::cluster::rpc::client::{
     node_service_delete_recovery_client, node_service_time_out_client, node_service_time_out_client_for_class,
     node_service_time_out_client_no_auth,
 };
+use crate::cluster::rpc::http_auth::{current_rpc_background_request, with_rpc_background_request};
 use crate::cluster::rpc::internode_data_transport::{
     InternodeDataTransport, NsScannerCapabilityRequest, NsScannerStreamRequest, ReadStreamRequest, WalkDirStreamRequest,
     WriteStreamRequest,
@@ -533,6 +534,7 @@ struct RetryingRemoteReader {
     reader: Option<FileReader>,
     transport: Arc<dyn InternodeDataTransport>,
     request: ReadStreamRequest,
+    background: bool,
     emitted: usize,
     retried: bool,
     initial_read_timeout: Option<Duration>,
@@ -553,6 +555,7 @@ impl RetryingRemoteReader {
             reader: Some(reader),
             transport,
             request,
+            background: current_rpc_background_request(),
             emitted: 0,
             retried: false,
             initial_read_timeout,
@@ -570,9 +573,10 @@ impl RetryingRemoteReader {
         let request = resumed_read_request(&self.request, self.emitted)?;
         let recovery_timeout = self.recovery_timeout;
         let transport = Arc::clone(&self.transport);
-        self.resume = Some(AbortOnDropTask::new(tokio::spawn(async move {
-            with_remote_read_recovery_timeout(recovery_timeout, transport.open_read_fresh(request)).await
-        })));
+        self.resume = Some(AbortOnDropTask::new(tokio::spawn(with_rpc_background_request(
+            self.background,
+            async move { with_remote_read_recovery_timeout(recovery_timeout, transport.open_read_fresh(request)).await },
+        ))));
         Ok(())
     }
 
@@ -680,6 +684,7 @@ struct RetryingRemoteChunkReader {
     reader: Option<rustfs_rio::ChunkReaderBox>,
     transport: Arc<dyn InternodeDataTransport>,
     request: ReadStreamRequest,
+    background: bool,
     emitted: usize,
     retried: bool,
     initial_read_timeout: Option<Duration>,
@@ -700,6 +705,7 @@ impl RetryingRemoteChunkReader {
             reader: Some(reader),
             transport,
             request,
+            background: current_rpc_background_request(),
             emitted: 0,
             retried: false,
             initial_read_timeout,
@@ -717,9 +723,10 @@ impl RetryingRemoteChunkReader {
         let request = resumed_read_request(&self.request, self.emitted)?;
         let recovery_timeout = self.recovery_timeout;
         let transport = Arc::clone(&self.transport);
-        self.resume = Some(AbortOnDropTask::new(tokio::spawn(async move {
-            with_remote_read_recovery_timeout(recovery_timeout, transport.open_read_chunks_fresh(request)).await
-        })));
+        self.resume = Some(AbortOnDropTask::new(tokio::spawn(with_rpc_background_request(
+            self.background,
+            async move { with_remote_read_recovery_timeout(recovery_timeout, transport.open_read_chunks_fresh(request)).await },
+        ))));
         Ok(())
     }
 
@@ -7071,6 +7078,98 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    async fn remote_read_retries_preserve_signed_admission_after_sender_scope_ends() {
+        use crate::cluster::rpc::{rpc_background_admission, verify_rpc_signature};
+        use crate::disk::with_background_resource_unit;
+        use rustfs_concurrency::adaptive::AdaptiveAdmission;
+        use std::num::NonZeroU16;
+
+        crate::runtime::sources::ensure_test_rpc_secret();
+        for background in [false, true] {
+            for chunked in [false, true] {
+                let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("retry listener");
+                let address = listener.local_addr().expect("retry address");
+                let server = tokio::spawn(async move {
+                    let (mut socket, _) = listener.accept().await.expect("fresh connection");
+                    let mut bytes = Vec::new();
+                    while !bytes.windows(4).any(|window| window == b"\r\n\r\n") {
+                        let mut chunk = [0; 1024];
+                        let read = socket.read(&mut chunk).await.expect("retry request");
+                        assert!(read > 0 && bytes.len() < 16 * 1024, "bounded complete headers");
+                        bytes.extend_from_slice(&chunk[..read]);
+                    }
+                    let request = std::str::from_utf8(&bytes).expect("HTTP request");
+                    let mut lines = request.split("\r\n");
+                    let target = lines
+                        .next()
+                        .expect("request line")
+                        .split_whitespace()
+                        .nth(1)
+                        .expect("target")
+                        .to_owned();
+                    let mut headers = http::HeaderMap::new();
+                    for line in lines.take_while(|line| !line.is_empty()) {
+                        let (name, value) = line.split_once(':').expect("header");
+                        headers.append(
+                            http::header::HeaderName::from_bytes(name.as_bytes()).expect("header name"),
+                            http::HeaderValue::from_str(value.trim()).expect("header value"),
+                        );
+                    }
+                    socket
+                        .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 6\r\nconnection: close\r\n\r\n456789")
+                        .await
+                        .expect("retry response");
+                    (target, headers)
+                });
+                let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+                let request = ReadStreamRequest {
+                    endpoint: format!("http://{address}"),
+                    ..resume_request(10)
+                };
+                let create = async {
+                    let transport = Arc::new(TcpHttpInternodeDataTransport);
+                    if chunked {
+                        Box::new(RetryingRemoteChunkReader::new_with_timeouts(
+                            resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec())),
+                            transport,
+                            request,
+                            None,
+                            None,
+                        )) as FileReader
+                    } else {
+                        Box::new(RetryingRemoteReader::new_with_timeouts(
+                            resume_step_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec())),
+                            transport,
+                            request,
+                            None,
+                            None,
+                        )) as FileReader
+                    }
+                };
+                let mut reader = if background {
+                    with_background_resource_unit(gate.try_acquire().expect("sender admission"), create).await
+                } else {
+                    create.await
+                };
+                assert_eq!(gate.snapshot().active, 0, "retry reader cannot retain sender disk capacity");
+                let mut output = Vec::new();
+                tokio::time::timeout(
+                    Duration::from_secs(5),
+                    with_rpc_background_request(!background, reader.read_to_end(&mut output)),
+                )
+                .await
+                .expect("bounded retry")
+                .expect("resumed body");
+                assert_eq!(output, b"0123456789", "chunked={chunked}, background={background}");
+                let (target, headers) = server.await.expect("captured retry");
+                assert!(target.contains("offset=11") && target.contains("length=6"), "{target}");
+                assert_eq!(rpc_background_admission(&headers).expect("admission marker"), background);
+                verify_rpc_signature(&target, &http::Method::GET, &headers).expect("signed retry class");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn remote_reader_resumes_from_emitted_bytes_without_duplicates() {
         let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::Data(b"456789".to_vec())]));

crates/ecstore/src/disk/disk_store.rs

@@ -1673,6 +1673,9 @@ impl LocalDiskWrapper {
         use crate::disk::io_admission::ResourceIoUnit;
 
         let Some(resources) = current_resource_unit() else {
+            if super::remote_io_admission::active() {
+                return Err(DiskError::Io(io::Error::from(io::ErrorKind::WouldBlock)));
+            }
             return Ok(operation.await);
         };
         let resources = if let Some(permits) = resources.as_set() {
@@ -2440,6 +2443,7 @@ impl LocalDiskWrapper {
             self.metrics.record_operation_latency(op, started.elapsed());
             self.record_result_error_metrics(&result);
             if result.is_ok() {
+                super::remote_io_admission::record_progress();
                 self.health.record_operation_success(&self.endpoint(), "operation_success");
                 self.metrics.record_mutation_success(mutation);
             }
@@ -2454,6 +2458,7 @@ impl LocalDiskWrapper {
                 self.record_result_error_metrics(&operation_result);
                 // Log success; the waiting guard balances every exit path.
                 if operation_result.is_ok() {
+                    super::remote_io_admission::record_progress();
                     self.health.record_operation_success(&self.endpoint(), "operation_success");
                     self.metrics.record_mutation_success(mutation);
                 }
@@ -3920,6 +3925,15 @@ mod tests {
 
     #[tokio::test]
     async fn local_disk_resource_identity_probes_keep_only_their_native_device_after_cancel() {
+        assert_native_identity_probe_cancellation(false).await;
+    }
+
+    #[tokio::test]
+    async fn remote_background_native_identity_cancellation_keeps_receiver_device() {
+        assert_native_identity_probe_cancellation(true).await;
+    }
+
+    async fn assert_native_identity_probe_cancellation(remote: bool) {
         use rustfs_concurrency::adaptive::{
             AdaptiveAdmission, FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure,
             permit_set::AdaptivePermitSetOwner,
@@ -3972,19 +3986,29 @@ mod tests {
                         completed_units: None,
                     }),
                 );
-                let parent = shared.try_acquire_with_node(&node).expect("planned devices");
+                let parent = (!remote).then(|| shared.try_acquire_with_node(&node).expect("planned devices"));
                 let task = tokio::spawn({
                     let wrapper = wrapper.clone();
+                    let node = node.clone();
                     async move {
-                        crate::disk::with_background_resource_set(parent.resources().clone(), async {
+                        let operation = async {
                             if online {
                                 assert!(wrapper.is_online().await);
                             } else {
                                 wrapper.get_disk_id().await.expect("disk identity");
                             }
-                        })
-                        .await;
-                        parent.complete();
+                        };
+                        if remote {
+                            crate::disk::with_remote_background_io(Some(crate::disk::RemoteBackgroundIo::new(node)), async {
+                                crate::disk::remote_io_admission::admit(&wrapper.physical_admission).expect("RPC disk admission");
+                                operation.await;
+                            })
+                            .await;
+                        } else {
+                            let parent = parent.expect("local operation admission");
+                            crate::disk::with_background_resource_set(parent.resources().clone(), operation).await;
+                            parent.complete();
+                        }
                     }
                 });
                 tokio::time::timeout(Duration::from_secs(5), started_rx)

crates/ecstore/src/disk/io_admission.rs

@@ -41,6 +41,9 @@ use tokio::sync::Notify;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
+pub(crate) mod remote;
+pub use remote::{RemoteBackgroundIo, with_remote_background_io};
+
 const BACKGROUND_FOREGROUND_BURST: u32 = 8;
 const BACKGROUND_MAX_WAIT: Duration = Duration::from_millis(100);
 
@@ -180,12 +183,24 @@ pub async fn with_background_resource_set<F: Future>(units: AdaptivePermitSet, f
 }
 
 pub(crate) fn current_resource_unit() -> Option<ResourceIoUnit> {
-    RESOURCE_IO_UNIT.try_with(Clone::clone).ok().or_else(|| {
-        DISK_IO_CONTEXT
-            .try_with(|context| context.resource_unit.clone())
-            .ok()
-            .flatten()
-    })
+    RESOURCE_IO_UNIT
+        .try_with(Clone::clone)
+        .ok()
+        .or_else(|| {
+            DISK_IO_CONTEXT
+                .try_with(|context| context.resource_unit.clone())
+                .ok()
+                .flatten()
+        })
+        .or_else(remote::resources)
+}
+
+pub(crate) fn has_resource_io_scope() -> bool {
+    RESOURCE_IO_UNIT.try_with(|_| ()).is_ok()
+        || DISK_IO_CONTEXT
+            .try_with(|context| context.resource_unit.is_some())
+            .unwrap_or(false)
+        || remote::active()
 }
 
 /// Transfer only resource ownership into an existing spawned task. Disk

crates/ecstore/src/disk/io_admission/remote.rs

@@ -0,0 +1,257 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::{PhysicalDiskBinding, ResourceIoUnit};
+use rustfs_concurrency::adaptive::{NodeAdmission, permit_set::NodeResourcePermit};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::io;
+use std::sync::Arc;
+
+tokio::task_local! {
+    static REMOTE_RESOURCES: RemoteBackgroundIo;
+}
+
+/// Receiver-owned admission for an authenticated background RPC. The owner must
+/// be the same node controller used by local Scanner and Heal work.
+pub struct RemoteBackgroundIo {
+    node: Arc<NodeAdmission>,
+    admitted: RefCell<Option<NodeResourcePermit>>,
+    progressed: Cell<bool>,
+    finished: Cell<bool>,
+}
+
+impl RemoteBackgroundIo {
+    pub fn new(node: Arc<NodeAdmission>) -> Self {
+        Self {
+            node,
+            admitted: RefCell::new(None),
+            progressed: Cell::new(false),
+            finished: Cell::new(false),
+        }
+    }
+
+    /// Move the request's reservations into its existing stream producer.
+    /// Only physical subsets may be cloned into native I/O; never its node unit.
+    pub fn take_current() -> Option<Self> {
+        REMOTE_RESOURCES
+            .try_with(|context| Self {
+                node: context.node.clone(),
+                admitted: RefCell::new(context.admitted.take()),
+                progressed: Cell::new(context.progressed.replace(false)),
+                finished: Cell::new(false),
+            })
+            .ok()
+    }
+
+    /// A failed response or cancelled producer must not credit earlier phases.
+    pub fn discard_current_progress() {
+        let _ = REMOTE_RESOURCES.try_with(|context| context.progressed.set(false));
+    }
+}
+
+impl Drop for RemoteBackgroundIo {
+    fn drop(&mut self) {
+        if self.finished.get()
+            && self.progressed.get()
+            && let Some(permit) = self.admitted.get_mut().take()
+        {
+            permit.complete();
+        }
+    }
+}
+
+/// Scope a verified request or its stream producer. An interrupted request
+/// returns node capacity while native jobs retain their selected physical units.
+pub async fn with_remote_background_io<F: Future>(context: Option<RemoteBackgroundIo>, operation: F) -> F::Output {
+    match context {
+        Some(context) => {
+            REMOTE_RESOURCES
+                .scope(context, async {
+                    let result = operation.await;
+                    REMOTE_RESOURCES.with(|context| context.finished.set(true));
+                    result
+                })
+                .await
+        }
+        None => operation.await,
+    }
+}
+
+pub(crate) fn active() -> bool {
+    REMOTE_RESOURCES.try_with(|_| ()).is_ok()
+}
+
+pub(crate) fn resources() -> Option<ResourceIoUnit> {
+    REMOTE_RESOURCES
+        .try_with(|context| {
+            context
+                .admitted
+                .borrow()
+                .as_ref()
+                .map(|permit| ResourceIoUnit::from(permit.resources().clone()))
+        })
+        .ok()
+        .flatten()
+}
+
+pub(crate) fn record_progress() {
+    let _ = REMOTE_RESOURCES.try_with(|context| context.progressed.set(true));
+}
+
+pub(crate) fn admit(binding: &PhysicalDiskBinding) -> io::Result<()> {
+    REMOTE_RESOURCES
+        .try_with(|context| {
+            let deferred = || io::Error::from(io::ErrorKind::WouldBlock);
+            let plan = binding.plan().ok_or_else(deferred)?;
+            let mut admitted = context.admitted.borrow_mut();
+            if let Some(permit) = admitted.as_ref() {
+                // Reuse the request's receipt for append/read/rename phases at
+                // capacity one. Additional attachments never wait under locks.
+                let native = plan.capture_or_acquire_native(permit.resources()).map_err(|_| deferred())?;
+                drop(native);
+                if !plan.is_current() {
+                    return Err(deferred());
+                }
+            } else {
+                *admitted = Some(plan.try_acquire(&context.node).ok_or_else(deferred)?);
+            }
+            Ok(())
+        })
+        .unwrap_or(Ok(()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rustfs_concurrency::adaptive::{
+        AdaptiveAdmission, FeedbackController, FeedbackObservation, ResourcePressure, permit_set::AdaptivePermitSetOwner,
+    };
+    use std::num::NonZeroU16;
+
+    fn node() -> Arc<NodeAdmission> {
+        let node = NodeAdmission::new(NonZeroU16::MIN);
+        node.update(
+            &mut FeedbackController::new(),
+            Some(FeedbackObservation {
+                generation: 1,
+                max_concurrency: 1,
+                pressure: ResourcePressure::Unknown,
+                completed_units: None,
+            }),
+        );
+        node
+    }
+
+    #[tokio::test]
+    async fn receiver_reuses_one_receipt_across_phases_and_releases_only_native_subsets() {
+        let node = node();
+        let a = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+        let b = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+        let shared = AdaptivePermitSetOwner::new(NonZeroU16::MIN, vec![a.clone(), b.clone()]).expect("device catalog");
+        let first = PhysicalDiskBinding::default();
+        let second = PhysicalDiskBinding::default();
+        first
+            .set_test_shared_capacity(shared.clone(), vec![a.clone()])
+            .expect("device A");
+        second.set_test_shared_capacity(shared, vec![b.clone()]).expect("device B");
+
+        let native = with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async {
+            assert!(resources().is_none(), "no physical work before target selection");
+            admit(&first).expect("first phase");
+            let admitted = resources().expect("admitted resources");
+            let native = first
+                .plan()
+                .expect("current plan")
+                .capture_native(admitted.as_set().expect("resource set"))
+                .expect("escaped native read");
+            admit(&first).expect("second phase reuses capacity while the first reader lives");
+            admit(&second).expect("additional selected device");
+            assert_eq!([node.snapshot().active, a.snapshot().active, b.snapshot().active], [1, 1, 1]);
+            record_progress();
+            native
+        })
+        .await;
+        assert_eq!([node.snapshot().active, a.snapshot().active, b.snapshot().active], [0, 1, 0]);
+        assert_eq!([node.completed(), a.completed(), b.completed()], [Some(1), Some(1), Some(1)]);
+        drop(native);
+        assert_eq!(a.snapshot().active, 0);
+    }
+
+    #[tokio::test]
+    async fn receiver_cancellation_returns_node_but_not_escaped_native_capacity() {
+        let node = node();
+        let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+        let binding = PhysicalDiskBinding::default();
+        binding.set_test_capacity(vec![gate.clone()]).expect("device plan");
+        let (sender, receiver) = tokio::sync::oneshot::channel();
+        let task = tokio::spawn(with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async move {
+            admit(&binding).expect("receiver admission");
+            let admitted = resources().expect("receiver resources");
+            let native = binding
+                .plan()
+                .expect("plan")
+                .capture_native(admitted.as_set().expect("set"))
+                .expect("native subset");
+            sender.send(native).expect("native owner receiver");
+            record_progress();
+            std::future::pending::<()>().await;
+        }));
+        let native = receiver.await.expect("native read ownership");
+        task.abort();
+        assert!(task.await.expect_err("cancel request").is_cancelled());
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 1]);
+        assert_eq!([node.completed(), gate.completed()], [Some(0), Some(0)]);
+        drop(native);
+        assert_eq!(gate.snapshot().active, 0);
+    }
+
+    #[tokio::test]
+    async fn receiver_failed_completion_does_not_credit_successful_preparation() {
+        let node = node();
+        let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+        let binding = PhysicalDiskBinding::default();
+        binding.set_test_capacity(vec![gate.clone()]).expect("device plan");
+        with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async {
+            admit(&binding).expect("receiver admission");
+            record_progress();
+            RemoteBackgroundIo::discard_current_progress();
+        })
+        .await;
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+        assert_eq!([node.completed(), gate.completed()], [Some(0), Some(0)]);
+    }
+
+    #[tokio::test]
+    async fn receiver_stream_transfer_moves_node_ownership_and_defers_unavailable_devices() {
+        let node = node();
+        let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 0);
+        let binding = PhysicalDiskBinding::default();
+        binding.set_test_capacity(vec![gate.clone()]).expect("device plan");
+        let context = with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async {
+            assert_eq!(admit(&binding).expect_err("closed device").kind(), io::ErrorKind::WouldBlock);
+            assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+            gate.set_limit(1);
+            admit(&binding).expect("opened device");
+            let context = RemoteBackgroundIo::take_current().expect("stream context");
+            assert!(resources().is_none(), "request cannot keep a second node owner");
+            context
+        })
+        .await;
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [1, 1]);
+        assert_eq!(node.completed(), Some(0));
+        with_remote_background_io(Some(context), async {
+            admit(&binding).expect("stream reuses the request's receipt");
+            binding
+                .set_test_capacity(vec![AdaptiveAdmission::new(NonZeroU16::MIN, 1)])
+                .expect("replace physical device");
+            assert_eq!(
+                admit(&binding).expect_err("old receipt cannot cross topology").kind(),
+                io::ErrorKind::WouldBlock
+            );
+        })
+        .await;
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+        assert_eq!(node.completed(), Some(0), "admission alone is not completed work");
+    }
+}

crates/ecstore/src/disk/mod.rs

@@ -31,12 +31,14 @@ pub mod os;
 
 #[cfg(test)]
 pub(crate) use io_admission::DiskIoClass;
+pub(crate) use io_admission::remote as remote_io_admission;
 pub use io_admission::{
-    PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan,
+    PhysicalDiskBinding, PhysicalDiskPlan, PhysicalDiskSample, PhysicalDiskSampler, PhysicalResourcePlan, RemoteBackgroundIo,
     spawn_blocking_with_resource_ownership, with_background_disk_io, with_background_resource_set, with_background_resource_unit,
+    with_remote_background_io,
 };
 pub(crate) use io_admission::{
-    PhysicalHealBinding, acquire_context_disk_io_permit, current_disk_io_context, current_resource_unit,
+    PhysicalHealBinding, acquire_context_disk_io_permit, current_disk_io_context, current_resource_unit, has_resource_io_scope,
     with_optional_disk_io_context, with_optional_resource_unit,
 };
 

crates/ecstore/src/store/mod.rs

@@ -456,8 +456,8 @@ pub(crate) mod utils;
 
 use peer::init_local_peer;
 pub use peer::{
-    BootstrapLocalTarget, all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks,
-    init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
+    BootstrapLocalTarget, all_local_disk, all_local_disk_path, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc,
+    get_disk_infos, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
     prewarm_local_disk_id_map_with_instance_ctx,
 };
 

crates/ecstore/src/store/peer.rs

@@ -99,6 +99,9 @@ impl ECStore {
         fi: FileInfo,
         expected: Uuid,
     ) -> DiskResult<()> {
+        if crate::disk::remote_io_admission::active() {
+            local_disk_candidate(&self.ctx, disk_ref).await?;
+        }
         let disk_ref = disk_ref.to_owned();
         let path = target.1.to_owned();
         self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move {
@@ -117,6 +120,9 @@ impl ECStore {
         opts: DeleteOptions,
         expected: Uuid,
     ) -> DiskResult<()> {
+        if crate::disk::remote_io_admission::active() {
+            local_disk_candidate(&self.ctx, disk_ref).await?;
+        }
         let disk_ref = disk_ref.to_owned();
         let path = target.1.to_owned();
         self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move {
@@ -144,6 +150,9 @@ impl ECStore {
         if expected.is_nil() || opts.undo_write || self.ctx.lock_manager().is_disabled() {
             return Err(DiskError::other("invalid incarnation-bound heal deletion"));
         }
+        if crate::disk::remote_io_admission::active() {
+            local_disk_candidate(&self.ctx, disk_ref).await?;
+        }
         let fence = Arc::new(
             self.acquire_bucket_incarnation_fence(target.0, expected)
                 .await
@@ -281,7 +290,9 @@ impl ECStore {
 // The optional ID is a cold lookup to cache only after final admission.
 async fn local_disk_candidate(ctx: &Arc<InstanceContext>, disk_ref: &str) -> DiskResult<(DiskStore, Option<Uuid>)> {
     let map = ctx.local_disk_map();
-    if let Some(disk) = map.read().await.get(disk_ref).and_then(Option::as_ref).cloned() {
+    let path_disk = map.read().await.get(disk_ref).and_then(Option::as_ref).cloned();
+    if let Some(disk) = path_disk {
+        admit_rpc_disk(&disk)?;
         return Ok((disk, None));
     }
     let disk_id = Uuid::parse_str(disk_ref).map_err(|_| DiskError::DiskNotFound)?;
@@ -290,11 +301,16 @@ async fn local_disk_candidate(ctx: &Arc<InstanceContext>, disk_ref: &str) -> Dis
         let cached_disk = map.read().await.get(&path).and_then(Option::as_ref).cloned();
         if let Some(disk) = cached_disk
             && matches!(disk.as_ref(), Disk::Local(_))
-            && disk.get_disk_id().await? == Some(disk_id)
         {
-            return Ok((disk, None));
+            admit_rpc_disk(&disk)?;
+            if disk.get_disk_id().await? == Some(disk_id) {
+                return Ok((disk, None));
+            }
         }
     }
+    if crate::disk::remote_io_admission::active() {
+        return Err(DiskError::Io(std::io::Error::from(std::io::ErrorKind::WouldBlock)));
+    }
     let disks: Vec<_> = map.read().await.values().filter_map(Clone::clone).collect();
     // Disk identity may perform format I/O. No registry guard spans this await.
     for disk in disks {
@@ -305,6 +321,17 @@ async fn local_disk_candidate(ctx: &Arc<InstanceContext>, disk_ref: &str) -> Dis
     Err(DiskError::DiskNotFound)
 }
 
+fn admit_rpc_disk(disk: &DiskStore) -> DiskResult<()> {
+    use crate::disk::remote_io_admission;
+    if remote_io_admission::active() {
+        let binding = disk
+            .physical_admission()
+            .ok_or_else(|| DiskError::Io(std::io::Error::from(std::io::ErrorKind::WouldBlock)))?;
+        remote_io_admission::admit(&binding).map_err(DiskError::Io)?;
+    }
+    Ok(())
+}
+
 async fn admit_local_disk(
     ctx: &Arc<InstanceContext>,
     disk: &DiskStore,
@@ -456,6 +483,18 @@ pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
     None
 }
 
+/// Resolve and admit an authenticated RPC's target before its storage locks.
+/// Unknown UUID mappings defer background work instead of probing unrelated disks.
+pub async fn find_local_disk_by_ref_for_rpc(disk_ref: &str) -> DiskResult<Option<DiskStore>> {
+    use crate::disk::remote_io_admission;
+
+    if !remote_io_admission::active() {
+        return Ok(find_local_disk_by_ref(disk_ref).await);
+    }
+    let (disk, _) = local_disk_candidate(&crate::runtime::global::current_ctx(), disk_ref).await?;
+    Ok(Some(disk))
+}
+
 pub async fn all_local_disk_path() -> Vec<String> {
     runtime_sources::local_disk_paths().await
 }
@@ -816,6 +855,92 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn remote_background_target_lookup_defers_before_io_and_reuses_multi_phase_admission() {
+        use crate::disk::{RemoteBackgroundIo, with_remote_background_io};
+        use rustfs_concurrency::adaptive::{
+            AdaptiveAdmission, FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure,
+        };
+        use std::num::NonZeroU16;
+
+        let ctx = Arc::new(InstanceContext::new());
+        let root = tempfile::tempdir().expect("target directory");
+        let id = Uuid::new_v4();
+        let disk = target_disk(&ctx, root.path(), id).await;
+        disk.make_volume("bucket").await.expect("fixture volume");
+        let gate = AdaptiveAdmission::new(NonZeroU16::MIN, 0);
+        disk.physical_admission()
+            .expect("local binding")
+            .set_test_capacity(vec![gate.clone()])
+            .expect("physical device");
+        let node = NodeAdmission::new(NonZeroU16::MIN);
+        node.update(
+            &mut FeedbackController::new(),
+            Some(FeedbackObservation {
+                generation: 1,
+                max_concurrency: 1,
+                pressure: ResourcePressure::Unknown,
+                completed_units: None,
+            }),
+        );
+        let path = disk.endpoint().to_string();
+        with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async {
+            assert!(
+                matches!(disk.read_all("bucket", "object").await,
+                Err(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock),
+                "no implicit target admission"
+            );
+            assert!(
+                matches!(local_disk_candidate(&ctx, &path).await,
+                Err(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock),
+                "closed receiver capacity"
+            );
+            gate.set_limit(1);
+            assert!(
+                matches!(local_disk_candidate(&ctx, &id.to_string()).await,
+                Err(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock),
+                "cold UUID cannot probe unrelated disks"
+            );
+            assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+            let (selected, _) = local_disk_candidate(&ctx, &path).await.expect("select known local path");
+            assert!(Arc::ptr_eq(&selected, &disk));
+            for body in [b"first phase".as_slice(), b"second phase".as_slice()] {
+                disk.write_all("bucket", "object", bytes::Bytes::copy_from_slice(body))
+                    .await
+                    .expect("write phase");
+                let (again, _) = local_disk_candidate(&ctx, &path)
+                    .await
+                    .expect("repeat target resolution at capacity one");
+                assert_eq!(again.read_all("bucket", "object").await.expect("read phase").as_ref(), body);
+                assert_eq!([node.snapshot().active, gate.snapshot().active], [1, 1]);
+            }
+        })
+        .await;
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+        assert_eq!([node.completed(), gate.completed()], [Some(1), Some(1)]);
+        assert!(record_local_disk_id_if_active(&ctx, &disk, id).await);
+        let replacement = target_disk(&ctx, root.path(), Uuid::new_v4()).await;
+        replacement
+            .physical_admission()
+            .expect("replacement binding")
+            .set_test_capacity(vec![gate.clone()])
+            .expect("replacement device");
+        with_remote_background_io(Some(RemoteBackgroundIo::new(node.clone())), async {
+            assert!(
+                matches!(local_disk_candidate(&ctx, &id.to_string()).await,
+                Err(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock),
+                "stale UUID must not resolve replacement data"
+            );
+        })
+        .await;
+        assert_eq!([node.snapshot().active, gate.snapshot().active], [0, 0]);
+        assert_eq!(
+            [node.completed(), gate.completed()],
+            [Some(1), Some(1)],
+            "identity checks alone are not work completion"
+        );
+    }
+
     #[tokio::test]
     async fn target_uuid_cache_cannot_admit_a_different_format_at_the_same_path() {
         let ctx = Arc::new(InstanceContext::new());

docs/operations/scanner-runtime-controls.md

@@ -500,6 +500,10 @@ The same sampling worker resolves local attachments through Linux sysfs, partiti
 
 Native directory work and the local metadata read retain only that disk's physical leaves after async cancellation. They do not retain the node scheduling unit. Later lifecycle/configuration work in object accounting does not inherit those physical units. Physical owners and carrier slots are preallocated during attachment refresh; object work does not discover devices or allocate a new resource group.
 
+Resource-controlled peer requests carry `x-rustfs-rpc-background-admission: v1`, bound to their HTTP or gRPC signature. Stream retries preserve that classification without retaining the sender's disk quota. The receiver uses its existing shared node owner and admits the selected local physical device before storage locks or I/O. Repeated phases reuse the request's reservation, including when capacity is one. Missing capacity, an unknown UUID mapping, or an unavailable physical plan defers the request instead of probing other disks or reporting object absence. Cancellation returns node capacity; native work retains only its selected physical resources until it finishes.
+
+Enable this experimental setting only after all peers support the admission marker. Older peers reject marked signatures, and a new peer with admission disabled rejects marked work; neither path silently runs it outside resource control. Requests without the marker retain their existing signature and foreground behavior. The receiver reports deferred admission as HTTP 503 or gRPC `RESOURCE_EXHAUSTED`.
+
 Calls without a coordinating cache lease retain immediate resource deferral and useful partial-cache handling. Existing remote single-session ownership and local-disk admission shares remain in force. Disabled admission creates no request channel. Enabled walks use one bounded channel, with a response allocation and cache revision reads only on contention; the uncontended per-entry path adds no global lookup or task. The streaming abandoned-child coordinator requires a separate ownership boundary: retaining a node unit while it waits for a bounded-channel consumer that needs the same unit would deadlock at capacity one.
 
 Foreground comparison uses the existing complete response-body windows and shared admission snapshots. Each active class needs at least 100 completions and a finite p99 interval; this permits the percentile rank but is not a statistical confidence guarantee. Comparisons require the same active classes and rates within 80–125% of a retained reference. A changed workload resets recovery. The reference is an observed workload window, not an independent background-off baseline. A p99 interval can justify stability only when its upper bound is within 5% of the reference lower bound; a lower bound above 105% of the reference upper bound is pressure. Overlapping/uncertain comparisons cannot justify growth. Missing queues, stale or repeated cuts, uncertain cut boundaries and low counts are not idle evidence.

rustfs/src/server/http.rs

@@ -313,6 +313,9 @@ struct RpcRequestPathService<S> {
     inner: S,
 }
 
+#[derive(Clone, Copy)]
+struct RpcBackgroundAdmissionReady;
+
 struct RpcCompletionBody<B> {
     inner: B,
     started_at: Instant,
@@ -424,6 +427,14 @@ where
 
     fn call(&mut self, mut req: HttpRequest<B>) -> Self::Future {
         let started_at = Instant::now();
+        let background = storage::rpc_background_admission(req.headers()).ok() == Some(true);
+        let resource_context = background
+            .then(crate::startup_background::background_node_admission)
+            .flatten()
+            .map(storage::RemoteBackgroundIo::new);
+        if resource_context.is_some() {
+            req.extensions_mut().insert(RpcBackgroundAdmissionReady);
+        }
         let target = RpcRequestTarget {
             uri: req.uri().clone(),
             method: req.method().clone(),
@@ -439,7 +450,17 @@ where
             });
         let future = self.inner.call(req);
         Box::pin(async move {
-            let mut response = future.await?;
+            let mut response = storage::with_remote_background_io(resource_context, async move {
+                let result = future.await;
+                let succeeded = result.as_ref().is_ok_and(|response| {
+                    response.status().is_success() && response.headers().get("grpc-status").is_none_or(|status| status == "0")
+                });
+                if !succeeded {
+                    storage::RemoteBackgroundIo::discard_current_progress();
+                }
+                result
+            })
+            .await?;
             if let Some(headers) = response_headers {
                 response.headers_mut().extend(headers);
             }
@@ -2736,6 +2757,13 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
         }
     })?;
 
+    if storage::rpc_background_admission(req.metadata().as_ref())
+        .map_err(|_| Status::unauthenticated("Invalid background admission request"))?
+        && req.extensions().get::<RpcBackgroundAdmissionReady>().is_none()
+    {
+        return Err(Status::resource_exhausted("background resource admission is not enabled"));
+    }
+
     let parent_context =
         global::get_text_map_propagator(|propagator| propagator.extract(&MetadataMapCarrier::new(req.metadata())));
     if parent_context.has_active_span() {
@@ -4579,6 +4607,264 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn authenticated_background_rpc_enforces_receiver_capacity_on_http_and_grpc() {
+        use rustfs_concurrency::adaptive::{
+            AdaptiveAdmission, FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure,
+        };
+        use std::num::NonZeroU16;
+        use storage::DiskAPI;
+
+        temp_env::async_with_vars(
+            [
+                ("RUSTFS_INTERNODE_PREWARM", Some("false")),
+                ("RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES", Some("0")),
+            ],
+            async {
+                let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string());
+                assert!(
+                    crate::startup_background::background_node_admission().is_none(),
+                    "requires an unpublished owner"
+                );
+                let root = tempfile::tempdir().expect("receiver disk");
+                tokio::fs::create_dir(root.path().join("bucket"))
+                    .await
+                    .expect("fixture bucket");
+                const DATA: &[u8] = b"receiver-owned resource";
+                tokio::fs::write(root.path().join("bucket/object"), DATA)
+                    .await
+                    .expect("fixture object");
+                let listener = TcpListener::bind("127.0.0.1:0").await.expect("receiver listener");
+                let address = listener.local_addr().expect("receiver address");
+                let mut endpoint =
+                    Endpoint::try_from(format!("http://{address}{}", root.path().display()).as_str()).expect("receiver endpoint");
+                endpoint.is_local = true;
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(0);
+                storage::init_local_disks(EndpointServerPools::from(vec![PoolEndpoints {
+                    legacy: false,
+                    set_count: 1,
+                    drives_per_set: 1,
+                    endpoints: Endpoints::from(vec![endpoint.clone()]),
+                    cmd_line: "receiver-admission-test".into(),
+                    platform: "test".into(),
+                }]))
+                .await
+                .expect("register receiver disk");
+                let disk = storage::find_local_disk_by_ref(&endpoint.to_string())
+                    .await
+                    .expect("registered local disk");
+                let physical = AdaptiveAdmission::new(NonZeroU16::MIN, 1);
+                let node_owner = NodeAdmission::new(NonZeroU16::MIN);
+                let sender_owner = NodeAdmission::new(NonZeroU16::MIN);
+                let previous_name = rustfs_common::get_global_local_node_name().await;
+                rustfs_common::set_global_local_node_name(&address.to_string()).await;
+                let node = InterceptedService::new(NodeServiceServer::new(make_server()), check_auth);
+                let fallback = tower::service_fn(|_| async { Ok::<_, Infallible>(Response::new(s3s::Body::empty())) });
+                let service = hybrid(
+                    InternodeRpcService::new(fallback),
+                    RpcRequestPathService::new(Routes::new(node).prepare()),
+                );
+                let server = tokio::spawn(async move {
+                    let mut connections = tokio::task::JoinSet::new();
+                    loop {
+                        let (socket, _) = listener.accept().await.expect("receiver connection");
+                        let service = service.clone();
+                        connections.spawn(async move {
+                            let _ = ConnBuilder::new(TokioExecutor::new())
+                                .serve_connection(TokioIo::new(socket), TowerToHyperService::new(service))
+                                .await;
+                        });
+                    }
+                });
+                let mut remote_endpoint = endpoint.clone();
+                remote_endpoint.is_local = false;
+                let remote = storage::new_disk(
+                    &remote_endpoint,
+                    &storage::DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await
+                .expect("production remote client");
+                let mut controller = FeedbackController::new();
+                for (generation, phase) in ["disabled", "physical_closed", "ready", "node_busy"].into_iter().enumerate() {
+                    if phase == "physical_closed" {
+                        let headers = storage::with_remote_background_io(
+                            Some(storage::RemoteBackgroundIo::new(sender_owner.clone())),
+                            async {
+                                storage::gen_tonic_signature_headers(
+                                    &address.to_string(),
+                                    "node_service.NodeService",
+                                    "ReadAll",
+                                    None,
+                                )
+                            },
+                        )
+                        .await
+                        .expect("early request signature");
+                        let mut request = HttpRequest::builder()
+                            .method(Method::POST)
+                            .uri(format!("http://{address}/node_service.NodeService/ReadAll"))
+                            .body(())
+                            .expect("early request");
+                        request.headers_mut().extend(headers);
+                        let mut early_service =
+                            RpcRequestPathService::new(tower::service_fn(|request: HttpRequest<()>| async move {
+                                let (parts, ()) = request.into_parts();
+                                let mut request = Request::new(());
+                                request.metadata_mut().as_mut().extend(parts.headers);
+                                *request.extensions_mut() = parts.extensions;
+                                assert_eq!(
+                                    check_auth(request)
+                                        .expect_err("request entered before admission was enabled")
+                                        .code(),
+                                    tonic::Code::ResourceExhausted
+                                );
+                                Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
+                            }));
+                        let early_request = early_service.call(request);
+                        crate::startup_background::install_background_node_admission_for_test(node_owner.clone());
+                        early_request.await.expect("startup race must fail closed");
+                    }
+                    node_owner.update(
+                        &mut controller,
+                        Some(FeedbackObservation {
+                            generation: u64::try_from(generation + 1).expect("small generation"),
+                            max_concurrency: 1,
+                            pressure: ResourcePressure::Unknown,
+                            completed_units: None,
+                        }),
+                    );
+                    physical.set_limit(if phase == "physical_closed" { 0 } else { 1 });
+                    disk.physical_admission()
+                        .expect("physical binding")
+                        .set_test_capacity(vec![physical.clone()])
+                        .expect("renew receiver device plan");
+                    let held = (phase == "node_busy").then(|| node_owner.try_acquire().expect("occupy receiver node"));
+                    for background in [false, true] {
+                        for stream in [false, true] {
+                            let operation = async {
+                                if stream {
+                                    let mut reader = remote.read_file_stream("bucket", "object", 0, DATA.len()).await?;
+                                    let mut bytes = Vec::new();
+                                    reader.read_to_end(&mut bytes).await?;
+                                    Ok(bytes)
+                                } else {
+                                    remote.read_all("bucket", "object").await.map(|bytes| bytes.to_vec())
+                                }
+                            };
+                            let result = tokio::time::timeout(
+                                Duration::from_secs(5),
+                                storage::with_remote_background_io(
+                                    background.then(|| storage::RemoteBackgroundIo::new(sender_owner.clone())),
+                                    operation,
+                                ),
+                            )
+                            .await
+                            .expect("receiver must defer without waiting under storage locks");
+                            if !background || phase == "ready" {
+                                assert_eq!(
+                                    result.unwrap_or_else(|error| panic!(
+                                        "{phase}, background={background}, stream={stream}: {error}"
+                                    )),
+                                    DATA,
+                                    "{phase}, background={background}, stream={stream}"
+                                );
+                            } else {
+                                assert!(
+                                    result.is_err(),
+                                    "{phase}, stream={stream}: background request bypassed receiver admission"
+                                );
+                            }
+                            assert_eq!(node_owner.snapshot().active, u32::from(held.is_some()));
+                            tokio::time::timeout(Duration::from_secs(1), async {
+                                while physical.snapshot().active != 0 {
+                                    tokio::task::yield_now().await;
+                                }
+                            })
+                            .await
+                            .expect("receiver native I/O and response body must drain");
+                            assert_eq!(physical.snapshot().active, 0, "completed response releases receiver device");
+                        }
+                    }
+                    drop(held);
+                }
+                assert_eq!(node_owner.completed(), Some(2), "only the two admitted background requests made progress");
+                assert_eq!(physical.completed(), Some(2));
+                let mut upload =
+                    reqwest::Url::parse(&format!("http://{address}/rustfs/rpc/put_file_stream")).expect("upload URL");
+                upload.query_pairs_mut().extend_pairs([
+                    ("disk", endpoint.to_string().as_str()),
+                    ("volume", "bucket"),
+                    ("path", "incomplete"),
+                    ("append", "false"),
+                    ("size", "4"),
+                ]);
+                let headers =
+                    storage::with_remote_background_io(Some(storage::RemoteBackgroundIo::new(sender_owner.clone())), async {
+                        storage::gen_signature_headers(upload.as_str(), &Method::PUT)
+                    })
+                    .await
+                    .expect("background upload signature");
+                let response = reqwest::Client::builder()
+                    .no_proxy()
+                    .timeout(Duration::from_secs(5))
+                    .build()
+                    .expect("upload client")
+                    .put(upload)
+                    .headers(headers)
+                    .body("bad")
+                    .send()
+                    .await
+                    .expect("incomplete upload response");
+                assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+                let message = response.text().await.expect("upload error body");
+                assert!(message.contains("body size mismatch"), "unexpected upload failure: {message}");
+                assert!(root.path().join("bucket/incomplete").exists(), "file creation completed before failure");
+                assert_eq!(node_owner.completed(), Some(2), "failed upload must not count its successful preparation");
+                assert_eq!(physical.completed(), Some(2));
+                let large = tokio::fs::File::create(root.path().join("bucket/large"))
+                    .await
+                    .expect("large response fixture");
+                const LARGE_LENGTH: usize = 64 * 1024 * 1024;
+                large
+                    .set_len(u64::try_from(LARGE_LENGTH).expect("fixture length"))
+                    .await
+                    .expect("sparse response size");
+                drop(large);
+                let mut reader = storage::with_remote_background_io(
+                    Some(storage::RemoteBackgroundIo::new(sender_owner.clone())),
+                    remote.read_file_stream("bucket", "large", 0, LARGE_LENGTH),
+                )
+                .await
+                .expect("admitted streaming response");
+                let mut first_byte = [1];
+                reader.read_exact(&mut first_byte).await.expect("response body started");
+                assert_eq!(first_byte, [0]);
+                assert_eq!(node_owner.snapshot().active, 0, "stream body cannot retain the handler's node unit");
+                assert_eq!(physical.snapshot().active, 1, "escaped response keeps its receiver disk capacity");
+                drop(reader);
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while physical.snapshot().active != 0 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("disconnect must retire the receiver stream and native reads");
+                remote.close().await.expect("close remote client");
+                disk.close().await.expect("stop receiver disk monitors");
+                server.abort();
+                assert!(server.await.expect_err("stop fixture server").is_cancelled());
+                rustfs_common::set_global_local_node_name(&previous_name).await;
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn rpc_auth_read_all_recovers_after_stale_boot_epoch() {

rustfs/src/startup_background.rs

@@ -27,7 +27,10 @@ use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use rustfs_heal::{
     create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager_with_workload_provider,
 };
-use std::{io::Result, sync::Arc};
+use std::{
+    io::Result,
+    sync::{Arc, OnceLock},
+};
 use tracing::{debug, info};
 
 const LOG_COMPONENT_MAIN: &str = "main";
@@ -35,6 +38,20 @@ const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
 const EVENT_ODM_BACKFILL_RECOVERY_CONFIGURED: &str = "odm_backfill_recovery_configured";
 
+static BACKGROUND_NODE_ADMISSION: OnceLock<Arc<rustfs_concurrency::adaptive::NodeAdmission>> = OnceLock::new();
+
+pub(crate) fn background_node_admission() -> Option<Arc<rustfs_concurrency::adaptive::NodeAdmission>> {
+    BACKGROUND_NODE_ADMISSION.get().cloned()
+}
+
+#[cfg(test)]
+pub(crate) fn install_background_node_admission_for_test(owner: Arc<rustfs_concurrency::adaptive::NodeAdmission>) {
+    assert!(
+        BACKGROUND_NODE_ADMISSION.set(owner).is_ok(),
+        "test requires an unpublished background owner"
+    );
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 struct BackgroundServicePlan {
     scanner_enabled: bool,
@@ -81,13 +98,15 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
         let admission = None;
         let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()).with_admission(admission.clone()));
         init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider.clone())).await?;
-        if let Some(admission) = &admission
-            && !rustfs_scanner::set_scanner_node_admission(admission.clone())
-        {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::AlreadyExists,
-                "scanner node admission is already initialized",
-            ));
+        if let Some(admission) = &admission {
+            if BACKGROUND_NODE_ADMISSION.set(admission.clone()).is_err()
+                || !rustfs_scanner::set_scanner_node_admission(admission.clone())
+            {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::AlreadyExists,
+                    "background node admission is already initialized",
+                ));
+            }
         }
         #[cfg(any(target_os = "linux", all(test, unix)))]
         if let Some(admission) = admission {

rustfs/src/storage/rpc/http_service.rs

@@ -16,10 +16,10 @@ use crate::server::RPC_PREFIX;
 use crate::storage::request_context::spawn_traced;
 use crate::storage::storage_api::rpc_consumer::http_service::{
     DEFAULT_READ_BUFFER_SIZE, DeleteOptions, DiskStore, NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse,
-    PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, StorageDiskRpcExt as _,
-    WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, check_and_record_signed_rpc_nonce, find_local_disk_by_ref,
-    sign_ns_scanner_capability_with_checkpoint_coverage, sign_put_file_capability, verify_put_file_auth_trailer,
-    verify_rpc_signature,
+    PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, RemoteBackgroundIo,
+    StorageDiskRpcExt as _, WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, check_and_record_signed_rpc_nonce,
+    find_local_disk_by_ref_for_rpc, rpc_background_admission, sign_ns_scanner_capability_with_checkpoint_coverage,
+    sign_put_file_capability, verify_put_file_auth_trailer, verify_rpc_signature, with_remote_background_io,
 };
 #[cfg(test)]
 use crate::storage::storage_api::rpc_consumer::http_service::{
@@ -424,39 +424,53 @@ async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
         return response;
     }
 
-    let method = req.method().clone();
-    let path = req.uri().path();
-
-    let response = match (method, path) {
-        (Method::GET, READ_FILE_STREAM_PATH) | (Method::HEAD, READ_FILE_STREAM_PATH) => handle_read_file(req).await,
-        (Method::GET, WALK_DIR_PATH) | (Method::HEAD, WALK_DIR_PATH) => handle_walk_dir(req).await,
-        (Method::GET, NS_SCANNER_PATH) => match parse_query::<NsScannerCapabilityQuery>(&req) {
-            Ok(query) if query.ns_scanner_protocol == Some(NS_SCANNER_PROTOCOL_VERSION) => match query.ns_scanner_challenge {
-                Some(challenge) if !challenge.is_nil() => ns_scanner_capability_response(
-                    challenge,
-                    query.ns_scanner_tier_registry_generation == Some(true),
-                    query.ns_scanner_checkpoint_coverage == Some(true),
-                ),
-                Some(_) | None => response_with_status(StatusCode::BAD_REQUEST, "namespace scanner challenge is invalid"),
-            },
-            Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "namespace scanner protocol is unsupported"),
-            Err(response) => *response,
+    let resource_context = match rpc_background_admission(req.headers()) {
+        Ok(false) => None,
+        Ok(true) => match crate::startup_background::background_node_admission() {
+            Some(owner) => Some(RemoteBackgroundIo::new(owner)),
+            None => return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "background resource admission is not enabled"),
         },
-        (Method::POST, NS_SCANNER_PATH) => handle_ns_scanner(req).await,
-        (Method::GET, PUT_FILE_CAPABILITY_PATH) => match parse_query::<PutFileCapabilityQuery>(&req) {
-            Ok(query) if query.put_file_capability == Some(PUT_FILE_CAPABILITY_VERSION) => {
-                match query.put_file_challenge.filter(|challenge| !challenge.is_nil()) {
-                    Some(challenge) => put_file_capability_response(challenge),
-                    None => response_with_status(StatusCode::BAD_REQUEST, "put_file capability challenge is invalid"),
-                }
-            }
-            Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "put_file capability is unsupported"),
-            Err(response) => *response,
-        },
-        (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req, false).await,
-        (Method::PUT, PUT_FILE_AUTH_STREAM_PATH) => handle_put_file(req, true).await,
-        _ => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"),
+        Err(_) => return response_with_status(StatusCode::FORBIDDEN, "invalid background admission request"),
     };
+    let response = with_remote_background_io(resource_context, async move {
+        let method = req.method().clone();
+        let path = req.uri().path();
+        let response = match (method, path) {
+            (Method::GET, READ_FILE_STREAM_PATH) | (Method::HEAD, READ_FILE_STREAM_PATH) => handle_read_file(req).await,
+            (Method::GET, WALK_DIR_PATH) | (Method::HEAD, WALK_DIR_PATH) => handle_walk_dir(req).await,
+            (Method::GET, NS_SCANNER_PATH) => match parse_query::<NsScannerCapabilityQuery>(&req) {
+                Ok(query) if query.ns_scanner_protocol == Some(NS_SCANNER_PROTOCOL_VERSION) => match query.ns_scanner_challenge {
+                    Some(challenge) if !challenge.is_nil() => ns_scanner_capability_response(
+                        challenge,
+                        query.ns_scanner_tier_registry_generation == Some(true),
+                        query.ns_scanner_checkpoint_coverage == Some(true),
+                    ),
+                    Some(_) | None => response_with_status(StatusCode::BAD_REQUEST, "namespace scanner challenge is invalid"),
+                },
+                Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "namespace scanner protocol is unsupported"),
+                Err(response) => *response,
+            },
+            (Method::POST, NS_SCANNER_PATH) => handle_ns_scanner(req).await,
+            (Method::GET, PUT_FILE_CAPABILITY_PATH) => match parse_query::<PutFileCapabilityQuery>(&req) {
+                Ok(query) if query.put_file_capability == Some(PUT_FILE_CAPABILITY_VERSION) => {
+                    match query.put_file_challenge.filter(|challenge| !challenge.is_nil()) {
+                        Some(challenge) => put_file_capability_response(challenge),
+                        None => response_with_status(StatusCode::BAD_REQUEST, "put_file capability challenge is invalid"),
+                    }
+                }
+                Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "put_file capability is unsupported"),
+                Err(response) => *response,
+            },
+            (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req, false).await,
+            (Method::PUT, PUT_FILE_AUTH_STREAM_PATH) => handle_put_file(req, true).await,
+            _ => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"),
+        };
+        if !response.status().is_success() {
+            RemoteBackgroundIo::discard_current_progress();
+        }
+        response
+    })
+    .await;
 
     if !response.status().is_success() {
         record_internode_rpc_error(operation);
@@ -638,7 +652,11 @@ async fn handle_read_file(req: Request<Incoming>) -> Response<Body> {
         Err(response) => return *response,
     };
 
-    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+    let disk = match find_local_disk_by_ref_for_rpc(&query.disk).await {
+        Ok(disk) => disk,
+        Err(_) => return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "background storage admission deferred"),
+    };
+    let Some(disk) = disk else {
         warn!(
             event = EVENT_RPC_REQUEST_REJECTED,
             component = LOG_COMPONENT_INTERNODE_RPC,
@@ -748,7 +766,11 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
         Ok(query) => query,
         Err(response) => return *response,
     };
-    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+    let disk = match find_local_disk_by_ref_for_rpc(&query.disk).await {
+        Ok(disk) => disk,
+        Err(_) => return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "background storage admission deferred"),
+    };
+    let Some(disk) = disk else {
         warn!(
             event = EVENT_RPC_REQUEST_REJECTED,
             component = LOG_COMPONENT_INTERNODE_RPC,
@@ -950,7 +972,11 @@ async fn handle_ns_scanner(req: Request<Incoming>) -> Response<Body> {
         );
         return response_with_status(StatusCode::CONFLICT, "namespace scanner server epoch is stale");
     }
-    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+    let disk = match find_local_disk_by_ref_for_rpc(&query.disk).await {
+        Ok(disk) => disk,
+        Err(_) => return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "background storage admission deferred"),
+    };
+    let Some(disk) = disk else {
         warn!(
             event = EVENT_RPC_REQUEST_REJECTED,
             component = LOG_COMPONENT_INTERNODE_RPC,
@@ -1239,15 +1265,19 @@ where
 {
     let (reader, writer) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
     let (mut completion_tx, completion_rx) = oneshot::channel();
-    spawn_traced(async move {
+    let resource_context = RemoteBackgroundIo::take_current();
+    spawn_traced(with_remote_background_io(resource_context, async move {
         tokio::select! {
             biased;
             result = producer(writer) => {
+                if result.is_err() {
+                    RemoteBackgroundIo::discard_current_progress();
+                }
                 let _ = completion_tx.send(result);
             }
-            _ = completion_tx.closed() => {}
+            _ = completion_tx.closed() => RemoteBackgroundIo::discard_current_progress(),
         }
-    });
+    }));
 
     let metrics = runtime_sources::current_internode_metrics();
     let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
@@ -1325,6 +1355,9 @@ where
     let (reader, writer) = tokio::io::duplex(NS_SCANNER_STREAM_BUFFER_SIZE);
     let (mut completion_tx, completion_rx) = oneshot::channel();
     let disconnect = CancellationToken::new();
+    // Scanner reacquires each operation outside its cache lease. Carrying the
+    // preflight node unit into that scheduler would deadlock at capacity one.
+    drop(RemoteBackgroundIo::take_current());
     spawn_traced(async move {
         let producer = producer(writer, disconnect.clone());
         tokio::pin!(producer);
@@ -1514,7 +1547,11 @@ async fn handle_put_file(req: Request<Incoming>, require_auth: bool) -> Response
         return response_with_status(StatusCode::FORBIDDEN, format!("invalid put_file auth: {e}"));
     }
 
-    let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
+    let disk = match find_local_disk_by_ref_for_rpc(&query.disk).await {
+        Ok(disk) => disk,
+        Err(_) => return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "background storage admission deferred"),
+    };
+    let Some(disk) = disk else {
         log_internode_rpc_response_failure!(
             StatusCode::BAD_REQUEST,
             &path,

rustfs/src/storage/rpc/node_service.rs

@@ -21,12 +21,13 @@ use crate::server::MODULE_SWITCHES_SIGNAL_SUBSYSTEM;
 use crate::storage::storage_api::ecstore_tier::tier_mutation_peer::{self, TierMutationPeerState as EcTierMutationPeerState};
 use crate::storage::storage_api::rpc_consumer::node_service::STORAGE_CLASS_SUB_SYS;
 #[cfg(test)]
-use crate::storage::storage_api::rpc_consumer::node_service::{CollectMetricsOpts, MetricType};
+use crate::storage::storage_api::rpc_consumer::node_service::{CollectMetricsOpts, MetricType, find_local_disk_by_ref};
 use crate::storage::storage_api::rpc_consumer::node_service::{
     DiskStore, ECStore, Error, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
-    SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION,
-    SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _,
-    StorageResult, all_local_disk_path, find_local_disk_by_ref, reload_transition_tier_config,
+    RemoteBackgroundIo, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION,
+    SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG,
+    SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StorageResult, all_local_disk_path, find_local_disk_by_ref_for_rpc,
+    reload_transition_tier_config, with_remote_background_io,
 };
 use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, ServerContextSlot, runtime_sources};
 use crate::storage::storage_api::{
@@ -1326,8 +1327,10 @@ impl NodeService {
         runtime_sources::current_object_store_handle_for_context(context.as_deref())
     }
 
-    async fn find_disk(&self, disk_path: &str) -> Option<DiskStore> {
-        find_local_disk_by_ref(disk_path).await
+    async fn find_disk(&self, disk_path: &str) -> Result<Option<DiskStore>, Status> {
+        find_local_disk_by_ref_for_rpc(disk_path)
+            .await
+            .map_err(|_| Status::resource_exhausted("background storage admission deferred"))
     }
 
     async fn all_disk(&self) -> Vec<String> {
@@ -1589,7 +1592,7 @@ impl Node for NodeService {
     async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<Self::WalkDirStream>, Status> {
         let request = request.into_inner();
         let (tx, rx) = mpsc::channel(128);
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let mut buf = Deserializer::new(Cursor::new(request.walk_dir_options));
             let opts = match Deserialize::deserialize(&mut buf) {
                 Ok(options) => options,
@@ -1597,10 +1600,12 @@ impl Node for NodeService {
                     return Err(Status::invalid_argument("invalid WalkDirOptions"));
                 }
             };
-            spawn(async {
+            let resource_context = RemoteBackgroundIo::take_current();
+            spawn(async move {
                 let (rd, mut wr) = tokio::io::duplex(64);
-                let job1 = spawn(async move {
+                let job1 = spawn(with_remote_background_io(resource_context, async move {
                     if let Err(err) = disk.walk_dir(opts, &mut wr).await {
+                        RemoteBackgroundIo::discard_current_progress();
                         error!(
                             component = LOG_COMPONENT_STORAGE,
                             subsystem = LOG_SUBSYSTEM_RPC,
@@ -1609,7 +1614,7 @@ impl Node for NodeService {
                             "walk_dir RPC failed"
                         );
                     }
-                });
+                }));
                 let job2 = spawn(async move {
                     let mut reader = MetacacheReader::new(rd);
 
@@ -6777,6 +6782,7 @@ mod tests {
         let disk = service
             .find_disk(&configured.endpoint().to_string())
             .await
+            .expect("RPC admission")
             .expect("RPC target disk");
         let volume = "evidence-rpc";
         disk.make_volume(volume).await.expect("create evidence bucket");
@@ -6831,6 +6837,7 @@ mod tests {
         let disk = service
             .find_disk(&configured.endpoint().to_string())
             .await
+            .expect("RPC admission")
             .expect("RPC target disk");
         let volume = "orphan-rpc";
         disk.make_volume(volume).await.expect("create orphan bucket");
@@ -7083,6 +7090,7 @@ mod tests {
         let local = service
             .find_disk(&remote_endpoint.to_string())
             .await
+            .expect("RPC admission")
             .expect("registered production local disk");
         let bucket = "guarded-delete-rpc";
         local.make_volume(bucket).await.expect("create guarded fixture volume");
@@ -7268,6 +7276,7 @@ mod tests {
         let survivor = service
             .find_disk(&survivor_endpoint.to_string())
             .await
+            .expect("RPC admission")
             .expect("registered survivor disk");
         if let Err(error) = survivor
             .make_volume(crate::storage::storage_api::ecstore_disk::RUSTFS_META_BUCKET)
@@ -7278,6 +7287,7 @@ mod tests {
         let replacement = service
             .find_disk(&endpoints[1].to_string())
             .await
+            .expect("RPC admission")
             .expect("registered replacement disk");
         let (shutdown, stopped) = tokio::sync::oneshot::channel();
         let server = tokio::spawn(async move {
@@ -7815,7 +7825,13 @@ mod tests {
         let service = make_server_for_context(Some(fixture.context.clone()));
         let mut local_disks = Vec::new();
         for endpoint in &endpoints {
-            local_disks.push(service.find_disk(&endpoint.to_string()).await.expect("registered disk"));
+            local_disks.push(
+                service
+                    .find_disk(&endpoint.to_string())
+                    .await
+                    .expect("RPC admission")
+                    .expect("registered disk"),
+            );
         }
         let (shutdown, stopped) = tokio::sync::oneshot::channel();
         let server = tokio::spawn(async move {
@@ -10971,7 +10987,7 @@ mod tests {
     #[tokio::test]
     async fn test_find_disk_method() {
         let service = create_test_node_service();
-        let disk = service.find_disk("non-existent-disk").await;
+        let disk = service.find_disk("non-existent-disk").await.expect("RPC admission");
         // Should return None for non-existent disk
         assert!(disk.is_none());
     }

rustfs/src/storage/rpc/node_service/disk.rs

@@ -480,7 +480,7 @@ impl NodeService {
             Some(store) => store.quota_mutation_fence_disk(&request.disk).await,
             None => self
                 .find_disk(&request.disk)
-                .await
+                .await?
                 .ok_or_else(|| DiskError::other("cannot find disk")),
         };
         let result = match disk {
@@ -502,7 +502,7 @@ impl NodeService {
         let request = request.into_inner();
         let token =
             SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
-        let result = match self.find_disk(&request.disk).await {
+        let result = match self.find_disk(&request.disk).await? {
             Some(disk) => disk.renew_snapshot_lease(&request.volume, &request.path, token).await,
             None => Err(DiskError::other("cannot find disk")),
         };
@@ -535,7 +535,7 @@ impl NodeService {
             Some(store) => store.quota_mutation_fence_disk(&request.disk).await,
             None => self
                 .find_disk(&request.disk)
-                .await
+                .await?
                 .ok_or_else(|| DiskError::other("cannot find disk")),
         };
         let result = match disk {
@@ -556,7 +556,7 @@ impl NodeService {
 
     pub(super) async fn handle_disk_info(&self, request: Request<DiskInfoRequest>) -> Result<Response<DiskInfoResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let opts = match serde_json::from_str::<DiskInfoOptions>(&request.opts) {
                 Ok(opts) => opts,
                 Err(err) => {
@@ -605,7 +605,7 @@ impl NodeService {
             "delete_volume",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.delete_volume(&request.volume, request.force).await {
                 Ok(_) => Ok(Response::new(DeleteVolumeResponse {
                     success: true,
@@ -629,7 +629,7 @@ impl NodeService {
         request: Request<ReadMultipleRequest>,
     ) -> Result<Response<ReadMultipleResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let read_multiple_req = match decode_msgpack_or_json::<ReadMultipleReq>(
                 &request.read_multiple_req_bin,
                 &request.read_multiple_req,
@@ -706,7 +706,7 @@ impl NodeService {
                     .saturating_add(request.batch_read_version_req_bin.len()),
             );
         }
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let decode_started = internode_stage_timer(attribution_enabled);
             let decoded_batch_read_version_req: DecodedRpcPayload<BatchReadVersionReq> = match decode_msgpack_or_json_with_source(
                 &request.batch_read_version_req_bin,
@@ -798,7 +798,7 @@ impl NodeService {
         let request = request.into_inner();
         let disk = self
             .find_disk(&request.disk)
-            .await
+            .await?
             .ok_or_else(|| Status::not_found("cannot find disk"))?;
         if request.probe_only {
             if !request.payload.is_empty() || !request.volume.is_empty() {
@@ -846,7 +846,7 @@ impl NodeService {
             "delete_versions",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let mut versions = Vec::with_capacity(request.versions.len());
             for (index, version) in request.versions.iter().enumerate() {
                 let version_bin = request.versions_bin.get(index).map(|b| b.as_ref()).unwrap_or(&[]);
@@ -998,7 +998,7 @@ impl NodeService {
                     .undo_local_write(&request.disk, &request.volume, &request.path, file_info, opts)
                     .await
             }
-        } else if let Some(disk) = self.find_disk(&request.disk).await {
+        } else if let Some(disk) = self.find_disk(&request.disk).await? {
             disk.delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts)
                 .await
         } else {
@@ -1027,7 +1027,7 @@ impl NodeService {
 
     pub(super) async fn handle_read_xl(&self, request: Request<ReadXlRequest>) -> Result<Response<ReadXlResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.read_xl(&request.volume, &request.path, request.read_data).await {
                 Ok(raw_file_info) => {
                     let raw_file_info_json = compat_response_json(&raw_file_info, false);
@@ -1095,7 +1095,7 @@ impl NodeService {
                     .saturating_add(request.opts_bin.len()),
             );
         }
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let request_had_msgpack_payload = !request.opts_bin.is_empty();
             let decode_started = internode_stage_timer(read_version_attribution_enabled);
             let opts = match decode_msgpack_or_json::<ReadOptions>(&request.opts_bin, &request.opts, "ReadOptions") {
@@ -1238,7 +1238,7 @@ impl NodeService {
                 error: result.err().map(Into::into),
             }));
         }
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
                 Ok(file_info) => file_info,
                 Err(err) => {
@@ -1276,7 +1276,7 @@ impl NodeService {
             "update_metadata",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
                 Ok(file_info) => file_info,
                 Err(err) => {
@@ -1320,7 +1320,7 @@ impl NodeService {
         request: Request<ReadMetadataRequest>,
     ) -> Result<Response<ReadMetadataResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.read_metadata(&request.volume, &request.path).await {
                 Ok(data) => Ok(Response::new(ReadMetadataResponse {
                     success: true,
@@ -1352,7 +1352,7 @@ impl NodeService {
             "delete_paths",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.delete_paths(&request.volume, &request.paths).await {
                 Ok(_) => Ok(Response::new(DeletePathsResponse {
                     success: true,
@@ -1376,7 +1376,7 @@ impl NodeService {
         request: Request<StatVolumeRequest>,
     ) -> Result<Response<StatVolumeResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.stat_volume(&request.volume).await {
                 Ok(volume_info) => match serde_json::to_string(&volume_info) {
                     Ok(volume_info) => Ok(Response::new(StatVolumeResponse {
@@ -1410,7 +1410,7 @@ impl NodeService {
         request: Request<ListVolumesRequest>,
     ) -> Result<Response<ListVolumesResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.list_volumes().await {
                 Ok(volume_infos) => {
                     let volume_infos = volume_infos
@@ -1459,7 +1459,7 @@ impl NodeService {
             "make_volume",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.make_volume(&request.volume).await {
                 Ok(_) => Ok(Response::new(MakeVolumeResponse {
                     success: true,
@@ -1488,7 +1488,7 @@ impl NodeService {
             "make_volumes",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.make_volumes(request.volumes.iter().map(|s| &**s).collect()).await {
                 Ok(_) => Ok(Response::new(MakeVolumesResponse {
                     success: true,
@@ -1715,7 +1715,7 @@ impl NodeService {
 
     pub(super) async fn handle_list_dir(&self, request: Request<ListDirRequest>) -> Result<Response<ListDirResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.list_dir("", &request.volume, &request.dir_path, request.count).await {
                 Ok(volumes) => Ok(Response::new(ListDirResponse {
                     success: true,
@@ -1751,7 +1751,7 @@ impl NodeService {
             "rename_file",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let result = if request.durable {
                 disk.rename_file_durable(&request.src_volume, &request.src_path, &request.dst_volume, &request.dst_path)
                     .await
@@ -1790,7 +1790,7 @@ impl NodeService {
             "rename_part",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk
                 .rename_part(
                     &request.src_volume,
@@ -1828,7 +1828,7 @@ impl NodeService {
             "prepare_part_transaction",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk
                 .prepare_part_transaction(
                     &request.src_volume,
@@ -1921,7 +1921,7 @@ impl NodeService {
             }
             _ => {}
         }
-        let result = if let Some(disk) = self.find_disk(&request.disk).await {
+        let result = if let Some(disk) = self.find_disk(&request.disk).await? {
             match operation {
                 DeleteRecoveryRpcOperation::Exists { object } => disk
                     .has_delete_transaction_evidence(&request.volume, &object)
@@ -2000,7 +2000,7 @@ impl NodeService {
             "settle_part_transaction",
         )?;
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let action = if request.rollback {
                 PartTransactionAction::Rollback
             } else {
@@ -2029,7 +2029,7 @@ impl NodeService {
         request: Request<CheckPartsRequest>,
     ) -> Result<Response<CheckPartsResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
                 Ok(file_info) => file_info,
                 Err(err) => {
@@ -2078,7 +2078,7 @@ impl NodeService {
         request: Request<ReadPartsRequest>,
     ) -> Result<Response<ReadPartsResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.read_parts(&request.bucket, &request.paths).await {
                 Ok(data) => {
                     let data = match rmp_serde::to_vec(&data) {
@@ -2118,7 +2118,7 @@ impl NodeService {
         request: Request<VerifyFileRequest>,
     ) -> Result<Response<VerifyFileResponse>, Status> {
         let request = request.into_inner();
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
                 Ok(file_info) => file_info,
                 Err(err) => {
@@ -2196,7 +2196,7 @@ impl NodeService {
                 error: result.err().map(Into::into),
             }));
         }
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             let options = match serde_json::from_str::<DeleteOptions>(&request.options) {
                 Ok(options) => options,
                 Err(err) => {
@@ -2280,7 +2280,7 @@ impl NodeService {
             INTERNODE_TRANSPORT_BACKEND_GRPC,
             data_len,
         );
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.write_all(&request.volume, &request.path, request.data).await {
                 Ok(_) => Ok(Response::new(WriteAllResponse {
                     success: true,
@@ -2331,7 +2331,7 @@ impl NodeService {
             INTERNODE_TRANSPORT_BACKEND_GRPC,
             data_len,
         );
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk
                 .compare_and_update_file(&request.volume, &request.path, request.expected, request.replacement)
                 .await
@@ -2379,7 +2379,7 @@ impl NodeService {
             INTERNODE_OPERATION_GRPC_READ_ALL,
             INTERNODE_TRANSPORT_BACKEND_GRPC,
         );
-        if let Some(disk) = self.find_disk(&request.disk).await {
+        if let Some(disk) = self.find_disk(&request.disk).await? {
             match disk.read_all(&request.volume, &request.path).await {
                 Ok(data) => {
                     metrics.record_sent_bytes_for_operation_and_backend(

rustfs/src/storage/storage_api.rs

@@ -269,9 +269,9 @@ pub(crate) mod rpc_consumer {
             PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, WALK_DIR_STREAM_COMPLETION_V1,
         };
         pub(crate) use super::super::{
-            DeleteOptions, DiskStore, StorageDiskRpcExt, WalkDirOptions, check_and_record_signed_rpc_nonce,
-            find_local_disk_by_ref, sign_ns_scanner_capability_with_checkpoint_coverage, sign_put_file_capability,
-            verify_put_file_auth_trailer, verify_rpc_signature,
+            DeleteOptions, DiskStore, RemoteBackgroundIo, StorageDiskRpcExt, WalkDirOptions, check_and_record_signed_rpc_nonce,
+            find_local_disk_by_ref_for_rpc, rpc_background_admission, sign_ns_scanner_capability_with_checkpoint_coverage,
+            sign_put_file_capability, verify_put_file_auth_trailer, verify_rpc_signature, with_remote_background_io,
         };
     }
 
@@ -285,11 +285,11 @@ pub(crate) mod rpc_consumer {
             BatchReadVersionReq, BatchReadVersionResp, CollectMetricsOpts, ConditionalFileUpdate, DeleteOptions, DiskError,
             DiskInfoOptions, DiskStore, ECStore, Error, FileInfoVersions, GuardedDeleteVersionsPayload, KMS_SIGNAL_SUBSYSTEM,
             LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq,
-            ReadMultipleResp, ReadOptions, SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG,
+            ReadMultipleResp, ReadOptions, RemoteBackgroundIo, SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG,
             SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt, StoragePeerS3ClientExt, TierDailyStatsWire, UpdateMetadataOpts,
-            all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, get_global_transition_state,
-            get_local_server_property, reload_bucket_metadata, reload_transition_tier_config, remove_bucket_metadata_if_absent,
-            validate_batch_read_version_item_count,
+            all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc,
+            get_global_transition_state, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config,
+            remove_bucket_metadata_if_absent, validate_batch_read_version_item_count, with_remote_background_io,
         };
         pub(crate) type StorageResult<T> = super::super::Result<T>;
 
@@ -498,8 +498,9 @@ pub(crate) mod ecstore_disk {
         DeleteRecoveryInspection, DeleteRecoveryPage, DeleteRecoveryRpcOperation, DeleteRecoveryRpcReply, DiskAPI, DiskInfo,
         DiskInfoOptions, DiskStore, FileInfoVersions, FileReader, FileWriter, GuardedDeleteVersionsPayload,
         GuardedFileInfoVersions, OldCurrentSize, PartTransactionAction, QUOTA_MUTATION_FENCE_METADATA_SUFFIX, RUSTFS_META_BUCKET,
-        ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo,
-        WalkDirOptions, get_object_disk_read_timeout, is_quota_mutation_fence_path, validate_batch_read_version_item_count,
+        ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteBackgroundIo, RenameDataResp, SnapshotLeaseToken,
+        UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, is_quota_mutation_fence_path,
+        validate_batch_read_version_item_count, with_remote_background_io,
     };
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::disk::{CurrentDeleteGeneration, DiskOption, new_disk};
@@ -580,9 +581,10 @@ pub(crate) mod ecstore_rpc {
         KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MAX_NETWORK_PROBE_BYTES, MAX_NETWORK_PROBE_DURATION, NetworkPeerProbeClient,
         NetworkPeerProbeError, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client,
         SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, check_and_record_signed_rpc_nonce,
-        decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience, sign_ns_scanner_capability_with_checkpoint_coverage,
-        sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
-        tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
+        decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience, rpc_background_admission,
+        sign_ns_scanner_capability_with_checkpoint_coverage, sign_put_file_capability, sign_tonic_rpc_response_proof,
+        tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
+        verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
         verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned,
         verify_tonic_rpc_signature_with_bootstrap,
     };
@@ -641,8 +643,8 @@ pub(crate) mod ecstore_storage {
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
     pub(crate) use rustfs_ecstore::api::storage::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
-        all_local_disk_path, find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients,
-        prewarm_local_disk_id_map_with_instance_ctx,
+        all_local_disk_path, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc, init_local_disks_with_instance_ctx,
+        init_lock_clients, prewarm_local_disk_id_map_with_instance_ctx,
     };
 }
 
@@ -685,8 +687,10 @@ pub(crate) fn try_current_local_node_name() -> Option<String> {
     crate::storage::runtime_sources::try_current_local_node_name()
 }
 
+pub(crate) use ecstore_disk::{RemoteBackgroundIo, with_remote_background_io};
 #[cfg(test)]
 pub(crate) use ecstore_rpc::gen_signature_headers;
+pub(crate) use ecstore_rpc::rpc_background_admission;
 pub(crate) use ecstore_rpc::sign_tonic_rpc_response_proof;
 #[cfg(test)]
 pub(crate) use ecstore_rpc::verify_tonic_rpc_response_proof;
@@ -2409,8 +2413,10 @@ pub(crate) async fn all_local_disk_path() -> Vec<String> {
     ecstore_storage::all_local_disk_path().await
 }
 
-pub(crate) async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
-    ecstore_storage::find_local_disk_by_ref(disk_ref).await
+pub(crate) use ecstore_storage::find_local_disk_by_ref;
+
+pub(crate) async fn find_local_disk_by_ref_for_rpc(disk_ref: &str) -> std::result::Result<Option<DiskStore>, DiskError> {
+    ecstore_storage::find_local_disk_by_ref_for_rpc(disk_ref).await
 }
 
 pub(crate) trait StorageVersioningConfigExt {

rustfs/src/storage_api.rs

@@ -183,8 +183,9 @@ pub(crate) mod server {
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::ecstore_disk::{DiskAPI, DiskOption, new_disk};
         pub(crate) use crate::storage::storage_api::{
-            ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge,
-            tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_tonic_rpc_signature_with_bootstrap,
+            RemoteBackgroundIo, ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, rpc_background_admission,
+            tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
+            verify_tonic_rpc_signature_with_bootstrap, with_remote_background_io,
         };
 
         pub(crate) fn try_current_local_node_name() -> Option<String> {
@@ -194,7 +195,8 @@ pub(crate) mod server {
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::{
             Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, ScannerScopedDirtyUsageAckEntry,
-            gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, verify_tonic_boot_epoch_response,
+            find_local_disk_by_ref, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
+            init_local_disks, verify_tonic_boot_epoch_response,
         };
 
         pub(crate) mod ecfs {