From 1067581e3157a740f943b8413285601f5f9d1d3f Mon Sep 17 00:00:00 2001
From: Aaron Zeisler <azeisler@launchdarkly.com>
Date: Mon, 20 Jul 2026 14:23:24 -0700
Subject: [PATCH] fix: prevent stream shutdown deadlock when Close races with
 event delivery

When Close (or a restart/error) fired while the decoder goroutine had just
decoded an event and was blocked sending it on the internal events channel,
discardCurrentStream deadlocked: it drained the errs channel before events,
but the decoder was stuck on the events send and never reached the code that
closes errs. Neither side progressed, so stream.Events was never closed and a
consumer draining it on shutdown would hang forever.

Drain both internal channels concurrently. The decoder may be blocked sending
on either channel, and once unblocked it always closes both, so draining them
together lets it terminate regardless of which send it was parked on.

Add a regression test that floods the stream and closes mid-delivery, which
reliably reproduces the hang before this change.
---
 stream.go                            | 26 ++++++++----
 stream_close_during_delivery_test.go | 62 ++++++++++++++++++++++++++++
 2 files changed, 81 insertions(+), 7 deletions(-)
 create mode 100644 stream_close_during_delivery_test.go

diff --git a/stream.go b/stream.go
index 5d886e0..416e113 100644
--- a/stream.go
+++ b/stream.go
@@ -322,13 +322,25 @@ NewStream:
 			if r != nil {
 				_ = r.Close()
 				r = nil
-				// allow the decoding goroutine to terminate
-				//nolint:revive // false positive, need to drain the channels here
-				for range errs {
-				}
-				//nolint:revive // false positive, need to drain the channels here
-				for range events {
-				}
+				// Allow the decoding goroutine to terminate. After r.Close it will either return an
+				// error from Decode (sending on errs, then closing both channels) or, if it had
+				// already decoded an event, still be blocked sending that event on events. We must
+				// therefore drain both channels concurrently: draining them sequentially deadlocks
+				// if the goroutine is blocked sending on the one we are not yet reading (e.g. we
+				// wait on errs while it is stuck on events), and neither side can make progress.
+				var wg sync.WaitGroup
+				wg.Add(2)
+				go func() {
+					defer wg.Done()
+					for range errs { //nolint:revive // draining until the channel is closed
+					}
+				}()
+				go func() {
+					defer wg.Done()
+					for range events { //nolint:revive // draining until the channel is closed
+					}
+				}()
+				wg.Wait()
 			}
 		}
 
diff --git a/stream_close_during_delivery_test.go b/stream_close_during_delivery_test.go
new file mode 100644
index 0000000..d3daf76
--- /dev/null
+++ b/stream_close_during_delivery_test.go
@@ -0,0 +1,62 @@
+package eventsource
+
+import (
+	"net/http/httptest"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/launchdarkly/go-test-helpers/v3/httphelpers"
+)
+
+// TestStreamCloseDuringEventDeliveryDoesNotDeadlock is a regression test for a deadlock in the
+// stream goroutine's shutdown path (discardCurrentStream). When Close raced with an event that had
+// just been decoded, discardCurrentStream drained the internal errs channel before events while the
+// decoder goroutine was blocked sending on events, so neither side progressed and stream.Events was
+// never closed. A consumer ranging over stream.Events to drain it on shutdown would then hang
+// forever. See discardCurrentStream in stream.go.
+//
+// The race is timing-dependent, so this floods the stream and closes mid-delivery, repeated many
+// times to make the interleaving overwhelmingly likely. On regression the consumer's drain never
+// finishes and the test times out, printing all goroutines to pinpoint the block.
+func TestStreamCloseDuringEventDeliveryDoesNotDeadlock(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		func() {
+			streamHandler, streamControl := httphelpers.SSEHandler(nil)
+			defer streamControl.Close()
+			httpServer := httptest.NewServer(streamHandler)
+			defer httpServer.Close()
+
+			stream := mustSubscribe(t, httpServer.URL)
+
+			// Flood the stream so the decoder is very likely mid-delivery of an event when we Close.
+			go func() {
+				for j := 0; j < 500; j++ {
+					streamControl.Enqueue(httphelpers.SSEEvent{ID: "id", Data: "data"})
+				}
+			}()
+
+			// A real consumer drains Events (and Errors) until they close on shutdown; that only
+			// happens once the stream goroutine exits cleanly, so this is what deadlocked.
+			drained := make(chan struct{})
+			go func() {
+				for range stream.Events {
+				}
+				for range stream.Errors {
+				}
+				close(drained)
+			}()
+
+			time.Sleep(2 * time.Millisecond)
+			stream.Close()
+
+			select {
+			case <-drained:
+			case <-time.After(5 * time.Second):
+				buf := make([]byte, 1<<20)
+				n := runtime.Stack(buf, true)
+				t.Fatalf("stream.Events was never closed after Close (deadlock) on iteration %d\n%s", i, buf[:n])
+			}
+		}()
+	}
+}