From 5047741bc0d2f6093b76adbdaa60ae3ca5f07c88 Mon Sep 17 00:00:00 2001
From: Atishyy27 <sethatishayjain@gmail.com>
Date: Sun, 16 Aug 2026 01:19:22 +0530
Subject: [PATCH] [broadcast] Use non-blocking send in broadcast() so one
 stalled subscriber cannot freeze the bus

broadcast() sent to each registered channel with a plain, blocking
ch <- m. If any one subscriber stops draining its channel - a dead
consumer, a panicked goroutine on the other end, anything that just
falls behind - the single run() goroutine wedges on that send. Since
Register, Unregister and every future Submit all go through that same
goroutine (the select in run()), one stuck subscriber freezes delivery
to every other subscriber and blocks all Register/Unregister calls,
indefinitely.

This pattern is inherited as-is from the reference implementation this
file's own header cites (github.com/dustin/go-broadcast, broadcaster.go
on master as of this writing) - upstream's broadcast() has the same
unprotected ch <- m, so this isn't a meshkit-introduced regression, but
it is a real gap in an exported, shared-library type.

Fix: wrap the send in select with a default case, so a subscriber that
isn't ready has its message dropped rather than blocking the shared
goroutine for everyone else.

Regression test registers a slow (never-drained) and a fast subscriber,
submits one message, and asserts the fast subscriber still receives it
and that a subsequent Register/Unregister each return within 2s. Verified
the test fails on the pre-fix code (fast subscriber times out) and passes
after the fix.

Signed-off-by: Atishyy27 <sethatishayjain@gmail.com>
---
 utils/broadcast/broadcaster.go      |  5 ++-
 utils/broadcast/broadcaster_test.go | 50 +++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 1 deletion(-)
 create mode 100644 utils/broadcast/broadcaster_test.go

diff --git a/utils/broadcast/broadcaster.go b/utils/broadcast/broadcaster.go
index 158bae67..19022800 100644
--- a/utils/broadcast/broadcaster.go
+++ b/utils/broadcast/broadcaster.go
@@ -52,7 +52,10 @@ type Broadcaster interface {
 
 func (b *broadcaster) broadcast(m BroadcastMessage) {
 	for ch := range b.outputs {
-		ch <- m
+		select {
+		case ch <- m:
+		default:
+		}
 	}
 }
 
diff --git a/utils/broadcast/broadcaster_test.go b/utils/broadcast/broadcaster_test.go
new file mode 100644
index 00000000..57725f51
--- /dev/null
+++ b/utils/broadcast/broadcaster_test.go
@@ -0,0 +1,50 @@
+package broadcast
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBroadcasterSlowSubscriberDoesNotBlockOthers(t *testing.T) {
+	b := NewBroadcaster(10)
+
+	// A subscriber that never drains its channel.
+	slow := make(chan BroadcastMessage)
+	b.Register(slow)
+
+	// A subscriber that does drain, buffered so Submit below doesn't need a
+	// live reader yet.
+	fast := make(chan BroadcastMessage, 1)
+	b.Register(fast)
+
+	b.Submit(BroadcastMessage{Type: "first"})
+
+	// The fast subscriber must still receive the message even though the
+	// slow one never reads it.
+	select {
+	case got := <-fast:
+		assert.Equal(t, "first", got.Type)
+	case <-time.After(2 * time.Second):
+		t.Fatal("fast subscriber never received the broadcast message")
+	}
+
+	// Register/Unregister go through the same goroutine as broadcast(), so
+	// they must not be blocked by the stalled slow subscriber either.
+	for _, op := range []struct {
+		name string
+		fn   func()
+	}{
+		{"Register", func() { b.Register(make(chan BroadcastMessage, 1)) }},
+		{"Unregister", func() { b.Unregister(slow) }},
+	} {
+		done := make(chan struct{})
+		go func() { op.fn(); close(done) }()
+		select {
+		case <-done:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("%s blocked for >2s behind the stalled subscriber", op.name)
+		}
+	}
+}