Limit inbound events and absorb shared broadcast bursts

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))
 }

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.

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

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

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.

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
 )

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=

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

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

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)
+		}
+	})
+}