fix(sentinel): set up streaming creds on failover (#4182)

* fix(sentinel): set up streaming creds on failover

* docs: add release note for failover streaming creds

* fix(pool): run OnRemove hooks in Filter

* test(sentinel): check failover pipeline pool hook

RELEASE-NOTES.md

@@ -55,6 +55,7 @@
   the cursor. Cursor ownership and reply-parsing changes are deferred to a
   separate PR.
 - **Per-node TLS verification for cluster URLs:** `ParseClusterURL` accepts `tls_verify_node_host=true` on `rediss://` URLs. The parameter clears `TLSConfig.ServerName`, so each node is verified against the host it is dialed at and not against the host of the first URL. Use it when each node has a certificate for its own host. The default is unchanged: every node is verified against the host of the first URL, so clusters that use one certificate and announce their nodes by IP continue to work. `rediss://` cluster URLs now also set `MinVersion: tls.VersionTLS12`, as `ParseURL` and `ParseFailoverURL` do ([#4145](https://github.com/redis/go-redis/pull/4145)).
+- **Failover client with `StreamingCredentialsProvider`:** `NewFailoverClient` now creates the streaming credentials manager, so new connections no longer panic in `initConn` ([#4182](https://github.com/redis/go-redis/pull/4182)).
 
 # 9.23.0 (2026-10-05)
 

derived_streaming_creds_test.go

@@ -109,3 +109,47 @@ func TestDerivedClientsColdInitConnDoesNotPanic(t *testing.T) {
 		})
 	}
 }
+
+func TestFailoverClientStreamingCredentials(t *testing.T) {
+	provider := &stubStreamingProvider{
+		credentials: auth.NewBasicCredentials("default", "token"),
+	}
+	c := NewFailoverClient(&FailoverOptions{
+		MasterName:                   "mymaster",
+		SentinelAddrs:                []string{"127.0.0.1:1"},
+		PoolSize:                     1,
+		StreamingCredentialsProvider: provider,
+		ReadTimeout:                  50 * time.Millisecond,
+		WriteTimeout:                 50 * time.Millisecond,
+	})
+	t.Cleanup(func() { _ = c.Close() })
+
+	if c.streamingCredentialsManager == nil {
+		t.Fatal("expected failover client to create a streamingCredentialsManager")
+	}
+
+	clientSide, serverSide := net.Pipe()
+	t.Cleanup(func() {
+		_ = clientSide.Close()
+		_ = serverSide.Close()
+	})
+	if err := c.initConn(context.Background(), pool.NewConn(clientSide)); err == nil {
+		t.Fatal("expected a handshake error from the unresponsive pipe")
+	}
+
+	// The pipeline pool must exist and carry the re-auth hook: removing a
+	// conn from it should drop that conn's listener from the manager.
+	pp := c.getPipelinePool()
+	if pp == nil {
+		t.Fatal("expected failover client to build a pipeline pool")
+	}
+	pipeClient, pipeServer := net.Pipe()
+	t.Cleanup(func() { _ = pipeServer.Close() })
+	cn := pool.NewConn(pipeClient)
+	before, _ := c.streamingCredentialsManager.Listener(cn, nil, nil)
+	pp.RemoveWithoutTurn(context.Background(), cn, nil)
+	after, _ := c.streamingCredentialsManager.Listener(cn, nil, nil)
+	if before == after {
+		t.Fatal("pipeline pool has no streaming credentials hook; initStreamingCredentials ran before the pipeline pool was built")
+	}
+}

internal/pool/hooks_test.go

@@ -355,3 +355,41 @@ func TestPutConnStateClosedByHookInvokesOnRemove(t *testing.T) {
 		t.Errorf("Expected OnRemove to fire once for StateClosed-on-Put, got %d", hook.onRemoveCalled)
 	}
 }
+
+// TestFilterInvokesOnRemove checks Filter runs OnRemove for the idle conns it
+// drops, so per-connection hook state (e.g. streaming credentials listeners)
+// does not leak on every Sentinel failover.
+func TestFilterInvokesOnRemove(t *testing.T) {
+	opt := &Options{
+		Dialer: func(ctx context.Context) (net.Conn, error) {
+			c, _ := net.Pipe()
+			return c, nil
+		},
+		PoolSize:           1,
+		MaxConcurrentDials: 1,
+		DialTimeout:        time.Second,
+		PoolTimeout:        time.Second,
+	}
+
+	p := NewConnPool(opt)
+	defer p.Close()
+
+	hook := &TestHook{ShouldPool: true, ShouldAccept: true}
+	p.AddPoolHook(hook)
+
+	ctx := context.Background()
+	cn, err := p.Get(ctx)
+	if err != nil {
+		t.Fatalf("Get failed: %v", err)
+	}
+	p.Put(ctx, cn)
+
+	_ = p.Filter(func(*Conn) bool { return true })
+
+	if hook.OnRemoveCalled != 1 {
+		t.Errorf("Expected OnRemove to fire once for Filter, got %d", hook.OnRemoveCalled)
+	}
+	if n := p.Len(); n != 0 {
+		t.Errorf("Expected Len=0 after Filter, got %d", n)
+	}
+}

internal/pool/pool.go

@@ -2050,7 +2050,6 @@ func (p *ConnPool) Filter(fn func(*Conn) bool) error {
 	ctx := context.Background()
 
 	p.connsMu.Lock()
-	defer p.connsMu.Unlock()
 
 	idleConnSet := make(map[*Conn]struct{}, len(p.idleConns))
 	for _, ic := range p.idleConns {
@@ -2058,26 +2057,40 @@ func (p *ConnPool) Filter(fn func(*Conn) bool) error {
 	}
 
 	var firstErr error
+	var toClose []*Conn
 	for _, cn := range p.conns {
 		if fn(cn) {
-			var err error
 			if _, isIdle := idleConnSet[cn]; isIdle {
-				// Idle connection - remove from pool and close.
+				// Idle connection - remove from pool, close it below.
 				p.removeConn(cn)
-				p.recordConnectionMetrics(ctx, cn, CloseReasonFailover, MetricStateIdle, getMetricConnectionCountCallback())
-				err = p.closeConn(cn)
+				toClose = append(toClose, cn)
 			} else {
 				// Used connection - set closeReason and close the connection.
 				// The connection remains in p.conns. When putConn() is called later,
 				// it will close the connection instead of pooling it.
 				cn.closeReason.Store(CloseReasonFailover)
-				err = cn.Close()
-			}
-			if err != nil && firstErr == nil {
-				firstErr = err
+				if err := cn.Close(); err != nil && firstErr == nil {
+					firstErr = err
+				}
 			}
 		}
 	}
+	countCb := getMetricConnectionCountCallback()
+	p.connsMu.Unlock()
+
+	// Run OnRemove so hooks drop per-connection state (e.g. streaming
+	// credentials listeners) for the idle conns removed here.
+	if hookManager := p.hookManager.Load(); hookManager != nil {
+		for _, cn := range toClose {
+			hookManager.ProcessOnRemove(ctx, cn, errors.New(CloseReasonFailover))
+		}
+	}
+	for _, cn := range toClose {
+		p.recordConnectionMetrics(ctx, cn, CloseReasonFailover, MetricStateIdle, countCb)
+		if err := p.closeConn(cn); err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
 	return firstErr
 }
 

redis.go

@@ -827,6 +827,31 @@ func (c *baseClient) buildPipelinePool(poolName string) (*pipelinePoolRef, error
 	return &pipelinePoolRef{pool: p, name: poolName}, nil
 }
 
+// initStreamingCredentials creates the streaming credentials manager and
+// registers its pool hook on the main and pipeline pools. Shared by the
+// NewClient and NewFailoverClient creation paths so they cannot drift; must
+// run after the pipeline pool is built.
+func (c *baseClient) initStreamingCredentials() {
+	if c.opt.StreamingCredentialsProvider == nil {
+		return
+	}
+	// Size the re-auth worker semaphore for the COMBINED ceiling of every pool
+	// the hook is registered on: the same hook drives the main pool AND the
+	// (possibly larger) dedicated pipeline pool, so a credential rotation must
+	// be able to re-AUTH connections from both concurrently. Sizing to only the
+	// main PoolSize would serialize pipeline re-auths behind PoolSize workers,
+	// leaving pipeline capacity unavailable well past reAuthTimeout.
+	workers := c.connPool.Size()
+	if pp := c.getPipelinePool(); pp != nil {
+		workers += pp.Size()
+	}
+	c.streamingCredentialsManager = streaming.NewManagerWithWorkers(c.connPool, c.opt.PoolTimeout, workers)
+	c.connPool.AddPoolHook(c.streamingCredentialsManager.PoolHook())
+	if pp := c.getPipelinePool(); pp != nil {
+		pp.AddPoolHook(c.streamingCredentialsManager.PoolHook())
+	}
+}
+
 func (c *baseClient) reAuthConnection() func(poolCn *pool.Conn, credentials auth.Credentials) error {
 	return func(poolCn *pool.Conn, credentials auth.Credentials) error {
 		var err error
@@ -2551,23 +2576,7 @@ func NewClient(opt *Options) *Client {
 		c.pipelinePool = ref
 	}
 
-	if opt.StreamingCredentialsProvider != nil {
-		// Size the re-auth worker semaphore for the COMBINED ceiling of every pool
-		// the hook is registered on: the same hook drives the main pool AND the
-		// (possibly larger) dedicated pipeline pool, so a credential rotation must
-		// be able to re-AUTH connections from both concurrently. Sizing to only the
-		// main PoolSize would serialize pipeline re-auths behind PoolSize workers,
-		// leaving pipeline capacity unavailable well past reAuthTimeout.
-		workers := c.connPool.Size()
-		if pp := c.getPipelinePool(); pp != nil {
-			workers += pp.Size()
-		}
-		c.streamingCredentialsManager = streaming.NewManagerWithWorkers(c.connPool, c.opt.PoolTimeout, workers)
-		c.connPool.AddPoolHook(c.streamingCredentialsManager.PoolHook())
-		if pp := c.getPipelinePool(); pp != nil {
-			pp.AddPoolHook(c.streamingCredentialsManager.PoolHook())
-		}
-	}
+	c.initStreamingCredentials()
 
 	// Create metadata independently of CSC. The zero config reuses the package
 	// default without allocating a store.

sentinel.go

@@ -708,6 +708,8 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
 		rdb.pipelinePool = ref
 	}
 
+	rdb.initStreamingCredentials()
+
 	// Register pools for OTel async gauge metrics, matching NewClient (the
 	// failover client previously registered none, so pool gauges were silent
 	// for the identical standalone setup). The pipeline pool is nil when not