From 8bb738207a428017daa7ee7a4cb853f09f127f50 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 16:00:56 +0200 Subject: [PATCH 1/8] fix: probe only read-idle WebSockets --- README.md | 18 ++-- jaws.go | 12 ++- lib/wire/wsio.go | 166 ++++++++++++++++++++---------- lib/wire/wsio_test.go | 234 +++++++++++++++++++++++++++--------------- request.go | 13 ++- request_test.go | 120 ++++++++++++++++------ serve.go | 8 +- 7 files changed, 380 insertions(+), 191 deletions(-) diff --git a/README.md b/README.md index db927318..a7a9f9f4 100644 --- a/README.md +++ b/README.md @@ -432,9 +432,11 @@ whole-second samples from the epoch established by `jaws.New()`. Retirement is checked only during maintenance passes, so it is not timed precisely from those events. -When `Jaws.WebSocketPingInterval` is positive, the same duration is passed -directly as each keepalive ping's timeout on an active WebSocket. Ping timing does -not use those activity samples or the maintenance schedule. +A WebSocket read that remains pending for `Jaws.WebSocketPingInterval` triggers +a keepalive ping. `requestTimeout` is passed directly as the ping timeout. +Data or a successful ping starts a new interval; time spent delivering an +already-read message for processing does not count as read-idle time. This timing +does not use the initial-render activity samples or the maintenance schedule. `*Request` values are borrowed lifecycle objects. Do not store them in application state or pass them to background goroutines; copy the required @@ -532,13 +534,13 @@ reported through `MustLog()`, which panics when no logger is configured. ### WebSocket keepalive ping -JaWS can periodically ping active WebSocket connections to detect peers -that disappeared without a close handshake. +JaWS can ping read-idle WebSocket connections to detect peers that disappeared +without a close handshake. Incoming data and successful pings defer the next +probe, and JaWS does not probe while delivering an already-read message for +processing. Set `Jaws.WebSocketPingInterval` to control this. The default is -`jaws.DefaultWebSocketPingInterval` (1 minute). A non-positive value disables -pings; the application must then detect and cancel unresponsive Requests to -bound queued updates. +`jaws.DefaultWebSocketPingInterval` (1 minute), and the value must be positive. ### Safe to call before `Serve()` diff --git a/jaws.go b/jaws.go index 7548492e..b1c14013 100644 --- a/jaws.go +++ b/jaws.go @@ -51,7 +51,7 @@ const ( // DefaultUpdateInterval is the default browser update interval. DefaultUpdateInterval = time.Millisecond * 100 - // DefaultWebSocketPingInterval is the default WebSocket keepalive ping interval. + // DefaultWebSocketPingInterval is the default WebSocket read-idle interval. DefaultWebSocketPingInterval = time.Minute // DefaultWebSocketTimeout is the timeout [Jaws.Serve] passes to [Jaws.ServeWithTimeout]. @@ -117,11 +117,13 @@ type Jaws struct { // function must return promptly and must not synchronously call this Jaws or // one of its Requests, or wait for work that does. See [Request.SetContext]. BaseContext context.Context - // WebSocketPingInterval controls keepalive pings on active WebSocket connections. + // WebSocketPingInterval controls read-idle keepalive pings. // - // It defaults to [DefaultWebSocketPingInterval]. A non-positive value disables - // pings; the application must then detect and cancel each unresponsive - // [Request] to bound queued updates. + // When a WebSocket read remains pending for this interval, JaWS pings the peer. + // A data message or successful ping restarts the interval. Time spent delivering + // an already-read message for processing does not count as read-idle time. + // + // It must be positive and defaults to [DefaultWebSocketPingInterval]. WebSocketPingInterval time.Duration MaxPendingRequestsPerIP int // Maximum number of unclaimed Requests per client IP. Defaults to DefaultMaxPendingRequestsPerIP. Set <=0 to disable the cap. webSocketTimeout time.Duration // timeout duration passed to ServeWith diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index c602e56f..43ed781e 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -5,6 +5,7 @@ import ( "context" "errors" "io" + "sync" "time" "github.com/coder/websocket" @@ -20,37 +21,132 @@ const writeBatchLimit = 32 * 1024 // Records are LF-terminated and delivered in order. A text message may contain // multiple records; malformed records are skipped independently. // +// A WebSocket read that remains pending for idleInterval triggers a ping bounded +// by pingTimeout. A successful pong starts another idle interval for the pending +// read. Time spent parsing or delivering an already-read message is not idle +// time. Data received during a pending ping supersedes a failed ping. +// idleInterval and pingTimeout must be positive. +// // Closes incomingMsgCh on exit. // -// Canceling ctx or closing doneCh interrupts reads in progress and is not -// reported through ccf. +// Canceling ctx or closing doneCh interrupts reads and pings in progress and is +// not reported through ccf. // // ccf may be nil, in which case errors are not reported and only the loop exits. -func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, incomingMsgCh chan<- WsMsg, ws *websocket.Conn) { - var typ websocket.MessageType - var txt []byte - var err error - defer close(incomingMsgCh) +func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, incomingMsgCh chan<- WsMsg, idleInterval, pingTimeout time.Duration, ws *websocket.Conn) { ctx, cancel := contextWithDone(ctx, doneCh) - defer cancel() - for err == nil { - // Only parse on a successful read; on error ws.Read returns no usable - // payload and the loop exits because the for condition fails. - if typ, txt, err = ws.Read(ctx); err == nil && typ == websocket.MessageText { - for record := range bytes.Lines(txt) { - if msg, ok := Parse(record); ok { + readResultCh := make(chan wsReadResult) + pingResultCh := make(chan error, 1) + var workers sync.WaitGroup + // coder/websocket requires a Reader to run concurrently with Ping. Keeping + // socket reads in one worker lets local delivery pause the idle timer while a + // pending read still handles control frames. The unbuffered channel bounds + // read-ahead during delivery to one complete WebSocket message. + workers.Go(func() { readWebSocket(ctx, readResultCh, ws) }) + + idleTimer := time.NewTimer(idleInterval) + idleTimerCh := idleTimer.C + armIdleTimer := func() { + idleTimer.Reset(idleInterval) + idleTimerCh = idleTimer.C + } + stopIdleTimer := func() { + idleTimer.Stop() + idleTimerCh = nil + } + var activityDuringPing bool + handleRead := func(result wsReadResult) (ok bool) { + pinging := idleTimerCh == nil + stopIdleTimer() + if result.err != nil { + reportError(ctx, doneCh, ccf, result.err) + return + } + if pinging { + activityDuringPing = true + } + if result.typ == websocket.MessageText { + for record := range bytes.Lines(result.txt) { + if msg, parsed := Parse(record); parsed { select { case <-ctx.Done(): return - case <-doneCh: - return case incomingMsgCh <- msg: } } } } + if !pinging { + armIdleTimer() + } + ok = true + return + } + + defer func() { + cancel() + workers.Wait() + stopIdleTimer() + close(incomingMsgCh) + }() + + for { + select { + case <-ctx.Done(): + return + case result := <-readResultCh: + if !handleRead(result) { + return + } + case <-idleTimerCh: + idleTimerCh = nil + activityDuringPing = false + // Ping waits for its pong, so it must not delay data that arrives while + // the probe is in flight. + workers.Go(func() { + pingctx, pingcancel := context.WithTimeout(ctx, pingTimeout) + err := ws.Ping(pingctx) + pingcancel() + pingResultCh <- err + }) + case err := <-pingResultCh: + if err != nil && !activityDuringPing { + // Activity already waiting at the deadline supersedes the probe even + // when its pong could not be consumed first. + select { + case result := <-readResultCh: + if !handleRead(result) { + return + } + default: + reportError(ctx, doneCh, ccf, err) + return + } + } + activityDuringPing = false + armIdleTimer() + } + } +} + +type wsReadResult struct { + typ websocket.MessageType + txt []byte + err error +} + +func readWebSocket(ctx context.Context, resultCh chan<- wsReadResult, ws *websocket.Conn) { + for { + typ, txt, err := ws.Read(ctx) + select { + case <-ctx.Done(): + return + case resultCh <- wsReadResult{typ: typ, txt: txt, err: err}: + } + if err != nil { + return + } } - reportError(ctx, doneCh, ccf, err) } // WriteLoop formats messages read from outboundMsgCh and writes them to the @@ -88,42 +184,6 @@ func WriteLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan s reportError(ctx, doneCh, ccf, err) } -// PingLoop sends periodic WebSocket pings and reports ping errors through ccf. -// -// Returns immediately when interval is non-positive. -// -// Canceling ctx or closing doneCh interrupts pings in progress and is not -// reported through ccf. -// -// ccf may be nil, in which case errors are not reported and only the loop exits. -func PingLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, interval, timeout time.Duration, ws *websocket.Conn) { - if interval <= 0 { - // A non-positive interval disables pinging: return without calling ccf, since - // there is no ping error to report and cancelling the connection would be wrong - // (the ctx.Done and doneCh cases below likewise return without ccf). - return - } - ctx, cancel := contextWithDone(ctx, doneCh) - defer cancel() - t := time.NewTicker(interval) - defer t.Stop() - - var err error - for err == nil { - select { - case <-ctx.Done(): - return - case <-doneCh: - return - case <-t.C: - pingctx, pingcancel := context.WithTimeout(ctx, timeout) - err = ws.Ping(pingctx) - pingcancel() - } - } - reportError(ctx, doneCh, ccf, err) -} - func contextWithDone(ctx context.Context, doneCh <-chan struct{}) (ioctx context.Context, cancel context.CancelFunc) { ioctx, cancel = context.WithCancel(ctx) go func() { diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index b5f09758..117d2430 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "slices" "strings" + "sync/atomic" "testing" "testing/synctest" "time" @@ -32,7 +33,7 @@ func TestReadLoop_RespectsContextDone(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, nil, jawsDoneCh, inCh, server) + ReadLoop(ctx, nil, jawsDoneCh, inCh, time.Hour, time.Hour, server) }() writeCtx, writeCancel := context.WithTimeout(t.Context(), 3*time.Second) @@ -63,7 +64,7 @@ func TestReadLoop_RespectsDone(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) close(loopDone) }() @@ -91,7 +92,7 @@ func TestReadLoop_RespectsDoneWhileReading(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) close(loopDone) }() @@ -144,7 +145,7 @@ func TestReadWriteLoop_RoundTrip(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, nil, doneCh, inCh, server) + ReadLoop(ctx, nil, doneCh, inCh, time.Hour, time.Hour, server) }() writeDoneCh := make(chan struct{}) go func() { @@ -184,7 +185,7 @@ func TestReadLoop_SkipsMalformedRecords(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) close(loopDone) }() if err := client.Write(ctx, websocket.MessageText, payload); err != nil { @@ -230,7 +231,7 @@ func TestReadLoop_BatchedDeliveryIsInterruptible(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) close(loopDone) }() if err := client.Write(ctx, websocket.MessageText, payload); err != nil { @@ -264,6 +265,141 @@ func TestReadLoop_BatchedDeliveryIsInterruptible(t *testing.T) { } } +func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + doneCh := make(chan struct{}) + inCh := make(chan WsMsg) + var pingCount atomic.Int32 + client, server := pipeWithDialOptions(t, websocket.DialOptions{ + OnPingReceived: func(context.Context, []byte) bool { + pingCount.Add(1) + return true + }, + }) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + client.CloseRead(ctx) + go func() { + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Second, server) + close(loopDone) + }() + + want := WsMsg{Jid: 1, What: what.Input, Data: "value"} + if err := client.Write(ctx, websocket.MessageText, want.Append(nil)); err != nil { + t.Fatal(err) + } + // The complete message is waiting for the application. This local delivery + // delay is not peer inactivity, so advancing well beyond both ping durations + // must not send a probe. + synctest.Wait() + time.Sleep(3 * time.Second) + synctest.Wait() + if got := pingCount.Load(); got != 0 { + t.Fatalf("pings while delivering = %d, want 0", got) + } + + if got := <-inCh; got != want { + t.Fatalf("message = %+v, want %+v", got, want) + } + synctest.Wait() + time.Sleep(3 * time.Second) + synctest.Wait() + if got := pingCount.Load(); got < 2 { + t.Fatalf("successful idle pings = %d, want at least 2", got) + } + if err := ctx.Err(); err != nil { + t.Fatalf("parent context was canceled: %v", err) + } + + close(doneCh) + synctest.Wait() + assertClosedNow(t, loopDone, "ReadLoop") + }) +} + +func TestReadLoop_ReadActivitySupersedesPingFailure(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + doneCh := make(chan struct{}) + inCh := make(chan WsMsg, 1) + firstPingCh := make(chan struct{}) + var pingCount atomic.Int32 + client, server := pipeWithDialOptions(t, websocket.DialOptions{ + OnPingReceived: func(context.Context, []byte) bool { + if pingCount.Add(1) == 1 { + close(firstPingCh) + return false + } + return true + }, + }) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + client.CloseRead(ctx) + go func() { + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, 10*time.Second, server) + close(loopDone) + }() + + <-firstPingCh + want := WsMsg{Jid: 1, What: what.Input, Data: "active"} + sentAt := time.Now() + if err := client.Write(ctx, websocket.MessageText, want.Append(nil)); err != nil { + t.Fatal(err) + } + if got := <-inCh; got != want { + t.Fatalf("message = %+v, want %+v", got, want) + } + if delay := time.Since(sentAt); delay != 0 { + t.Fatalf("delivery during ping was delayed by %v", delay) + } + // The first ping deliberately receives no pong. The completed data read + // supersedes that probe, so its stale error must not cancel the connection. + synctest.Wait() + time.Sleep(10 * time.Second) + synctest.Wait() + if err := ctx.Err(); err != nil { + t.Fatalf("read activity did not supersede ping failure: %v", context.Cause(ctx)) + } + + close(doneCh) + synctest.Wait() + assertClosedNow(t, loopDone, "ReadLoop") + }) +} + +func TestReadLoop_ReportsUnresponsivePeer(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + doneCh := make(chan struct{}) + inCh := make(chan WsMsg) + client, server := pipeWithDialOptions(t, websocket.DialOptions{ + OnPingReceived: func(context.Context, []byte) bool { + return false // Read the ping but deliberately suppress the required pong. + }, + }) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + client.CloseRead(ctx) + go func() { + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Second, server) + close(loopDone) + }() + + // The peer consumes the ping but violates the protocol by withholding its pong. + time.Sleep(2 * time.Second) + synctest.Wait() + assertClosedNow(t, loopDone, "ReadLoop") + if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("context cause = %T(%v), want deadline exceeded", err, err) + } + }) +} + func TestWriteLoop_SendsThePayload(t *testing.T) { outCh := make(chan WsMsg) jawsDoneCh := make(chan struct{}) @@ -583,7 +719,7 @@ func TestReadLoop_ReportsError(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, cancel, jawsDoneCh, inCh, server) + ReadLoop(ctx, cancel, jawsDoneCh, inCh, time.Hour, time.Hour, server) }() waitDone(t, readDoneCh, "ReadLoop after read error") @@ -610,59 +746,11 @@ func TestReportError_IgnoresContextDone(t *testing.T) { }, errors.New("websocket closed")) } -func TestPingLoop_NonPositiveIntervalReturns(t *testing.T) { - client, server := pipe(t) - defer func() { _ = client.CloseNow() }() - defer func() { _ = server.CloseNow() }() - - // A non-positive interval must return immediately without starting a ticker or - // invoking ccf; calling PingLoop directly (not in a goroutine) proves it does - // not block, and a non-nil ccf would panic if it were called. - PingLoop(t.Context(), nil, make(chan struct{}), 0, time.Millisecond, server) -} - -func TestPingLoop_RespectsContextDone(t *testing.T) { - jawsDoneCh := make(chan struct{}) - client, server := pipe(t) - defer func() { _ = client.CloseNow() }() - defer func() { _ = server.CloseNow() }() - - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - - pingDoneCh := make(chan struct{}) - go func() { - defer close(pingDoneCh) - PingLoop(ctx, nil, jawsDoneCh, time.Millisecond*10, time.Millisecond*10, server) - }() - - cancel() - waitDone(t, pingDoneCh, "PingLoop after context cancel") -} - -func TestPingLoop_RespectsDone(t *testing.T) { - jawsDoneCh := make(chan struct{}) - client, server := pipe(t) - defer func() { _ = client.CloseNow() }() - defer func() { _ = server.CloseNow() }() - - ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) - defer cancel() - - pingDoneCh := make(chan struct{}) - go func() { - defer close(pingDoneCh) - PingLoop(ctx, nil, jawsDoneCh, time.Millisecond, time.Millisecond, server) - }() - - close(jawsDoneCh) - waitDone(t, pingDoneCh, "PingLoop after done close") -} - -func TestPingLoop_RespectsDoneWhileWaitingForPong(t *testing.T) { +func TestReadLoop_RespectsDoneWhileWaitingForPong(t *testing.T) { synctest.Test(t, func(t *testing.T) { ctx, cancel := context.WithCancelCause(t.Context()) doneCh := make(chan struct{}) + inCh := make(chan WsMsg) pingSeen := make(chan struct{}) client, server := pipeWithDialOptions(t, websocket.DialOptions{ OnPingReceived: func(context.Context, []byte) bool { @@ -675,47 +763,23 @@ func TestPingLoop_RespectsDoneWhileWaitingForPong(t *testing.T) { client.CloseRead(ctx) go func() { - PingLoop(ctx, cancel, doneCh, time.Second, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Hour, server) close(loopDone) }() <-pingSeen - // PingLoop is waiting for the deliberately omitted pong. + // The idle watchdog is waiting for the deliberately omitted pong while its + // socket reader remains in the concurrent read required by Ping. synctest.Wait() close(doneCh) synctest.Wait() - assertClosedNow(t, loopDone, "PingLoop") + assertClosedNow(t, loopDone, "ReadLoop") if err := ctx.Err(); err != nil { t.Fatalf("parent context was canceled: %v", err) } }) } -func TestPingLoop_ReportsErrorWhenPeerDoesNotPong(t *testing.T) { - jawsDoneCh := make(chan struct{}) - client, server := pipe(t) - defer func() { _ = client.CloseNow() }() - defer func() { _ = server.CloseNow() }() - - ctx, cancel := context.WithCancelCause(t.Context()) - - pingDoneCh := make(chan struct{}) - go func() { - defer close(pingDoneCh) - PingLoop(ctx, cancel, jawsDoneCh, 20*time.Millisecond, 20*time.Millisecond, server) - }() - - waitDone(t, pingDoneCh, "PingLoop after ping timeout") - - err := context.Cause(ctx) - if err == nil { - t.Fatal("expected context cause from ping failure") - } - if !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("%T(%v)", err, err) - } -} - func waitDone(t *testing.T, doneCh <-chan struct{}, what string) { t.Helper() select { diff --git a/request.go b/request.go index b36de996..edf1ad4e 100644 --- a/request.go +++ b/request.go @@ -1146,7 +1146,7 @@ func (rq *Request) stopServe() { // runWebSocket subscribes rq, runs its connect callback, and processes the // accepted WebSocket when the callback succeeds. -func (rq *Request) runWebSocket(ws *websocket.Conn, pingInterval, wsTimeout time.Duration) (err error) { +func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, pingTimeout time.Duration) (err error) { // Subscribe before onConnect so broadcasts from the callback are buffered for // this Request. Browser input and outbound writes do not start until the // callback succeeds. @@ -1177,9 +1177,8 @@ func (rq *Request) runWebSocket(ws *websocket.Conn, pingInterval, wsTimeout time // cancellation; the canceled context suppresses errors from the other loops. cancelRequest := rq.cancel outboundMsgCh := make(chan wire.WsMsg, cap(pendingSubscription)) - go wire.ReadLoop(ctx, cancelRequest, rq.Jaws.Done(), incomingMsgCh, ws) // closes incomingMsgCh - go wire.WriteLoop(ctx, cancelRequest, rq.Jaws.Done(), outboundMsgCh, ws) // calls ws.Close() - go wire.PingLoop(ctx, cancelRequest, rq.Jaws.Done(), pingInterval, wsTimeout, ws) + go wire.ReadLoop(ctx, cancelRequest, rq.Jaws.Done(), incomingMsgCh, idleInterval, pingTimeout, ws) // closes incomingMsgCh + go wire.WriteLoop(ctx, cancelRequest, rq.Jaws.Done(), outboundMsgCh, ws) // calls ws.Close() broadcastMsgCh := pendingSubscription pendingSubscription = nil // Production deliberately discards the recovered value so a loop panic stays @@ -1202,8 +1201,8 @@ func (rq *Request) runWebSocket(ws *websocket.Conn, pingInterval, wsTimeout time func (rq *Request) ServeHTTP(w http.ResponseWriter, r *http.Request) { if rq.startServe() { defer rq.stopServe() - pingInterval := rq.Jaws.WebSocketPingInterval - wsTimeout := rq.Jaws.getWebSocketTimeout() + idleInterval := rq.Jaws.WebSocketPingInterval + pingTimeout := rq.Jaws.getWebSocketTimeout() if strings.HasSuffix(r.URL.Path, "/noscript") { w.WriteHeader(http.StatusNoContent) rq.cancel(ErrJavascriptDisabled) @@ -1230,7 +1229,7 @@ func (rq *Request) ServeHTTP(w http.ResponseWriter, r *http.Request) { ws, err = websocket.Accept(acceptWriter, acceptRequest, nil) if err == nil { ws.SetReadLimit(webSocketReadLimit) - if err = rq.runWebSocket(ws, pingInterval, wsTimeout); err != nil { + if err = rq.runWebSocket(ws, idleInterval, pingTimeout); err != nil { // A ConnectFn failure is terminal. Cancel before touching the socket so // a non-reading peer cannot retain the Request, then close without a // handshake because no WebSocket processing loops were started. diff --git a/request_test.go b/request_test.go index 25603088..d3ffb500 100644 --- a/request_test.go +++ b/request_test.go @@ -4320,10 +4320,37 @@ func TestWS_PingDisconnectsUnresponsiveClient(t *testing.T) { waitForRequestCount(t, ts.jw, 0, testTimeout) } -func TestWS_PingDisabledKeepsIdleConnection(t *testing.T) { +func TestWS_SlowLocalUpdateDoesNotDisconnectResponsiveClient(t *testing.T) { + const ( + idleInterval = 40 * time.Millisecond + pingTimeout = 20 * time.Millisecond + stallTime = 4 * (idleInterval + pingTimeout) + ) + ts := newTestServer(t) defer ts.Close() - ts.jw.WebSocketPingInterval = 0 + ts.jw.WebSocketPingInterval = idleInterval + ts.jw.webSocketTimeout = pingTimeout + + updateStarted := make(chan struct{}) + releaseUpdateCh := make(chan struct{}) + releaseUpdate := sync.OnceFunc(func() { close(releaseUpdateCh) }) + defer releaseUpdate() + + item := &testUi{} + item.updateFn = func(*Element) { + // Register performs the first update synchronously. Block only the dirty + // update run by the Request processing loop. + if atomic.LoadInt32(&item.updateCalled) == 2 { + close(updateStarted) + <-releaseUpdateCh + } + } + inputHandled := make(chan string, 1) + id := testRequestWriter{rq: ts.rq, Writer: io.Discard}.Register(item, func(_ *Element, value string) error { + inputHandled <- value + return nil + }) conn, resp, err := ts.Dial() if err != nil { @@ -4332,6 +4359,26 @@ func TestWS_PingDisabledKeepsIdleConnection(t *testing.T) { if resp.StatusCode != http.StatusSwitchingProtocols { t.Error(resp.StatusCode) } + // Browsers handle Ping and Pong control frames below the JavaScript API. + // Keep a real client Reader active so this peer drains both application and + // control frames while the server-side Request loop is stalled. + clientReadCtx, cancelClientRead := context.WithCancel(ts.ctx) + clientReadErrCh := make(chan error, 1) + clientReadDoneCh := make(chan struct{}) + go func() { + defer close(clientReadDoneCh) + for { + if _, _, readErr := conn.Read(clientReadCtx); readErr != nil { + clientReadErrCh <- readErr + return + } + } + }() + defer func() { + cancelClientRead() + _ = conn.CloseNow() + <-clientReadDoneCh + }() select { case <-ts.connectedCh: @@ -4339,20 +4386,49 @@ func TestWS_PingDisabledKeepsIdleConnection(t *testing.T) { t.Fatal("timeout waiting for websocket connect") } - // This test drives a real WebSocket connection (real network I/O), so it runs - // on the real clock and cannot use a synctest bubble. Give the connection a - // moment to settle, then confirm the request counts are stable. - time.Sleep(150 * time.Millisecond) - total, active := ts.jw.RequestCounts() - if total != 1 || active != 1 { - t.Fatalf("RequestCounts() = %d, %d, want 1, 1", total, active) + ts.rq.Dirty(item) + select { + case <-updateStarted: + case <-time.After(testTimeout): + t.Fatal("timeout waiting for dirty update to start") } - if got := ts.jw.RequestCount(); got != total { - t.Fatalf("RequestCount() = %d, want %d", got, total) + + writeCtx, writeCancel := context.WithTimeout(ts.ctx, testTimeout) + inputMsg := wire.WsMsg{Jid: id, What: what.Input, Data: "value"} + err = conn.Write(writeCtx, websocket.MessageText, inputMsg.Append(nil)) + writeCancel() + if err != nil { + t.Fatal(err) } - _ = conn.CloseNow() - waitForRequestCounts(t, ts.jw, 0, 0, testTimeout) + // updateStarted proves the Request loop cannot receive the input. Keep the + // real loopback connection in that state across several complete probe windows: + // once ReadLoop receives the frame, it must wait for the local update to finish, + // but that local wait is not evidence that the client stopped responding. + requestCtx := ts.rq.Context() + stallTimer := time.NewTimer(stallTime) + select { + case <-requestCtx.Done(): + stallTimer.Stop() + t.Fatalf("responsive client disconnected during local update: %v", context.Cause(requestCtx)) + case readErr := <-clientReadErrCh: + stallTimer.Stop() + t.Fatalf("responsive client reader stopped during local update: %v", readErr) + case <-stallTimer.C: + } + + releaseUpdate() + select { + case value := <-inputHandled: + if value != "value" { + t.Fatalf("input value = %q, want %q", value, "value") + } + case <-time.After(testTimeout): + t.Fatal("timeout waiting for input after update completed") + } + if err := requestCtx.Err(); err != nil { + t.Fatalf("request canceled after processing input: %v", context.Cause(requestCtx)) + } } // TestWS_SetContextCancellationClosesIdleConnection proves production @@ -4364,7 +4440,7 @@ func TestWS_SetContextCancellationClosesIdleConnection(t *testing.T) { logger := &eventErrorLogger{} ts := newTestServerWithSession(t, false, logger) defer ts.Close() - ts.jw.WebSocketPingInterval = 0 + ts.jw.WebSocketPingInterval = time.Hour observed := &observedDoneContext{ Context: ts.rq.Context(), @@ -4453,22 +4529,6 @@ func waitForRequestCount(t *testing.T, jw *Jaws, want int, timeout time.Duration } } -func waitForRequestCounts(t *testing.T, jw *Jaws, wantTotal, wantActive int, timeout time.Duration) { - t.Helper() - deadline := time.Now().Add(timeout) - for { - total, active := jw.RequestCounts() - if total == wantTotal && active == wantActive { - return - } - if time.Now().After(deadline) { - total, active = jw.RequestCounts() - t.Fatalf("RequestCounts() = %d, %d, want %d, %d", total, active, wantTotal, wantActive) - } - time.Sleep(5 * time.Millisecond) - } -} - // TestReleaseBuffersLockedZeroesWsQueue verifies releaseBuffersLocked releases the // queued wire message payloads before the storage is returned to the buffer pool, // mirroring its clear() of todoDirt and elems. A bare [:0] reslice would leave the diff --git a/serve.go b/serve.go index a33491b2..05f5c270 100644 --- a/serve.go +++ b/serve.go @@ -45,9 +45,11 @@ func (jw *Jaws) getWebSocketTimeout() (t time.Duration) { // whole-second samples from the epoch established by [New]. Retirement is checked // only during maintenance passes, so it is not timed precisely from those events. // -// When [Jaws.WebSocketPingInterval] is positive, each keepalive ping on an active -// WebSocket uses requestTimeout directly as its timeout. Ping timing does not use -// those activity samples or the maintenance schedule. +// A WebSocket read that remains pending for [Jaws.WebSocketPingInterval] triggers +// a keepalive ping. requestTimeout bounds each ping. Data or a successful ping +// starts a new interval; time spent delivering an already-read message for +// processing does not count as read-idle time. This timing does not use the +// initial-render activity samples or maintenance schedule. // // It is intended to run on its own goroutine and returns when [Jaws.Close] is // called. Errors reported through [Jaws.Log] are queued without waiting for From f46bd9a8c5e813b2c15bbf40fa1becb5c52bfd7b Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 16:57:48 +0200 Subject: [PATCH 2/8] fix: bound stalled WebSocket writes --- README.md | 7 +++-- lib/wire/wsio.go | 26 +++++++---------- lib/wire/wsio_test.go | 54 ++++++++++++++++++++++++++-------- request.go | 10 +++---- request_test.go | 67 +++++++++++++++++++++++++++++++++++++++++++ serve.go | 8 +++--- 6 files changed, 133 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index a7a9f9f4..ba284139 100644 --- a/README.md +++ b/README.md @@ -433,8 +433,8 @@ checked only during maintenance passes, so it is not timed precisely from those events. A WebSocket read that remains pending for `Jaws.WebSocketPingInterval` triggers -a keepalive ping. `requestTimeout` is passed directly as the ping timeout. -Data or a successful ping starts a new interval; time spent delivering an +a keepalive ping. `requestTimeout` bounds each ping and each outbound WebSocket +write. Data or a successful ping starts a new interval; time spent delivering an already-read message for processing does not count as read-idle time. This timing does not use the initial-render activity samples or the maintenance schedule. @@ -537,7 +537,8 @@ reported through `MustLog()`, which panics when no logger is configured. JaWS can ping read-idle WebSocket connections to detect peers that disappeared without a close handshake. Incoming data and successful pings defer the next probe, and JaWS does not probe while delivering an already-read message for -processing. +processing. An outbound WebSocket write that remains blocked for the configured +request timeout also ends the Request. Set `Jaws.WebSocketPingInterval` to control this. The default is `jaws.DefaultWebSocketPingInterval` (1 minute), and the value must be positive. diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index 43ed781e..62daf568 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -24,7 +24,7 @@ const writeBatchLimit = 32 * 1024 // A WebSocket read that remains pending for idleInterval triggers a ping bounded // by pingTimeout. A successful pong starts another idle interval for the pending // read. Time spent parsing or delivering an already-read message is not idle -// time. Data received during a pending ping supersedes a failed ping. +// time. Data processed while a ping is pending supersedes a failed ping. // idleInterval and pingTimeout must be positive. // // Closes incomingMsgCh on exit. @@ -56,6 +56,8 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st } var activityDuringPing bool handleRead := func(result wsReadResult) (ok bool) { + // A nil timer channel denotes the one in-flight ping. Capture that state + // before stopIdleTimer also clears the channel during ordinary delivery. pinging := idleTimerCh == nil stopIdleTimer() if result.err != nil { @@ -100,7 +102,6 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st } case <-idleTimerCh: idleTimerCh = nil - activityDuringPing = false // Ping waits for its pong, so it must not delay data that arrives while // the probe is in flight. workers.Go(func() { @@ -111,17 +112,8 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st }) case err := <-pingResultCh: if err != nil && !activityDuringPing { - // Activity already waiting at the deadline supersedes the probe even - // when its pong could not be consumed first. - select { - case result := <-readResultCh: - if !handleRead(result) { - return - } - default: - reportError(ctx, doneCh, ccf, err) - return - } + reportError(ctx, doneCh, ccf, err) + return } activityDuringPing = false armIdleTimer() @@ -154,13 +146,15 @@ func readWebSocket(ctx context.Context, resultCh chan<- wsReadResult, ws *websoc // // Consecutive queued records may be coalesced into one text message. // +// Each write is bounded by writeTimeout, which must be positive. +// // Closes the WebSocket on exit. // // Canceling ctx or closing doneCh interrupts writes in progress and is not // reported through ccf. // // ccf may be nil, in which case errors are not reported and only the loop exits. -func WriteLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, outboundMsgCh <-chan WsMsg, ws *websocket.Conn) { +func WriteLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, outboundMsgCh <-chan WsMsg, writeTimeout time.Duration, ws *websocket.Conn) { defer func() { _ = ws.Close(websocket.StatusNormalClosure, "") }() ctx, cancel := contextWithDone(ctx, doneCh) defer cancel() @@ -175,10 +169,12 @@ func WriteLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan s if !ok { return } + writectx, writecancel := context.WithTimeout(ctx, writeTimeout) var wc io.WriteCloser - if wc, err = ws.Writer(ctx, websocket.MessageText); err == nil { + if wc, err = ws.Writer(writectx, websocket.MessageText); err == nil { err = writeData(wc, msg, outboundMsgCh) } + writecancel() } } reportError(ctx, doneCh, ccf, err) diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index 117d2430..b219c203 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -150,7 +150,7 @@ func TestReadWriteLoop_RoundTrip(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, doneCh, outCh, client) + WriteLoop(ctx, nil, doneCh, outCh, time.Hour, client) }() var got []WsMsg @@ -306,8 +306,8 @@ func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { synctest.Wait() time.Sleep(3 * time.Second) synctest.Wait() - if got := pingCount.Load(); got < 2 { - t.Fatalf("successful idle pings = %d, want at least 2", got) + if got := pingCount.Load(); got != 3 { + t.Fatalf("successful idle pings = %d, want 3", got) } if err := ctx.Err(); err != nil { t.Fatalf("parent context was canceled: %v", err) @@ -413,7 +413,7 @@ func TestWriteLoop_SendsThePayload(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() var mt websocket.MessageType @@ -468,7 +468,7 @@ func TestWriteLoop_ConcatenatesMessages(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() mt, b, err := client.Read(ctx) @@ -503,7 +503,7 @@ func TestWriteLoop_ConcatenatesMessagesClosedChannel(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() mt, b, err := client.Read(ctx) @@ -551,7 +551,7 @@ func TestWriteLoop_SplitsAtBatchLimit(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() var frames [][]byte @@ -602,7 +602,7 @@ func TestWriteLoop_RespectsContext(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() cancel() @@ -623,7 +623,7 @@ func TestWriteLoop_RespectsDone(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() close(jawsDoneCh) @@ -645,7 +645,7 @@ func TestWriteLoop_RespectsDoneWhileWriting(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - WriteLoop(ctx, cancel, doneCh, outCh, server) + WriteLoop(ctx, cancel, doneCh, outCh, time.Hour, server) close(loopDone) }() @@ -660,6 +660,36 @@ func TestWriteLoop_RespectsDoneWhileWriting(t *testing.T) { }) } +func TestWriteLoop_ReportsUnresponsivePeer(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + doneCh := make(chan struct{}) + outCh := make(chan WsMsg, 1) + outCh <- WsMsg{ + Jid: jid.Jid(1234), + What: what.Inner, + Data: strings.Repeat("x", writeBatchLimit), + } + client, server := pipe(t) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + go func() { + WriteLoop(ctx, cancel, doneCh, outCh, time.Second, server) + close(loopDone) + }() + + // The peer does not read, so the WebSocket write remains blocked until its + // operation timeout closes the connection. + time.Sleep(time.Second) + synctest.Wait() + assertClosedNow(t, loopDone, "WriteLoop") + if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("context cause = %T(%v), want deadline exceeded", err, err) + } + }) +} + func TestWriteLoop_RespectsOutboundClosed(t *testing.T) { outCh := make(chan WsMsg) jawsDoneCh := make(chan struct{}) @@ -674,7 +704,7 @@ func TestWriteLoop_RespectsOutboundClosed(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, nil, jawsDoneCh, outCh, server) + WriteLoop(ctx, nil, jawsDoneCh, outCh, time.Hour, server) }() close(outCh) @@ -694,7 +724,7 @@ func TestWriteLoop_ReportsError(t *testing.T) { writeDoneCh := make(chan struct{}) go func() { defer close(writeDoneCh) - WriteLoop(ctx, cancel, jawsDoneCh, outCh, server) + WriteLoop(ctx, cancel, jawsDoneCh, outCh, time.Hour, server) }() outCh <- WsMsg{Jid: jid.Jid(1234)} diff --git a/request.go b/request.go index edf1ad4e..5501f4fb 100644 --- a/request.go +++ b/request.go @@ -1146,7 +1146,7 @@ func (rq *Request) stopServe() { // runWebSocket subscribes rq, runs its connect callback, and processes the // accepted WebSocket when the callback succeeds. -func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, pingTimeout time.Duration) (err error) { +func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, wsTimeout time.Duration) (err error) { // Subscribe before onConnect so broadcasts from the callback are buffered for // this Request. Browser input and outbound writes do not start until the // callback succeeds. @@ -1177,8 +1177,8 @@ func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, pingTimeout ti // cancellation; the canceled context suppresses errors from the other loops. cancelRequest := rq.cancel outboundMsgCh := make(chan wire.WsMsg, cap(pendingSubscription)) - go wire.ReadLoop(ctx, cancelRequest, rq.Jaws.Done(), incomingMsgCh, idleInterval, pingTimeout, ws) // closes incomingMsgCh - go wire.WriteLoop(ctx, cancelRequest, rq.Jaws.Done(), outboundMsgCh, ws) // calls ws.Close() + go wire.ReadLoop(ctx, cancelRequest, rq.Jaws.Done(), incomingMsgCh, idleInterval, wsTimeout, ws) // closes incomingMsgCh + go wire.WriteLoop(ctx, cancelRequest, rq.Jaws.Done(), outboundMsgCh, wsTimeout, ws) // calls ws.Close() broadcastMsgCh := pendingSubscription pendingSubscription = nil // Production deliberately discards the recovered value so a loop panic stays @@ -1202,7 +1202,7 @@ func (rq *Request) ServeHTTP(w http.ResponseWriter, r *http.Request) { if rq.startServe() { defer rq.stopServe() idleInterval := rq.Jaws.WebSocketPingInterval - pingTimeout := rq.Jaws.getWebSocketTimeout() + wsTimeout := rq.Jaws.getWebSocketTimeout() if strings.HasSuffix(r.URL.Path, "/noscript") { w.WriteHeader(http.StatusNoContent) rq.cancel(ErrJavascriptDisabled) @@ -1229,7 +1229,7 @@ func (rq *Request) ServeHTTP(w http.ResponseWriter, r *http.Request) { ws, err = websocket.Accept(acceptWriter, acceptRequest, nil) if err == nil { ws.SetReadLimit(webSocketReadLimit) - if err = rq.runWebSocket(ws, idleInterval, pingTimeout); err != nil { + if err = rq.runWebSocket(ws, idleInterval, wsTimeout); err != nil { // A ConnectFn failure is terminal. Cancel before touching the socket so // a non-reading peer cannot retain the Request, then close without a // handshake because no WebSocket processing loops were started. diff --git a/request_test.go b/request_test.go index d3ffb500..b40cbff6 100644 --- a/request_test.go +++ b/request_test.go @@ -4320,6 +4320,70 @@ func TestWS_PingDisconnectsUnresponsiveClient(t *testing.T) { waitForRequestCount(t, ts.jw, 0, testTimeout) } +func TestWS_WriteTimeoutDisconnectsNonReadingClient(t *testing.T) { + const writeTimeout = 50 * time.Millisecond + + ts := newTestServer(t) + defer ts.Close() + ts.jw.WebSocketPingInterval = time.Hour + ts.jw.webSocketTimeout = writeTimeout + + updateStarted := make(chan struct{}) + releaseUpdateCh := make(chan struct{}) + releaseUpdate := sync.OnceFunc(func() { close(releaseUpdateCh) }) + defer releaseUpdate() + + item := &testUi{} + item.updateFn = func(elem *Element) { + if atomic.LoadInt32(&item.updateCalled) == 2 { + close(updateStarted) + <-releaseUpdateCh + innerHTML := template.HTML(strings.Repeat("x", 256*1024)) + for range 64 { + elem.SetInner(innerHTML) + } + } + } + id := testRequestWriter{rq: ts.rq, Writer: io.Discard}.Register(item, func(*Element, string) error { return nil }) + + conn, resp, err := ts.Dial() + if err != nil { + t.Fatal(err) + } + defer func() { _ = conn.CloseNow() }() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Error(resp.StatusCode) + } + // This peer never calls Read, so it consumes neither application nor control + // frames after the handshake. + select { + case <-ts.connectedCh: + case <-time.After(testTimeout): + t.Fatal("timeout waiting for websocket connect") + } + + ts.rq.Dirty(item) + select { + case <-updateStarted: + case <-time.After(testTimeout): + t.Fatal("timeout waiting for dirty update to start") + } + + writeCtx, writeCancel := context.WithTimeout(ts.ctx, testTimeout) + inputMsg := wire.WsMsg{Jid: id, What: what.Input, Data: "value"} + err = conn.Write(writeCtx, websocket.MessageText, inputMsg.Append(nil)) + writeCancel() + if err != nil { + t.Fatal(err) + } + + // Releasing the update queues more output than the connection and bounded + // outbound channel can absorb. The peer can still send the input above, but + // its blocked receive direction must tear down the Request within writeTimeout. + releaseUpdate() + waitForRequestCount(t, ts.jw, 0, testTimeout) +} + func TestWS_SlowLocalUpdateDoesNotDisconnectResponsiveClient(t *testing.T) { const ( idleInterval = 40 * time.Millisecond @@ -4392,6 +4456,9 @@ func TestWS_SlowLocalUpdateDoesNotDisconnectResponsiveClient(t *testing.T) { case <-time.After(testTimeout): t.Fatal("timeout waiting for dirty update to start") } + if total, active := ts.jw.RequestCounts(); total != 1 || active != 1 { + t.Fatalf("RequestCounts() = %d, %d, want 1, 1", total, active) + } writeCtx, writeCancel := context.WithTimeout(ts.ctx, testTimeout) inputMsg := wire.WsMsg{Jid: id, What: what.Input, Data: "value"} diff --git a/serve.go b/serve.go index 05f5c270..1c0cb9cd 100644 --- a/serve.go +++ b/serve.go @@ -46,10 +46,10 @@ func (jw *Jaws) getWebSocketTimeout() (t time.Duration) { // only during maintenance passes, so it is not timed precisely from those events. // // A WebSocket read that remains pending for [Jaws.WebSocketPingInterval] triggers -// a keepalive ping. requestTimeout bounds each ping. Data or a successful ping -// starts a new interval; time spent delivering an already-read message for -// processing does not count as read-idle time. This timing does not use the -// initial-render activity samples or maintenance schedule. +// a keepalive ping. requestTimeout bounds each ping and each outbound WebSocket +// write. Data or a successful ping starts a new interval; time spent delivering +// an already-read message for processing does not count as read-idle time. This +// timing does not use the initial-render activity samples or maintenance schedule. // // It is intended to run on its own goroutine and returns when [Jaws.Close] is // called. Errors reported through [Jaws.Log] are queued without waiting for From 931b1664eb7236eb3c6c3ab80a47f69b555d3586 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 17:02:35 +0200 Subject: [PATCH 3/8] refactor: inline WebSocket read handling --- lib/wire/wsio.go | 53 +++++++++++++++++++++--------------------------- 1 file changed, 23 insertions(+), 30 deletions(-) diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index 62daf568..a1648ab7 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -55,35 +55,6 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st idleTimerCh = nil } var activityDuringPing bool - handleRead := func(result wsReadResult) (ok bool) { - // A nil timer channel denotes the one in-flight ping. Capture that state - // before stopIdleTimer also clears the channel during ordinary delivery. - pinging := idleTimerCh == nil - stopIdleTimer() - if result.err != nil { - reportError(ctx, doneCh, ccf, result.err) - return - } - if pinging { - activityDuringPing = true - } - if result.typ == websocket.MessageText { - for record := range bytes.Lines(result.txt) { - if msg, parsed := Parse(record); parsed { - select { - case <-ctx.Done(): - return - case incomingMsgCh <- msg: - } - } - } - } - if !pinging { - armIdleTimer() - } - ok = true - return - } defer func() { cancel() @@ -97,9 +68,31 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st case <-ctx.Done(): return case result := <-readResultCh: - if !handleRead(result) { + // A nil timer channel denotes the one in-flight ping. Capture that state + // before stopIdleTimer also clears the channel during ordinary delivery. + pinging := idleTimerCh == nil + stopIdleTimer() + if result.err != nil { + reportError(ctx, doneCh, ccf, result.err) return } + if pinging { + activityDuringPing = true + } + if result.typ == websocket.MessageText { + for record := range bytes.Lines(result.txt) { + if msg, parsed := Parse(record); parsed { + select { + case <-ctx.Done(): + return + case incomingMsgCh <- msg: + } + } + } + } + if !pinging { + armIdleTimer() + } case <-idleTimerCh: idleTimerCh = nil // Ping waits for its pong, so it must not delay data that arrives while From d743d83e4c1a0836ff3b85569f48279422d60863 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 18:05:06 +0200 Subject: [PATCH 4/8] docs: clarify WebSocket ping interval contract --- README.md | 3 ++- jaws.go | 3 ++- lib/wire/wsio.go | 3 ++- serve.go | 2 ++ 4 files changed, 8 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index ba284139..5a9be83b 100644 --- a/README.md +++ b/README.md @@ -541,7 +541,8 @@ processing. An outbound WebSocket write that remains blocked for the configured request timeout also ends the Request. Set `Jaws.WebSocketPingInterval` to control this. The default is -`jaws.DefaultWebSocketPingInterval` (1 minute), and the value must be positive. +`jaws.DefaultWebSocketPingInterval` (1 minute). The value must be greater than +zero; non-positive values are invalid and do not disable probing. ### Safe to call before `Serve()` diff --git a/jaws.go b/jaws.go index b1c14013..29268c63 100644 --- a/jaws.go +++ b/jaws.go @@ -123,7 +123,8 @@ type Jaws struct { // A data message or successful ping restarts the interval. Time spent delivering // an already-read message for processing does not count as read-idle time. // - // It must be positive and defaults to [DefaultWebSocketPingInterval]. + // It must be greater than zero; non-positive values are invalid and do not + // disable probing. It defaults to [DefaultWebSocketPingInterval]. WebSocketPingInterval time.Duration MaxPendingRequestsPerIP int // Maximum number of unclaimed Requests per client IP. Defaults to DefaultMaxPendingRequestsPerIP. Set <=0 to disable the cap. webSocketTimeout time.Duration // timeout duration passed to ServeWith diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index a1648ab7..23d8af8e 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -25,7 +25,8 @@ const writeBatchLimit = 32 * 1024 // by pingTimeout. A successful pong starts another idle interval for the pending // read. Time spent parsing or delivering an already-read message is not idle // time. Data processed while a ping is pending supersedes a failed ping. -// idleInterval and pingTimeout must be positive. +// idleInterval and pingTimeout must be greater than zero; non-positive values +// are invalid. // // Closes incomingMsgCh on exit. // diff --git a/serve.go b/serve.go index 1c0cb9cd..254bb1da 100644 --- a/serve.go +++ b/serve.go @@ -50,6 +50,8 @@ func (jw *Jaws) getWebSocketTimeout() (t time.Duration) { // write. Data or a successful ping starts a new interval; time spent delivering // an already-read message for processing does not count as read-idle time. This // timing does not use the initial-render activity samples or maintenance schedule. +// [Jaws.WebSocketPingInterval] must be greater than zero; non-positive values +// are invalid and do not disable probing. // // It is intended to run on its own goroutine and returns when [Jaws.Close] is // called. Errors reported through [Jaws.Log] are queued without waiting for From 6de0ada4a910612a10bde7e2453e2b3d593c82e7 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 18:05:17 +0200 Subject: [PATCH 5/8] test: pin per-write WebSocket deadlines --- lib/wire/wsio_test.go | 75 +++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 73 insertions(+), 2 deletions(-) diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index b219c203..09e747fe 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -662,6 +662,8 @@ func TestWriteLoop_RespectsDoneWhileWriting(t *testing.T) { func TestWriteLoop_ReportsUnresponsivePeer(t *testing.T) { synctest.Test(t, func(t *testing.T) { + const writeTimeout = time.Second + ctx, cancel := context.WithCancelCause(t.Context()) doneCh := make(chan struct{}) outCh := make(chan WsMsg, 1) @@ -675,13 +677,24 @@ func TestWriteLoop_ReportsUnresponsivePeer(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - WriteLoop(ctx, cancel, doneCh, outCh, time.Second, server) + WriteLoop(ctx, cancel, doneCh, outCh, writeTimeout, server) close(loopDone) }() // The peer does not read, so the WebSocket write remains blocked until its // operation timeout closes the connection. - time.Sleep(time.Second) + synctest.Wait() + time.Sleep(writeTimeout - time.Nanosecond) + synctest.Wait() + select { + case <-loopDone: + t.Fatal("WriteLoop returned before writeTimeout") + default: + } + if err := context.Cause(ctx); err != nil { + t.Fatalf("context cause before writeTimeout = %v, want nil", err) + } + time.Sleep(time.Nanosecond) synctest.Wait() assertClosedNow(t, loopDone, "WriteLoop") if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { @@ -690,6 +703,64 @@ func TestWriteLoop_ReportsUnresponsivePeer(t *testing.T) { }) } +func TestWriteLoop_RenewsTimeoutForEachMessage(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const writeTimeout = time.Second + + ctx, cancel := context.WithCancelCause(t.Context()) + doneCh := make(chan struct{}) + outCh := make(chan WsMsg) + client, server := pipe(t) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + go func() { + WriteLoop(ctx, cancel, doneCh, outCh, writeTimeout, server) + close(loopDone) + }() + + outCh <- WsMsg{Jid: jid.Jid(1234), What: what.Inner, Data: "first"} + if _, _, err := client.Read(ctx); err != nil { + t.Fatal(err) + } + synctest.Wait() + + // Block a second write across the first write's former deadline. A fresh + // operation deadline leaves the loop running until this write is read. + time.Sleep(3 * writeTimeout / 4) + second := WsMsg{ + Jid: jid.Jid(1234), + What: what.Inner, + Data: strings.Repeat("x", writeBatchLimit/2), + } + outCh <- second + synctest.Wait() + time.Sleep(writeTimeout / 2) + synctest.Wait() + select { + case <-loopDone: + t.Fatal("WriteLoop reused the first write deadline") + default: + } + if err := context.Cause(ctx); err != nil { + t.Fatalf("context cause = %v, want nil", err) + } + + if _, _, err := client.Read(ctx); err != nil { + t.Fatal(err) + } + synctest.Wait() + if err := context.Cause(ctx); err != nil { + t.Fatalf("context cause = %v, want nil", err) + } + + close(outCh) + _ = client.CloseNow() + synctest.Wait() + assertClosedNow(t, loopDone, "WriteLoop") + }) +} + func TestWriteLoop_RespectsOutboundClosed(t *testing.T) { outCh := make(chan WsMsg) jawsDoneCh := make(chan struct{}) From d99ce71a2f683a49ece2945c38dc2db5428a7924 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 19:44:54 +0200 Subject: [PATCH 6/8] refactor: simplify WebSocket heartbeat coordination --- README.md | 8 ++--- lib/wire/wsio.go | 23 +++++--------- lib/wire/wsio_test.go | 73 +++++++------------------------------------ serve.go | 10 ++---- 4 files changed, 25 insertions(+), 89 deletions(-) diff --git a/README.md b/README.md index 5a9be83b..dd4281df 100644 --- a/README.md +++ b/README.md @@ -432,11 +432,9 @@ whole-second samples from the epoch established by `jaws.New()`. Retirement is checked only during maintenance passes, so it is not timed precisely from those events. -A WebSocket read that remains pending for `Jaws.WebSocketPingInterval` triggers -a keepalive ping. `requestTimeout` bounds each ping and each outbound WebSocket -write. Data or a successful ping starts a new interval; time spent delivering an -already-read message for processing does not count as read-idle time. This timing -does not use the initial-render activity samples or the maintenance schedule. +On an active WebSocket, `requestTimeout` bounds each keepalive ping and outbound +write. See [WebSocket keepalive ping](#websocket-keepalive-ping) for probe +scheduling. `*Request` values are borrowed lifecycle objects. Do not store them in application state or pass them to background goroutines; copy the required diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index 23d8af8e..0bfe2f36 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -51,16 +51,12 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st idleTimer.Reset(idleInterval) idleTimerCh = idleTimer.C } - stopIdleTimer := func() { - idleTimer.Stop() - idleTimerCh = nil - } var activityDuringPing bool defer func() { cancel() workers.Wait() - stopIdleTimer() + idleTimer.Stop() close(incomingMsgCh) }() @@ -69,17 +65,15 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st case <-ctx.Done(): return case result := <-readResultCh: - // A nil timer channel denotes the one in-flight ping. Capture that state - // before stopIdleTimer also clears the channel during ordinary delivery. - pinging := idleTimerCh == nil - stopIdleTimer() + // A nil timer channel denotes the one in-flight ping while this select is + // active. Keep it nil so parsing and delivery do not count as read-idle; + // Reset afterward discards any expiry that occurs meanwhile. + activityDuringPing = idleTimerCh == nil + idleTimerCh = nil if result.err != nil { reportError(ctx, doneCh, ccf, result.err) return } - if pinging { - activityDuringPing = true - } if result.typ == websocket.MessageText { for record := range bytes.Lines(result.txt) { if msg, parsed := Parse(record); parsed { @@ -91,7 +85,7 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st } } } - if !pinging { + if !activityDuringPing { armIdleTimer() } case <-idleTimerCh: @@ -157,8 +151,6 @@ func WriteLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan s select { case <-ctx.Done(): return - case <-doneCh: - return case msg, ok := <-outboundMsgCh: if !ok { return @@ -188,6 +180,7 @@ func contextWithDone(ctx context.Context, doneCh <-chan struct{}) (ioctx context func reportError(ctx context.Context, doneCh <-chan struct{}, ccf context.CancelCauseFunc, err error) { if ccf != nil { + // Check doneCh directly because contextWithDone propagates it asynchronously. select { case <-ctx.Done(): case <-doneCh: diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index 09e747fe..34789bb6 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -660,50 +660,7 @@ func TestWriteLoop_RespectsDoneWhileWriting(t *testing.T) { }) } -func TestWriteLoop_ReportsUnresponsivePeer(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - const writeTimeout = time.Second - - ctx, cancel := context.WithCancelCause(t.Context()) - doneCh := make(chan struct{}) - outCh := make(chan WsMsg, 1) - outCh <- WsMsg{ - Jid: jid.Jid(1234), - What: what.Inner, - Data: strings.Repeat("x", writeBatchLimit), - } - client, server := pipe(t) - loopDone := make(chan struct{}) - defer closeWireBubble(cancel, client, server)() - - go func() { - WriteLoop(ctx, cancel, doneCh, outCh, writeTimeout, server) - close(loopDone) - }() - - // The peer does not read, so the WebSocket write remains blocked until its - // operation timeout closes the connection. - synctest.Wait() - time.Sleep(writeTimeout - time.Nanosecond) - synctest.Wait() - select { - case <-loopDone: - t.Fatal("WriteLoop returned before writeTimeout") - default: - } - if err := context.Cause(ctx); err != nil { - t.Fatalf("context cause before writeTimeout = %v, want nil", err) - } - time.Sleep(time.Nanosecond) - synctest.Wait() - assertClosedNow(t, loopDone, "WriteLoop") - if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("context cause = %T(%v), want deadline exceeded", err, err) - } - }) -} - -func TestWriteLoop_RenewsTimeoutForEachMessage(t *testing.T) { +func TestWriteLoop_TimeoutIsPerWrite(t *testing.T) { synctest.Test(t, func(t *testing.T) { const writeTimeout = time.Second @@ -726,38 +683,30 @@ func TestWriteLoop_RenewsTimeoutForEachMessage(t *testing.T) { synctest.Wait() // Block a second write across the first write's former deadline. A fresh - // operation deadline leaves the loop running until this write is read. + // operation deadline leaves the loop running for a full writeTimeout. time.Sleep(3 * writeTimeout / 4) - second := WsMsg{ + outCh <- WsMsg{ Jid: jid.Jid(1234), What: what.Inner, - Data: strings.Repeat("x", writeBatchLimit/2), + Data: strings.Repeat("x", writeBatchLimit), } - outCh <- second synctest.Wait() - time.Sleep(writeTimeout / 2) + time.Sleep(writeTimeout - time.Nanosecond) synctest.Wait() select { case <-loopDone: - t.Fatal("WriteLoop reused the first write deadline") + t.Fatal("WriteLoop returned before the second write's writeTimeout") default: } if err := context.Cause(ctx); err != nil { - t.Fatalf("context cause = %v, want nil", err) - } - - if _, _, err := client.Read(ctx); err != nil { - t.Fatal(err) - } - synctest.Wait() - if err := context.Cause(ctx); err != nil { - t.Fatalf("context cause = %v, want nil", err) + t.Fatalf("context cause before writeTimeout = %v, want nil", err) } - - close(outCh) - _ = client.CloseNow() + time.Sleep(time.Nanosecond) synctest.Wait() assertClosedNow(t, loopDone, "WriteLoop") + if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("context cause = %T(%v), want deadline exceeded", err, err) + } }) } diff --git a/serve.go b/serve.go index 254bb1da..c759e042 100644 --- a/serve.go +++ b/serve.go @@ -45,13 +45,9 @@ func (jw *Jaws) getWebSocketTimeout() (t time.Duration) { // whole-second samples from the epoch established by [New]. Retirement is checked // only during maintenance passes, so it is not timed precisely from those events. // -// A WebSocket read that remains pending for [Jaws.WebSocketPingInterval] triggers -// a keepalive ping. requestTimeout bounds each ping and each outbound WebSocket -// write. Data or a successful ping starts a new interval; time spent delivering -// an already-read message for processing does not count as read-idle time. This -// timing does not use the initial-render activity samples or maintenance schedule. -// [Jaws.WebSocketPingInterval] must be greater than zero; non-positive values -// are invalid and do not disable probing. +// requestTimeout also bounds each WebSocket keepalive ping and outbound write. +// These operation deadlines are independent of maintenance timing; see +// [Jaws.WebSocketPingInterval] for probe scheduling. // // It is intended to run on its own goroutine and returns when [Jaws.Close] is // called. Errors reported through [Jaws.Log] are queued without waiting for From 5ed7f1cfb038fa14107db2c4e3ba5639157e4f23 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 20:12:15 +0200 Subject: [PATCH 7/8] test: pin WebSocket heartbeat durations --- lib/wire/wsio_test.go | 54 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 8 deletions(-) diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index 34789bb6..22ad86a8 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -267,6 +267,11 @@ func TestReadLoop_BatchedDeliveryIsInterruptible(t *testing.T) { func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { synctest.Test(t, func(t *testing.T) { + const ( + idleInterval = time.Second + pingTimeout = 10 * time.Second + ) + ctx, cancel := context.WithCancelCause(t.Context()) doneCh := make(chan struct{}) inCh := make(chan WsMsg) @@ -282,7 +287,7 @@ func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Second, server) + ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server) close(loopDone) }() @@ -291,10 +296,10 @@ func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { t.Fatal(err) } // The complete message is waiting for the application. This local delivery - // delay is not peer inactivity, so advancing well beyond both ping durations + // delay is not peer inactivity, so advancing across several idle intervals // must not send a probe. synctest.Wait() - time.Sleep(3 * time.Second) + time.Sleep(3 * idleInterval) synctest.Wait() if got := pingCount.Load(); got != 0 { t.Fatalf("pings while delivering = %d, want 0", got) @@ -304,10 +309,10 @@ func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { t.Fatalf("message = %+v, want %+v", got, want) } synctest.Wait() - time.Sleep(3 * time.Second) + time.Sleep(5 * idleInterval / 2) synctest.Wait() - if got := pingCount.Load(); got != 3 { - t.Fatalf("successful idle pings = %d, want 3", got) + if got := pingCount.Load(); got != 2 { + t.Fatalf("successful idle pings = %d, want 2", got) } if err := ctx.Err(); err != nil { t.Fatalf("parent context was canceled: %v", err) @@ -373,11 +378,18 @@ func TestReadLoop_ReadActivitySupersedesPingFailure(t *testing.T) { func TestReadLoop_ReportsUnresponsivePeer(t *testing.T) { synctest.Test(t, func(t *testing.T) { + const ( + idleInterval = time.Second + pingTimeout = 4 * time.Second + ) + ctx, cancel := context.WithCancelCause(t.Context()) doneCh := make(chan struct{}) inCh := make(chan WsMsg) + pingSeen := make(chan struct{}) client, server := pipeWithDialOptions(t, websocket.DialOptions{ OnPingReceived: func(context.Context, []byte) bool { + close(pingSeen) return false // Read the ping but deliberately suppress the required pong. }, }) @@ -386,12 +398,38 @@ func TestReadLoop_ReportsUnresponsivePeer(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Second, server) + ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server) close(loopDone) }() + synctest.Wait() + time.Sleep(idleInterval - time.Nanosecond) + synctest.Wait() + select { + case <-pingSeen: + t.Fatal("ping sent before idleInterval") + default: + } + time.Sleep(time.Nanosecond) + synctest.Wait() + select { + case <-pingSeen: + default: + t.Fatal("ping not sent at idleInterval") + } + // The peer consumes the ping but violates the protocol by withholding its pong. - time.Sleep(2 * time.Second) + time.Sleep(pingTimeout - time.Nanosecond) + synctest.Wait() + select { + case <-loopDone: + t.Fatal("ReadLoop returned before pingTimeout") + default: + } + if err := context.Cause(ctx); err != nil { + t.Fatalf("context cause before pingTimeout = %v, want nil", err) + } + time.Sleep(time.Nanosecond) synctest.Wait() assertClosedNow(t, loopDone, "ReadLoop") if err := context.Cause(ctx); !errors.Is(err, context.DeadlineExceeded) { From 63a2600ba0b633e48beaad8f20ab06473b08dcca Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Fri, 14 Aug 2026 20:27:27 +0200 Subject: [PATCH 8/8] docs: tighten WebSocket timeout documentation --- README.md | 17 ++++++++--------- jaws.go | 8 ++++---- lib/wire/wsio.go | 12 ++++++------ request_test.go | 2 +- serve.go | 6 +++--- 5 files changed, 22 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index dd4281df..f1b1211e 100644 --- a/README.md +++ b/README.md @@ -532,15 +532,14 @@ reported through `MustLog()`, which panics when no logger is configured. ### WebSocket keepalive ping -JaWS can ping read-idle WebSocket connections to detect peers that disappeared -without a close handshake. Incoming data and successful pings defer the next -probe, and JaWS does not probe while delivering an already-read message for -processing. An outbound WebSocket write that remains blocked for the configured -request timeout also ends the Request. - -Set `Jaws.WebSocketPingInterval` to control this. The default is -`jaws.DefaultWebSocketPingInterval` (1 minute). The value must be greater than -zero; non-positive values are invalid and do not disable probing. +JaWS pings read-idle WebSocket connections to detect peers that disappear +without a close handshake. Incoming data and successful pings restart the idle +interval. Time spent parsing or delivering already-read data does not count +toward it. + +`Jaws.WebSocketPingInterval` defaults to +`jaws.DefaultWebSocketPingInterval` (1 minute) and must be positive; +non-positive values do not disable probing. ### Safe to call before `Serve()` diff --git a/jaws.go b/jaws.go index 29268c63..460b09ae 100644 --- a/jaws.go +++ b/jaws.go @@ -120,11 +120,11 @@ type Jaws struct { // WebSocketPingInterval controls read-idle keepalive pings. // // When a WebSocket read remains pending for this interval, JaWS pings the peer. - // A data message or successful ping restarts the interval. Time spent delivering - // an already-read message for processing does not count as read-idle time. + // Incoming data or a successful ping restarts the interval. Time spent parsing + // or delivering already-read data does not count toward it. // - // It must be greater than zero; non-positive values are invalid and do not - // disable probing. It defaults to [DefaultWebSocketPingInterval]. + // It defaults to [DefaultWebSocketPingInterval] and must be positive; + // non-positive values do not disable probing. WebSocketPingInterval time.Duration MaxPendingRequestsPerIP int // Maximum number of unclaimed Requests per client IP. Defaults to DefaultMaxPendingRequestsPerIP. Set <=0 to disable the cap. webSocketTimeout time.Duration // timeout duration passed to ServeWith diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index 0bfe2f36..a5b99b45 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -22,11 +22,10 @@ const writeBatchLimit = 32 * 1024 // multiple records; malformed records are skipped independently. // // A WebSocket read that remains pending for idleInterval triggers a ping bounded -// by pingTimeout. A successful pong starts another idle interval for the pending -// read. Time spent parsing or delivering an already-read message is not idle -// time. Data processed while a ping is pending supersedes a failed ping. -// idleInterval and pingTimeout must be greater than zero; non-positive values -// are invalid. +// by pingTimeout. Incoming data or a successful ping restarts the idle interval. +// Time spent parsing or delivering an already-read message does not count toward +// it. If a message is processed while a ping is pending, that ping's failure is +// ignored. idleInterval and pingTimeout must be positive. // // Closes incomingMsgCh on exit. // @@ -134,7 +133,8 @@ func readWebSocket(ctx context.Context, resultCh chan<- wsReadResult, ws *websoc // // Consecutive queued records may be coalesced into one text message. // -// Each write is bounded by writeTimeout, which must be positive. +// Each WebSocket write has its own writeTimeout deadline; writeTimeout must be +// positive. // // Closes the WebSocket on exit. // diff --git a/request_test.go b/request_test.go index b40cbff6..8d524ce6 100644 --- a/request_test.go +++ b/request_test.go @@ -4379,7 +4379,7 @@ func TestWS_WriteTimeoutDisconnectsNonReadingClient(t *testing.T) { // Releasing the update queues more output than the connection and bounded // outbound channel can absorb. The peer can still send the input above, but - // its blocked receive direction must tear down the Request within writeTimeout. + // once an outbound write blocks, its timeout must tear down the Request. releaseUpdate() waitForRequestCount(t, ts.jw, 0, testTimeout) } diff --git a/serve.go b/serve.go index c759e042..5a0cd3f7 100644 --- a/serve.go +++ b/serve.go @@ -45,9 +45,9 @@ func (jw *Jaws) getWebSocketTimeout() (t time.Duration) { // whole-second samples from the epoch established by [New]. Retirement is checked // only during maintenance passes, so it is not timed precisely from those events. // -// requestTimeout also bounds each WebSocket keepalive ping and outbound write. -// These operation deadlines are independent of maintenance timing; see -// [Jaws.WebSocketPingInterval] for probe scheduling. +// requestTimeout also bounds each WebSocket keepalive ping and outbound write, +// independently of the maintenance schedule. See [Jaws.WebSocketPingInterval] +// for probe scheduling. // // It is intended to run on its own goroutine and returns when [Jaws.Close] is // called. Errors reported through [Jaws.Log] are queued without waiting for