only async reauth

internal/auth/streaming/conn_reauth_credentials_listener.go

@@ -35,25 +35,11 @@ func (c *ConnReAuthCredentialsListener) OnNext(credentials auth.Credentials) {
 		return
 	}
 
-	// this connection is not in use, so we can re-authenticate it
+	// Always use async reauth to avoid complex pool semaphore issues
-	if c.conn.Used.CompareAndSwap(false, true) {
+	// The synchronous path can cause deadlocks in the pool's semaphore mechanism
-		// try to acquire the connection for background operation
+	// when called from the Subscribe goroutine, especially with small pool sizes.
-		if c.conn.Usable.CompareAndSwap(true, false) {
+	// The connection pool hook will re-authenticate the connection when it is
-			err := c.reAuth(c.conn, credentials)
+	// returned to the pool in a clean, idle state.
-			if err != nil {
-				c.OnError(err)
-			}
-			c.conn.Usable.Store(true)
-			c.conn.Used.Store(false)
-			return
-		}
-		c.conn.Used.Store(false)
-	}
-	// else if the connection is in use, mark it for re-authentication
-	// and connection pool hook will re-authenticate it when it is returned to the pool
-	// or in case the connection WAS in the pool, but handoff is in progress, the pool hook
-	// will re-authenticate it when the handoff is complete
-	// and the connection is acquired from the pool
 	c.manager.MarkForReAuth(c.conn, func(err error) {
 		if err != nil {
 			c.OnError(err)

internal/auth/streaming/pool_hook.go

@@ -92,8 +92,11 @@ func (r *ReAuthPoolHook) OnPut(_ context.Context, conn *pool.Conn) (bool, bool,
 			var err error
 			timeout := time.After(r.reAuthTimeout)
 
-			// Try to acquire the connection (set Usable to false)
+			// Try to acquire the connection
-			for !conn.Usable.CompareAndSwap(true, false) {
+			// We need to ensure the connection is both Usable and not Used
+			// to prevent data races with concurrent operations
+			acquired := false
+			for !acquired {
 				select {
 				case <-timeout:
 					// Timeout occurred, cannot acquire connection
@@ -101,7 +104,12 @@ func (r *ReAuthPoolHook) OnPut(_ context.Context, conn *pool.Conn) (bool, bool,
 					reAuthFn(err)
 					return
 				default:
-					time.Sleep(time.Millisecond)
+					// Try to acquire: set Usable=false only if Used=false
+					if !conn.Used.Load() && conn.Usable.CompareAndSwap(true, false) {
+						acquired = true
+					} else {
+						time.Sleep(time.Millisecond)
+					}
 				}
 			}
 

pubsub.go

@@ -465,7 +465,6 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
 	}
 
 	// Don't hold the lock to allow subscriptions and pings.
-
 	cn, err := c.connWithLock(ctx)
 	if err != nil {
 		return nil, err

redis.go

@@ -305,6 +305,8 @@ func (c *baseClient) reAuthConnection() func(poolCn *pool.Conn, credentials auth
 	return func(poolCn *pool.Conn, credentials auth.Credentials) error {
 		var err error
 		username, password := credentials.BasicAuth()
+
+		// Use background context - timeout is handled by ReadTimeout in WithReader/WithWriter
 		ctx := context.Background()
 
 		connPool := pool.NewSingleConnPool(c.connPool, poolCn)