From 79b90bda03583d6ef3ba7d0d41d44af076b5ec04 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Fri, 23 Oct 2026 06:46:18 -0400
Subject: [PATCH] fix: shutdown ends the live-event streams so a redeploy
 drains instead of waiting out its deadline (#5293)

http.Server.Shutdown waits for in-flight requests but never cancels them,
and an SSE stream only ends when its client leaves, so any open tab or
signed-in phone held every redeploy for the full 10s and the old task
exited 1. With the swarm's stop-first update that was ~10s of downtime per
deploy. The bus now closes on shutdown, ending each stream; clients
reconnect to the next instance.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 cmd/minstrel/main.go               |  5 ++++
 internal/api/events_stream.go      |  3 ++-
 internal/eventbus/eventbus.go      | 23 +++++++++++++++++
 internal/eventbus/eventbus_test.go | 41 ++++++++++++++++++++++++++++++
 4 files changed, 71 insertions(+), 1 deletion(-)

diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go
index 2298f3fa..252772a4 100644
--- a/cmd/minstrel/main.go
+++ b/cmd/minstrel/main.go
@@ -481,6 +481,11 @@ func run() error {
 		// server's limitRequestBody middleware instead.
 		IdleTimeout: 120 * time.Second,
 	}
+	// Shutdown waits for in-flight requests but never cancels them, and an
+	// SSE stream only ends when its client leaves. Closing the bus ends
+	// every stream as shutdown begins, so the drain waits only on real work
+	// such as an audio stream; clients reconnect to the next instance (#5293).
+	httpServer.RegisterOnShutdown(bus.Close)
 
 	errCh := make(chan error, 1)
 	go func() {
diff --git a/internal/api/events_stream.go b/internal/api/events_stream.go
index ff58deb0..9af3d8d6 100644
--- a/internal/api/events_stream.go
+++ b/internal/api/events_stream.go
@@ -105,7 +105,8 @@ func (h *handlers) handleEventsStream(w http.ResponseWriter, r *http.Request) {
 
 		case e, ok := <-ch:
 			if !ok {
-				// Bus closed our subscription (rare; only on full shutdown).
+				// The bus closed at shutdown. Ending here lets the
+				// server drain; the client reconnects to the next one.
 				return
 			}
 			// Scope filter: forward only events scoped to this user or
diff --git a/internal/eventbus/eventbus.go b/internal/eventbus/eventbus.go
index 1f3504e6..0280e5a2 100644
--- a/internal/eventbus/eventbus.go
+++ b/internal/eventbus/eventbus.go
@@ -38,6 +38,7 @@ type Event struct {
 type Bus struct {
 	mu          sync.RWMutex
 	subscribers map[chan Event]struct{}
+	closed      bool
 }
 
 // New returns a Bus with no subscribers.
@@ -74,6 +75,13 @@ func (b *Bus) Subscribe(buffer int) (<-chan Event, func()) {
 	}
 	ch := make(chan Event, buffer)
 	b.mu.Lock()
+	if b.closed {
+		// A subscriber arriving during shutdown ends at once, like the
+		// ones Close already ended.
+		b.mu.Unlock()
+		close(ch)
+		return ch, func() {}
+	}
 	b.subscribers[ch] = struct{}{}
 	b.mu.Unlock()
 	unsub := func() {
@@ -87,6 +95,21 @@ func (b *Bus) Subscribe(buffer int) (<-chan Event, func()) {
 	return ch, unsub
 }
 
+// Close ends every subscription by closing its channel, and any later
+// Subscribe gets a channel that is already closed. The server calls it as
+// shutdown begins: an SSE stream otherwise runs until its client leaves,
+// and holds the graceful shutdown open for its whole deadline. Publish
+// after Close is a no-op. Safe to call more than once.
+func (b *Bus) Close() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.closed = true
+	for ch := range b.subscribers {
+		delete(b.subscribers, ch)
+		close(ch)
+	}
+}
+
 // SubscriberCount returns the current number of active subscribers. For
 // tests + observability; not load-bearing in hot paths.
 func (b *Bus) SubscriberCount() int {
diff --git a/internal/eventbus/eventbus_test.go b/internal/eventbus/eventbus_test.go
index 65898b77..a5e267d3 100644
--- a/internal/eventbus/eventbus_test.go
+++ b/internal/eventbus/eventbus_test.go
@@ -148,3 +148,44 @@ func TestPublish_ConcurrentWritersAndSubscribers(t *testing.T) {
 	}
 	subWG.Wait()
 }
+
+// Close is what lets the server shut down with SSE clients connected: each
+// stream returns when its channel closes (#5293).
+func TestClose_EndsEverySubscription(t *testing.T) {
+	t.Parallel()
+	b := New()
+	ch1, unsub1 := b.Subscribe(4)
+	ch2, unsub2 := b.Subscribe(4)
+
+	b.Close()
+
+	for i, ch := range []<-chan Event{ch1, ch2} {
+		if _, ok := <-ch; ok {
+			t.Fatalf("subscriber %d: channel still open after Close", i)
+		}
+	}
+	if got := b.SubscriberCount(); got != 0 {
+		t.Fatalf("SubscriberCount=%d after Close, want 0", got)
+	}
+	// The handlers' deferred unsubscribes still run; they must not
+	// double-close.
+	unsub1()
+	unsub2()
+	b.Close()
+}
+
+func TestSubscribe_AfterCloseIsAlreadyClosed(t *testing.T) {
+	t.Parallel()
+	b := New()
+	b.Close()
+
+	ch, unsub := b.Subscribe(4)
+	defer unsub()
+	if _, ok := <-ch; ok {
+		t.Fatal("a subscription made after Close should already be closed")
+	}
+	if got := b.SubscriberCount(); got != 0 {
+		t.Fatalf("SubscriberCount=%d, want 0", got)
+	}
+	b.Publish(Event{Kind: "after.close"}) // must not panic
+}