diff --git a/README.md b/README.md index db927318..f1b1211e 100644 --- a/README.md +++ b/README.md @@ -432,9 +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. -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. +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 @@ -532,13 +532,14 @@ 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 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. -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.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 7548492e..460b09ae 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,14 @@ 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. + // Incoming data or a successful ping restarts the interval. Time spent parsing + // or delivering already-read data does not count toward it. + // + // 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 c602e56f..a5b99b45 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,111 @@ 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. 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. // -// 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 { - select { - case <-ctx.Done(): - return - case <-doneCh: - return - case incomingMsgCh <- msg: + 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 + } + var activityDuringPing bool + + defer func() { + cancel() + workers.Wait() + idleTimer.Stop() + close(incomingMsgCh) + }() + + for { + select { + case <-ctx.Done(): + return + case result := <-readResultCh: + // 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 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 !activityDuringPing { + armIdleTimer() + } + case <-idleTimerCh: + idleTimerCh = nil + // 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 { + 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 @@ -58,13 +133,16 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st // // Consecutive queued records may be coalesced into one text message. // +// Each WebSocket write has its own writeTimeout deadline; writeTimeout 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() @@ -73,52 +151,16 @@ 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 } + 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) } - } - } - 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() + writecancel() } } reportError(ctx, doneCh, ccf, err) @@ -138,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 b5f09758..22ad86a8 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,12 +145,12 @@ 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() { defer close(writeDoneCh) - WriteLoop(ctx, nil, doneCh, outCh, client) + WriteLoop(ctx, nil, doneCh, outCh, time.Hour, client) }() var got []WsMsg @@ -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,179 @@ 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) + 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, idleInterval, pingTimeout, 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 across several idle intervals + // must not send a probe. + synctest.Wait() + time.Sleep(3 * idleInterval) + 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(5 * idleInterval / 2) + synctest.Wait() + 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) + } + + 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) { + 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. + }, + }) + loopDone := make(chan struct{}) + defer closeWireBubble(cancel, client, server)() + + client.CloseRead(ctx) + go func() { + 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(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) { + 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{}) @@ -277,7 +451,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 @@ -332,7 +506,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) @@ -367,7 +541,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) @@ -415,7 +589,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 @@ -466,7 +640,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() @@ -487,7 +661,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) @@ -509,7 +683,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) }() @@ -524,6 +698,56 @@ func TestWriteLoop_RespectsDoneWhileWriting(t *testing.T) { }) } +func TestWriteLoop_TimeoutIsPerWrite(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 for a full writeTimeout. + time.Sleep(3 * writeTimeout / 4) + outCh <- WsMsg{ + Jid: jid.Jid(1234), + What: what.Inner, + Data: strings.Repeat("x", writeBatchLimit), + } + synctest.Wait() + time.Sleep(writeTimeout - time.Nanosecond) + synctest.Wait() + select { + case <-loopDone: + t.Fatal("WriteLoop returned before the second write's 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_RespectsOutboundClosed(t *testing.T) { outCh := make(chan WsMsg) jawsDoneCh := make(chan struct{}) @@ -538,7 +762,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) @@ -558,7 +782,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)} @@ -583,7 +807,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 +834,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 +851,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..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, pingInterval, wsTimeout 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,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, 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 +1201,7 @@ 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 + idleInterval := rq.Jaws.WebSocketPingInterval wsTimeout := rq.Jaws.getWebSocketTimeout() if strings.HasSuffix(r.URL.Path, "/noscript") { w.WriteHeader(http.StatusNoContent) @@ -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, 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 25603088..8d524ce6 100644 --- a/request_test.go +++ b/request_test.go @@ -4320,10 +4320,101 @@ func TestWS_PingDisconnectsUnresponsiveClient(t *testing.T) { waitForRequestCount(t, ts.jw, 0, testTimeout) } -func TestWS_PingDisabledKeepsIdleConnection(t *testing.T) { +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 + // once an outbound write blocks, its timeout must tear down the Request. + releaseUpdate() + waitForRequestCount(t, ts.jw, 0, testTimeout) +} + +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 +4423,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 +4450,52 @@ 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 { + ts.rq.Dirty(item) + select { + case <-updateStarted: + 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) } - 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 +4507,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 +4596,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..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. // -// 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. +// 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