diff --git a/internal/home/client.go b/internal/home/client.go index d16a9b1e7..a8c0a729a 100644 --- a/internal/home/client.go +++ b/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) { diff --git a/internal/home/client_test.go b/internal/home/client_test.go index 973969acb..133a6ee61 100644 --- a/internal/home/client_test.go +++ b/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) + } }) } } diff --git a/sdk/cliproxy/service_home.go b/sdk/cliproxy/service_home.go index f13be7494..10f8fdf92 100644 --- a/sdk/cliproxy/service_home.go +++ b/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 } }