From 17cafcd7a1ce724746d786a79dffef175e4698a1 Mon Sep 17 00:00:00 2001 From: Johan Lindh Date: Wed, 7 Oct 2026 09:03:03 +0200 Subject: [PATCH] Limit inbound events and absorb shared broadcast bursts --- broadcast.go | 10 ++ doc/browser.md | 3 + doc/deployment.md | 15 +++ doc/transport.md | 14 ++- errors.go | 3 + go.mod | 1 + go.sum | 2 + jaws.go | 10 ++ lib/wire/wsio.go | 19 ++- lib/wire/wsio_rate_test.go | 156 ++++++++++++++++++++++++ lib/wire/wsio_test.go | 24 ++-- request.go | 13 +- request_rate_test.go | 242 +++++++++++++++++++++++++++++++++++++ 13 files changed, 492 insertions(+), 20 deletions(-) create mode 100644 lib/wire/wsio_rate_test.go create mode 100644 request_rate_test.go diff --git a/broadcast.go b/broadcast.go index ae94fd4d..5a25ebff 100644 --- a/broadcast.go +++ b/broadcast.go @@ -25,6 +25,12 @@ import ( // // All convenience helpers on [Jaws] that call Broadcast inherit this requirement. // +// Broadcast commands are not coalesced. A full recipient queue cancels that +// Request with [ErrRequestOverloaded], except that unaddressed [what.Update] +// messages are skipped. Per-event broadcasts on shared tags let clients drive +// every recipient's bandwidth requirements. Keep that byte rate within the +// slowest supported client link, or use [Jaws.Dirty] to coalesce shared-state updates. +// // [what.Replace] and [what.Remove] are reported as [ErrReplaceNotBroadcastable] // and [ErrRemoveNotBroadcastable], and are not sent. Use [Element.Replace], // [Jaws.Delete], or [Element.Remove] instead. @@ -357,6 +363,10 @@ func (jw *Jaws) Delete(target any) { } // Append calls the JavaScript appendChild method on all HTML elements matching target. +// +// Each call broadcasts a separate command. Per-event appends on shared tags must +// fit every recipient's bandwidth; see [Jaws.Broadcast]. Prefer [Jaws.Dirty] for +// coalesced shared-state updates. func (jw *Jaws) Append(target any, html template.HTML) { jw.broadcastTo(target, what.Append, string(html)) } diff --git a/doc/browser.md b/doc/browser.md index dd745c2b..415d92ee 100644 --- a/doc/browser.md +++ b/doc/browser.md @@ -111,6 +111,9 @@ path must already exist, and invalid or reserved paths throw. Bound array writes must target an existing canonical index. The server accepts or reconciles proposals through the [JavaScript binding API](ui/jsvar.md). +Both reads and writes that submit bound values consume the Request's +[`MaxEventRate` budget](deployment.md#capacity). + Binding nodes provide the store name and initial JSON during the initial page render. The live values then reside on `window`. Server patches can replace the root, replace a subtree, or delete an object property. diff --git a/doc/deployment.md b/doc/deployment.md index 837d8307..10863a28 100644 --- a/doc/deployment.md +++ b/doc/deployment.md @@ -105,6 +105,21 @@ client IP, and permits each request key to be claimed once. ## Capacity +`MaxEventRate` limits incoming `Click`, `ContextMenu`, `Input`, and `JsVar` +records together to 100 per second per Request by default. Zero selects the +default; a negative value disables the limit. Records are paced in order without +being dropped, with an initial 10 ms wait at the default rate. `Remove` consumes +no event budget but cannot overtake a waiting event. Fast range-slider or +JavaScript activity can accumulate delay; configure the rate for the +application's workload. + +The per-Request limit does not bound aggregate broadcasts from many clients. +Broadcast helpers such as `Append`, `SetInner`, and `Alert` send a command per +call. On shared tags, each client's events can increase every subscriber's +required bandwidth. Requests whose queues fill are disconnected with +`ErrRequestOverloaded`. Prefer `Dirty`/`Update` coalescing for shared state; +per-event broadcast byte rates must fit the slowest supported client link. + `MaxPendingRequestsPerIP` defaults to 100. IPv4 clients share a bucket by address; IPv6 uses a /64, with the well-known `64:ff9b::/96` NAT64 prefix mapped to its embedded IPv4 address. At capacity a new page retires the oldest idle diff --git a/doc/transport.md b/doc/transport.md index 3daba153..7b6fbde4 100644 --- a/doc/transport.md +++ b/doc/transport.md @@ -127,10 +127,18 @@ the Request connection. The resulting error is retained as the Request's cancellation cause and passed to `Jaws.Log`. The client does not split input, JavaScript proposals, click data, or removal reports across messages. +`Jaws.MaxEventRate` paces `Click`, `ContextMenu`, `Input`, and `JsVar` records +per Request, including records batched in one WebSocket message. The default is +100 records per second; zero selects the default and a negative value disables +it. Events wait in order without being dropped; pacing can also delay the first +event. `Remove` is uncharged but remains in record order. This limit applies to +WebSocket input, not in-process broadcasts or the test harness. See +[capacity](deployment.md#capacity) for shared broadcasts. + `wire.ReadLoop` sends keepalive pings when a read remains idle. Parsing and -delivery time do not count as read-idle time. Incoming data or a successful ping -restarts the interval. Data received while a ping is pending makes a failure -of that ping irrelevant to the connection. +delivery time, including rate-limit waits, do not count as read-idle time. +Incoming data or a successful ping restarts the interval. Data received while +a ping is pending makes a failure of that ping irrelevant to the connection. `wire.WriteLoop` combines queued records into a text message until reaching its 32 KiB flush threshold. It appends whole records, so a batch can exceed the diff --git a/errors.go b/errors.go index bd7013ea..6da0f1fe 100644 --- a/errors.go +++ b/errors.go @@ -25,6 +25,9 @@ var ErrValueUnchanged = errors.New("value unchanged") // // The cancellation cause reachable via [context.Cause] on [Request.Context] // wraps this sentinel, so callers can match it with [errors.Is]. +// A Request can overload on its own events or on broadcasts from other clients. +// Shared broadcasts must fit each recipient's processing and network capacity; +// see [Jaws.Broadcast]. var ErrRequestOverloaded = errors.New("request overloaded") // ErrValueNotFinite indicates that a [Request] was cancelled by a non-finite UI value. diff --git a/go.mod b/go.mod index a73c8380..5ea8d881 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/coder/websocket v1.8.15 github.com/linkdata/deadlock v0.5.5 github.com/linkdata/jq v0.6.0 + github.com/linkdata/rate v1.17.1 github.com/linkdata/secureheaders v1.5.0 github.com/linkdata/staticserve v1.1.8 ) diff --git a/go.sum b/go.sum index 39d5a26d..f3e05a91 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/linkdata/deadlock v0.5.5 h1:d6O+rzEqasSfamGDA8u7bjtaq7hOX8Ha4Zn36Wxrk github.com/linkdata/deadlock v0.5.5/go.mod h1:tXb28stzAD3trzEEK0UJWC+rZKuobCoPktPYzebb1u0= github.com/linkdata/jq v0.6.0 h1:yxAeWsknIWmt7Wbr5CIEkj2i1WwC2DnJDZlOrPxJkr0= github.com/linkdata/jq v0.6.0/go.mod h1:b76MMuWybyXrVEKSHJkv+r5IEattnghPf1bNRJ5PNlE= +github.com/linkdata/rate v1.17.1 h1:b3ld6CaQQzDxk/ChR7x1iDGMQIFyWIZAKQknZWAmiuc= +github.com/linkdata/rate v1.17.1/go.mod h1:VMPlHbG1mu28l0QsvRF3cMN0/pmOmEW+7vq622fx54Y= github.com/linkdata/secureheaders v1.5.0 h1:4kI290hrVUYqluxF8pQfRYpbZuorSd9+8GlMgrLcrK4= github.com/linkdata/secureheaders v1.5.0/go.mod h1:50TiKmPaWki8gIZXmBxrLUinyJst12YuN3EbsXZMO+4= github.com/linkdata/staticserve v1.1.8 h1:qpm0dAaAsik1YrETSiMoS7ciBL9qPVNUFcfdio/ol84= diff --git a/jaws.go b/jaws.go index 4ea1c0e0..4c2fa41a 100644 --- a/jaws.go +++ b/jaws.go @@ -61,6 +61,9 @@ const ( // DefaultMaxPendingRequestsPerIP is the default pending Request limit per client bucket. DefaultMaxPendingRequestsPerIP = 100 + + // DefaultMaxEventRate is the default inbound event limit per Request, in records per second. + DefaultMaxEventRate = 100 ) type subscription struct { @@ -149,6 +152,13 @@ type Jaws struct { // It defaults to [DefaultWebSocketPingInterval] and must be positive; // non-positive values do not disable probing. WebSocketPingInterval time.Duration + // MaxEventRate limits incoming event records per second per Request. + // + // Click, ContextMenu, Input, and JsVar share the limit. Events are paced in + // order without dropping them; pacing can also delay the first event. + // Remove consumes no event budget but cannot overtake a waiting event. + // Zero selects [DefaultMaxEventRate]; a negative value disables the limit. + MaxEventRate int32 // MaxSessions limits the number of registered Sessions. // // A non-positive value disables the cap, which is the default. Sessions diff --git a/lib/wire/wsio.go b/lib/wire/wsio.go index c3e96580..f79289a7 100644 --- a/lib/wire/wsio.go +++ b/lib/wire/wsio.go @@ -8,6 +8,7 @@ import ( "github.com/coder/websocket" "github.com/linkdata/jaws/lib/what" + "github.com/linkdata/rate" ) // writeBatchLimit is the threshold at which WriteLoop stops appending whole @@ -20,6 +21,12 @@ const writeBatchLimit = 32 * 1024 // Records are LF-terminated and delivered in order. A text message may contain // multiple records; malformed records are skipped independently. // +// maxEventRate limits Click, ContextMenu, Input, and JsVar records per second. +// Non-positive values disable the limit. Events are paced in record order; +// pacing can also delay the first event. Remove and other records consume no +// event budget but cannot overtake a waiting event. Pacing time does not count +// as read-idle time. +// // 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 @@ -32,8 +39,9 @@ const writeBatchLimit = 32 * 1024 // 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, idleInterval, pingTimeout time.Duration, ws *websocket.Conn) { +func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan struct{}, incomingMsgCh chan<- WsMsg, idleInterval, pingTimeout time.Duration, ws *websocket.Conn, maxEventRate int32) { ctx, cancel := contextWithDone(ctx, doneCh) + limiter := rate.Limiter{CloseCh: ctx.Done()} readResultCh := make(chan wsReadResult) pingResultCh := make(chan error, 1) var workers sync.WaitGroup @@ -75,6 +83,15 @@ func ReadLoop(ctx context.Context, ccf context.CancelCauseFunc, doneCh <-chan st if result.typ == websocket.MessageText { for record := range bytes.Lines(result.txt) { if msg, parsed := Parse(record); parsed { + switch msg.What { + case what.Click, what.ContextMenu, what.Input, what.JsVar: + if maxEventRate > 0 { + limiter.Wait(&maxEventRate) + if ctx.Err() != nil { + return + } + } + } select { case <-ctx.Done(): return diff --git a/lib/wire/wsio_rate_test.go b/lib/wire/wsio_rate_test.go new file mode 100644 index 00000000..2749c0eb --- /dev/null +++ b/lib/wire/wsio_rate_test.go @@ -0,0 +1,156 @@ +package wire + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/coder/websocket" + "github.com/linkdata/jaws/lib/what" +) + +func TestReadLoopEventRate(t *testing.T) { + for _, maxRate := range []int32{-1, 0, 100} { + for _, batched := range []bool{false, true} { + t.Run(fmt.Sprintf("rate=%d/batched=%t", maxRate, batched), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + client, server := pipe(t) + defer closeWireBubble(cancel, client, server)() + inCh := make(chan WsMsg) + go ReadLoop(ctx, cancel, nil, inCh, time.Hour, time.Hour, server, maxRate) + want := []WsMsg{ + {Jid: 1, What: what.Remove, Data: "Jid.2"}, + {Jid: 1, What: what.Input, Data: "first"}, + {Jid: 1, What: what.Remove, Data: "Jid.3"}, + {Jid: 1, What: what.Click, Data: "1 1 0 click"}, + {Jid: 1, What: what.ContextMenu, Data: "1 1 0 menu"}, + {Jid: 1, What: what.JsVar, Data: "value=2"}, + {Jid: 1, What: what.Remove, Data: "Jid.4"}, + } + go func() { + var payload []byte + for _, msg := range want { + payload = msg.Append(payload) + if !batched { + if err := client.Write(ctx, websocket.MessageText, payload); err != nil { + t.Error(err) + return + } + payload = payload[:0] + } + } + if batched { + if err := client.Write(ctx, websocket.MessageText, payload); err != nil { + t.Error(err) + } + } + }() + start := time.Now() + var elapsed time.Duration + for _, msg := range want { + if maxRate > 0 && msg.What != what.Remove { + elapsed += time.Second / time.Duration(maxRate) + } + if got := <-inCh; got != msg { + t.Fatalf("record = %+v, want %+v", got, msg) + } + if got := time.Since(start); got != elapsed { + t.Fatalf("%v arrived after %v, want %v", msg.What, got, elapsed) + } + } + // Idle time permits one immediate event, not an accumulated burst. + time.Sleep(time.Second) + start = time.Now() + msg := want[1] + if err := client.Write(ctx, websocket.MessageText, msg.Append(msg.Append(nil))); err != nil { + t.Fatal(err) + } + for i := range 2 { + if got := <-inCh; got != msg { + t.Fatalf("record after idle = %+v", got) + } + elapsed = 0 + if maxRate > 0 { + elapsed = time.Duration(i) * time.Second / time.Duration(maxRate) + } + if got := time.Since(start); got != elapsed { + t.Fatalf("record %d after idle arrived after %v, want %v", i, got, elapsed) + } + } + }) + }) + } + } +} + +func TestReadLoopRateLimitShutdown(t *testing.T) { + for _, maxRate := range []int32{10, 100} { + for _, stop := range []string{"context", "done"} { + t.Run(fmt.Sprintf("rate=%d/stop=%s", maxRate, stop), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + client, server := pipe(t) + defer closeWireBubble(cancel, client, server)() + inCh := make(chan WsMsg, 2) + done := make(chan struct{}) + loopDone := make(chan struct{}) + go func() { + ReadLoop(ctx, cancel, done, inCh, time.Hour, time.Hour, server, maxRate) + close(loopDone) + }() + msg := WsMsg{Jid: 1, What: what.Input} + if err := client.Write(ctx, websocket.MessageText, msg.Append(msg.Append(nil))); err != nil { + t.Fatal(err) + } + synctest.Wait() + start := time.Now() + if stop == "done" { + close(done) + } else { + cancel(nil) + } + <-loopDone + if elapsed := time.Since(start); elapsed > 20*time.Millisecond { + t.Fatalf("shutdown took %v", elapsed) + } + if msg, ok := <-inCh; ok { + t.Fatalf("delivered record after shutdown: %+v", msg) + } + if stop == "done" && ctx.Err() != nil { + t.Fatalf("shutdown reported a transport error: %v", context.Cause(ctx)) + } + }) + }) + } + } +} + +func TestReadLoopDoesNotPingWhileRateLimited(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + var pings atomic.Int32 + client, server := pipeWithDialOptions(t, websocket.DialOptions{ + OnPingReceived: func(context.Context, []byte) bool { pings.Add(1); return true }, + }) + defer closeWireBubble(cancel, client, server)() + client.CloseRead(ctx) + inCh := make(chan WsMsg) + go ReadLoop(ctx, cancel, nil, inCh, time.Millisecond, time.Second, server, 10) + msg := WsMsg{Jid: 1, What: what.Input} + if err := client.Write(ctx, websocket.MessageText, msg.Append(msg.Append(nil))); err != nil { + t.Fatal(err) + } + for range 2 { + if got := <-inCh; got != msg { + t.Fatalf("record = %+v", got) + } + } + if got := pings.Load(); got != 0 { + t.Fatalf("pings during pacing = %d", got) + } + }) +} diff --git a/lib/wire/wsio_test.go b/lib/wire/wsio_test.go index e4cb1101..be1392ef 100644 --- a/lib/wire/wsio_test.go +++ b/lib/wire/wsio_test.go @@ -36,7 +36,7 @@ func TestReadLoop_RespectsContextDone(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, nil, jawsDoneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, nil, jawsDoneCh, inCh, time.Hour, time.Hour, server, 0) }() writeCtx, writeCancel := context.WithTimeout(t.Context(), 3*time.Second) @@ -67,7 +67,7 @@ func TestReadLoop_RespectsDone(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server, 0) close(loopDone) }() @@ -95,7 +95,7 @@ func TestReadLoop_RespectsDoneWhileReading(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server, 0) close(loopDone) }() @@ -148,7 +148,7 @@ func TestReadWriteLoop_RoundTrip(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, nil, doneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, nil, doneCh, inCh, time.Hour, time.Hour, server, 0) }() writeDoneCh := make(chan struct{}) go func() { @@ -188,7 +188,7 @@ func TestReadLoop_SkipsMalformedRecords(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server, 0) close(loopDone) }() if err := client.Write(ctx, websocket.MessageText, payload); err != nil { @@ -234,7 +234,7 @@ func TestReadLoop_BatchedDeliveryIsInterruptible(t *testing.T) { defer closeWireBubble(cancel, client, server)() go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Hour, time.Hour, server, 0) close(loopDone) }() if err := client.Write(ctx, websocket.MessageText, payload); err != nil { @@ -290,7 +290,7 @@ func TestReadLoop_DoesNotPingWhileDelivering(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server) + ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server, 0) close(loopDone) }() @@ -348,7 +348,7 @@ func TestReadLoop_ReadActivitySupersedesPingFailure(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Second, 10*time.Second, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, 10*time.Second, server, 0) close(loopDone) }() @@ -401,7 +401,7 @@ func TestReadLoop_ReportsUnresponsivePeer(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server) + ReadLoop(ctx, cancel, doneCh, inCh, idleInterval, pingTimeout, server, 0) close(loopDone) }() @@ -698,7 +698,7 @@ func TestWriteLoop_ReloadClosesConnection(t *testing.T) { readDone := make(chan struct{}) go func() { defer close(readDone) - ReadLoop(ioctx, cancelIO, nil, inCh, time.Hour, time.Hour, server) + ReadLoop(ioctx, cancelIO, nil, inCh, time.Hour, time.Hour, server, 0) }() // After delivering the first record, ReadLoop blocks delivering the // second. Closing the socket alone cannot unblock that delivery. @@ -931,7 +931,7 @@ func TestReadLoop_ReportsError(t *testing.T) { readDoneCh := make(chan struct{}) go func() { defer close(readDoneCh) - ReadLoop(ctx, cancel, jawsDoneCh, inCh, time.Hour, time.Hour, server) + ReadLoop(ctx, cancel, jawsDoneCh, inCh, time.Hour, time.Hour, server, 0) }() waitDone(t, readDoneCh, "ReadLoop after read error") @@ -975,7 +975,7 @@ func TestReadLoop_RespectsDoneWhileWaitingForPong(t *testing.T) { client.CloseRead(ctx) go func() { - ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Hour, server) + ReadLoop(ctx, cancel, doneCh, inCh, time.Second, time.Hour, server, 0) close(loopDone) }() diff --git a/request.go b/request.go index e18fb3d3..4d54dc64 100644 --- a/request.go +++ b/request.go @@ -1201,10 +1201,11 @@ func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, wsTimeout time numElems := len(rq.elems) rq.mu.RUnlock() // Size the broadcast buffer with headroom that scales with the page's element - // count. mustBroadcast (see Jaws.Serve) sends here non-blocking and, if the send + // count, with a floor of 64 to absorb synchronized bursts on small pages. + // mustBroadcast (see Jaws.Serve) sends here non-blocking and, if the send // would block, kills the subscription and cancels this request for every message // except a coalescible nil-destination Update, which it drops instead. - pendingSubscription := rq.Jaws.subscribe(rq, 4+numElems*4) + pendingSubscription := rq.Jaws.subscribe(rq, max(64, 4+numElems*4)) defer func() { // onConnect is user code and may return an error or panic. Release its // subscription unless process took responsibility for doing so. @@ -1226,8 +1227,12 @@ func (rq *Request) runWebSocket(ws *websocket.Conn, idleInterval, wsTimeout time rq.cancel(err) } outboundMsgCh := make(chan wire.WsMsg, cap(pendingSubscription)) - go wire.ReadLoop(ctx, disconnect, rq.Jaws.Done(), incomingMsgCh, idleInterval, wsTimeout, ws) // closes incomingMsgCh - go wire.WriteLoop(ctx, disconnect, rq.Jaws.Done(), outboundMsgCh, wsTimeout, ws) // closes the WebSocket + maxEventRate := rq.Jaws.MaxEventRate + if maxEventRate == 0 { + maxEventRate = DefaultMaxEventRate + } + go wire.ReadLoop(ctx, disconnect, rq.Jaws.Done(), incomingMsgCh, idleInterval, wsTimeout, ws, maxEventRate) // closes incomingMsgCh + go wire.WriteLoop(ctx, disconnect, rq.Jaws.Done(), outboundMsgCh, wsTimeout, ws) // closes the WebSocket broadcastMsgCh := pendingSubscription pendingSubscription = nil // Production deliberately discards the recovered value so a loop panic stays diff --git a/request_rate_test.go b/request_rate_test.go new file mode 100644 index 00000000..df04ef9c --- /dev/null +++ b/request_rate_test.go @@ -0,0 +1,242 @@ +package jaws + +import ( + "bytes" + "context" + "fmt" + "html/template" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/linkdata/jaws/lib/tag" + "github.com/linkdata/jaws/lib/what" + "github.com/linkdata/jaws/lib/wire" + "github.com/linkdata/rate" +) + +// broadcastFlood uses real WebSockets and a continuously reading one-Element +// subscriber. Senders do not subscribe to the shared tag. No receiver delays or +// artificial scheduling apply. +type broadcastFlood struct { + ctx context.Context + victim *Request + senders []*websocket.Conn + received atomic.Int64 + changed chan struct{} + readDone chan struct{} +} + +func newBroadcastFlood(tb testing.TB, maxRate int32, senders int) *broadcastFlood { + tb.Helper() + jw, err := New() + if err != nil { + tb.Fatal(err) + } + jw.MaxEventRate = maxRate + ctx, cancel := context.WithTimeout(tb.Context(), 30*time.Second) + go jw.Serve() + srv := httptest.NewServer(jw) + f := &broadcastFlood{ctx: ctx, changed: make(chan struct{}, 1), readDone: make(chan struct{})} + var conns []*websocket.Conn + tb.Cleanup(func() { + cancel() + for _, conn := range conns { + _ = conn.CloseNow() + } + jw.Close() + srv.Close() + }) + const shared = tag.Tag("shared-log") + const entry = template.HTML("

aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa

") + connect := func(sender bool) (*Request, *websocket.Conn) { + initial := httptest.NewRequest(http.MethodGet, srv.URL, nil).WithContext(ctx) + initial.RemoteAddr = "127.0.0.1:12345" + rw := httptest.NewRecorder() + rq := jw.NewRequest(rw, initial) + writer := testRequestWriter{rq: rq, Writer: rw} + if sender { + writer.Register(&testUi{}, func(*Element, string) error { + jw.Append(shared, entry) + return nil + }) + } else { + writer.Register(&testUi{}, shared) + } + connected := make(chan struct{}) + rq.SetConnectFn(func(*Request) error { close(connected); return nil }) + conn, _, dialErr := websocket.Dial(ctx, strings.Replace(srv.URL, "http", "ws", 1)+"/jaws/"+rq.JawsKeyString(), &websocket.DialOptions{ + HTTPHeader: http.Header{"Origin": {srv.URL}}, + }) + if dialErr != nil { + tb.Fatal(dialErr) + } + conns = append(conns, conn) + select { + case <-connected: + case <-ctx.Done(): + tb.Fatal(ctx.Err()) + } + return rq, conn + } + var victimConn *websocket.Conn + f.victim, victimConn = connect(false) + go func() { + defer close(f.readDone) + for { + _, data, readErr := victimConn.Read(ctx) + if readErr != nil { + return + } + for record := range bytes.Lines(data) { + if msg, ok := wire.Parse(record); ok && msg.What == what.Append { + f.received.Add(1) + } + } + f.signal() + } + }() + for range senders { + _, conn := connect(true) + conn.CloseRead(ctx) + f.senders = append(f.senders, conn) + } + return f +} + +func (f *broadcastFlood) signal() { + select { + case f.changed <- struct{}{}: + default: + } +} + +func (f *broadcastFlood) wait(want int64) (err error) { + for f.received.Load() < want { + select { + case <-f.changed: + case <-f.readDone: + return fmt.Errorf("victim disconnected after %d/%d appends: %w", f.received.Load(), want, context.Cause(f.victim.Context())) + case <-f.ctx.Done(): + return f.ctx.Err() + } + } + return +} + +func TestWS_EventRateAndBroadcastBurst(t *testing.T) { + for _, tt := range []struct { + senders int + rate int32 + }{ + {senders: 1}, + {senders: 16}, + {senders: 1, rate: 25}, + } { + t.Run(fmt.Sprintf("senders=%d/rate=%d", tt.senders, tt.rate), func(t *testing.T) { + f := newBroadcastFlood(t, tt.rate, tt.senders) + const records = 20 + msg := wire.WsMsg{Jid: 1, What: what.Input, Data: "send"} + var payload []byte + for range records { + payload = msg.Append(payload) + } + start := time.Now() + var writers sync.WaitGroup + for _, conn := range f.senders { + writers.Go(func() { + if err := conn.Write(f.ctx, websocket.MessageText, payload); err != nil { + t.Error(err) + } + }) + } + writers.Wait() + if err := f.wait(int64(tt.senders * records)); err != nil { + t.Fatal(err) + } + maxRate := tt.rate + if maxRate == 0 { + maxRate = DefaultMaxEventRate + } + if elapsed := time.Since(start); elapsed < (records-1)*time.Second/time.Duration(maxRate) { + t.Fatalf("event batch bypassed the rate limit: %v", elapsed) + } + if err := f.victim.Context().Err(); err != nil { + t.Fatalf("victim cancelled: %v", context.Cause(f.victim.Context())) + } + }) + } +} + +// BenchmarkWSBroadcastBurst measures synchronized senders against a continuously +// reading one-Element victim. Pacing is client-side so it isolates buffer capacity. +// Run with -cpu=1,4 to keep 16 senders in both cases and -benchtime below 30s. +// Compare delivery and disconnect metrics; ns/op includes client pacing. +func BenchmarkWSBroadcastBurst(b *testing.B) { + for _, eventsPerSecond := range []int32{25, 400} { + b.Run(fmt.Sprintf("rate=%d", eventsPerSecond), func(b *testing.B) { + parallelism := max(1, 16/runtime.GOMAXPROCS(0)) + senders := parallelism * runtime.GOMAXPROCS(0) + f := newBroadcastFlood(b, -1, senders) + ctx, cancel := context.WithCancel(f.ctx) + ticks := make([]chan struct{}, senders) + for i := range ticks { + ticks[i] = make(chan struct{}, 1) + } + clockDone := make(chan struct{}) + go func() { + defer close(clockDone) + limiter := rate.Limiter{CloseCh: ctx.Done()} + for ctx.Err() == nil { + limiter.Wait(&eventsPerSecond) + for _, tick := range ticks { + select { + case tick <- struct{}{}: + default: + } + } + } + }() + var worker atomic.Int32 + event := wire.WsMsg{Jid: 1, What: what.Input, Data: "send"} + msg := event.Append(nil) + b.SetParallelism(parallelism) + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + i := int(worker.Add(1)) - 1 + for pb.Next() { + select { + case <-ticks[i]: + case <-ctx.Done(): + b.Error(ctx.Err()) + return + } + if err := f.senders[i].Write(ctx, websocket.MessageText, msg); err != nil { + b.Error(err) + return + } + } + }) + cancel() + <-clockDone + err := f.wait(int64(b.N)) + b.StopTimer() + if err != nil && f.victim.Context().Err() == nil { + b.Fatal(err) + } + var disconnected float64 + if f.victim.Context().Err() != nil { + disconnected = 1 + } + b.ReportMetric(disconnected, "victim-disconnects") + b.ReportMetric(100*float64(f.received.Load())/float64(b.N), "delivered-%") + }) + } +}