fix: prevent stream shutdown deadlock when Close races with event delivery

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()
 			}
 		}
 

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])
+			}
+		}()
+	}
+}