diff --git a/driver/redis/cluster.go b/driver/redis/cluster.go index 3a539f80f..64e04446f 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 @@ -236,14 +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 existing, ok := c.nodes[addr]; ok { + if c.nodes == nil { c.mu.Unlock() - _ = client.Close() + _ = n.client.Close() + return nil, ErrClosed + } + if existing, ok := c.nodes[n.addr]; ok { + c.mu.Unlock() + _ = n.client.Close() return existing, nil } - c.nodes[addr] = n + c.nodes[n.addr] = n c.mu.Unlock() return n, nil } @@ -336,8 +351,20 @@ 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() + // 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 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..fa3b859c3 --- /dev/null +++ b/driver/redis/cluster_close_refresh_1104_test.go @@ -0,0 +1,169 @@ +package redis + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/goceleris/celeris/driver/internal/async" +) + +// 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 +} + +// 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 +} + +// 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". It must return ErrClosed and store nothing. +func TestClusterRefreshAfterCloseDoesNotWriteToTheNodes1104(t *testing.T) { + const iters = 30 + panics, wrongErr, stored := 0, 0, 0 + for range iters { + cc, v, _ := closedClusterWithSlots1104(t) + var err error + if p := panicOf1104(func() { err = cc.applyTopology(context.Background(), v) }); p != nil { + panics++ + continue + } + if !errors.Is(err, ErrClosed) { + wrongErr++ + } + if !nodesNil1104(cc) { + stored++ + } + } + 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 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, wrongErr, stored := 0, 0, 0 + for range iters { + cc, _, node1 := closedClusterWithSlots1104(t) + var err error + if p := panicOf1104(func() { _, err = cc.getOrCreateNode(node1.Addr()) }); p != nil { + panics++ + continue + } + 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 _, derr := dup.Do(context.Background(), "PING"); !clientClosed1104(derr) { + t.Errorf("the duplicate client was not closed: %v", derr) + } +} diff --git a/driver/redis/sentinel.go b/driver/redis/sentinel.go index f4d1a14db..d387a8935 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. @@ -449,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 { @@ -459,7 +469,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 @@ -478,10 +488,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 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..fda4732d4 --- /dev/null +++ b/driver/redis/sentinel_close_switch_1104_test.go @@ -0,0 +1,219 @@ +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) + } +} + +// 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) + } +}