fix(events): coalesce drop warnings; raise subscriber buffer 50→500

A slow SSE consumer falling behind on a bursty event stream
(logs.updated emits one event per tailed log line) overflows the
50-slot subscriber channel and produces one WARN per dropped event,
flooding gearbox-agent's journal with hundreds of identical lines.

- Raise per-subscriber Events channel from 50 to 500 to absorb typical
  log bursts without dropping.
- When drops do happen, aggregate per (subscriber, event_type) and log
  at most once per 10s with a cumulative count + window duration, so a
  burst is one line instead of N.
- Clean up the drop tracker when a subscriber unsubscribes so it
  doesn't grow unbounded with churning SSE clients.

Related: #60

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: sarg3nt

gearbox/internal/framework/events/hub.go

@@ -92,6 +92,35 @@ type Subscriber struct {
 	done     chan struct{}
 }
 
+// subscriberEventBufferSize is the per-subscriber Events channel capacity.
+// Sized to absorb bursts of high-frequency events (e.g. logs.updated emits
+// one event per log line tailed on the agent) without dropping. A slow SSE
+// consumer that stays behind this for the full window will still drop, but
+// the warning is throttled — see dropLogInterval.
+const subscriberEventBufferSize = 500
+
+// dropLogInterval is the minimum gap between consecutive "channel full"
+// warnings for the same (subscriber, event_type) pair. Drops still happen
+// at the same rate; only the log output is coalesced so a bursty consumer
+// produces one line per interval instead of one per dropped event.
+const dropLogInterval = 10 * time.Second
+
+// dropKey identifies a stream of drops to coalesce. Per-(subscriber, event-type)
+// so an overflow on logs.updated doesn't suppress a separate overflow on
+// metrics.updated for the same subscriber.
+type dropKey struct {
+	subID     string
+	eventType EventType
+}
+
+// dropAggregator tracks drops between log emissions for one dropKey.
+// Only accessed from the Hub's run goroutine, so no synchronization.
+type dropAggregator struct {
+	count      int64
+	firstSeen  time.Time
+	lastLogged time.Time
+}
+
 // Hub manages event distribution to subscribers.
 type Hub struct {
 	subscribers map[string]*Subscriber
@@ -101,6 +130,9 @@ type Hub struct {
 	register    chan *Subscriber
 	unregister  chan *Subscriber
 	done        chan struct{}
+	// subscriberDrops is owned by run() and must not be touched from other
+	// goroutines.
+	subscriberDrops map[dropKey]*dropAggregator
 }
 
 // NewHub creates a new event hub.
@@ -110,12 +142,13 @@ func NewHub(logger *slog.Logger) *Hub {
 	}
 
 	h := &Hub{
-		subscribers: make(map[string]*Subscriber),
-		logger:      logger,
-		broadcast:   make(chan Event, 100),
-		register:    make(chan *Subscriber),
-		unregister:  make(chan *Subscriber),
-		done:        make(chan struct{}),
+		subscribers:     make(map[string]*Subscriber),
+		logger:          logger,
+		broadcast:       make(chan Event, 100),
+		register:        make(chan *Subscriber),
+		unregister:      make(chan *Subscriber),
+		done:            make(chan struct{}),
+		subscriberDrops: make(map[dropKey]*dropAggregator),
 	}
 
 	return h
@@ -159,6 +192,7 @@ func (h *Hub) run() {
 				h.logger.Debug("subscriber unregistered", "id", sub.ID)
 			}
 			h.mu.Unlock()
+			h.cleanupDropTracker(sub.ID)
 
 		case event := <-h.broadcast:
 			h.mu.RLock()
@@ -172,10 +206,7 @@ func (h *Hub) run() {
 				select {
 				case sub.Events <- event:
 				default:
-					// Channel full, skip this event for this subscriber
-					h.logger.Warn("subscriber event channel full, dropping event",
-						"subscriber_id", sub.ID,
-						"event_type", event.Type)
+					h.recordDrop(sub.ID, event.Type)
 				}
 			}
 			h.mu.RUnlock()
@@ -188,14 +219,61 @@ func (h *Hub) Subscribe(id string, serverID string) *Subscriber {
 	sub := &Subscriber{
 		ID:       id,
 		ServerID: serverID,
-		Events:   make(chan Event, 50),
+		Events:   make(chan Event, subscriberEventBufferSize),
 		done:     make(chan struct{}),
 	}
 
 	h.register <- sub
 	return sub
 }
 
+// recordDrop accounts for one dropped event and emits a coalesced warning
+// at most once per dropLogInterval per (subscriber, event type). The first
+// drop in a window logs immediately so a fresh problem isn't hidden; later
+// drops accumulate until the interval elapses, then log with the count.
+// Called only from run(); no synchronization on subscriberDrops.
+func (h *Hub) recordDrop(subID string, eventType EventType) {
+	h.recordDropAt(subID, eventType, time.Now())
+}
+
+// recordDropAt is the testable form of recordDrop. now is the timestamp the
+// caller wants treated as "now" — production code passes time.Now(), tests
+// pass a controlled time so throttling behavior is deterministic.
+func (h *Hub) recordDropAt(subID string, eventType EventType, now time.Time) {
+	key := dropKey{subID: subID, eventType: eventType}
+
+	agg, ok := h.subscriberDrops[key]
+	if !ok {
+		agg = &dropAggregator{firstSeen: now}
+		h.subscriberDrops[key] = agg
+	}
+	agg.count++
+
+	if now.Sub(agg.lastLogged) < dropLogInterval {
+		return
+	}
+
+	h.logger.Warn("subscriber event channel full, dropping event(s)",
+		"subscriber_id", subID,
+		"event_type", eventType,
+		"dropped", agg.count,
+		"window", now.Sub(agg.firstSeen).Round(time.Millisecond))
+	agg.lastLogged = now
+	agg.firstSeen = now
+	agg.count = 0
+}
+
+// cleanupDropTracker removes any drop-aggregator entries for a subscriber
+// that has unsubscribed, so the map doesn't grow with churning subscribers.
+// Called only from run().
+func (h *Hub) cleanupDropTracker(subID string) {
+	for k := range h.subscriberDrops {
+		if k.subID == subID {
+			delete(h.subscriberDrops, k)
+		}
+	}
+}
+
 // Unsubscribe removes a subscriber.
 func (h *Hub) Unsubscribe(sub *Subscriber) {
 	select {

gearbox/internal/framework/events/hub_test.go

@@ -0,0 +1,173 @@
+package events
+
+import (
+	"bytes"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// newTestHub returns a Hub backed by a buffered slog handler so tests can
+// inspect emitted warnings.
+func newTestHub(t *testing.T) (*Hub, *bytes.Buffer, *sync.Mutex) {
+	t.Helper()
+	var buf bytes.Buffer
+	var mu sync.Mutex
+	handler := slog.NewTextHandler(&lockedWriter{w: &buf, mu: &mu}, &slog.HandlerOptions{Level: slog.LevelDebug})
+	logger := slog.New(handler)
+	return NewHub(logger), &buf, &mu
+}
+
+// lockedWriter serialises writes from goroutines that share the slog handler
+// so the buffer doesn't get torn output under -race.
+type lockedWriter struct {
+	w  *bytes.Buffer
+	mu *sync.Mutex
+}
+
+func (l *lockedWriter) Write(p []byte) (int, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.w.Write(p)
+}
+
+func countLines(buf *bytes.Buffer, mu *sync.Mutex, substr string) int {
+	mu.Lock()
+	defer mu.Unlock()
+	return strings.Count(buf.String(), substr)
+}
+
+func TestSubscriberEventBufferIsLarge(t *testing.T) {
+	hub, _, _ := newTestHub(t)
+	hub.Start()
+	defer hub.Stop()
+
+	sub := hub.Subscribe("sub-1", "")
+	if got := cap(sub.Events); got != subscriberEventBufferSize {
+		t.Fatalf("subscriber event channel capacity = %d, want %d", got, subscriberEventBufferSize)
+	}
+}
+
+func TestRecordDropFirstDropLogsImmediately(t *testing.T) {
+	hub, buf, mu := newTestHub(t)
+	t0 := time.Date(2026, 5, 13, 12, 0, 0, 0, time.UTC)
+
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0)
+
+	if got := countLines(buf, mu, "subscriber event channel full"); got != 1 {
+		t.Fatalf("expected 1 warning on first drop, got %d. log=%q", got, buf.String())
+	}
+}
+
+func TestRecordDropCoalescesWithinWindow(t *testing.T) {
+	hub, buf, mu := newTestHub(t)
+	t0 := time.Date(2026, 5, 13, 12, 0, 0, 0, time.UTC)
+
+	// First drop logs.
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0)
+	// 99 more drops within the window — should not log again.
+	for i := 1; i <= 99; i++ {
+		hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0.Add(time.Duration(i)*time.Millisecond))
+	}
+
+	if got := countLines(buf, mu, "subscriber event channel full"); got != 1 {
+		t.Fatalf("expected 1 coalesced warning, got %d. log=%q", got, buf.String())
+	}
+}
+
+func TestRecordDropEmitsAfterIntervalWithCount(t *testing.T) {
+	hub, buf, mu := newTestHub(t)
+	t0 := time.Date(2026, 5, 13, 12, 0, 0, 0, time.UTC)
+
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0)
+	for i := 1; i <= 49; i++ {
+		hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0.Add(time.Duration(i)*time.Millisecond))
+	}
+	// Cross the interval: the next drop should log again, this time with
+	// dropped=50 (the 49 coalesced + this one).
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0.Add(dropLogInterval+time.Millisecond))
+
+	if got := countLines(buf, mu, "subscriber event channel full"); got != 2 {
+		t.Fatalf("expected 2 warnings after crossing interval, got %d. log=%q", got, buf.String())
+	}
+	if !strings.Contains(buf.String(), "dropped=50") {
+		t.Fatalf("expected coalesced count of 50 in second warning, log=%q", buf.String())
+	}
+}
+
+func TestRecordDropPerSubscriberPerEventTypeIsIndependent(t *testing.T) {
+	hub, buf, mu := newTestHub(t)
+	t0 := time.Date(2026, 5, 13, 12, 0, 0, 0, time.UTC)
+
+	// Three distinct (subscriber, event_type) streams in the same instant —
+	// each should produce its own first-drop warning.
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0)
+	hub.recordDropAt("sub-1", EventTypeMetricsUpdated, t0)
+	hub.recordDropAt("sub-2", EventTypeLogsUpdated, t0)
+
+	if got := countLines(buf, mu, "subscriber event channel full"); got != 3 {
+		t.Fatalf("expected 3 independent first-drop warnings, got %d. log=%q", got, buf.String())
+	}
+}
+
+func TestCleanupDropTrackerRemovesSubscriberEntries(t *testing.T) {
+	hub, _, _ := newTestHub(t)
+	t0 := time.Date(2026, 5, 13, 12, 0, 0, 0, time.UTC)
+
+	hub.recordDropAt("sub-1", EventTypeLogsUpdated, t0)
+	hub.recordDropAt("sub-1", EventTypeMetricsUpdated, t0)
+	hub.recordDropAt("sub-2", EventTypeLogsUpdated, t0)
+
+	if got := len(hub.subscriberDrops); got != 3 {
+		t.Fatalf("expected 3 tracker entries, got %d", got)
+	}
+
+	hub.cleanupDropTracker("sub-1")
+
+	if got := len(hub.subscriberDrops); got != 1 {
+		t.Fatalf("expected 1 tracker entry after cleanup, got %d", got)
+	}
+	if _, ok := hub.subscriberDrops[dropKey{subID: "sub-2", eventType: EventTypeLogsUpdated}]; !ok {
+		t.Fatal("sub-2 entry should have survived cleanup of sub-1")
+	}
+}
+
+func TestSlowSubscriberDropsAreCoalesced(t *testing.T) {
+	// End-to-end through the broadcast loop: a subscriber that never drains
+	// will overflow, but we should still emit only one warning per event-type
+	// during the burst (no clock fast-forwarding here, so we stay inside
+	// dropLogInterval).
+	hub, buf, mu := newTestHub(t)
+	hub.Start()
+	defer hub.Stop()
+
+	sub := hub.Subscribe("slow-sub", "")
+	// Never read sub.Events.
+	_ = sub
+
+	// Publish well past the buffer size so a flood of drops occurs.
+	for i := 0; i < subscriberEventBufferSize*3; i++ {
+		hub.Publish(Event{Type: EventTypeLogsUpdated})
+	}
+
+	// Let the broadcast loop drain.
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if countLines(buf, mu, "subscriber event channel full") >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	got := countLines(buf, mu, "subscriber event channel full")
+	if got == 0 {
+		t.Fatalf("expected at least one drop warning under burst, got none. log=%q", buf.String())
+	}
+	// Under the throttle, a single burst should produce far fewer warnings
+	// than dropped events. Allow some slack but assert it's not 1-per-drop.
+	if got > 5 {
+		t.Fatalf("expected coalesced warning(s), got %d (no throttling?). log=%q", got, buf.String())
+	}
+}