fix(eventbus): deliver events on a bounded per-subscriber worker

The previous fix dispatched each event to every subscriber with a bare
`go safeCall`. That unblocked the dispatch loop, but removed the bus's
backpressure: under a login-attempt flood (which both notifier subscribers
process without rate-limiting) with email/Telegram enabled, every attempt
spawned handler goroutines that each block on network I/O for up to ~30s,
with no bound — a goroutine and outbound-connection storm. It also let a
subscriber's handler run concurrently with itself, racing the Telegram
notifier's lazily-cached hostname.

Give each subscriber its own bounded queue drained by a single worker
goroutine. Dispatch does a non-blocking send per subscriber (dropping only
that subscriber's event when its queue is full), so a slow subscriber still
can't stall the others, concurrency is bounded to one in-flight handler per
subscriber, per-subscriber event order is preserved, and Stop again waits
for in-flight handlers to finish.

internal/eventbus/bus.go

@@ -10,10 +10,19 @@ import (
 // DefaultBufferSize is the number of events the bus can hold before Publish starts dropping.
 const DefaultBufferSize = 256
 
-// subscriber pairs an ID with its event handler.
+// subscriberQueueSize bounds how many undelivered events a single subscriber may
+// hold before the newest are dropped. Each subscriber drains its own queue on a
+// dedicated worker goroutine, so a slow subscriber can neither stall delivery to
+// the others nor make the bus spawn an unbounded number of goroutines.
+const subscriberQueueSize = 64
+
+// subscriber pairs an ID with its event handler and the per-subscriber worker
+// state used to deliver events to it serially, without blocking the dispatch loop.
 type subscriber struct {
 	id      string
 	handler func(Event)
+	queue   chan Event
+	quit    chan struct{}
 }
 
 // Bus is a minimal in-process pub/sub event bus backed by a buffered channel.
@@ -21,7 +30,7 @@ type subscriber struct {
 // subscribers; per-event filtering is the subscriber's responsibility.
 type Bus struct {
 	ch   chan Event
-	subs []subscriber
+	subs []*subscriber
 	mu   sync.RWMutex
 	done chan struct{}
 	wg   sync.WaitGroup
@@ -41,27 +50,38 @@ func New(bufSize int) *Bus {
 	return b
 }
 
-// Subscribe registers a handler that receives every published event.
-// The id is used for Unsubscribe; it must be unique across active subscribers.
-// Subscribing with an already-registered id replaces the previous handler.
+// Subscribe registers a handler that receives every published event on its own
+// worker goroutine. The id is used for Unsubscribe; it must be unique across
+// active subscribers. Subscribing with an already-registered id replaces the
+// previous subscriber, stopping its worker.
 func (b *Bus) Subscribe(id string, handler func(Event)) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	for i, s := range b.subs {
 		if s.id == id {
-			b.subs[i].handler = handler
-			return
+			close(s.quit)
+			b.subs = append(b.subs[:i], b.subs[i+1:]...)
+			break
 		}
 	}
-	b.subs = append(b.subs, subscriber{id: id, handler: handler})
+	s := &subscriber{
+		id:      id,
+		handler: handler,
+		queue:   make(chan Event, subscriberQueueSize),
+		quit:    make(chan struct{}),
+	}
+	b.subs = append(b.subs, s)
+	b.wg.Add(1)
+	go b.runWorker(s)
 }
 
-// Unsubscribe removes a subscriber by id. Safe to call with unknown id.
+// Unsubscribe removes a subscriber by id and stops its worker. Safe to call with an unknown id.
 func (b *Bus) Unsubscribe(id string) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	for i, s := range b.subs {
 		if s.id == id {
+			close(s.quit)
 			b.subs = append(b.subs[:i], b.subs[i+1:]...)
 			return
 		}
@@ -82,10 +102,12 @@ func (b *Bus) Publish(e Event) {
 }
 
 // dispatch is the fan-out loop. It reads events from the channel and hands each
-// one to every subscriber's handler in its own goroutine, so a subscriber whose
-// handler blocks on network I/O (the email and Telegram notifiers can block for
-// tens of seconds) cannot stall delivery of unrelated, higher-value events such
-// as xray.crash or node.down.
+// one to every subscriber's queue with a non-blocking send, so a subscriber
+// whose handler blocks on network I/O (the email and Telegram notifiers can
+// block for tens of seconds) can neither stall delivery of unrelated, higher-
+// value events such as xray.crash or node.down, nor force the bus to spawn an
+// unbounded number of goroutines under load. A subscriber whose queue is full
+// drops the event, keeping the bus non-blocking and its memory bounded.
 func (b *Bus) dispatch() {
 	defer b.wg.Done()
 	for {
@@ -95,12 +117,30 @@ func (b *Bus) dispatch() {
 				return
 			}
 			b.mu.RLock()
-			subs := make([]subscriber, len(b.subs))
-			copy(subs, b.subs)
-			b.mu.RUnlock()
-			for _, s := range subs {
-				go safeCall(s.handler, e)
+			for _, s := range b.subs {
+				select {
+				case s.queue <- e:
+				default:
+					logger.Warning("eventbus: subscriber ", s.id, " queue full, dropping ", e.Type)
+				}
 			}
+			b.mu.RUnlock()
+		case <-b.done:
+			return
+		}
+	}
+}
+
+// runWorker delivers queued events to one subscriber serially, so a subscriber
+// never runs concurrently with itself and observes events in publication order.
+func (b *Bus) runWorker(s *subscriber) {
+	defer b.wg.Done()
+	for {
+		select {
+		case e := <-s.queue:
+			safeCall(s.handler, e)
+		case <-s.quit:
+			return
 		case <-b.done:
 			return
 		}
@@ -117,9 +157,9 @@ func safeCall(fn func(Event), e Event) {
 	fn(e)
 }
 
-// Stop shuts down the bus: the dispatch goroutine exits and any events still
-// buffered may be dropped. Handler goroutines already spawned for delivered
-// events run to completion on their own. Safe to call once.
+// Stop shuts down the bus: the dispatch loop and every subscriber worker exit
+// after finishing any handler already in progress, and any events still buffered
+// or queued may be dropped. Safe to call once.
 func (b *Bus) Stop() {
 	close(b.done)
 	b.wg.Wait()

internal/eventbus/bus_test.go

@@ -174,6 +174,43 @@ func TestBusBlockingSubscriberDoesNotStallOthers(t *testing.T) {
 	close(release)
 }
 
+func TestBusSubscriberRunsSerially(t *testing.T) {
+	b := New(16)
+	defer b.Stop()
+
+	var inFlight atomic.Int32
+	var maxSeen atomic.Int32
+	var wg sync.WaitGroup
+	const n = 8
+	wg.Add(n)
+
+	b.Subscribe("serial", func(Event) {
+		cur := inFlight.Add(1)
+		for {
+			m := maxSeen.Load()
+			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
+				break
+			}
+		}
+		time.Sleep(5 * time.Millisecond)
+		inFlight.Add(-1)
+		wg.Done()
+	})
+
+	for i := 0; i < n; i++ {
+		b.Publish(Event{Type: EventXrayCrash})
+	}
+
+	select {
+	case <-waitDone(&wg):
+	case <-time.After(2 * time.Second):
+		t.Fatal("subscriber did not process all events")
+	}
+	if got := maxSeen.Load(); got != 1 {
+		t.Fatalf("subscriber ran concurrently with itself: max in-flight = %d, want 1", got)
+	}
+}
+
 func TestBusBufferFull(t *testing.T) {
 	b := New(2)
 	defer b.Stop()