From bb43b7a30fc8ad12b1d40890cce1a55f5425e8ec Mon Sep 17 00:00:00 2001
From: hkfires <10558748+hkfires@users.noreply.github.com>
Date: Tue, 28 Jul 2026 08:34:43 +0800
Subject: [PATCH] feat(home): add membership takeover eligibility handling and
 update related tests

---
 internal/home/client.go      | 17 ++++++++++++++---
 internal/home/client_test.go | 23 +++++++++++++++++++++--
 2 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/internal/home/client.go b/internal/home/client.go
index fde6ed8f..a7b8fa01 100644
--- a/internal/home/client.go
+++ b/internal/home/client.go
@@ -1628,6 +1628,15 @@ func (c *Client) subscriptionParameters() ([]string, time.Duration) {
 	return args, cfg.CPAHeartbeatTimeout
 }
 
+func (c *Client) markMembershipTakeoverEligible() {
+	if c == nil {
+		return
+	}
+	if !c.recoveryState.CompareAndSwap(uint32(recoveryStateStable), uint32(recoveryStateTakeoverEligible)) {
+		c.recoveryState.CompareAndSwap(uint32(recoveryStateSwitching), uint32(recoveryStateSwitchingTakeover))
+	}
+}
+
 func (c *Client) rebuildCommandPoolAndProbe(ctx context.Context) error {
 	c.promoteSubscription()
 	if errPing := c.Ping(ctx); errPing != nil {
@@ -1726,6 +1735,10 @@ func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func(
 		}
 		return c.endConfigSubscriberLifetimeWithSubscription(errACK, pubsub, "failed ACK")
 	}
+	// A protocol-one ACK means Home already committed this membership. Preserve it if the command probe fails.
+	if len(args) > 1 {
+		c.markMembershipTakeoverEligible()
+	}
 
 	if errProbe := c.rebuildCommandPoolAndProbe(ctx); errProbe != nil {
 		if ctx.Err() == nil {
@@ -1745,9 +1758,7 @@ func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func(
 		if errReceive != nil {
 			if ctx.Err() == nil {
 				if c.heartbeatOK.Load() {
-					if !c.recoveryState.CompareAndSwap(uint32(recoveryStateStable), uint32(recoveryStateTakeoverEligible)) {
-						c.recoveryState.CompareAndSwap(uint32(recoveryStateSwitching), uint32(recoveryStateSwitchingTakeover))
-					}
+					c.markMembershipTakeoverEligible()
 				}
 				if isTimeoutError(errReceive) {
 					c.markSubscriptionTimeout()
diff --git a/internal/home/client_test.go b/internal/home/client_test.go
index 4cfd9251..af414c58 100644
--- a/internal/home/client_test.go
+++ b/internal/home/client_test.go
@@ -1968,9 +1968,9 @@ func TestRunConfigSubscriberLifetimeRebuildsFreshCommandPoolBeforeReady(t *testi
 	}
 }
 
-func TestRunConfigSubscriberLifetimeDoesNotReadyWhenFreshCommandProbeFails(t *testing.T) {
+func TestRunConfigSubscriberLifetimePreservesTakeoverWhenFreshCommandProbeFails(t *testing.T) {
 	configPayload := "host: 127.0.0.1\n"
-	client, _ := newRedisCommandTestClient(t, func(args []string) string {
+	client, commands := newRedisCommandTestClient(t, func(args []string) string {
 		switch {
 		case len(args) >= 1 && strings.EqualFold(args[0], "HELLO"):
 			return "%6\r\n$6\r\nserver\r\n$5\r\nredis\r\n$5\r\nproto\r\n:3\r\n$2\r\nid\r\n:1\r\n$4\r\nmode\r\n$10\r\nstandalone\r\n$4\r\nrole\r\n$6\r\nmaster\r\n$7\r\nmodules\r\n*0\r\n"
@@ -1984,6 +1984,10 @@ func TestRunConfigSubscriberLifetimeDoesNotReadyWhenFreshCommandProbeFails(t *te
 			return "+OK\r\n"
 		}
 	})
+	lifecycle := config.CredentialConcurrencyConfig{LifecycleConfigRevision: 9}
+	if errSet := client.SetLifecycleConfig(lifecycle); errSet != nil {
+		t.Fatal(errSet)
+	}
 	ready := make(chan struct{}, 1)
 	errRun := client.RunConfigSubscriberLifetime(context.Background(), func([]byte) error { return nil }, func() { ready <- struct{}{} })
 	if errRun == nil {
@@ -2000,6 +2004,21 @@ func TestRunConfigSubscriberLifetimeDoesNotReadyWhenFreshCommandProbeFails(t *te
 	if commandClient != nil || subscriptionClient != nil {
 		t.Fatalf("clients retained after fresh command probe failure: command=%v subscription=%v", commandClient != nil, subscriptionClient != nil)
 	}
+	if got := recoveryState(client.recoveryState.Load()); got != recoveryStateTakeoverEligible {
+		t.Fatalf("recovery state = %d, want %d", got, recoveryStateTakeoverEligible)
+	}
+	if got := findRedisCommand(commands.All(), "SUBSCRIBE"); !reflect.DeepEqual(got, []string{"subscribe", "config", "9"}) {
+		t.Fatalf("initial SUBSCRIBE wire command = %#v, want []string{\"subscribe\", \"config\", \"9\"}", got)
+	}
+
+	next := client.NewLifetime()
+	if errSet := next.SetLifecycleConfig(lifecycle); errSet != nil {
+		t.Fatal(errSet)
+	}
+	args, _ := next.subscriptionParameters()
+	if !reflect.DeepEqual(args, []string{"config", "9", "takeover"}) {
+		t.Fatalf("replacement SUBSCRIBE args = %#v, want takeover", args)
+	}
 }
 
 func findRedisCommand(commands [][]string, commandName string) []string {
-- 
2.51.2