gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-23 20:25:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Space websocket state broadcasts at least 2s apart per audience

Browse Source 
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
This commit is contained in:
rcourtman
2026-09-24 11:21:20 +01:00
parent a6a189dd9b
commit 0bf66d7c72
3 changed files with 187 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/release-control/v6/internal/subsystems/api-contracts.md
+14
View File
@@ -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
internal/websocket/hub.go
+50 -29
View File
@@ -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 = &current
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 {
internal/websocket/hub_test.go
+123
View File
@@ -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)
}
}