fix: shutdown ends the live-event streams so a redeploy drains instead of waiting out its deadline (#5293)
release / govulncheck (push) Successful in 24s
release / web (push) Successful in 1m49s
release / go (push) Successful in 2m3s
release / integration (push) Successful in 6m1s
release / android (push) Successful in 6m25s
release / Build signed APK (releases and dev) (push) Successful in 6m24s
release / Attach APK to the Release (tag releases only) (push) Skipped
release / Build + push container image (push) Successful in 15s
release / Verify release artifacts (tag releases only) (push) Skipped
release / govulncheck (push) Successful in 24s
release / web (push) Successful in 1m49s
release / go (push) Successful in 2m3s
release / integration (push) Successful in 6m1s
release / android (push) Successful in 6m25s
release / Build signed APK (releases and dev) (push) Successful in 6m24s
release / Attach APK to the Release (tag releases only) (push) Skipped
release / Build + push container image (push) Successful in 15s
release / Verify release artifacts (tag releases only) (push) Skipped
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>
This commit is contained in:
@@ -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() {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user