From 4e1913ebac1e29dcc5236cf40ce35fd64d568d38 Mon Sep 17 00:00:00 2001 From: Eric Litman Date: Sat, 3 Oct 2026 16:42:05 -0400 Subject: [PATCH] test(executor): fix two websocket test-server races behind CI flakes (MASTRA-771) Root cause 1, keepalive tests (Codex and xAI KeepalivePingDuringUpload WithSession, Sessionless and NonstreamSessionless): the fake upstream wrote response.completed and closed the connection as soon as the pong arrived, before the client's held payload write ran. The client read loop (readUpstreamLoop) then saw the close and invalidated the connection, so the write failed with "use of closed network connection". The server now replies only after it has read the request. In the WithSession variants the same race would send the retry back through the hook and panic on a second close(inWriteHook). Root cause 2, TestWebsocketRetryBindFailureClearsActiveSessionState: the server numbered connections after Upgrade, so a handler descheduled after the handshake could take a later number. The third connection then got number 2, was closed at once, and the second request failed with close 1006. The counter now runs before Upgrade, in dial order. Evidence (GOMAXPROCS=1, stress -p 32, 16 CPU burners): - keepalive Sessionless tests: 153/20000 failures before, all the CI signature; 0/60000 after. - retry bind test: 52/6000 before (1006 or reset on the second request); 0/18000 after. - Deterministic: a 100 ms sleep after the pong wait fails 25/25 runs before and 0/25 after. A 200 ms sleep after the first Upgrade fails 15/15 before and 0/15 after. Production code is unchanged. internal/runtime/executor is identical to upstream v8.0.13. Co-Authored-By: Claude Opus 5.5 --- .../codex_websockets_executor_test.go | 36 +++++++++---------- .../executor/websocket_session_target_test.go | 3 +- .../executor/xai_websockets_executor_test.go | 22 ++++++------ 3 files changed, 32 insertions(+), 29 deletions(-) diff --git a/internal/runtime/executor/codex_websockets_executor_test.go b/internal/runtime/executor/codex_websockets_executor_test.go index 47e957f3f..530a109ee 100644 --- a/internal/runtime/executor/codex_websockets_executor_test.go +++ b/internal/runtime/executor/codex_websockets_executor_test.go @@ -2667,15 +2667,11 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_WithSession(t *testing.T) { return nil }) - // Start server reader loop so server processes control frames. - readErrCh := make(chan error, 1) + // The reader answers control frames until the client request arrives. + requestRead := make(chan struct{}) go func() { - for { - if _, _, errRead := conn.ReadMessage(); errRead != nil { - readErrCh <- errRead - return - } - } + _, _, _ = conn.ReadMessage() + close(requestRead) }() // Wait until client has entered writeMessage and is actively holding writeMu. @@ -2701,6 +2697,8 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_WithSession(t *testing.T) { return } + // Reply only after the request arrives, so the reply and close cannot race the held write. + <-requestRead // Now send terminal response. respPayload := []byte(`{"type":"response.completed","response":{"id":"resp-1","status":"completed","output":[]}}`) _ = conn.WriteMessage(websocket.TextMessage, respPayload) @@ -2766,12 +2764,11 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_Sessionless(t *testing.T) { return nil }) + // The reader answers control frames until the client request arrives. + requestRead := make(chan struct{}) go func() { - for { - if _, _, errRead := conn.ReadMessage(); errRead != nil { - return - } - } + _, _, _ = conn.ReadMessage() + close(requestRead) }() // Wait until client has entered writeMessage on sessionless path. @@ -2797,6 +2794,8 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_Sessionless(t *testing.T) { return } + // Reply only after the request arrives, so the reply and close cannot race the held write. + <-requestRead respPayload := []byte(`{"type":"response.completed","response":{"id":"resp-1","status":"completed","output":[]}}`) _ = conn.WriteMessage(websocket.TextMessage, respPayload) })) @@ -2858,12 +2857,11 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_NonstreamSessionless(t *testi return nil }) + // The reader answers control frames until the client request arrives. + requestRead := make(chan struct{}) go func() { - for { - if _, _, errRead := conn.ReadMessage(); errRead != nil { - return - } - } + _, _, _ = conn.ReadMessage() + close(requestRead) }() // Wait until client has entered writeMessage on nonstream path. @@ -2889,6 +2887,8 @@ func TestCodexWebsockets_KeepalivePingDuringUpload_NonstreamSessionless(t *testi return } + // Reply only after the request arrives, so the reply and close cannot race the held write. + <-requestRead respPayload := []byte(`{"type":"response.completed","response":{"id":"resp-1","status":"completed","output":[]}}`) _ = conn.WriteMessage(websocket.TextMessage, respPayload) })) diff --git a/internal/runtime/executor/websocket_session_target_test.go b/internal/runtime/executor/websocket_session_target_test.go index 2422bb0e8..382faac2f 100644 --- a/internal/runtime/executor/websocket_session_target_test.go +++ b/internal/runtime/executor/websocket_session_target_test.go @@ -232,12 +232,13 @@ func TestWebsocketRetryBindFailureClearsActiveSessionState(t *testing.T) { upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} var connections atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Count before the handshake completes so numbers follow dial order. + connection := connections.Add(1) conn, errUpgrade := upgrader.Upgrade(w, r, nil) if errUpgrade != nil { t.Errorf("upgrade websocket: %v", errUpgrade) return } - connection := connections.Add(1) defer func() { _ = conn.Close() }() if connection == 1 { _, _, _ = conn.ReadMessage() diff --git a/internal/runtime/executor/xai_websockets_executor_test.go b/internal/runtime/executor/xai_websockets_executor_test.go index ebf42156f..de16dd133 100644 --- a/internal/runtime/executor/xai_websockets_executor_test.go +++ b/internal/runtime/executor/xai_websockets_executor_test.go @@ -2214,12 +2214,11 @@ func TestXAIWebsockets_KeepalivePingDuringUpload_WithSession(t *testing.T) { return nil }) + // The reader answers control frames until the client request arrives. + requestRead := make(chan struct{}) go func() { - for { - if _, _, errReadLoop := conn.ReadMessage(); errReadLoop != nil { - return - } - } + _, _, _ = conn.ReadMessage() + close(requestRead) }() select { @@ -2242,6 +2241,8 @@ func TestXAIWebsockets_KeepalivePingDuringUpload_WithSession(t *testing.T) { return } + // Reply only after the request arrives, so the reply and close cannot race the held write. + <-requestRead respPayload := []byte(`{"type":"response.done","response":{"id":"resp-1","status":"completed","output":[]}}`) _ = conn.WriteMessage(websocket.TextMessage, respPayload) })) @@ -2310,12 +2311,11 @@ func TestXAIWebsockets_KeepalivePingDuringUpload_Sessionless(t *testing.T) { return nil }) + // The reader answers control frames until the client request arrives. + requestRead := make(chan struct{}) go func() { - for { - if _, _, errReadLoop := conn.ReadMessage(); errReadLoop != nil { - return - } - } + _, _, _ = conn.ReadMessage() + close(requestRead) }() select { @@ -2338,6 +2338,8 @@ func TestXAIWebsockets_KeepalivePingDuringUpload_Sessionless(t *testing.T) { return } + // Reply only after the request arrives, so the reply and close cannot race the held write. + <-requestRead respPayload := []byte(`{"type":"response.done","response":{"id":"resp-1","status":"completed","output":[]}}`) _ = conn.WriteMessage(websocket.TextMessage, respPayload) }))