From 0061892d8799cb593dacc36af35d672e6db2c9b9 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Wed, 15 Jul 2026 06:27:26 +0200 Subject: [PATCH] fix(eventbus): deliver events on a bounded per-subscriber worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 82 ++++++++++++++++++++++++++--------- internal/eventbus/bus_test.go | 37 ++++++++++++++++ 2 files changed, 98 insertions(+), 21 deletions(-) diff --git a/internal/eventbus/bus.go b/internal/eventbus/bus.go index fe4324ae3..b9cd34c9a 100644 --- a/internal/eventbus/bus.go +++ b/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() diff --git a/internal/eventbus/bus_test.go b/internal/eventbus/bus_test.go index 76e80fa7e..57a3cce72 100644 --- a/internal/eventbus/bus_test.go +++ b/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()