still hunting this deadlock

Author: ndyakov

hitless/pool_hook.go

@@ -388,24 +388,29 @@ func (ph *PoolHook) queueHandoff(conn *pool.Conn) error {
 	}
 
 	select {
-	case ph.handoffQueue <- request:
-		// Store in pending map
-		ph.pending.Store(request.ConnID, request.SeqID)
-		return nil
+	// priority to shutdown
 	case <-ph.shutdown:
-		ph.pending.Delete(request.ConnID)
 		return errors.New("shutdown")
 	default:
-		// Queue is full - log and attempt scaling
-		if ph.config != nil && ph.config.LogLevel >= 1 { // Warning level
-			internal.Logger.Printf(context.Background(),
-				"hitless: handoff queue is full (%d/%d), attempting timeout queuing and scaling workers",
-				len(ph.handoffQueue), cap(ph.handoffQueue))
+		select {
+		case <-ph.shutdown:
+			return errors.New("shutdown")
+		case ph.handoffQueue <- request:
+			// Store in pending map
+			ph.pending.Store(request.ConnID, request.SeqID)
+			return nil
+		default:
+			// Queue is full - log and attempt scaling
+			if ph.config != nil && ph.config.LogLevel >= 1 { // Warning level
+				internal.Logger.Printf(context.Background(),
+					"hitless: handoff queue is full (%d/%d), attempting timeout queuing and scaling workers",
+					len(ph.handoffQueue), cap(ph.handoffQueue))
+			}
 		}
-
-		// Scale up workers to handle the load
-		go ph.scaleUpWorkers()
 	}
+
+	// Scale up workers to handle the load
+	go ph.scaleUpWorkers()
 	return errors.New("queue full")
 }
 

redis.go

@@ -932,18 +932,14 @@ func NewClient(opt *Options) *Client {
 	// Create connection pool with the hooks (nil if hitless upgrades disabled)
 	c.connPool = newConnPool(opt, c.dialHook)
 
-	// Initialize hitless upgrades first if enabled to get the connection processor
-	if opt.HitlessUpgradeConfig.IsEnabled() {
-		if opt.Protocol != 3 {
-			internal.Logger.Printf(context.Background(), "hitless: RESP3 protocol required for hitless upgrades, but Protocol is %d", opt.Protocol)
-		} else {
-			err := c.enableHitlessUpgrades()
-			if err != nil {
-				internal.Logger.Printf(context.Background(), "hitless: failed to initialize hitless upgrades: %v", err)
-				if opt.HitlessUpgradeConfig.Enabled == hitless.MaintNotificationsEnabled {
-					// panic so we fail fast without breaking existing clients api
-					panic(fmt.Errorf("failed to enable hitless upgrades: %w", err))
-				}
+	// Initialize hitless upgrades first if enabled and protocol is RESP3
+	if opt.HitlessUpgradeConfig.IsEnabled() && opt.Protocol == 3 {
+		err := c.enableHitlessUpgrades()
+		if err != nil {
+			internal.Logger.Printf(context.Background(), "hitless: failed to initialize hitless upgrades: %v", err)
+			if opt.HitlessUpgradeConfig.Enabled == hitless.MaintNotificationsEnabled {
+				// panic so we fail fast without breaking existing clients api
+				panic(fmt.Errorf("failed to enable hitless upgrades: %w", err))
 			}
 		}
 	}

search_test.go

@@ -13,7 +13,6 @@ import (
 	. "github.com/bsm/gomega"
 	"github.com/redis/go-redis/v9"
 	"github.com/redis/go-redis/v9/helper"
-	"github.com/redis/go-redis/v9/hitless"
 )
 
 func WaitForIndexing(c *redis.Client, index string) {
@@ -3334,14 +3333,7 @@ var _ = Describe("RediSearch FT.Config with Resp2 and Resp3", Label("search", "N
 	var clientResp3 *redis.Client
 	BeforeEach(func() {
 		clientResp2 = redis.NewClient(&redis.Options{Addr: ":6379", Protocol: 2})
-		clientResp3 = redis.NewClient(&redis.Options{
-			Addr: ":6379",
-			Protocol: 3,
-			UnstableResp3: true,
-			HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
-				Enabled: hitless.MaintNotificationsDisabled,
-			},
-		})
+		clientResp3 = redis.NewClient(&redis.Options{Addr: ":6379", Protocol: 3, UnstableResp3: true})
 		Expect(clientResp3.FlushDB(ctx).Err()).NotTo(HaveOccurred())
 	})
 
@@ -3381,21 +3373,8 @@ var _ = Describe("RediSearch commands Resp 3", Label("search"), func() {
 	var client2 *redis.Client
 
 	BeforeEach(func() {
-		client = redis.NewClient(&redis.Options{
-			Addr: ":6379",
-			Protocol: 3,
-			UnstableResp3: true,
-			HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
-				Enabled: hitless.MaintNotificationsDisabled,
-			},
-		})
-		client2 = redis.NewClient(&redis.Options{
-			Addr: ":6379",
-			Protocol: 3,
-			HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
-				Enabled: hitless.MaintNotificationsDisabled,
-			},
-		})
+		client = redis.NewClient(&redis.Options{Addr: ":6379", Protocol: 3, UnstableResp3: true})
+		client2 = redis.NewClient(&redis.Options{Addr: ":6379", Protocol: 3})
 		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
 	})
 

universal_test.go

@@ -5,7 +5,6 @@ import (
 	. "github.com/bsm/gomega"
 
 	"github.com/redis/go-redis/v9"
-	"github.com/redis/go-redis/v9/hitless"
 )
 
 var _ = Describe("UniversalClient", func() {
@@ -54,9 +53,6 @@ var _ = Describe("UniversalClient", func() {
 			Addrs:         cluster.addrs(),
 			Protocol:      3,
 			UnstableResp3: true,
-			HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
-				Enabled: hitless.MaintNotificationsDisabled,
-			},
 		})
 		Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())
 		a := func() { client.FTInfo(ctx, "all").Result() }
@@ -68,9 +64,6 @@ var _ = Describe("UniversalClient", func() {
 			Addrs:         cluster.addrs(),
 			Protocol:      3,
 			UnstableResp3: true,
-			HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
-				Enabled: hitless.MaintNotificationsDisabled,
-			},
 		})
 		Expect(client.Ping(ctx).Err()).NotTo(HaveOccurred())
 		a := func() { client.FTInfo(ctx, "all").Result() }