From 55895f78fc36640fa9ea63ce6289840eadd669db Mon Sep 17 00:00:00 2001
From: Natalie Bridgers <natalie@stream.place>
Date: Wed, 30 Sep 2026 19:43:45 -0500
Subject: [PATCH] simplify bus overflow subscriptions

---
 pkg/api/websocket.go                          |  6 +---
 pkg/bus/bus.go                                | 13 +++----
 pkg/bus/bus_test.go                           | 34 ++++++++-----------
 pkg/media/key_revocation.go                   |  5 +--
 pkg/replication/iroh_replicator/kv.go         |  5 +--
 .../websocketrep/websocket_replicator.go      |  5 +--
 6 files changed, 22 insertions(+), 46 deletions(-)

diff --git a/pkg/api/websocket.go b/pkg/api/websocket.go
index 33766d295..1d71c5119 100644
--- a/pkg/api/websocket.go
+++ b/pkg/api/websocket.go
@@ -179,11 +179,7 @@ func (a *StreamplaceAPI) HandleWebsocket(ctx context.Context) httprouter.Handle
 
 			for {
 				select {
-				case msg, ok := <-ch:
-					if !ok {
-						log.Warn(ctx, "websocket bus subscription exceeded its backlog")
-						return
-					}
+				case msg := <-ch:
 					send(msg)
 				case msg := <-initialBurst:
 					send(msg)
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index 1c9579f09..d4ad96a8d 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -72,7 +72,6 @@ func (s *subscriber) close() {
 }
 
 func (s *subscriber) deliver() {
-	defer close(s.ch)
 	for {
 		s.mu.Lock()
 		for len(s.queue) == 0 && !s.closed {
@@ -131,20 +130,16 @@ func NewBus() *Bus {
 }
 
 func (b *Bus) Subscribe(user string) <-chan Message {
-	return b.subscribe(user, 0)
+	return b.subscribeWithOverflow(user, 0, nil)
 }
 
-// SubscribeWithBacklogLimit bounds queued messages and closes the subscription
-// when a consumer falls behind. Use this for clients that can reconnect and
-// receive a fresh snapshot after overflow.
+// SubscribeWithBacklogLimit bounds queued messages and calls onOverflow when a
+// consumer falls behind. Use this for clients that can reconnect and receive a
+// fresh snapshot after overflow.
 func (b *Bus) SubscribeWithBacklogLimit(user string, onOverflow func()) <-chan Message {
 	return b.subscribeWithOverflow(user, maxQueuedMessages, onOverflow)
 }
 
-func (b *Bus) subscribe(user string, limit int) <-chan Message {
-	return b.subscribeWithOverflow(user, limit, nil)
-}
-
 func (b *Bus) subscribeWithOverflow(user string, limit int, onOverflow func()) <-chan Message {
 	if b == nil {
 		return make(<-chan Message)
diff --git a/pkg/bus/bus_test.go b/pkg/bus/bus_test.go
index 5fe7e3b52..212b23955 100644
--- a/pkg/bus/bus_test.go
+++ b/pkg/bus/bus_test.go
@@ -25,36 +25,30 @@ func TestPublishPreservesSubscriptionOrder(t *testing.T) {
 	}
 }
 
-func TestPublishClosesSlowSubscriberAtBacklogLimit(t *testing.T) {
+func TestPublishCallsOverflowForSlowSubscriber(t *testing.T) {
 	b := NewBus()
 	overflowed := false
 	ch := b.SubscribeWithBacklogLimit("repo", func() {
 		overflowed = true
 	})
+	b.mu.Lock()
+	sub := b.clients["repo"][0]
+	b.mu.Unlock()
 	defer b.Unsubscribe("repo", ch)
 
 	for i := 0; i <= maxQueuedMessages+cap(ch)+1; i++ {
 		b.Publish("repo", i)
 	}
 
-	count := 0
-	timer := time.NewTimer(time.Second)
-	defer timer.Stop()
-	for {
-		select {
-		case _, ok := <-ch:
-			if !ok {
-				if !overflowed {
-					t.Fatal("overflow callback was not called")
-				}
-				if count > maxQueuedMessages+cap(ch) {
-					t.Fatalf("received %d messages before subscription closed", count)
-				}
-				return
-			}
-			count++
-		case <-timer.C:
-			t.Fatal("timed out waiting for slow subscription to close")
-		}
+	if !overflowed {
+		t.Fatal("overflow callback was not called")
+	}
+	sub.mu.Lock()
+	defer sub.mu.Unlock()
+	if !sub.closed {
+		t.Fatal("slow subscriber remained active")
+	}
+	if len(sub.queue) != 0 {
+		t.Fatalf("slow subscriber retained %d queued messages after overflow", len(sub.queue))
 	}
 }
diff --git a/pkg/media/key_revocation.go b/pkg/media/key_revocation.go
index fc5e35cb0..6dcfbbd3b 100644
--- a/pkg/media/key_revocation.go
+++ b/pkg/media/key_revocation.go
@@ -50,10 +50,7 @@ func (mm *MediaManager) watchKeyRevocation(ctx context.Context, streamer, did st
 		select {
 		case <-ctx.Done():
 			return
-		case msg, ok := <-sub:
-			if !ok {
-				return
-			}
+		case msg := <-sub:
 			switch v := msg.(type) {
 			case *model.SigningKey:
 				if v.RevokedAt != nil && v.DID == did {
diff --git a/pkg/replication/iroh_replicator/kv.go b/pkg/replication/iroh_replicator/kv.go
index 95c2faf29..92136ada3 100644
--- a/pkg/replication/iroh_replicator/kv.go
+++ b/pkg/replication/iroh_replicator/kv.go
@@ -270,10 +270,7 @@ func (swarm *IrohSwarm) startBusSubscribe(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case msg, ok := <-busCh:
-			if !ok {
-				return fmt.Errorf("bus subscription exceeded its backlog")
-			}
+		case msg := <-busCh:
 			if view, ok := msg.(*placestream.BroadcastDefs_BroadcastOriginView); ok {
 				log.Debug(ctx, "got broadcast origin view", "view", view)
 				err = swarm.handleOriginMessage(ctx, view)
diff --git a/pkg/replication/websocketrep/websocket_replicator.go b/pkg/replication/websocketrep/websocket_replicator.go
index 83ccbef23..0fbd21aba 100644
--- a/pkg/replication/websocketrep/websocket_replicator.go
+++ b/pkg/replication/websocketrep/websocket_replicator.go
@@ -181,10 +181,7 @@ func (r *WebsocketReplicator) startBusSubscribe(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case msg, ok := <-busCh:
-			if !ok {
-				return fmt.Errorf("bus subscription exceeded its backlog")
-			}
+		case msg := <-busCh:
 			if view, ok := msg.(*placestream.BroadcastDefs_BroadcastOriginView); ok {
 				log.Debug(ctx, "got broadcast origin view", "view", view)
 				err = r.handleOriginMessage(ctx, view)
-- 
2.54.0