feat(home): add NewLifetime method and improve reconnect failover handling

- Introduced `NewLifetime` to create fresh `Client` instances while maintaining cluster failover state.
- Enhanced reconnect failure tracking with `markReconnectFailure` conditions for various connection stages.
- Reset reconnect failure counter on successful reconnections to ensure stable client behavior.
- Updated tests to verify failover mechanics, subscription handling, and heartbeat recovery scenarios.
- Improved lifetime management in `service_home` with reuse of `previousClient` for seamless client transitions.

internal/home/client.go

@@ -157,6 +157,22 @@ func New(homeCfg config.HomeConfig) *Client {
 	}
 }
 
+// NewLifetime creates a fresh client while preserving cluster failover state.
+func (c *Client) NewLifetime() *Client {
+	if c == nil {
+		return nil
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return &Client{
+		homeCfg:           c.homeCfg,
+		seedHost:          c.seedHost,
+		seedPort:          c.seedPort,
+		clusterNodes:      append([]clusterNode(nil), c.clusterNodes...),
+		reconnectFailures: c.reconnectFailures,
+	}
+}
+
 func (c *Client) Enabled() bool {
 	if c == nil {
 		return false
@@ -1568,11 +1584,17 @@ func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func(
 
 	c.closeBootstrapPools()
 	if errEnsure := c.ensureClients(); errEnsure != nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("connect")
+		}
 		return c.endConfigSubscriberLifetime(errEnsure)
 	}
 
 	raw, errGet := c.GetConfig(ctx)
 	if errGet != nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("config fetch")
+		}
 		return c.endConfigSubscriberLifetime(errGet)
 	}
 	if errApply := onConfig(raw); errApply != nil {
@@ -1581,21 +1603,34 @@ func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func(
 
 	sub, errSubClient := c.subscriptionClient()
 	if errSubClient != nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("subscribe client")
+		}
 		return c.endConfigSubscriberLifetime(errSubClient)
 	}
 	args, receiveTimeout := c.subscriptionParameters()
 	pubsub := sub.Subscribe(ctx, args...)
 	if pubsub == nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("subscribe")
+		}
 		return c.endConfigSubscriberLifetime(ErrNotConnected)
 	}
 
 	if errACK := receiveSubscriptionACKs(ctx, pubsub, receiveTimeout, args[:1]); errACK != nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("subscribe")
+		}
 		return c.endConfigSubscriberLifetimeWithSubscription(errACK, pubsub, "failed ACK")
 	}
 
 	if errProbe := c.rebuildCommandPoolAndProbe(ctx); errProbe != nil {
+		if ctx.Err() == nil {
+			c.markReconnectFailure("command probe")
+		}
 		return c.endConfigSubscriberLifetimeWithSubscription(errProbe, pubsub, "fresh command probe failure")
 	}
+	c.resetReconnectFailures()
 	c.heartbeatOK.Store(true)
 	if onReady != nil {
 		onReady()
@@ -1605,6 +1640,13 @@ func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func(
 		_, receiveTimeout = c.subscriptionParameters()
 		event, errReceive := pubsub.ReceiveTimeout(ctx, receiveTimeout)
 		if errReceive != nil {
+			if ctx.Err() == nil {
+				if isTimeoutError(errReceive) {
+					c.markSubscriptionTimeout()
+				} else {
+					c.markReconnectFailure("subscription")
+				}
+			}
 			return c.endConfigSubscriberLifetimeWithSubscription(errReceive, pubsub, "heartbeat loss")
 		}
 		switch msg := event.(type) {

internal/home/client_test.go

@@ -174,6 +174,49 @@ func TestFailoverAfterReconnectFailureDisabledDoesNotSwitchToClusterNode(t *test
 	}
 }
 
+func TestNewLifetimePreservesClusterFailoverState(t *testing.T) {
+	client := New(config.HomeConfig{Enabled: true, Host: "seed.example.com", Port: 8327})
+	client.mu.Lock()
+	client.homeCfg.Host = "failed.example.com"
+	client.clusterNodes = []clusterNode{
+		{IP: "failed.example.com", Port: 8327, ClientCount: 1},
+		{IP: "healthy.example.com", Port: 8327, ClientCount: 2},
+	}
+	client.reconnectFailures = homeReconnectFailoverThreshold - 1
+	client.mu.Unlock()
+	client.Close()
+
+	next := client.NewLifetime()
+	if next == nil {
+		t.Fatal("NewLifetime() = nil")
+	}
+	if got, _ := next.addr(); got != "failed.example.com:8327" {
+		t.Fatalf("addr() = %q, want failed.example.com:8327", got)
+	}
+	next.mu.Lock()
+	seedHost, seedPort := next.seedHost, next.seedPort
+	nodes := append([]clusterNode(nil), next.clusterNodes...)
+	failures := next.reconnectFailures
+	next.mu.Unlock()
+	if seedHost != "seed.example.com" || seedPort != 8327 {
+		t.Fatalf("seed = %s:%d, want seed.example.com:8327", seedHost, seedPort)
+	}
+	if !reflect.DeepEqual(nodes, []clusterNode{
+		{IP: "failed.example.com", Port: 8327, ClientCount: 1},
+		{IP: "healthy.example.com", Port: 8327, ClientCount: 2},
+	}) {
+		t.Fatalf("cluster nodes = %#v", nodes)
+	}
+	if failures != homeReconnectFailoverThreshold-1 {
+		t.Fatalf("reconnect failures = %d, want %d", failures, homeReconnectFailoverThreshold-1)
+	}
+
+	switched, addr := next.failoverAfterReconnectFailure()
+	if !switched || addr != "healthy.example.com:8327" {
+		t.Fatalf("failover = %t, %q, want true, healthy.example.com:8327", switched, addr)
+	}
+}
+
 func TestBuildKVSetArgs(t *testing.T) {
 	args, errArgs := buildKVSetArgs("key", []byte("value"), KVSetOptions{EX: 2 * time.Second, NX: true})
 	if errArgs != nil {
@@ -1130,7 +1173,10 @@ func TestRunConfigSubscriberLifetimeReturnsAfterHeartbeatLoss(t *testing.T) {
 	if errPort != nil {
 		t.Fatalf("parse listener port: %v", errPort)
 	}
-	client := New(config.HomeConfig{Enabled: true, Host: host, Port: port, DisableClusterDiscovery: true})
+	client := New(config.HomeConfig{Enabled: true, Host: host, Port: port})
+	client.mu.Lock()
+	client.clusterNodes = []clusterNode{{IP: "failover.example.com", Port: 8327}}
+	client.mu.Unlock()
 
 	ready := make(chan struct{}, 1)
 	errRun := client.RunConfigSubscriberLifetime(context.Background(), func(raw []byte) error {
@@ -1154,6 +1200,9 @@ func TestRunConfigSubscriberLifetimeReturnsAfterHeartbeatLoss(t *testing.T) {
 	if client.HeartbeatOK() {
 		t.Fatal("HeartbeatOK() = true after heartbeat loss")
 	}
+	if got, _ := client.addr(); got != "failover.example.com:8327" {
+		t.Fatalf("addr() = %q, want failover.example.com:8327 after heartbeat timeout", got)
+	}
 	client.mu.Lock()
 	commandClient, subscriptionClient := client.cmd, client.sub
 	client.mu.Unlock()
@@ -1190,6 +1239,11 @@ func TestRunConfigSubscriberLifetimeRejectsInvalidSubscriptionACK(t *testing.T)
 					return "+OK\r\n"
 				}
 			})
+			client.mu.Lock()
+			client.homeCfg.DisableClusterDiscovery = false
+			client.clusterNodes = []clusterNode{{IP: "failover.example.com", Port: 8327}}
+			client.reconnectFailures = homeReconnectFailoverThreshold - 1
+			client.mu.Unlock()
 			errRun := client.RunConfigSubscriberLifetime(context.Background(), func([]byte) error { return nil }, nil)
 			if errRun == nil {
 				t.Fatal("RunConfigSubscriberLifetime() error = nil, want invalid ACK rejection")
@@ -1197,6 +1251,9 @@ func TestRunConfigSubscriberLifetimeRejectsInvalidSubscriptionACK(t *testing.T)
 			if command := findRedisCommand(commands.All(), "PING"); command != nil {
 				t.Fatalf("PING command = %#v, want no command pool exposure before valid ACK", command)
 			}
+			if got, _ := client.addr(); got != "failover.example.com:8327" {
+				t.Fatalf("addr() = %q, want failover.example.com:8327 after repeated subscription failure", got)
+			}
 		})
 	}
 }

sdk/cliproxy/service_home.go

@@ -491,9 +491,15 @@ func (s *Service) runHomeSubscriber(homeCtx context.Context, parentCtx context.C
 		close(supervisor.done)
 	}()
 
+	var previousClient *home.Client
 	for homeCtx.Err() == nil {
 		supervisor.setPublisherCompletion(nil)
-		client := home.New(homeCfg)
+		client := previousClient
+		if client == nil {
+			client = home.New(homeCfg)
+		} else {
+			client = client.NewLifetime()
+		}
 		client.SetManagedLifetime(true)
 		registry := executionregistry.New()
 		releaseCtx, releaseCancel := context.WithCancel(context.WithoutCancel(homeCtx))
@@ -577,6 +583,7 @@ func (s *Service) runHomeSubscriber(homeCtx context.Context, parentCtx context.C
 		if !published.Load() && errRun != nil && !waitForHomeSubscriberRetry(homeCtx, homeSubscriberPreAckRetryBackoff) {
 			return
 		}
+		previousClient = client
 	}
 }