feat(storage): support per-instance internode data transports (#9257)

* feat(storage): support per-instance internode data transports

* perf(rpc): avoid staging and client cache overhead on TCP writes

.github/workflows/ci.yml

@@ -649,6 +649,10 @@ jobs:
             flags: "--features sftp"
           - name: http3
             flags: "--features http3,swift"
+          - name: internode-transport
+            flags: "--features rustfs/internode-transport"
+            packages: "-p rustfs-ecstore -p rustfs-rio"
+            tests: "--no-tests fail -E 'test(peer_transport_)'"
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
@@ -674,7 +678,7 @@ jobs:
         run: |
           ./scripts/ci/resource_sampler.sh start clippy
           trap './scripts/ci/resource_sampler.sh stop' EXIT
-          cargo clippy -p rustfs -p rustfs-protocols --all-targets ${{ matrix.features.flags }} -- -D warnings
+          cargo clippy -p rustfs ${{ matrix.features.packages || '-p rustfs-protocols' }} --all-targets ${{ matrix.features.flags }} -- -D warnings
 
       - name: Run tests with ${{ matrix.features.name }}
         id: protocol-tests
@@ -691,7 +695,7 @@ jobs:
           mkdir -p artifacts/protocol-tests
           ./scripts/ci/resource_sampler.sh start nextest
           trap './scripts/ci/resource_sampler.sh stop' EXIT
-          cargo nextest run --profile ci -p rustfs -p rustfs-protocols ${{ matrix.features.flags }} \
+          cargo nextest run --profile ci -p rustfs ${{ matrix.features.packages || '-p rustfs-protocols' }} ${{ matrix.features.flags }} ${{ matrix.features.tests }} \
             2>&1 | tee artifacts/protocol-tests/nextest.log
 
       - name: Upload protocol test reports and diagnostics

crates/ecstore/Cargo.toml

@@ -30,6 +30,7 @@ documentation = "https://docs.rs/rustfs-ecstore/latest/rustfs_ecstore/"
 workspace = true
 
 [features]
+internode-transport = []
 default = []
 fips = ["rustfs-s3-client/fips", "rustfs-utils/fips", "rustfs-tls-runtime/fips"]
 diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"]

crates/ecstore/src/api/mod.rs

@@ -638,7 +638,14 @@ pub mod rio {
 }
 
 pub mod rpc {
+    #[cfg(feature = "internode-transport")]
+    pub use crate::cluster::rpc::http_auth::ExternalStreamAuth;
     pub use crate::cluster::rpc::http_auth::with_rpc_background_request;
+    #[cfg(feature = "internode-transport")]
+    pub use crate::cluster::rpc::internode_data_transport::{
+        InternodeDataTransport, InternodeDataTransportCapabilities, NsScannerCapabilityRequest, NsScannerStreamRequest,
+        ReadStreamRequest, TcpHttpInternodeDataTransport, WalkDirStreamRequest, WriteStreamRequest,
+    };
     pub use crate::cluster::rpc::{
         AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MAX_NETWORK_PROBE_BYTES, MAX_NETWORK_PROBE_DURATION,
         NetworkPeerProbeClient, NetworkPeerProbeError, NetworkPeerProbeMeasurement, NetworkPeerTarget, PEER_RESTDRY_RUN,
@@ -699,6 +706,8 @@ pub mod store_list {
 pub mod storage {
     pub use crate::core::pools::{HealLifecycleExpiryContext, POOL_META_NAME};
     pub use crate::store::HealWalkVersion;
+    #[cfg(feature = "internode-transport")]
+    pub use crate::store::external_rpc_disk;
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
         all_local_disk_path, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc, init_local_disks,

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

@@ -1399,6 +1399,73 @@ pub fn build_auth_headers(url: &str, method: &Method, headers: &mut HeaderMap) -
     Ok(())
 }
 
+/// Signs only external stream controls targeting members of this instance's topology.
+#[cfg(feature = "internode-transport")]
+#[derive(Clone)]
+pub struct ExternalStreamAuth {
+    peers: std::sync::Arc<std::collections::HashSet<url::Origin>>,
+}
+
+#[cfg(feature = "internode-transport")]
+impl ExternalStreamAuth {
+    /// The authenticated HTTP route used by peer stream controls.
+    pub const PATH: &'static str = "/rustfs/rpc/external_stream";
+
+    /// Capture the topology already configured for this host; a missing topology has no peers.
+    pub fn new(instance: &crate::runtime::instance::InstanceContext) -> Self {
+        let mut peers = std::collections::HashSet::new();
+        if let Some(topology) = instance.endpoints() {
+            for pool in topology.0 {
+                for endpoint in pool.endpoints.as_ref() {
+                    if matches!(endpoint.url.scheme(), "http" | "https") {
+                        peers.insert(endpoint.url.origin());
+                    }
+                }
+            }
+        }
+        Self {
+            peers: std::sync::Arc::new(peers),
+        }
+    }
+
+    /// Sign a fresh POST for one configured origin, request body digest and replay nonce.
+    /// The caller must send it to the returned URL without redirects or resubmission.
+    pub fn sign_control(&self, endpoint: &str, digest: &[u8; 32]) -> std::io::Result<(String, HeaderMap)> {
+        let mut url = url::Url::parse(endpoint).map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?;
+        if !url.username().is_empty()
+            || url.password().is_some()
+            || !self.peers.contains(&url.origin())
+            || url.path() != "/"
+            || url.query().is_some()
+            || url.fragment().is_some()
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::PermissionDenied,
+                "invalid internode control endpoint",
+            ));
+        }
+        url.set_path(Self::PATH);
+        url.query_pairs_mut()
+            .append_pair("body_sha256", &hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower))
+            .append_pair("request_nonce", &Uuid::new_v4().to_string());
+        let url = url.to_string();
+        let mut headers = HeaderMap::new();
+        headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/octet-stream"));
+        build_auth_headers(&url, &Method::POST, &mut headers)?;
+        Ok((url, headers))
+    }
+
+    /// Derive the operation key for this stream protocol and exact request digest.
+    /// The shared RPC signing credential is never exposed to the stream provider.
+    pub fn transfer_key(&self, request_digest: &[u8; 32]) -> std::io::Result<[u8; 32]> {
+        let secret = get_shared_secret()?;
+        let mut mac = <HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).map_err(std::io::Error::other)?;
+        mac.update(b"rustfs-external-stream-key-v1\0");
+        mac.update(request_digest);
+        Ok(mac.finalize().into_bytes().into())
+    }
+}
+
 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();

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

@@ -220,6 +220,11 @@ impl InternodeDataTransportCapabilities {
 }
 
 #[derive(Debug, Clone)]
+#[cfg_attr(
+    feature = "internode-transport",
+    derive(serde::Serialize, serde::Deserialize),
+    serde(deny_unknown_fields)
+)]
 pub struct ReadStreamRequest {
     pub endpoint: String,
     pub disk: String,
@@ -231,6 +236,11 @@ pub struct ReadStreamRequest {
 }
 
 #[derive(Debug, Clone)]
+#[cfg_attr(
+    feature = "internode-transport",
+    derive(serde::Serialize, serde::Deserialize),
+    serde(deny_unknown_fields)
+)]
 pub struct WriteStreamRequest {
     pub endpoint: String,
     pub disk: String,
@@ -1243,6 +1253,38 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "internode-transport")]
+    #[test]
+    fn peer_transport_is_selected_once_per_instance() {
+        use crate::runtime::instance::InstanceContext;
+        let first = InstanceContext::new();
+        let second = InstanceContext::new();
+        let provider: Arc<dyn InternodeDataTransport> = Arc::new(LegacyTestTransport);
+        first
+            .select_internode_data_transport(provider.clone(), Some(" LEGACY-TEST "))
+            .expect("select external transport");
+        let selected = first.internode_data_transport().expect("first instance selection");
+        assert!(Arc::ptr_eq(selected, &provider));
+        assert!(second.internode_data_transport().is_none());
+        assert_eq!(
+            first
+                .select_internode_data_transport(provider.clone(), Some("tcp"))
+                .expect_err("already configured")
+                .kind(),
+            io::ErrorKind::AlreadyExists
+        );
+        second
+            .select_internode_data_transport(provider.clone(), None)
+            .expect("ordinary TCP");
+        assert_eq!(
+            second.internode_data_transport().expect("ordinary selection").name(),
+            DEFAULT_INTERNODE_DATA_TRANSPORT
+        );
+        let invalid = InstanceContext::new();
+        assert!(invalid.select_internode_data_transport(provider, Some("unknown")).is_err());
+        assert!(invalid.internode_data_transport().is_none());
+    }
+
     #[tokio::test]
     async fn legacy_transport_defaults_namespace_scanner_to_unsupported() {
         let transport = LegacyTestTransport;

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

@@ -997,6 +997,8 @@ pub struct RemoteDisk {
     #[cfg(test)]
     recovery_monitor_teardown_hook: Arc<tokio::sync::Mutex<Option<Arc<RecoveryMonitorTeardownHook>>>>,
     data_transport: Arc<dyn InternodeDataTransport>,
+    #[cfg(feature = "internode-transport")]
+    external_streams: bool,
     /// Bound the per-operation failure warnings of this peer. An unreachable
     /// peer fails every in-flight shard operation; the counters keep the exact
     /// numbers per operation. Per handle, so one peer cannot hide another's.
@@ -1438,6 +1440,8 @@ impl RemoteDisk {
             #[cfg(test)]
             recovery_monitor_teardown_hook: Arc::new(tokio::sync::Mutex::new(None)),
             data_transport,
+            #[cfg(feature = "internode-transport")]
+            external_streams: false,
             timeout_logs: rustfs_utils::LogThrottle::new(REMOTE_DISK_RPC_FAILURE_LOG_INTERVAL_MS),
             network_error_logs: rustfs_utils::LogThrottle::new(REMOTE_DISK_RPC_FAILURE_LOG_INTERVAL_MS),
         };
@@ -1452,6 +1456,14 @@ impl RemoteDisk {
         Ok(disk)
     }
 
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn with_instance_transport(mut self) -> Self {
+        // Built-in names are reserved by the instance registration API. A host provider
+        // owns its open, data verification and cancellation as one operation.
+        self.external_streams = !rustfs_config::KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS.contains(&self.data_transport.name());
+        self
+    }
+
     pub fn runtime_state(&self) -> RuntimeDriveHealthState {
         self.health.runtime_state()
     }
@@ -1475,6 +1487,10 @@ impl RemoteDisk {
     }
 
     async fn open_write_with_retry(&self, request: WriteStreamRequest) -> Result<FileWriter> {
+        #[cfg(feature = "internode-transport")]
+        if self.external_streams {
+            return self.data_transport.open_write(request).await;
+        }
         let mut attempt = 1;
         let mut last_retry_classification = None;
         loop {
@@ -4206,6 +4222,10 @@ impl DiskAPI for RemoteDisk {
             length,
             stall_timeout: timeouts.body_stall,
         };
+        #[cfg(feature = "internode-transport")]
+        if self.external_streams {
+            return self.data_transport.open_read(request).await;
+        }
         let reader = self.open_read_with_retry(request.clone()).await?;
         Ok(Box::new(RetryingRemoteReader::new_with_timeouts(
             reader,
@@ -4237,6 +4257,10 @@ impl DiskAPI for RemoteDisk {
             length,
             stall_timeout: timeouts.body_stall,
         };
+        #[cfg(feature = "internode-transport")]
+        if self.external_streams {
+            return self.data_transport.open_read_chunks(request).await;
+        }
         let reader = self.open_read_chunks_with_retry(request.clone()).await?;
         Ok(reader.map(|reader| {
             Box::new(RetryingRemoteChunkReader::new_with_timeouts(
@@ -10394,6 +10418,81 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_open_returns_the_original_error_after_one_attempt() {
+        let failure = || {
+            OpenWriteTestStep::Error(DiskError::from(rustfs_rio::new_test_internode_http_io_error(
+                rustfs_rio::InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE),
+            )))
+        };
+        let writer = RetryingOpenWriteInternodeDataTransport::with_steps(vec![failure(), OpenWriteTestStep::Success]);
+        let disk = new_remote_disk_with_transport(Arc::new(writer.clone()))
+            .await
+            .with_instance_transport();
+        let error = disk
+            .create_file("", "bucket", "shard", 3)
+            .await
+            .err()
+            .expect("provider write failure");
+        assert_eq!(
+            error.internode_http_error_kind(),
+            Some(rustfs_rio::InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE))
+        );
+        assert_eq!(writer.calls().len(), 1);
+        let reader = RetryingOpenReadInternodeDataTransport::with_steps(vec![failure(), OpenWriteTestStep::Success]);
+        let disk = new_remote_disk_with_transport(Arc::new(reader.clone()))
+            .await
+            .with_instance_transport();
+        let error = disk
+            .read_file_stream("bucket", "shard", 0, 3)
+            .await
+            .err()
+            .expect("provider read failure");
+        assert_eq!(
+            error.internode_http_error_kind(),
+            Some(rustfs_rio::InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE))
+        );
+        assert_eq!(reader.calls().len(), 1);
+    }
+
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_body_failures_never_open_a_fresh_stream() {
+        let provider = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::PartialThenReset(b"x".to_vec())]));
+        let disk = new_remote_disk_with_transport(provider.clone())
+            .await
+            .with_instance_transport();
+        let mut reader = disk.read_file_stream("bucket", "shard", 0, 3).await.expect("reader");
+        let mut bytes = Vec::new();
+        let error = tokio::io::AsyncReadExt::read_to_end(&mut reader, &mut bytes)
+            .await
+            .expect_err("stream reset");
+        assert_eq!(error.kind(), std_io::ErrorKind::ConnectionReset);
+        assert_eq!(bytes, b"x");
+        assert_eq!(provider.read_requests.lock().expect("recorded transport calls").len(), 1);
+        assert!(provider.fresh_read_requests.lock().expect("recorded fresh calls").is_empty());
+        let provider = Arc::new(ResumeTransport::with_chunk_steps(vec![ResumeReadStep::PartialThenReset(b"x".to_vec())]));
+        let disk = new_remote_disk_with_transport(provider.clone())
+            .await
+            .with_instance_transport();
+        let mut reader = disk
+            .read_file_stream_chunks("bucket", "shard", 0, 3)
+            .await
+            .expect("chunked stream")
+            .expect("supported");
+        let chunk = std::future::poll_fn(|cx| Pin::new(reader.as_mut()).poll_read_chunk(cx, 16))
+            .await
+            .expect("first chunk");
+        assert_eq!(chunk, Some(Bytes::from_static(b"x")));
+        let error = std::future::poll_fn(|cx| Pin::new(reader.as_mut()).poll_read_chunk(cx, 16))
+            .await
+            .expect_err("chunk stream reset");
+        assert_eq!(error.kind(), std_io::ErrorKind::ConnectionReset);
+        assert_eq!(provider.chunk_requests.lock().expect("recorded transport calls").len(), 1);
+        assert!(provider.fresh_chunk_requests.lock().expect("recorded fresh calls").is_empty());
+    }
+
     #[tokio::test]
     #[serial(internode_metrics)]
     async fn test_remote_disk_create_file_retries_once_on_retryable_open_write_error() {

crates/ecstore/src/core/sets.rs

@@ -14,6 +14,8 @@
 // limitations under the License.
 
 use crate::disk::error_reduce::count_errs;
+#[cfg(any(test, feature = "test-util"))]
+use crate::disk::new_disk;
 use crate::error::{Error, Result, is_all_volume_not_found, is_err_object_not_found, is_err_strict_volume_not_found};
 use crate::layout::set_heal::{formats_to_drives_info, new_heal_format_sets};
 use crate::multipart_listing::paginate_multipart_listing;
@@ -28,7 +30,6 @@ use crate::{
         DiskAPI, DiskOption, DiskStore,
         error::DiskError,
         format::{DistributionAlgoVersion, FormatV3},
-        new_disk,
     },
     error::StorageError,
     layout::endpoints::{Endpoints, PoolEndpoints},
@@ -1105,6 +1106,7 @@ impl Sets {
         F: FnMut() -> bool + Send,
     {
         let (disks, init_errs) = init_storage_disks_with_errors(
+            &self.ctx,
             &self.endpoints.endpoints,
             &DiskOption {
                 cleanup: false,
@@ -1310,6 +1312,7 @@ async fn _close_storage_disks(disks: &[Option<DiskStore>]) {
 }
 
 async fn init_storage_disks_with_errors(
+    instance: &InstanceContext,
     endpoints: &Endpoints,
     opts: &DiskOption,
 ) -> (Vec<Option<DiskStore>>, Vec<Option<DiskError>>) {
@@ -1318,7 +1321,7 @@ async fn init_storage_disks_with_errors(
     // let errs = Arc::new(RwLock::new(vec![None; endpoints.as_ref().len()]));
     let mut futures = Vec::with_capacity(endpoints.as_ref().len());
     for endpoint in endpoints.as_ref().iter() {
-        futures.push(new_disk(endpoint, opts));
+        futures.push(crate::disk::new_disk_for_instance(instance, endpoint, opts));
 
         // let ep = endpoint.clone();
         // let opt = opts.clone();

crates/ecstore/src/disk/mod.rs

@@ -1567,13 +1567,36 @@ impl Disk {
 }
 
 pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
-    new_disk_with_health_tracker(ep, opt, None).await
+    new_disk_with_health_tracker(
+        ep,
+        opt,
+        None,
+        #[cfg(feature = "internode-transport")]
+        None,
+    )
+    .await
+}
+
+pub(crate) async fn new_disk_for_instance(
+    _instance: &crate::runtime::instance::InstanceContext,
+    ep: &Endpoint,
+    opt: &DiskOption,
+) -> Result<DiskStore> {
+    new_disk_with_health_tracker(
+        ep,
+        opt,
+        None,
+        #[cfg(feature = "internode-transport")]
+        Some(_instance),
+    )
+    .await
 }
 
 pub(crate) async fn new_disk_with_health_tracker(
     ep: &Endpoint,
     opt: &DiskOption,
     reconnect: Option<disk_store::ReconnectDiskHealthState>,
+    #[cfg(feature = "internode-transport")] instance: Option<&crate::runtime::instance::InstanceContext>,
 ) -> Result<DiskStore> {
     if ep.is_local {
         let s = LocalDisk::new(ep, opt.cleanup).await?;
@@ -1583,8 +1606,15 @@ pub(crate) async fn new_disk_with_health_tracker(
             reconnect,
         )))))
     } else {
-        let data_transport = build_internode_data_transport_from_env();
-        let remote_disk = RemoteDisk::new(ep, opt, data_transport?).await?;
+        #[cfg(feature = "internode-transport")]
+        let remote_disk = match instance.and_then(|ctx| ctx.internode_data_transport()) {
+            Some(transport) => RemoteDisk::new(ep, opt, Arc::clone(transport))
+                .await?
+                .with_instance_transport(),
+            None => RemoteDisk::new(ep, opt, build_internode_data_transport_from_env()?).await?,
+        };
+        #[cfg(not(feature = "internode-transport"))]
+        let remote_disk = RemoteDisk::new(ep, opt, build_internode_data_transport_from_env()?).await?;
         Ok(Arc::new(Disk::Remote(Box::new(remote_disk))))
     }
 }

crates/ecstore/src/runtime/instance.rs

@@ -234,6 +234,8 @@ pub struct InstanceContext {
     /// never observe a torn intermediate state.
     erasure_kind: RwLock<SetupType>,
     list_read_budget: OnceLock<ListReadBudget>,
+    #[cfg(feature = "internode-transport")]
+    internode_data_transport: OnceLock<Arc<dyn crate::api::rpc::InternodeDataTransport>>,
     /// This instance's namespace lock manager (Phase 5 Slice 3, backlog#939).
     ///
     /// Owned per-instance so two instances no longer share a lock namespace
@@ -387,6 +389,49 @@ impl InstanceContext {
         Self::with_lock_manager(Arc::new(GlobalLockManager::new()))
     }
 
+    /// Select this instance's transport before creating or reconnecting remote disks.
+    /// Returns an error for a repeated installation or an unsupported configured name.
+    #[cfg(feature = "internode-transport")]
+    pub fn install_internode_data_transport(
+        &self,
+        provider: Arc<dyn crate::api::rpc::InternodeDataTransport>,
+    ) -> std::io::Result<()> {
+        let configured = std::env::var(rustfs_config::ENV_RUSTFS_INTERNODE_DATA_TRANSPORT).ok();
+        self.select_internode_data_transport(provider, configured.as_deref())
+    }
+
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn select_internode_data_transport(
+        &self,
+        provider: Arc<dyn crate::api::rpc::InternodeDataTransport>,
+        configured: Option<&str>,
+    ) -> std::io::Result<()> {
+        let name = provider.name();
+        if name.is_empty()
+            || name.len() > 64
+            || !name.bytes().all(|byte| byte.is_ascii_lowercase() || byte == b'-')
+            || rustfs_config::KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS.contains(&name)
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "invalid external internode transport name",
+            ));
+        }
+        let selected = match configured.map(str::trim).filter(|v| !v.is_empty()) {
+            Some(value) if value.eq_ignore_ascii_case(name) => provider,
+            value => crate::cluster::rpc::internode_data_transport::build_internode_data_transport(value)
+                .map_err(std::io::Error::other)?,
+        };
+        self.internode_data_transport
+            .set(selected)
+            .map_err(|_| std::io::Error::new(std::io::ErrorKind::AlreadyExists, "internode transport already installed"))
+    }
+
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn internode_data_transport(&self) -> Option<&Arc<dyn crate::api::rpc::InternodeDataTransport>> {
+        self.internode_data_transport.get()
+    }
+
     /// Bound native listing fanout independently of GET body permits. A walk
     /// keeps its shared grant until all of its disk producers have terminated.
     /// Acquire before creating the merge channels or taking object locks.
@@ -414,6 +459,8 @@ impl InstanceContext {
         Self {
             erasure_kind: RwLock::new(SetupType::Unknown),
             list_read_budget: OnceLock::new(),
+            #[cfg(feature = "internode-transport")]
+            internode_data_transport: OnceLock::new(),
             lock_manager,
             region: OnceLock::new(),
             deployment_id: OnceLock::new(),

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

@@ -536,7 +536,7 @@ impl SetDisks {
             .as_ref()
             .and_then(|disk| disk.local_health_tracker_epoch_for_reconnect());
 
-        let (new_disk, fm) = match Self::connect_endpoint(ep, previous_health).await {
+        let (new_disk, fm) = match self.connect_endpoint(ep, previous_health).await {
             Ok(res) => res,
             Err(e) => {
                 warn!("renew_disk: connect_endpoint err {:?}", &e);
@@ -702,6 +702,7 @@ impl SetDisks {
     }
 
     pub(in crate::set_disk) async fn connect_endpoint(
+        &self,
         ep: &Endpoint,
         reconnect: Option<disk::disk_store::ReconnectDiskHealthState>,
     ) -> disk::error::Result<(DiskStore, FormatV3)> {
@@ -712,6 +713,8 @@ impl SetDisks {
                 health_check: true,
             },
             reconnect,
+            #[cfg(feature = "internode-transport")]
+            Some(&self.ctx),
         )
         .await?;
 

crates/ecstore/src/set_disk/read.rs

@@ -40,7 +40,7 @@ use crate::diagnostics::get::{
     get_stage_timer_if_enabled, mark_get_object_downstream_closed, record_get_object_pipeline_failure,
     record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
-use crate::disk::{DiskAPI, new_disk};
+use crate::disk::{DiskAPI, new_disk_for_instance};
 use crate::io_support::bitrot::{BitrotReaderStageMetrics, DeferredReaderStripeHandle, object_mmap_read_enabled};
 use crate::set_disk::coding;
 use crate::set_disk::runtime_sources;
@@ -419,7 +419,8 @@ impl SetDisks {
         // An unregistered probe must not leave a health monitor running after
         // failure/cancellation. Only an identity-validated published handle gets
         // monitoring. No replacement formatting or healing is initiated here.
-        let probe = new_disk(
+        let probe = new_disk_for_instance(
+            &self.ctx,
             endpoint,
             &DiskOption {
                 cleanup: false,
@@ -431,7 +432,8 @@ impl SetDisks {
         if self.find_disk_index(&format)? != (self.set_index, index) {
             return Err(Error::CorruptedFormat);
         }
-        let disk = new_disk(
+        let disk = new_disk_for_instance(
+            &self.ctx,
             endpoint,
             &DiskOption {
                 cleanup: false,

crates/ecstore/src/store/init_format.rs

@@ -27,7 +27,6 @@ use crate::{
         DiskInfoOptions, DiskOption, DiskStore, FORMAT_CONFIG_FILE, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET,
         error::DiskError,
         format::{FormatBackend, FormatErasureVersion, FormatMetaVersion, FormatV3},
-        new_disk,
     },
     layout::endpoints::Endpoints,
 };
@@ -65,7 +64,7 @@ pub async fn init_disks(
             match registered {
                 Some(disk) if disk.endpoint() == *ep => Ok(disk),
                 Some(_) => Err(DiskError::InconsistentDisk),
-                None => new_disk(ep, opt).await,
+                None => crate::disk::new_disk_for_instance(instance_ctx, ep, opt).await,
             }
         });
     }

crates/ecstore/src/store/mod.rs

@@ -456,6 +456,8 @@ mod rebalance;
 pub(crate) use rebalance::PoolSpaceForTest;
 pub(crate) mod utils;
 
+#[cfg(feature = "internode-transport")]
+pub use peer::external_rpc_disk;
 use peer::init_local_peer;
 pub use peer::{
     BootstrapLocalTarget, all_local_disk, all_local_disk_path, find_local_disk_by_ref, find_local_disk_by_ref_for_rpc,

crates/ecstore/src/store/peer.rs

@@ -522,6 +522,14 @@ pub async fn find_local_disk_by_ref_for_rpc(disk_ref: &str) -> DiskResult<Option
     Ok(Some(disk))
 }
 
+/// Resolve the target from the verified listener's instance and acquire any background admission.
+#[cfg(feature = "internode-transport")]
+pub async fn external_rpc_disk(ctx: &Arc<InstanceContext>, disk_ref: &str) -> DiskResult<DiskStore> {
+    let (disk, id) = local_disk_candidate(ctx, disk_ref).await?;
+    admit_local_disk(ctx, &disk, id, false).await?;
+    Ok(disk)
+}
+
 pub async fn all_local_disk_path() -> Vec<String> {
     runtime_sources::local_disk_paths().await
 }
@@ -708,6 +716,26 @@ mod tests {
         }])
     }
 
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_only_resolves_the_listeners_own_storage() {
+        let first = Arc::new(InstanceContext::new());
+        let second = Arc::new(InstanceContext::new());
+        let dir = tempfile::tempdir().expect("local disk");
+        let id = Uuid::new_v4();
+        let disk = target_disk(&first, dir.path(), id).await;
+        for reference in [disk.endpoint().to_string(), id.to_string()] {
+            let found = super::external_rpc_disk(&first, &reference)
+                .await
+                .expect("first instance owns the disk");
+            assert!(Arc::ptr_eq(&found, &disk));
+            assert!(matches!(
+                super::external_rpc_disk(&second, &reference).await,
+                Err(DiskError::DiskNotFound)
+            ));
+        }
+    }
+
     async fn target_disk(ctx: &Arc<InstanceContext>, root: &std::path::Path, id: Uuid) -> DiskStore {
         let mut format = crate::layout::format::FormatV3::new(1, 1);
         format.erasure.this = id;

crates/rio/Cargo.toml

@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-rio/latest/rustfs_rio/"
 workspace = true
 
 [features]
+internode-transport = []
 default = []
 hotpath = [
     "hotpath/hotpath",

crates/rio/src/http_reader.rs

@@ -606,7 +606,7 @@ fn apply_http_data_plane_liveness(builder: reqwest::ClientBuilder) -> reqwest::C
         .http2_keep_alive_interval(None::<std::time::Duration>)
 }
 
-async fn build_http_client(
+async fn build_http_client<const DIRECT: bool>(
     disable_proxy: bool,
     tuning: InternodeHttpClientTuning,
     outbound_tls: &rustfs_tls_runtime::GlobalPublishedOutboundTlsState,
@@ -614,6 +614,11 @@ async fn build_http_client(
     let mut builder = apply_http_data_plane_liveness(Client::builder().connect_timeout(std::time::Duration::from_secs(5)));
     builder = apply_http_client_tuning(builder, tuning);
 
+    if DIRECT {
+        builder = builder
+            .redirect(reqwest::redirect::Policy::none())
+            .retry(reqwest::retry::never());
+    }
     if disable_proxy {
         builder = builder.no_proxy();
     }
@@ -692,7 +697,16 @@ fn should_disable_proxy_for_url(url: &str, tuning: InternodeHttpClientTuning) ->
     }
 }
 
-async fn get_http_client(url: &str) -> io::Result<Client> {
+const fn http_client_cache<const DIRECT: bool>() -> &'static arc_swap::ArcSwapOption<CachedClients> {
+    #[cfg(feature = "internode-transport")]
+    if DIRECT {
+        static DIRECT_CLIENT_CACHE: arc_swap::ArcSwapOption<CachedClients> = arc_swap::ArcSwapOption::const_empty();
+        return &DIRECT_CLIENT_CACHE;
+    }
+    &CLIENT_CACHE
+}
+
+async fn get_http_client<const DIRECT: bool>(url: &str) -> io::Result<Client> {
     let tuning = internode_http_client_tuning();
     // Reuse HTTP connection pools while honoring the configured internode proxy
     // policy. The legacy profile only bypasses loopback URLs to preserve defaults.
@@ -702,7 +716,8 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
     // the full PEM + identity bytes when the TLS state hasn't changed.
     let generation = crate::http_runtime_sources::outbound_tls_generation();
 
-    let previous = CLIENT_CACHE.load_full();
+    let cache = const { http_client_cache::<DIRECT>() };
+    let previous = cache.load_full();
     if let Some(cached) = previous.as_ref() {
         if cached.generation == generation {
             return Ok(cached.client_for(disable_proxy));
@@ -713,8 +728,8 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
     // Cache miss or stale generation — load full outbound TLS state.
     let outbound_tls = crate::http_runtime_sources::outbound_tls_state().await;
 
-    let built = match build_http_client(false, tuning, &outbound_tls).await {
-        Ok(client) => match build_http_client(true, tuning, &outbound_tls).await {
+    let built = match build_http_client::<DIRECT>(false, tuning, &outbound_tls).await {
+        Ok(client) => match build_http_client::<DIRECT>(true, tuning, &outbound_tls).await {
             Ok(no_proxy_client) => Ok((client, no_proxy_client)),
             Err(err) => Err(err),
         },
@@ -725,7 +740,7 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
         Err(err) => {
             // Prefer serving with the previous TLS generation over failing the
             // request outright; the stale-generation metric already fired above.
-            if let Some(cached) = previous {
+            if !DIRECT && let Some(cached) = previous {
                 warn!(
                     error = %err,
                     stale_generation = cached.generation,
@@ -746,7 +761,7 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
     // Guard against races: only overwrite the cache if it is empty or contains an
     // older generation, so a slower task cannot regress the TLS state after a
     // faster task already cached a newer generation.
-    CLIENT_CACHE.rcu(|current| match current {
+    cache.rcu(|current| match current {
         Some(existing) if existing.generation > generation => Some(existing.clone()),
         _ => Some(cached.clone()),
     });
@@ -757,7 +772,7 @@ async fn get_fresh_http_client(url: &str) -> io::Result<Client> {
     let tuning = internode_http_client_tuning();
     let disable_proxy = should_disable_proxy_for_url(url, tuning);
     let outbound_tls = crate::http_runtime_sources::outbound_tls_state().await;
-    build_http_client(disable_proxy, tuning, &outbound_tls).await
+    build_http_client::<false>(disable_proxy, tuning, &outbound_tls).await
 }
 
 fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext {
@@ -999,6 +1014,14 @@ pin_project! {
     }
 }
 
+#[derive(Clone, Copy)]
+enum ReaderConnection {
+    Pooled,
+    Fresh,
+    #[cfg(feature = "internode-transport")]
+    Direct,
+}
+
 impl HttpReader {
     pub async fn new(url: String, method: Method, headers: HeaderMap, body: Option<Vec<u8>>) -> io::Result<Self> {
         // http_log!("[HttpReader::new] url: {url}, method: {method:?}, headers: {headers:?}");
@@ -1015,15 +1038,22 @@ impl HttpReader {
         Self::with_capacity_and_stall_timeout(url, method, headers, body, 0, stall_timeout).await
     }
 
-    pub async fn new_fresh_connection_with_stall_timeout(
+    /// Send one authenticated POST without HTTP redirects or automatic resubmission.
+    /// Uses the current internode TLS configuration and a pooled connection for the selected peer.
+    #[cfg(feature = "internode-transport")]
+    pub async fn post_without_redirects(
         url: String,
-        method: Method,
         headers: HeaderMap,
-        body: Option<Vec<u8>>,
-        stall_timeout: Option<Duration>,
+        body: Vec<u8>,
+        stall_timeout: Duration,
     ) -> io::Result<Self> {
-        let init = Self::open(&url, &method, &headers, body, stall_timeout, true).await?;
-        Ok(Self {
+        let method = Method::POST;
+        let init = Self::open(&url, &method, &headers, Some(body), Some(stall_timeout), ReaderConnection::Direct).await?;
+        Ok(Self::from_response(url, method, headers, init))
+    }
+
+    fn from_response(url: String, method: Method, headers: HeaderMap, init: HttpReaderInit) -> Self {
+        Self {
             inner: StreamReader::new(init.stream),
             url,
             method,
@@ -1034,7 +1064,18 @@ impl HttpReader {
             stall_timeout: init.stall_timeout,
             request_started: init.request_started,
             duration_recorded: false,
-        })
+        }
+    }
+
+    pub async fn new_fresh_connection_with_stall_timeout(
+        url: String,
+        method: Method,
+        headers: HeaderMap,
+        body: Option<Vec<u8>>,
+        stall_timeout: Option<Duration>,
+    ) -> io::Result<Self> {
+        let init = Self::open(&url, &method, &headers, body, stall_timeout, ReaderConnection::Fresh).await?;
+        Ok(Self::from_response(url, method, headers, init))
     }
 
     /// Create a new HttpReader from a URL. The request is performed immediately.
@@ -1056,19 +1097,8 @@ impl HttpReader {
         _read_buf_size: usize,
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
-        let init = Self::open(&url, &method, &headers, body, stall_timeout, false).await?;
-        Ok(Self {
-            inner: StreamReader::new(init.stream),
-            url,
-            method,
-            headers,
-            track_internode_metrics: init.track_internode_metrics,
-            internode_operation: init.internode_operation,
-            stall_timer: None,
-            stall_timeout: init.stall_timeout,
-            request_started: init.request_started,
-            duration_recorded: false,
-        })
+        let init = Self::open(&url, &method, &headers, body, stall_timeout, ReaderConnection::Pooled).await?;
+        Ok(Self::from_response(url, method, headers, init))
     }
 
     async fn open(
@@ -1077,14 +1107,15 @@ impl HttpReader {
         headers: &HeaderMap,
         body: Option<Vec<u8>>,
         stall_timeout: Option<Duration>,
-        force_fresh_connection: bool,
+        connection: ReaderConnection,
     ) -> io::Result<HttpReaderInit> {
         let track_internode_metrics = is_internode_rpc_url(url);
         let internode_operation = internode_rpc_operation(url);
-        let client = if force_fresh_connection {
-            get_fresh_http_client(url).await
-        } else {
-            get_http_client(url).await
+        let client = match connection {
+            ReaderConnection::Fresh => get_fresh_http_client(url).await,
+            ReaderConnection::Pooled => get_http_client::<false>(url).await,
+            #[cfg(feature = "internode-transport")]
+            ReaderConnection::Direct => get_http_client::<true>(url).await,
         }
         .inspect_err(|_| {
             record_internode_error(track_internode_metrics, internode_operation);
@@ -1221,7 +1252,7 @@ impl HttpChunkReader {
         body: Option<Vec<u8>>,
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
-        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, false).await?;
+        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, ReaderConnection::Pooled).await?;
         Ok(Self {
             inner: init.stream,
             current: None,
@@ -1242,7 +1273,7 @@ impl HttpChunkReader {
         body: Option<Vec<u8>>,
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
-        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, true).await?;
+        let init = HttpReader::open(&url, &method, &headers, body, stall_timeout, ReaderConnection::Fresh).await?;
         Ok(Self {
             inner: init.stream,
             current: None,
@@ -1505,7 +1536,7 @@ impl HttpWriter {
             //     "[HttpWriter::spawn] sending HTTP request: url={url_clone}, method={method_clone:?}, headers={headers_clone:?}"
             // );
 
-            let client = match get_http_client(&url_clone).await {
+            let client = match get_http_client::<false>(&url_clone).await {
                 Ok(client) => client,
                 Err(err) => {
                     record_internode_error(track_internode_metrics, internode_operation);

crates/rio/src/http_reader/protocol_tests.rs

@@ -152,7 +152,7 @@ impl TlsPeer {
             None,
         );
         tuning.protocol = protocol;
-        build_http_client(true, tuning, &self.outbound_tls)
+        build_http_client::<false>(true, tuning, &self.outbound_tls)
             .await
             .expect("build production TLS client")
     }

rustfs/Cargo.toml

@@ -57,6 +57,7 @@ name = "swift_object_integration_test"
 required-features = ["swift"]
 
 [features]
+internode-transport = ["rustfs-ecstore/internode-transport", "rustfs-rio/internode-transport"]
 default = ["ftps", "webdav", "gcs"]
 fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips", "rustfs-ecstore/fips", "rustfs-protocols/fips", "rustfs-s3-client/fips", "rustfs-targets/fips", "rustfs-utils/fips"]
 gcs = ["rustfs-ecstore/gcs", "dep:google-cloud-auth"]

rustfs/src/app/context/server_slot.rs

@@ -34,6 +34,8 @@ use std::sync::{Arc, OnceLock};
 pub struct ServerContextSlot {
     app_context: OnceLock<Arc<AppContext>>,
     bootstrap_target: Option<BootstrapLocalTarget>,
+    #[cfg(feature = "internode-transport")]
+    peer_control: std::sync::Weak<crate::app::storage_api::context::InstalledControlService>,
     heal_topology_fingerprint: Arc<tokio::sync::OnceCell<String>>,
 }
 
@@ -52,6 +54,8 @@ impl ServerContextSlot {
         Arc::new(Self {
             app_context: OnceLock::new(),
             bootstrap_target: None,
+            #[cfg(feature = "internode-transport")]
+            peer_control: std::sync::Weak::new(),
             heal_topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()),
         })
     }
@@ -64,6 +68,23 @@ impl ServerContextSlot {
         })
     }
 
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn with_peer_control(
+        ctx: Arc<InstanceContext>,
+        control: Option<&Arc<crate::app::storage_api::context::InstalledControlService>>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            bootstrap_target: Some(BootstrapLocalTarget::new(ctx)),
+            peer_control: control.map(Arc::downgrade).unwrap_or_default(),
+            ..Self::default()
+        })
+    }
+
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn peer_control(&self) -> Option<Arc<crate::app::storage_api::context::InstalledControlService>> {
+        self.peer_control.upgrade()
+    }
+
     /// Install this server's application context (once). Returns `false` if
     /// the slot was already installed; the first installation wins, matching
     /// the process-global singleton's `get_or_init` semantics.

rustfs/src/app/storage_api.rs

@@ -1854,6 +1854,8 @@ pub(crate) mod context {
         TableDataPlanePublicationGuards,
     };
     pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext;
+    #[cfg(feature = "internode-transport")]
+    pub(crate) use crate::storage::storage_api::rpc_consumer::InstalledControlService;
     pub(crate) use crate::storage::storage_api::{BootstrapLocalTarget, ECStore, EndpointServerPools, InstanceContext};
     #[cfg(test)]
     pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};

rustfs/src/server/http.rs

@@ -3089,7 +3089,11 @@ fn process_connection(
         let http_service = SwiftService::new(true, None, http_service);
         let http_service =
             EarlyResponseBodyService::new(http_service, request_body_idle_timeout).with_task_owner(task_owner.clone());
+        #[cfg(not(feature = "internode-transport"))]
         let http_service = InternodeRpcService::new(http_service);
+        #[cfg(feature = "internode-transport")]
+        let http_service =
+            InternodeRpcService::with_external_control(http_service, if is_console { None } else { server_ctx.peer_control() });
 
         let external_service = hybrid(http_service.clone(), rpc_service.clone());
         let internode_service = hybrid(http_service, rpc_service);

rustfs/src/startup_entrypoint.rs

@@ -55,8 +55,40 @@ const STARTUP_THREAD_NAME: &str = "rustfs-main";
 /// release build; 8 MiB keeps the budget Linux and macOS already had.
 const STARTUP_THREAD_STACK_SIZE: usize = 8 * 1024 * 1024;
 
+#[cfg(feature = "internode-transport")]
+use crate::storage_api::startup::storage::{
+    ExternalControlService, InstalledControlService, InstanceContext, InternodeDataTransport, PeerControlClient,
+};
+#[cfg(feature = "internode-transport")]
+use std::sync::Arc;
+
+#[cfg(feature = "internode-transport")]
+type BuildPeerTransport = Box<dyn FnOnce(Arc<InstanceContext>) -> Result<Arc<InstalledControlService>> + Send>;
+
+/// Runs the server with one instance-owned peer stream provider. The factory receives
+/// the node's authenticated control client after its topology has been initialized.
+/// Outbound selection uses the configured internode transport. The provider handles
+/// inbound controls until server shutdown, which waits for its workers by a deadline.
+#[cfg(feature = "internode-transport")]
+pub fn run_process_with_peer_transport<T>(build: impl FnOnce(PeerControlClient) -> Result<Arc<T>> + Send + 'static) -> ExitCode
+where
+    T: InternodeDataTransport + ExternalControlService + 'static,
+{
+    let install = Box::new(move |instance: Arc<InstanceContext>| {
+        let provider = build(PeerControlClient::new(&instance))?;
+        let control = InstalledControlService::new(provider.clone(), instance.clone())?;
+        instance.install_internode_data_transport(provider)?;
+        Ok(Arc::new(control))
+    });
+    run_on_startup_thread(move || run_process_on_current_thread(Some(install)))
+}
+
 pub fn run_process() -> ExitCode {
-    run_on_startup_thread(run_process_on_current_thread)
+    #[cfg(not(feature = "internode-transport"))]
+    let run = run_process_on_current_thread;
+    #[cfg(feature = "internode-transport")]
+    let run = || run_process_on_current_thread(None);
+    run_on_startup_thread(run)
 }
 
 /// Run `f` on a thread whose stack is [`STARTUP_THREAD_STACK_SIZE`] on every
@@ -74,10 +106,13 @@ fn run_on_startup_thread<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stat
     }
 }
 
-fn run_process_on_current_thread() -> ExitCode {
+fn run_process_on_current_thread(#[cfg(feature = "internode-transport")] build: Option<BuildPeerTransport>) -> ExitCode {
     // Building the process runtime is a startup fatal boundary.
     let (runtime, dial9_guard) = crate::server::build_tokio_runtime().expect("Failed to build Tokio runtime");
-    let result = runtime.block_on(async_main());
+    let result = runtime.block_on(async_main(
+        #[cfg(feature = "internode-transport")]
+        build,
+    ));
 
     // Preserve the existing trace flush/seal order before runtime teardown.
     // A disabled telemetry guard remains a no-op.
@@ -109,7 +144,7 @@ fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display) {
     eprintln!("{}", format_fatal_stderr_message(context, error));
 }
 
-async fn async_main() -> Result<()> {
+async fn async_main(#[cfg(feature = "internode-transport")] build: Option<BuildPeerTransport>) -> Result<()> {
     #[cfg(feature = "e2e-test-hooks")]
     if let Ok(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_PROBE") {
         let nonce = uuid::Uuid::parse_str(&nonce).map_err(Error::other)?;
@@ -207,7 +242,13 @@ async fn async_main() -> Result<()> {
         Err(StartupServerPreflightError::Other(err)) => return Err(err),
     }
 
-    match run(*config).await {
+    match run(
+        *config,
+        #[cfg(feature = "internode-transport")]
+        build,
+    )
+    .await
+    {
         Ok(_) => Ok(()),
         Err(e) => {
             error!(
@@ -1983,20 +2024,55 @@ fn print_license_report(report: &crate::connect::LicenseReport) -> Result<()> {
     Ok(())
 }
 
-#[instrument(skip(config))]
-async fn run(config: Config) -> Result<()> {
+#[instrument(skip_all)]
+async fn run(config: Config, #[cfg(feature = "internode-transport")] build: Option<BuildPeerTransport>) -> Result<()> {
     // Single-instance startup threads the process bootstrap context through
     // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future
     // multi-instance server constructs its own context here instead.
     let instance_ctx = bootstrap_instance_ctx();
+    let listen = init_startup_listen_context(&config, &instance_ctx).await?;
+    let endpoint_pools = init_startup_storage_foundation(&listen.server_address, &config.volumes, &instance_ctx).await?;
+    #[cfg(not(feature = "internode-transport"))]
+    let server_ctx = ServerContextSlot::with_instance_context(instance_ctx.clone());
+    #[cfg(feature = "internode-transport")]
+    let mut peer_control = build.map(|install| install(instance_ctx.clone())).transpose()?;
+    #[cfg(feature = "internode-transport")]
+    let server_ctx = ServerContextSlot::with_peer_control(instance_ctx.clone(), peer_control.as_ref());
+    let result = run_server(
+        config,
+        listen,
+        endpoint_pools,
+        instance_ctx,
+        server_ctx,
+        #[cfg(feature = "internode-transport")]
+        &mut peer_control,
+    )
+    .await;
+    #[cfg(feature = "internode-transport")]
+    if let Some(control) = peer_control {
+        let shutdown = control.shutdown(Instant::now() + Duration::from_secs(10)).await;
+        return match (result, shutdown) {
+            (Err(run), Err(close)) => Err(Error::new(run.kind(), format!("{run}; peer transport shutdown: {close}"))),
+            (result, Ok(())) => result,
+            (Ok(()), Err(close)) => Err(close),
+        };
+    }
+    result
+}
+
+async fn run_server(
+    config: Config,
+    listen: StartupListenContext,
+    endpoint_pools: crate::storage_api::startup::storage::EndpointServerPools,
+    instance_ctx: std::sync::Arc<crate::storage_api::startup::storage::InstanceContext>,
+    server_ctx: std::sync::Arc<ServerContextSlot>,
+    #[cfg(feature = "internode-transport")] peer_control: &mut Option<Arc<InstalledControlService>>,
+) -> Result<()> {
     let StartupListenContext {
         readiness,
         server_addr,
         server_address,
-    } = init_startup_listen_context(&config, &instance_ctx).await?;
-
-    let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes, &instance_ctx).await?;
-    let server_ctx = ServerContextSlot::with_instance_context(instance_ctx.clone());
+    } = listen;
     let StartupHttpServers {
         state_manager,
         s3_shutdown_tx,
@@ -2048,17 +2124,21 @@ async fn run(config: Config) -> Result<()> {
     )
     .await?;
 
-    run_startup_runtime_lifecycle(StartupRuntimeLifecycle {
-        server_address,
-        state_manager,
-        s3_shutdown_tx,
-        console_shutdown_tx,
-        capacity_tasks,
-        service_runtime,
-        store,
-        shutdown_token: ctx,
-        readiness,
-    })
+    run_startup_runtime_lifecycle(
+        StartupRuntimeLifecycle {
+            server_address,
+            state_manager,
+            s3_shutdown_tx,
+            console_shutdown_tx,
+            capacity_tasks,
+            service_runtime,
+            store,
+            shutdown_token: ctx,
+            readiness,
+        },
+        #[cfg(feature = "internode-transport")]
+        peer_control,
+    )
     .await
 }
 

rustfs/src/startup_lifecycle.rs

@@ -21,7 +21,7 @@ use crate::{
     startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap},
     startup_runtime_sources,
     startup_services::StartupServiceRuntime,
-    startup_shutdown::{StartupBackgroundTasks, run_startup_shutdown_sequence},
+    startup_shutdown::{StartupShutdownServices, run_startup_shutdown_sequence},
 };
 use rustfs_common::GlobalReadiness;
 use rustfs_object_capacity::capacity_manager::CapacityBackgroundTasks;
@@ -191,7 +191,12 @@ impl std::error::Error for StartupShutdownFailure {
     }
 }
 
-pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifecycle) -> Result<()> {
+pub(crate) async fn run_startup_runtime_lifecycle(
+    lifecycle: StartupRuntimeLifecycle,
+    #[cfg(feature = "internode-transport")] peer_control: &mut Option<
+        Arc<crate::storage_api::startup::storage::InstalledControlService>,
+    >,
+) -> Result<()> {
     let StartupRuntimeLifecycle {
         server_address,
         state_manager,
@@ -238,9 +243,11 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
         s3_shutdown_tx,
         console_shutdown_tx,
         optional_runtimes,
-        StartupBackgroundTasks {
+        StartupShutdownServices {
             capacity: capacity_tasks,
             table_maintenance,
+            #[cfg(feature = "internode-transport")]
+            peer_control: peer_control.take(),
         },
         shutdown_token,
     )

rustfs/src/startup_shutdown.rs

@@ -486,10 +486,16 @@ pub(crate) struct FrontShutdownReport {
     optional_expected: Option<usize>,
     optional_joined: bool,
     optional_retired: usize,
+    #[cfg(feature = "internode-transport")]
+    peer_error: Option<std::io::Error>,
 }
 
 impl FrontShutdownReport {
     fn producers_retired(&self) -> bool {
+        #[cfg(feature = "internode-transport")]
+        if self.peer_error.is_some() {
+            return false;
+        }
         self.s3.retired()
             && self.console.retired()
             && self
@@ -498,6 +504,10 @@ impl FrontShutdownReport {
     }
 
     fn allows_marker_clear(&self) -> bool {
+        #[cfg(feature = "internode-transport")]
+        if self.peer_error.is_some() {
+            return false;
+        }
         self.s3.allows_marker_clear()
             && self.console.allows_marker_clear()
             && self
@@ -513,11 +523,19 @@ impl FrontShutdownReport {
         let deadline_missed = [&self.s3, &self.console]
             .into_iter()
             .any(|observation| observation.http12().is_some_and(|outcome| outcome.deadline_missed()));
+        #[cfg(feature = "internode-transport")]
+        let deadline_missed = deadline_missed
+            || self
+                .peer_error
+                .as_ref()
+                .is_some_and(|error| error.kind() == std::io::ErrorKind::TimedOut);
         let incomplete_leaf_proof = !self.producers_retired();
         Err(Box::new(FrontShutdownFailure {
             s3: self.s3.error(),
             console: self.console.error(),
             optional: self.optional,
+            #[cfg(feature = "internode-transport")]
+            peer_error: self.peer_error,
             deadline_missed,
             incomplete_leaf_proof,
         }))
@@ -526,6 +544,8 @@ impl FrontShutdownReport {
 
 #[derive(Debug)]
 pub(crate) struct FrontShutdownFailure {
+    #[cfg(feature = "internode-transport")]
+    peer_error: Option<std::io::Error>,
     s3: Option<ShutdownWaitError>,
     console: Option<ShutdownWaitError>,
     optional: Vec<ShutdownWaitError>,
@@ -550,6 +570,10 @@ impl std::fmt::Display for FrontShutdownFailure {
                 write!(f, "; {service}: {error}")?;
             }
         }
+        #[cfg(feature = "internode-transport")]
+        if let Some(error) = &self.peer_error {
+            write!(f, "; peer transport: {error}")?;
+        }
         for error in &self.optional {
             write!(f, "; optional_protocol: {error}")?;
         }
@@ -559,6 +583,10 @@ impl std::fmt::Display for FrontShutdownFailure {
 
 impl std::error::Error for FrontShutdownFailure {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        #[cfg(feature = "internode-transport")]
+        if let Some(error) = &self.peer_error {
+            return Some(error);
+        }
         self.s3
             .as_ref()
             .or(self.console.as_ref())
@@ -597,7 +625,9 @@ pub(crate) async fn join_startup_fronts(
     report.optional_joined = true;
 }
 
-pub(crate) struct StartupBackgroundTasks {
+pub(crate) struct StartupShutdownServices {
+    #[cfg(feature = "internode-transport")]
+    pub(crate) peer_control: Option<std::sync::Arc<crate::storage_api::startup::storage::InstalledControlService>>,
     pub(crate) capacity: Option<CapacityBackgroundTasks>,
     pub(crate) table_maintenance: Option<crate::table_catalog::PeriodicMaintenanceRuntime>,
 }
@@ -608,10 +638,12 @@ pub(crate) async fn run_startup_shutdown_sequence(
     mut s3_shutdown_handle: Option<ShutdownHandle>,
     mut console_shutdown_handle: Option<ShutdownHandle>,
     optional_runtimes: OptionalRuntimeServices,
-    background_tasks: StartupBackgroundTasks,
+    background_tasks: StartupShutdownServices,
     ctx: CancellationToken,
 ) -> (CapacityShutdownReport, std::io::Result<()>, FrontShutdownReport) {
-    let StartupBackgroundTasks {
+    let StartupShutdownServices {
+        #[cfg(feature = "internode-transport")]
+        peer_control,
         capacity: capacity_tasks,
         table_maintenance,
     } = background_tasks;
@@ -645,6 +677,19 @@ pub(crate) async fn run_startup_shutdown_sequence(
         handle.signal();
     }
 
+    // HTTP listeners have been signalled. A detached data worker may still own
+    // a disk writer, so collect its shutdown result before stopping audit and
+    // notification services or certifying a clean shutdown.
+    #[cfg(feature = "internode-transport")]
+    let peer_error = if let Some(control) = peer_control {
+        control
+            .shutdown(std::time::Instant::now() + std::time::Duration::from_secs(10))
+            .await
+            .err()
+    } else {
+        None
+    };
+
     let mut capacity_report = CapacityShutdownReport {
         configured: capacity_tasks.is_some(),
         timed_out: false,
@@ -789,7 +834,11 @@ pub(crate) async fn run_startup_shutdown_sequence(
         state = "stopping",
         "RustFS server stopping"
     );
-    let mut front_failures = FrontShutdownReport::default();
+    let mut front_failures = FrontShutdownReport {
+        #[cfg(feature = "internode-transport")]
+        peer_error,
+        ..Default::default()
+    };
     join_startup_fronts(
         &mut s3_shutdown_handle,
         &mut console_shutdown_handle,
@@ -1104,6 +1153,36 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_worker_timeout_preserves_unclean_marker_and_original_error() {
+        let report = super::FrontShutdownReport {
+            peer_error: Some(std::io::Error::new(std::io::ErrorKind::TimedOut, "worker still owns a disk writer")),
+            ..Default::default()
+        };
+        let clear = std::cell::Cell::new(false);
+        let waited = std::cell::Cell::new(false);
+        let result = finish_detached_mutations_and_clear_unclean_shutdown_markers(
+            true,
+            &Ok(()),
+            &report,
+            Duration::from_secs(1),
+            |_| async {
+                waited.set(true);
+                true
+            },
+            async {
+                clear.set(true);
+            },
+        )
+        .await;
+        assert!(!result);
+        assert!(!waited.get() && !clear.get());
+        let failure = report.into_result().unwrap_err();
+        assert!(failure.deadline_missed());
+        assert_eq!(failure.peer_error.unwrap().kind(), std::io::ErrorKind::TimedOut);
+    }
+
     #[tokio::test]
     async fn early_positive_tls_receipt_preserves_late_original_join_error() {
         assert_early_tls_receipt_final_join(true).await;

rustfs/src/storage/rpc/external_transport.rs

@@ -0,0 +1,448 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+//! Authenticated, transport-neutral ports for optional internode stream providers.
+//! RPC authentication and local storage semantics remain owned by RustFS.
+
+pub use crate::storage::storage_api::ecstore_internode::{
+    FileReader, FileWriter, InternodeDataTransport, InternodeDataTransportCapabilities, NsScannerCapabilityRequest,
+    NsScannerCapabilityResponse, NsScannerStreamRequest, ReadStreamRequest, TcpHttpInternodeDataTransport, TransportError,
+    WalkDirStreamRequest, WriteStreamRequest,
+};
+use crate::storage::storage_api::{InstanceContext, RemoteBackgroundIo, with_remote_background_io};
+use crate::storage::storage_api::{check_and_record_signed_rpc_nonce, ecstore_internode};
+use async_trait::async_trait;
+use bytes::Bytes;
+use http::{Request, StatusCode};
+use http_body_util::{BodyExt, Limited};
+use serde::Deserialize;
+use sha2::{Digest, Sha256};
+use std::{
+    io,
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+use tokio::sync::Semaphore;
+
+/// Signed node-control route; the enclosing RPC service verifies its signature.
+pub const CONTROL_PATH: &str = ecstore_internode::ExternalStreamAuth::PATH;
+/// Includes the opaque provider offer and the storage operation binding.
+pub const MAX_CONTROL_BYTES: usize = 2 * 1024 * 1024;
+
+/// Capabilities issued only after signature, body digest and replay validation.
+pub struct AuthenticatedStreams {
+    request_digest: [u8; 32],
+    key: [u8; 32],
+    operation: Mutex<Option<StorageOperation>>,
+}
+
+struct StorageOperation {
+    instance: Arc<InstanceContext>,
+    background: Option<RemoteBackgroundIo>,
+}
+
+impl AuthenticatedStreams {
+    /// Identity of the exact signed control body.
+    pub fn request_digest(&self) -> [u8; 32] {
+        self.request_digest
+    }
+
+    /// A purpose-separated key for response and data-channel authentication.
+    pub fn transfer_key(&self) -> io::Result<[u8; 32]> {
+        Ok(self.key)
+    }
+
+    async fn run<T, Fut>(&self, operation: impl FnOnce(Arc<InstanceContext>) -> Fut) -> Result<T, TransportError>
+    where
+        Fut: std::future::Future<Output = Result<T, TransportError>>,
+    {
+        // One control describes one local operation. Its worker inherits the receiver's admission.
+        let storage = self
+            .operation
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .take()
+            .ok_or_else(|| TransportError::Io(io::Error::new(io::ErrorKind::InvalidInput, "control already used")))?;
+        with_remote_background_io(storage.background, async {
+            let result = operation(storage.instance).await;
+            if result.is_err() {
+                RemoteBackgroundIo::discard_current_progress();
+            }
+            result
+        })
+        .await
+    }
+
+    /// Opens a local reader with the authenticated disk scope and metadata compatibility rules.
+    pub async fn read(&self, request: ReadStreamRequest) -> Result<FileReader, TransportError> {
+        self.run(|instance| async move { super::http_service::read_external_stream(&instance, request).await })
+            .await
+    }
+
+    /// Publishes owned chunks after the provider reports verified EOF. Cancellation,
+    /// transport failure or an unverified end must surface as an I/O error.
+    pub async fn write_chunks(
+        &self,
+        request: WriteStreamRequest,
+        reader: rustfs_rio::ChunkReaderBox,
+    ) -> Result<u64, TransportError> {
+        self.run(|instance| async move { super::http_service::write_external_chunks(&instance, request, reader).await })
+            .await
+    }
+}
+
+/// An embedding binary supplies control framing and the native data protocol.
+#[async_trait]
+pub trait ExternalControlService: Send + Sync {
+    /// Maximum authenticated inbound controls, including their detached workers.
+    fn max_operations(&self) -> usize;
+    /// The returned control stream owns no object payload. Retain `admission` in
+    /// native workers until storage and checked cleanup finish, including disconnects.
+    async fn open(
+        &self,
+        control: Bytes,
+        storage: AuthenticatedStreams,
+        admission: Arc<dyn Send + Sync>,
+    ) -> io::Result<FileReader>;
+    /// Stops admission and reports physical resource retention by the deadline.
+    async fn shutdown(&self, deadline: Instant) -> io::Result<()>;
+}
+
+/// Resources and disk namespace belonging to one listener.
+pub(crate) struct InstalledControlService {
+    service: Arc<dyn ExternalControlService>,
+    admission: Arc<Semaphore>,
+    instance: Arc<InstanceContext>,
+    auth: ecstore_internode::ExternalStreamAuth,
+}
+
+impl InstalledControlService {
+    pub(crate) fn new(service: Arc<dyn ExternalControlService>, instance: Arc<InstanceContext>) -> io::Result<Self> {
+        let limit = service.max_operations();
+        if limit == 0 || limit > Semaphore::MAX_PERMITS {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid internode control capacity"));
+        }
+        Ok(Self {
+            service,
+            admission: Arc::new(Semaphore::new(limit)),
+            auth: ecstore_internode::ExternalStreamAuth::new(&instance),
+            instance,
+        })
+    }
+
+    pub(crate) async fn shutdown(&self, deadline: Instant) -> io::Result<()> {
+        self.admission.close();
+        tokio::time::timeout_at(deadline.into(), self.service.shutdown(deadline))
+            .await
+            .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "peer transport shutdown exceeded its deadline"))?
+    }
+}
+
+/// Host capability for sending signed controls using the standard RPC credentials.
+#[derive(Clone)]
+pub struct PeerControlClient(ecstore_internode::ExternalStreamAuth);
+
+impl PeerControlClient {
+    pub(crate) fn new(instance: &InstanceContext) -> Self {
+        Self(ecstore_internode::ExternalStreamAuth::new(instance))
+    }
+
+    /// Derives the data authentication key for one control body using the node credential.
+    pub fn transfer_key(&self, request_digest: &[u8; 32]) -> io::Result<[u8; 32]> {
+        self.0.transfer_key(request_digest)
+    }
+
+    /// Sends one signed control using the existing pooled node HTTP client, TLS,
+    /// proxy configuration and error classification. This function performs no replay.
+    pub async fn open_control(&self, endpoint: &str, body: Bytes, deadline: Instant) -> io::Result<FileReader> {
+        if body.len() > MAX_CONTROL_BYTES {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "internode control exceeds limit"));
+        }
+        let (url, headers) = self.0.sign_control(endpoint, &Sha256::digest(&body).into())?;
+        let remaining = deadline
+            .checked_duration_since(Instant::now())
+            .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "internode operation deadline elapsed"))?;
+        let reader = tokio::time::timeout_at(
+            deadline.into(),
+            rustfs_rio::HttpReader::post_without_redirects(url, headers, body.to_vec(), remaining),
+        )
+        .await
+        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "internode control deadline elapsed"))??;
+        Ok(Box::new(reader))
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ControlQuery {
+    body_sha256: String,
+    request_nonce: uuid::Uuid,
+}
+
+pub(super) async fn handle<B>(installed: &InstalledControlService, request: Request<B>) -> Result<FileReader, ControlRejection>
+where
+    B: http_body::Body + Send + 'static,
+    B::Error: std::error::Error + Send + Sync + 'static,
+{
+    let query = match request
+        .uri()
+        .query()
+        .and_then(|query| serde_urlencoded::from_str::<ControlQuery>(query).ok())
+    {
+        Some(query)
+            if !query.request_nonce.is_nil()
+                && query.body_sha256.len() == 64
+                && query
+                    .body_sha256
+                    .bytes()
+                    .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) =>
+        {
+            query
+        }
+        _ => return Err(response(StatusCode::BAD_REQUEST, "invalid internode control scope")),
+    };
+    let admission = match Arc::clone(&installed.admission).try_acquire_owned() {
+        Ok(permit) => Arc::new(permit),
+        Err(_) => return Err(response(StatusCode::TOO_MANY_REQUESTS, "internode control capacity exhausted")),
+    };
+    let (parts, body) = request.into_parts();
+    let body = match tokio::time::timeout(Duration::from_secs(15), Limited::new(body, MAX_CONTROL_BYTES).collect()).await {
+        Ok(Ok(body)) => body.to_bytes(),
+        Ok(Err(_)) => return Err(response(StatusCode::PAYLOAD_TOO_LARGE, "invalid internode control body")),
+        Err(_) => return Err(response(StatusCode::REQUEST_TIMEOUT, "internode control body deadline elapsed")),
+    };
+    let request_digest: [u8; 32] = Sha256::digest(&body).into();
+    if query.body_sha256 != hex_simd::encode_to_string(request_digest, hex_simd::AsciiCase::Lower) {
+        return Err(response(StatusCode::FORBIDDEN, "invalid internode control digest"));
+    }
+    if let Err(error) =
+        check_and_record_signed_rpc_nonce(&parts.headers, query.request_nonce, CONTROL_PATH, "external_stream", "external")
+    {
+        if crate::storage::storage_api::rpc_auth_is_resource_exhausted(&error) {
+            return Err(response(StatusCode::SERVICE_UNAVAILABLE, "internode replay admission deferred"));
+        }
+        return Err(response(StatusCode::FORBIDDEN, "invalid internode control nonce"));
+    }
+    let key = installed
+        .auth
+        .transfer_key(&request_digest)
+        .map_err(|_| response(StatusCode::SERVICE_UNAVAILABLE, "internode control key unavailable"))?;
+    let storage = AuthenticatedStreams {
+        request_digest,
+        key,
+        operation: Mutex::new(Some(StorageOperation {
+            instance: Arc::clone(&installed.instance),
+            background: RemoteBackgroundIo::take_current(),
+        })),
+    };
+    installed
+        .service
+        .open(body, storage, admission)
+        .await
+        .map_err(|_| response(StatusCode::SERVICE_UNAVAILABLE, "internode stream provider refused control"))
+}
+
+#[derive(Debug)]
+pub(super) struct ControlRejection {
+    pub(super) status: StatusCode,
+    pub(super) message: &'static str,
+}
+fn response(status: StatusCode, message: &'static str) -> ControlRejection {
+    ControlRejection { status, message }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{AuthenticatedStreams, ExternalControlService, FileReader, InstalledControlService, PeerControlClient};
+    use crate::storage::storage_api::InstanceContext;
+    use crate::storage::storage_api::{
+        Endpoint,
+        ecstore_layout::{Endpoints, PoolEndpoints},
+    };
+    use async_trait::async_trait;
+    use bytes::Bytes;
+    use http::StatusCode;
+    use sha2::{Digest as _, Sha256};
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::{
+        io,
+        sync::Arc,
+        time::{Duration, Instant},
+    };
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    #[derive(Default)]
+    struct TestService {
+        opened: AtomicUsize,
+        keep_running: bool,
+    }
+
+    #[async_trait]
+    impl ExternalControlService for TestService {
+        fn max_operations(&self) -> usize {
+            1
+        }
+        async fn open(
+            &self,
+            body: Bytes,
+            storage: AuthenticatedStreams,
+            _admission: Arc<dyn Send + Sync>,
+        ) -> io::Result<FileReader> {
+            let digest: [u8; 32] = Sha256::digest(&body).into();
+            assert_eq!(storage.request_digest(), digest);
+            self.opened.fetch_add(1, Ordering::Relaxed);
+            Ok(Box::new(std::io::Cursor::new(body)))
+        }
+        async fn shutdown(&self, _: Instant) -> io::Result<()> {
+            if self.keep_running {
+                std::future::pending::<()>().await;
+            }
+            Ok(())
+        }
+    }
+
+    fn instance(endpoint: &str) -> Arc<InstanceContext> {
+        let instance = Arc::new(InstanceContext::new());
+        let disk = Endpoint::try_from(format!("{endpoint}/disk").as_str()).expect("configured disk endpoint");
+        instance.set_endpoints(
+            vec![PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 1,
+                endpoints: Endpoints::from(vec![disk]),
+                cmd_line: String::new(),
+                platform: String::new(),
+            }]
+            .into(),
+        );
+        instance
+    }
+
+    #[tokio::test]
+    async fn peer_transport_shutdown_closes_admission_and_bounds_pending_workers() {
+        let service = TestService {
+            keep_running: true,
+            ..TestService::default()
+        };
+        let installed =
+            InstalledControlService::new(Arc::new(service), Arc::new(InstanceContext::new())).expect("valid capacity");
+        let error = installed.shutdown(Instant::now()).await.expect_err("worker remains active");
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(installed.admission.is_closed());
+    }
+
+    #[tokio::test]
+    async fn peer_transport_control_checks_signature_digest_replay_and_instance() {
+        use hyper_util::{rt::TokioIo, service::TowerToHyperService};
+        let _ = rustfs_credentials::set_global_rpc_secret("external-control-test-secret".into());
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("RPC listener");
+        let base = format!("http://{}", listener.local_addr().expect("RPC address"));
+        let instance = instance(&base);
+        let recorder = Arc::new(TestService::default());
+        let service = Arc::new(InstalledControlService::new(recorder.clone(), instance.clone()).expect("control service"));
+        let server = tokio::spawn(async move {
+            loop {
+                let (socket, _) = listener.accept().await.expect("RPC connect");
+                let fallback = tower::service_fn(|_| async {
+                    Ok::<_, std::convert::Infallible>(http::Response::new(rustfs_s3_types::Body::empty()))
+                });
+                let rpc = super::super::http_service::InternodeRpcService::with_external_control(fallback, Some(service.clone()));
+                tokio::spawn(async move {
+                    let _ = hyper::server::conn::http1::Builder::new()
+                        .serve_connection(TokioIo::new(socket), TowerToHyperService::new(rpc))
+                        .await;
+                });
+            }
+        });
+        let sender = PeerControlClient::new(&instance);
+        let other = PeerControlClient::new(&InstanceContext::new());
+        let body = Bytes::from_static(b"control");
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let refused = other
+            .open_control(&base, body.clone(), deadline)
+            .await
+            .err()
+            .expect("another instance has no such peer");
+        assert_eq!(refused.kind(), io::ErrorKind::PermissionDenied);
+        let (url, headers) = sender
+            .0
+            .sign_control(&base, &Sha256::digest(&body).into())
+            .expect("signed request");
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .timeout(Duration::from_secs(5))
+            .build()
+            .expect("test HTTP client");
+        let status = client
+            .post(&url)
+            .body(body.clone())
+            .send()
+            .await
+            .expect("unsigned response")
+            .status();
+        assert_eq!(status, StatusCode::FORBIDDEN);
+        let status = client
+            .post(&url)
+            .headers(headers.clone())
+            .body("forged")
+            .send()
+            .await
+            .expect("digest response")
+            .status();
+        assert_eq!(status, StatusCode::FORBIDDEN);
+        assert_eq!(recorder.opened.load(Ordering::Relaxed), 0);
+        let response = client
+            .post(&url)
+            .headers(headers.clone())
+            .body(body.clone())
+            .send()
+            .await
+            .expect("signed response");
+        assert_eq!(response.status(), StatusCode::OK);
+        assert_eq!(response.bytes().await.expect("control response"), body);
+        let replay = client
+            .post(&url)
+            .headers(headers)
+            .body(body.clone())
+            .send()
+            .await
+            .expect("replay response");
+        assert_eq!(replay.status(), StatusCode::FORBIDDEN);
+        assert_eq!(recorder.opened.load(Ordering::Relaxed), 1);
+        let mut reply = sender
+            .open_control(&base, body.clone(), deadline)
+            .await
+            .expect("fresh signed control");
+        let mut bytes = Vec::new();
+        reply.read_to_end(&mut bytes).await.expect("control response body");
+        assert_eq!(bytes.as_slice(), &body[..]);
+        assert_eq!(recorder.opened.load(Ordering::Relaxed), 2);
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn peer_transport_control_does_not_forward_signed_headers_or_body_to_redirects() {
+        let _ = rustfs_credentials::set_global_rpc_secret("external-control-test-secret".into());
+        let first = TcpListener::bind("127.0.0.1:0").await.expect("registered peer");
+        let second = TcpListener::bind("127.0.0.1:0").await.expect("redirect target");
+        let base = format!("http://{}", first.local_addr().expect("registered address"));
+        let second_addr = second.local_addr().expect("redirect address");
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = first.accept().await.expect("control connection");
+            let mut request = [0; 4096];
+            let n = socket.read(&mut request).await.expect("control request");
+            assert!(String::from_utf8_lossy(&request[..n]).contains("x-rustfs-signature:"));
+            socket.write_all(format!("HTTP/1.1 307 Temporary Redirect\r\nLocation: http://{second_addr}/capture\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.expect("redirect response");
+        });
+        let sender = PeerControlClient::new(&instance(&base));
+        let body = Bytes::from_static(b"control");
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let rejected = tokio::select! {
+            result = sender.open_control(&base, body, deadline) => result.is_err(),
+            _ = second.accept() => panic!("redirect forwarded the signed request"),
+        };
+        assert!(rejected);
+        server.await.expect("registered peer served the redirect");
+    }
+}

rustfs/src/storage/rpc/http_service.rs

@@ -332,11 +332,25 @@ macro_rules! log_internode_put_file_stage_failure {
 #[derive(Clone)]
 pub struct InternodeRpcService<S> {
     inner: S,
+    #[cfg(feature = "internode-transport")]
+    external: Option<Arc<super::external_transport::InstalledControlService>>,
 }
 
 impl<S> InternodeRpcService<S> {
     pub fn new(inner: S) -> Self {
-        Self { inner }
+        Self {
+            inner,
+            #[cfg(feature = "internode-transport")]
+            external: None,
+        }
+    }
+
+    #[cfg(feature = "internode-transport")]
+    pub(crate) fn with_external_control(
+        inner: S,
+        external: Option<Arc<super::external_transport::InstalledControlService>>,
+    ) -> Self {
+        Self { inner, external }
     }
 }
 
@@ -486,7 +500,17 @@ where
             return Box::pin(async move { inner.call(req).await });
         }
 
-        Box::pin(async move { Ok(handle_internode_rpc(req).await.map(ResBody::from)) })
+        #[cfg(feature = "internode-transport")]
+        let external = self.external.clone();
+        Box::pin(async move {
+            Ok(handle_internode_rpc(
+                req,
+                #[cfg(feature = "internode-transport")]
+                external,
+            )
+            .await
+            .map(ResBody::from))
+        })
     }
 }
 
@@ -494,7 +518,10 @@ fn is_internode_rpc_path(path: &str) -> bool {
     path.starts_with(RPC_PREFIX)
 }
 
-async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
+async fn handle_internode_rpc(
+    req: Request<Incoming>,
+    #[cfg(feature = "internode-transport")] external: Option<Arc<super::external_transport::InstalledControlService>>,
+) -> Response<Body> {
     let operation = internode_http_operation(req.uri().path());
     let started_at = Instant::now();
     if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) {
@@ -523,6 +550,17 @@ async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
         let method = req.method().clone();
         let path = req.uri().path();
         let response = match (method, path) {
+            #[cfg(feature = "internode-transport")]
+            (Method::POST, super::external_transport::CONTROL_PATH) => match external {
+                Some(installed) => match super::external_transport::handle(&installed, req).await {
+                    Ok(reader) => {
+                        let stream = tokio_util::io::ReaderStream::with_capacity(reader, 64 * 1024);
+                        Response::new(Body::from_stream(stream))
+                    }
+                    Err(error) => response_with_status(error.status, error.message),
+                },
+                None => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"),
+            },
             (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) {
@@ -820,6 +858,101 @@ where
     response
 }
 
+#[cfg(feature = "internode-transport")]
+pub(super) async fn read_external_stream(
+    instance: &Arc<crate::storage::storage_api::InstanceContext>,
+    request: super::external_transport::ReadStreamRequest,
+) -> Result<FileReader, DiskError> {
+    let disk = crate::storage::storage_api::ecstore_internode::external_rpc_disk(instance, &request.disk).await?;
+    let reader =
+        read_file_stream_with_legacy_meta_fallback(&disk, &request.volume, &request.path, request.offset, request.length).await?;
+    let length = (request.length != 0)
+        .then(|| u64::try_from(request.length))
+        .transpose()
+        .map_err(DiskError::other)?;
+    let reader = match length {
+        Some(length) => Box::new(reader.take(length)) as FileReader,
+        None => reader,
+    };
+    let completion = RemoteBackgroundIo::take_current_completion();
+    if length.is_none() && completion.is_none() {
+        return Ok(reader);
+    }
+    Ok(Box::new(ExternalReadCompletion {
+        reader,
+        remaining: length,
+        completion,
+    }))
+}
+
+#[cfg(feature = "internode-transport")]
+struct ExternalReadCompletion {
+    reader: FileReader,
+    remaining: Option<u64>,
+    completion: Option<rustfs_concurrency::adaptive::permit_set::NodeResourceCompletion>,
+}
+
+#[cfg(feature = "internode-transport")]
+impl tokio::io::AsyncRead for ExternalReadCompletion {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut tokio::io::ReadBuf<'_>) -> Poll<io::Result<()>> {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+        let before = buf.filled().len();
+        if let Err(error) = std::task::ready!(Pin::new(&mut self.reader).poll_read(cx, buf)) {
+            self.completion.take();
+            return Poll::Ready(Err(error));
+        }
+        let read = buf.filled().len() - before;
+        if read == 0 && self.remaining.is_some_and(|remaining| remaining > 0) {
+            self.completion.take();
+            return Poll::Ready(Err(io::Error::new(io::ErrorKind::UnexpectedEof, "truncated internode read")));
+        }
+        let done = match &mut self.remaining {
+            Some(remaining) => {
+                *remaining = remaining.saturating_sub(u64::try_from(read).unwrap_or(u64::MAX));
+                *remaining == 0
+            }
+            None => read == 0,
+        };
+        if done && let Some(completion) = self.completion.take() {
+            completion.complete();
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(feature = "internode-transport")]
+pub(super) async fn write_external_chunks(
+    instance: &Arc<crate::storage::storage_api::InstanceContext>,
+    request: super::external_transport::WriteStreamRequest,
+    mut reader: rustfs_rio::ChunkReaderBox,
+) -> Result<u64, DiskError> {
+    let body = stream::poll_fn(move |cx| {
+        Pin::new(reader.as_mut())
+            .poll_read_chunk(cx, DEFAULT_READ_BUFFER_SIZE)
+            .map(Result::transpose)
+    });
+    let disk = crate::storage::storage_api::ecstore_internode::external_rpc_disk(instance, &request.disk).await?;
+    let query = PutFileQuery {
+        disk: request.disk,
+        volume: request.volume,
+        path: request.path,
+        append: request.append,
+        size: request.size,
+        ..Default::default()
+    };
+    write_staged_put_file(&disk, body, &query, StagedWriteAuth::VerifiedStream)
+        .await
+        .map_err(|(_, error)| match error.downcast::<DiskError>() {
+            Ok(error) => *error,
+            Err(error) => match error.downcast::<io::Error>() {
+                Ok(error) => DiskError::from(*error),
+                Err(error) => DiskError::other(error),
+            },
+        })
+}
+
 fn read_file_body_stream<R>(
     reader: R,
     length: usize,
@@ -1555,12 +1688,35 @@ async fn remove_put_file_staging(disk: &DiskStore, volume: &str, path: &str) ->
     .map_err(Into::into)
 }
 
-async fn write_authenticated_put_file<S, E>(
+fn write_authenticated_put_file<S, E>(
     disk: &DiskStore,
     body: S,
     query: &PutFileQuery,
     nonce: uuid::Uuid,
     url: &str,
+) -> impl Future<Output = Result<u64, (&'static str, BoxError)>>
+where
+    S: futures::TryStream<Ok = Bytes, Error = E> + Unpin,
+    E: Into<BoxError>,
+{
+    write_staged_put_file(disk, body, query, StagedWriteAuth::Signed { nonce, url })
+}
+
+#[derive(Clone, Copy)]
+enum StagedWriteAuth<'a> {
+    Signed {
+        nonce: uuid::Uuid,
+        url: &'a str,
+    },
+    #[cfg(feature = "internode-transport")]
+    VerifiedStream,
+}
+
+async fn write_staged_put_file<S, E>(
+    disk: &DiskStore,
+    body: S,
+    query: &PutFileQuery,
+    auth: StagedWriteAuth<'_>,
 ) -> Result<u64, (&'static str, BoxError)>
 where
     S: futures::TryStream<Ok = Bytes, Error = E> + Unpin,
@@ -1575,8 +1731,18 @@ where
     };
 
     let result = async {
+        let staging_size = query.size;
+        #[cfg(feature = "internode-transport")]
+        let append_size = match auth {
+            StagedWriteAuth::VerifiedStream if query.append && query.size > 0 => Some(query.size.unsigned_abs()),
+            _ => None,
+        };
+        // A provider's append length excludes the existing prefix. A positive total-size
+        // hint would let the local writer close before it consumes the new payload.
+        #[cfg(feature = "internode-transport")]
+        let staging_size = if append_size.is_some() { 0 } else { staging_size };
         let mut file = disk
-            .create_file("", &query.volume, &staging_path, query.size)
+            .create_file("", &query.volume, &staging_path, staging_size)
             .await
             .map_err(|err| ("create_staging", Box::new(err) as BoxError))?;
 
@@ -1592,10 +1758,20 @@ where
             }
         }
 
-        let copied = write_put_file_body_chunks_to_writer(body, &mut file, query, Some(nonce), url)
-            .await
-            .map_err(|err| ("write_body", Box::new(err) as BoxError))?;
-        if put_body_size_mismatch(query, copied) {
+        let copied = match auth {
+            StagedWriteAuth::Signed { nonce, url } => {
+                write_put_file_body_chunks_to_writer(body, &mut file, query, Some(nonce), url)
+                    .await
+                    .map_err(BoxError::from)
+            }
+            #[cfg(feature = "internode-transport")]
+            StagedWriteAuth::VerifiedStream => write_verified_chunks(body, &mut file).await,
+        }
+        .map_err(|err| ("write_body", err))?;
+        let size_mismatch = put_body_size_mismatch(query, copied);
+        #[cfg(feature = "internode-transport")]
+        let size_mismatch = size_mismatch || append_size.is_some_and(|size| size != copied);
+        if size_mismatch {
             return Err((
                 "verify_size",
                 Box::new(io::Error::new(
@@ -1618,6 +1794,18 @@ where
         Ok(copied) => Ok(copied),
         Err((stage, primary)) => {
             if let Err(cleanup) = remove_put_file_staging(disk, &query.volume, &staging_path).await {
+                #[cfg(feature = "internode-transport")]
+                if matches!(auth, StagedWriteAuth::VerifiedStream) {
+                    // A missing staging file is expected when creation failed. Keep the typed
+                    // primary disk error for the peer's quorum and recovery decisions.
+                    if !matches!(
+                        cleanup.downcast_ref::<DiskError>(),
+                        Some(DiskError::FileNotFound | DiskError::VolumeNotFound)
+                    ) {
+                        log_internode_put_file_stage_failure!("cleanup_staging", query, cleanup);
+                    }
+                    return Err((stage, primary));
+                }
                 return Err((
                     stage,
                     Box::new(io::Error::other(format!("{primary}; staging cleanup failed: {cleanup}"))) as BoxError,
@@ -1628,6 +1816,25 @@ where
     }
 }
 
+#[cfg(feature = "internode-transport")]
+async fn write_verified_chunks<S, E, W>(mut body: S, writer: &mut W) -> Result<u64, BoxError>
+where
+    S: futures::TryStream<Ok = Bytes, Error = E> + Unpin,
+    E: Into<BoxError>,
+    W: tokio::io::AsyncWrite + Unpin,
+{
+    // Keep the provider's registered window alive until the disk writer consumes it.
+    let mut copied = 0_u64;
+    while let Some(chunk) = body.try_next().await.map_err(Into::<BoxError>::into)? {
+        writer.write_all(&chunk).await?;
+        let count = u64::try_from(chunk.len()).map_err(io::Error::other)?;
+        copied = copied
+            .checked_add(count)
+            .ok_or_else(|| io::Error::other("stream length overflow"))?;
+    }
+    Ok(copied)
+}
+
 async fn handle_put_file(req: Request<Incoming>, require_auth: bool) -> Response<Body> {
     let method = req.method().clone();
     let path = req.uri().path().to_string();
@@ -3045,6 +3252,114 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_bounded_read_reports_truncation_without_a_background_permit() {
+        let dir = tempfile::tempdir().expect("temporary disk");
+        let path = dir.path().join("shard");
+        tokio::fs::write(&path, b"abcd").await.expect("initial shard");
+        let file = tokio::fs::File::open(&path).await.expect("open shard");
+        let mut reader = super::ExternalReadCompletion {
+            reader: Box::new(file.take(4)),
+            remaining: Some(4),
+            completion: None,
+        };
+        let mut data = [0; 2];
+        reader.read_exact(&mut data).await.expect("first bytes");
+        assert_eq!(&data, b"ab");
+        tokio::fs::OpenOptions::new()
+            .write(true)
+            .open(&path)
+            .await
+            .expect("open for truncation")
+            .set_len(2)
+            .await
+            .expect("truncate shard");
+        assert_eq!(reader.read(&mut []).await.expect("empty caller buffer"), 0);
+        assert_eq!(
+            reader.read(&mut data).await.expect_err("premature EOF").kind(),
+            io::ErrorKind::UnexpectedEof
+        );
+    }
+
+    #[cfg(feature = "internode-transport")]
+    #[tokio::test]
+    async fn peer_transport_staged_write_waits_for_verified_eof_and_preserves_append() {
+        let (disk, dir) = new_put_file_test_disk().await;
+        let mut query = PutFileQuery {
+            volume: "bucket".into(),
+            path: "part".into(),
+            size: 3,
+            ..Default::default()
+        };
+        super::write_staged_put_file(
+            &disk,
+            iter([Ok::<_, io::Error>(Bytes::from_static(b"old"))]),
+            &query,
+            super::StagedWriteAuth::VerifiedStream,
+        )
+        .await
+        .expect("publish original bytes");
+        for append in [false, true] {
+            query.append = append;
+            let unverified = iter([
+                Ok(Bytes::from_static(b"new")),
+                Err(io::Error::new(io::ErrorKind::InvalidData, "unverified EOF")),
+            ]);
+            let (stage, cause) = super::write_staged_put_file(&disk, unverified, &query, super::StagedWriteAuth::VerifiedStream)
+                .await
+                .expect_err("final verification failed");
+            assert_eq!(stage, "write_body");
+            assert_eq!(
+                cause.downcast_ref::<io::Error>().expect("original verification error").kind(),
+                io::ErrorKind::InvalidData
+            );
+            assert_eq!(disk.read_all("bucket", "part").await.expect("existing bytes"), Bytes::from_static(b"old"));
+        }
+        for incorrect in [Bytes::from_static(b"no"), Bytes::from_static(b"extra")] {
+            let (stage, cause) = super::write_staged_put_file(
+                &disk,
+                iter([Ok::<_, io::Error>(incorrect)]),
+                &query,
+                super::StagedWriteAuth::VerifiedStream,
+            )
+            .await
+            .expect_err("declared append length must match");
+            assert_eq!(stage, "verify_size");
+            assert_eq!(
+                cause.downcast_ref::<io::Error>().expect("length error").kind(),
+                io::ErrorKind::UnexpectedEof
+            );
+            assert_eq!(
+                disk.read_all("bucket", "part")
+                    .await
+                    .expect("original after a length mismatch"),
+                Bytes::from_static(b"old")
+            );
+        }
+        super::write_staged_put_file(
+            &disk,
+            iter([Ok::<_, io::Error>(Bytes::from_static(b"new"))]),
+            &query,
+            super::StagedWriteAuth::VerifiedStream,
+        )
+        .await
+        .expect("append verified bytes");
+        assert_eq!(
+            disk.read_all("bucket", "part").await.expect("appended bytes"),
+            Bytes::from_static(b"oldnew")
+        );
+        for entry in std::fs::read_dir(dir.path().join("bucket")).expect("bucket entries") {
+            assert!(
+                !entry
+                    .expect("entry")
+                    .file_name()
+                    .to_string_lossy()
+                    .starts_with(".rustfs-put-")
+            );
+        }
+    }
+
     #[tokio::test]
     async fn authenticated_create_truncation_preserves_existing_local_file() {
         let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string());

rustfs/src/storage/rpc/mod.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(feature = "internode-transport")]
+pub mod external_transport;
 pub mod http_service;
 pub mod node_service;
 

rustfs/src/storage/storage_api.rs

@@ -283,6 +283,10 @@ pub(crate) mod request_context_consumer {
 
 pub(crate) mod rpc_consumer {
     pub(crate) use super::super::rpc::InternodeRpcService;
+    #[cfg(feature = "internode-transport")]
+    pub(crate) use super::super::rpc::external_transport::{
+        ExternalControlService, InstalledControlService, InternodeDataTransport, PeerControlClient,
+    };
 
     pub(crate) mod http_service {
         #[cfg(test)]
@@ -632,6 +636,20 @@ pub(crate) mod ecstore_rio {
     };
 }
 
+#[cfg(feature = "internode-transport")]
+pub(crate) mod ecstore_internode {
+    pub use ecstore_rpc::{
+        ExternalStreamAuth, InternodeDataTransport, InternodeDataTransportCapabilities, NsScannerCapabilityRequest,
+        NsScannerStreamRequest, ReadStreamRequest, TcpHttpInternodeDataTransport, WalkDirStreamRequest, WriteStreamRequest,
+    };
+    use rustfs_ecstore::api::rpc as ecstore_rpc;
+    pub(crate) use rustfs_ecstore::api::storage::external_rpc_disk;
+    pub type TransportError = super::DiskError;
+    pub type FileReader = super::FileReader;
+    pub type FileWriter = super::FileWriter;
+    pub type NsScannerCapabilityResponse = super::storage_contracts::NsScannerCapabilityResponse;
+}
+
 pub(crate) mod ecstore_rpc {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::rpc::verify_ns_scanner_capability_with_checkpoint_coverage;

rustfs/src/storage_api.rs

@@ -485,6 +485,10 @@ pub(crate) mod startup {
         pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::{
             INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV,
         };
+        #[cfg(feature = "internode-transport")]
+        pub(crate) use crate::storage::storage_api::rpc_consumer::{
+            ExternalControlService, InstalledControlService, InternodeDataTransport, PeerControlClient,
+        };
         pub(crate) use crate::storage::storage_api::{
             ECStore, EndpointServerPools, InstanceContext, bootstrap_instance_ctx, global_config_init_error_is_deterministic,
             init_background_replication, init_compression_total_memory_from_backend, init_ecstore_config, init_global_config_sys,

scripts/ci_gate.py

@@ -741,7 +741,8 @@ def check_workflow(root: Path) -> list[str]:
     matrix = jobs.get("test-and-lint-protocols", {}).get("strategy", {}).get("matrix", {})
     features = matrix.get("features", [])
     if not isinstance(features, list) or {item.get("name"): item.get("flags") for item in features} != {
-            "swift": "--features swift", "sftp": "--features sftp", "http3": "--features http3,swift"} or len(features) != 3:
+            "swift": "--features swift", "sftp": "--features sftp", "http3": "--features http3,swift",
+            "internode-transport": "--features rustfs/internode-transport"} or len(features) != 4:
         errors.append("protocol matrix must retain every required feature configuration")
     triggers = document.get("on", {})
     if isinstance(triggers, dict) and isinstance(triggers.get("pull_request"), dict) and any(key in triggers["pull_request"] for key in ("paths", "paths-ignore")):

scripts/test_ci_plan.py

@@ -567,7 +567,8 @@ class GateTests(unittest.TestCase):
             jobs['classify-changes']['outputs'][selection] = f"${{{{ steps.scope.outputs.{selection} || 'true' }}}}"
             jobs[job]['steps'].append({'name': 'Upload report', 'uses': 'actions/upload-artifact', 'with': {'name': job}})
         jobs['test-and-lint-protocols']['strategy'] = {'matrix': {'features': [
-            {'name': 'swift', 'flags': '--features swift'}, {'name': 'sftp', 'flags': '--features sftp'}, {'name': 'http3', 'flags': '--features http3,swift'}]}}
+            {'name': 'swift', 'flags': '--features swift'}, {'name': 'sftp', 'flags': '--features sftp'}, {'name': 'http3', 'flags': '--features http3,swift'},
+            {'name': 'internode-transport', 'flags': '--features rustfs/internode-transport'}]}}
         jobs['required-checks'] = {'name': 'Test and Lint', 'if': "always() && (github.event_name != 'pull_request' || github.event.action != 'closed')",
                                    'needs': list(gate.ALWAYS_JOBS + gate.CODE_JOBS + gate.OPTIONAL_JOBS),
                                    'steps': [{'shell': 'bash', 'env': {'CI_NEEDS': '${{ toJSON(needs) }}'}, 'run': 'python3 scripts/ci_gate.py verify'}]}