feat(runtime): multiplex TCP response streams

docs/design-docs/request-plane.md

@@ -99,6 +99,26 @@ Additional TCP-specific environment variables:
 - `DYN_TCP_CONNECT_TIMEOUT`: Connect timeout for TCP client (default: 3 seconds)
 - `DYN_TCP_CHANNEL_BUFFER`: Request channel buffer size for TCP client (default: 100)
 
+#### TCP Response Multiplexing
+
+The TCP response path always uses the versioned `tcp_response_mux_v1` protocol. Each worker process maintains four persistent response connections to each frontend it communicates with, and logical responses share those connections. For example, eight worker processes paired with one frontend maintain 32 physical response connections after warmup. Request streams remain on their existing dedicated sockets.
+
+Response connections use the mux codec from the first byte. A worker sends a binary `ConnectionHello` containing the protocol version and frontend UUID, and the frontend returns an empty `ConnectionReady`. Every subsequent frame has a 21-byte header containing the payload length, frame kind, and logical-stream UUID. Successful stream prologues and reset frames have empty payloads; a failed prologue carries its error as UTF-8 text.
+
+Workers submit prologue and reset frames through an urgent writer lane and submit data and end frames through an ordered bounded lane. Data frames are batched for up to 1 ms by default to reduce small TCP writes and per-packet operating system overhead. Set `DYN_TCP_RESPONSE_BATCH_INTERVAL_MS=0` for opportunistic batching without an intentional delay. Values above 100 ms or malformed values prevent runtime initialization.
+
+Per-stream credits isolate slow consumers. Each logical stream can queue at most eight ordered frames, and a bounded connection queue limits total userspace memory. The mux does not add a connection-level credit window: TCP provides physical-connection backpressure. Host pools live for the worker-process lifetime, and a process-wide maintenance task warms new frontend pools to four connections and replaces failed connections.
+
+The response mux supports the following tuning variables:
+
+- `DYN_TCP_RESPONSE_BATCH_INTERVAL_MS`: Maximum data batching delay in milliseconds (default: `1`, maximum: `100`)
+- `DYN_TCP_RESPONSE_BATCH_MAX_BYTES`: Maximum encoded bytes per batch (default: `65536`)
+- `DYN_TCP_RESPONSE_BATCH_MAX_FRAMES`: Maximum frames per batch (default: `64`)
+- `DYN_TCP_RESPONSE_STREAM_WINDOW_BYTES`: Initial per-stream flow-control window (default: `262144`)
+- `DYN_TCP_RESPONSE_PACKET_METRICS`: Enable Linux TCP segment diagnostics (`0` or `1`, default: `0`)
+
+Response mux versions do not fall back to the former dedicated response protocol. Upgrade frontends and workers together so both sides support `tcp_response_mux_v1`; mixed versions reject the response connection during its handshake.
+
 ### Using NATS
 
 NATS provides a brokered request plane and can also carry KV events and router replica synchronization over NATS Core.

lib/runtime/src/component/endpoint.rs

@@ -139,6 +139,7 @@ impl EndpointConfigBuilder {
         let metrics_labels: Option<Vec<(&str, &str)>> = metrics_labels
             .as_ref()
             .map(|v| v.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect());
+        handler.set_response_mux_client(endpoint.drt().response_mux_client())?;
         // Add metrics to the handler. The endpoint provides additional information to the handler.
         handler.add_metrics(&endpoint, metrics_labels.as_deref())?;
 

lib/runtime/src/config/environment_names.rs

@@ -625,6 +625,21 @@ pub mod tcp_response_stream {
     /// Host/interface for the TCP response stream server.
     /// If unset, the server auto-detects a routable local IP.
     pub const DYN_TCP_RESPONSE_STREAM_HOST: &str = "DYN_TCP_RESPONSE_STREAM_HOST";
+
+    /// Maximum time response data may wait for cross-stream batching.
+    pub const DYN_TCP_RESPONSE_BATCH_INTERVAL_MS: &str = "DYN_TCP_RESPONSE_BATCH_INTERVAL_MS";
+
+    /// Maximum encoded bytes in one response batch.
+    pub const DYN_TCP_RESPONSE_BATCH_MAX_BYTES: &str = "DYN_TCP_RESPONSE_BATCH_MAX_BYTES";
+
+    /// Maximum logical frames in one response batch.
+    pub const DYN_TCP_RESPONSE_BATCH_MAX_FRAMES: &str = "DYN_TCP_RESPONSE_BATCH_MAX_FRAMES";
+
+    /// Per-stream response flow-control window in bytes.
+    pub const DYN_TCP_RESPONSE_STREAM_WINDOW_BYTES: &str = "DYN_TCP_RESPONSE_STREAM_WINDOW_BYTES";
+
+    /// Enables diagnostic TCP_INFO data-segment accounting for response sockets.
+    pub const DYN_TCP_RESPONSE_PACKET_METRICS: &str = "DYN_TCP_RESPONSE_PACKET_METRICS";
 }
 
 /// Event Plane transport environment variables
@@ -872,6 +887,11 @@ mod tests {
             // TCP Response Stream
             tcp_response_stream::DYN_TCP_RESPONSE_STREAM_PORT,
             tcp_response_stream::DYN_TCP_RESPONSE_STREAM_HOST,
+            tcp_response_stream::DYN_TCP_RESPONSE_BATCH_INTERVAL_MS,
+            tcp_response_stream::DYN_TCP_RESPONSE_BATCH_MAX_BYTES,
+            tcp_response_stream::DYN_TCP_RESPONSE_BATCH_MAX_FRAMES,
+            tcp_response_stream::DYN_TCP_RESPONSE_STREAM_WINDOW_BYTES,
+            tcp_response_stream::DYN_TCP_RESPONSE_PACKET_METRICS,
             // Event Plane
             event_plane::DYN_EVENT_PLANE,
             event_plane::DYN_EVENT_PLANE_CODEC,

lib/runtime/src/distributed.rs

@@ -50,6 +50,7 @@ pub struct DistributedRuntime {
     nats_client: Option<transports::nats::Client>,
     network_manager: Arc<NetworkManager>,
     tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
+    response_mux_client: Arc<crate::pipeline::network::tcp::mux::client::ResponseMuxClientPool>,
     system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
     request_plane: RequestPlaneMode,
 
@@ -193,11 +194,20 @@ impl DistributedRuntime {
             request_plane,
         );
 
+        let response_mux_config =
+            crate::pipeline::network::tcp::mux::initialize_response_mux_config()?;
+        let response_mux_client =
+            crate::pipeline::network::tcp::mux::client::ResponseMuxClientPool::new(
+                runtime.child_token(),
+                response_mux_config,
+            );
+
         let distributed_runtime = Self {
             runtime,
             network_manager: Arc::new(network_manager),
             nats_client,
             tcp_server: Arc::new(OnceCell::new()),
+            response_mux_client,
             system_status_server: Arc::new(OnceLock::new()),
             discovery_client,
             discovery_metadata,
@@ -213,6 +223,8 @@ impl DistributedRuntime {
             event_transport_kind,
         };
 
+        crate::metrics::response_mux::ensure_registered(&distributed_runtime.metrics_registry);
+
         // Initialize the uptime gauge in SystemHealth
         distributed_runtime
             .system_health
@@ -400,6 +412,12 @@ impl DistributedRuntime {
             .clone())
     }
 
+    pub fn response_mux_client(
+        &self,
+    ) -> Arc<crate::pipeline::network::tcp::mux::client::ResponseMuxClientPool> {
+        self.response_mux_client.clone()
+    }
+
     /// Get the network manager
     ///
     /// The network manager consolidates all network configuration and provides

lib/runtime/src/metrics.rs

@@ -9,6 +9,7 @@
 pub mod frontend_perf;
 pub mod prometheus_names;
 pub mod request_plane;
+pub mod response_mux;
 pub mod tokio_perf;
 pub mod transport_metrics;
 pub mod work_handler_perf;

lib/runtime/src/metrics/response_mux.rs

@@ -0,0 +1,220 @@
+// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Bounded-cardinality metrics for the multiplexed TCP response transport.
+
+use std::sync::{Mutex, RwLock, Weak};
+
+use once_cell::sync::Lazy;
+use prometheus::{
+    Histogram, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGaugeVec, Opts,
+};
+
+use crate::MetricsRegistry;
+
+pub static ACTIVE_CONNECTIONS: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_active_connections",
+            "Active physical TCP response-mux connections",
+        ),
+        &["role"],
+    )
+    .expect("response mux active connection gauge")
+});
+
+pub static CONNECTIONS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_connections_total",
+            "Physical TCP response-mux connection lifecycle events",
+        ),
+        &["role", "result"],
+    )
+    .expect("response mux connection counter")
+});
+
+pub static ACTIVE_STREAMS: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_active_streams",
+            "Active logical response streams",
+        ),
+        &["role"],
+    )
+    .expect("response mux active stream gauge")
+});
+
+pub static SETUP_SECONDS: Lazy<Histogram> = Lazy::new(|| {
+    Histogram::with_opts(
+        HistogramOpts::new(
+            "dynamo_tcp_response_mux_setup_seconds",
+            "Time from request dispatch to logical response-stream prologue",
+        )
+        .buckets(vec![
+            0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0,
+        ]),
+    )
+    .expect("response mux setup histogram")
+});
+
+pub static FRAMES_PER_WRITE: Lazy<HistogramVec> = Lazy::new(|| {
+    HistogramVec::new(
+        HistogramOpts::new(
+            "dynamo_tcp_response_mux_frames_per_write",
+            "Logical response-mux frames encoded into each physical write",
+        )
+        .buckets(vec![1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0]),
+        &["role"],
+    )
+    .expect("response mux frames-per-write histogram")
+});
+
+pub static CONFIGURED_BATCH_INTERVAL_MS: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_configured_batch_interval_ms",
+            "Configured response data batching interval in milliseconds",
+        ),
+        &["role"],
+    )
+    .expect("response mux configured batch interval gauge")
+});
+
+pub static WRITE_CALLS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_write_calls_total",
+            "Physical response-mux TCP write calls",
+        ),
+        &["role"],
+    )
+    .expect("response mux write call counter")
+});
+
+pub static DATA_SEGMENTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_data_segments_total",
+            "Kernel TCP data segments sent on response sockets when packet metrics are enabled",
+        ),
+        &["transport", "role"],
+    )
+    .expect("response TCP data segment counter")
+});
+
+pub static RESETS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_resets_total",
+            "Logical response streams reset by role and reason",
+        ),
+        &["role", "reason"],
+    )
+    .expect("response mux reset counter")
+});
+
+pub static RECONNECTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_reconnects_total",
+            "Replacement physical response-mux connections",
+        ),
+        &["role"],
+    )
+    .expect("response mux reconnect counter")
+});
+
+pub static STALLS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "dynamo_tcp_response_mux_stalls_total",
+            "Response producer stalls by bounded admission point",
+        ),
+        &["kind"],
+    )
+    .expect("response mux stall counter")
+});
+
+pub static CONNECTION_LOST_STREAMS_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
+    IntCounter::new(
+        "dynamo_tcp_response_mux_connection_lost_streams_total",
+        "Logical streams failed by physical response-mux connection loss",
+    )
+    .expect("response mux connection-lost stream counter")
+});
+
+static REGISTERED: Lazy<Mutex<Vec<Weak<RwLock<prometheus::Registry>>>>> =
+    Lazy::new(|| Mutex::new(Vec::new()));
+
+pub fn ensure_registered(registry: &MetricsRegistry) {
+    {
+        let mut registered = REGISTERED.lock().expect("response mux registry lock");
+        registered.retain(|candidate| candidate.strong_count() > 0);
+        let identity = std::sync::Arc::downgrade(&registry.prometheus_registry);
+        if registered
+            .iter()
+            .any(|candidate| Weak::ptr_eq(candidate, &identity))
+        {
+            return;
+        }
+        registered.push(identity);
+    }
+
+    registry.add_metric_or_warn(
+        Box::new(ACTIVE_CONNECTIONS.clone()),
+        "response_mux_active_connections",
+    );
+    registry.add_metric_or_warn(
+        Box::new(CONNECTIONS_TOTAL.clone()),
+        "response_mux_connections_total",
+    );
+    registry.add_metric_or_warn(
+        Box::new(ACTIVE_STREAMS.clone()),
+        "response_mux_active_streams",
+    );
+    registry.add_metric_or_warn(
+        Box::new(SETUP_SECONDS.clone()),
+        "response_mux_setup_seconds",
+    );
+    registry.add_metric_or_warn(
+        Box::new(FRAMES_PER_WRITE.clone()),
+        "response_mux_frames_per_write",
+    );
+    registry.add_metric_or_warn(
+        Box::new(CONFIGURED_BATCH_INTERVAL_MS.clone()),
+        "response_mux_configured_batch_interval_ms",
+    );
+    registry.add_metric_or_warn(
+        Box::new(WRITE_CALLS_TOTAL.clone()),
+        "response_mux_write_calls_total",
+    );
+    registry.add_metric_or_warn(
+        Box::new(DATA_SEGMENTS_TOTAL.clone()),
+        "response_data_segments_total",
+    );
+    registry.add_metric_or_warn(Box::new(RESETS_TOTAL.clone()), "response_mux_resets_total");
+    registry.add_metric_or_warn(
+        Box::new(RECONNECTS_TOTAL.clone()),
+        "response_mux_reconnects_total",
+    );
+    registry.add_metric_or_warn(Box::new(STALLS_TOTAL.clone()), "response_mux_stalls_total");
+    registry.add_metric_or_warn(
+        Box::new(CONNECTION_LOST_STREAMS_TOTAL.clone()),
+        "response_mux_connection_lost_streams_total",
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn metrics_register_with_multiple_registries() {
+        let first = MetricsRegistry::new();
+        let second = MetricsRegistry::new();
+        ensure_registered(&first);
+        ensure_registered(&first);
+        ensure_registered(&second);
+    }
+}