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