Merge branch 'master' into feat/8.8-xnack

Author: elena-kolevska

auth/auth.go

@@ -9,12 +9,30 @@ type StreamingCredentialsProvider interface {
 	// Subscribe subscribes to the credentials provider for updates.
 	// It returns the current credentials, a cancel function to unsubscribe from the provider,
 	// and an error if any.
+	//
+	// Implementations MUST be idempotent with respect to listener identity:
+	// subscribing the same listener value more than once must not produce
+	// duplicate notifications and must not create multiple independent
+	// subscriptions that each need to be cancelled separately. Every
+	// UnsubscribeFunc returned for a given listener must cancel that
+	// listener's subscription; calling any one of them must be sufficient to
+	// stop updates to that listener, and calling subsequent ones must be a
+	// safe no-op. Callers (including go-redis internals) may retain only
+	// the most recently returned UnsubscribeFunc and rely on it to fully
+	// unsubscribe the listener.
+	//
 	// TODO(ndyakov): Should we add context to the Subscribe method?
 	Subscribe(listener CredentialsListener) (Credentials, UnsubscribeFunc, error)
 }
 
 // UnsubscribeFunc is a function that is used to cancel the subscription to the credentials provider.
 // It is used to unsubscribe from the provider when the credentials are no longer needed.
+//
+// Per the StreamingCredentialsProvider.Subscribe contract, if the same
+// listener is subscribed multiple times, every UnsubscribeFunc returned for
+// that listener must fully unsubscribe it on first invocation, and
+// subsequent invocations (from any of the equivalent UnsubscribeFuncs) must
+// be a safe no-op.
 type UnsubscribeFunc func() error
 
 // CredentialsListener is an interface that defines the methods for a credentials listener.

error.go

@@ -105,6 +105,12 @@ func shouldRetry(err error, retryTimeout bool) bool {
 	// Check for timeout errors (works with wrapped errors)
 	if isTimeout, hasTimeoutFlag := isTimeoutError(err); isTimeout {
 		if hasTimeoutFlag {
+			// A dial error means the TCP connection was never established and the
+			// command was never sent to the server, so retry is always safe
+			var opErr *net.OpError
+			if errors.As(err, &opErr) && opErr.Op == "dial" {
+				return true
+			}
 			return retryTimeout
 		}
 		return true

error_test.go

@@ -3,9 +3,11 @@ package redis_test
 import (
 	"context"
 	"io"
+	"net"
 
 	. "github.com/bsm/ginkgo/v2"
 	. "github.com/bsm/gomega"
+
 	"github.com/redis/go-redis/v9"
 	"github.com/redis/go-redis/v9/internal/proto"
 )
@@ -24,11 +26,9 @@ func (t testTimeout) Error() string {
 
 var _ = Describe("error", func() {
 	BeforeEach(func() {
-
 	})
 
 	AfterEach(func() {
-
 	})
 
 	It("should retry", func() {
@@ -39,15 +39,16 @@ var _ = Describe("error", func() {
 			context.Canceled:         false,
 			context.DeadlineExceeded: false,
 			redis.ErrPoolTimeout:     true,
+			&net.OpError{Op: "dial"}: true,
 			// Use typed errors instead of plain errors.New()
-			proto.ParseErrorReply([]byte("-ERR max number of clients reached")):                      true,
-			proto.ParseErrorReply([]byte("-LOADING Redis is loading the dataset in memory")):         true,
-			proto.ParseErrorReply([]byte("-READONLY You can't write against a read only replica")):                                                                                      true,
+			proto.ParseErrorReply([]byte("-ERR max number of clients reached")):                                                                                   true,
+			proto.ParseErrorReply([]byte("-LOADING Redis is loading the dataset in memory")):                                                                      true,
+			proto.ParseErrorReply([]byte("-READONLY You can't write against a read only replica")):                                                                true,
 			proto.ParseErrorReply([]byte("-ERR Error running script (call to f_abc123): @user_script:1: -READONLY You can't write against a read only replica.")): true,
-			proto.ParseErrorReply([]byte("-CLUSTERDOWN The cluster is down")):                        true,
-			proto.ParseErrorReply([]byte("-TRYAGAIN Command cannot be processed, please try again")): true,
-			proto.ParseErrorReply([]byte("-NOREPLICAS Not enough good replicas to write")):           true,
-			proto.ParseErrorReply([]byte("-ERR other")):                                              false,
+			proto.ParseErrorReply([]byte("-CLUSTERDOWN The cluster is down")):                                                                                     true,
+			proto.ParseErrorReply([]byte("-TRYAGAIN Command cannot be processed, please try again")):                                                              true,
+			proto.ParseErrorReply([]byte("-NOREPLICAS Not enough good replicas to write")):                                                                        true,
+			proto.ParseErrorReply([]byte("-ERR other")):                                                                                                           false,
 		}
 
 		for err, expected := range data {

internal/hscan/hscan_test.go

@@ -48,6 +48,19 @@ type TimeData struct {
 	Time *TimeRFC3339Nano `redis:"login"`
 }
 
+type binaryUnmarshaler struct {
+	s string
+}
+
+func (b *binaryUnmarshaler) UnmarshalBinary(data []byte) error {
+	b.s = string(data)
+	return nil
+}
+
+type BinaryData struct {
+	Binary *binaryUnmarshaler `redis:"binary"`
+}
+
 type i []interface{}
 
 func TestGinkgoSuite(t *testing.T) {
@@ -217,4 +230,9 @@ var _ = Describe("Scan", func() {
 		Expect(Scan(&tt, i{"time"}, i{now.Format(time.RFC3339Nano)})).NotTo(HaveOccurred())
 		Expect(now.Unix()).To(Equal(tt.Time.Unix()))
 	})
+	It("Implements BinaryUnmarshaler", func() {
+		var bd BinaryData
+		Expect(Scan(&bd, i{"binary"}, i{"hello"})).NotTo(HaveOccurred())
+		Expect(bd.Binary.s).To(Equal("hello"))
+	})
 })

internal/hscan/structmap.go

@@ -109,6 +109,8 @@ func (s StructValue) Scan(key string, value string) error {
 			return scan.ScanRedis(value)
 		case encoding.TextUnmarshaler:
 			return scan.UnmarshalText(util.StringToBytes(value))
+		case encoding.BinaryUnmarshaler:
+			return scan.UnmarshalBinary(util.StringToBytes(value))
 		}
 	}
 

internal/pool/conn.go

@@ -102,7 +102,6 @@ type Conn struct {
 
 	pooled    bool
 	pubsub    bool
-	closed    atomic.Bool
 	createdAt time.Time
 	expiresAt time.Time
 	poolName  string // Name of the pool this connection belongs to (for metrics)
@@ -576,6 +575,41 @@ func (cn *Conn) getEffectiveWriteTimeout(normalTimeout time.Duration) time.Durat
 	}
 }
 
+// SetOnClose installs fn as the callback invoked exactly once when this
+// connection is closed (via Conn.Close).
+//
+// IMPORTANT: SetOnClose OVERWRITES any previously installed callback — it
+// does not compose, chain, or deduplicate. A Conn has room for a single
+// onClose hook by design, because its lifecycle is bounded (a Conn is
+// created, optionally re-initialized on its own net.Conn, and then closed
+// once) and the pool's OnRemove hooks handle any registry-level cleanup
+// that must survive the net.Conn being swapped.
+//
+// This has a subtle implication for per-connection subscriptions such as
+// the unsubscribe function returned by StreamingCredentialsProvider
+// (e.g. EntraID token rotation): if SetOnClose is called twice on the
+// same Conn with DIFFERENT unsubscribe closures — for example because
+// initConn ran a second time and obtained a fresh Subscribe() —
+// the previous unsubscribe is dropped and will NEVER run, leaking a
+// subscription on the provider. Callers must therefore ensure either:
+//
+//   - the provider's Subscribe is idempotent for the same listener (the
+//     streaming credentials Manager deduplicates listeners by connection
+//     id, so re-Subscribe returns an equivalent unsubscribe), OR
+//   - the previous callback has already been invoked before SetOnClose is
+//     called again.
+//
+// Design note: unlike the client-level onCloseHooks registry (see
+// redis.baseClient), there is intentionally NO named-hook dedup or
+// multi-callback support on Conn. This is a deliberate trade-off to keep
+// the Conn object slim — a pool can hold thousands of Conn values and
+// each one is a hot allocation, so paying for a sync.Mutex plus a
+// map[string]func() error per connection to support a feature that would
+// only be used by at most one subsystem today (streaming credentials) is
+// not worth the per-connection memory and allocation cost. For a single
+// Conn there is at most one meaningful close callback at any point in
+// time, and a richer registry here would not even solve the "stale
+// closure" hazard described above.
 func (cn *Conn) SetOnClose(fn func() error) {
 	cn.onClose = fn
 }
@@ -882,12 +916,10 @@ func (cn *Conn) WithWriter(
 }
 
 func (cn *Conn) IsClosed() bool {
-	return cn.closed.Load() || cn.stateMachine.GetState() == StateClosed
+	return cn.stateMachine.GetState() == StateClosed
 }
 
 func (cn *Conn) Close() error {
-	cn.closed.Store(true)
-
 	// Transition to CLOSED state
 	cn.stateMachine.Transition(StateClosed)
 

internal/pool/main_test.go

@@ -121,3 +121,22 @@ func (d *dummyRawConn) Close() {
 	d.closed = true
 	d.mu.Unlock()
 }
+
+// failCloseConn wraps a dummyConn but returns a configured error on Close().
+// This is used to simulate TLS closeNotify errors or network timeouts during connection close.
+type failCloseConn struct {
+	*dummyConn
+	closeErr error
+}
+
+func (f *failCloseConn) Close() error {
+	// Close the underlying raw conn so that connCheck sees a closed fd,
+	// matching real-world behavior (e.g., TLS closeNotify failure after
+	// the OS-level socket is torn down).
+	f.dummyConn.rawConn.Close()
+	return f.closeErr
+}
+
+func (f *failCloseConn) SyscallConn() (syscall.RawConn, error) {
+	return f.dummyConn.SyscallConn()
+}

internal/pool/pool.go

@@ -1572,6 +1572,7 @@ func (p *ConnPool) Close() error {
 	}
 
 	var firstErr error
+	nowNs := time.Now().UnixNano()
 	p.connsMu.Lock()
 
 	// Emit -1 for each connection. Since all idle↔used transitions happen
@@ -1583,6 +1584,10 @@ func (p *ConnPool) Close() error {
 	}
 	ctx := context.Background()
 	for _, cn := range p.conns {
+		// Check health before closing, since closeConn invalidates the
+		// underlying fd and would make connCheck (inside isHealthyConn)
+		// always fail with EBADF.
+		healthy := p.isHealthyConn(cn, nowNs)
 		if cb != nil {
 			if _, isIdle := idleSet[cn.GetID()]; isIdle {
 				cb(ctx, -1, cn, "idle", false)
@@ -1594,7 +1599,11 @@ func (p *ConnPool) Close() error {
 			closedCb(ctx, cn, "pool_shutdown", nil)
 		}
 		if err := p.closeConn(cn); err != nil && firstErr == nil {
-			firstErr = err
+			// Suppress close errors for stale connections, consistent
+			// with how Get() handles them (see CloseReasonStale path).
+			if healthy {
+				firstErr = err
+			}
 		}
 	}
 	p.conns = nil

internal/pool/pool_test.go

@@ -1397,6 +1397,65 @@ var _ = Describe("queuedNewConn", func() {
 			"dialsQueue should be empty after all requests complete")
 	})
 
+	Describe("Close suppresses errors for stale connections", func() {
+		It("should not return close error when connection is stale due to ConnMaxIdleTime", func() {
+			closeErr := fmt.Errorf("tls: failed to send closeNotify alert (but connection was closed anyway): write tcp: connection timed out")
+
+			testPool := pool.NewConnPool(&pool.Options{
+				Dialer: func(ctx context.Context) (net.Conn, error) {
+					return &failCloseConn{
+						dummyConn: &dummyConn{rawConn: new(dummyRawConn)},
+						closeErr:  closeErr,
+					}, nil
+				},
+				PoolSize:           1,
+				MaxConcurrentDials: 1,
+				DialTimeout:        1 * time.Second,
+				PoolTimeout:        1 * time.Second,
+				ConnMaxIdleTime:    50 * time.Millisecond,
+			})
+
+			// Get a connection, put it back, then wait for it to become stale
+			cn, err := testPool.Get(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			testPool.Put(ctx, cn)
+
+			// Wait for the connection to exceed ConnMaxIdleTime
+			time.Sleep(100 * time.Millisecond)
+
+			// Close the pool — stale connection close errors should be suppressed
+			err = testPool.Close()
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("should return close error when connection is still healthy", func() {
+			closeErr := fmt.Errorf("tls: failed to send closeNotify alert")
+
+			testPool := pool.NewConnPool(&pool.Options{
+				Dialer: func(ctx context.Context) (net.Conn, error) {
+					return &failCloseConn{
+						dummyConn: &dummyConn{rawConn: new(dummyRawConn)},
+						closeErr:  closeErr,
+					}, nil
+				},
+				PoolSize:           1,
+				MaxConcurrentDials: 1,
+				DialTimeout:        1 * time.Second,
+				PoolTimeout:        1 * time.Second,
+				ConnMaxIdleTime:    10 * time.Minute, // Long idle time — connection stays healthy
+			})
+
+			// Get a connection and put it back
+			cn, err := testPool.Get(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			testPool.Put(ctx, cn)
+
+			// Close the pool — healthy connection close error should be returned
+			err = testPool.Close()
+			Expect(err).To(MatchError(closeErr))
+		})
+	})
+
 	Describe("calcConnExpiresAt", func() {
 		// Case 1: lifetime <= 0 returns noExpiration
 		It("returns noExpiration when ConnMaxLifetime is not positive", func() {