From 0bf66d7c724f307d2de35f4b691d9463d05d2522 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:21:20 +0100 Subject: [PATCH] Space websocket state broadcasts at least 2s apart per audience Every accepted agent report sends a state invalidation, and each state broadcast rebuilds and diffs the full frontend state, about 130 ms at 2,080 synthetic resources. The hub coalesced invalidations with a 100 ms debounce that restarted on every signal, so reports spaced more than 100 ms apart each produced a broadcast, while signals arriving closer than the window could postpone broadcasts until they stopped (#2199). With 200 agents reporting every 30 seconds, that was roughly 3.4 broadcasts a second, about half a core whenever a browser was open. Coalescing is now a throttle. The first invalidation after a quiet period still broadcasts after the 100 ms window. Later invalidations for the same audience (all clients, or one tenant) replace the pending request without rescheduling it, and broadcast no sooner than 2 seconds after that audience's previous broadcast. That caps broadcast work near 7% of a core at the same scale. Two seconds matches the report-driven store refresh interval, so faster broadcasts mostly re-sent the same resources. Refs #2199 --- .../v6/internal/subsystems/api-contracts.md | 14 ++ internal/websocket/hub.go | 79 ++++++----- internal/websocket/hub_test.go | 123 ++++++++++++++++++ 3 files changed, 187 insertions(+), 29 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index e79dbd405f..fa90b4332d 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -698,6 +698,20 @@ and slow-client eviction must use the same synchronized send/close ownership; recovering a send-on-closed panic is not a substitute for a race-free channel lifecycle. +Invalidation coalescing is a throttle, not a restart-on-signal debounce. The +first invalidation after a quiet period broadcasts after the 100 ms coalesce +window; later invalidations for the same audience (all clients, or one tenant) +replace the pending request without rescheduling it, and are broadcast no +sooner than `defaultStateBroadcastInterval` (2 seconds) after that audience's +previous broadcast. Every accepted agent report sends an invalidation and each +broadcast resolves and diffs the full frontend state (about 130 ms at 2,080 +synthetic resources), so the debounce let broadcast work scale with agent +count, and signals closer together than the window could postpone broadcasts +indefinitely (#2199). `TestStateBroadcastsSpacedByInterval`, +`TestTenantStateBroadcastsSpacedByInterval`, and +`TestStateBroadcastNotStarvedBySteadySignals` pin the spacing, latest-state, +and no-starvation behaviour. + Browser clients declare their inbound state-frame ceiling on the WebSocket upgrade URL before initial-state construction begins. If a complete snapshot or a later delta exceeds that ceiling, the hub sends a compact diff --git a/internal/websocket/hub.go b/internal/websocket/hub.go index 7c07ff1e63..0b2dd1a396 100644 --- a/internal/websocket/hub.go +++ b/internal/websocket/hub.go @@ -614,8 +614,20 @@ type Hub struct { tenantCoalesceGeneration map[string]uint64 stateBroadcastWake chan struct{} stateBroadcastDone chan struct{} + // stateBroadcastInterval is the minimum spacing between state broadcasts + // to one audience (all clients, or one tenant); zero disables spacing. + stateBroadcastInterval time.Duration + lastStateBroadcast time.Time + tenantLastStateBroadcast map[string]time.Time } +// defaultStateBroadcastInterval spaces state broadcasts to one audience. Each +// broadcast rebuilds and diffs the full frontend state, and every accepted +// agent report requests one, so a 100 ms debounce let broadcasts scale with +// agent count (#2199). Report-driven store refreshes run at most once per +// 2 seconds, so broadcasting more often mostly re-sent the same resources. +const defaultStateBroadcastInterval = 2 * time.Second + // Message represents a WebSocket message type Message struct { Type string `json:"type"` @@ -702,6 +714,8 @@ func NewHub(getState func(orgID string) interface{}) *Hub { tenantCoalesceGeneration: make(map[string]uint64), stateBroadcastWake: make(chan struct{}, 1), stateBroadcastDone: make(chan struct{}), + stateBroadcastInterval: defaultStateBroadcastInterval, + tenantLastStateBroadcast: make(map[string]time.Time), } } @@ -1218,6 +1232,7 @@ func (h *Hub) markGlobalStateBroadcastReady(generation uint64) { h.coalesceReady = h.coalescePending h.coalescePending = nil h.coalesceTimer = nil + h.lastStateBroadcast = time.Now() h.coalesceMutex.Unlock() h.wakeStateBroadcastWorker() } @@ -1237,6 +1252,7 @@ func (h *Hub) markTenantStateBroadcastReady(orgID string, generation uint64) { delete(h.tenantCoalescePending, orgID) delete(h.tenantCoalesceTimers, orgID) delete(h.tenantCoalesceGeneration, orgID) + h.tenantLastStateBroadcast[orgID] = time.Now() h.coalesceMutex.Unlock() h.wakeStateBroadcastWorker() } @@ -1401,6 +1417,20 @@ func (h *Hub) marshalBroadcastMessage(msg Message, orgID string) ([]byte, bool) return data, true } +// stateBroadcastDelayLocked returns how long a newly requested state +// broadcast waits: coalesceWindow to merge simultaneous signals, and at least +// stateBroadcastInterval after the audience's previous broadcast. The caller +// holds coalesceMutex. +func (h *Hub) stateBroadcastDelayLocked(last time.Time) time.Duration { + delay := h.coalesceWindow + if !last.IsZero() && h.stateBroadcastInterval > 0 { + if wait := h.stateBroadcastInterval - time.Since(last); wait > delay { + delay = wait + } + } + return delay +} + // runBroadcastSequencer handles sequenced broadcasts with coalescing for rapid state updates func (h *Hub) runBroadcastSequencer() { for { @@ -1410,22 +1440,18 @@ func (h *Hub) runBroadcastSequencer() { if msg.Type == "rawData" { h.coalesceMutex.Lock() - // Cancel pending timer if exists - if h.coalesceTimer != nil { - h.coalesceTimer.Stop() - } - - // Update pending message + // The latest state wins. A due broadcast is not rescheduled, so + // a steady stream of signals cannot postpone it. current := msg h.coalescePending = ¤t - h.nextGeneration++ - generation := h.nextGeneration - h.coalesceGeneration = generation - - // Set timer to send after coalesce window - h.coalesceTimer = time.AfterFunc(h.coalesceWindow, func() { - h.markGlobalStateBroadcastReady(generation) - }) + if h.coalesceTimer == nil { + h.nextGeneration++ + generation := h.nextGeneration + h.coalesceGeneration = generation + h.coalesceTimer = time.AfterFunc(h.stateBroadcastDelayLocked(h.lastStateBroadcast), func() { + h.markGlobalStateBroadcastReady(generation) + }) + } h.coalesceMutex.Unlock() } else { @@ -1440,23 +1466,18 @@ func (h *Hub) runBroadcastSequencer() { if tb.Message.Type == "rawData" { h.coalesceMutex.Lock() - // Cancel pending timer for this tenant if exists - if timer := h.tenantCoalesceTimers[tb.OrgID]; timer != nil { - timer.Stop() - } - - // Update pending message for this tenant + // The latest state wins; a due tenant broadcast keeps its time. msgCopy := tb.Message h.tenantCoalescePending[tb.OrgID] = &msgCopy - h.nextGeneration++ - generation := h.nextGeneration - h.tenantCoalesceGeneration[tb.OrgID] = generation - - // Set timer to send after coalesce window - orgID := tb.OrgID // Capture for closure - h.tenantCoalesceTimers[orgID] = time.AfterFunc(h.coalesceWindow, func() { - h.markTenantStateBroadcastReady(orgID, generation) - }) + if h.tenantCoalesceTimers[tb.OrgID] == nil { + h.nextGeneration++ + generation := h.nextGeneration + h.tenantCoalesceGeneration[tb.OrgID] = generation + orgID := tb.OrgID // Capture for closure + h.tenantCoalesceTimers[orgID] = time.AfterFunc(h.stateBroadcastDelayLocked(h.tenantLastStateBroadcast[orgID]), func() { + h.markTenantStateBroadcastReady(orgID, generation) + }) + } h.coalesceMutex.Unlock() } else { diff --git a/internal/websocket/hub_test.go b/internal/websocket/hub_test.go index 9dd250bec7..1c186f6e7d 100644 --- a/internal/websocket/hub_test.go +++ b/internal/websocket/hub_test.go @@ -1245,3 +1245,126 @@ func TestBroadcastCurrentStateToTenantResolvesTenantState(t *testing.T) { t.Fatalf("lazy tenant state org = %v, want org-123", payload["org"]) } } + +// startStateBroadcastHarness runs the sequencer and state worker for hub and +// registers one client, optionally in a tenant. +func startStateBroadcastHarness(t *testing.T, hub *Hub, orgID string) *Client { + t.Helper() + client := &Client{hub: hub, send: make(chan []byte, 16), id: "state-rate-client", orgID: orgID} + hub.mu.Lock() + hub.clients[client] = true + if orgID != "" { + if hub.clientsByTenant[orgID] == nil { + hub.clientsByTenant[orgID] = make(map[*Client]bool) + } + hub.clientsByTenant[orgID][client] = true + } + hub.mu.Unlock() + sequencerDone := make(chan struct{}) + go func() { + hub.runBroadcastSequencer() + close(sequencerDone) + }() + go hub.runStateBroadcastWorker() + t.Cleanup(func() { + close(hub.stopChan) + <-sequencerDone + <-hub.stateBroadcastDone + }) + return client +} + +func receiveStateValue(t *testing.T, client *Client, within time.Duration) (string, bool) { + t.Helper() + select { + case data := <-client.send: + var msg Message + if err := json.Unmarshal(data, &msg); err != nil { + t.Fatalf("unmarshal message: %v", err) + } + payload, _ := msg.Data.(map[string]interface{}) + value, _ := payload["value"].(string) + return value, true + case <-time.After(within): + return "", false + } +} + +// Every accepted agent report requests a state broadcast, and each broadcast +// rebuilds the full frontend state. Broadcasts to one audience are spaced by +// stateBroadcastInterval and carry the latest state (#2199). +func TestStateBroadcastsSpacedByInterval(t *testing.T) { + hub := NewHub(nil) + hub.coalesceWindow = 5 * time.Millisecond + hub.stateBroadcastInterval = 300 * time.Millisecond + client := startStateBroadcastHarness(t, hub, "") + + hub.broadcastSeq <- Message{Type: "rawData", Data: map[string]string{"value": "first"}} + if value, ok := receiveStateValue(t, client, time.Second); !ok || value != "first" { + t.Fatalf("first broadcast = %q (received %v), want an immediate first broadcast", value, ok) + } + start := time.Now() + for _, value := range []string{"second", "third", "fourth"} { + hub.broadcastSeq <- Message{Type: "rawData", Data: map[string]string{"value": value}} + } + if value, ok := receiveStateValue(t, client, 150*time.Millisecond); ok { + t.Fatalf("broadcast %q arrived %v after the previous one, inside the interval", value, time.Since(start)) + } + if value, ok := receiveStateValue(t, client, 2*time.Second); !ok || value != "fourth" { + t.Fatalf("spaced broadcast = %q (received %v), want the latest state", value, ok) + } + if value, ok := receiveStateValue(t, client, 100*time.Millisecond); ok { + t.Fatalf("burst produced a second spaced broadcast %q", value) + } +} + +// A signal arriving while a broadcast is due must not push it back. With the +// earlier restart-on-signal debounce, signals closer together than the window +// postponed every broadcast until they stopped. +func TestStateBroadcastNotStarvedBySteadySignals(t *testing.T) { + hub := NewHub(nil) + hub.coalesceWindow = 60 * time.Millisecond + hub.stateBroadcastInterval = 0 + client := startStateBroadcastHarness(t, hub, "") + + stop := time.After(600 * time.Millisecond) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + received := 0 + for streaming := true; streaming; { + select { + case <-ticker.C: + hub.broadcastSeq <- Message{Type: "rawData", Data: map[string]string{"value": "tick"}} + case <-client.send: + received++ + case <-stop: + streaming = false + } + } + if received == 0 { + t.Fatal("no state broadcast while signals kept arriving every 10 ms") + } +} + +func TestTenantStateBroadcastsSpacedByInterval(t *testing.T) { + hub := NewHub(nil) + hub.coalesceWindow = 5 * time.Millisecond + hub.stateBroadcastInterval = 300 * time.Millisecond + client := startStateBroadcastHarness(t, hub, "acme") + + send := func(value string) { + hub.tenantBroadcast <- TenantBroadcast{OrgID: "acme", Message: Message{Type: "rawData", Data: map[string]string{"value": value}}} + } + send("first") + if value, ok := receiveStateValue(t, client, time.Second); !ok || value != "first" { + t.Fatalf("first tenant broadcast = %q (received %v)", value, ok) + } + send("second") + send("third") + if value, ok := receiveStateValue(t, client, 150*time.Millisecond); ok { + t.Fatalf("tenant broadcast %q arrived inside the interval", value) + } + if value, ok := receiveStateValue(t, client, 2*time.Second); !ok || value != "third" { + t.Fatalf("spaced tenant broadcast = %q (received %v), want the latest state", value, ok) + } +}