From 523a68fd0b05207e715ede45de313e38ff300d2e Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 17:52:01 +0100 Subject: [PATCH 01/11] Fix races in cluster tests and shrink CI matrix The cluster suite failed in roughly one of six matrix cells per push, and fail-fast turned that single flake into a fully red run. TestMasterOnly asserted on a debug event within a fixed 800ms window, which races the background control loop: an iteration that read the master-only set before the test unset it emits the event after the events were reset. Both fallback-to-slave tests read a value from a replica right after writing it to the master, without waiting for asynchronous replication. TestAllReturns_ GoodMoving counted any error as a failure while running MIGRATE in a loop, although MIGRATE blocks both servers well beyond the 200ms IO timeout the test used. Test_justToCover relied on a hostname in a real TLD resolving to nothing. Cluster behaviour does not depend on the go version, so the cluster stage now runs on the oldest and the newest go only; the other stages keep the full matrix. --- .github/workflows/ci.yml | 14 ++++++- Makefile | 2 +- rediscluster/cluster_test.go | 71 ++++++++++++++++++++++++++++++++---- testbed/cluster.go | 4 +- 4 files changed, 79 insertions(+), 12 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7d10318..311efc3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,15 +2,28 @@ name: CI on: [push] +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + jobs: build: runs-on: ubuntu-latest + timeout-minutes: 25 strategy: + fail-fast: false matrix: go: ["1.19", "1.20", "1.21", "1.22", "1.23", "1.24"] stage: [testredis, testconn, testcluster] + exclude: + # Cluster suite spawns a 7-node redis cluster and takes ~7 minutes per + # matrix cell, while its behaviour does not depend on the go version. + - {go: "1.20", stage: testcluster} + - {go: "1.21", stage: testcluster} + - {go: "1.22", stage: testcluster} + - {go: "1.23", stage: testcluster} steps: @@ -43,4 +56,3 @@ jobs: - name: Build run: make ${{ matrix.stage }} - diff --git a/Makefile b/Makefile index 94f4e02..93c9ab0 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ test: testcluster testconn testredis rm redis-$(REDIS_VERSION) -rf testredis: /tmp/redis-server/redis-server - PATH=/tmp/redis-server/:${PATH} go test ./redis + PATH=/tmp/redis-server/:${PATH} go test -count 1 ./redis testconn: /tmp/redis-server/redis-server killall redis-server || true diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index d340c6f..37b0398 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -81,6 +81,59 @@ func (s *Suite) AsError(v interface{}) *errorx.Error { return v.(*errorx.Error) } +func containsEvent(events []string, event string) bool { + for _, ev := range events { + if ev == event { + return true + } + } + return false +} + +// waitDebugEvent waits for event to be emitted by cluster's background control loop. +func (s *Suite) waitDebugEvent(event string, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for !containsEvent(DebugEvents(), event) { + if time.Now().After(deadline) { + s.r().Contains(DebugEvents(), event) + } + time.Sleep(10 * time.Millisecond) + } +} + +// waitNoDebugEvent waits for a window of quiet duration without event being emitted. +// Control loop iteration started before the state change under test may still emit +// the event, so a single window is retried until timeout. +func (s *Suite) waitNoDebugEvent(event string, quiet, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for { + DebugEventsReset() + time.Sleep(quiet) + events := DebugEvents() + if !containsEvent(events, event) { + return + } + if time.Now().After(deadline) { + s.r().NotContains(events, event) + } + } +} + +// waitReplicated waits until master acknowledges that a replica caught up with it. +func (s *Suite) waitReplicated(master int, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for { + res := s.cl.Node[master].Do("WAIT", 1, 100) + if n, ok := res.(int64); ok && n >= 1 { + return + } + if time.Now().After(deadline) { + s.r().Failf("replica didn't catch up", "node %d: WAIT returned %v", master, res) + } + time.Sleep(10 * time.Millisecond) + } +} + var defopts = redisconn.Opts{ IOTimeout: 200 * time.Millisecond, } @@ -167,7 +220,7 @@ func (s *Suite) Test_justToCover() { opts.CheckInterval = 0 opts.MovedRetries = 11 opts.WaitToMigrate = time.Microsecond - cl, err = NewCluster(s.ctx, []string{"never-known-lost-my-host.badubadu.duba:43210"}, opts) + cl, err = NewCluster(s.ctx, []string{"no-such-host-xyzzy.invalid:43210"}, opts) s.r().Nil(cl) s.r().Error(err) @@ -365,7 +418,8 @@ func (s *Suite) TestFallbackToSlaveStop() { sconn := redis.SyncCtx{cl.WithPolicy(MasterAndSlaves)} key := slotkey("toslave", s.keys[1], "stop") - sconn.Do(s.ctx, "SET", key, "1") + s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) + s.waitReplicated(0, 5*time.Second) s.cl.Node[0].Stop() // test read from replica @@ -395,7 +449,8 @@ func (s *Suite) TestFallbackToSlaveTimeout() { sconn := redis.SyncCtx{cl.WithPolicy(PreferSlaves)} key := slotkey("toslave", s.keys[1], "timeout") - sconn.Do(s.ctx, "SET", key, "1") + s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) + s.waitReplicated(0, 5*time.Second) s.cl.Node[0].Pause() // test read from replica @@ -466,14 +521,11 @@ func (s *Suite) TestMasterOnly() { err = redisclusterutil.SetMasterOnly(cl, "", []uint16{1, 2}) s.r().Nil(err) - time.Sleep(clustopts.CheckInterval * 2) - s.Contains(DebugEvents(), "automatic masteronly") + s.waitDebugEvent("automatic masteronly", 10*time.Second) err = redisclusterutil.UnsetMasterOnly(cl, "", []uint16{1, 2}) s.r().Nil(err) - DebugEventsReset() - time.Sleep(clustopts.CheckInterval * 2) - s.NotContains(DebugEvents(), "automatic masteronly") + s.waitNoDebugEvent("automatic masteronly", clustopts.CheckInterval*2, 10*time.Second) }() } } @@ -678,6 +730,9 @@ Loop: func (s *Suite) TestAllReturns_GoodMoving() { opts := clustopts opts.CheckInterval = 4 * time.Second + // MIGRATE blocks both source and destination server, and this test runs it + // in a loop against a cluster loaded with N goroutines. + opts.HostOpts.IOTimeout = 2 * time.Second cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) s.r().Nil(err) defer cl.Close() diff --git a/testbed/cluster.go b/testbed/cluster.go index 1ba738e..b58c710 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -41,7 +41,7 @@ func NewCluster(startport uint16) *Cluster { cl.Node[i].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[i].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "500", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", @@ -262,7 +262,7 @@ func (cl *Cluster) StartSeventhNode() { cl.Node[6].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[6].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "500", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", From e54f090581309b1ec85ea3c7425d285810d5ccc5 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 17:52:12 +0100 Subject: [PATCH 02/11] Add temporary cluster stress job --- .github/workflows/ci.yml | 42 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 311efc3..544bb6f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -56,3 +56,45 @@ jobs: - name: Build run: make ${{ matrix.stage }} + + # TEMPORARY: flake-rate measurement, remove before merge. + cluster-stress: + + runs-on: ubuntu-latest + timeout-minutes: 25 + + strategy: + fail-fast: false + matrix: + run: [1, 2, 3, 4, 5, 6] + + steps: + + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: "1.24" + + - name: Cache go modules + uses: actions/cache@v4 + with: + path: ~/go/pkg/mod + key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} + restore-keys: | + ${{ runner.os }}-go- + + - name: Cache redis build + uses: actions/cache@v4 + with: + path: | + /tmp/redis-server + key: ${{ runner.os }}-redis-${{ hashFiles('Makefile') }} + + - name: Install redis + run: make /tmp/redis-server/redis-server + + - name: Build + run: make testcluster From 75663beb828691faecea5b9bf7af5340f9ef875b Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 18:10:53 +0100 Subject: [PATCH 03/11] Wait for replication before reading filled keys from replicas TestAllReturns_Good and TestAllReturns_Bad read with MasterAndSlaves policy keys that fillMany wrote to masters 10ms earlier, so a replica lagging behind answered with a miss. In TestAllReturns_Good such a goroutine returned without signalling the channel the test waits on, which turned the mismatch into a 10 minute test binary timeout instead of a failed assertion. Revert the cluster node timeout back to 200ms: the replication race explains the observed failures, and slower failure detection only prolongs the tests that stop nodes. --- rediscluster/cluster_test.go | 38 +++++++++++++++++++++++++----------- testbed/cluster.go | 4 ++-- 2 files changed, 29 insertions(+), 13 deletions(-) diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index 37b0398..2ae3586 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -119,16 +119,27 @@ func (s *Suite) waitNoDebugEvent(event string, quiet, timeout time.Duration) { } } -// waitReplicated waits until master acknowledges that a replica caught up with it. -func (s *Suite) waitReplicated(master int, timeout time.Duration) { +// waitReplicated waits until a replica acknowledges writes to the shard serving slot. +// Node roles are not stable across tests, so the master is the node that accepts a write. +func (s *Suite) waitReplicated(slot int, timeout time.Duration) { + // WAIT only accounts for writes issued by the calling connection, hence the probe. + probe := slotkey("replprobe", s.keys[slot]) deadline := time.Now().Add(timeout) for { - res := s.cl.Node[master].Do("WAIT", 1, 100) - if n, ok := res.(int64); ok && n >= 1 { - return + for i := range s.cl.Node { + node := &s.cl.Node[i] + if !node.RunningNow() { + continue + } + if redis.AsError(node.Do("SET", probe, "1")) != nil { + continue + } + if n, ok := node.Do("WAIT", 1, 100).(int64); ok && n >= 1 { + return + } } if time.Now().After(deadline) { - s.r().Failf("replica didn't catch up", "node %d: WAIT returned %v", master, res) + s.r().Fail("no replica caught up with master of slot " + strconv.Itoa(slot)) } time.Sleep(10 * time.Millisecond) } @@ -419,7 +430,7 @@ func (s *Suite) TestFallbackToSlaveStop() { key := slotkey("toslave", s.keys[1], "stop") s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) - s.waitReplicated(0, 5*time.Second) + s.waitReplicated(1, 30*time.Second) s.cl.Node[0].Stop() // test read from replica @@ -450,7 +461,7 @@ func (s *Suite) TestFallbackToSlaveTimeout() { key := slotkey("toslave", s.keys[1], "timeout") s.r().Equal("OK", sconn.Do(s.ctx, "SET", key, "1")) - s.waitReplicated(0, 5*time.Second) + s.waitReplicated(1, 30*time.Second) s.cl.Node[0].Pause() // test read from replica @@ -661,11 +672,16 @@ func (s *Suite) fillMany(sconn redis.SyncCtx, prefix string) { for _, res := range ress { s.r().Equal("OK", res) } - time.Sleep(10 * time.Millisecond) + // Tests read filled keys with MasterAndSlaves policy. + for _, slot := range []int{0, 5500, 11000} { + s.waitReplicated(slot, 30*time.Second) + } } func (s *Suite) TestAllReturns_Good() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + opts := clustopts + opts.HostOpts.IOTimeout = 2 * time.Second + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) s.r().Nil(err) defer cl.Close() @@ -679,6 +695,7 @@ func (s *Suite) TestAllReturns_Good() { for i := 0; i < N; i++ { go func(i int) { + defer func() { ch <- struct{}{} }() for j := 0; j < K; j++ { skey := s.keys[(i*N+j)*127%NumSlots] key := slotkey("allgood", skey) @@ -709,7 +726,6 @@ func (s *Suite) TestAllReturns_Good() { return } } - ch <- struct{}{} }(i) } diff --git a/testbed/cluster.go b/testbed/cluster.go index b58c710..1ba738e 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -41,7 +41,7 @@ func NewCluster(startport uint16) *Cluster { cl.Node[i].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[i].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "500", + "--cluster-node-timeout", "200", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", @@ -262,7 +262,7 @@ func (cl *Cluster) StartSeventhNode() { cl.Node[6].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[6].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "500", + "--cluster-node-timeout", "200", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", From aaa29b41beb47f60cedf4659c96212f88d9a43da Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 18:25:42 +0100 Subject: [PATCH 04/11] Tolerate connectivity errors while slots migrate A shard has no connection to serve a request for a short while after the slot it serves moves to a node the client has not connected to yet, so TestAllReturns_GoodMoving now retries such requests instead of counting them as wrong answers. Its load is also lowered: with 400 goroutines saturating the runner, redis did not converge between the moves and the testbed gave up waiting. An explicit ReconnectPause keeps the dead window after a broken connection short, since it otherwise follows IOTimeout, which both TestAllReturns tests raise to survive a MIGRATE blocking the server they talk to. --- rediscluster/cluster_test.go | 47 +++++++++++++++++++++++++++++++++--- testbed/cluster.go | 2 +- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index 2ae3586..c61e985 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -177,6 +177,41 @@ func (s *Suite) TestConnectDisconnected() { s.r().NotNil(err) } +func isConnectivityError(res interface{}) bool { + rerr := redis.AsErrorx(res) + return rerr != nil && rerr.HasTrait(redis.ErrTraitConnectivity) +} + +// doRetrying repeats a request while it fails with a connectivity error, which a +// shard is allowed to answer with while slots migrate between nodes. +func (s *Suite) doRetrying(sconn redis.SyncCtx, ctx context.Context, cmd string, args ...interface{}) interface{} { + var res interface{} + for i := 0; i < 20; i++ { + if res = sconn.Do(ctx, cmd, args...); !isConnectivityError(res) { + break + } + time.Sleep(50 * time.Millisecond) + } + return res +} + +// sendManyRetrying is doRetrying for a batch of requests. +func (s *Suite) sendManyRetrying(sconn redis.SyncCtx, ctx context.Context, reqs []redis.Request) []interface{} { + var ress []interface{} + for i := 0; i < 20; i++ { + ress = sconn.SendMany(ctx, reqs) + retry := false + for _, res := range ress { + retry = retry || isConnectivityError(res) + } + if !retry { + break + } + time.Sleep(50 * time.Millisecond) + } + return ress +} + func slotkey(prefix, slot string, suffix ...string) string { if len(suffix) == 0 { return prefix + "{" + slot + "}" @@ -681,6 +716,8 @@ func (s *Suite) fillMany(sconn redis.SyncCtx, prefix string) { func (s *Suite) TestAllReturns_Good() { opts := clustopts opts.HostOpts.IOTimeout = 2 * time.Second + // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. + opts.HostOpts.ReconnectPause = 100 * time.Millisecond cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) s.r().Nil(err) defer cl.Close() @@ -749,6 +786,8 @@ func (s *Suite) TestAllReturns_GoodMoving() { // MIGRATE blocks both source and destination server, and this test runs it // in a loop against a cluster loaded with N goroutines. opts.HostOpts.IOTimeout = 2 * time.Second + // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. + opts.HostOpts.ReconnectPause = 100 * time.Millisecond cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) s.r().Nil(err) defer cl.Close() @@ -761,7 +800,9 @@ func (s *Suite) TestAllReturns_GoodMoving() { log.Println("Started seventh") defer s.cl.StopSeventhNode() - const N = 400 + // Migrations need the cluster to converge between the moves, which it does not + // do while every core is busy serving the load this test generates. + const N = 100 ch := make(chan struct{}, N) var good uint32 var bad uint32 @@ -774,7 +815,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { for j := 0; atomic.LoadUint32(&stop) == 0; j++ { skey := s.keys[(i*N+j)*127%NumSlots] key := slotkey("allgoodmove", skey) - res := sconn.Do(ctx, "GET", key) + res := s.doRetrying(sconn, ctx, "GET", key) if !s.Equal([]byte(skey), res) { log.Println("Res ", res) atomic.AddUint32(&bad, 1) @@ -793,7 +834,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { redis.Req("SET", keya, keyb), redis.Req("GET", keyb), } - ress := sconn.SendMany(ctx, reqs) + ress := s.sendManyRetrying(sconn, ctx, reqs) if !s.Equal("OK", ress[0]) { log.Println("Ress[0] ", ress[0]) diff --git a/testbed/cluster.go b/testbed/cluster.go index 1ba738e..2d6d2ac 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -105,7 +105,7 @@ func RaiseClusterPanic() { // WaitClusterOk wait for cluster configuration to be stable. func (cl *Cluster) WaitClusterOk() { i := 0 - t := time.AfterFunc(30*time.Second, RaiseClusterPanic) + t := time.AfterFunc(60*time.Second, RaiseClusterPanic) defer t.Stop() for !cl.ClusterOk() { if i++; i == 10 { From bb08e97735af48fb15c9182a9f09f20b78520b22 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 18:36:38 +0100 Subject: [PATCH 05/11] Keep the load of TestAllReturns_GoodMoving at 400 goroutines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The keys a goroutine walks are derived from the goroutine count, so lowering it also lowers how often the migrating slots are requested — with 100 the test stopped seeing MOVED at all. --- rediscluster/cluster_test.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index c61e985..c1a6cb1 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -800,9 +800,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { log.Println("Started seventh") defer s.cl.StopSeventhNode() - // Migrations need the cluster to converge between the moves, which it does not - // do while every core is busy serving the load this test generates. - const N = 100 + const N = 400 ch := make(chan struct{}, N) var good uint32 var bad uint32 From 0e815610608ca35c8482e1fbc2693dc8690e97e5 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 18:48:24 +0100 Subject: [PATCH 06/11] Give the cluster bus and the TLS handshake room on a loaded runner With a 200ms node timeout the TLS cluster failed to take the seventh node in within a minute, since nodes declare each other failed faster than the bus finishes its handshakes when the runner is busy. For the same reason a TLS connection could not answer PING within the 200ms the plain tests use. WaitClusterOk now reports what every node thought about the cluster before it gives up, which is otherwise unrecoverable from CI logs. --- redisconn/conn_test.go | 6 ++++-- testbed/cluster.go | 29 ++++++++++++++++++++++++++--- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/redisconn/conn_test.go b/redisconn/conn_test.go index d0d4f8f..22c4fe9 100644 --- a/redisconn/conn_test.go +++ b/redisconn/conn_test.go @@ -115,8 +115,10 @@ func (s *Suite) TestConnects() { } func (s *Suite) TestConnectsTls() { + // A TLS handshake does not fit into the IO timeout the other tests use. + const tlsTimeout = time.Second tlsopts := Opts{ - IOTimeout: defopts.IOTimeout, + IOTimeout: tlsTimeout, TLSEnabled: true, TLSConfig: &tls.Config{ InsecureSkipVerify: true, @@ -125,7 +127,7 @@ func (s *Suite) TestConnectsTls() { conn, err := Connect(s.ctx, s.s.TlsAddr(), tlsopts) s.r().Nil(err) defer conn.Close() - s.goodPing(conn, 0) + s.goodPing(conn, tlsTimeout) } func (s *Suite) TestConnectsDb() { diff --git a/testbed/cluster.go b/testbed/cluster.go index 2d6d2ac..d0fbdb1 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -7,6 +7,7 @@ import ( "time" "github.com/joomcode/redispipe/rediscluster/redisclusterutil" + "github.com/joomcode/redispipe/redisdumb" ) // Node is wrapper for Server with its NodeId @@ -41,7 +42,7 @@ func NewCluster(startport uint16) *Cluster { cl.Node[i].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[i].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "1000", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", @@ -102,10 +103,32 @@ func RaiseClusterPanic() { panic("cluster didn't stabilize") } +// DumpState reports what every node thinks about the cluster. +func (cl *Cluster) DumpState() { + for i := range cl.Node { + n := &cl.Node[i] + if !n.RunningNow() { + log.Printf("node %d (%s): not running", i, n.Addr()) + continue + } + // The node's own connection belongs to the goroutine that polls it. + conn := redisdumb.Conn{ + Addr: n.Conn.Addr, + TlsAddr: n.Conn.TlsAddr, + TLSEnabled: n.Conn.TLSEnabled, + TLSConfig: n.Conn.TLSConfig, + } + log.Printf("node %d (%s): %v\n%v", i, n.Addr(), conn.Do("CLUSTER INFO"), conn.Do("CLUSTER NODES")) + } +} + // WaitClusterOk wait for cluster configuration to be stable. func (cl *Cluster) WaitClusterOk() { i := 0 - t := time.AfterFunc(60*time.Second, RaiseClusterPanic) + t := time.AfterFunc(60*time.Second, func() { + cl.DumpState() + RaiseClusterPanic() + }) defer t.Stop() for !cl.ClusterOk() { if i++; i == 10 { @@ -262,7 +285,7 @@ func (cl *Cluster) StartSeventhNode() { cl.Node[6].Args = []string{ "--cluster-enabled", "yes", "--cluster-config-file", "node-" + cl.Node[6].PortStr(effectivePort) + ".conf", - "--cluster-node-timeout", "200", + "--cluster-node-timeout", "1000", "--cluster-slave-validity-factor", "1000", "--slave-serve-stale-data", "yes", "--cluster-require-full-coverage", "no", From 4cc65d9361be72a7f877d1cfd72007050ce0a125 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 19:01:23 +0100 Subject: [PATCH 07/11] Move test redis ports out of the ephemeral port range Linux hands out 32768-60999 to outgoing connections, and the tests open hundreds of them, so a client socket could take 43216 before the seventh node tried to listen on it. Ports are chosen below the range now, cluster bus ports (port + 10000) included. --- redis/example_test.go | 10 +++---- rediscluster/bench/bench_test.go | 16 +++++------ rediscluster/cluster_test.go | 48 ++++++++++++++++---------------- redisconn/bench/bench_test.go | 16 +++++------ redisconn/conn_test.go | 4 +-- testbed/cluster.go | 10 ++++++- 6 files changed, 56 insertions(+), 48 deletions(-) diff --git a/redis/example_test.go b/redis/example_test.go index fc82c5a..d65a3f0 100644 --- a/redis/example_test.go +++ b/redis/example_test.go @@ -51,9 +51,9 @@ func ExampleAsError() { } func ExampleScanner() { - defer runServer(46231)() + defer runServer(21050)() ctx := context.Background() - conn, _ := redisconn.Connect(ctx, "127.0.0.1:46231", redisconn.Opts{ + conn, _ := redisconn.Connect(ctx, "127.0.0.1:21050", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) sync := redis.Sync{conn} @@ -79,9 +79,9 @@ func ExampleScanner() { } func ExampleSync() { - defer runServer(46231)() + defer runServer(21050)() ctx := context.Background() - conn, _ := redisconn.Connect(ctx, "127.0.0.1:46231", redisconn.Opts{ + conn, _ := redisconn.Connect(ctx, "127.0.0.1:21050", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) sync := redis.Sync{conn} @@ -114,7 +114,7 @@ func ExampleSync() { // OK // OK // ["1" "2"] - // redispipe.result: WRONGTYPE Operation against a key holding the wrong kind of value {request: Req("HSET", ["key1" "field1" "val1"]), address: 127.0.0.1:46231} + // redispipe.result: WRONGTYPE Operation against a key holding the wrong kind of value {request: Req("HSET", ["key1" "field1" "val1"]), address: 127.0.0.1:21050} // // ['\x02' '\x01' "2" "1"] } diff --git a/rediscluster/bench/bench_test.go b/rediscluster/bench/bench_test.go index a57c943..4e6e97e 100644 --- a/rediscluster/bench/bench_test.go +++ b/rediscluster/bench/bench_test.go @@ -29,11 +29,11 @@ func benchCluster(port int) func() { } func BenchmarkSerialGetSet(b *B) { - defer benchCluster(45000)() + defer benchCluster(21080)() rng := rand.New(rand.NewSource(1)) b.Run("radix_pause0", func(b *B) { rdxv2, err := radix.NewCluster( - []string{"127.0.0.1:45000"}, + []string{"127.0.0.1:21080"}, radix.ClusterPoolFunc(func(network, addr string) (radix.Client, error) { return radix.NewPool(network, addr, 4, radix.PoolPipelineWindow(0, 0)) @@ -72,7 +72,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -96,7 +96,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe_pause0", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -122,7 +122,7 @@ func BenchmarkSerialGetSet(b *B) { } func BenchmarkParallelGetSet(b *B) { - defer benchCluster(45000)() + defer benchCluster(21080)() parallel := runtime.GOMAXPROCS(0) * 8 i := uint32(1) @@ -137,7 +137,7 @@ func BenchmarkParallelGetSet(b *B) { } b.Run("radix", func(b *B) { - rdx2, err := radix.NewCluster([]string{"127.0.0.1:45000"}) + rdx2, err := radix.NewCluster([]string{"127.0.0.1:21080"}) defer rdx2.Close() if err != nil { b.Fatal(err) @@ -170,7 +170,7 @@ func BenchmarkParallelGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:45000"}, rediscluster.Opts{ + pipe, err := rediscluster.NewCluster(context.Background(), []string{"127.0.0.1:21080"}, rediscluster.Opts{ Logger: rediscluster.NoopLogger{}, HostOpts: redisconn.Opts{ Logger: redisconn.NoopLogger{}, @@ -196,7 +196,7 @@ func BenchmarkParallelGetSet(b *B) { func newRedigo() *redigo.Cluster { c, err := redigo.NewCluster(&redigo.Options{ - StartNodes: []string{"127.0.0.1:45000"}, + StartNodes: []string{"127.0.0.1:21080"}, ConnTimeout: time.Minute, KeepAlive: 128, AliveTime: time.Minute, diff --git a/rediscluster/cluster_test.go b/rediscluster/cluster_test.go index c1a6cb1..3c6bf47 100644 --- a/rediscluster/cluster_test.go +++ b/rediscluster/cluster_test.go @@ -39,7 +39,7 @@ type Suite struct { func (s *Suite) SetupSuite() { testbed.InitDir(".") - s.cl = testbed.NewCluster(43210) + s.cl = testbed.NewCluster(21100) s.keys = make([]string, NumSlots) cnt := 0 for i := 0; cnt < NumSlots; i++ { @@ -173,7 +173,7 @@ func TestCluster(t *testing.T) { func (s *Suite) TestConnectDisconnected() { s.cl.Stop() - _, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + _, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().NotNil(err) } @@ -236,7 +236,7 @@ func (s *Suite) slotnode(slot int) *testbed.Node { } func (s *Suite) TestBasicOps() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() scl := redis.SyncCtx{cl} @@ -266,21 +266,21 @@ func (s *Suite) Test_justToCover() { opts.CheckInterval = 0 opts.MovedRetries = 11 opts.WaitToMigrate = time.Microsecond - cl, err = NewCluster(s.ctx, []string{"no-such-host-xyzzy.invalid:43210"}, opts) + cl, err = NewCluster(s.ctx, []string{"no-such-host-xyzzy.invalid:21100"}, opts) s.r().Nil(cl) s.r().Error(err) opts.CheckInterval = 11 * time.Minute opts.MovedRetries = 1 opts.WaitToMigrate = time.Second - cl, err = NewCluster(s.ctx, []string{"127.0.0.1:43200"}, opts) + cl, err = NewCluster(s.ctx, []string{"127.0.0.1:21090"}, opts) s.r().Nil(cl) s.r().Error(err) opts = clustopts opts.ConnsPerHost = 1 opts.Handle = new(struct{}) - cl, err = NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err = NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -336,7 +336,7 @@ func (c *cancelledFuture) Resolve(res interface{}, n uint64) { } func (s *Suite) TestSendMany() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() scl := redis.SyncCtx{cl} @@ -366,7 +366,7 @@ func (s *Suite) TestSendMany() { func (s *Suite) TestTransactionNormal() { // copy fo connection test - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -417,7 +417,7 @@ func (s *Suite) TestScan() { opts := clustopts opts.HostOpts.IOTimeout = time.Second - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -457,7 +457,7 @@ func (a alwaysZero) Current() uint32 { func (s *Suite) TestFallbackToSlaveStop() { opts := longcheckopts opts.RoundRobinSeed = alwaysZero{} - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -488,7 +488,7 @@ func (s *Suite) TestFallbackToSlaveStop() { func (s *Suite) TestFallbackToSlaveTimeout() { opts := longcheckopts opts.RoundRobinSeed = alwaysZero{} - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -519,7 +519,7 @@ func (s *Suite) TestFallbackToSlaveTimeout() { } func (s *Suite) TestGetMoved() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -537,7 +537,7 @@ func (s *Suite) TestGetMoved() { } func (s *Suite) TestSetMoved() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -557,7 +557,7 @@ func (s *Suite) TestSetMoved() { } func (s *Suite) TestMasterOnly() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -577,7 +577,7 @@ func (s *Suite) TestMasterOnly() { } func (s *Suite) TestAsk() { - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, longcheckopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, longcheckopts) s.r().Nil(err) defer cl.Close() @@ -606,7 +606,7 @@ func (s *Suite) TestAskTransaction() { opts := longcheckopts opts.MovedRetries = 4 - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -673,7 +673,7 @@ func (s *Suite) TestMovedTransaction() { opts := longcheckopts opts.MovedRetries = 4 - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -718,7 +718,7 @@ func (s *Suite) TestAllReturns_Good() { opts.HostOpts.IOTimeout = 2 * time.Second // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. opts.HostOpts.ReconnectPause = 100 * time.Millisecond - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -788,7 +788,7 @@ func (s *Suite) TestAllReturns_GoodMoving() { opts.HostOpts.IOTimeout = 2 * time.Second // ReconnectPause defaults to DialTimeout*2, which follows IOTimeout. opts.HostOpts.ReconnectPause = 100 * time.Millisecond - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, opts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, opts) s.r().Nil(err) defer cl.Close() @@ -878,7 +878,7 @@ func (s *Suite) TestAllReturns_Bad() { s.ctx, s.ctxcancel = context.WithTimeout(context.Background(), 10*time.Minute) DebugDisable = true - cl, err := NewCluster(s.ctx, []string{"127.0.0.1:43210"}, clustopts) + cl, err := NewCluster(s.ctx, []string{"127.0.0.1:21100"}, clustopts) s.r().Nil(err) defer cl.Close() @@ -1034,9 +1034,9 @@ Loop: func (s *Suite) TestConnectWithDeadAddresses() { addrs := []string{ - "127.0.0.1:43200", // dead - "127.0.0.1:43210", // live - "127.0.0.1:43201", // dead + "127.0.0.1:21090", // dead + "127.0.0.1:21100", // live + "127.0.0.1:21091", // dead } cl, err := NewCluster(s.ctx, addrs, clustopts) s.Nil(err) @@ -1051,7 +1051,7 @@ func (s *Suite) TestConnectWithDeadAddresses() { func (s *Suite) TestConnectWithUnresolvableAddresses() { addrs := []string{ "no-such-host-xyzzy.invalid:6379", // guaranteed NXDOMAIN - "127.0.0.1:43210", // live + "127.0.0.1:21100", // live } cl, err := NewCluster(s.ctx, addrs, clustopts) s.Nil(err) diff --git a/redisconn/bench/bench_test.go b/redisconn/bench/bench_test.go index 144b434..8b711f3 100644 --- a/redisconn/bench/bench_test.go +++ b/redisconn/bench/bench_test.go @@ -25,9 +25,9 @@ func benchServer(port int) func() { } func BenchmarkSerialGetSet(b *B) { - defer benchServer(45678)() + defer benchServer(21070)() b.Run("radix", func(b *B) { - rdxv2, err := radix.Dial("tcp", "127.0.0.1:45678") + rdxv2, err := radix.Dial("tcp", "127.0.0.1:21070") if err != nil { b.Fatal(err) return @@ -59,7 +59,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) defer pipe.Close() @@ -79,7 +79,7 @@ func BenchmarkSerialGetSet(b *B) { }) b.Run("redispipe_pause0", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, WritePause: -1, }) @@ -101,7 +101,7 @@ func BenchmarkSerialGetSet(b *B) { } func BenchmarkParallelGetSet(b *B) { - defer benchServer(45678)() + defer benchServer(21070)() parallel := runtime.GOMAXPROCS(0) * 2 do := func(b *B, fn func()) { @@ -114,7 +114,7 @@ func BenchmarkParallelGetSet(b *B) { } b.Run("radix", func(b *B) { - rdx2, err := radix.NewPool("tcp", "127.0.0.1:45678", parallel) + rdx2, err := radix.NewPool("tcp", "127.0.0.1:21070", parallel) if err != nil { b.Fatal(err) } @@ -152,7 +152,7 @@ func BenchmarkParallelGetSet(b *B) { }) b.Run("redispipe", func(b *B) { - pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:45678", redisconn.Opts{ + pipe, err := redisconn.Connect(context.Background(), "127.0.0.1:21070", redisconn.Opts{ Logger: redisconn.NoopLogger{}, }) if err != nil { @@ -173,7 +173,7 @@ func BenchmarkParallelGetSet(b *B) { } func newRedigo() redigo.Conn { - c, err := redigo.Dial("tcp", "127.0.0.1:45678") + c, err := redigo.Dial("tcp", "127.0.0.1:21070") if err != nil { panic(err) } diff --git a/redisconn/conn_test.go b/redisconn/conn_test.go index 22c4fe9..0bcd7c3 100644 --- a/redisconn/conn_test.go +++ b/redisconn/conn_test.go @@ -29,8 +29,8 @@ type Suite struct { func (s *Suite) SetupSuite() { testbed.InitDir(".") - s.s.Port = 45678 - s.s.TlsPort = 55678 + s.s.Port = 21060 + s.s.TlsPort = 21061 s.s.Start() } diff --git a/testbed/cluster.go b/testbed/cluster.go index d0fbdb1..4dde360 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -3,6 +3,7 @@ package testbed import ( "bytes" "crypto/tls" + "fmt" "log" "time" @@ -103,6 +104,13 @@ func RaiseClusterPanic() { panic("cluster didn't stabilize") } +func dumpResult(res interface{}) string { + if buf, ok := res.([]byte); ok { + return string(buf) + } + return fmt.Sprintf("%v", res) +} + // DumpState reports what every node thinks about the cluster. func (cl *Cluster) DumpState() { for i := range cl.Node { @@ -118,7 +126,7 @@ func (cl *Cluster) DumpState() { TLSEnabled: n.Conn.TLSEnabled, TLSConfig: n.Conn.TLSConfig, } - log.Printf("node %d (%s): %v\n%v", i, n.Addr(), conn.Do("CLUSTER INFO"), conn.Do("CLUSTER NODES")) + log.Printf("node %d (%s): %s\n%s", i, n.Addr(), dumpResult(conn.Do("CLUSTER INFO")), dumpResult(conn.Do("CLUSTER NODES"))) } } From 105c944b4a544071621c9a16ca4bd2aff1b45128 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 19:13:16 +0100 Subject: [PATCH 08/11] Report what keeps the test cluster from stabilizing The dump of every node's view does not say which of the conditions WaitClusterOk polls for was not met, and the ones it polls for are not observable after the fact. --- testbed/cluster.go | 42 ++++++++++++++++++++++++++++++------------ 1 file changed, 30 insertions(+), 12 deletions(-) diff --git a/testbed/cluster.go b/testbed/cluster.go index 4dde360..d7a8334 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "fmt" "log" + "sync/atomic" "time" "github.com/joomcode/redispipe/rediscluster/redisclusterutil" @@ -20,6 +21,8 @@ type Node struct { // Cluster is a tool for starting/stopping redis cluster for tests. type Cluster struct { Node []Node + + lastNotOk atomic.Value } // NewCluster instantiate cluster of 6 nodes (3 masters and 3 slaves). @@ -113,6 +116,9 @@ func dumpResult(res interface{}) string { // DumpState reports what every node thinks about the cluster. func (cl *Cluster) DumpState() { + if reason, ok := cl.lastNotOk.Load().(string); ok { + log.Printf("cluster is not ok: %s", reason) + } for i := range cl.Node { n := &cl.Node[i] if !n.RunningNow() { @@ -148,6 +154,14 @@ func (cl *Cluster) WaitClusterOk() { // ClusterOk checks cluster configuration. func (cl *Cluster) ClusterOk() bool { + reason := cl.clusterNotOkReason() + cl.lastNotOk.Store(reason) + return reason == "" +} + +// clusterNotOkReason returns an empty string when the cluster configuration is stable, +// and what stands in the way otherwise. +func (cl *Cluster) clusterNotOkReason() string { stopped := []int{} for i := range cl.Node { if !cl.Node[i].RunningNow() { @@ -155,6 +169,7 @@ func (cl *Cluster) ClusterOk() bool { } } var hashsum uint64 + var hashnode int for i := range cl.Node { if !cl.Node[i].RunningNow() { continue @@ -162,24 +177,24 @@ func (cl *Cluster) ClusterOk() bool { res := cl.Node[i].Do("CLUSTER INFO") buf, ok := res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: CLUSTER INFO: %v", i, res) } if !bytes.Contains(buf, []byte("cluster_state:ok")) { - return false + return fmt.Sprintf("node %d: cluster state is not ok", i) } res = cl.Node[i].Do("INFO REPLICATION") buf, ok = res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: INFO REPLICATION: %v", i, res) } if !bytes.Contains(buf, []byte("role:master")) && !bytes.Contains(buf, []byte("master_link_status:up")) { - return false + return fmt.Sprintf("node %d: replica is not in sync with its master", i) } res = cl.Node[i].Do("CLUSTER NODES") buf, ok = res.([]byte) if !ok { - return false + return fmt.Sprintf("node %d: CLUSTER NODES: %v", i, res) } masters := 0 for _, line := range bytes.Split(buf, []byte("\n")) { @@ -187,7 +202,7 @@ func (cl *Cluster) ClusterOk() bool { for _, j := range stopped { if bytes.HasPrefix(line, cl.Node[j].NodeId) { if !bytes.Contains(line, []byte("fail ")) { - return false + return fmt.Sprintf("node %d: stopped node %d is not marked as failed", i, j) } hasStopped = true } @@ -196,18 +211,21 @@ func (cl *Cluster) ClusterOk() bool { masters++ } } - if masters != 3+(len(cl.Node)-6) { - return false - + if want := 3 + (len(cl.Node) - 6); masters != want { + return fmt.Sprintf("node %d: sees %d masters instead of %d", i, masters, want) + } + infos, err := redisclusterutil.ParseClusterNodes(res) + if err != nil { + return fmt.Sprintf("node %d: unparsable CLUSTER NODES: %v", i, err) } - infos, _ := redisclusterutil.ParseClusterNodes(res) hash := infos.HashSum() if hash != hashsum && hashsum != 0 { - return false + return fmt.Sprintf("node %d: sees a configuration different from node %d", i, hashnode) } hashsum = hash + hashnode = i } - return true + return "" } // AttemptFailover tries to issue CLUSTER FAILOVER FORCE to slaves of falled masters. From 0fd94dd7521d9f25b8ddf7b53cb44768ebfa9963 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 19:24:31 +0100 Subject: [PATCH 09/11] Tell every master who owns a slot after it moved Only the source and the destination were told, so a following migration of the same slot could outrun the gossip carrying the previous one. The cluster then had masters pointing at the new owner and the new owner pointing back, with nobody claiming the slot, and it never agreed on the configuration again. --- testbed/cluster.go | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/testbed/cluster.go b/testbed/cluster.go index d7a8334..bdf9ed3 100644 --- a/testbed/cluster.go +++ b/testbed/cluster.go @@ -266,8 +266,18 @@ func (cl *Cluster) CancelMoveSlot(slot int) { // FinishMoveSlot finalizes slot migration func (cl *Cluster) FinishMoveSlot(slot, from, to int) { - cl.Node[to].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) - cl.Node[from].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + cl.Node[to].DoSure("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + cl.Node[from].DoSure("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + // The rest of the masters would learn the new owner from the epoch bump below, + // but a next migration of the same slot may outrun it, and then neither the old + // nor the new owner claims the slot and the cluster never agrees again. + // Replicas answer with an error, which is why these are not DoSure. + for i := range cl.Node { + if i == to || i == from || !cl.Node[i].RunningNow() { + continue + } + cl.Node[i].Do("CLUSTER SETSLOT", slot, "NODE", cl.Node[to].NodeId) + } cl.Node[to].Do("CLUSTER BUMPEPOCH", "BROADCAST") // proprietary extension cl.Node[to].Do("CLUSTER BUMPEPOCH") } From 169f9ef20e2546b918e1895eea0e4a8fccfc38c1 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 19:55:08 +0100 Subject: [PATCH 10/11] Remove the temporary cluster stress job --- .github/workflows/ci.yml | 42 ---------------------------------------- 1 file changed, 42 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 544bb6f..311efc3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -56,45 +56,3 @@ jobs: - name: Build run: make ${{ matrix.stage }} - - # TEMPORARY: flake-rate measurement, remove before merge. - cluster-stress: - - runs-on: ubuntu-latest - timeout-minutes: 25 - - strategy: - fail-fast: false - matrix: - run: [1, 2, 3, 4, 5, 6] - - steps: - - - name: Checkout - uses: actions/checkout@v4 - - - name: Set up Go - uses: actions/setup-go@v5 - with: - go-version: "1.24" - - - name: Cache go modules - uses: actions/cache@v4 - with: - path: ~/go/pkg/mod - key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} - restore-keys: | - ${{ runner.os }}-go- - - - name: Cache redis build - uses: actions/cache@v4 - with: - path: | - /tmp/redis-server - key: ${{ runner.os }}-redis-${{ hashFiles('Makefile') }} - - - name: Install redis - run: make /tmp/redis-server/redis-server - - - name: Build - run: make testcluster From b3a5bb99d7e558387b356fb817b46ce48ab68ff0 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Fri, 11 Sep 2026 20:30:05 +0100 Subject: [PATCH 11/11] Test with go up to 1.27 --- .github/workflows/ci.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 311efc3..020e117 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ jobs: strategy: fail-fast: false matrix: - go: ["1.19", "1.20", "1.21", "1.22", "1.23", "1.24"] + go: ["1.19", "1.20", "1.21", "1.22", "1.23", "1.24", "1.25", "1.26", "1.27"] stage: [testredis, testconn, testcluster] exclude: # Cluster suite spawns a 7-node redis cluster and takes ~7 minutes per @@ -24,6 +24,9 @@ jobs: - {go: "1.21", stage: testcluster} - {go: "1.22", stage: testcluster} - {go: "1.23", stage: testcluster} + - {go: "1.24", stage: testcluster} + - {go: "1.25", stage: testcluster} + - {go: "1.26", stage: testcluster} steps: