wip

Author: ndyakov

adapters.go

@@ -77,7 +77,7 @@ func (oa *optionsAdapter) NewDialer() func(context.Context) (net.Conn, error) {
 	baseDialer := oa.options.NewDialer()
 	return func(ctx context.Context) (net.Conn, error) {
 		// Extract network and address from the options
-		network := "tcp"
+		network := oa.options.Network
 		addr := oa.options.Addr
 		return baseDialer(ctx, network, addr)
 	}

hitless/hitless_manager.go

@@ -213,15 +213,19 @@ func (hm *HitlessManager) Close() error {
 		return nil // Already closed
 	}
 
-	// Shutdown the pool hook
-	err := hm.poolHooksRef.Shutdown(context.Background())
-	if err != nil {
-		// was not able to close pool hook, keep closed state false
-		hm.closed.Store(false)
-		return err
+	// Shutdown the pool hook if it exists
+	if hm.poolHooksRef != nil {
+		err := hm.poolHooksRef.Shutdown(context.Background())
+		if err != nil {
+			// was not able to close pool hook, keep closed state false
+			hm.closed.Store(false)
+			return err
+		}
+		// Remove the pool hook from the pool
+		if hm.pool != nil {
+			hm.pool.RemovePoolHook(hm.poolHooksRef)
+		}
 	}
-	// Remove the pool hook from the pool
-	hm.pool.RemovePoolHook(hm.poolHooksRef)
 
 	// Clear all active operations
 	hm.activeMovingOps.Range(func(key, value interface{}) bool {

hitless/hitless_manager_test.go

@@ -75,7 +75,7 @@ func TestHitlessManagerRefactoring(t *testing.T) {
 		config := DefaultConfig()
 		client := &MockClient{options: &MockOptions{}}
 
-		manager, err := NewHitlessManager(client, config)
+		manager, err := NewHitlessManager(client, nil, config)
 		if err != nil {
 			t.Fatalf("Failed to create hitless manager: %v", err)
 		}
@@ -136,7 +136,7 @@ func TestHitlessManagerRefactoring(t *testing.T) {
 		config := DefaultConfig()
 		client := &MockClient{options: &MockOptions{}}
 
-		manager, err := NewHitlessManager(client, config)
+		manager, err := NewHitlessManager(client, nil, config)
 		if err != nil {
 			t.Fatalf("Failed to create hitless manager: %v", err)
 		}
@@ -178,7 +178,7 @@ func TestHitlessManagerRefactoring(t *testing.T) {
 		config := DefaultConfig()
 		client := &MockClient{options: &MockOptions{}}
 
-		manager, err := NewHitlessManager(client, config)
+		manager, err := NewHitlessManager(client, nil, config)
 		if err != nil {
 			t.Fatalf("Failed to create hitless manager: %v", err)
 		}

hitless/notification_handler.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/interfaces"
 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/push"
@@ -109,26 +110,38 @@ func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx pus
 		return ErrInvalidNotification
 	}
 
-	// TODO(hitless): if newEndpoint is empty, we should schedule a handoff to the current endpoint in timeS/2 seconds
+	deadline := time.Now().Add(time.Duration(timeS) * time.Second)
+	// If newEndpoint is empty, we should schedule a handoff to the current endpoint in timeS/2 seconds
+	if newEndpoint == "" || newEndpoint == internal.RedisNull {
+		// same as current endpoint
+		newEndpoint = snh.manager.options.GetAddr()
+		// delay the handoff for timeS/2 seconds to the same endpoint
+		// do this in a goroutine to avoid blocking the notification handler
+		go func() {
+			time.Sleep(time.Duration(timeS/2) * time.Second)
+			snh.markConnForHandoff(poolConn, newEndpoint, seqID, deadline)
+		}()
+		return nil
+	}
+
+	return snh.markConnForHandoff(poolConn, newEndpoint, seqID, deadline)
+}
 
-	// Mark the connection for handoff
-	if err := poolConn.MarkForHandoff(newEndpoint, seqID); err != nil {
+func (snh *NotificationHandler) markConnForHandoff(conn *pool.Conn, newEndpoint string, seqID int64, deadline time.Time) error {
+	if err := conn.MarkForHandoff(newEndpoint, seqID); err != nil {
 		// Connection is already marked for handoff, which is acceptable
 		// This can happen if multiple MOVING notifications are received for the same connection
 		return nil
 	}
-
 	// Optionally track in hitless manager for monitoring/debugging
 	if snh.manager != nil {
-		connID := poolConn.GetID()
-		deadline := time.Now().Add(time.Duration(timeS) * time.Second)
+		connID := conn.GetID()
 
 		// Track the operation (ignore errors since this is optional)
 		_ = snh.manager.TrackMovingOperationWithConnID(context.Background(), newEndpoint, deadline, seqID, connID)
 	} else {
 		return fmt.Errorf("hitless: manager not initialized")
 	}
-
 	return nil
 }
 

hitless/pool_hook_test.go

@@ -91,6 +91,14 @@ func (mp *mockPool) Stats() *pool.Stats {
 	return &pool.Stats{}
 }
 
+func (mp *mockPool) AddPoolHook(hook pool.PoolHook) {
+	// Mock implementation - do nothing
+}
+
+func (mp *mockPool) RemovePoolHook(hook pool.PoolHook) {
+	// Mock implementation - do nothing
+}
+
 func (mp *mockPool) Close() error {
 	return nil
 }

options.go

@@ -640,19 +640,7 @@ func getUserPassword(u *url.URL) (string, string) {
 	return user, password
 }
 
-func newConnPoolConfig(opt *Options) *pool.Config {
-	return &pool.Config{
-		PoolFIFO:        opt.PoolFIFO || opt.HitlessUpgradeConfig.IsEnabled(), // Always FIFO with hitless upgrades
-		PoolSize:        opt.PoolSize,
-		PoolTimeout:     opt.PoolTimeout,
-		DialTimeout:     opt.DialTimeout,
-		MinIdleConns:    opt.MinIdleConns,
-		MaxIdleConns:    opt.MaxIdleConns,
-		MaxActiveConns:  opt.MaxActiveConns,
-		ConnMaxIdleTime: opt.ConnMaxIdleTime,
-		ConnMaxLifetime: opt.ConnMaxLifetime,
-	}
-}
+
 
 func newConnPool(
 	opt *Options,