From dc2cc4da439307d926d1c940eac666a62b7db485 Mon Sep 17 00:00:00 2001 From: Tai An Date: Thu, 16 Jul 2026 08:25:31 -0700 Subject: [PATCH] fix(audio-transform): serialize WebSocket writes to avoid concurrent-write panic (#10857) * fix(audio-transform): serialize WebSocket writes to avoid concurrent-write panic AudioTransformStreamEndpoint writes to the same Gorilla WebSocket connection from two goroutines: the backend-forwarding goroutine emits binary PCM frames (and can call sendWSError on a backend recv error), while the read loop calls sendWSError for malformed mid-stream JSON or a backend send failure. Gorilla WebSocket permits only one concurrent writer, so these writers race and can panic with "concurrent write to websocket connection", resetting the client session; a -race build reports the data race directly. Wrap the connection in a lockedConn that serializes WriteMessage behind a mutex, mirroring the existing lockedConn used by the openresponses WebSocket endpoint. Reads stay on the single read loop, so only writes need the lock. Fixes #10844 Signed-off-by: Tai An * chore: empty commit to re-trigger checks Signed-off-by: Anai-Guo --------- Signed-off-by: Tai An Signed-off-by: Anai-Guo Co-authored-by: Anai-Guo --- .../http/endpoints/localai/audio_transform.go | 22 +++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/core/http/endpoints/localai/audio_transform.go b/core/http/endpoints/localai/audio_transform.go index a11c6595a..2868afed6 100644 --- a/core/http/endpoints/localai/audio_transform.go +++ b/core/http/endpoints/localai/audio_transform.go @@ -36,6 +36,23 @@ var audioTransformWSUpgrader = websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, } +// lockedConn wraps a WebSocket connection so the backend-forwarding goroutine +// and the read loop can both emit frames without tripping Gorilla's +// single-concurrent-writer rule. Without this the two writers race and can +// panic with "concurrent write to websocket connection" (and a -race build +// flags the data race directly). Reads stay on the single read loop, so only +// writes need serialising. Mirrors the openresponses endpoint's lockedConn. +type lockedConn struct { + *websocket.Conn + writeMu sync.Mutex +} + +func (lc *lockedConn) WriteMessage(messageType int, data []byte) error { + lc.writeMu.Lock() + defer lc.writeMu.Unlock() + return lc.Conn.WriteMessage(messageType, data) +} + const ( // audioTransformWSReadLimit is the per-message ceiling on inbound WS // frames. With 16 kHz / 256-sample / s16-stereo (1024 B/frame) the @@ -158,10 +175,11 @@ func AudioTransformEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, // @Router /audio/transformations/stream [get] func AudioTransformStreamEndpoint(app *application.Application) echo.HandlerFunc { return func(c echo.Context) error { - ws, err := audioTransformWSUpgrader.Upgrade(c.Response(), c.Request(), nil) + rawWS, err := audioTransformWSUpgrader.Upgrade(c.Response(), c.Request(), nil) if err != nil { return err } + ws := &lockedConn{Conn: rawWS} defer func() { _ = ws.Close() }() ws.SetReadLimit(audioTransformWSReadLimit) @@ -404,7 +422,7 @@ func splitStereoFrameInto(buf []byte, fmt_ proto.AudioTransformStreamConfig_Samp return *audio, *ref } -func sendWSError(ws *websocket.Conn, msg string) { +func sendWSError(ws *lockedConn, msg string) { payload, _ := json.Marshal(schema.AudioTransformStreamControl{ Type: schema.AudioTransformCtrlError, Error: msg,