diff --git a/internal/home/client.go b/internal/home/client.go index 3ac162ac..8f0dcc28 100644 --- a/internal/home/client.go +++ b/internal/home/client.go @@ -1307,7 +1307,7 @@ func (c *Client) concurrencyReleaseClient() (*redis.Client, error) { return nil, ErrDispatchFenced } state := recoveryState(c.recoveryState.Load()) - if state == recoveryStateSwitching || state == recoveryStateSwitchingTakeover { + if state == recoveryStateTakeoverEligible || state == recoveryStateSwitching || state == recoveryStateSwitchingTakeover { return nil, ErrNotConnected } if !c.Enabled() { @@ -1320,7 +1320,7 @@ func (c *Client) concurrencyReleaseClient() (*redis.Client, error) { return nil, ErrDispatchFenced } state = recoveryState(c.recoveryState.Load()) - if state == recoveryStateSwitching || state == recoveryStateSwitchingTakeover { + if state == recoveryStateTakeoverEligible || state == recoveryStateSwitching || state == recoveryStateSwitchingTakeover { return nil, ErrNotConnected } if c.release != nil { diff --git a/internal/home/client_test.go b/internal/home/client_test.go index b6ea2984..173bb533 100644 --- a/internal/home/client_test.go +++ b/internal/home/client_test.go @@ -243,18 +243,31 @@ func TestEnsureClientsWaitsForPreviousTargetClose(t *testing.T) { client.Close() } -func TestConcurrencyReleaseDoesNotOpenSwitchingTarget(t *testing.T) { - client := New(config.HomeConfig{Enabled: true, Host: "next.example.com", Port: 8327}) - client.recoveryState.Store(uint32(recoveryStateSwitching)) - errRelease := client.PushConcurrencyRelease(context.Background(), ConcurrencyReleaseFrame{CredentialID: "cred-a", Model: "model-a", ReleaseSeq: 1}) - if !errors.Is(errRelease, ErrNotConnected) { - t.Fatalf("PushConcurrencyRelease() error = %v, want %v", errRelease, ErrNotConnected) - } - client.mu.Lock() - releaseClient := client.release - client.mu.Unlock() - if releaseClient != nil { - t.Fatal("release client was opened before the switched target became ready") +func TestConcurrencyReleaseDoesNotOpenBeforeMembershipReady(t *testing.T) { + tests := []struct { + name string + state recoveryState + }{ + {name: "takeover pending", state: recoveryStateTakeoverEligible}, + {name: "target switching", state: recoveryStateSwitching}, + {name: "target switching with takeover", state: recoveryStateSwitchingTakeover}, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + client := New(config.HomeConfig{Enabled: true, Host: "next.example.com", Port: 8327}) + client.recoveryState.Store(uint32(testCase.state)) + errRelease := client.PushConcurrencyRelease(context.Background(), ConcurrencyReleaseFrame{CredentialID: "cred-a", Model: "model-a", ReleaseSeq: 1}) + if !errors.Is(errRelease, ErrNotConnected) { + t.Fatalf("PushConcurrencyRelease() error = %v, want %v", errRelease, ErrNotConnected) + } + client.mu.Lock() + releaseClient := client.release + client.mu.Unlock() + if releaseClient != nil { + t.Fatal("release client was opened before the membership became ready") + } + }) } }