feat(router): Multiplex and reuse WebSocket subgraph connections

[enisdenjo]

Closes #1236

Subgraph graphql-transport-ws operations can share an initialized connection when they have the same subgraph, resolved WebSocket endpoint, and inbound connection identity. This replaces one physical WebSocket per operation with one multiplexed connection while preserving independent GraphQL operation streams.

The implementation also adds explicit traffic-shaping controls for whether connections are reused and how queries and mutations are transported.

Configuration

WebSocket behavior is configured under traffic_shaping.all.websocket or overridden per subgraph:

subscriptions:
  websocket:
    subgraphs:
      reviews:
        path: /reviews/ws

traffic_shaping:
  all:
    pool_idle_timeout: 50s # is default
    websocket:
      reuse_connections: true # is default
      execute_mode: reuse_existing # default is "http"

reuse_connections defaults to true and controls whether WebSocket operations multiplex over matching pooled connections. Setting it to false preserves one connection per WebSocket operation.

execute_mode controls queries and mutations:

• http is the default and always uses HTTP
• reuse_existing prefers a matching initialized WebSocket and immediately uses HTTP when none exists or initialization is still in progress
• websocket executes over WebSocket, initializing or joining a pooled connection when reuse is enabled, or opening a dedicated connection when reuse is disabled

Per-subgraph WebSocket fields inherit omitted values from traffic_shaping.all.websocket.

Pooled WebSockets use the existing effective pool_idle_timeout. A per-subgraph timeout overrides traffic_shaping.all.pool_idle_timeout for both HTTP and WebSocket pools. Active WebSocket operations prevent idle expiry.

Configuration examples and routing behavior

Preserve HTTP execution and pool subscriptions

subscriptions:
  enabled: true
  websocket:
    subgraphs:
      reviews:
        path: /reviews/ws

traffic_shaping:
  all:
    websocket:
      reuse_connections: true
      execute_mode: http

What happens:

• subscriptions to reviews use WebSocket
• subscriptions with the same resolved WebSocket endpoint and connection identity share one initialized connection
• queries and mutations always use HTTP, even when a matching WebSocket is open
• this is the default execute behavior; both WebSocket fields can be omitted because reuse_connections defaults to true and execute_mode defaults to http

Opportunistically reuse a subscription connection

subscriptions:
  enabled: true
  websocket:
    subgraphs:
      reviews:
        path: /reviews/ws

traffic_shaping:
  all:
    pool_idle_timeout: 50s
    websocket:
      reuse_connections: true
      execute_mode: reuse_existing
  router:
    dedupe:
      headers:
        include: [authorization, cookie, x-tenant]

What happens:

• a subscription may initialize the pooled WebSocket
• later queries and mutations with the same subgraph, resolved WebSocket endpoint, and connection fingerprint use that connection
• queries and mutations use HTTP when the connection is absent, closed, expired, or still waiting for connection_ack
• a query or mutation never creates or waits for a WebSocket in this mode
• once an initialized connection is selected, execution stays on WebSocket and failures are not retried over HTTP

Send every operation over WebSocket

subscriptions:
  enabled: true
  websocket:
    subgraphs:
      reviews:
        path: /reviews/ws

traffic_shaping:
  all:
    websocket:
      reuse_connections: true
      execute_mode: websocket

What happens:

• subscriptions, queries, and mutations use WebSocket for WebSocket-enabled subgraphs
• the first operation initializes a pooled connection
• concurrent operations for the same resolved WebSocket endpoint and identity join that initialization
• later matching operations multiplex over the initialized connection
• subgraphs without WebSocket subscription configuration continue using their configured transport

Force a dedicated WebSocket per operation

subscriptions:
  enabled: true
  websocket:
    subgraphs:
      reviews:
        path: /reviews/ws

traffic_shaping:
  all:
    websocket:
      reuse_connections: false
      execute_mode: websocket

What happens:

• subscriptions, queries, and mutations use WebSocket
• every operation opens and owns a separate WebSocket
• no connection is inserted into the shared pool
• connection fingerprinting is avoided when reuse is disabled globally and no subgraph override enables it

Override one subgraph

traffic_shaping:
  all:
    pool_idle_timeout: 50s
    websocket:
      reuse_connections: true
      execute_mode: reuse_existing
  subgraphs:
    payments:
      pool_idle_timeout: 5s
      websocket:
        reuse_connections: false
        execute_mode: websocket

What happens:

• other WebSocket-enabled subgraphs opportunistically reuse initialized connections and inherit the 50 second idle timeout
• payments sends queries, mutations, and subscriptions over dedicated WebSockets
• payments inherits nothing for the two explicitly overridden WebSocket fields
• its 5 second pool_idle_timeout also remains the effective timeout for its HTTP connection pool

Connection identity

Request identity is now split into two typed fingerprints:

• ConnectionFingerprint covers the inbound method, path, selected headers, and supergraph identity
• InboundRequestFingerprint extends it with the normalized operation, variables, and extensions for request deduplication

This keeps request deduplication separate from connection reuse. Different operations are not treated as the same request, but they can still multiplex over the same connection.

Header selection follows traffic_shaping.router.dedupe.headers, including when request deduplication is disabled. Operators using a custom header policy must include every inbound header that can affect connection-scoped authentication, authorization, cookies, or tenant identity.

The router avoids computing a connection fingerprint when neither request deduplication nor WebSocket reuse needs one. Inbound client WebSockets compute it once and reuse it across operations.

WebSocket pool

The shared pool coordinates connections per subgraph, resolved WebSocket endpoint, and connection fingerprint.

Concurrent operations that are allowed to initialize a matching connection join one initialization attempt. The executor becomes available only after connection_ack. Failed or canceled attempts wake their waiters and remove only their own pool generation.

The underlying ntex WebSocket client remains on its local runtime because it is not Send. Shared executors communicate with that owner through a bounded command channel. Operations are registered serially and their response streams drain concurrently through the existing subscription buffer.

Pool removal checks exact command-channel ownership, so an expired or failed connection cannot remove a newer replacement.

Query and mutation routing

Plugin executor decisions continue to win. Pool routing happens only when hooks did not return a response or replace the original HTTP executor.

For execute_mode: reuse_existing, the router resolves the endpoint once and uses the corresponding cached subscription executor's parsed WebSocket endpoint to look up an initialized matching connection. This keeps endpoint expressions and overrides isolated without parsing the endpoint again. A missing subscription executor or connecting pool entry is an immediate miss, and execution stays on HTTP.

For execute_mode: websocket, the configured WebSocket executor handles queries and mutations. With reuse enabled it can initialize or join the pool. With reuse disabled it opens a dedicated connection for the operation.

After WebSocket execution is selected, command, timeout, transport, dispatcher, and empty-stream failures are returned to the client. The router does not retry the operation over HTTP, avoiding mutation replay after an uncertain WebSocket send.

WebSocket client lifecycle

WsClient now has explicit connected and initialized states. Only a connected client can run protocol initialization, and only an initialized client can start operations.

The client now:

• propagates initialization and subscribe send failures
• removes partially registered operations when a send is canceled or fails
• rejects subscription ID exhaustion instead of wrapping
• exposes dispatcher completion to the pool owner
• forwards transport failures as typed operation errors
• preserves GraphQL protocol error messages as GraphQL responses
• accepts frames up to 16 MiB instead of ntex's 64 KiB default

Execute timeout now covers WebSocket execution, including connection initialization where applicable, command submission, and the first response.

Buffering and cancellation

Pooled operations preserve the existing bounded buffering and drop-on-backpressure behavior.

The drainer now also waits for the downstream channel to close. This immediately drops a silent upstream stream when its consumer disappears.

For pooled execute, dropping the receiver after the first response sends protocol complete, ends only that logical operation, and allows the physical connection to become idle without waiting for another subgraph message.

Telemetry

The pool exposes the following low-cardinality metrics:

Metric	Type	Attributes	What it shows
hive.router.websocket_pool.connections.active	up-down counter	subgraph.name	currently initialized physical pooled connections
hive.router.websocket_pool.connections.initializations_total	counter	subgraph.name, result=success|error	completed connection initialization attempts
hive.router.websocket_pool.connections.initialization_waiters_total	counter	subgraph.name	operations that joined an initialization already in progress
hive.router.websocket_pool.connections.lookups_total	counter	subgraph.name, result=hit|miss	reuse_existing attempts and whether an initialized connection was found
hive.router.websocket_pool.connections.closed_total	counter	subgraph.name, websocket_pool.connection.close_reason=idle|dispatcher|pool_dropped	why pooled physical connections ended
hive.router.websocket_pool.operations.active	up-down counter	subgraph.name, websocket_pool.operation.type=execute|subscribe	currently active logical operations multiplexed over the pool
hive.router.websocket_pool.operations.started_total	counter	subgraph.name, websocket_pool.operation.type=execute|subscribe	logical operations started through pooled connections

These metrics distinguish physical connection lifetime from logical operation lifetime. That makes it possible to derive and monitor:

• lookup hit ratio: hit lookups / all lookups. A low ratio in reuse_existing mode means most queries and mutations still use HTTP. This can indicate that subscriptions rarely overlap with those requests, connection fingerprints are too fragmented, or the idle timeout is too short.
• initialization failure ratio: error initializations / all initializations. This surfaces DNS, TCP, TLS, WebSocket upgrade, connection_init, and acknowledgement problems before they are hidden among ordinary operation failures.
• single-flight effectiveness: initialization waiters compared with successful initializations. More waiters per initialization means concurrent operations are sharing handshake work instead of opening duplicate connections. A sustained spike can also show initialization latency or a burst of cold identities.
• multiplexing factor: active logical operations divided by active physical connections, or the same comparison over rates of started operations and successful initializations. Values above one demonstrate concurrent multiplexing and quantify how many operation sockets are being avoided.
• connection churn: the rate of successful initializations compared with the rate of closed connections. High churn with mostly idle closures suggests increasing pool_idle_timeout may improve reuse. High dispatcher closure rates point to subgraph or network instability instead.
• pool utilization by operation type: execute versus subscribe values from operations.active and operations.started_total. This shows whether the pool is serving only its original subscription workload or is also carrying queries and mutations.
• pool footprint by subgraph: connections.active grouped by subgraph.name. This identifies subgraphs or identity fragmentation that create unexpectedly many physical connections.
• cold-start pressure: the rate of initializations plus initialization waiters. This helps correlate request latency with WebSocket handshakes and distinguish a traffic burst from connection instability.

Metrics are labeled by subgraph and bounded status values only. Fingerprints, headers, and endpoints are intentionally excluded to avoid leaking identity data and creating unbounded cardinality.

Pool initialization, reuse, lookup, eviction, idle closure, dispatcher failure, and stale initialization events also use the dedicated router::websocket_pool logging target. Logs provide the endpoint and event context needed to investigate a metric anomaly without turning endpoint or connection identity into metric dimensions.

Should we move the websocket subgraph config?

WebSocket endpoint paths still live at subscriptions.websocket.subgraphs.<name>.path. WebSocket is now also a transport for queries and mutations, so that hierarchy may no longer describe the full feature accurately.

Consider moving the path to a transport-level location such as websocket.subgraphs.<name>.path.

+websocket:
+  subgraphs:
+    reviews:
+      path: /reviews/ws
subscriptions:
  enabled: true
- websocket:
-   subgraphs:
-     reviews:
-       path: /reviews/ws

TODO

• docs
• product update

[Copilot]

Pull request overview

This PR introduces WebSocket connection pooling/multiplexing for graphql-transport-ws subgraph operations, allowing multiple GraphQL operation streams to share a single physical WebSocket per subgraph and inbound connection identity. It also adds traffic-shaping configuration that controls whether WebSocket connections are reused and whether queries/mutations can execute over WebSocket.

Changes:

• Add traffic_shaping.*.websocket configuration (reuse_connections, execute_mode) plus helpers to resolve effective per-subgraph values.
• Implement a shared WebSocket pool/executor to single-flight initialization and multiplex logical operations over one connection.
• Add WebSocket pool telemetry (metrics + logging target) and extensive e2e coverage for pooling, reuse behavior, and idle timeout behavior.

Reviewed changes

Copilot reviewed 32 out of 32 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
lib/router-config/src/traffic_shaping.rs	Adds WebSocket traffic-shaping config types and helper accessors (reuse, execute mode, idle timeout).
lib/internal/src/telemetry/metrics/websocket_pool_metrics.rs	New metrics wrapper for pool connection/operation lifecycle and lookup/initialization events.
lib/internal/src/telemetry/metrics/mod.rs	Wires WebSocketPoolMetrics into the shared Metrics struct.
lib/internal/src/telemetry/metrics/catalog.rs	Registers new metric names/labels and their allowed attribute sets.
lib/internal/src/telemetry/logging/targets.rs	Adds router::websocket_pool logging target.
lib/executor/src/executors/websocket.rs	Routes execution/subscribe through the pool when reuse is enabled; updates WsClient API usage and timeout handling.
lib/executor/src/executors/websocket_pool.rs	New pooled WebSocket implementation: single-flight init, multiplexing, idle expiry, and operation submission/dispatch.
lib/executor/src/executors/websocket_client.rs	Refactors WsClient into connected/initialized states; improves error propagation and dispatcher completion signaling.
lib/executor/src/executors/subscription_buffer.rs	Ensures upstream drains cancel promptly when downstream receiver closes.
lib/executor/src/executors/mod.rs	Exposes the new websocket_pool module.
lib/executor/src/executors/map.rs	Adds query/mutation routing based on WebSocketExecuteMode and integrates pool lookups.
lib/executor/src/executors/http.rs	Updates tests for new connection_fingerprint field on SubgraphExecutionRequest.
lib/executor/src/executors/graphql_transport_ws.rs	Replaces payload builder with From<SubgraphExecutionRequest> for subscribe payload creation.
lib/executor/src/executors/error.rs	Adds typed WebSocket client operation error variant.
lib/executor/src/executors/common.rs	Introduces ConnectionFingerprint and InboundRequestFingerprint; threads fingerprint through execution requests.
lib/executor/src/execution/plan.rs	Plumbs connection_fingerprint through plan execution options and subgraph execution requests.
e2e/src/websocket.rs	Updates e2e tests to new WsClient API and stream item error type.
e2e/src/websocket_pool.rs	New end-to-end test suite validating pooling, reuse modes, and idle-timeout behavior.
e2e/src/testkit/mod.rs	Adds async per-path delay support for deterministic pool initialization timing tests.
e2e/src/telemetry/subscription_metrics.rs	Updates tests to new WsClient subscribe API and item error handling.
e2e/src/subscriptions.rs	Updates tests to new WsClient subscribe API and item error handling.
e2e/src/lib.rs	Registers the new websocket pool e2e module.
e2e/src/demand_control/enforcement.rs	Updates demand-control tests to new WsClient subscribe API and item error handling.
e2e/configs/websocket_pool.yaml	New config fixture for websocket pool e2e tests.
bin/router/src/shared_state.rs	Switches in-flight request map keying to InboundRequestFingerprint.
bin/router/src/pipeline/websocket_server.rs	Computes/threads connection fingerprint into WS request handling and dedupe fingerprinting.
bin/router/src/pipeline/mod.rs	Splits connection vs request fingerprinting; threads connection_fingerprint through pipeline execution.
bin/router/src/pipeline/execution.rs	Plumbs connection_fingerprint into planned execution.
bench/subgraphs/lib.rs	Adds /ws subscription routes for additional subgraphs in bench server.
.changeset/multiplex_and_reuse_websocket_subgraph_connections.md	Release notes for multiplexing/reuse behavior and telemetry.
.changeset/add_websocket_connection_reuse_and_execution_mode_configuration.md	Release notes for new traffic shaping WebSocket configuration fields.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 32 out of 32 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (3)

lib/executor/src/executors/map.rs:419

• endpoint_str is already resolved once earlier in execute specifically to avoid evaluating endpoint expressions multiple times. Re-resolving it here can evaluate expressions twice and cause HTTP/WebSocket transport selection to use different destinations than the initial resolution.

                        let endpoint_str = self.resolve_endpoint(subgraph_name, client_request)?;
                        executor =
                            self.get_or_create_subscription_executor(subgraph_name, &endpoint_str)?;

bin/router/src/pipeline/mod.rs:415

• any_websocket_connection_reuse_enabled() defaults to true, so this will compute a connection_fingerprint on every HTTP request even when no subgraph WebSocket endpoints are configured (i.e., subscriptions.websocket is unset) and request dedupe is disabled. That adds avoidable hashing/allocation overhead to the hot path. Consider gating WebSocket reuse fingerprinting on having any subscriptions.websocket configuration present.

        let websocket_reuse_enabled = shared_state
            .router_config
            .traffic_shaping
            .any_websocket_connection_reuse_enabled();

bin/router/src/pipeline/websocket_server.rs:508

• any_websocket_connection_reuse_enabled() defaults to true, so this will compute a connection_fingerprint for inbound WebSocket operations even when no subgraph WebSocket endpoints are configured (no subscriptions.websocket) and request dedupe is disabled. Gate fingerprinting for multiplexing on subscriptions.websocket.is_some() to avoid unnecessary work.

                  let websocket_reuse_enabled = shared_state
                      .router_config
                      .traffic_shaping
                      .any_websocket_connection_reuse_enabled();

[github-actions]

🐋 This PR was built and pushed to the following Docker images:

Image Names: ghcr.io/graphql-hive/router

Platforms: linux/amd64,linux/arm64

Image Tags: ghcr.io/graphql-hive/router:pr-1345 ghcr.io/graphql-hive/router:sha-0ec4001

[Copilot]

Pull request overview

Copilot reviewed 32 out of 32 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (3)

bin/router/src/pipeline/mod.rs:425

• any_websocket_connection_reuse_enabled() defaults to true (via unwrap_or(true)), so websocket_reuse_enabled becomes true even when there are no WebSocket-enabled subgraphs configured. That makes the router compute connection_fingerprint for every request (even with request dedupe disabled), which looks like an avoidable per-request hashing/sorting cost.

Consider additionally gating WebSocket reuse fingerprinting on whether subscriptions.websocket is configured for any subgraph (i.e. websocket.all.is_some() || !websocket.subgraphs.is_empty()), so the default reuse_connections: true doesn’t impose overhead when WebSocket transport isn’t in use at all.

        let websocket_reuse_enabled = shared_state
            .router_config
            .traffic_shaping
            .any_websocket_connection_reuse_enabled();

        // establish a connection fingerprint only if dedupe or multiplexing is enabled
        let connection_fingerprint = (request_dedupe_enabled || websocket_reuse_enabled)
            .then(|| connection_fingerprint(
                req.method(),
                req.path(),
                &request_headers,
                &shared_state.in_flight_requests_header_policy,
                supergraph.snapshot.cache_id,
            ));

bin/router/src/pipeline/websocket_server.rs:517

• Same as the HTTP pipeline: websocket_reuse_enabled is derived only from traffic shaping, and because reuse_connections defaults to true, this path computes a connection_fingerprint for every inbound client WebSocket operation even when no subgraph has WebSocket transport configured.

Consider gating this on whether any WebSocket subgraph config exists under subscriptions.websocket (all set or subgraphs non-empty), so WebSocket clients don’t pay fingerprint hashing cost unless subgraph WebSocket reuse can actually happen.

                  let request_dedupe_enabled =
                      shared_state.router_config.traffic_shaping.router.dedupe.enabled;
                  let websocket_reuse_enabled = shared_state
                      .router_config
                      .traffic_shaping
                      .any_websocket_connection_reuse_enabled();

                  let connection_fingerprint = (request_dedupe_enabled || websocket_reuse_enabled)
                    .then(|| connection_fingerprint(
                      &Method::POST,
                      ws_uri.path(),
                      headers.as_ref(),
                      &shared_state.in_flight_requests_header_policy,
                      supergraph.snapshot.cache_id,
                    ));

lib/executor/src/executors/websocket.rs:176

• The per-operation timeout is applied only while awaiting the oneshot receiver (tokio::time::timeout(timeout, response)), but the actual WebSocket task spawned on the ntex runtime keeps running if the timeout fires (it continues connecting/handshaking/subscribing and will only fail to tx.send because the receiver was dropped). This can leave a live WebSocket connection/task doing work after the request has already timed out.

Consider enforcing the timeout inside the spawned task (wrapping the whole connect/init/subscribe flow) or adding an explicit cancellation signal so the spawned task drops the WsClient promptly when the caller times out.

        let response = async {
            rx.await
                .map_err(|_| SubgraphExecutorError::WebSocketArbiterChannelClosed)?
        };
        match timeout {
            Some(timeout) => tokio::time::timeout(timeout, response).await?,
            None => response.await,
        }

[Copilot]

Pull request overview

Copilot reviewed 32 out of 32 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

lib/executor/src/executors/websocket_pool.rs:510

• In execute, the WebSocket pool metrics guard is created before submit() succeeds, so operations.started_total (incremented inside active_operation) will also be incremented for operations that fail to start (e.g., command channel closed / subscribe send failed). Also, responses.recv() yields SubscriptionItem = Result<SubgraphResponse, SubgraphExecutorError>, but the code returns that inner Result without unwrapping it, which should fail to type-check (or would otherwise return a nested Result). Create the guard only after submit() succeeds and flatten the received item with an extra ?.

        let _operation_guard = self
            .telemetry_context
            .metrics
            .websocket_pool
            .active_operation(&self.id.subgraph_name, WebSocketPoolOperationType::Execute);

bin/router/src/pipeline/mod.rs:415

• any_websocket_connection_reuse_enabled() currently evaluates to true by default (reuse_connections.unwrap_or(true)), which makes websocket_reuse_enabled true even when no subgraph WebSocket endpoints are configured. That causes connection_fingerprint to be computed for every request (even when request dedupe is off and WebSocket routing is impossible), adding avoidable per-request hashing overhead. Gate websocket_reuse_enabled on whether any subscriptions.websocket subgraph configuration exists.

        let websocket_reuse_enabled = shared_state
            .router_config
            .traffic_shaping
            .any_websocket_connection_reuse_enabled();

bin/router/src/pipeline/websocket_server.rs:508

• Same issue as the HTTP pipeline: websocket_reuse_enabled is derived only from traffic-shaping defaults and is true by default, which makes connection_fingerprint computation happen even when no subscriptions.websocket subgraphs are configured (so the fingerprint cannot be used for pooling). Gate this on whether any WebSocket subgraph subscription config exists to avoid unnecessary hashing on every inbound WS operation.

                  let websocket_reuse_enabled = shared_state
                      .router_config
                      .traffic_shaping
                      .any_websocket_connection_reuse_enabled();

[Copilot]

Pull request overview

Copilot reviewed 32 out of 32 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

bin/router/src/pipeline/websocket_server.rs:508

• As written, websocket_reuse_enabled becomes true by default (because reuse_connections defaults to true), which causes connection_fingerprint computation for inbound router WebSocket operations even when there are no WebSocket-enabled subgraphs configured under subscriptions.websocket. This is unnecessary work and can be avoided by gating reuse on whether any WebSocket subgraph config exists.

                  let websocket_reuse_enabled = shared_state
                      .router_config
                      .traffic_shaping
                      .any_websocket_connection_reuse_enabled();

bin/router/src/pipeline/mod.rs:415

• any_websocket_connection_reuse_enabled() defaults to true (when reuse_connections is omitted), so websocket_reuse_enabled becomes true even when no subgraph has WebSocket subscriptions configured. That makes the router compute a connection_fingerprint for every HTTP request even when it can never be used (no WebSocket pool lookups/initializations possible), adding avoidable hashing overhead.

        let websocket_reuse_enabled = shared_state
            .router_config
            .traffic_shaping
            .any_websocket_connection_reuse_enabled();