fix: carry streaming credentials manager to Conn and Tx (#4155)

* fix: carry streaming credentials manager to Conn and Tx

Conn(), EphemeralConn(), and Watch build derived clients that call
initConn on cold connections. initConn uses streamingCredentialsManager
when StreamingCredentialsProvider is set; without the parent's manager
those paths nil-pointer on Listener (issue #4154).

Copy the manager the same way clone() already does for WithTimeout.

Disclosure: drafted with Grok assistance; I reviewed and tested.

* test: check EphemeralConn manager identity and drive cold initConn

Apply review suggestions on #4155: assert EphemeralConn shares the
parent's streamingCredentialsManager, and drive the cold initConn path of
Conn, EphemeralConn and Tx against an unresponsive net.Pipe so the test
panics without the fix and returns a handshake error with it.

Disclosure: drafted with AI assistance (Grok); I reviewed and ran it.

derived_streaming_creds_test.go

@@ -0,0 +1,111 @@
+package redis
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9/auth"
+	"github.com/redis/go-redis/v9/internal/pool"
+)
+
+type stubStreamingProvider struct {
+	mu          sync.Mutex
+	credentials auth.Credentials
+}
+
+func (s *stubStreamingProvider) Subscribe(listener auth.CredentialsListener) (auth.Credentials, auth.UnsubscribeFunc, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.credentials, func() error { return nil }, nil
+}
+
+func TestDerivedClientsCarryStreamingCredentialsManager(t *testing.T) {
+	provider := &stubStreamingProvider{
+		credentials: auth.NewBasicCredentials("default", "token"),
+	}
+	c := NewClient(&Options{
+		Addr:                         "127.0.0.1:1",
+		PoolSize:                     1,
+		StreamingCredentialsProvider: provider,
+	})
+	t.Cleanup(func() { _ = c.Close() })
+
+	if c.streamingCredentialsManager == nil {
+		t.Fatal("expected client to create a streamingCredentialsManager")
+	}
+
+	conn := c.Conn()
+	t.Cleanup(func() { _ = conn.Close() })
+	if conn.streamingCredentialsManager == nil {
+		t.Fatal("Conn() did not inherit streamingCredentialsManager; cold initConn would panic (#4154)")
+	}
+	if conn.streamingCredentialsManager != c.streamingCredentialsManager {
+		t.Fatal("Conn() should share the parent's streamingCredentialsManager")
+	}
+
+	eph := c.EphemeralConn()
+	t.Cleanup(func() { _ = eph.Close() })
+	if eph.streamingCredentialsManager == nil {
+		t.Fatal("EphemeralConn() did not inherit streamingCredentialsManager (#4154)")
+	}
+	if eph.streamingCredentialsManager != c.streamingCredentialsManager {
+		t.Fatal("EphemeralConn() did not inherit the parent's streamingCredentialsManager (#4154)")
+	}
+
+	tx := c.newTx()
+	t.Cleanup(func() { _ = tx.Close(context.Background()) })
+	if tx.streamingCredentialsManager == nil {
+		t.Fatal("newTx()/Watch did not inherit streamingCredentialsManager (#4154)")
+	}
+	if tx.streamingCredentialsManager != c.streamingCredentialsManager {
+		t.Fatal("newTx() should share the parent's streamingCredentialsManager")
+	}
+}
+
+// TestDerivedClientsColdInitConnDoesNotPanic drives the cold initConn path
+// of each derived client against an unresponsive net.Pipe. Before #4154 was
+// fixed this panicked with a nil streamingCredentialsManager inside
+// Listener; now it gets past Listener and fails the handshake on timeout.
+func TestDerivedClientsColdInitConnDoesNotPanic(t *testing.T) {
+	provider := &stubStreamingProvider{
+		credentials: auth.NewBasicCredentials("default", "token"),
+	}
+	c := NewClient(&Options{
+		Addr:                         "127.0.0.1:1",
+		PoolSize:                     1,
+		StreamingCredentialsProvider: provider,
+		ReadTimeout:                  50 * time.Millisecond,
+		WriteTimeout:                 50 * time.Millisecond,
+	})
+	t.Cleanup(func() { _ = c.Close() })
+
+	conn := c.Conn()
+	t.Cleanup(func() { _ = conn.Close() })
+	eph := c.EphemeralConn()
+	t.Cleanup(func() { _ = eph.Close() })
+	tx := c.newTx()
+	t.Cleanup(func() { _ = tx.Close(context.Background()) })
+
+	for _, tc := range []struct {
+		name string
+		base *baseClient
+	}{
+		{name: "Conn", base: &conn.baseClient},
+		{name: "EphemeralConn", base: &eph.baseClient},
+		{name: "Tx", base: &tx.baseClient},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			clientSide, serverSide := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientSide.Close()
+				_ = serverSide.Close()
+			})
+			if err := tc.base.initConn(context.Background(), pool.NewConn(clientSide)); err == nil {
+				t.Fatal("expected a handshake error from the unresponsive pipe")
+			}
+		})
+	}
+}

redis.go

@@ -2726,6 +2726,10 @@ func (c *Client) conn(discardOnClose bool) *Conn {
 	// initializes a pool conn, the close hook it installs still evicts from the
 	// parent cache (its own csc is nil).
 	conn.baseClient.cscPoolHook = c.baseClient.cscPoolHook
+	// Carry the parent's streaming-credentials manager so initConn can create a
+	// credentials listener when a cold Conn/EphemeralConn initializes a fresh
+	// pool connection (issue #4154).
+	conn.baseClient.streamingCredentialsManager = c.streamingCredentialsManager
 	return conn
 }
 

tx.go

@@ -57,6 +57,10 @@ func (c *Client) newTx() *Tx {
 			// evict from the parent cache.
 			cscPoolHook: c.cscPoolHook,
 			cscActive:   c.cscActive,
+			// Carry the parent's streaming-credentials manager so initConn can
+			// create a credentials listener when Watch initializes a fresh
+			// connection (issue #4154).
+			streamingCredentialsManager: c.streamingCredentialsManager,
 		},
 	}
 	tx.init()