streaming: prevent reader Close from deadlocking on a fatal read error (#78)

* streaming: prevent reader Close from deadlocking on fatal read error

On a fatal read error (e.g. the underlying stream key being destroyed)
the reader read loop called Close synchronously. Close waits on the read
goroutine via wait.Wait, so calling it from that same goroutine deadlocked
and leaked the reader and its Redis connection; any external Close blocked
forever too.

Trigger the shutdown asynchronously so the read goroutine can return and
release the wait group. Close is already idempotent via sync.Once, so this
is safe alongside a concurrent external Close.

Adds a regression test that drives the real read loop through a simulated
fatal error (via a small xreadFn seam) and asserts the reader closes
instead of hanging. The test hangs on pre-fix code.

* streaming: make reader read seam a package var

Move the test-only read hook from a Reader struct field to a package
variable so it no longer pollutes Reader. Behavior is unchanged; it
defaults to (*Reader).xread and is only overridden in tests.

Author: raphael

streaming/reader.go

@@ -224,19 +224,28 @@ func (r *Reader) start() {
 	})
 }
 
+// xreadFn fetches the next batch of events for a reader. It is a package
+// variable (rather than a struct field) so tests can simulate read errors
+// without polluting Reader; it defaults to (*Reader).xread.
+var xreadFn = (*Reader).xread
+
 // read reads events from the streams and sends them to the reader channel.
 func (r *Reader) read() {
 	ctx := context.Background()
 	defer r.cleanup()
 	for {
-		streamsEvents, err := r.xread(ctx)
+		streamsEvents, err := xreadFn(r, ctx)
 		if r.isClosing() {
 			return
 		}
 		if err != nil {
 			if err := handleReadError(err, r.logger); err != nil {
 				r.logger.Error(fmt.Errorf("fatal error while reading events: %w, stopping", err))
-				r.Close()
+				// Close waits on this goroutine via wait.Wait, so calling it
+				// synchronously here would deadlock and leak the reader and its
+				// Redis connection. Trigger the shutdown asynchronously and let
+				// this goroutine return so cleanup can release the wait group.
+				pulse.Go(r.logger, r.Close)
 				return
 			}
 			continue

streaming/reader_test.go

@@ -1,6 +1,8 @@
 package streaming
 
 import (
+	"context"
+	"fmt"
 	"strconv"
 	"strings"
 	"testing"
@@ -124,6 +126,31 @@ func TestCleanupReader(t *testing.T) {
 	assert.Eventually(t, func() bool { return rdb.Exists(ctx, s.key).Val() == 0 }, max, delay)
 }
 
+func TestReaderCloseOnFatalReadError(t *testing.T) {
+	testName := strings.Replace(t.Name(), "/", "_", -1)
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, true, testName)
+	ctx := ptesting.NewTestContext(t)
+	s, err := NewStream(testName, rdb, options.WithStreamLogger(pulse.ClueLogger(ctx)))
+	require.NoError(t, err)
+	// Simulate a fatal read error (e.g. the underlying stream key being
+	// destroyed) before the read goroutine starts. The read loop reacts to a
+	// fatal error by closing the reader; because Close waits on the read
+	// goroutine, it must run asynchronously or it would deadlock and leak the
+	// reader and its Redis connection.
+	defer func(orig func(*Reader, context.Context) ([]redis.XStream, error)) { xreadFn = orig }(xreadFn)
+	xreadFn = func(*Reader, context.Context) ([]redis.XStream, error) {
+		return nil, fmt.Errorf("stream key no longer exists")
+	}
+
+	reader, err := s.NewReader(ctx, options.WithReaderBlockDuration(testBlockDuration))
+	require.NoError(t, err)
+	reader.Subscribe()
+
+	require.Eventually(t, func() bool { return reader.IsClosed() }, max, delay,
+		"reader did not close after a fatal read error (Close likely deadlocked on its own read goroutine)")
+}
+
 func TestAddReaderStream(t *testing.T) {
 	testName := strings.Replace(t.Name(), "/", "_", -1)
 	rdb := ptesting.NewRedisClient(t)