From ac8f75a16a558f9ff7c21964f9db474b4e7091d5 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:15:40 +0100 Subject: [PATCH 1/6] driver/redis: tests for a SentinelClient closed during handleSwitchMaster (Refs #1104) Two defects, both fail here and are fixed by the next commit. A Close that lands while dialMaster is in flight (its own timeout, not closeCh) leaves the new client installed on the closed SentinelClient, and open: 20 of 20 runs. A failed dial followed by a Close, with the retry pause over too, is a tie between closeCh and the timer, and the loop dialled the master again about half of the time: 52 of 100 runs. The first pause of handleSwitchMaster is a variable (sentinelSwitchBackoffBase) so the test can make it end at once; its default is the 100 ms it was. --- driver/redis/sentinel.go | 7 +- .../redis/sentinel_close_switch_1104_test.go | 185 ++++++++++++++++++ 2 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 driver/redis/sentinel_close_switch_1104_test.go diff --git a/driver/redis/sentinel.go b/driver/redis/sentinel.go index f4d1a14db..81e22ad07 100644 --- a/driver/redis/sentinel.go +++ b/driver/redis/sentinel.go @@ -44,6 +44,11 @@ type SentinelConfig struct { // subscription. var sentinelDownGrace = 5 * time.Second +// sentinelSwitchBackoffBase is the first pause of handleSwitchMaster between +// two dials of the new master (it doubles, up to 2 s). A variable so a test can +// make the pause end at once. +var sentinelSwitchBackoffBase = 100 * time.Millisecond + // SentinelClient is a Redis client managed by Redis Sentinel. It discovers the // current primary via Sentinel, connects to it, and automatically reconnects // to the new primary when a failover is detected. @@ -459,7 +464,7 @@ func (s *SentinelClient) handleSwitchMaster(payload string) { } newAddr := net.JoinHostPort(parts[3], parts[4]) - bo := async.NewBackoff(100*time.Millisecond, 2*time.Second) + bo := async.NewBackoff(sentinelSwitchBackoffBase, 2*time.Second) const maxRetries = 3 var newClient *Client var lastErr error diff --git a/driver/redis/sentinel_close_switch_1104_test.go b/driver/redis/sentinel_close_switch_1104_test.go new file mode 100644 index 000000000..d6bb5f182 --- /dev/null +++ b/driver/redis/sentinel_close_switch_1104_test.go @@ -0,0 +1,185 @@ +package redis + +import ( + "bufio" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// gatedMaster1104 is a Redis master whose reply to ROLE waits for release, so a +// test can call SentinelClient.Close while handleSwitchMaster's dialMaster is +// in flight (ROLE is the last step of the dial). role is what ROLE answers. It +// counts the ROLE requests it saw and the connections the client closed. +type gatedMaster1104 struct { + ln net.Listener + role string + seen chan struct{} // receives once per ROLE request (buffered) + release chan struct{} + roles atomic.Int32 + closed atomic.Int32 + mu sync.Mutex + conns []net.Conn +} + +func startGatedMaster1104(t *testing.T, role string) *gatedMaster1104 { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + g := &gatedMaster1104{ln: ln, role: role, seen: make(chan struct{}, 16), release: make(chan struct{})} + t.Cleanup(func() { + _ = ln.Close() + g.mu.Lock() + for _, c := range g.conns { + _ = c.Close() + } + g.mu.Unlock() + }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + g.mu.Lock() + g.conns = append(g.conns, c) + g.mu.Unlock() + go g.serve(c) + } + }() + return g +} + +func (g *gatedMaster1104) Addr() string { return g.ln.Addr().String() } + +func (g *gatedMaster1104) serve(c net.Conn) { + defer func() { g.closed.Add(1); _ = c.Close() }() + r := bufio.NewReader(c) + w := bufio.NewWriter(c) + var buf []string + for { + cmd, err := readCommandInto(r, buf[:0]) + if err != nil { + return + } + buf = cmd + if len(cmd) == 0 { + continue + } + switch strings.ToUpper(cmd[0]) { + case "HELLO": + proto := 2 + if len(cmd) > 1 && cmd[1] == "3" { + proto = 3 + } + handleHELLO(w, proto) + case "ROLE": + g.roles.Add(1) + select { + case g.seen <- struct{}{}: + default: + } + <-g.release + writeArrayHeader(w, 3) + writeBulk(w, g.role) + writeInt(w, 0) + writeArrayHeader(w, 0) + case "PING": + writeSimple(w, "PONG") + default: + writeSimple(w, "OK") + } + if w.Flush() != nil { + return + } + } +} + +// switchPayload1104 is a +switch-master payload naming addr as the new master. +func switchPayload1104(addr string) string { + host, port, _ := net.SplitHostPort(addr) + return "mymaster 127.0.0.1 1 " + host + " " + port +} + +// closeDuringDial1104 runs handleSwitchMaster against g and calls Close while +// its dial is held at ROLE; it returns once handleSwitchMaster has returned. +func closeDuringDial1104(t *testing.T, g *gatedMaster1104) *SentinelClient { + t.Helper() + master := startFakeMaster(t) + sent := startFakeSentinel(t, master.Addr()) + sc, err := NewSentinelClient(SentinelConfig{MasterName: "mymaster", SentinelAddrs: []string{sent.Addr()}, DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + go func() { sc.handleSwitchMaster(switchPayload1104(g.Addr())); close(done) }() + select { + case <-g.seen: + case <-time.After(5 * time.Second): + t.Fatal("the dial of the new master never reached ROLE") + } + _ = sc.Close() + close(g.release) + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("handleSwitchMaster did not return") + } + return sc +} + +// A Close that lands while handleSwitchMaster's dialMaster is in flight (it runs +// under its own timeout, not closeCh) must not leave the new client installed on +// the closed SentinelClient: nothing would ever close it. Before the fix +// handleSwitchMaster stored it with no look at the closed flag. +func TestSentinelCloseDuringSwitchMasterDialInstallsNothing1104(t *testing.T) { + const iters = 20 + installed, open := 0, 0 + for range iters { + g := startGatedMaster1104(t, "master") + sc := closeDuringDial1104(t, g) + sc.mu.RLock() + if sc.client != nil { + installed++ + } + sc.mu.RUnlock() + // The client the dial made is closed, not just left out. + deadline := time.Now().Add(500 * time.Millisecond) + for g.closed.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if g.closed.Load() == 0 { + open++ + } + } + if installed != 0 || open != 0 { + t.Errorf("after a Close during the dial: a client was installed on the closed SentinelClient in %d of %d runs, and its connection stayed open in %d", installed, iters, open) + } +} + +// A failed dial followed by a Close, with the retry pause over as well (a base of +// 1 ns), is a tie between closeCh and the timer; Go picks at random, so the +// loop dialled the master again after Close about half of the time. It must +// not: exactly one ROLE request, the one that was in flight when Close ran. +func TestSentinelSwitchMasterDoesNotDialAfterClose1104(t *testing.T) { + old := sentinelSwitchBackoffBase + sentinelSwitchBackoffBase = time.Nanosecond + t.Cleanup(func() { sentinelSwitchBackoffBase = old }) + const iters = 100 + extra := 0 + for range iters { + g := startGatedMaster1104(t, "slave") // the dial fails: not a master + _ = closeDuringDial1104(t, g) + if n := g.roles.Load(); n != 1 { + extra++ + } + } + if extra != 0 { + t.Errorf("handleSwitchMaster dialled the new master again after Close in %d of %d runs", extra, iters) + } +} From ff1f37604d52bf9c39ea86497c1598b89ecbd2fa Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:15:58 +0100 Subject: [PATCH 2/6] driver/redis: handleSwitchMaster installs nothing on, and does not dial for, a closed SentinelClient (Refs #1104) Under s.mu, look at the closed flag before storing the new client and close the client when it is set: Close sets the flag first, so either handleSwitchMaster sees it or Close finds the stored client and closes it. After the retry pause, a closed client wins a tie with the timer, so no further dial is made. --- driver/redis/sentinel.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/driver/redis/sentinel.go b/driver/redis/sentinel.go index 81e22ad07..fb00299a4 100644 --- a/driver/redis/sentinel.go +++ b/driver/redis/sentinel.go @@ -483,10 +483,26 @@ func (s *SentinelClient) handleSwitchMaster(payload string) { case <-s.closeCh: return } + // closeCh wins a tie with the timer: select picks at random when + // both are ready, and another dial would connect to the master + // after Close. + if s.closed.Load() { + return + } } } s.mu.Lock() + // Close sets closed before it takes s.mu to clear s.client, and a dial + // runs under its own timeout, not closeCh, so a Close can land while + // dialMaster is in flight. Nothing closes a client installed after it. + if s.closed.Load() { + s.mu.Unlock() + if newClient != nil { + _ = newClient.Close() + } + return + } old := s.client if newClient != nil { s.client = newClient From bfc657b733cc75a36819e394512bea11ffdb995b Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:16:22 +0100 Subject: [PATCH 3/6] driver/redis: tests for a ClusterClient refresh or redirect after Close (Refs #1104) closeNodes leaves c.nodes nil. A refresh whose CLUSTER SLOTS reply came before Close and takes c.mu after it, and getOrCreateNode for a redirect, write into that nil map and end the process with a panic, after dialling a node nothing closes. Both fail here, 30 of 30 runs each; the next commit fixes them. refreshTopology is split so the part under c.mu (applyTopology) can run at the moment Close has just returned; no behaviour change. --- driver/redis/cluster.go | 7 ++ .../redis/cluster_close_refresh_1104_test.go | 95 +++++++++++++++++++ 2 files changed, 102 insertions(+) create mode 100644 driver/redis/cluster_close_refresh_1104_test.go diff --git a/driver/redis/cluster.go b/driver/redis/cluster.go index 3a539f80f..5f2f631a8 100644 --- a/driver/redis/cluster.go +++ b/driver/redis/cluster.go @@ -336,6 +336,13 @@ func (c *ClusterClient) refreshTopology(ctx context.Context) error { if v.Type != protocol.TyArray { return fmt.Errorf("CLUSTER SLOTS: unexpected reply type %s", v.Type) } + return c.applyTopology(ctx, v) +} + +// applyTopology rebuilds the slot-to-node mapping from a CLUSTER SLOTS reply. +// It is the part of refreshTopology that runs under c.mu, split out so a test +// can run it at the moment a Close has just returned. +func (c *ClusterClient) applyTopology(ctx context.Context, v *Value) error { c.mu.Lock() defer c.mu.Unlock() diff --git a/driver/redis/cluster_close_refresh_1104_test.go b/driver/redis/cluster_close_refresh_1104_test.go new file mode 100644 index 000000000..b5369c6e8 --- /dev/null +++ b/driver/redis/cluster_close_refresh_1104_test.go @@ -0,0 +1,95 @@ +package redis + +import ( + "context" + "errors" + "testing" + "time" +) + +// closedClusterWithSlots1104 returns a cluster client that has been closed and +// a CLUSTER SLOTS reply it fetched before the Close: the state of a refresh whose +// Do finished just before Close and has not yet taken c.mu. +func closedClusterWithSlots1104(t *testing.T) (*ClusterClient, *Value, *fakeClusterNode) { + t.Helper() + node1, _, _ := setupCluster(t) + cc, err := NewClusterClient(ClusterConfig{Addrs: []string{node1.Addr()}, DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatalf("NewClusterClient: %v", err) + } + v, err := cc.anyNode().client.Do(context.Background(), "CLUSTER", "SLOTS") + if err != nil { + t.Fatalf("CLUSTER SLOTS: %v", err) + } + if err := cc.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + return cc, v, node1 +} + +// connsOf1104 is how many connections the fake node has accepted so far. +func connsOf1104(n *fakeClusterNode) int { + n.fake.mu.Lock() + defer n.fake.mu.Unlock() + return len(n.fake.conns) +} + +// panicOf1104 runs f and returns the panic it ended with, or nil. +func panicOf1104(f func()) (panicked any) { + defer func() { panicked = recover() }() + f() + return nil +} + +// A refresh that takes c.mu after Close wrote into the nil c.nodes that +// closeNodes had left and ended the process with "assignment to entry in nil +// map" (after dialling a node nothing would close). It must return ErrClosed +// and dial nothing. +func TestClusterRefreshAfterCloseDoesNotWriteToTheNodes1104(t *testing.T) { + const iters = 30 + panics, dialled, wrongErr := 0, 0, 0 + for range iters { + cc, v, node1 := closedClusterWithSlots1104(t) + before := connsOf1104(node1) + var err error + if p := panicOf1104(func() { err = cc.applyTopology(context.Background(), v) }); p != nil { + panics++ + continue + } + if connsOf1104(node1) != before { + dialled++ + } + if !errors.Is(err, ErrClosed) { + wrongErr++ + } + } + if panics != 0 || dialled != 0 || wrongErr != 0 { + t.Errorf("a topology refresh after Close: %d of %d panicked, %d dialled a node, %d did not return ErrClosed", panics, iters, dialled, wrongErr) + } +} + +// getOrCreateNode is the other writer of c.nodes (a MOVED or ASK redirect to an +// address not known yet): after Close it dialled the node and stored it in the +// nil map, which panics. It must return ErrClosed and dial nothing. +func TestClusterGetOrCreateNodeAfterCloseDoesNotWriteToTheNodes1104(t *testing.T) { + const iters = 30 + panics, dialled, wrongErr := 0, 0, 0 + for range iters { + cc, _, node1 := closedClusterWithSlots1104(t) + before := connsOf1104(node1) + var err error + if p := panicOf1104(func() { _, err = cc.getOrCreateNode(node1.Addr()) }); p != nil { + panics++ + continue + } + if connsOf1104(node1) != before { + dialled++ + } + if !errors.Is(err, ErrClosed) { + wrongErr++ + } + } + if panics != 0 || dialled != 0 || wrongErr != 0 { + t.Errorf("getOrCreateNode after Close: %d of %d panicked, %d dialled a node, %d did not return ErrClosed", panics, iters, dialled, wrongErr) + } +} From 6a98787ebb60a17aa45252c9a0c0b6873221339c Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:16:55 +0100 Subject: [PATCH 4/6] driver/redis: a ClusterClient refresh or redirect after Close returns ErrClosed (Refs #1104) applyTopology and getOrCreateNode look at c.nodes under c.mu before they dial or write: closeNodes leaves it nil, and a write into it is a panic. getOrCreateNode also looks at the closed flag before it dials, and closes a client it dialled while Close ran. --- driver/redis/cluster.go | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/driver/redis/cluster.go b/driver/redis/cluster.go index 5f2f631a8..88a66e3e7 100644 --- a/driver/redis/cluster.go +++ b/driver/redis/cluster.go @@ -224,6 +224,9 @@ func (c *ClusterClient) getOrCreateNodeAs(addr string, isPrimary bool) (*cluster if ok { return n, nil } + if c.closed.Load() { + return nil, ErrClosed + } var ( client *Client err error @@ -238,6 +241,11 @@ func (c *ClusterClient) getOrCreateNodeAs(addr string, isPrimary bool) (*cluster } n = &clusterNode{addr: addr, client: client, primary: isPrimary} c.mu.Lock() + if c.nodes == nil { // Close ran while this dialled (see applyTopology) + c.mu.Unlock() + _ = client.Close() + return nil, ErrClosed + } if existing, ok := c.nodes[addr]; ok { c.mu.Unlock() _ = client.Close() @@ -345,6 +353,11 @@ func (c *ClusterClient) refreshTopology(ctx context.Context) error { func (c *ClusterClient) applyTopology(ctx context.Context, v *Value) error { c.mu.Lock() defer c.mu.Unlock() + // closeNodes leaves c.nodes nil: a refresh whose reply came before Close + // gets here after it, and must neither dial nor write into the nil map. + if c.nodes == nil { + return ErrClosed + } wantReplicas := c.cfg.ReadOnly || c.cfg.RouteByLatency From b7843526f840b2c765d30b9ba16e399623d849fa Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:36:43 +0100 Subject: [PATCH 5/6] driver/redis: tests for the checks in getOrCreateNode and for a switch-master that runs after Close (Refs #1104) The cluster tests asserted that no node was dialled, which cannot fail: a Client dials lazily. They assert ErrClosed and that nothing was stored in c.nodes now. The part of getOrCreateNode under c.mu is split out as installNode (no behaviour change) and tested on its own, so removing its check turns a test red; the early closed check cannot be reached alone (without it installNode answers the same), and the test says so. handleSwitchMaster, called after Close with a switch message already queued (the pick watchFailover's select can make), dialled the new master once: the new test fails, 50 of 50 runs. The next commit fixes it. --- driver/redis/cluster.go | 19 ++- .../redis/cluster_close_refresh_1104_test.go | 126 ++++++++++++++---- .../redis/sentinel_close_switch_1104_test.go | 34 +++++ 3 files changed, 147 insertions(+), 32 deletions(-) diff --git a/driver/redis/cluster.go b/driver/redis/cluster.go index 88a66e3e7..64e04446f 100644 --- a/driver/redis/cluster.go +++ b/driver/redis/cluster.go @@ -239,19 +239,26 @@ func (c *ClusterClient) getOrCreateNodeAs(addr string, isPrimary bool) (*cluster if err != nil { return nil, err } - n = &clusterNode{addr: addr, client: client, primary: isPrimary} + return c.installNode(&clusterNode{addr: addr, client: client, primary: isPrimary}) +} + +// installNode stores a node a redirect just dialled and returns the node to +// use. If the address was stored meanwhile it returns that one, and if Close ran +// while the node dialled (closeNodes left c.nodes nil, and a write into it is a +// panic) it returns ErrClosed; in both cases n's client is closed. +func (c *ClusterClient) installNode(n *clusterNode) (*clusterNode, error) { c.mu.Lock() - if c.nodes == nil { // Close ran while this dialled (see applyTopology) + if c.nodes == nil { c.mu.Unlock() - _ = client.Close() + _ = n.client.Close() return nil, ErrClosed } - if existing, ok := c.nodes[addr]; ok { + if existing, ok := c.nodes[n.addr]; ok { c.mu.Unlock() - _ = client.Close() + _ = n.client.Close() return existing, nil } - c.nodes[addr] = n + c.nodes[n.addr] = n c.mu.Unlock() return n, nil } diff --git a/driver/redis/cluster_close_refresh_1104_test.go b/driver/redis/cluster_close_refresh_1104_test.go index b5369c6e8..fa3b859c3 100644 --- a/driver/redis/cluster_close_refresh_1104_test.go +++ b/driver/redis/cluster_close_refresh_1104_test.go @@ -5,6 +5,8 @@ import ( "errors" "testing" "time" + + "github.com/goceleris/celeris/driver/internal/async" ) // closedClusterWithSlots1104 returns a cluster client that has been closed and @@ -27,13 +29,6 @@ func closedClusterWithSlots1104(t *testing.T) (*ClusterClient, *Value, *fakeClus return cc, v, node1 } -// connsOf1104 is how many connections the fake node has accepted so far. -func connsOf1104(n *fakeClusterNode) int { - n.fake.mu.Lock() - defer n.fake.mu.Unlock() - return len(n.fake.conns) -} - // panicOf1104 runs f and returns the panic it ended with, or nil. func panicOf1104(f func()) (panicked any) { defer func() { panicked = recover() }() @@ -41,55 +36,134 @@ func panicOf1104(f func()) (panicked any) { return nil } +// clientClosed1104 reports whether err is what a command on a closed Client +// answers. +func clientClosed1104(err error) bool { + return errors.Is(err, ErrClosed) || errors.Is(err, async.ErrPoolClosed) +} + +// nodesNil1104 reports whether closeNodes' nil c.nodes is still nil, i.e. nothing +// was stored after Close. +func nodesNil1104(c *ClusterClient) bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.nodes == nil +} + // A refresh that takes c.mu after Close wrote into the nil c.nodes that // closeNodes had left and ended the process with "assignment to entry in nil -// map" (after dialling a node nothing would close). It must return ErrClosed -// and dial nothing. +// map". It must return ErrClosed and store nothing. func TestClusterRefreshAfterCloseDoesNotWriteToTheNodes1104(t *testing.T) { const iters = 30 - panics, dialled, wrongErr := 0, 0, 0 + panics, wrongErr, stored := 0, 0, 0 for range iters { - cc, v, node1 := closedClusterWithSlots1104(t) - before := connsOf1104(node1) + cc, v, _ := closedClusterWithSlots1104(t) var err error if p := panicOf1104(func() { err = cc.applyTopology(context.Background(), v) }); p != nil { panics++ continue } - if connsOf1104(node1) != before { - dialled++ - } if !errors.Is(err, ErrClosed) { wrongErr++ } + if !nodesNil1104(cc) { + stored++ + } } - if panics != 0 || dialled != 0 || wrongErr != 0 { - t.Errorf("a topology refresh after Close: %d of %d panicked, %d dialled a node, %d did not return ErrClosed", panics, iters, dialled, wrongErr) + if panics != 0 || wrongErr != 0 || stored != 0 { + t.Errorf("a topology refresh after Close: %d of %d panicked, %d did not return ErrClosed, %d stored a node", panics, iters, wrongErr, stored) } } // getOrCreateNode is the other writer of c.nodes (a MOVED or ASK redirect to an -// address not known yet): after Close it dialled the node and stored it in the -// nil map, which panics. It must return ErrClosed and dial nothing. +// address not known yet): after Close it must return ErrClosed and store +// nothing. It gets there through its early closed check, which is not reachable +// on its own: without it the node is dialled (a Client dials lazily, so nothing +// observable happens) and installNode answers ErrClosed too, with the same +// result; the check only saves building a client for a closed cluster. The +// check under c.mu is tested through installNode below. func TestClusterGetOrCreateNodeAfterCloseDoesNotWriteToTheNodes1104(t *testing.T) { const iters = 30 - panics, dialled, wrongErr := 0, 0, 0 + panics, wrongErr, stored := 0, 0, 0 for range iters { cc, _, node1 := closedClusterWithSlots1104(t) - before := connsOf1104(node1) var err error if p := panicOf1104(func() { _, err = cc.getOrCreateNode(node1.Addr()) }); p != nil { panics++ continue } - if connsOf1104(node1) != before { - dialled++ - } if !errors.Is(err, ErrClosed) { wrongErr++ } + if !nodesNil1104(cc) { + stored++ + } + } + if panics != 0 || wrongErr != 0 || stored != 0 { + t.Errorf("getOrCreateNode after Close: %d of %d panicked, %d did not return ErrClosed, %d stored a node", panics, iters, wrongErr, stored) + } +} + +// installNode is the part of getOrCreateNode under c.mu, which a Close that ran +// while the node dialled reaches with no early check in the way: it must return +// ErrClosed, store nothing and close the client it was given. Its other two +// outcomes are pinned too, so the test is of the function and not only of this +// case. +func TestClusterInstallNode1104(t *testing.T) { + const iters = 30 + panics, wrongErr, stored, open := 0, 0, 0, 0 + for range iters { + cc, _, node1 := closedClusterWithSlots1104(t) + client, err := cc.dialNode(node1.Addr()) + if err != nil { + t.Fatalf("dialNode: %v", err) + } + n := &clusterNode{addr: node1.Addr(), client: client, primary: true} + var got *clusterNode + if p := panicOf1104(func() { got, err = cc.installNode(n) }); p != nil { + panics++ + _ = client.Close() + continue + } + if !errors.Is(err, ErrClosed) || got != nil { + wrongErr++ + } + if !nodesNil1104(cc) { + stored++ + } + if _, derr := client.Do(context.Background(), "PING"); !clientClosed1104(derr) { + open++ // the client was not closed + _ = client.Close() + } + } + if panics != 0 || wrongErr != 0 || stored != 0 || open != 0 { + t.Errorf("installNode after Close: %d of %d panicked, %d did not return (nil, ErrClosed), %d stored a node, %d left the client open", panics, iters, wrongErr, stored, open) + } + + // On a live cluster a new address is stored and returned, and an address that + // is there already returns the stored node and closes the new client. + node1, _, _ := setupCluster(t) + cc, err := NewClusterClient(ClusterConfig{Addrs: []string{node1.Addr()}, DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = cc.Close() }() + fresh, err := cc.dialNode("127.0.0.1:1") + if err != nil { + t.Fatal(err) + } + n := &clusterNode{addr: "127.0.0.1:1", client: fresh, primary: true} + if got, err := cc.installNode(n); err != nil || got != n { + t.Fatalf("installNode(new address) = %v, %v; want the node itself", got, err) + } + dup, err := cc.dialNode("127.0.0.1:1") + if err != nil { + t.Fatal(err) + } + if got, err := cc.installNode(&clusterNode{addr: "127.0.0.1:1", client: dup}); err != nil || got != n { + t.Fatalf("installNode(known address) = %v, %v; want the stored node", got, err) } - if panics != 0 || dialled != 0 || wrongErr != 0 { - t.Errorf("getOrCreateNode after Close: %d of %d panicked, %d dialled a node, %d did not return ErrClosed", panics, iters, dialled, wrongErr) + if _, derr := dup.Do(context.Background(), "PING"); !clientClosed1104(derr) { + t.Errorf("the duplicate client was not closed: %v", derr) } } diff --git a/driver/redis/sentinel_close_switch_1104_test.go b/driver/redis/sentinel_close_switch_1104_test.go index d6bb5f182..fda4732d4 100644 --- a/driver/redis/sentinel_close_switch_1104_test.go +++ b/driver/redis/sentinel_close_switch_1104_test.go @@ -183,3 +183,37 @@ func TestSentinelSwitchMasterDoesNotDialAfterClose1104(t *testing.T) { t.Errorf("handleSwitchMaster dialled the new master again after Close in %d of %d runs", extra, iters) } } + +// A +switch-master that was already queued when Close ran: watchFailover's +// select between the message and closeCh picks at random, so handleSwitchMaster +// can run after Close, and it dialled the new master once before it looked at +// anything. It must make no dial at all: no ROLE request, nothing installed. The +// test calls it after Close, which is the state the lost pick leaves. +func TestSentinelSwitchMasterAfterCloseMakesNoDial1104(t *testing.T) { + const iters = 50 + dialled, installed := 0, 0 + for range iters { + g := startGatedMaster1104(t, "master") + close(g.release) // a ROLE would be answered at once + master := startFakeMaster(t) + sent := startFakeSentinel(t, master.Addr()) + sc, err := NewSentinelClient(SentinelConfig{MasterName: "mymaster", SentinelAddrs: []string{sent.Addr()}, DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + _ = sc.Close() + sc.handleSwitchMaster(switchPayload1104(g.Addr())) + // handleSwitchMaster dials synchronously, so a dial has reached ROLE by now. + if g.roles.Load() != 0 { + dialled++ + } + sc.mu.RLock() + if sc.client != nil { + installed++ + } + sc.mu.RUnlock() + } + if dialled != 0 || installed != 0 { + t.Errorf("handleSwitchMaster on a closed SentinelClient: dialled the new master in %d of %d runs, installed a client in %d", dialled, iters, installed) + } +} From 784d60af2aac91363a222e158087f3a97d5c489c Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 12:36:48 +0100 Subject: [PATCH 6/6] driver/redis: handleSwitchMaster makes no dial for a closed SentinelClient (Refs #1104) A switch message already queued when Close ran can still be picked by watchFailover's select, and the first dial of handleSwitchMaster had no look at the closed flag (the retry dial and the install did). Return at entry. --- driver/redis/sentinel.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/driver/redis/sentinel.go b/driver/redis/sentinel.go index fb00299a4..d387a8935 100644 --- a/driver/redis/sentinel.go +++ b/driver/redis/sentinel.go @@ -454,6 +454,11 @@ var ErrSentinelUnhealthy = errors.New("celeris-redis: sentinel client unhealthy, // fail, the client is marked unhealthy and returns errors until the next // successful +switch-master. func (s *SentinelClient) handleSwitchMaster(payload string) { + // watchFailover's select between a queued message and closeCh picks at + // random, so this can run after Close: make no dial for a closed client. + if s.closed.Load() { + return + } // Format: " " parts := splitFields(payload) if len(parts) < 5 {