diff --git a/driver/redis/cluster.go b/driver/redis/cluster.go index 64e04446f..8d31e385b 100644 --- a/driver/redis/cluster.go +++ b/driver/redis/cluster.go @@ -490,6 +490,11 @@ func (c *ClusterClient) refreshLoop() { for { select { case <-ticker.C: + // closeCh wins a tie with the ticker: select picks at random + // when both are ready, and a refresh after Close dials nodes. + if c.closed.Load() { + return + } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) _ = c.refreshTopology(ctx) cancel() @@ -956,16 +961,41 @@ func (c *ClusterClient) Publish(ctx context.Context, channel, message string) (i } // Subscribe opens a pub/sub connection on any cluster node and subscribes to -// channels (SUBSCRIBE is cluster-wide). See [Client.Subscribe]: when the node is -// found down on the first SUBSCRIBE the PubSub and a nil error are returned and -// the reconnect retries that one node address. +// channels (SUBSCRIBE is node-local, and the cluster delivers every PUBLISH to +// every node). See [Client.Subscribe]: when the node is found down on the first +// SUBSCRIBE the PubSub and a nil error are returned and the reconnect replays +// the channels. +// +// The PubSub follows its subscription: when the node that holds it goes away, +// it subscribes again on another node and keeps delivering to the same +// [PubSub.Channel]. A message published while it switches is lost (a gap), and +// the PubSub itself delivers from one conn at a time: it stops listening to the +// old conn before the new one subscribes. The cluster, though, sends a PUBLISH +// to every node over its bus asynchronously, so a message in flight around the +// switch can in rare cases reach the old and the new node and be delivered +// twice. While the +// PubSub has no conn its control calls ([PubSub.Subscribe] and the others) +// return [ErrPubSubConnDown]; nothing else is surfaced as long as a node can be +// reached, and the PubSub retries with backoff, with no limit, until +// [PubSub.Close]. When the ClusterClient is closed, a PubSub whose conn is down +// (or goes down afterwards) gives up: its Channel() is closed and its control +// calls return [ErrClosed]. A PubSub whose conn stays up after the close (the +// event loop is shared with another Client, or WithEngine) keeps it until +// [PubSub.Close]. func (c *ClusterClient) Subscribe(ctx context.Context, channels ...string) (*PubSub, error) { - // Regular SUBSCRIBE is cluster-wide: pick any node. + // Regular SUBSCRIBE is node-local: pick any node. node := c.anyNode() if node == nil { return nil, errors.New("celeris-redis: no cluster nodes available") } - return node.client.Subscribe(ctx, channels...) + ps, err := c.openPubSub(ctx, node.client, "") + if err != nil { + return nil, err + } + if len(channels) == 0 { + return ps, nil + } + return ps.firstControl(ps.Subscribe(ctx, channels...)) } // SSubscribe opens a pub/sub connection and subscribes to shard channels @@ -974,9 +1004,25 @@ func (c *ClusterClient) Subscribe(ctx context.Context, channels ...string) (*Pub // SHOULD hash to the same slot (use hash tags); mixing slots in one PubSub is // undefined behavior in cluster mode. Like [Client.Subscribe], it returns the // PubSub and a nil error when the conn is found down on the first SSUBSCRIBE: -// the reconnect replays the channels, to the same node address. It does not -// follow a slot move, so a caller that wants the new owner closes the PubSub -// and calls SSubscribe again. +// the reconnect replays the channels. +// +// The PubSub follows its slot: when the owner goes away, or the slot moves (the +// server drops the subscription with an unsolicited sunsubscribe, or answers +// the SSUBSCRIBE with -MOVED), it refreshes the cluster topology, subscribes +// again on the new owner and keeps delivering to the same [PubSub.Channel], +// with the PubSub's reconnect backoff between attempts and no limit. A message +// published while it switches is lost (a gap); one owner serves a slot, so the +// switch cannot deliver a message twice. One PubSub serves one slot, the slot of +// its first channel; a channel of another slot added to it later is not served. +// If another node owns that slot, the SSUBSCRIBE is answered -MOVED, which is not +// reported, and the channel is not subscribed. If the same node owns it, the node +// accepts it and delivers it, until the next reroute of any kind, whose replay +// sends only the channels of the served slot and leaves it out; the move of that +// slot does not reroute the PubSub. If the first channel is removed while a +// channel of another node's slot remains, that channel is not subscribed until +// the next reroute of any kind, whose replay serves the re-selected slot. See +// [ClusterClient.Subscribe] for what the caller sees while it is down and when +// the ClusterClient is closed. func (c *ClusterClient) SSubscribe(ctx context.Context, channels ...string) (*PubSub, error) { if len(channels) == 0 { return nil, errors.New("celeris-redis: SSubscribe requires at least one channel") @@ -989,7 +1035,7 @@ func (c *ClusterClient) SSubscribe(ctx context.Context, channels ...string) (*Pu if node == nil { return nil, errors.New("celeris-redis: no cluster nodes available") } - ps, err := node.client.newPubSub(ctx) + ps, err := c.openPubSub(ctx, node.client, channels[0]) if err != nil { return nil, err } diff --git a/driver/redis/cluster_node.go b/driver/redis/cluster_node.go index 7854b25c1..2eced54a1 100644 --- a/driver/redis/cluster_node.go +++ b/driver/redis/cluster_node.go @@ -10,13 +10,55 @@ type clusterNode struct { client *Client primary bool // failing is set when the node is unreachable; cleared after a successful - // command or topology refresh confirms it is alive. - failing atomic.Bool + // command or topology refresh confirms it is alive. A PubSub dial also + // sets it, and clears only what it set (see failFlag). + failing failFlag // latency stores the last-measured round-trip time in nanoseconds, updated // during topology refresh. Used by RouteByLatency to pick the fastest node. latency atomic.Int64 } +// failFlag is clusterNode.failing: a flag that also remembers whether it was last +// set by a failed PubSub dial. One atomic word holds both bits, so every change +// is one atomic operation and none can be lost to another: +// +// 0 healthy +// failingBit failing, set by the command path or found by a topology refresh +// failingBit|byPubSubBit failing, set by a failed PubSub dial +// +// Store, which the command path and the topology refresh use, takes the node +// over: it replaces the word, so a later successful PubSub dial leaves it alone. +// markPubSub replaces it with the PubSub's own verdict. clearPubSub clears the +// flag only if the word is exactly the PubSub's own mark (one compare and +// swap), so it never erases a mark that changed after it looked. +type failFlag struct { + w atomic.Uint32 +} + +const ( + failingBit uint32 = 1 + byPubSubBit uint32 = 2 +) + +// Load reports whether the node is marked failing. +func (f *failFlag) Load() bool { return f.w.Load()&failingBit != 0 } + +// Store sets the flag as the command path or the topology refresh found it. +func (f *failFlag) Store(v bool) { + if v { + f.w.Store(failingBit) + return + } + f.w.Store(0) +} + +// markPubSub sets the flag because a PubSub dial failed. +func (f *failFlag) markPubSub() { f.w.Store(failingBit | byPubSubBit) } + +// clearPubSub clears the flag after a successful PubSub dial, but only if a +// failed PubSub dial set it and nothing has changed it since. +func (f *failFlag) clearPubSub() { f.w.CompareAndSwap(failingBit|byPubSubBit, 0) } + // Close tears down the node's client. func (n *clusterNode) Close() error { if n.client != nil { diff --git a/driver/redis/conn.go b/driver/redis/conn.go index 2cec6b0c5..d9ed7aa03 100644 --- a/driver/redis/conn.go +++ b/driver/redis/conn.go @@ -761,6 +761,21 @@ func (c *redisConn) releaseResult(req *redisRequest) { putRequest(req) } +// testHookHandshakeRelease, when non-nil, runs for every request handshake gives +// back to the pool. It is nil in production (handshake runs once per conn, so +// the check is off every hot path); a test sets it to see that a handshake that +// failed for any reason but a server error reply gives nothing back (the reader +// may still be filling the request in), and resets it to nil when it is done. +var testHookHandshakeRelease atomic.Pointer[func(req *redisRequest)] + +// releaseHandshakeReq is releaseResult for handshake's requests. +func (c *redisConn) releaseHandshakeReq(req *redisRequest) { + if h := testHookHandshakeRelease.Load(); h != nil { + (*h)(req) + } + c.releaseResult(req) +} + // handshake runs HELLO (with optional fallback to AUTH+SELECT) to negotiate // protocol version and authentication. func (c *redisConn) handshake(ctx context.Context) error { @@ -784,7 +799,7 @@ func (c *redisConn) handshake(ctx context.Context) error { } req, err := c.exec(ctx, helloArgs...) if err == nil { - defer c.releaseResult(req) + defer c.releaseHandshakeReq(req) c.state.proto.Store(int32(targetProto)) // After HELLO OK, select db. if c.cfg.DB != 0 { @@ -794,12 +809,18 @@ func (c *redisConn) handshake(ctx context.Context) error { } return nil } - // Release the request from the failed exec — it may hold pooled Reader - // slices that would otherwise leak. - c.releaseResult(req) // If HELLO was rejected, fall back to legacy auth + select. var rerr *RedisError if errors.As(err, &rerr) { + // The server's error reply was dispatched: dispatch is done with req. + // Any other error (the caller's context ended, the conn closed) leaves + // the request to the garbage collector, as every other caller of exec + // does: the reader goroutine may still be filling it in, and a request + // given back to the pool now could be reused by another conn and + // woken by that late finish with a reply that is not its own. Nothing + // leaks: req.owned is never set in production code, so releaseResult + // only skips a sync.Pool reuse here. + c.releaseHandshakeReq(req) c.state.proto.Store(2) return c.legacyAuth(ctx) } diff --git a/driver/redis/handshake_ctx_ended_887_unix_test.go b/driver/redis/handshake_ctx_ended_887_unix_test.go new file mode 100644 index 000000000..e6487067f --- /dev/null +++ b/driver/redis/handshake_ctx_ended_887_unix_test.go @@ -0,0 +1,126 @@ +//go:build linux || darwin + +package redis + +// A Do whose context ends while a fresh conn's HELLO reply is arriving: exec +// returns the context's error, but the reader goroutine already popped the +// request and is filling it in (state.go dispatch). handshake gave that request +// back to the pool on every failed exec, so another Do could take the same +// object, and the reader's late finish woke it before its own reply came: an +// empty reply with a nil error on a healthy conn, and -race names the two +// writers (found by the review of celeris#887 / PR #1154; the code is from +// April, 5cb0f9b). +// +// The window is the microseconds around the context's end, so the cell is a +// stress test: 1500 rounds with random delays and timeouts, and it is only +// meaningful under -race (-count=20 is the bar; without -race it counts wrong +// replies, which is probabilistic). It is left out of the Windows job, whose +// timer granularity would make it pass without testing anything. +// +// It is meant to run once per CI job. Every round opens a connection, and a +// local -count=20 uses about half of the macOS ephemeral port range (TIME_WAIT); +// run it with a modest -count, and leave a minute between big runs. + +import ( + "bufio" + "context" + "math/rand/v2" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestHandshakeWhoseContextEndedDoesNotRecycleItsRequestWhileTheReaderHoldsIt887(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer func() { _ = c.Close() }() + r := bufio.NewReader(c) + w := bufio.NewWriter(c) + for { + cmd, err := readCommandInto(r, nil) + if err != nil { + return + } + switch strings.ToUpper(cmd[0]) { + case "HELLO": + time.Sleep(time.Duration(rand.IntN(300)) * time.Microsecond) + handleHELLO(w, 3) + case "PING": + writeSimple(w, "PONG") + default: + writeSimple(w, "OK") + } + if w.Flush() != nil { + return + } + } + }(c) + } + }() + addr := ln.Addr().String() + + // A healthy client whose PING replies are checked while the others' + // handshakes are cut short: a recycled request wakes one of them with a + // reply that is not its own. + checker, err := NewClient(addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = checker.Close() }) + var bad atomic.Int64 + stop := make(chan struct{}) + var wg sync.WaitGroup + for g := 0; g < 2; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + v, err := checker.Do(context.Background(), "PING") + if err != nil { + // (A dial error: the loopback runs out of ports under this + // churn on macOS. Only a reply that is not the conn's own + // counts.) + continue + } + if v == nil || string(v.Str) != "PONG" { + if bad.Add(1) <= 3 { + t.Logf("the checker's PING got v=%+v, want PONG", v) + } + } + } + }() + } + for i := 0; i < 1000; i++ { + c, err := NewClient(addr) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(50+rand.IntN(300))*time.Microsecond) + _, _ = c.Do(ctx, "PING") + cancel() + _ = c.Close() + } + close(stop) + wg.Wait() + if n := bad.Load(); n != 0 { + t.Errorf("%d PINGs on a healthy conn got a reply that was not theirs: a request of a failed handshake was recycled while its reader still held it", n) + } +} diff --git a/driver/redis/handshake_release_887_test.go b/driver/redis/handshake_release_887_test.go new file mode 100644 index 000000000..b41e8465d --- /dev/null +++ b/driver/redis/handshake_release_887_test.go @@ -0,0 +1,75 @@ +package redis + +// The contract behind the handshake fix, deterministic and on every OS (the +// stress test next to it, handshake_ctx_ended_887_unix_test.go, is the -race +// guard on linux and darwin): handshake gives its request back to the pool only +// when the server's reply was dispatched. A handshake that ends because the +// caller's context ended leaves the request alone, since the reader goroutine +// may still be filling it in. + +import ( + "bufio" + "context" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestHandshakeGivesBackNothingWhenTheContextEnds887(t *testing.T) { + srv := startFakeRedis(t, func(cmd []string, w *bufio.Writer) { + // HELLO is never answered. + }) + var released atomic.Int64 + hook := func(*redisRequest) { released.Add(1) } + testHookHandshakeRelease.Store(&hook) + t.Cleanup(func() { testHookHandshakeRelease.Store(nil) }) + + c, err := NewClient(srv.Addr()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + if _, err := c.Do(ctx, "PING"); err == nil { + t.Fatal("Do against a server that never answers HELLO returned no error") + } + if n := released.Load(); n != 0 { + t.Errorf("a handshake that ended with the caller's context gave %d request(s) back to the pool while its reader could still hold them", n) + } +} + +// The other side: an error reply (a server that does not know HELLO) was +// dispatched, so its request is given back and the handshake falls back to +// the legacy login. +func TestHandshakeGivesBackTheRequestOfAnErrorReply887(t *testing.T) { + srv := startFakeRedis(t, func(cmd []string, w *bufio.Writer) { + switch strings.ToUpper(cmd[0]) { + case "HELLO": + writeError(w, "ERR unknown command 'HELLO'") + case "PING": + writeSimple(w, "PONG") + default: + writeSimple(w, "OK") + } + }) + var released atomic.Int64 + hook := func(*redisRequest) { released.Add(1) } + testHookHandshakeRelease.Store(&hook) + t.Cleanup(func() { testHookHandshakeRelease.Store(nil) }) + + c, err := NewClient(srv.Addr()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := c.Ping(ctx); err != nil { + t.Fatalf("Ping after the HELLO fallback: %v", err) + } + if n := released.Load(); n != 1 { + t.Errorf("the handshake gave %d request(s) back, want 1 (the error reply's)", n) + } +} diff --git a/driver/redis/pubsub.go b/driver/redis/pubsub.go index 3c881c1bd..795f4e1ff 100644 --- a/driver/redis/pubsub.go +++ b/driver/redis/pubsub.go @@ -3,6 +3,8 @@ package redis import ( "context" "errors" + "strconv" + "strings" "sync" "sync/atomic" "time" @@ -53,7 +55,14 @@ type Message struct { // [Client.Subscribe], [Client.PSubscribe] and [ClusterClient.SSubscribe] apply // the same rule to their first call: they return the PubSub and a nil error, // not [ErrPubSubConnDown]; the PubSub then retries the address of that one -// node (it does not follow a cluster slot move). Any other error +// node. A PubSub of a [ClusterClient] or a [SentinelClient] is the exception: +// its reconnect asks the client for the node to dial each time, so when its +// conn goes down it follows a node that is gone and a sentinel failover, and a +// cluster slot that moved is noticed even though the conn stays up (see +// [ClusterClient.Subscribe] and [ClusterClient.SSubscribe]). When that client +// is closed, a PubSub whose conn is down, or goes down afterwards, gives up and +// its Channel() closes; one whose conn stays up (a loop shared with another +// Client, or WithEngine, keeps it open) keeps it until [PubSub.Close]. Any other error // from the write (for example [ErrQueueFull] on a live connection) means the // command was not sent and nothing replays it before the next reconnect. // [ErrClosed] means the PubSub itself is closed, also when [PubSub.Close] raced @@ -62,6 +71,23 @@ type Message struct { type PubSub struct { client *Client + // route, when not nil, chooses the client every reconnect dials: a cluster + // or sentinel PubSub follows its subscription to another node. Nil (a + // standalone Client) means client, for ever. Set before the PubSub is + // published and never changed. + route pubsubRoute + // connClient is the client whose pool conn came from, so the conn is + // released to that pool and no other. Guarded by mu, set with conn. + connClient *Client + // routeCur is the client the loop tried last (or the one the conn lost + // came from). Owned by the constructor and then by the one running + // reconnect loop, each loop handed over through reconnectingMu. + routeCur *Client + // ctx is cancelled by Close, so a dial or a topology refresh of the + // reconnect loop that is in flight ends with it. + ctx context.Context + cancel context.CancelFunc + // ctlMu keeps a control write and the release of its conn apart // (celeris#929). A conn is written by its file-descriptor number, and the // number goes back to the kernel the moment the conn is closed, so a write @@ -87,15 +113,26 @@ type PubSub struct { subs map[string]struct{} psubs map[string]struct{} shardSubs map[string]struct{} - msgCh chan *Message - closeCh chan struct{} - closed atomic.Bool + // servedSlot is the slot of the shard channels a cluster PubSub serves (the + // slot of the first one added); hasServed says it is set. Guarded by mu; use + // servedSlotLocked. + servedSlot uint16 + hasServed bool + msgCh chan *Message + closeCh chan struct{} + closed atomic.Bool // reconnecting is set while a reconnect goroutine is active. Guards // against double-fires from both onRecv-error and onClose in the same // conn teardown. reconnectingMu sync.Mutex reconnecting bool + // pendingReroute records a trigger that came from a conn that is up (an + // unsolicited sunsubscribe, a -MOVED) while a reconnect loop was running: + // the conn's close hook cannot report it, so the loop's tail starts one + // more. Cleared when the loop installs its conn: whatever the old conn + // said before that is answered by the replay. + pendingReroute bool // The outage state below is guarded by reconnectingMu (celeris#887 items // 29 and 30). An outage starts at the first drop of a conn and ends when a // replayed conn has stayed up for pubsubStableAfter. The backoff runs over @@ -177,7 +214,28 @@ func (ps *PubSub) firstControl(err error) (*PubSub, error) { // the hook to fire on (no t.Parallel), and resets it to nil when it is done. var testHookBeforeBind atomic.Pointer[func(conn *redisConn)] +// testHookReplayed, when non-nil, runs in reconnectLoop after a conn was +// installed and replayed, before the old conn is released and before the loop +// returns (reconnecting is still set, ctlMu is held). It is nil in production; +// a test sets it to hold the loop there, to make a trigger arrive from the new +// conn during the handover or to look at the old conn while it is still open +// (celeris#887 item 31). A test that sets it resets it to nil when it is done. +var testHookReplayed atomic.Pointer[func(ps *PubSub)] + +// testHookBeforeInstall, when non-nil, runs in reconnectLoop after a conn was +// dialled and before the loop takes ctlMu to install it (no lock held). It is +// nil in production; a test sets it to close the PubSub in that window +// (celeris#887 item 31) and resets it to nil when it is done. +var testHookBeforeInstall atomic.Pointer[func(ps *PubSub)] + func (c *Client) newPubSub(ctx context.Context) (*PubSub, error) { + return c.newPubSubRouted(ctx, nil) +} + +// newPubSubRouted is newPubSub for a PubSub that follows its subscription: +// route chooses the client of every reconnect (nil: always c). The first conn +// is dialled on c. +func (c *Client) newPubSubRouted(ctx context.Context, route pubsubRoute) (*PubSub, error) { conn, err := c.pool.acquirePubSub(ctx) if err != nil { return nil, err @@ -185,15 +243,21 @@ func (c *Client) newPubSub(ctx context.Context) (*PubSub, error) { if h := testHookBeforeBind.Load(); h != nil { (*h)(conn) } + psCtx, cancel := context.WithCancel(context.Background()) ps := &PubSub{ - client: c, - conn: conn, - subs: map[string]struct{}{}, - psubs: map[string]struct{}{}, - shardSubs: map[string]struct{}{}, - msgCh: make(chan *Message, defaultPubSubChanBuf), - closeCh: make(chan struct{}), - backoff: async.NewBackoff(50*time.Millisecond, 5*time.Second), + client: c, + route: route, + connClient: c, + routeCur: c, + ctx: psCtx, + cancel: cancel, + conn: conn, + subs: map[string]struct{}{}, + psubs: map[string]struct{}{}, + shardSubs: map[string]struct{}{}, + msgCh: make(chan *Message, defaultPubSubChanBuf), + closeCh: make(chan struct{}), + backoff: async.NewBackoff(50*time.Millisecond, 5*time.Second), // Read once: the goroutines of the PubSub never touch the package // variable, which tests set. stableAfter: pubsubStableAfter, @@ -218,8 +282,10 @@ func (ps *PubSub) bindConn(conn *redisConn) { conn.setPubSubCloseHook(ps.onConnDrop) } -// Channel returns the read-only message stream. It remains open until Close -// is called or the reconnect loop exhausts its retries. +// Channel returns the read-only message stream. It remains open until Close is +// called, or until the [ClusterClient] or [SentinelClient] the PubSub follows is +// closed while its conn is down (the reconnect loop then gives up). The reconnect +// loop has no retry budget: it does not give up on its own. func (ps *PubSub) Channel() <-chan *Message { return ps.msgCh } @@ -282,6 +348,9 @@ func (ps *PubSub) PUnsubscribe(_ context.Context, patterns ...string) error { func (ps *PubSub) SSubscribe(_ context.Context, channels ...string) error { return ps.control("SSUBSCRIBE", channels, func() { for _, ch := range channels { + if len(ps.shardSubs) == 0 { + ps.servedSlot, ps.hasServed = Slot(ch), true + } ps.shardSubs[ch] = struct{}{} } }) @@ -294,11 +363,15 @@ func (ps *PubSub) SUnsubscribe(_ context.Context, channels ...string) error { return ps.control("SUNSUBSCRIBE", channels, func() { if len(channels) == 0 { ps.shardSubs = map[string]struct{}{} + ps.hasServed = false return } for _, ch := range channels { delete(ps.shardSubs, ch) } + if len(ps.shardSubs) == 0 { + ps.hasServed = false + } }) } @@ -376,8 +449,10 @@ func (ps *PubSub) Close() error { return nil } close(ps.closeCh) + // Cut short a dial or a topology refresh of the reconnect loop. + ps.cancel() ps.mu.Lock() - conn := ps.conn + conn, client := ps.conn, ps.connClient ps.conn = nil ps.mu.Unlock() if conn != nil { @@ -389,7 +464,7 @@ func (ps *PubSub) Close() error { // number under it, so none can start after the release either. ps.detach(conn) ps.ctlMu.Lock() - ps.client.pool.releasePubSub(conn) + client.pool.releasePubSub(conn) ps.ctlMu.Unlock() } ps.closeMsgCh() @@ -414,6 +489,7 @@ func (ps *PubSub) takeConn(conn *redisConn) bool { return false } ps.conn = nil + ps.connClient = nil return true } @@ -464,16 +540,26 @@ func (ps *PubSub) deliver(m *Message) bool { // once: harmless. Comparing the dropped conn with ps.conn would avoid that // extra dial and cost a signature change in the hook plumbing and a second // lock on this path. No subscription is lost by the extra reconnect; a message -// published in the swap window (the new conn's SUBSCRIBE is sent without -// waiting for the server's reply, then the old, live conn is released) may be -// lost, and one published just before the old conn is released may arrive -// twice, as on any reconnect. -func (ps *PubSub) onConnDrop(_ error) { +// published in the swap window (the old conn is cut off from the PubSub before +// the new conn is bound, and the new conn's SUBSCRIBE is sent without waiting +// for the server's reply) may be lost. The PubSub delivers from one conn at a +// time, so the swap does not deliver a message twice by itself. +func (ps *PubSub) onConnDrop(_ error) { ps.trigger(false) } + +// trigger starts a reconnect loop unless one is running or the PubSub is +// closed. live is true for a trigger that comes from a conn that is up (the +// server dropped the subscription, or refused it): the close hook will not +// report that one, so when a loop is already running it is remembered +// (pendingReroute) and the loop's tail starts another. +func (ps *PubSub) trigger(live bool) { if ps.closed.Load() { return } ps.reconnectingMu.Lock() if ps.reconnecting { + if live { + ps.pendingReroute = true + } ps.reconnectingMu.Unlock() return } @@ -519,23 +605,36 @@ func (ps *PubSub) downFor() time.Duration { } // reconnectLoop re-dials the pubsub conn and replays subscriptions with -// exponential backoff. It terminates when the pubsub is closed or the backoff -// reaches its cap and a configured retry budget is exhausted (here we keep -// retrying indefinitely, but every iteration honors closeCh). +// exponential backoff. It terminates when the pubsub is closed, or when the +// route says there is nowhere to dial for good (the ClusterClient or +// SentinelClient was closed: the loop gives up and closes the PubSub). There is +// no retry budget: it keeps retrying indefinitely, but every iteration honors +// closeCh. +// +// Each attempt asks the route which client to dial (a standalone PubSub +// always dials its own), so a subscription follows a node that is gone and a +// slot that moved (celeris#887 item 31). The conn it installs replaces the +// one in ps.conn, which may be dead or may still be up (a trigger from a live +// conn): the old conn is cut off from the PubSub before the new one is bound +// and replayed, so a message is delivered by one conn or the other, never +// both. func (ps *PubSub) reconnectLoop() { defer func() { ps.reconnectingMu.Lock() ps.reconnecting = false + again := ps.pendingReroute + ps.pendingReroute = false ps.reconnectingMu.Unlock() // A close hook that fired while reconnecting was set (the new conn // died during the handover) was ignored by onConnDrop, and the hook // fires once: look at the conn now, after the flag is down. The // conn's closed flag is stored before its hook runs, so no drop is - // missed between the two. + // missed between the two. A trigger from a conn that is up (again) is + // not seen by that check, it was recorded for this point. ps.mu.Lock() cur := ps.conn ps.mu.Unlock() - if cur != nil && cur.closed.Load() { + if again || (cur != nil && cur.closed.Load()) { ps.onConnDrop(nil) } }() @@ -552,41 +651,68 @@ func (ps *PubSub) reconnectLoop() { ps.reconnectingMu.Lock() ps.attempt = attempt + 1 ps.reconnectingMu.Unlock() - select { - case <-time.After(delay): - case <-ps.closeCh: + if !ps.sleep(delay) { return } - if ps.closed.Load() { - return + // The first attempt of an outage goes back to the node it lost (a + // blip is the common case); a later one, or a conn that died soon + // after its replay, looks elsewhere. + client, err := ps.pickClient(attempt > 0) + if err != nil { + if errors.Is(err, ErrClosed) { + // The owner of the route is closed: nothing to re-route to, + // ever. Give up the way the PubSub doc says: Channel() closes. + _ = ps.Close() + return + } + continue } + ps.routeCur = client // Give the dial a bounded window so a hard-down server doesn't - // block reconnect forever on a single attempt. - dialCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - conn, err := ps.client.pool.acquirePubSub(dialCtx) + // block reconnect forever on a single attempt. Close cuts it short. + dialCtx, cancel := context.WithTimeout(ps.ctx, 5*time.Second) + conn, err := client.pool.acquirePubSub(dialCtx) cancel() + if ps.route != nil && ps.ctx.Err() == nil { + // (Not after Close: a dial it cut short says nothing of the node.) + ps.route.dialed(client, err == nil) + } if err != nil { continue } + if h := testHookBeforeInstall.Load(); h != nil { + (*h)(ps) + } ps.ctlMu.Lock() ps.mu.Lock() if ps.closed.Load() { ps.mu.Unlock() ps.ctlMu.Unlock() // Never installed: nothing else holds conn. - ps.client.pool.releasePubSub(conn) + client.pool.releasePubSub(conn) return } // Install conn under ctlMu and mu together, and resubscribe on it // before ctlMu is let go: a caller's control write then goes to // conn, after the replay, and Close cannot release conn under the // replay's writes. - old := ps.conn + old, oldClient := ps.conn, ps.connClient + if old != nil { + // Cut the old conn off before the new one is bound: from here on + // only conn delivers. The old conn may still be up (the server + // dropped one subscription of it, or the client follows a master + // switch), and a PUBLISH reaches every node. + old.state.router.clear() + } + ps.reconnectingMu.Lock() + ps.pendingReroute = false + ps.reconnectingMu.Unlock() // Bind first, so a Close that takes conn as soon as it is published // detaches a conn that is already bound, and cannot be re-bound // after its detach. ps.bindConn(conn) ps.conn = conn + ps.connClient = client subs := make([]string, 0, len(ps.subs)) for s := range ps.subs { subs = append(subs, s) @@ -596,7 +722,15 @@ func (ps *PubSub) reconnectLoop() { psubs = append(psubs, p) } ssubs := make([]string, 0, len(ps.shardSubs)) + // A cluster serves one slot per SSUBSCRIBE (-CROSSSLOT otherwise), and + // one PubSub serves one slot: replay the channels of that slot, so a + // channel of another slot cannot take the rest down with it. + served, onlyServed := ps.servedSlotLocked() + onlyServed = onlyServed && ps.route != nil && ps.route.slotted() for s := range ps.shardSubs { + if onlyServed && Slot(s) != served { + continue + } ssubs = append(ssubs, s) } ps.mu.Unlock() @@ -629,20 +763,24 @@ func (ps *PubSub) reconnectLoop() { if ok && conn.closed.Load() { ok = false } + if h := testHookReplayed.Load(); h != nil { + (*h)(ps) + } var failed *redisConn if !ok && ps.takeConn(conn) { failed = conn // else Close took it and releases it } ps.ctlMu.Unlock() // The dropped conn is out of ps.conn, no writer can hold it (writers - // hold ctlMu and read ps.conn under it): release it, once. + // hold ctlMu and read ps.conn under it): release it, once, to the + // pool it came from. if old != nil { ps.detach(old) - ps.client.pool.releasePubSub(old) + oldClient.pool.releasePubSub(old) } if failed != nil { ps.detach(failed) - ps.client.pool.releasePubSub(failed) + client.pool.releasePubSub(failed) } if ok { // No Reset: the backoff starts over when a drop finds this conn @@ -655,3 +793,144 @@ func (ps *PubSub) reconnectLoop() { } } } + +// sleep waits for d and reports whether the loop should go on: false once the +// PubSub is closed. closeCh wins every tie with the timer: a select picks at +// random when both are ready, and an attempt after Close would dial a conn +// nobody owns. +func (ps *PubSub) sleep(d time.Duration) bool { + select { + case <-ps.closeCh: + return false + default: + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ps.closeCh: + return false + case <-t.C: + } + return !ps.closed.Load() +} + +// pickClient returns the client the next dial goes to. moveOn says the last +// client tried did not work out (a failed dial or a replay that did not hold), +// so a route that has a choice picks another. +func (ps *PubSub) pickClient(moveOn bool) (*Client, error) { + if ps.route == nil { + return ps.client, nil + } + ps.mu.Lock() + // The route follows the slot the PubSub serves, by one of its channels. + shard := "" + if slot, ok := ps.servedSlotLocked(); ok { + for ch := range ps.shardSubs { + if Slot(ch) == slot && (shard == "" || ch < shard) { + shard = ch + } + } + } + ps.mu.Unlock() + return ps.route.pick(ps.ctx, shard, ps.routeCur, moveOn) +} + +// servedSlotLocked returns the slot of the shard channels the PubSub serves: +// the slot of the first one added, kept while a channel of that slot is still +// tracked, otherwise the slot of the smallest tracked channel. ok is false when +// there is no shard channel. The caller holds mu. +func (ps *PubSub) servedSlotLocked() (slot uint16, ok bool) { + if len(ps.shardSubs) == 0 { + ps.hasServed = false + return 0, false + } + if ps.hasServed { + for ch := range ps.shardSubs { + if Slot(ch) == ps.servedSlot { + return ps.servedSlot, true + } + } + } + first := "" + for ch := range ps.shardSubs { + if first == "" || ch < first { + first = ch + } + } + ps.servedSlot, ps.hasServed = Slot(first), true + return ps.servedSlot, true +} + +// onShardUnsubscribed is the router's report, from the conn whose router is +// from, of a "sunsubscribe" push for ch. A control call updates the tracked set +// before it writes, so a channel that is still in shardSubs was not removed by +// the caller: the server dropped it (its slot moved to another node) and the +// PubSub subscribes again where the slot lives now. A channel that is gone from +// the set is the ack of the caller's own SUnsubscribe. One for a channel +// outside the served slot is ignored (see below). (A SUnsubscribe(x) +// followed at once by a SSubscribe(x) can make the ack look like the server's: +// one extra reroute, harmless.) +// +// It runs on the event-loop worker: mu and reconnectingMu only, never ctlMu. A +// report from a conn that is not ps.conn any more (one the reroute loop has +// replaced, its report still in flight) is ignored: the replay already answers +// it. mu is held across trigger, as the loop holds it across the install that +// clears pendingReroute, so a report is either before the install (and cleared +// by it) or after it (and ignored). +func (ps *PubSub) onShardUnsubscribed(from *pubsubRouter, ch string) { + if ps.route == nil { + return + } + ps.mu.Lock() + defer ps.mu.Unlock() + if ps.conn == nil || ps.conn.state.router != from { + return + } + // Only the slot the PubSub serves: a channel of another slot that the same + // node happened to accept is dropped from the replay anyway, and its slot + // moving away is no reason to rebuild the subscription that is served. + if served, ok := ps.servedSlotLocked(); ok && Slot(ch) == served { + if _, still := ps.shardSubs[ch]; still { + ps.trigger(true) + } + } +} + +// onControlError is the router's report, from the conn whose router is from, of +// an error reply in pubsub mode. The replies of the control writes are not +// matched to their commands, so the rule is by content, on a PubSub that serves +// shard channels of a cluster: +// - -MOVED for the slot the PubSub serves: the node does not serve it (the +// slot moved, or the topology was stale), so the subscription did not take: +// look for the owner again, with the backoff. A -MOVED for another slot +// answers something else (Redis answers a SUNSUBSCRIBE of a channel it holds +// no subscription for, in a slot it does not serve, with -MOVED; a channel +// of a second slot added to the PubSub is not served) and is ignored. +// - -CLUSTERDOWN: the cluster is between owners; look again, with the backoff. +// +// Anything else (a -CROSSSLOT, an unknown command) is the caller's to fix: +// retrying would loop. Same context and locking as onShardUnsubscribed. +func (ps *PubSub) onControlError(from *pubsubRouter, msg string) { + if ps.route == nil || !ps.route.slotted() { + return + } + ps.mu.Lock() + defer ps.mu.Unlock() + if ps.conn == nil || ps.conn.state.router != from { + return + } + served, ok := ps.servedSlotLocked() + if !ok { + return + } + switch { + case strings.HasPrefix(msg, "MOVED "): + if f := strings.Fields(msg); len(f) >= 2 { + if n, err := strconv.Atoi(f[1]); err == nil && n == int(served) { + ps.trigger(true) + } + } + case strings.HasPrefix(msg, "CLUSTERDOWN"): + ps.trigger(true) + } +} diff --git a/driver/redis/pubsub_reroute_887_test.go b/driver/redis/pubsub_reroute_887_test.go new file mode 100644 index 000000000..de4e061bd --- /dev/null +++ b/driver/redis/pubsub_reroute_887_test.go @@ -0,0 +1,369 @@ +package redis + +// celeris#887 item 31 (decision D3, option A): a cluster PubSub follows its +// subscription. SUBSCRIBE is node-local, so when the node holding it goes away +// the PubSub subscribes again on another node; a sharded subscription follows +// its slot (a gone owner, a slot that moved: the server's unsolicited +// "sunsubscribe", a -MOVED answer to the re-subscribe). The caller keeps +// reading the same Channel(); a message published while the PubSub switches is +// lost (a gap), never delivered twice. + +import ( + "context" + "testing" + "time" +) + +func newRPClusterClient(t *testing.T, cl *rpCluster) *ClusterClient { + t.Helper() + cc, err := NewClusterClient(ClusterConfig{Addrs: cl.addrs(), DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cc.Close() }) + return cc +} + +func rpCtx(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + return ctx +} + +// A regular SUBSCRIBE lives on one node. When that node goes away the PubSub +// subscribes on another one and keeps delivering to the same channel. +func TestClusterSubscribeFollowsALostNode887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "the subscription on a node", 5*time.Second, func() bool { return cl.holder("sub") != nil }) + first := cl.holder("sub") + cl.publish("news", "m1") + poll(t, "m1", 5*time.Second, func() bool { return rc.count("m1") == 1 }) + + first.kill() + poll(t, "the subscription on another node", 10*time.Second, func() bool { + h := cl.holder("sub") + return h != nil && h != first + }) + cl.publish("news", "m2") + poll(t, "m2 on the same channel", 5*time.Second, func() bool { return rc.count("m2") >= 1 }) + time.Sleep(100 * time.Millisecond) + if n := rc.count("m2"); n != 1 { + t.Errorf("m2 delivered %d times, want exactly 1", n) + } + if n := cl.subscribers("sub"); n != 1 { + t.Errorf("%d conns hold the subscription, want 1", n) + } +} + +// A node that is down when Subscribe is called does not fail the call: any +// live node will do for a regular SUBSCRIBE. +func TestClusterSubscribeSkipsADeadNode887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + cl.nodes[0].kill() + for i := 0; i < 15; i++ { + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatalf("call %d: Subscribe: %v", i, err) + } + _ = ps.Close() + } +} + +// A sharded subscription follows its slot when the slot moves: the old owner +// sends "sunsubscribe" and keeps the conn, and the PubSub subscribes on the new +// owner. The old conn is released. +func TestClusterSSubscribeFollowsASlotMove887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" // slot 12182 + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + if from == to { + t.Fatal("test setup: slot already on the target") + } + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + cl.spublish(ch, "m1") + poll(t, "m1", 5*time.Second, func() bool { return rc.count("m1") == 1 }) + + cl.moveSlot(slot, to, 0) + poll(t, "SSUBSCRIBE on the new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + cl.spublish(ch, "m2") + poll(t, "m2 on the same channel", 5*time.Second, func() bool { return rc.count("m2") == 1 }) + poll(t, "the conn to the old owner released", 5*time.Second, func() bool { + return from.conn(func(*rpConn) bool { return true }) == nil || from.subscribers("ssub") == 0 + }) + if n := cl.subscribers("ssub"); n != 1 { + t.Errorf("%d conns hold the shard subscription, want 1", n) + } +} + +// The owner of a slot goes away and a replica takes its slots a little later: +// until then CLUSTER SLOTS still names the dead node. The PubSub keeps trying +// with backoff and ends on the new owner. +func TestClusterSSubscribeFollowsAFailover887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + from := cl.owner(int(Slot(ch))) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + + from.kill() + time.Sleep(400 * time.Millisecond) + cl.failover(from, to, 0) + poll(t, "SSUBSCRIBE on the new owner", 15*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + cl.spublish(ch, "m1") + poll(t, "m1", 5*time.Second, func() bool { return rc.count("m1") == 1 }) +} + +// The slot moves but the nodes still answer CLUSTER SLOTS with the old table +// for a while. The re-subscribe goes to the old owner and is answered -MOVED, +// which used to be dropped in pubsub mode (the subscription was dead and +// nothing said so). The MOVED reroutes again, with backoff, until the table +// catches up. +func TestClusterSSubscribeMovedAnswerReroutes887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + + cl.moveSlot(slot, to, 3) + poll(t, "SSUBSCRIBE on the new owner", 15*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + cl.spublish(ch, "m1") + poll(t, "m1", 5*time.Second, func() bool { return rc.count("m1") == 1 }) + if got := from.subCmds.Load(); got < 2 { + t.Errorf("the old owner saw %d SSUBSCRIBE commands, want the replay to have hit it (stale table) too", got) + } +} + +// Two PubSubs on different nodes, one of the nodes goes away: only that one +// moves. The other keeps its conn and gets every message once. +func TestClusterTwoSubscriptionsOneLost887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + var ps1, ps2 *PubSub + var n1, n2 *rpNode + for try := 0; try < 40; try++ { + a, err := cc.Subscribe(rpCtx(t), "one") + if err != nil { + t.Fatal(err) + } + poll(t, "first subscription", 5*time.Second, func() bool { return cl.subscribers("sub") == 1 }) + h1 := cl.holder("sub") + b, err := cc.Subscribe(rpCtx(t), "two") + if err != nil { + t.Fatal(err) + } + poll(t, "second subscription", 5*time.Second, func() bool { return cl.subscribers("sub") == 2 }) + var h2 *rpNode + for _, nd := range cl.nodes { + if nd != h1 && nd.subscribers("sub") == 1 { + h2 = nd + } + } + if h2 != nil { + ps1, ps2, n1, n2 = a, b, h1, h2 + break + } + _ = a.Close() + _ = b.Close() + poll(t, "both closed", 5*time.Second, func() bool { return cl.subscribers("sub") == 0 }) + } + if ps1 == nil { + t.Fatal("could not place the two subscriptions on different nodes") + } + t.Cleanup(func() { _ = ps1.Close(); _ = ps2.Close() }) + rc1, rc2 := collect(ps1), collect(ps2) + conn2 := n2.conn(func(c *rpConn) bool { c.mu.Lock(); defer c.mu.Unlock(); return len(c.subs) > 0 }) + + n1.kill() + poll(t, "the first subscription on a live node", 10*time.Second, func() bool { + h := cl.holder("sub") + return cl.subscribers("sub") == 2 && h != nil && h != n1 + }) + cl.publish("one", "a") + cl.publish("two", "b") + poll(t, "both messages", 5*time.Second, func() bool { return rc1.count("a") >= 1 && rc2.count("b") >= 1 }) + time.Sleep(100 * time.Millisecond) + if rc1.count("a") != 1 || rc2.count("b") != 1 { + t.Errorf("deliveries: one/a=%d two/b=%d, want 1 and 1", rc1.count("a"), rc2.count("b")) + } + conn2.mu.Lock() + still := len(conn2.subs) + conn2.mu.Unlock() + if still != 1 { + t.Errorf("the subscription that was not lost was disturbed (its conn has %d subscriptions)", still) + } +} + +// A server that moves the slot away again at every SSUBSCRIBE: the PubSub +// keeps trying, but follows the existing backoff (50 ms, 100 ms, ... doubling), +// it does not spin. +func TestClusterSSubscribeFlappingStaysInsideTheBackoff887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + owner := cl.owner(int(Slot(ch))) + var subs func() int64 + owner.setOnSSub(func(n *rpNode, c *rpConn, ch string) { + // The slot "moves" right after every subscription: the owner takes + // the subscription back with a sunsubscribe, and stays the owner. + n.sunsubscribeSlot(int(Slot(ch))) + }) + subs = owner.subCmds.Load + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + time.Sleep(2500 * time.Millisecond) + n := subs() + t.Logf("%d SSUBSCRIBE commands in 2.5 s", n) + // 50+100+200+400+800+1600 ms of delay: about 7 attempts. A loop that + // starts over from the base delay would make about 40. + if n > 12 { + t.Errorf("%d SSUBSCRIBEs in 2.5 s: the reroute did not back off", n) + } + if n < 3 { + t.Errorf("only %d SSUBSCRIBEs in 2.5 s: the unsolicited sunsubscribe did not make the PubSub subscribe again", n) + } +} + +// The old conn is cut off from the PubSub before the new conn is bound, so a +// message the cluster delivers on both while the PubSub switches reaches the +// caller once. Here the new node publishes a probe on a channel the old conn +// is still subscribed to, right after it acks the replay. +func TestClusterRerouteDeliversASwitchWindowMessageOnce887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + if err := ps.Subscribe(rpCtx(t), "x"); err != nil { + t.Fatal(err) + } + rc := collect(ps) + poll(t, "both subscriptions on the owner", 5*time.Second, func() bool { + return from.subscribers("ssub") == 1 && from.subscribers("sub") == 1 + }) + to.setAfterSub(func(n *rpNode, c *rpConn, verb string) { + if verb == "SUBSCRIBE" { + cl.publish("x", "probe") // reaches the old conn (still open) and the new one + } + }) + cl.moveSlot(slot, to, 0) + poll(t, "the probe", 10*time.Second, func() bool { return rc.count("probe") >= 1 }) + time.Sleep(300 * time.Millisecond) + if n := rc.count("probe"); n != 1 { + t.Errorf("the probe was delivered %d times, want 1 (a gap is allowed, a duplicate is not)", n) + } +} + +// A conn that stalls its HELLO (a node that accepts and says nothing) must not +// keep the reroute goroutine alive after Close: the dial is cut short. +func TestClusterPubSubCloseCutsAStalledRerouteDial887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + poll(t, "the subscription", 5*time.Second, func() bool { return cl.holder("sub") != nil }) + first := cl.holder("sub") + for _, nd := range cl.nodes { + if nd != first { + nd.mu.Lock() + nd.hang = true + nd.mu.Unlock() + } + } + first.kill() + // The reroute dials the other nodes, which never answer. + time.Sleep(600 * time.Millisecond) + start := time.Now() + _ = ps.Close() + if d := time.Since(start); d > time.Second { + t.Errorf("Close took %v", d) + } + poll(t, "the reroute loop to end", 2*time.Second, func() bool { return rpLoopIdle(ps) }) +} + +// Close races a reroute at every point of it, and the conns are all released. +func TestClusterPubSubCloseRacesAReroute887(t *testing.T) { + for i := 0; i < 60; i++ { + cl := newRPCluster(t, 3) + cc, err := NewClusterClient(ClusterConfig{Addrs: cl.addrs(), DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + poll(t, "the subscription", 5*time.Second, func() bool { return cl.holder("sub") != nil }) + first := cl.holder("sub") + first.kill() + time.Sleep(time.Duration(i) * 4 * time.Millisecond) // 0 .. 240 ms: before, between and after the attempts + _ = ps.Close() + _ = cc.Close() + for _, nd := range cl.nodes { + nd.kill() + } + poll(t, "no reroute loop", 3*time.Second, func() bool { return rpLoopIdle(ps) }) + } +} + +// A PubSub that is not closed by its owner is closed with the ClusterClient: the +// reroute has nowhere to go, the loop gives up, and Channel() is closed. +func TestClusterCloseEndsThePubSubReroute887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "the subscription", 5*time.Second, func() bool { return cl.holder("sub") != nil }) + cl.holder("sub").kill() + _ = cc.Close() + poll(t, "Channel() closed", 5*time.Second, func() bool { return rc.closed.Load() }) + poll(t, "no reroute loop", 2*time.Second, func() bool { return rpLoopIdle(ps) }) +} diff --git a/driver/redis/pubsub_reroute_fake_887_test.go b/driver/redis/pubsub_reroute_fake_887_test.go new file mode 100644 index 000000000..236e6acb4 --- /dev/null +++ b/driver/redis/pubsub_reroute_fake_887_test.go @@ -0,0 +1,638 @@ +package redis + +// A fake redis cluster for the pub/sub re-routing tests (celeris#887 item 31). +// It speaks just enough RESP to host PubSub conns: HELLO, CLUSTER SLOTS, +// (P/S)SUBSCRIBE and the unsubscribes with their acks, PUBLISH and SPUBLISH +// fan-out, and the three things a real Redis 7 cluster does that the tests +// need: a node that goes away (listener and conns closed), a slot that moves +// (the old owner sends "sunsubscribe 0" to the clients subscribed to a +// channel of that slot and keeps their conns), and a node that is not the +// owner of a slot answering SSUBSCRIBE with -MOVED. + +import ( + "bufio" + "fmt" + "net" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +type rpCluster struct { + t *testing.T + nodes []*rpNode + + mu sync.Mutex + cur [16384]*rpNode // current owner of every slot + old [16384]*rpNode // what a stale CLUSTER SLOTS reply still shows + // stale is how many more CLUSTER SLOTS replies show old, not cur. + stale int + // slotsReplies counts the CLUSTER SLOTS replies given. + slotsReplies atomic.Int64 +} + +type rpNode struct { + cl *rpCluster + ln net.Listener + addr string + + mu sync.Mutex + conns map[*rpConn]struct{} + dead bool + accepted int + // hang makes the node accept conns and never answer them (a blackholed + // host: a dial that waits for the HELLO reply). + hang bool + // afterSub, when set, runs on the serve goroutine after the ack of a + // (P/S)SUBSCRIBE was flushed, with no lock held. Guarded by mu; set it + // with setAfterSub. + afterSub func(n *rpNode, c *rpConn, verb string) + // onSSub, when set, runs after an SSUBSCRIBE of ch was acked. Guarded by + // mu; set it with setOnSSub. + onSSub func(n *rpNode, c *rpConn, ch string) + // subCmds counts SUBSCRIBE/PSUBSCRIBE/SSUBSCRIBE commands received. + subCmds atomic.Int64 + // ssubErrs are error replies the next SSUBSCRIBE commands get, one each, + // before the node answers them normally. Guarded by mu. + ssubErrs []string +} + +func (n *rpNode) setAfterSub(f func(n *rpNode, c *rpConn, verb string)) { + n.mu.Lock() + n.afterSub = f + n.mu.Unlock() +} + +func (n *rpNode) setOnSSub(f func(n *rpNode, c *rpConn, ch string)) { + n.mu.Lock() + n.onSSub = f + n.mu.Unlock() +} + +func (n *rpNode) setSSubErrs(errs ...string) { + n.mu.Lock() + n.ssubErrs = errs + n.mu.Unlock() +} + +type rpConn struct { + nc net.Conn + w *bufio.Writer + wm sync.Mutex + proto int + + mu sync.Mutex + subs map[string]struct{} + psubs map[string]struct{} + ssubs map[string]struct{} +} + +func newRPCluster(t *testing.T, n int) *rpCluster { + t.Helper() + cl := &rpCluster{t: t} + for i := 0; i < n; i++ { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + nd := &rpNode{cl: cl, ln: ln, addr: ln.Addr().String(), conns: map[*rpConn]struct{}{}} + cl.nodes = append(cl.nodes, nd) + } + for s := 0; s < 16384; s++ { + o := cl.nodes[s*n/16384] + cl.cur[s] = o + cl.old[s] = o + } + for _, nd := range cl.nodes { + go nd.accept() + } + t.Cleanup(func() { + for _, nd := range cl.nodes { + nd.kill() + } + }) + return cl +} + +func (cl *rpCluster) addrs() []string { + out := make([]string, len(cl.nodes)) + for i, n := range cl.nodes { + out[i] = n.addr + } + return out +} + +// owner returns the current owner of slot. +func (cl *rpCluster) owner(slot int) *rpNode { + cl.mu.Lock() + defer cl.mu.Unlock() + return cl.cur[slot] +} + +// moveSlot gives slot to to. The old owner tells its subscribers of channels +// in that slot with "sunsubscribe 0" and keeps their conns, as Redis 7 +// does. staleReplies is how many CLUSTER SLOTS replies still show the old +// table. +func (cl *rpCluster) moveSlot(slot int, to *rpNode, staleReplies int) { + cl.mu.Lock() + from := cl.cur[slot] + cl.old = cl.cur + cl.cur[slot] = to + cl.stale = staleReplies + cl.mu.Unlock() + from.sunsubscribeSlot(slot) +} + +// failover gives every slot of from to to, as a promoted replica does. +func (cl *rpCluster) failover(from, to *rpNode, staleReplies int) { + cl.mu.Lock() + cl.old = cl.cur + for s := range cl.cur { + if cl.cur[s] == from { + cl.cur[s] = to + } + } + cl.stale = staleReplies + cl.mu.Unlock() +} + +// publish is PUBLISH: every live node delivers to its subscribers of ch. +// It returns the number of deliveries. +func (cl *rpCluster) publish(ch, payload string) int { + n := 0 + for _, nd := range cl.nodes { + n += nd.deliver("message", ch, payload, func(c *rpConn) bool { + c.mu.Lock() + defer c.mu.Unlock() + _, ok := c.subs[ch] + return ok + }) + } + return n +} + +// spublish is SPUBLISH: the owner of the slot delivers to its subscribers. +func (cl *rpCluster) spublish(ch, payload string) int { + return cl.owner(int(Slot(ch))).deliver("smessage", ch, payload, func(c *rpConn) bool { + c.mu.Lock() + defer c.mu.Unlock() + _, ok := c.ssubs[ch] + return ok + }) +} + +// subscribers counts the live conns of every node that hold any subscription +// of the kind ("sub", "psub", "ssub"). +func (cl *rpCluster) subscribers(kind string) int { + n := 0 + for _, nd := range cl.nodes { + n += nd.subscribers(kind) + } + return n +} + +// liveConns counts every open conn of every node. +func (cl *rpCluster) liveConns() int { + n := 0 + for _, nd := range cl.nodes { + nd.mu.Lock() + n += len(nd.conns) + nd.mu.Unlock() + } + return n +} + +// holder returns the node holding a conn subscribed with kind, or nil. +func (cl *rpCluster) holder(kind string) *rpNode { + for _, nd := range cl.nodes { + if nd.subscribers(kind) > 0 { + return nd + } + } + return nil +} + +func (n *rpNode) accept() { + for { + nc, err := n.ln.Accept() + if err != nil { + return + } + n.mu.Lock() + if n.dead { + n.mu.Unlock() + _ = nc.Close() + return + } + n.accepted++ + hang := n.hang + c := &rpConn{nc: nc, w: bufio.NewWriter(nc), proto: 2, + subs: map[string]struct{}{}, psubs: map[string]struct{}{}, ssubs: map[string]struct{}{}} + n.conns[c] = struct{}{} + n.mu.Unlock() + if hang { + continue + } + go n.serve(c) + } +} + +// kill takes the node away: the listener and every conn close. +func (n *rpNode) kill() { + n.mu.Lock() + n.dead = true + conns := n.conns + n.conns = map[*rpConn]struct{}{} + n.mu.Unlock() + _ = n.ln.Close() + for c := range conns { + _ = c.nc.Close() + } +} + +func (n *rpNode) acceptedConns() int { + n.mu.Lock() + defer n.mu.Unlock() + return n.accepted +} + +func (n *rpNode) forget(c *rpConn) { + n.mu.Lock() + delete(n.conns, c) + n.mu.Unlock() +} + +func (n *rpNode) subscribers(kind string) int { + n.mu.Lock() + cs := make([]*rpConn, 0, len(n.conns)) + for c := range n.conns { + cs = append(cs, c) + } + n.mu.Unlock() + cnt := 0 + for _, c := range cs { + c.mu.Lock() + var l int + switch kind { + case "sub": + l = len(c.subs) + case "psub": + l = len(c.psubs) + default: + l = len(c.ssubs) + } + c.mu.Unlock() + if l > 0 { + cnt++ + } + } + return cnt +} + +func (n *rpNode) conn(pred func(*rpConn) bool) *rpConn { + n.mu.Lock() + defer n.mu.Unlock() + for c := range n.conns { + if pred(c) { + return c + } + } + return nil +} + +func (n *rpNode) deliver(kind, ch, payload string, match func(*rpConn) bool) int { + n.mu.Lock() + cs := make([]*rpConn, 0, len(n.conns)) + for c := range n.conns { + cs = append(cs, c) + } + n.mu.Unlock() + cnt := 0 + for _, c := range cs { + if !match(c) { + continue + } + c.wm.Lock() + c.pushStrs(kind, ch, payload) + _ = c.w.Flush() + c.wm.Unlock() + cnt++ + } + return cnt +} + +// sunsubscribeSlot is what a node does when it stops owning slot: every +// subscriber of a channel of the slot gets "sunsubscribe 0" and keeps +// its conn. +func (n *rpNode) sunsubscribeSlot(slot int) { + n.mu.Lock() + cs := make([]*rpConn, 0, len(n.conns)) + for c := range n.conns { + cs = append(cs, c) + } + n.mu.Unlock() + for _, c := range cs { + c.mu.Lock() + var chs []string + for ch := range c.ssubs { + if int(Slot(ch)) == slot { + chs = append(chs, ch) + delete(c.ssubs, ch) + } + } + left := len(c.ssubs) + c.mu.Unlock() + for _, ch := range chs { + c.wm.Lock() + c.pushAck("sunsubscribe", ch, int64(left)) + _ = c.w.Flush() + c.wm.Unlock() + } + } +} + +// pushStrs writes a push of bulk strings (RESP3 ">", RESP2 array). The caller +// holds wm. +func (c *rpConn) pushStrs(items ...string) { + if c.proto == 3 { + writePush(c.w, items...) + return + } + writeArrayBulks(c.w, items...) +} + +// pushAck writes a subscribe-family ack: [kind, channel, count]. The caller +// holds wm. +func (c *rpConn) pushAck(kind, ch string, count int64) { + if c.proto == 3 { + _ = c.w.WriteByte('>') + } else { + _ = c.w.WriteByte('*') + } + _, _ = c.w.WriteString("3\r\n") + writeBulk(c.w, kind) + if ch == "" { + writeNullBulk(c.w) + } else { + writeBulk(c.w, ch) + } + writeInt(c.w, count) +} + +func (n *rpNode) serve(c *rpConn) { + defer func() { + _ = c.nc.Close() + n.forget(c) + }() + r := bufio.NewReader(c.nc) + var args []string + for { + cmd, err := readCommandInto(r, args[:0]) + if err != nil { + return + } + args = cmd + if len(cmd) == 0 { + continue + } + verb := strings.ToUpper(cmd[0]) + c.wm.Lock() + after := n.handle(c, verb, cmd) + err = c.w.Flush() + c.wm.Unlock() + if err != nil { + return + } + if after != nil { + after() + } + } +} + +// handle answers one command. The caller holds c.wm and flushes. The returned +// func, when not nil, runs after the flush with no lock held. +func (n *rpNode) handle(c *rpConn, verb string, cmd []string) func() { + switch verb { + case "HELLO": + proto := 2 + if len(cmd) > 1 && cmd[1] == "3" { + proto = 3 + } + c.proto = proto + handleHELLO(c.w, proto) + case "PING": + writeSimple(c.w, "PONG") + case "CLUSTER": + if len(cmd) >= 2 && strings.EqualFold(cmd[1], "SLOTS") { + n.cl.writeSlots(c.w) + return nil + } + writeError(c.w, "ERR unknown cluster subcommand") + case "SUBSCRIBE", "PSUBSCRIBE": + n.subCmds.Add(1) + c.mu.Lock() + set, kind := c.subs, "subscribe" + if verb == "PSUBSCRIBE" { + set, kind = c.psubs, "psubscribe" + } + for _, ch := range cmd[1:] { + set[ch] = struct{}{} + c.pushAck(kind, ch, int64(len(set))) + } + c.mu.Unlock() + n.mu.Lock() + f := n.afterSub + n.mu.Unlock() + if f != nil { + return func() { f(n, c, verb) } + } + case "UNSUBSCRIBE", "PUNSUBSCRIBE": + c.mu.Lock() + set, kind := c.subs, "unsubscribe" + if verb == "PUNSUBSCRIBE" { + set, kind = c.psubs, "punsubscribe" + } + names := cmd[1:] + if len(names) == 0 { + for k := range set { + names = append(names, k) + } + } + for _, ch := range names { + delete(set, ch) + c.pushAck(kind, ch, int64(len(set))) + } + c.mu.Unlock() + case "ROLE": + writeArrayHeader(c.w, 3) + writeBulk(c.w, "master") + writeInt(c.w, 0) + writeArrayHeader(c.w, 0) + case "SSUBSCRIBE": + n.subCmds.Add(1) + n.mu.Lock() + var forced string + if len(n.ssubErrs) > 0 { + forced, n.ssubErrs = n.ssubErrs[0], n.ssubErrs[1:] + } + n.mu.Unlock() + if forced != "" { + writeError(c.w, forced) + return nil + } + for _, ch := range cmd[1:] { + if Slot(ch) != Slot(cmd[1]) { + writeError(c.w, "CROSSSLOT Keys in request don't hash to the same slot") + return nil + } + } + for _, ch := range cmd[1:] { + slot := int(Slot(ch)) + if o := n.cl.owner(slot); o != n { + writeError(c.w, fmt.Sprintf("MOVED %d %s", slot, o.addr)) + return nil + } + } + c.mu.Lock() + for _, ch := range cmd[1:] { + c.ssubs[ch] = struct{}{} + c.pushAck("ssubscribe", ch, int64(len(c.ssubs))) + } + c.mu.Unlock() + chs := append([]string(nil), cmd[1:]...) + n.mu.Lock() + f, g := n.afterSub, n.onSSub + n.mu.Unlock() + if f != nil || g != nil { + return func() { + if f != nil { + f(n, c, verb) + } + if g != nil { + for _, ch := range chs { + g(n, c, ch) + } + } + } + } + case "SUNSUBSCRIBE": + // Like Redis 7, a node that does not serve the slot of a channel it + // holds no subscription for answers -MOVED. + for _, ch := range cmd[1:] { + c.mu.Lock() + _, held := c.ssubs[ch] + c.mu.Unlock() + if slot := int(Slot(ch)); !held && n.cl.owner(slot) != n { + writeError(c.w, fmt.Sprintf("MOVED %d %s", slot, n.cl.owner(slot).addr)) + return nil + } + } + c.mu.Lock() + names := cmd[1:] + if len(names) == 0 { + for k := range c.ssubs { + names = append(names, k) + } + } + for _, ch := range names { + delete(c.ssubs, ch) + c.pushAck("sunsubscribe", ch, int64(len(c.ssubs))) + } + c.mu.Unlock() + default: + writeSimple(c.w, "OK") + } + return nil +} + +// writeSlots answers CLUSTER SLOTS from the current table, or the old one +// while stale replies are owed. +func (cl *rpCluster) writeSlots(w *bufio.Writer) { + cl.slotsReplies.Add(1) + cl.mu.Lock() + tab := cl.cur + if cl.stale > 0 { + cl.stale-- + tab = cl.old + } + cl.mu.Unlock() + type rng struct { + lo, hi int + o *rpNode + } + var rs []rng + for s := 0; s < 16384; s++ { + if tab[s] == nil { + continue + } + if k := len(rs) - 1; k >= 0 && rs[k].o == tab[s] && rs[k].hi == s-1 { + rs[k].hi = s + continue + } + rs = append(rs, rng{s, s, tab[s]}) + } + writeArrayHeader(w, len(rs)) + for _, r := range rs { + host, port, _ := splitHostPort(r.o.addr) + p, _ := strconv.Atoi(port) + writeArrayHeader(w, 3) + writeInt(w, int64(r.lo)) + writeInt(w, int64(r.hi)) + writeArrayHeader(w, 3) + writeBulk(w, host) + writeInt(w, int64(p)) + writeBulk(w, "node-"+r.o.addr) + } +} + +// ---- collectors and polling ---- + +// rpCollector reads a PubSub's channel and counts each payload. +type rpCollector struct { + mu sync.Mutex + counts map[string]int + closed atomic.Bool +} + +func collect(ps *PubSub) *rpCollector { + rc := &rpCollector{counts: map[string]int{}} + go func() { + for m := range ps.Channel() { + rc.mu.Lock() + rc.counts[m.Payload]++ + rc.mu.Unlock() + } + rc.closed.Store(true) + }() + return rc +} + +func (rc *rpCollector) count(payload string) int { + rc.mu.Lock() + defer rc.mu.Unlock() + return rc.counts[payload] +} + +// poll waits for cond, failing the test after d. +func poll(t *testing.T, what string, d time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out after %v waiting for %s", d, what) + } + time.Sleep(5 * time.Millisecond) + } +} + +// rpLoopIdle reports whether ps has no reconnect loop running. After Close the +// loop of this PubSub must end (checking the PubSub's own flag, not every +// goroutine of the process, keeps other tests' leftovers out of it). +func rpLoopIdle(ps *PubSub) bool { + ps.reconnectingMu.Lock() + defer ps.reconnectingMu.Unlock() + return !ps.reconnecting +} diff --git a/driver/redis/pubsub_reroute_internal_887_test.go b/driver/redis/pubsub_reroute_internal_887_test.go new file mode 100644 index 000000000..dfb2b65b5 --- /dev/null +++ b/driver/redis/pubsub_reroute_internal_887_test.go @@ -0,0 +1,343 @@ +package redis + +// Cells of celeris#887 item 31 that need the PubSub's unexported route and +// hooks: what is and is not a trigger (RESP3 and RESP2), and a trigger that +// arrives during the handover. + +import ( + "context" + "strconv" + "sync/atomic" + "testing" + "time" +) + +// sameClientRoute is a route that always answers with the same client: it +// makes a PubSub of a plain Client behave like a cluster PubSub for the +// triggers, so the cells below run on one fake node. +type sameClientRoute struct{ c *Client } + +func (r sameClientRoute) pick(context.Context, string, *Client, bool) (*Client, error) { + return r.c, nil +} +func (sameClientRoute) dialed(*Client, bool) {} +func (sameClientRoute) slotted() bool { return true } + +// routedPubSub opens a RESP2 PubSub on the only node of cl with a route. +func routedPubSub(t *testing.T, cl *rpCluster, resp2 bool) *PubSub { + t.Helper() + var opts []Option + if resp2 { + opts = append(opts, WithForceRESP2()) + } + c, err := NewClient(cl.nodes[0].addr, opts...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ps, err := c.newPubSubRouted(rpCtx(t), sameClientRoute{c}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + return ps +} + +// An unsolicited sunsubscribe of a channel the PubSub still wants makes it +// subscribe again, on RESP3 pushes and on RESP2 arrays alike. +func TestPubSubSunsubscribeOfAWantedChannelResubscribes887(t *testing.T) { + for _, resp2 := range []bool{false, true} { + name := "resp3" + if resp2 { + name = "resp2" + } + t.Run(name, func(t *testing.T) { + cl := newRPCluster(t, 1) + ps := routedPubSub(t, cl, resp2) + if err := ps.SSubscribe(rpCtx(t), "foo"); err != nil { + t.Fatal(err) + } + node := cl.nodes[0] + // (The fake counts a command when it receives it, before it records the + // subscription: wait for both, or the sunsubscribe below finds nothing.) + poll(t, "the first SSUBSCRIBE", 5*time.Second, func() bool { return node.subCmds.Load() == 1 && node.subscribers("ssub") == 1 }) + node.sunsubscribeSlot(int(Slot("foo"))) + poll(t, "the second SSUBSCRIBE", 5*time.Second, func() bool { return node.subscribers("ssub") == 1 && node.subCmds.Load() == 2 }) + }) + } +} + +// The ack of the PubSub's own SUnsubscribe is not a trigger. +func TestPubSubSunsubscribeAckIsNotATrigger887(t *testing.T) { + cl := newRPCluster(t, 1) + ps := routedPubSub(t, cl, false) + node := cl.nodes[0] + if err := ps.SSubscribe(rpCtx(t), "foo"); err != nil { + t.Fatal(err) + } + poll(t, "the SSUBSCRIBE", 5*time.Second, func() bool { return node.subscribers("ssub") == 1 }) + accepted := node.acceptedConns() + if err := ps.SUnsubscribe(rpCtx(t), "foo"); err != nil { + t.Fatal(err) + } + poll(t, "the SUNSUBSCRIBE", 5*time.Second, func() bool { return node.subscribers("ssub") == 0 }) + time.Sleep(400 * time.Millisecond) + if n := node.subCmds.Load(); n != 1 { + t.Errorf("%d SSUBSCRIBEs, want 1: the ack of the caller's own SUnsubscribe made the PubSub subscribe again", n) + } + if n := node.acceptedConns(); n != accepted { + t.Errorf("the PubSub dialled %d more conns after its own SUnsubscribe", n-accepted) + } +} + +// -MOVED and -CLUSTERDOWN answers to an SSUBSCRIBE are triggers (the +// subscription did not take); a -CROSSSLOT is the caller's mistake and is not +// retried. +func TestPubSubControlErrorTriggers887(t *testing.T) { + for _, tc := range []struct { + errReply string + retry bool + }{ + {"MOVED 12182 127.0.0.1:1", true}, + {"CLUSTERDOWN The cluster is down", true}, + {"CROSSSLOT Keys in request don't hash to the same slot", false}, + {"ERR unknown command 'ssubscribe'", false}, + } { + t.Run(tc.errReply, func(t *testing.T) { + cl := newRPCluster(t, 1) + node := cl.nodes[0] + node.setSSubErrs(tc.errReply) + ps := routedPubSub(t, cl, false) + if err := ps.SSubscribe(rpCtx(t), "foo"); err != nil { + t.Fatal(err) + } + if tc.retry { + poll(t, "the second SSUBSCRIBE after the error", 5*time.Second, func() bool { return node.subscribers("ssub") == 1 }) + return + } + time.Sleep(600 * time.Millisecond) + if n := node.subCmds.Load(); n != 1 { + t.Errorf("%d SSUBSCRIBEs after %q, want 1: a caller's error must not loop", n, tc.errReply) + } + }) + } +} + +// A trigger from the new conn that arrives while the reroute loop is still +// running (the handover) is not lost: the conn is up, so no close hook says +// it, and the loop's tail starts one more reroute. Here the slot moves a +// second time right after the replay on the first new owner; the hook holds the +// loop until the unsolicited sunsubscribe has reached the PubSub. +func TestClusterRerouteTriggerDuringTheHandoverIsKept887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + slot := int(Slot(ch)) + a := cl.owner(slot) + b, c := cl.nodes[0], cl.nodes[1] + if a == b || a == c { + t.Fatal("test setup: the owner is one of the targets") + } + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return a.subscribers("ssub") == 1 }) + + var held atomic.Int64 + hook := func(p *PubSub) { + if p != ps { + return + } + if held.Add(1) > 1 { + return + } + // Hold the first loop until the trigger of the new conn was recorded. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + p.reconnectingMu.Lock() + pending := p.pendingReroute + p.reconnectingMu.Unlock() + if pending { + return + } + time.Sleep(time.Millisecond) + } + } + testHookReplayed.Store(&hook) + t.Cleanup(func() { testHookReplayed.Store(nil) }) + // The slot moves a second time right after b acks the replay. + b.setOnSSub(func(n *rpNode, _ *rpConn, ch string) { + b.setOnSSub(nil) + cl.moveSlot(slot, c, 0) + }) + cl.moveSlot(slot, b, 0) + poll(t, "SSUBSCRIBE on the last owner", 15*time.Second, func() bool { return c.subscribers("ssub") == 1 }) + if held.Load() == 0 { + t.Error("the loop's hook never ran") + } +} + +// The old conn is cut off before the new one is bound: with the loop held after +// the replay (the old conn still open, not yet released), a message the cluster +// delivers on both reaches the caller once. This is the hold the public test of +// the same name cannot make: there the loop has released the old conn before +// the probe comes back. +func TestClusterRerouteOldConnIsCutOffWhileTheLoopIsHeld887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + if err := ps.Subscribe(rpCtx(t), "x"); err != nil { + t.Fatal(err) + } + rc := collect(ps) + poll(t, "both subscriptions on the owner", 5*time.Second, func() bool { + return from.subscribers("ssub") == 1 && from.subscribers("sub") == 1 + }) + to.setAfterSub(func(n *rpNode, c *rpConn, verb string) { + if verb == "SUBSCRIBE" { + cl.publish("x", "probe") + } + }) + hook := func(p *PubSub) { + if p != ps { + return + } + // The old conn is still open: wait for the probe, then give a + // duplicate time to arrive through the old conn. + deadline := time.Now().Add(3 * time.Second) + for rc.count("probe") == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + time.Sleep(200 * time.Millisecond) + } + testHookReplayed.Store(&hook) + t.Cleanup(func() { testHookReplayed.Store(nil) }) + cl.moveSlot(slot, to, 0) + poll(t, "SSUBSCRIBE on the new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + time.Sleep(400 * time.Millisecond) + if n := rc.count("probe"); n != 1 { + t.Errorf("the probe was delivered %d times, want 1", n) + } +} + +// Close wins every tie with the backoff timer: a loop that finds both ready +// must not go on to dial a conn nobody owns. +func TestPubSubSleepIsClosedFirst887(t *testing.T) { + cl := newRPCluster(t, 1) + c, err := NewClient(cl.nodes[0].addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ps, err := c.Subscribe(rpCtx(t)) + if err != nil { + t.Fatal(err) + } + _ = ps.Close() + for i := 0; i < 300; i++ { + if ps.sleep(0) { + t.Fatalf("iteration %d: sleep returned true on a closed PubSub (a select with both ready picked the timer)", i) + } + } +} + +// Close lands after the reroute loop dialled and before it installs the conn: +// the loop sees the PubSub closed under the lock, installs nothing and releases +// the conn it dialled (a conn installed on a closed PubSub would be nobody's to +// release). +func TestClusterRerouteDoesNotInstallAConnOnAClosedPubSub887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ps, err := cc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + poll(t, "the subscription", 5*time.Second, func() bool { return cl.holder("sub") != nil }) + first := cl.holder("sub") + + reached := make(chan struct{}) + proceed := make(chan struct{}) + var once atomic.Bool + hook := func(p *PubSub) { + if p != ps || !once.CompareAndSwap(false, true) { + return + } + close(reached) + <-proceed + } + testHookBeforeInstall.Store(&hook) + t.Cleanup(func() { testHookBeforeInstall.Store(nil) }) + + first.kill() + select { + case <-reached: + case <-time.After(10 * time.Second): + t.Fatal("the reroute loop never reached the install") + } + _ = ps.Close() + close(proceed) + poll(t, "the reroute loop to end", 2*time.Second, func() bool { return rpLoopIdle(ps) }) + ps.mu.Lock() + conn := ps.conn + ps.mu.Unlock() + if conn != nil { + t.Errorf("the loop installed a conn on a closed PubSub") + } +} + +// A report from the old conn that was in flight when the loop installed the new +// one (the router's callback read its target before the install cleared the +// router, then waited for mu) must not start a second reroute: the replay +// already answers it. The loop is held after the install and the old conn's +// router reports in that window. +func TestClusterRerouteIgnoresAReportFromTheReplacedConn887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + + ps.mu.Lock() + oldRouter := ps.conn.state.router + ps.mu.Unlock() + var held atomic.Int64 + hook := func(p *PubSub) { + if p != ps || held.Add(1) > 1 { + return + } + // After the install: the replaced conn's late reports. + p.onShardUnsubscribed(oldRouter, ch) + p.onControlError(oldRouter, "MOVED "+strconv.Itoa(slot)+" 127.0.0.1:1") + p.reconnectingMu.Lock() + pending := p.pendingReroute + p.reconnectingMu.Unlock() + if pending { + t.Error("a report from the replaced conn set pendingReroute") + } + } + testHookReplayed.Store(&hook) + t.Cleanup(func() { testHookReplayed.Store(nil) }) + cl.moveSlot(slot, to, 0) + poll(t, "SSUBSCRIBE on the new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + time.Sleep(700 * time.Millisecond) + if n := to.subCmds.Load(); n != 1 { + t.Errorf("the new owner saw %d SSUBSCRIBE commands, want 1: a late report of the old conn made another reroute", n) + } +} diff --git a/driver/redis/pubsub_reroute_more_887_test.go b/driver/redis/pubsub_reroute_more_887_test.go new file mode 100644 index 000000000..d3a2a3e36 --- /dev/null +++ b/driver/redis/pubsub_reroute_more_887_test.go @@ -0,0 +1,245 @@ +package redis + +// More cells of celeris#887 item 31: what is and is not a trigger, a trigger +// that arrives during the handover, a control call racing a reroute, RESP2, and +// the sentinel's master PubSub. + +import ( + "context" + "testing" + "time" +) + +// A standalone PubSub (no route) does not act on any of it: a sunsubscribe is +// the caller's business, as before. +func TestStandalonePubSubIgnoresSunsubscribe887(t *testing.T) { + cl := newRPCluster(t, 1) + c, err := NewClient(cl.nodes[0].addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ps, err := c.Subscribe(rpCtx(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + _ = ps.SSubscribe(rpCtx(t), "foo") + node := cl.nodes[0] + poll(t, "the SSUBSCRIBE", 5*time.Second, func() bool { return node.subscribers("ssub") == 1 }) + node.sunsubscribeSlot(int(Slot("foo"))) + time.Sleep(400 * time.Millisecond) + if n := node.subCmds.Load(); n != 1 { + t.Errorf("a standalone PubSub subscribed again (%d SSUBSCRIBEs)", n) + } +} + +// A control call racing a reroute is neither lost nor undone: whatever the +// order, the channel the caller removed is not subscribed again anywhere, and +// the one it added is. +func TestClusterControlCallRacingARerouteLeavesTheCallerWithItsSet887(t *testing.T) { + for i := 0; i < 20; i++ { + cl := newRPCluster(t, 3) + cc, err := NewClusterClient(ClusterConfig{Addrs: cl.addrs(), DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + const ch, ch2 = "{foo}1", "{foo}2" // same slot + slot := int(Slot(ch)) + from := cl.owner(slot) + to := cl.nodes[0] + if from == to { + to = cl.nodes[1] + } + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + t.Fatal(err) + } + poll(t, "SSUBSCRIBE", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + done := make(chan struct{}) + go func() { + defer close(done) + time.Sleep(time.Duration(i) * 200 * time.Microsecond) + _ = ps.SSubscribe(context.Background(), ch2) + _ = ps.SUnsubscribe(context.Background(), ch) + }() + cl.moveSlot(slot, to, 0) + <-done + // Settled: only ch2 is subscribed, on the new owner. + poll(t, "ch2 on the new owner, ch gone", 10*time.Second, func() bool { + c := to.conn(func(c *rpConn) bool { + c.mu.Lock() + defer c.mu.Unlock() + _, a := c.ssubs[ch2] + _, b := c.ssubs[ch] + return a && !b + }) + return c != nil && cl.subscribers("ssub") == 1 + }) + _ = ps.Close() + _ = cc.Close() + } +} + +// Closing the PubSub at every point of a reroute releases every conn: no +// server-side conn is left behind (the pool's counts are not touched either). +func TestClusterPubSubCloseLeavesNoConnBehind887(t *testing.T) { + for i := 0; i < 40; i++ { + cl := newRPCluster(t, 3) + cc, err := NewClusterClient(ClusterConfig{Addrs: cl.addrs(), DialTimeout: 2 * time.Second}) + if err != nil { + t.Fatal(err) + } + // The table must hold the three nodes before anything else is asked of + // it (a CLUSTER SLOTS reply that is not one empties it, and the first + // symptom is an SSubscribe that finds no node: seen once, on Windows). + cc.mu.RLock() + var table []string + for addr := range cc.nodes { + table = append(table, addr) + } + cc.mu.RUnlock() + if len(table) != 3 { + t.Fatalf("iteration %d: after NewClusterClient the client's node table is %v, want the three nodes %v (CLUSTER SLOTS replies given: %d)", + i, table, cl.addrs(), cl.slotsReplies.Load()) + } + const ch = "foo" + slot := int(Slot(ch)) + to := cl.nodes[0] + ps, err := cc.SSubscribe(rpCtx(t), ch) + if err != nil { + // (Seen once, on Windows, as "no cluster nodes available": say what + // the client's table looked like.) + cc.mu.RLock() + nodes, owner := len(cc.nodes), cc.slots[slot] != nil + cc.mu.RUnlock() + t.Fatalf("iteration %d: SSubscribe: %v (the client knows %d nodes, slot owner known: %v, CLUSTER SLOTS replies given: %d)", + i, err, nodes, owner, cl.slotsReplies.Load()) + } + poll(t, "SSUBSCRIBE", 5*time.Second, func() bool { return cl.subscribers("ssub") == 1 }) + cl.moveSlot(slot, to, 0) + poll(t, "SSUBSCRIBE on the second node", 5*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + // A second move: the conn the loop releases now came from a node that + // is not the one the PubSub started on. + third := cl.nodes[1] + if third == to || third == cl.owner(slot) { + third = cl.nodes[2] + } + cl.moveSlot(slot, third, 0) + time.Sleep(time.Duration(i) * 3 * time.Millisecond) + _ = ps.Close() + // Every conn went back to the pool it came from: no node's pubsub pool + // counts one still open, and none counts one too few (a conn released + // to the pool of the client the PubSub started on would do both). + cc.mu.RLock() + pools := map[string]*redisPool{} + for addr, n := range cc.nodes { + pools[addr] = n.client.pool + } + cc.mu.RUnlock() + // (Close does not wait for a dial of the loop that is in flight: it + // releases its conn a moment later. A conn that is never released, or + // released to the wrong pool, does not settle.) + poll(t, "every pubsub pool back to zero open conns", 5*time.Second, func() bool { + for _, pl := range pools { + if pl.pubsub.Stats().Open != 0 { + return false + } + } + return true + }) + _ = cc.Close() + poll(t, "every conn closed", 5*time.Second, func() bool { return cl.liveConns() == 0 }) + } +} + +// A SentinelClient's PubSub follows a failover: it subscribes on the new master +// and keeps delivering to the same channel. Before, it retried the old master's +// closed client for ever. +func TestSentinelPubSubFollowsAFailover887(t *testing.T) { + cl := newRPCluster(t, 2) + m1, m2 := cl.nodes[0], cl.nodes[1] + sentinel := startFakeSentinel(t, m1.addr) + sc, err := NewSentinelClient(SentinelConfig{ + MasterName: "mymaster", + SentinelAddrs: []string{sentinel.Addr()}, + DialTimeout: 2 * time.Second, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sc.Close() }) + ps, err := sc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "the subscription on the master", 5*time.Second, func() bool { return m1.subscribers("sub") == 1 }) + + // The SentinelClient subscribes to +switch-master in the background: a + // switch pushed before that subscription exists is never heard. + poll(t, "the SentinelClient subscribed to +switch-master", 5*time.Second, func() bool { + sentinel.mu.Lock() + defer sentinel.mu.Unlock() + return len(sentinel.subWriters) > 0 + }) + m1.kill() + sentinel.SetMaster(m2.addr) + poll(t, "the subscription on the new master", 15*time.Second, func() bool { return m2.subscribers("sub") == 1 }) + cl.publish("news", "m1") + poll(t, "m1 on the same channel", 5*time.Second, func() bool { return rc.count("m1") == 1 }) +} + +// Closing the SentinelClient ends its PubSubs' reroute: Channel() closes. +func TestSentinelCloseEndsThePubSubReroute887(t *testing.T) { + cl := newRPCluster(t, 1) + sentinel := startFakeSentinel(t, cl.nodes[0].addr) + sc, err := NewSentinelClient(SentinelConfig{ + MasterName: "mymaster", + SentinelAddrs: []string{sentinel.Addr()}, + DialTimeout: 2 * time.Second, + }) + if err != nil { + t.Fatal(err) + } + ps, err := sc.Subscribe(rpCtx(t), "news") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "the subscription", 5*time.Second, func() bool { return cl.nodes[0].subscribers("sub") == 1 }) + _ = sc.Close() + poll(t, "Channel() closed", 5*time.Second, func() bool { return rc.closed.Load() }) + poll(t, "no reroute loop", 2*time.Second, func() bool { return rpLoopIdle(ps) }) +} + +// Two unsolicited sunsubscribes of the same move (one per channel of the slot) +// make one reroute, not two: the second arrives while the loop is running, from +// the old conn, and the loop's replay already answers it. A flag kept from +// before the install would start another loop (another gap, another dial). +func TestClusterRerouteAnswersEveryChannelOfAMoveWithOneLoop887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch1, ch2 = "{foo}1", "{foo}2" + slot := int(Slot(ch1)) + from := cl.owner(slot) + to := cl.nodes[0] + if from == to { + to = cl.nodes[1] + } + ps, err := cc.SSubscribe(rpCtx(t), ch1, ch2) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + poll(t, "both channels on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + cl.moveSlot(slot, to, 0) + poll(t, "SSUBSCRIBE on the new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + // A needless second loop would dial again after its (longer) backoff. + time.Sleep(700 * time.Millisecond) + if n := to.subCmds.Load(); n != 1 { + t.Errorf("the new owner saw %d SSUBSCRIBE commands, want 1: the second sunsubscribe of the move made another reroute", n) + } +} diff --git a/driver/redis/pubsub_reroute_r1_887_test.go b/driver/redis/pubsub_reroute_r1_887_test.go new file mode 100644 index 000000000..3bd779760 --- /dev/null +++ b/driver/redis/pubsub_reroute_r1_887_test.go @@ -0,0 +1,220 @@ +package redis + +// Review round 1 of celeris#887 item 31 (PR #1154): the cells below compile on +// the first head and fail there (F2, F2b, F3, F4, F7), or pin a gap the first +// head's tests had (F1: a mutant survived). + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// F1. The first head's guard for the closed check before the install used a +// PubSub with a channel, whose replay (it checks closed before every write) +// released the conn anyway. With no channel the replay writes nothing, so only +// the check before the install keeps a conn off a closed PubSub. +func TestClusterRerouteDoesNotInstallOnAClosedPubSubWithNoChannels887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ps, err := cc.Subscribe(rpCtx(t)) // no channels: the replay writes nothing + if err != nil { + t.Fatal(err) + } + ps.mu.Lock() + home := ps.connClient + ps.mu.Unlock() + var homeAddr string + cc.mu.RLock() + for a, n := range cc.nodes { + if n.client == home { + homeAddr = a + } + } + cc.mu.RUnlock() + reached, proceed := make(chan struct{}), make(chan struct{}) + var once atomic.Bool + hook := func(p *PubSub) { + if p != ps || !once.CompareAndSwap(false, true) { + return + } + close(reached) + <-proceed + } + testHookBeforeInstall.Store(&hook) + t.Cleanup(func() { testHookBeforeInstall.Store(nil) }) + for _, n := range cl.nodes { + if n.addr == homeAddr { + n.kill() + } + } + select { + case <-reached: + case <-time.After(10 * time.Second): + t.Fatal("the reroute loop never reached the install") + } + _ = ps.Close() + close(proceed) + poll(t, "the reroute loop to end", 2*time.Second, func() bool { return rpLoopIdle(ps) }) + ps.mu.Lock() + conn := ps.conn + ps.mu.Unlock() + if conn != nil { + t.Errorf("a conn was installed on a closed PubSub (nobody releases it)") + } + cc.mu.RLock() + var pools []*redisPool + for _, nd := range cc.nodes { + pools = append(pools, nd.client.pool) + } + cc.mu.RUnlock() + poll(t, "every pubsub pool back to zero open conns", 3*time.Second, func() bool { + for _, pl := range pools { + if pl.pubsub.Stats().Open != 0 { + return false + } + } + return true + }) +} + +// F2. A dial that fails because the CALLER's context ended says nothing about +// the node: it must not be marked failing (anyNode steers Publish, keyless +// commands and the topology probe away from failing nodes). +func TestClusterSubscribeWithACancelledContextDoesNotMarkANodeFailing887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _ = cc.Subscribe(ctx, "x") + cc.mu.RLock() + defer cc.mu.RUnlock() + for addr, n := range cc.nodes { + if n.failing.Load() { + t.Errorf("healthy node %s marked failing by a Subscribe whose context was cancelled", addr) + } + } +} + +// F2b. A PubSub dial that succeeds clears only the failing mark a PubSub dial +// set; one the command path set (a network error on a command) is the command +// path's to clear. +func TestPubSubDialReportClearsOnlyTheMarkItSet887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + route := clusterPubSubRoute{cc} + cc.mu.RLock() + var n *clusterNode + for _, nd := range cc.nodes { + n = nd + break + } + cc.mu.RUnlock() + + n.failing.Store(true) // the command path's verdict + route.dialed(n.client, true) + if !n.failing.Load() { + t.Error("a successful PubSub dial cleared the failing mark the command path set") + } + n.failing.Store(false) + + route.dialed(n.client, false) // the PubSub path's own verdict ... + if !n.failing.Load() { + t.Fatal("a failed PubSub dial did not mark the node failing (the next attempt would not move on)") + } + route.dialed(n.client, true) // ... and its own recovery + if n.failing.Load() { + t.Error("a successful PubSub dial did not clear the mark a failed PubSub dial set") + } +} + +// F3. Real Redis answers SUNSUBSCRIBE of a channel it holds no subscription for, +// in a slot it does not serve, with -MOVED. That is an answer to the caller's +// harmless call, not a sign that the subscription is lost: it must not rebuild +// a healthy subscription (a delivery gap for every channel of the PubSub). +func TestClusterForeignSUnsubscribeDoesNotRerouteAHealthySubscription887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const a, foreign = "{a}x", "{b}never" // slots 15495 and 3300: different nodes + if cl.owner(int(Slot(a))) == cl.owner(int(Slot(foreign))) { + t.Fatal("test setup: the two slots share a node") + } + ps, err := cc.SSubscribe(rpCtx(t), a) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + holder := cl.owner(int(Slot(a))) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return holder.subscribers("ssub") == 1 }) + accepted, cmds := holder.acceptedConns(), holder.subCmds.Load() + + if err := ps.SUnsubscribe(rpCtx(t), foreign); err != nil { + t.Fatal(err) + } + time.Sleep(700 * time.Millisecond) + if n := holder.acceptedConns(); n != accepted { + t.Errorf("the PubSub dialled %d more conns after a SUnsubscribe of a channel it never had: a healthy subscription was rebuilt", n-accepted) + } + if n := holder.subCmds.Load(); n != cmds { + t.Errorf("the owner saw %d more SSUBSCRIBEs", n-cmds) + } + cl.spublish(a, "m1") + poll(t, "m1", 5*time.Second, func() bool { return rc.count("m1") == 1 }) +} + +// F4. Two slots on one PubSub (documented undefined behaviour): the channel of +// the first slot keeps working, as on the base. The first head sent both in one +// SSUBSCRIBE after a reroute, got -CROSSSLOT, and left both dead. +func TestClusterSecondSlotOnOnePubSubDoesNotKillTheFirst887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const a, b = "{a}x", "{b}y" + slotA := int(Slot(a)) + from := cl.owner(slotA) + if from == cl.owner(int(Slot(b))) { + t.Fatal("test setup: the two slots share a node") + } + ps, err := cc.SSubscribe(rpCtx(t), a) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return from.subscribers("ssub") == 1 }) + if err := ps.SSubscribe(rpCtx(t), b); err != nil { // answered -MOVED: b is not served + t.Fatal(err) + } + time.Sleep(900 * time.Millisecond) // a reroute (and its replay) would have happened by now + cl.spublish(a, "m1") + poll(t, "m1 on the first channel", 5*time.Second, func() bool { return rc.count("m1") == 1 }) + + // The same after the first channel's slot moves: the reroute replays the + // channel of the served slot, not both. + to := cl.nodes[1] + if to == from || to == cl.owner(int(Slot(b))) { + to = cl.nodes[0] + } + cl.moveSlot(slotA, to, 0) + poll(t, "SSUBSCRIBE of the first channel on the new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + cl.spublish(a, "m2") + poll(t, "m2 on the first channel", 5*time.Second, func() bool { return rc.count("m2") == 1 }) +} + +// F7. A dead owner that the (still stale) topology keeps naming is dialled once +// by the constructor, not once per node of the cluster, each with a topology +// refresh in front of it. +func TestClusterSSubscribeTriesADeadOwnerOnce887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const ch = "foo" + cl.owner(int(Slot(ch))).kill() + before := cl.slotsReplies.Load() + if _, err := cc.SSubscribe(rpCtx(t), ch); err == nil { + t.Fatal("SSubscribe on a dead owner returned no error") + } + if n := cl.slotsReplies.Load() - before; n > 1 { + t.Errorf("the constructor refreshed the topology %d times for one dead owner (it names the same node every time), want at most 1", n) + } +} diff --git a/driver/redis/pubsub_reroute_r2_887_test.go b/driver/redis/pubsub_reroute_r2_887_test.go new file mode 100644 index 000000000..0bbd2b973 --- /dev/null +++ b/driver/redis/pubsub_reroute_r2_887_test.go @@ -0,0 +1,125 @@ +package redis + +// Review round 2 of celeris#887 item 31 (PR #1154): the cells compile on the +// round-1 head and fail there (N2, N5), or kill mutants that survived its +// suite (N6: every earlier cell had a first channel that also sorts first). + +import ( + "testing" + "time" +) + +// N2. A channel of a second slot that the SAME node also owns is accepted by the +// node; when that slot moves away the node sends "sunsubscribe" for it. It is +// not the slot the PubSub serves, so it must not rebuild the served +// subscription (a gap on every served channel). +func TestClusterSunsubscribeOfANonServedSlotDoesNotRerouteTheServedOne887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const a, f = "{a}x", "foo" // slots 15495 and 12182: both on nodes[2] + sa, sf := int(Slot(a)), int(Slot(f)) + holder := cl.owner(sa) + if holder != cl.owner(sf) { + t.Fatalf("setup: slots %d and %d are on different nodes", sa, sf) + } + ps, err := cc.SSubscribe(rpCtx(t), a) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return holder.subscribers("ssub") == 1 }) + if err := ps.SSubscribe(rpCtx(t), f); err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + cl.spublish(f, "f1") + poll(t, "f1 (the second slot is accepted by the same node)", 3*time.Second, func() bool { return rc.count("f1") == 1 }) + accepted := holder.acceptedConns() + + cl.moveSlot(sf, cl.nodes[0], 0) // the NON-served slot moves away + time.Sleep(1200 * time.Millisecond) + if n := holder.acceptedConns() - accepted; n != 0 { + t.Errorf("the move of a slot the PubSub does not serve made it dial %d more conns: the served subscription was rebuilt (a gap on %s)", n, a) + } + cl.spublish(a, "a1") + poll(t, "a1", 3*time.Second, func() bool { return rc.count("a1") == 1 }) +} + +// N6. The served slot is the slot of the FIRST channel, even when a later +// channel of another slot sorts lower. +func TestClusterFirstChannelNotSmallestStaysServed887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const z, a = "{z}q", "{a}x" + if cl.owner(int(Slot(z))) == cl.owner(int(Slot(a))) { + t.Fatalf("setup: same node (%d, %d)", Slot(z), Slot(a)) + } + ps, err := cc.SSubscribe(rpCtx(t), z) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return cl.owner(int(Slot(z))).subscribers("ssub") == 1 }) + _ = ps.SSubscribe(rpCtx(t), a) // -MOVED for a's slot: ignored + time.Sleep(1200 * time.Millisecond) + cl.spublish(z, "z1") + poll(t, "z1 on the first channel", 3*time.Second, func() bool { return rc.count("z1") == 1 }) +} + +// N6. The reroute follows the served slot even when a channel of another slot +// sorts lower: a pick by the smallest channel of any slot would chase the other +// slot's owner, get -MOVED for the served slot and loop for ever. +func TestClusterRerouteFollowsTheServedSlotNotTheSmallestChannel887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const z, a = "{z}q", "{a}x" + sz := int(Slot(z)) + oz, oa := cl.owner(sz), cl.owner(int(Slot(a))) + if oz == oa { + t.Fatal("setup: same node") + } + var third *rpNode + for _, n := range cl.nodes { + if n != oz && n != oa { + third = n + } + } + ps, err := cc.SSubscribe(rpCtx(t), z) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return oz.subscribers("ssub") == 1 }) + _ = ps.SSubscribe(rpCtx(t), a) + time.Sleep(300 * time.Millisecond) + cl.moveSlot(sz, third, 0) + poll(t, "the served channel on the new owner", 10*time.Second, func() bool { return third.subscribers("ssub") == 1 }) + cl.spublish(z, "z1") + poll(t, "z1", 3*time.Second, func() bool { return rc.count("z1") == 1 }) +} + +// N5. The failing mark the command path sets after a PubSub dial failed is the +// command path's: a later successful PubSub dial must not clear it (the +// PubSub's own mark went stale when the command path took the node over). +func TestPubSubDialSuccessDoesNotClearAMarkTheCommandPathSetAfterwards887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + route := clusterPubSubRoute{cc} + cc.mu.RLock() + var n *clusterNode + for _, nd := range cc.nodes { + n = nd + break + } + cc.mu.RUnlock() + route.dialed(n.client, false) // the PubSub path's verdict + n.failing.Store(false) // a topology probe of this node found it alive + n.failing.Store(true) // a command hit a network error + route.dialed(n.client, true) // a PubSub dial succeeded + if !n.failing.Load() { + t.Error("a successful PubSub dial cleared the failing mark the command path set after the PubSub's own") + } +} diff --git a/driver/redis/pubsub_reroute_r3_887_test.go b/driver/redis/pubsub_reroute_r3_887_test.go new file mode 100644 index 000000000..55bb1566b --- /dev/null +++ b/driver/redis/pubsub_reroute_r3_887_test.go @@ -0,0 +1,147 @@ +package redis + +// Review round 3 of celeris#887 item 31 (PR #1154): a mutant that survived +// round 2's suite (the sunsubscribe handler using the cached served slot), and +// the failing flag's state machine. + +import ( + "runtime" + "sync" + "sync/atomic" + "testing" + "time" +) + +// m1. The served slot is re-selected when its channels are gone: with {a}x and +// foo on the same node, SUnsubscribe({a}x) leaves foo as the only channel, and +// the move of foo's slot must be followed. A handler that read the cached served +// slot instead of re-deriving it would ignore foo's sunsubscribe, and the +// subscription would be lost until some other reroute. +func TestClusterServedSlotIsReselectedWhenItsChannelsAreGone887(t *testing.T) { + cl := newRPCluster(t, 3) + cc := newRPClusterClient(t, cl) + const a, f = "{a}x", "foo" // slots 15495 and 12182: both on nodes[2] + holder := cl.owner(int(Slot(a))) + if holder != cl.owner(int(Slot(f))) { + t.Fatal("setup: the two slots are on different nodes") + } + ps, err := cc.SSubscribe(rpCtx(t), a) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ps.Close() }) + rc := collect(ps) + poll(t, "SSUBSCRIBE on the owner", 5*time.Second, func() bool { return holder.subscribers("ssub") == 1 }) + if err := ps.SSubscribe(rpCtx(t), f); err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + if err := ps.SUnsubscribe(rpCtx(t), a); err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + + to := cl.nodes[0] + cl.moveSlot(int(Slot(f)), to, 0) + poll(t, "the remaining channel followed to its new owner", 10*time.Second, func() bool { return to.subscribers("ssub") == 1 }) + cl.spublish(f, "f1") + poll(t, "f1", 3*time.Second, func() bool { return rc.count("f1") == 1 }) +} + +// The failing flag, sequentially: a PubSub success undoes a PubSub failure and +// nothing else. +func TestFailFlagStateMachine887(t *testing.T) { + var f failFlag + f.markPubSub() + if !f.Load() { + t.Fatal("markPubSub did not set the flag") + } + f.clearPubSub() + if f.Load() { + t.Fatal("clearPubSub did not undo markPubSub") + } + f.Store(true) // the command path + f.clearPubSub() + if !f.Load() { + t.Fatal("clearPubSub undid the command path's mark") + } + f.markPubSub() + f.Store(true) // the command path takes the node over + f.clearPubSub() + if !f.Load() { + t.Fatal("clearPubSub undid a mark the command path set after the PubSub's") + } + f.markPubSub() + f.Store(false) // a topology probe found it alive + f.clearPubSub() + f.Store(true) + f.clearPubSub() + if !f.Load() { + t.Fatal("a stale PubSub mark survived a Store(false) and made a later command mark clearable") + } +} + +// The flag under concurrency. A command's Store(true) and a PubSub success that +// run at the same time leave the flag set whatever their order (Store first: +// the success finds a mark that is not its own; the success first: Store sets +// it afterwards). A clear that read the PubSub's mark, was overtaken by the +// command's whole Store, and then wrote false would erase the command's fresh +// mark. +func TestFailFlagCommandMarkIsNotLostToAConcurrentPubSubSuccess887(t *testing.T) { + const rounds = 300000 + var f failFlag + var start atomic.Int32 + var lost atomic.Int64 + var wg sync.WaitGroup + round := make(chan struct{}) + done := make(chan struct{}, 2) + wg.Add(2) + run := func(op func()) { + defer wg.Done() + for i := 0; i < rounds; i++ { + <-round + for start.Load() == 0 { + runtime.Gosched() + } + op() + done <- struct{}{} + } + } + go run(func() { f.clearPubSub() }) + go run(func() { f.Store(true) }) + for i := 0; i < rounds; i++ { + f.markPubSub() + start.Store(0) + round <- struct{}{} + round <- struct{}{} + start.Store(1) + <-done + <-done + if !f.Load() { + lost.Add(1) + } + } + wg.Wait() + if n := lost.Load(); n != 0 { + t.Errorf("%d of %d rounds lost the command path's mark to a concurrent PubSub success", n, rounds) + } +} + +func BenchmarkFailFlagStore887(b *testing.B) { + var f failFlag + b.ReportAllocs() + for b.Loop() { + f.Store(false) + } +} + +func BenchmarkAnyNode887(b *testing.B) { + c := &ClusterClient{nodes: map[string]*clusterNode{}} + for i := 0; i < 6; i++ { + c.nodes[string(rune('a'+i))] = &clusterNode{} + } + b.ReportAllocs() + for b.Loop() { + _ = c.anyNode() + } +} diff --git a/driver/redis/pubsub_route.go b/driver/redis/pubsub_route.go new file mode 100644 index 000000000..0a03b0461 --- /dev/null +++ b/driver/redis/pubsub_route.go @@ -0,0 +1,189 @@ +package redis + +import ( + "context" + "errors" + "slices" + "strings" +) + +// pubsubRoute chooses the client a PubSub's next conn is dialled on, so a +// subscription follows its node when the node is gone and its slot when the +// slot moves (celeris#887 item 31). A PubSub of a standalone Client has none: +// it dials that client for ever. +// +// pick and dialed run on the PubSub's reconnect goroutine with no PubSub lock +// held. +type pubsubRoute interface { + // pick returns the client to dial. shard is a shard channel the PubSub + // holds ("" if none): the client must be the owner of its slot. cur is + // the client the PubSub last used and moveOn is true when it did not work + // out; for a subscription that can live on any node a route then returns + // another. An error that matches [ErrClosed] means the route's owner is + // closed and nothing will ever work (the loop gives up); any other error is + // retried with the backoff. + pick(ctx context.Context, shard string, cur *Client, moveOn bool) (*Client, error) + // dialed reports the outcome of a dial on cl. + dialed(cl *Client, ok bool) + // slotted reports whether the route serves shard channels by slot (a + // cluster): the PubSub then serves one slot and ignores a -MOVED for + // another. + slotted() bool +} + +// errPubSubNoOwner is a transient pick failure: the topology names no node for +// the slot, or none is known. +var errPubSubNoOwner = errors.New("celeris-redis: no cluster node for the pubsub subscription") + +// clusterPubSubRoute routes the PubSubs of a ClusterClient. +// +// - A regular SUBSCRIBE or PSUBSCRIBE is node-local and the cluster bus +// delivers every PUBLISH to every node, so any node will do: the PubSub +// stays on its node while that works and otherwise goes to the next one +// (by address) that is not marked failing. +// - A sharded subscription lives on the owner of its slot: the topology is +// refreshed (bounded by the dial timeout) and the owner of the slot read. +type clusterPubSubRoute struct{ c *ClusterClient } + +func (r clusterPubSubRoute) pick(ctx context.Context, shard string, cur *Client, moveOn bool) (*Client, error) { + c := r.c + if c.closed.Load() { + return nil, ErrClosed + } + if shard != "" { + bound := c.cfg.DialTimeout + if bound <= 0 { + bound = defaultDialTimeout + } + rctx, cancel := context.WithTimeout(ctx, bound) + // A failed refresh leaves the table as it was: the dial tells. + _ = c.refreshTopology(rctx) + cancel() + if c.closed.Load() { + return nil, ErrClosed + } + n := c.route(Slot(shard)) + if n == nil { + // The table names no owner (the SSubscribe constructor does the + // same): any node, which answers -MOVED if it is the wrong one, + // and the PubSub asks again. + n = c.anyNode() + } + if n == nil { + return nil, errPubSubNoOwner + } + return n.client, nil + } + return c.pubsubNode(cur, moveOn) +} + +func (clusterPubSubRoute) slotted() bool { return true } + +// dialed records what a PubSub dial found out about a node. A failure marks the +// node failing (so the next attempt moves on) and remembers that the PubSub +// path did it; a success clears the mark only if the PubSub path's failure was +// the last thing to change it (a mark the command path or a topology probe has +// set or cleared since is theirs). +func (r clusterPubSubRoute) dialed(cl *Client, ok bool) { + c := r.c + c.mu.RLock() + defer c.mu.RUnlock() + for _, n := range c.nodes { + if n.client == cl { + if !ok { + n.failing.markPubSub() + } else { + n.failing.clearPubSub() + } + return + } + } +} + +// pubsubNode picks the node of a regular subscription: cur while it works, +// otherwise the next one after it, by address, that is not marked failing (or +// the next one, if all are). +func (c *ClusterClient) pubsubNode(cur *Client, moveOn bool) (*Client, error) { + c.mu.RLock() + nodes := make([]*clusterNode, 0, len(c.nodes)) + for _, n := range c.nodes { + nodes = append(nodes, n) + } + c.mu.RUnlock() + if len(nodes) == 0 { + if c.closed.Load() { + return nil, ErrClosed + } + return nil, errPubSubNoOwner + } + slices.SortFunc(nodes, func(a, b *clusterNode) int { return strings.Compare(a.addr, b.addr) }) + at := -1 + for i, n := range nodes { + if n.client == cur { + at = i + } + } + if !moveOn && at >= 0 && !nodes[at].failing.Load() { + return cur, nil + } + for i := 1; i <= len(nodes); i++ { + if n := nodes[(at+i+len(nodes))%len(nodes)]; !n.failing.Load() { + return n.client, nil + } + } + return nodes[(at+1+len(nodes))%len(nodes)].client, nil +} + +// openPubSub opens the PubSub of a ClusterClient. It dials first and, if that +// node is down, the others in turn (a regular subscription) or the owner the +// refreshed topology names (a shard channel: shard is its first channel), so a +// node that is gone does not fail the call while another would serve it. +func (c *ClusterClient) openPubSub(ctx context.Context, first *Client, shard string) (*PubSub, error) { + route := clusterPubSubRoute{c} + cl := first + tried := map[*Client]struct{}{cl: {}} + var lastErr error + for try := 0; ; try++ { + ps, err := cl.newPubSubRouted(ctx, route) + if err == nil { + route.dialed(cl, true) + return ps, nil + } + // A dial that failed because the caller's context ended says nothing + // about the node. + if ctx.Err() == nil { + route.dialed(cl, false) + } + lastErr = err + c.mu.RLock() + n := len(c.nodes) + c.mu.RUnlock() + if ctx.Err() != nil || try+1 >= max(n, 2) || c.closed.Load() { + return nil, lastErr + } + next, perr := route.pick(ctx, shard, cl, true) + if perr != nil { + return nil, lastErr + } + // Each node at most once: a stale topology keeps naming a dead owner, + // and dialling it again (after a refresh) only spends the caller's time. + if _, seen := tried[next]; seen { + return nil, lastErr + } + tried[next] = struct{}{} + cl = next + } +} + +// sentinelPubSubRoute routes the PubSubs of a SentinelClient: the current +// master, so a PubSub follows a failover instead of dialling the old master's +// closed client for ever. +type sentinelPubSubRoute struct{ s *SentinelClient } + +func (r sentinelPubSubRoute) pick(context.Context, string, *Client, bool) (*Client, error) { + return r.s.getClient() +} + +func (r sentinelPubSubRoute) dialed(*Client, bool) {} + +func (sentinelPubSubRoute) slotted() bool { return false } diff --git a/driver/redis/sentinel.go b/driver/redis/sentinel.go index d387a8935..327244df3 100644 --- a/driver/redis/sentinel.go +++ b/driver/redis/sentinel.go @@ -765,13 +765,25 @@ func (s *SentinelClient) Publish(ctx context.Context, channel, message string) ( // Subscribe is [Client.Subscribe] on the current master: it returns the // PubSub with a nil error when the conn is found down on the first SUBSCRIBE, -// and the reconnect replays the channels. +// and the reconnect replays the channels. The reconnect dials the master the +// SentinelClient has at that moment, so after a failover the PubSub follows the +// new master instead of retrying the old one, once its conn to the old master +// has dropped (a conn that stays up, on a loop shared with another Client or +// WithEngine, is not moved). When the SentinelClient is closed, a PubSub whose +// conn is down (or goes down afterwards) gives up: Channel() is closed. func (s *SentinelClient) Subscribe(ctx context.Context, channels ...string) (*PubSub, error) { c, err := s.getClient() if err != nil { return nil, err } - return c.Subscribe(ctx, channels...) + ps, err := c.newPubSubRouted(ctx, sentinelPubSubRoute{s}) + if err != nil { + return nil, err + } + if len(channels) == 0 { + return ps, nil + } + return ps.firstControl(ps.Subscribe(ctx, channels...)) } func (s *SentinelClient) Do(ctx context.Context, args ...any) (*Value, error) { diff --git a/driver/redis/state.go b/driver/redis/state.go index 391a55c52..4ea2bfb89 100644 --- a/driver/redis/state.go +++ b/driver/redis/state.go @@ -147,6 +147,30 @@ func (r *pubsubRouter) clear() { r.mu.Unlock() } +// shardUnsubscribed reports a "sunsubscribe" push for ch to the PubSub: it is +// either the ack of the PubSub's own SUnsubscribe or the server telling the +// client that it no longer serves the channel's slot (celeris#887 item 31). +func (r *pubsubRouter) shardUnsubscribed(ch string) { + r.mu.Lock() + t := r.target + r.mu.Unlock() + if t != nil { + t.onShardUnsubscribed(r, ch) + } +} + +// controlError reports an error reply that arrived in pubsub mode, where no +// command waits for a reply (the control writes are not tracked): a -MOVED to +// an SSUBSCRIBE, say (celeris#887 item 31). +func (r *pubsubRouter) controlError(msg string) { + r.mu.Lock() + t := r.target + r.mu.Unlock() + if t != nil { + t.onControlError(r, msg) + } +} + //nolint:unparam // bool return kept for test introspection and backpressure func (r *pubsubRouter) deliver(m *Message) bool { r.mu.Lock() @@ -295,7 +319,18 @@ func (s *redisState) dispatch(v protocol.Value) error { head := s.bridge.Pop() if head == nil { if s.currentMode() == modePubSub { + // Nothing waits for a reply here (the control writes are not + // tracked), so an error reply is dropped, except that a + // cluster PubSub learns from it that its subscription did not + // take (a -MOVED to an SSUBSCRIBE). + var emsg string + if v.Type == protocol.TyError || v.Type == protocol.TyBlobErr { + emsg = string(v.Str) + } s.reader.Release(v) + if emsg != "" { + s.router.controlError(emsg) + } return nil } s.reader.Release(v) @@ -652,8 +687,12 @@ func (s *redisState) routePush(v protocol.Value) { payload := string(v.Array[2].Str) s.router.deliver(&Message{Channel: ch, Payload: payload, Shard: true}) } + case "sunsubscribe": + // The ack of a SUnsubscribe, or the server dropping a shard channel + // whose slot moved to another node: the PubSub tells them apart. + s.router.shardUnsubscribed(string(v.Array[1].Str)) case "subscribe", "unsubscribe", "psubscribe", "punsubscribe", - "ssubscribe", "sunsubscribe": + "ssubscribe": // Control acks — drop silently; PubSub tracks subs independently. } } diff --git a/test/conformance/redis/pubsub_reroute_cluster_r1_test.go b/test/conformance/redis/pubsub_reroute_cluster_r1_test.go new file mode 100644 index 000000000..d27b0e6d0 --- /dev/null +++ b/test/conformance/redis/pubsub_reroute_cluster_r1_test.go @@ -0,0 +1,134 @@ +//go:build redis_cluster + +package redis_test + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/goceleris/celeris/driver/redis" +) + +type r1Counter struct { + mu sync.Mutex + n map[string]int +} + +func (c *r1Counter) get(k string) int { c.mu.Lock(); defer c.mu.Unlock(); return c.n[k] } + +func r1Collect(ps *redis.PubSub) *r1Counter { + c := &r1Counter{n: map[string]int{}} + go func() { + for m := range ps.Channel() { + c.mu.Lock() + c.n[m.Channel+"/"+m.Payload]++ + c.mu.Unlock() + } + }() + return c +} + +// Two slots on one PubSub (documented undefined behaviour): the channel of the +// first slot keeps delivering, as it did before the PubSub followed its slot. +// The second channel is on another node (the SSUBSCRIBE is answered -MOVED) and +// is not served. A reroute that replayed both channels in one SSUBSCRIBE would +// get -CROSSSLOT from Redis and leave both dead. +func TestClusterSecondSlotOnOnePubSubKeepsTheFirstDelivering(t *testing.T) { + cc := newClusterClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + const a, b = "{l22-a}.x", "{l22-b}.y" + if redis.Slot(a) == redis.Slot(b) { + t.Fatal("test setup: same slot") + } + ps, err := cc.SSubscribe(ctx, a) + if err != nil { + t.Fatal(err) + } + defer func() { _ = ps.Close() }() + rc := r1Collect(ps) + time.Sleep(300 * time.Millisecond) + _ = ps.SSubscribe(ctx, b) // not served when it is on another node; the call itself returns nil + time.Sleep(2 * time.Second) + for i := 0; i < 3; i++ { + _, _ = cc.SPublish(ctx, a, fmt.Sprintf("m%d", i)) + time.Sleep(200 * time.Millisecond) + } + time.Sleep(500 * time.Millisecond) + for i := 0; i < 3; i++ { + if n := rc.get(fmt.Sprintf("%s/m%d", a, i)); n != 1 { + t.Errorf("%s m%d delivered %d times, want 1: the first channel died with the second", a, i, n) + } + } +} + +// ssubConnIDs returns the ids of the connections of the node that hold a shard +// subscription (CLIENT LIST shows ssub=N). +func ssubConnIDs(ctx context.Context, t *testing.T, c *redis.Client) []string { + t.Helper() + v, err := c.Do(ctx, "CLIENT", "LIST") + if err != nil { + t.Fatalf("CLIENT LIST: %v", err) + } + var ids []string + for _, line := range strings.Split(string(v.Str), "\n") { + if !strings.Contains(line, " ssub=") || strings.Contains(line, " ssub=0 ") { + continue + } + for _, f := range strings.Fields(line) { + if strings.HasPrefix(f, "id=") { + ids = append(ids, f) + } + } + } + return ids +} + +// Redis answers SUNSUBSCRIBE of a channel the connection holds no subscription +// for, in a slot the node does not serve, with -MOVED. That answers the +// caller's harmless call; it must not rebuild the healthy subscription: the +// connection that holds it stays the same one. +func TestClusterForeignSUnsubscribeKeepsTheSubscriptionConn(t *testing.T) { + addrs := clusterAddrsFromEnv(t) + cc := newClusterClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + const a, foreign = "{l22-a}.x", "{l22-b}.never" + seed, err := redis.NewClient(addrs[0]) + if err != nil { + t.Fatal(err) + } + defer func() { _ = seed.Close() }() + ownerAddr, _, _, _ := clusterOwnerOf(ctx, t, seed, int(redis.Slot(a))) + owner, err := redis.NewClient(ownerAddr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = owner.Close() }() + if ownerAddr2, _, _, _ := clusterOwnerOf(ctx, t, seed, int(redis.Slot(foreign))); ownerAddr2 == ownerAddr { + t.Fatal("test setup: the foreign slot is served by the same node") + } + + ps, err := cc.SSubscribe(ctx, a) + if err != nil { + t.Fatal(err) + } + defer func() { _ = ps.Close() }() + time.Sleep(500 * time.Millisecond) + before := ssubConnIDs(ctx, t, owner) + if len(before) != 1 { + t.Fatalf("the owner has %d connections holding a shard subscription before the call, want 1", len(before)) + } + if err := ps.SUnsubscribe(ctx, foreign); err != nil { + t.Fatal(err) + } + time.Sleep(1500 * time.Millisecond) + after := ssubConnIDs(ctx, t, owner) + if len(after) != 1 || after[0] != before[0] { + t.Errorf("the subscription's connection was %v, now %v: a SUnsubscribe of a channel the PubSub never had rebuilt it", before, after) + } +} diff --git a/test/conformance/redis/pubsub_reroute_cluster_r2_test.go b/test/conformance/redis/pubsub_reroute_cluster_r2_test.go new file mode 100644 index 000000000..4a1456ce6 --- /dev/null +++ b/test/conformance/redis/pubsub_reroute_cluster_r2_test.go @@ -0,0 +1,82 @@ +//go:build redis_cluster + +package redis_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/goceleris/celeris/driver/redis" +) + +// A channel of a second slot that the same node owns is accepted by the node. +// When that slot moves away the node sends sunsubscribe for it; the PubSub +// serves the slot of its first channel, so the move must not rebuild the +// subscription it serves: the connection that holds it stays the same one. +func TestClusterMoveOfASecondSlotOnTheSameNodeKeepsTheSubscriptionConn(t *testing.T) { + addrs := clusterAddrsFromEnv(t) + cc := newClusterClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + seed, err := redis.NewClient(addrs[0]) + if err != nil { + t.Fatal(err) + } + defer func() { _ = seed.Close() }() + + const a = "{l22-a}.x" + ownerAddr, ownerID, otherAddr, otherID := clusterOwnerOf(ctx, t, seed, int(redis.Slot(a))) + // A channel of another slot that the same node serves. + var second string + for i := 0; i < 200 && second == ""; i++ { + cand := fmt.Sprintf("{l22-n2-%d}.y", i) + if redis.Slot(cand) == redis.Slot(a) { + continue + } + if addr, _, _, _ := clusterOwnerOf(ctx, t, seed, int(redis.Slot(cand))); addr == ownerAddr { + second = cand + } + } + if second == "" { + t.Fatal("test setup: no second slot on the owner of the first") + } + owner, err := redis.NewClient(ownerAddr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = owner.Close() }() + other, err := redis.NewClient(otherAddr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = other.Close() }() + + ps, err := cc.SSubscribe(ctx, a) + if err != nil { + t.Fatal(err) + } + defer func() { _ = ps.Close() }() + if err := ps.SSubscribe(ctx, second); err != nil { + t.Fatal(err) + } + time.Sleep(500 * time.Millisecond) + before := ssubConnIDs(ctx, t, owner) + if len(before) != 1 { + t.Fatalf("the owner has %d connections holding a shard subscription, want 1", len(before)) + } + + slot2 := int(redis.Slot(second)) + moveEmptySlot(ctx, t, owner, other, ownerID, otherID, slot2) + defer func() { + cctx, ccancel := context.WithTimeout(context.Background(), 20*time.Second) + defer ccancel() + moveEmptySlot(cctx, t, other, owner, otherID, ownerID, slot2) + }() + time.Sleep(1500 * time.Millisecond) + after := ssubConnIDs(ctx, t, owner) + if len(after) != 1 || after[0] != before[0] { + t.Errorf("the subscription's connection was %v, now %v: the move of a slot the PubSub does not serve rebuilt it", before, after) + } +} diff --git a/test/conformance/redis/pubsub_reroute_cluster_test.go b/test/conformance/redis/pubsub_reroute_cluster_test.go new file mode 100644 index 000000000..db3708d94 --- /dev/null +++ b/test/conformance/redis/pubsub_reroute_cluster_test.go @@ -0,0 +1,146 @@ +//go:build redis_cluster + +package redis_test + +import ( + "context" + "fmt" + "strconv" + "sync" + "testing" + "time" + + "github.com/goceleris/celeris/driver/redis" +) + +// clusterOwnerOf returns the host:port and the node id of the primary that owns +// slot, and those of another primary. +func clusterOwnerOf(ctx context.Context, t *testing.T, c *redis.Client, slot int) (ownerAddr, ownerID, otherAddr, otherID string) { + t.Helper() + v, err := c.Do(ctx, "CLUSTER", "SLOTS") + if err != nil { + t.Fatalf("CLUSTER SLOTS: %v", err) + } + for _, r := range v.Array { + if len(r.Array) < 3 || len(r.Array[2].Array) < 3 { + continue + } + p := r.Array[2].Array + addr := fmt.Sprintf("%s:%d", p[0].Str, p[1].Int) + id := string(p[2].Str) + if slot >= int(r.Array[0].Int) && slot <= int(r.Array[1].Int) { + ownerAddr, ownerID = addr, id + } else if otherAddr == "" { + otherAddr, otherID = addr, id + } + } + if ownerAddr == "" || otherAddr == "" { + t.Fatalf("CLUSTER SLOTS names no owner for slot %d or no second primary", slot) + } + return +} + +// moveEmptySlot moves slot from one primary to another, the way redis-cli does +// for a slot with no keys: IMPORTING on the target, MIGRATING on the source, +// then NODE on the target and on the source. +func moveEmptySlot(ctx context.Context, t *testing.T, from, to *redis.Client, fromID, toID string, slot int) { + t.Helper() + s := strconv.Itoa(slot) + for _, step := range []struct { + c *redis.Client + args []any + }{ + {to, []any{"CLUSTER", "SETSLOT", s, "IMPORTING", fromID}}, + {from, []any{"CLUSTER", "SETSLOT", s, "MIGRATING", toID}}, + {to, []any{"CLUSTER", "SETSLOT", s, "NODE", toID}}, + {from, []any{"CLUSTER", "SETSLOT", s, "NODE", toID}}, + } { + if _, err := step.c.Do(ctx, step.args...); err != nil { + t.Fatalf("%v: %v", step.args, err) + } + } +} + +// A shard subscription follows its slot on a real cluster: when the slot moves +// the server sends an unsolicited sunsubscribe and keeps the conn, and the +// PubSub subscribes again on the new owner and keeps delivering to the same +// channel (celeris#887 item 31). The slot goes back at the end. +func TestClusterSSubscribeFollowsASlotMove(t *testing.T) { + addrs := clusterAddrsFromEnv(t) + cc := newClusterClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + const ch = "{l22-reroute}.ch" + slot := int(redis.Slot(ch)) + seed, err := redis.NewClient(addrs[0]) + if err != nil { + t.Fatal(err) + } + defer func() { _ = seed.Close() }() + ownerAddr, ownerID, otherAddr, otherID := clusterOwnerOf(ctx, t, seed, slot) + owner, err := redis.NewClient(ownerAddr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = owner.Close() }() + other, err := redis.NewClient(otherAddr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = other.Close() }() + + ps, err := cc.SSubscribe(ctx, ch) + if err != nil { + t.Fatalf("SSubscribe: %v", err) + } + defer func() { _ = ps.Close() }() + var mu sync.Mutex + got := map[string]int{} + go func() { + for m := range ps.Channel() { + mu.Lock() + got[m.Payload]++ + mu.Unlock() + } + }() + count := func(p string) int { mu.Lock(); defer mu.Unlock(); return got[p] } + + // publishUntil publishes distinct payloads every 100 ms until one arrives: + // what is published while the PubSub switches is lost, so one message + // does not prove anything. + publishUntil := func(prefix string) []string { + var sent []string + deadline := time.Now().Add(30 * time.Second) + for i := 0; time.Now().Before(deadline); i++ { + p := fmt.Sprintf("%s-%d", prefix, i) + sent = append(sent, p) + _, _ = cc.SPublish(ctx, ch, p) + time.Sleep(100 * time.Millisecond) + for _, q := range sent { + if count(q) > 0 { + return sent + } + } + } + t.Fatalf("no message of %q arrived within 30 s", prefix) + return nil + } + publishUntil("before") + + moveEmptySlot(ctx, t, owner, other, ownerID, otherID, slot) + // The slot goes back before the clients above are closed (defers run in + // reverse order, a t.Cleanup would run after them). + defer func() { + cctx, ccancel := context.WithTimeout(context.Background(), 20*time.Second) + defer ccancel() + moveEmptySlot(cctx, t, other, owner, otherID, ownerID, slot) + }() + sent := publishUntil("after-move") + time.Sleep(300 * time.Millisecond) + for _, p := range sent { + if n := count(p); n > 1 { + t.Errorf("payload %q delivered %d times, want at most once", p, n) + } + } +}