Skip to content

Commit 4cfa6a0

Browse files
committed
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 <antai12232931@outlook.com>
1 parent a6cf67c commit 4cfa6a0

1 file changed

Lines changed: 20 additions & 2 deletions

File tree

core/http/endpoints/localai/audio_transform.go

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,23 @@ var audioTransformWSUpgrader = websocket.Upgrader{
3636
CheckOrigin: func(r *http.Request) bool { return true },
3737
}
3838

39+
// lockedConn wraps a WebSocket connection so the backend-forwarding goroutine
40+
// and the read loop can both emit frames without tripping Gorilla's
41+
// single-concurrent-writer rule. Without this the two writers race and can
42+
// panic with "concurrent write to websocket connection" (and a -race build
43+
// flags the data race directly). Reads stay on the single read loop, so only
44+
// writes need serialising. Mirrors the openresponses endpoint's lockedConn.
45+
type lockedConn struct {
46+
*websocket.Conn
47+
writeMu sync.Mutex
48+
}
49+
50+
func (lc *lockedConn) WriteMessage(messageType int, data []byte) error {
51+
lc.writeMu.Lock()
52+
defer lc.writeMu.Unlock()
53+
return lc.Conn.WriteMessage(messageType, data)
54+
}
55+
3956
const (
4057
// audioTransformWSReadLimit is the per-message ceiling on inbound WS
4158
// frames. With 16 kHz / 256-sample / s16-stereo (1024 B/frame) the
@@ -158,10 +175,11 @@ func AudioTransformEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader,
158175
// @Router /audio/transformations/stream [get]
159176
func AudioTransformStreamEndpoint(app *application.Application) echo.HandlerFunc {
160177
return func(c echo.Context) error {
161-
ws, err := audioTransformWSUpgrader.Upgrade(c.Response(), c.Request(), nil)
178+
rawWS, err := audioTransformWSUpgrader.Upgrade(c.Response(), c.Request(), nil)
162179
if err != nil {
163180
return err
164181
}
182+
ws := &lockedConn{Conn: rawWS}
165183
defer func() { _ = ws.Close() }()
166184
ws.SetReadLimit(audioTransformWSReadLimit)
167185

@@ -404,7 +422,7 @@ func splitStereoFrameInto(buf []byte, fmt_ proto.AudioTransformStreamConfig_Samp
404422
return *audio, *ref
405423
}
406424

407-
func sendWSError(ws *websocket.Conn, msg string) {
425+
func sendWSError(ws *lockedConn, msg string) {
408426
payload, _ := json.Marshal(schema.AudioTransformStreamControl{
409427
Type: schema.AudioTransformCtrlError,
410428
Error: msg,

0 commit comments

Comments
 (0)