From 66b239da7dd7b2c06032cf22abee49125352b8be Mon Sep 17 00:00:00 2001
From: MHSanaei <ho3ein.sanaei@gmail.com>
Date: Wed, 15 Jul 2026 04:34:06 +0200
Subject: [PATCH] fix(eventbus): dispatch each subscriber in its own goroutine
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The fan-out loop called every subscriber's handler sequentially on the
single dispatch goroutine. The email and Telegram notifiers block on
network I/O for tens of seconds (or minutes when the remote is slow), so
one slow subscriber stalled the whole loop: the 256-slot channel then
filled and Publish silently dropped later events — including high-value
xray.crash and node.down notifications unrelated to the slow handler.

Hand each delivered event to every handler in its own goroutine so a
blocking subscriber can no longer stall delivery to the others. safeCall
already recovers panics, so a detached handler cannot take down the bus.
---
 internal/eventbus/bus.go      | 15 +++++++++------
 internal/eventbus/bus_test.go | 25 +++++++++++++++++++++++++
 2 files changed, 34 insertions(+), 6 deletions(-)

diff --git a/internal/eventbus/bus.go b/internal/eventbus/bus.go
index 6a20669f0..fe4324ae3 100644
--- a/internal/eventbus/bus.go
+++ b/internal/eventbus/bus.go
@@ -81,9 +81,11 @@ func (b *Bus) Publish(e Event) {
 	}
 }
 
-// dispatch is the fan-out loop. It reads events from the channel and calls
-// every subscriber's handler sequentially. Handlers run on the dispatch
-// goroutine — they must not block.
+// 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.
 func (b *Bus) dispatch() {
 	defer b.wg.Done()
 	for {
@@ -97,7 +99,7 @@ func (b *Bus) dispatch() {
 			copy(subs, b.subs)
 			b.mu.RUnlock()
 			for _, s := range subs {
-				safeCall(s.handler, e)
+				go safeCall(s.handler, e)
 			}
 		case <-b.done:
 			return
@@ -115,8 +117,9 @@ func safeCall(fn func(Event), e Event) {
 	fn(e)
 }
 
-// Stop shuts down the bus: the dispatch goroutine exits, in-flight handlers
-// finish, and any events still buffered may be dropped. Safe to call once.
+// 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.
 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 43ae785d5..76e80fa7e 100644
--- a/internal/eventbus/bus_test.go
+++ b/internal/eventbus/bus_test.go
@@ -149,6 +149,31 @@ func TestBusPanicRecovery(t *testing.T) {
 	}
 }
 
+func TestBusBlockingSubscriberDoesNotStallOthers(t *testing.T) {
+	b := New(16)
+	defer b.Stop()
+
+	release := make(chan struct{})
+	b.Subscribe("blocking", func(e Event) {
+		<-release
+	})
+
+	fast := make(chan struct{}, 1)
+	b.Subscribe("fast", func(e Event) {
+		fast <- struct{}{}
+	})
+
+	b.Publish(Event{Type: EventXrayCrash})
+
+	select {
+	case <-fast:
+	case <-time.After(time.Second):
+		close(release)
+		t.Fatal("a blocking subscriber stalled event delivery to another subscriber")
+	}
+	close(release)
+}
+
 func TestBusBufferFull(t *testing.T) {
 	b := New(2)
 	defer b.Stop()