Skip to content

feat(router): per-message-deflate with context takeover support for websockets #2549

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dbinnersley wants to merge 2 commits into from
Closed

feat(router): per-message-deflate with context takeover support for websockets #2549

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8ad1be3
Add support for per message deflate to websocket connectinos
dbinnersley Jan 5, 2026
82111cc
add client window bits config, more tests
dbinnersley Feb 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion router-tests/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ require (
github.com/MicahParks/jwkset v0.11.0
github.com/buger/jsonparser v1.1.1
github.com/cloudflare/backoff v0.0.0-20240920015135-e46b80a3a7d0
github.com/coder/websocket v1.8.13
github.com/golang-jwt/jwt/v5 v5.2.2
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.1
Expand Down Expand Up @@ -58,7 +59,6 @@ require (
github.com/cep21/circuit/v4 v4.0.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/cilium/ebpf v0.16.0 // indirect
github.com/coder/websocket v1.8.13 // indirect
github.com/containerd/cgroups/v3 v3.0.2 // indirect
github.com/containerd/stargz-snapshotter/estargz v0.16.3 // indirect
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
Expand Down
40 changes: 35 additions & 5 deletions router-tests/testenv/testenv.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2357,11 +2357,7 @@ type GraphQLError struct {

const maxSocketRetries = 5

func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
dialer := websocket.Dialer{
Subprotocols: []string{"graphql-transport-ws"},
}

func (e *Environment) graphQLWebsocketDialWithRetry(dialer websocket.Dialer, header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
waitBetweenRetriesInMs := rand.Intn(10)
timeToSleep := time.Duration(waitBetweenRetriesInMs) * time.Millisecond

Expand Down Expand Up @@ -2392,6 +2388,21 @@ func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query ur
return nil, nil, err
}

func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
Subprotocols: []string{"graphql-transport-ws"},
}, header, query)
}

// GraphQLWebsocketDialWithCompressionRetry is like GraphQLWebsocketDialWithRetry but enables
// permessage-deflate compression negotiation on the client side.
func (e *Environment) GraphQLWebsocketDialWithCompressionRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
Subprotocols: []string{"graphql-transport-ws"},
EnableCompression: true,
}, header, query)
}
Comment on lines +2391 to +2404
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot Feb 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

# First, check the imports in testenv.go to see which websocket package is being used
head -100 router-tests/testenv/testenv.go | grep -E "^import|websocket"

Repository: wundergraph/cosmo

Length of output: 101

🏁 Script executed:

# Check for websocket imports in the file
rg 'import.*websocket' router-tests/testenv/testenv.go -A 5 -B 5

Repository: wundergraph/cosmo

Length of output: 43

🏁 Script executed:

# Look for the websocket package declaration
rg '^import' router-tests/testenv/testenv.go -A 30

Repository: wundergraph/cosmo

Length of output: 453

🏁 Script executed:

# Search for DefaultDialer usage in the codebase
rg 'DefaultDialer' --type go

Repository: wundergraph/cosmo

Length of output: 174

🏁 Script executed:

# Check websocket.Dialer usage patterns in testenv.go
rg 'websocket\.Dialer' router-tests/testenv/testenv.go -B 2 -A 5

Repository: wundergraph/cosmo

Length of output: 1487

🏁 Script executed:

# Check the gorilla/websocket version in go.mod
rg 'github.com/gorilla/websocket' go.mod

Repository: wundergraph/cosmo

Length of output: 107

🏁 Script executed:

# Find go.mod
find . -name go.mod -type f | head -5

Repository: wundergraph/cosmo

Length of output: 182

🏁 Script executed:

# Also check if there's a go.sum to see websocket version
find . -name go.sum -type f | head -5

Repository: wundergraph/cosmo

Length of output: 182

🏁 Script executed:

# Search for websocket version in any go.mod
rg 'gorilla/websocket' -g 'go.mod'

Repository: wundergraph/cosmo

Length of output: 291

🏁 Script executed:

# Check if there's any comment or documentation about websocket.Dialer in the codebase
rg 'websocket.Dialer' -B 3 -A 8 | head -100

Repository: wundergraph/cosmo

Length of output: 4277

🌐 Web query:

gorilla/websocket v1.5.1 DefaultDialer settings HandshakeTimeout

💡 Result:

In gorilla/websocket v1.5.1, websocket.DefaultDialer is defined with:

  • Proxy: http.ProxyFromEnvironment
  • HandshakeTimeout: 45 * time.Second (i.e., 45s) [1]

Dialer.HandshakeTimeout only limits how long the initial WebSocket opening handshake is allowed to take (“duration for the handshake to complete”)—it is not an idle timeout and it does not govern ping/pong intervals after the connection is established [1]. If your connection drops after being idle, that’s typically due to your app’s read deadlines / ping-pong handling or an intermediate proxy/load balancer timeout, not HandshakeTimeout.

Sources:
[1] Go package docs (DefaultDialer value and HandshakeTimeout field doc) https://pkg.go.dev/github.com/gorilla/websocket (shows DefaultDialer with HandshakeTimeout: 45 * time.Second and describes HandshakeTimeout)

🏁 Script executed:

# Look at the full implementation of graphQLWebsocketDialWithRetry
rg 'func \(e \*Environment\) graphQLWebsocketDialWithRetry' router-tests/testenv/testenv.go -A 40

Repository: wundergraph/cosmo

Length of output: 1480

Preserve DefaultDialer settings when customizing the WebSocket dialer.

The zero-value websocket.Dialer lacks HandshakeTimeout (45 seconds from websocket.DefaultDialer), which means a stalled upgrade handshake will hang indefinitely instead of timing out. The retry loop cannot compensate since there's no per-attempt timeout. Clone websocket.DefaultDialer and override only the fields you need:

♻️ Suggested fix 
 func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
-	return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
-		Subprotocols: []string{"graphql-transport-ws"},
-	}, header, query)
+	dialer := *websocket.DefaultDialer
+	dialer.Subprotocols = []string{"graphql-transport-ws"}
+	return e.graphQLWebsocketDialWithRetry(dialer, header, query)
 }
 
 // GraphQLWebsocketDialWithCompressionRetry is like GraphQLWebsocketDialWithRetry but enables
 // permessage-deflate compression negotiation on the client side.
 func (e *Environment) GraphQLWebsocketDialWithCompressionRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
-	return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
-		Subprotocols:      []string{"graphql-transport-ws"},
-		EnableCompression: true,
-	}, header, query)
+	dialer := *websocket.DefaultDialer
+	dialer.Subprotocols = []string{"graphql-transport-ws"}
+	dialer.EnableCompression = true
+	return e.graphQLWebsocketDialWithRetry(dialer, header, query)
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
Subprotocols: []string{"graphql-transport-ws"},
}, header, query)
}
// GraphQLWebsocketDialWithCompressionRetry is like GraphQLWebsocketDialWithRetry but enables
// permessage-deflate compression negotiation on the client side.
func (e *Environment) GraphQLWebsocketDialWithCompressionRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
return e.graphQLWebsocketDialWithRetry(websocket.Dialer{
Subprotocols: []string{"graphql-transport-ws"},
EnableCompression: true,
}, header, query)
}
func (e *Environment) GraphQLWebsocketDialWithRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
dialer := *websocket.DefaultDialer
dialer.Subprotocols = []string{"graphql-transport-ws"}
return e.graphQLWebsocketDialWithRetry(dialer, header, query)
}
// GraphQLWebsocketDialWithCompressionRetry is like GraphQLWebsocketDialWithRetry but enables
// permessage-deflate compression negotiation on the client side.
func (e *Environment) GraphQLWebsocketDialWithCompressionRetry(header http.Header, query url.Values) (*websocket.Conn, *http.Response, error) {
dialer := *websocket.DefaultDialer
dialer.Subprotocols = []string{"graphql-transport-ws"}
dialer.EnableCompression = true
return e.graphQLWebsocketDialWithRetry(dialer, header, query)
}
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@router-tests/testenv/testenv.go` around lines 2391 - 2404, The current
GraphQLWebsocketDialWithRetry and GraphQLWebsocketDialWithCompressionRetry
construct a zero-value websocket.Dialer which omits defaults like
HandshakeTimeout; update both to start from a copy of websocket.DefaultDialer
(e.g., d := *websocket.DefaultDialer) and then set/override only the needed
fields (Subprotocols and EnableCompression for
GraphQLWebsocketDialWithCompressionRetry) before passing d to
graphQLWebsocketDialWithRetry so the default HandshakeTimeout and other sensible
settings are preserved.


func (e *Environment) InitGraphQLWebSocketConnection(header http.Header, query url.Values, initialPayload json.RawMessage) *websocket.Conn {
conn, _, err := e.GraphQLWebsocketDialWithRetry(header, query)
require.NoError(e.t, err)
Expand All @@ -2409,6 +2420,25 @@ func (e *Environment) InitGraphQLWebSocketConnection(header http.Header, query u
return conn
}

// InitGraphQLWebSocketConnectionWithCompression initializes a WebSocket connection with
// permessage-deflate compression negotiation enabled on the client side.
func (e *Environment) InitGraphQLWebSocketConnectionWithCompression(header http.Header, query url.Values, initialPayload json.RawMessage) (*websocket.Conn, *http.Response) {
conn, resp, err := e.GraphQLWebsocketDialWithCompressionRetry(header, query)
require.NoError(e.t, err)
e.t.Cleanup(func() {
_ = conn.Close()
})
err = conn.WriteJSON(WebSocketMessage{
Type: "connection_init",
Payload: initialPayload,
})
require.NoError(e.t, err)
var ack WebSocketMessage
require.NoError(e.t, ReadAndCheckJSON(e.t, conn, &ack))
require.Equal(e.t, "connection_ack", ack.Type)
return conn, resp
}

func (e *Environment) GraphQLSubscriptionOverSSE(ctx context.Context, request GraphQLRequest, handler func(data string)) {
req, err := e.newGraphQLRequestOverGET(e.GraphQLRequestURL(), request)
if err != nil {
Expand Down
Loading