feat(cluster): add RouteByLatencyTolerance to spread reads across equally-close nodes (#3973)

* feat(cluster): add RouteByLatencyTolerance to spread reads across equally-close nodes RouteByLatency picks the strict minimum, so when several nodes are effectively equidistant every client converges on the same one and the rest take no read traffic. Latency is the mean of ten pings of a sub-millisecond operation, refreshed at most every 10s, so a difference well inside the measurement noise can decide the whole read load for the next interval. Observed on a sentinel-backed failover cluster client spread over three availability zones: GET rates across a five-replica set ran 97.7, 71.4, 7.6, 4.3 and 0.17 per second - a 590x spread - while per-node CPU stayed flat, because the split follows the ranking rather than any real difference in capacity. Zone locality worked; the distribution within a zone did not. RouteByLatencyTolerance widens the selection from "the fastest node" to "any node within this much of the fastest", round-robining between them with the existing ShardPicker. Nodes an availability zone away sit far outside any sensible tolerance, so locality is preserved. Zero, the default, keeps the current strict-minimum behaviour, and the option has no effect unless RouteByLatency is set. Also plumbed through FailoverOptions so sentinel users can reach it. Also fixes a latent bug in the same function: closestNonFailingNode was only assigned while scanning for a new minimum, so a healthy node that was not the outright fastest could be skipped entirely, and selection fell through to the all-nodes-are-failing path even though a healthy node existed. Healthy and overall minima are now tracked separately, covered by TestSlotClosestNodePrefersSlowerHealthyNode. * fix(cluster): sample latency once, and expose tolerance through the remaining option paths Addresses the review on #3973. The tolerance filter read Latency() a second time, after the pass that computed the minimum. The background probe updates latency concurrently, so the fastest node could move above minLatency+tolerance between the two reads, leaving the candidate set empty. slotClosestNode then returned (nil, nil) and callers dereferenced node.Client. Latency and health are now sampled once per node and the filter uses the captured values, so the candidate set always contains at least the node that established the minimum. The comparison is also written as a subtraction rather than minLatency+tolerance, which would overflow for a very large tolerance. TestSlotClosestNodeToleranceNeverReturnsNilWithHealthyNode churns latencies from a second goroutine while selecting, and fails on the previous two-pass implementation. The option is now also reachable from the three configuration paths that mirror RouteByLatency: route_by_latency_tolerance in ParseClusterURL and ParseFailoverURL - both parsers reject unknown keys, so the URL form previously failed outright - and UniversalOptions, which feeds both Cluster() and Failover(). * perf(cluster): skip tolerance sampling when the option is unset Addresses the review on #3973. The candidate slice was built on every command regardless of RouteByLatencyTolerance, then discarded by the tolerance <= 0 branch, so existing RouteByLatency users paid for work whose result was thrown away. Collection is now gated on tolerance > 0, and the slice is pre-sized when it is used, so the tolerance path does not grow it either. Latency is still read once per node in the same pass, so the guarantee from the previous commit holds: the candidate set cannot disagree with the minimum it is compared against. TestSlotClosestNodeToleranceDisabledDoesNotAllocate pins this with testing.AllocsPerRun. Without the gate it reports 3 allocations per call; with it, zero. * fix(cluster): keep tolerance rotation per slot Addresses the review on #3973. The tolerance path rotated through candidates with the single ClusterOptions.ShardPicker counter, which is shared by every slot. On a real cluster that counter is advanced by the other slots between two visits to the same one, so a regular interleaving leaves each slot landing on the same index every time - two slots with two candidates each alternate to 0,0,0... and 1,1,1..., and one replica per shard takes all of that shard's reads. The balancing this option exists to provide is then absent exactly where there is more than one slot to balance. Rotation now lives on clusterSlot, so each slot advances its own cursor. The state is rebuilt on every cluster-state reload, which is fine: it only decides which of several equally-close replicas answers next. slotEntry is factored out of slotNodes to reach the slot rather than just its nodes. TestSlotClosestNodeToleranceRotatesPerSlot alternates two slots and asserts both candidates are used in each. With a shared counter it reports 100/0 for the first slot. * fix(cluster): reduce the tolerance rotation cursor in the unsigned domain The per-slot rotation cursor is a uint32 and wraps. Converting it to int before the modulo is only safe while int is 64 bits: on a 32-bit build a cursor past 2^31 becomes negative, Go's % keeps the dividend's sign, and indexing candidates panics. Reproduced rather than reasoned about. With the cursor primed to math.MaxInt32, GOARCH=386 gives: panic: runtime error: index out of range [-1] osscluster.go:1068 Indexing with the uint32 directly keeps the reduction unsigned, and the same test passes on 386 and amd64. Only reachable once a slot has more than one in-band candidate and has served ~2^31 reads, so it needs both the tolerance option and a very hot slot -- but it is a hard panic in the read path, not degraded routing. * test(cluster): add the 6-node integration coverage for RouteByLatencyTolerance .claude/specs/cluster-routing.md:147-153 requires cluster-routing changes to add a test that spins the 6-node clusterScenario, uses keys verified to land on different shards, and exercises both the happy path and the redirect path. The tolerance option only had in-memory selector tests, so nothing checked it against real CLUSTER SLOTS and MOVED behaviour. Adds a Describe block mirroring the RouteByLatency one with a 1s tolerance, wide enough that every healthy node for a slot is in band and the per-slot rotation is what is under test. It calls assertClusterClient(), which already covers GET/SET/DEL, redirects for both via SwapNodes, and key distribution. The shard keys are discovered rather than hardcoded. keyN does not spread across the three masters -- key2 (slot 4998) and key3 (slot 935) both hash into 0-5460 -- and the first version of this test failed on exactly that. Hardcoding luckier keys would work today and silently stop covering three shards if the slot ranges ever moved, so the test walks candidates until it has one key per master and asserts it found three. Verified against the real cluster from docker-compose --profile all: 45 specs pass across both RouteByLatency blocks. * refactor(cluster): split the tolerance path out of slotClosestNode Addresses @ndyakov's review. slotClosestNode goes back to exactly what is on master - same signature, same body, byte-identical - and the tolerance logic moves to a new slotNodeWithinLatency(slot, tolerance). slotReadOnlyNode dispatches between them on RouteByLatencyTolerance > 0. This is better than the tolerance parameter it replaces: with the option unset, existing latency routing now runs literally the same code as before rather than the same code plus a branch, so there is nothing to argue about on the hot path. Also renames clusterSlot.rotation to nodesByLatencyRotation, so the field says what it is for. One consequence worth flagging: the earlier version of this PR also fixed a latent bug in slotClosestNode, where closestNonFailingNode is only updated inside the 'is this the new minimum' branch, so a healthy-but-slower node never wins over a faster failing one. Restoring the function byte-for-byte puts that bug back. It is pre-existing and unrelated to tolerance, so it does not belong in this PR - the test that covered it now points at slotNodeWithinLatency, which retains the fix. Happy to raise it separately. * fix(cluster): skip loading nodes in the tolerance band Codex caught a real gap. slotSlaveNode and slotShardPickerSlaveNode both filter on !Failing() && !Loading(); slotClosestNode only checks Failing(), and slotNodeWithinLatency inherited that. It matters more with a tolerance than without. Under a strict minimum a loading replica has to be the fastest node to be chosen at all; within a band it only has to be close, so widening the candidate set widens the window for picking a node that answers LOADING. And a replica is loading precisely after the full resync that a too-small replication backlog causes, so this is not a rare state. Filtered at the candidate stage rather than in the sampling pass, because Loading() costs a PING with a 100ms timeout whenever the node is not already known loaded - doing it per node per command would be a real cost, whereas doing it for the handful inside the band is bounded. If every candidate is loading the strict-minimum healthy node is returned, so the filter can never empty the set or return nil. Test helper now marks nodes loaded, which is what steady state looks like and keeps the suite off a 100ms dial per candidate. Added TestSlotNodeWithinLatencyKeepsUnconfirmedNode to pin the other direction: loaded=0 means 'not yet confirmed', not 'loading', and such a node must stay selectable - a connection error is not a LOADING reply. * fix(cluster): fall back outside the band when it is entirely loading Both Codex and Cursor Bugbot caught the same real bug in the loading filter I added, and they are right. When every node inside the tolerance band is loading, the fallback returned closestHealthyNode - but that node is the band minimum, so it is in the band and was just observed loading. The caller then gets -LOADING even though a ready node exists outside the band. With MaxRedirects=0 the request fails outright; with the default budget it can still exhaust retries when the band holds more loading nodes than there are retries. Now the exceptional path widens the search: it scans the healthy nodes outside the band and returns the closest ready one. In a multi-AZ layout that is typically the master, still serving while the local replicas reload together after a resync - which is exactly the shape this option is aimed at, and exactly what happens after the full resync a small replication backlog causes. The extra Loading() calls stay off the hot path. This branch is only reached when the whole band is loading, and the latency comparison is evaluated before Loading() so most candidates are skipped anyway. If nothing is ready anywhere it still returns closestHealthyNode, so the caller always gets a node to retry against - the behaviour this path had before the loading filter existed. * refactor(cluster): rename the rotation field to latencyBandNodeCursor Picking one and stopping, per @ndyakov. nodesByLatencyRotation was misleading in the way he first flagged: it reads as nodes ordered by latency, with the cursor walking them fastest-first. That is not what happens. Latency decides only which nodes are in the tolerance band; inside it they are treated as equally close and picked in turn. Swapping Rotation for Cursor fixes the 'what is this' half and leaves the misleading half in place, which is why nodesByLatencyCursor and nodeByLatencyCursor were not quite it either. latencyBandNodeCursor reads as: a cursor, over nodes, in the latency band. 'band' is already the vocabulary in the surrounding comments, the PR description and the review thread, so it is continuity rather than a fourth coinage. The field comment now says explicitly that latency governs membership and not order, so the next reader does not have to infer it. * fix(cluster): keep strict selection until a latency probe lands newClusterNodeWithNodeAddress stores math.MaxUint32 as each node's initial latency and measures asynchronously, so for the ~150-250ms until the first of ten probes completes every node reports the same sentinel value. Every difference is then zero, so any positive tolerance admitted the whole slot and round-robined startup reads across distant zones - the opposite of what RouteByLatency is for, and worst for short-lived clients that spend a large share of their traffic in that window. Track whether any healthy node carries a measurement and fall back to the strict pick until one does. A mix needs no special case: an unmeasured node's sentinel latency is ~4295s, far outside the band of any measured node. updateLatency stores the value and the timestamp together, so a node with a latency but no timestamp does not occur in production; the test helper now sets both, and the two new tests cover the all-unmeasured and mixed cases. * fix(cluster): derive unmeasured state from the sampled latency The previous guard read LastLatencyMeasurement() separately from the latency it was guarding. updateLatency publishes the two as independent stores - latency first, timestamp second - so a probe landing between the two reads made the set look measured while every sampled latency was still the initial sentinel. That is precisely the case the guard exists to catch, so it had a scheduling-dependent hole. Compare the already-sampled latency against the sentinel instead, which is consistent by construction and needs no second read. Equality against a named constant rather than a >= threshold, because maximumNodeLatency (1 minute, every ping failed) is a real measurement and must keep taking the band path. newClusterNodeWithNodeAddress now stores that same constant so the two cannot drift. The test helper's timestamp is reverted, being unread again. --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>
2026-10-24 07:59:20 +03:00 · 2026-08-27 16:33:45 +02:00
parent 18837034a1
commit 6aedc4ce78
5 changed files with 700 additions and 12 deletions
--- a/osscluster.go
+++ b/osscluster.go
@@ -62,6 +62,17 @@ type ClusterOptions struct {
 	// Allows routing read-only commands to the closest master or slave node.
 	// It automatically enables ReadOnly.
 	RouteByLatency bool
+	// RouteByLatencyTolerance widens RouteByLatency from "the single fastest node" to
+	// "any node within this much of the fastest", round-robining between them.
+	//
+	// RouteByLatency alone takes a strict minimum, so when several nodes are equally
+	// close - replicas sharing an availability zone, say - every client picks the same
+	// one and the others take no read traffic. Latency is estimated from ten pings and
+	// refreshed at most every 10s, so a difference well inside the noise can decide the
+	// whole read load for the next interval.
+	//
+	// Zero keeps the strict-minimum behaviour. Has no effect unless RouteByLatency is set.
+	RouteByLatencyTolerance time.Duration
 	// Allows routing read-only commands to the random master or slave node.
 	// It automatically enables ReadOnly.
 	RouteRandomly bool
@@ -379,6 +390,7 @@ func setupClusterQueryParams(u *url.URL, o *ClusterOptions) (*ClusterOptions, er
 	o.MaxRedirects = q.int("max_redirects")
 	o.ReadOnly = q.bool("read_only")
 	o.RouteByLatency = q.bool("route_by_latency")
+	o.RouteByLatencyTolerance = q.duration("route_by_latency_tolerance")
 	o.RouteRandomly = q.bool("route_randomly")
 	o.MaxRetries = q.int("max_retries")
 	o.MinRetryBackoff = q.duration("min_retry_backoff")
@@ -518,7 +530,7 @@ func newClusterNodeWithNodeAddress(clOpt *ClusterOptions, addr, nodeAddress stri
 		Client: clOpt.NewClient(opt),
 	}

-	node.latency.Store(math.MaxUint32)
+	node.latency.Store(unmeasuredNodeLatencyMicros)
 	if clOpt.RouteByLatency {
 		go node.updateLatency()
 	}
@@ -536,6 +548,14 @@ func (n *clusterNode) Close() error {

 const maximumNodeLatency = 1 * time.Minute

+// Latency held by a node from creation until its first probe completes. Deliberately distinct
+// from maximumNodeLatency, which marks a node whose pings all failed - the two must stay
+// separable, so this is compared for equality rather than as a ">= huge" threshold.
+const (
+	unmeasuredNodeLatencyMicros uint32 = math.MaxUint32
+	unmeasuredNodeLatency              = time.Duration(unmeasuredNodeLatencyMicros) * time.Microsecond
+)
+
 func (n *clusterNode) updateLatency() {
 	const numProbe = 10
 	var dur uint64
@@ -817,6 +837,13 @@ type clusterSlot struct {
 	start int
 	end   int
 	nodes []*clusterNode
+
+	// Round-robin cursor over the nodes inside the RouteByLatencyTolerance band. Latency
+	// decides only which nodes are in the band; within it they are treated as equally close
+	// and picked in turn, so this does not order them. Per slot on purpose: a counter shared
+	// across slots is advanced by the other slots between two visits to this one, so with a
+	// regular interleaving each slot keeps landing on the same candidate index.
+	latencyBandNodeCursor atomic.Uint32
 }

 type clusterState struct {
@@ -1027,6 +1054,147 @@ func (c *clusterState) slotClosestNode(slot int) (*clusterNode, error) {
 	return c.nodes.Random()
 }

+// slotNodeWithinLatency picks from the healthy nodes whose latency is within tolerance of the
+// fastest one, rotating across them so equally-close nodes share the read traffic. Used only
+// when RouteByLatencyTolerance is set; slotClosestNode keeps the strict-minimum behaviour and
+// is left untouched for everyone else.
+func (c *clusterState) slotNodeWithinLatency(slot int, tolerance time.Duration) (*clusterNode, error) {
+	entry := c.slotEntry(slot)
+	if entry == nil || len(entry.nodes) == 0 {
+		return c.nodes.Random()
+	}
+	nodes := entry.nodes
+
+	// Latency and health are sampled once per node. The background probe updates them
+	// concurrently, so re-reading would let the candidate set disagree with the minimum it
+	// is compared against - and could leave that set empty.
+	type sample struct {
+		node    *clusterNode
+		latency time.Duration
+	}
+
+	var (
+		healthy            = make([]sample, 0, len(nodes))
+		closestNode        *clusterNode
+		closestHealthyNode *clusterNode
+		minLatency         = time.Duration(math.MaxInt64)
+		minHealthyLatency  = time.Duration(math.MaxInt64)
+		anyHealthyMeasured bool
+	)
+
+	for _, n := range nodes {
+		latency := n.Latency()
+		if latency < minLatency {
+			closestNode, minLatency = n, latency
+		}
+
+		if n.Failing() {
+			continue
+		}
+
+		healthy = append(healthy, sample{node: n, latency: latency})
+
+		// Derived from the latency just captured, not from a second read of the node.
+		// updateLatency publishes the latency and its timestamp as two separate stores, so
+		// reading the timestamp here could observe a probe that landed after the latency
+		// above was sampled - marking the set measured while every sampled latency is still
+		// a sentinel, which is exactly the case this guards.
+		if latency != unmeasuredNodeLatency {
+			anyHealthyMeasured = true
+		}
+
+		// Tracked separately: a healthy node that is not the outright fastest must still be
+		// preferred over a failing one.
+		if latency < minHealthyLatency {
+			closestHealthyNode, minHealthyLatency = n, latency
+		}
+	}
+
+	if closestHealthyNode != nil {
+		// Until the first probe lands, every node still holds the sentinel latency stored by
+		// newClusterNodeWithNodeAddress, so every difference is zero and any positive tolerance
+		// would admit the whole slot - scattering startup reads across distant zones instead of
+		// preserving locality. Keep strict selection until at least one measurement exists; a
+		// mix needs no special case, since an unmeasured node's sentinel latency puts it far
+		// outside the band of any measured one.
+		if !anyHealthyMeasured {
+			return closestHealthyNode, nil
+		}
+
+		candidates := make([]*clusterNode, 0, len(healthy))
+		for _, s := range healthy {
+			// Subtraction rather than minHealthyLatency+tolerance, which would overflow for
+			// a very large tolerance. Always true for closestHealthyNode, so candidates is
+			// never empty here.
+			if s.latency-minHealthyLatency <= tolerance {
+				candidates = append(candidates, s.node)
+			}
+		}
+
+		// Drop nodes that are still loading, matching slotSlaveNode. Checked here rather than
+		// in the pass above so Loading() - which can cost a Ping when the node is not known
+		// loaded - is only paid for nodes actually eligible for this read. Widening the
+		// candidate set is what makes this matter: under a strict minimum a loading replica
+		// has to be the fastest to be picked, within a tolerance band it merely has to be
+		// close, and a replica is loading precisely after the full resync that a too-small
+		// replication backlog causes.
+		ready := candidates[:0]
+		for _, n := range candidates {
+			if !n.Loading() {
+				ready = append(ready, n)
+			}
+		}
+
+		if len(ready) == 0 {
+			// Every node inside the band is loading. closestHealthyNode is itself in the band,
+			// so returning it hands back a node we just observed loading. Widen the search to
+			// the healthy nodes outside the band and take the closest ready one - which in a
+			// multi-AZ layout is typically the master, still serving while the local replicas
+			// reload together after a resync. Only reached in this exceptional case, so the
+			// extra Loading() calls are off the hot path, and the latency comparison is
+			// evaluated first so most of them are skipped.
+			var (
+				fallback        *clusterNode
+				fallbackLatency = time.Duration(math.MaxInt64)
+			)
+			for _, s := range healthy {
+				if s.latency-minHealthyLatency <= tolerance {
+					continue // in the band, already known to be loading
+				}
+				if s.latency < fallbackLatency && !s.node.Loading() {
+					fallback, fallbackLatency = s.node, s.latency
+				}
+			}
+			if fallback != nil {
+				return fallback, nil
+			}
+
+			// Nothing is ready anywhere. Return the closest healthy node so the caller still
+			// gets a node to retry against, which is what this path did before the filter.
+			return closestHealthyNode, nil
+		}
+
+		if len(ready) == 1 {
+			return ready[0], nil
+		}
+
+		// Reduced in the unsigned domain: the cursor wraps, and on 32-bit builds converting a
+		// value past 2^31 to int before the modulo would make the index negative.
+		return ready[(entry.latencyBandNodeCursor.Add(1)-1)%uint32(len(ready))], nil
+	}
+
+	// Every node is failing. Fall back to the least-slow one, so a transient failure does
+	// not take the slot down.
+	if minLatency < maximumNodeLatency && closestNode != nil {
+		internal.Logger.Printf(context.TODO(), "redis: all nodes are marked as failed, picking the temporarily failing node with lowest latency")
+		return closestNode, nil
+	}
+
+	// If all nodes are having the maximum latency(all pings are failing) - return a random node across the cluster
+	internal.Logger.Printf(context.TODO(), "redis: pings to all nodes are failing, picking a random node across the cluster")
+	return c.nodes.Random()
+}
+
 func (c *clusterState) slotRandomNode(slot int) (*clusterNode, error) {
 	nodes := c.slotNodes(slot)
 	if len(nodes) == 0 {
@@ -1067,7 +1235,7 @@ func (c *clusterState) slotShardPickerSlaveNode(slot int, shardPicker routing.Sh
 	return nodes[0], nil
 }

-func (c *clusterState) slotNodes(slot int) []*clusterNode {
+func (c *clusterState) slotEntry(slot int) *clusterSlot {
 	i := sort.Search(len(c.slots), func(i int) bool {
 		return c.slots[i].end >= slot
 	})
@@ -1076,6 +1244,13 @@ func (c *clusterState) slotNodes(slot int) []*clusterNode {
 	}
 	x := c.slots[i]
 	if slot >= x.start && slot <= x.end {
+		return x
+	}
+	return nil
+}
+
+func (c *clusterState) slotNodes(slot int) []*clusterNode {
+	if x := c.slotEntry(slot); x != nil {
 		return x.nodes
 	}
 	return nil
@@ -2982,6 +3157,9 @@ func (c *ClusterClient) cmdNodeWithShardPicker(

 func (c *ClusterClient) slotReadOnlyNode(state *clusterState, slot int) (*clusterNode, error) {
 	if c.opt.RouteByLatency {
+		if c.opt.RouteByLatencyTolerance > 0 {
+			return state.slotNodeWithinLatency(slot, c.opt.RouteByLatencyTolerance)
+		}
 		return state.slotClosestNode(slot)
 	}
 	if c.opt.RouteRandomly {
--- a/osscluster_latency_tolerance_test.go
+++ b/osscluster_latency_tolerance_test.go
@@ -0,0 +1,414 @@
+package redis
+
+import (
+	"math"
+	"testing"
+	"time"
+)
+
+func toleranceTestNode(t *testing.T, latency time.Duration, failing bool) *clusterNode {
+	t.Helper()
+
+	n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})}
+	t.Cleanup(func() { _ = n.Client.Close() })
+
+	// Steady state: the node has answered a PING at least once, so Loading() takes its
+	// cached path instead of dialling a dead address for 100ms per candidate.
+	n.loaded.Store(1)
+	n.latency.Store(uint32(latency / time.Microsecond))
+	if failing {
+		n.MarkAsFailing()
+	}
+
+	return n
+}
+
+func toleranceTestState(nodes ...*clusterNode) *clusterState {
+	return &clusterState{slots: []*clusterSlot{{start: 0, end: 16383, nodes: nodes}}}
+}
+
+func TestRouteByLatencyToleranceZeroKeepsStrictMinimum(t *testing.T) {
+	fastest := toleranceTestNode(t, 100*time.Microsecond, false)
+	state := toleranceTestState(
+		fastest,
+		toleranceTestNode(t, 120*time.Microsecond, false),
+		toleranceTestNode(t, 900*time.Microsecond, false),
+	)
+
+	for i := 0; i < 20; i++ {
+		got, err := state.slotClosestNode(0)
+		if err != nil {
+			t.Fatalf("slotClosestNode: %v", err)
+		}
+		if got != fastest {
+			t.Fatalf("iteration %d: expected the fastest node with zero tolerance", i)
+		}
+	}
+}
+
+func TestSlotNodeWithinLatencySpreadsAcrossCloseNodes(t *testing.T) {
+	a := toleranceTestNode(t, 100*time.Microsecond, false)
+	b := toleranceTestNode(t, 120*time.Microsecond, false)
+	far := toleranceTestNode(t, 900*time.Microsecond, false)
+	state := toleranceTestState(a, b, far)
+
+	counts := map[*clusterNode]int{}
+
+	for i := 0; i < 100; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		counts[got]++
+	}
+
+	if counts[far] != 0 {
+		t.Fatalf("node outside the tolerance received %d requests", counts[far])
+	}
+	if counts[a] == 0 || counts[b] == 0 {
+		t.Fatalf("expected both close nodes to be used, got a=%d b=%d", counts[a], counts[b])
+	}
+	if diff := counts[a] - counts[b]; diff > 1 || diff < -1 {
+		t.Fatalf("expected an even split across close nodes, got a=%d b=%d", counts[a], counts[b])
+	}
+}
+
+func TestSlotNodeWithinLatencySkipsFailingNodes(t *testing.T) {
+	healthy := toleranceTestNode(t, 120*time.Microsecond, false)
+	state := toleranceTestState(toleranceTestNode(t, 100*time.Microsecond, true), healthy)
+
+	for i := 0; i < 10; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got != healthy {
+			t.Fatal("a failing node was selected")
+		}
+	}
+}
+
+// The fastest node being unhealthy must not stop a slower healthy one from being chosen.
+// Previously the healthy candidate was only recorded while scanning for a new minimum, so
+// a healthy-but-slower node could be skipped and selection fell through to the
+// all-nodes-failing path.
+// Exercises slotNodeWithinLatency rather than slotClosestNode: preferring a slower healthy
+// node over a faster failing one is behaviour of the new path. slotClosestNode is left as it
+// was on master, latent bug included - see the PR discussion.
+func TestSlotNodeWithinLatencyPrefersSlowerHealthyNode(t *testing.T) {
+	healthy := toleranceTestNode(t, 800*time.Microsecond, false)
+	state := toleranceTestState(toleranceTestNode(t, 100*time.Microsecond, true), healthy)
+
+	got, err := state.slotNodeWithinLatency(0, 0)
+	if err != nil {
+		t.Fatalf("slotNodeWithinLatency: %v", err)
+	}
+	if got != healthy {
+		t.Fatal("expected the slower healthy node rather than the failing fastest one")
+	}
+}
+
+func TestSlotNodeWithinLatencyNeverReturnsNilWithHealthyNode(t *testing.T) {
+	nodes := []*clusterNode{
+		toleranceTestNode(t, 100*time.Microsecond, false),
+		toleranceTestNode(t, 120*time.Microsecond, false),
+		toleranceTestNode(t, 900*time.Microsecond, false),
+	}
+	state := toleranceTestState(nodes...)
+
+	stop := make(chan struct{})
+	defer close(stop)
+
+	// Churn the latencies while selection runs, including pushing the fastest node above
+	// the band it established.
+	go func() {
+		for i := 0; ; i++ {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			nodes[i%len(nodes)].latency.Store(uint32(100 + (i*37)%5000))
+		}
+	}()
+
+	for i := 0; i < 5000; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got == nil {
+			t.Fatal("returned nil despite healthy nodes being available")
+		}
+	}
+}
+
+func TestSlotNodeWithinLatencyHandlesHugeTolerance(t *testing.T) {
+	a := toleranceTestNode(t, 100*time.Microsecond, false)
+	state := toleranceTestState(a, toleranceTestNode(t, 900*time.Microsecond, false))
+
+	// A tolerance near the maximum must not overflow into a negative comparison.
+	got, err := state.slotNodeWithinLatency(0, time.Duration(math.MaxInt64))
+	if err != nil {
+		t.Fatalf("slotNodeWithinLatency: %v", err)
+	}
+	if got == nil {
+		t.Fatal("returned nil for a very large tolerance")
+	}
+}
+
+func TestRouteByLatencyToleranceFromClusterURL(t *testing.T) {
+	o, err := ParseClusterURL("redis://localhost:6379?route_by_latency=true&route_by_latency_tolerance=250us")
+	if err != nil {
+		t.Fatalf("ParseClusterURL: %v", err)
+	}
+	if !o.RouteByLatency {
+		t.Fatal("RouteByLatency not parsed")
+	}
+	if o.RouteByLatencyTolerance != 250*time.Microsecond {
+		t.Fatalf("tolerance = %v, want 250us", o.RouteByLatencyTolerance)
+	}
+}
+
+func TestRouteByLatencyToleranceFromFailoverURL(t *testing.T) {
+	o, err := ParseFailoverURL("redis://localhost:26379?master_name=mymaster&route_by_latency=true&route_by_latency_tolerance=1ms")
+	if err != nil {
+		t.Fatalf("ParseFailoverURL: %v", err)
+	}
+	if o.RouteByLatencyTolerance != time.Millisecond {
+		t.Fatalf("tolerance = %v, want 1ms", o.RouteByLatencyTolerance)
+	}
+}
+
+func TestRouteByLatencyToleranceFromUniversalOptions(t *testing.T) {
+	o := &UniversalOptions{
+		Addrs:                   []string{"localhost:6379"},
+		RouteByLatency:          true,
+		RouteByLatencyTolerance: 300 * time.Microsecond,
+	}
+
+	if got := o.Cluster().RouteByLatencyTolerance; got != 300*time.Microsecond {
+		t.Fatalf("Cluster() tolerance = %v, want 300us", got)
+	}
+
+	o.MasterName = "mymaster"
+	if got := o.Failover().RouteByLatencyTolerance; got != 300*time.Microsecond {
+		t.Fatalf("Failover() tolerance = %v, want 300us", got)
+	}
+}
+
+// With the option unset, selection must stay on the pre-existing hot path: no candidate
+// slice, no allocation per command.
+func TestSlotClosestNodeStillDoesNotAllocate(t *testing.T) {
+	state := toleranceTestState(
+		toleranceTestNode(t, 100*time.Microsecond, false),
+		toleranceTestNode(t, 120*time.Microsecond, false),
+		toleranceTestNode(t, 900*time.Microsecond, false),
+	)
+
+	allocs := testing.AllocsPerRun(200, func() {
+		if _, err := state.slotClosestNode(0); err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+	})
+
+	if allocs != 0 {
+		t.Fatalf("zero tolerance allocated %.0f times per call, want 0", allocs)
+	}
+}
+
+// Rotation state is per slot. A counter shared across slots is advanced by the other slots
+// between two visits to this one, so with a regular interleaving each slot keeps landing on
+// the same candidate index and one replica per shard takes all of that shard's reads.
+func TestSlotNodeWithinLatencyRotatesPerSlot(t *testing.T) {
+	nodesFor := func() []*clusterNode {
+		return []*clusterNode{
+			toleranceTestNode(t, 100*time.Microsecond, false),
+			toleranceTestNode(t, 120*time.Microsecond, false),
+		}
+	}
+	a, b := nodesFor(), nodesFor()
+	state := &clusterState{slots: []*clusterSlot{
+		{start: 0, end: 100, nodes: a},
+		{start: 101, end: 200, nodes: b},
+	}}
+
+	counts := map[*clusterNode]int{}
+	for i := 0; i < 100; i++ {
+		for _, slot := range []int{50, 150} { // alternate slots, the interleaving that breaks a shared counter
+			got, err := state.slotNodeWithinLatency(slot, 50*time.Microsecond)
+			if err != nil {
+				t.Fatalf("slotNodeWithinLatency: %v", err)
+			}
+			counts[got]++
+		}
+	}
+
+	for name, nodes := range map[string][]*clusterNode{"slot 0-100": a, "slot 101-200": b} {
+		if counts[nodes[0]] == 0 || counts[nodes[1]] == 0 {
+			t.Fatalf("%s did not rotate: %d / %d", name, counts[nodes[0]], counts[nodes[1]])
+		}
+	}
+}
+
+// The rotation cursor is a uint32 and wraps. Reducing it to an index must stay in the unsigned
+// domain: on 32-bit builds int is 32 bits, so converting a cursor past 2^31 before the modulo
+// yields a negative index and panics. Run with GOARCH=386 to exercise that.
+func TestSlotNodeWithinLatencyRotationPastIntMax(t *testing.T) {
+	first := toleranceTestNode(t, 100*time.Microsecond, false)
+	second := toleranceTestNode(t, 110*time.Microsecond, false)
+	state := toleranceTestState(first, second)
+	state.slots[0].latencyBandNodeCursor.Store(math.MaxInt32)
+
+	seen := map[*clusterNode]int{}
+	for i := 0; i < 8; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got != first && got != second {
+			t.Fatalf("unexpected node %p", got)
+		}
+		seen[got]++
+	}
+
+	if len(seen) != 2 {
+		t.Fatalf("rotation stalled across the uint32 boundary: %v", seen)
+	}
+}
+
+// A node that has not yet been confirmed loaded must still be selectable: Loading() only
+// reports true for an actual LOADING reply, and a connection error is not one. Guards
+// against the tolerance filter over-excluding and shrinking the rotation to nothing.
+func TestSlotNodeWithinLatencyKeepsUnconfirmedNode(t *testing.T) {
+	fast := toleranceTestNode(t, 100*time.Microsecond, false)
+	unconfirmed := toleranceTestNode(t, 110*time.Microsecond, false)
+	unconfirmed.loaded.Store(0)
+
+	state := toleranceTestState(fast, unconfirmed)
+
+	seen := map[*clusterNode]bool{}
+	for i := 0; i < 8; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got == nil {
+			t.Fatal("slotNodeWithinLatency returned nil")
+		}
+		seen[got] = true
+	}
+
+	if !seen[unconfirmed] {
+		t.Fatal("node with loaded=0 was excluded from the rotation; a connection error is not a LOADING reply")
+	}
+}
+
+// Before the background probes finish, every node still carries the math.MaxUint32 sentinel
+// from newClusterNodeWithNodeAddress. All differences are then zero, so a positive tolerance
+// would admit the entire slot and round-robin startup reads across distant zones. Until one
+// measurement lands, the tolerance path must behave exactly like the strict one.
+func TestSlotNodeWithinLatencyKeepsStrictOrderBeforeFirstProbe(t *testing.T) {
+	unprobed := func() *clusterNode {
+		n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})}
+		t.Cleanup(func() { _ = n.Client.Close() })
+		n.loaded.Store(1)
+		n.latency.Store(unmeasuredNodeLatencyMicros) // as newClusterNodeWithNodeAddress does
+		return n
+	}
+
+	first, second, third := unprobed(), unprobed(), unprobed()
+	state := toleranceTestState(first, second, third)
+
+	for i := 0; i < 50; i++ {
+		got, err := state.slotNodeWithinLatency(0, time.Second)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got != first {
+			t.Fatalf("iteration %d: rotated across unmeasured nodes; want the strict pick", i)
+		}
+	}
+
+	strict, err := state.slotClosestNode(0)
+	if err != nil {
+		t.Fatalf("slotClosestNode: %v", err)
+	}
+	if strict != first {
+		t.Fatalf("slotClosestNode disagrees with the tolerance path before the first probe")
+	}
+}
+
+// Once any healthy node has a real measurement the band is meaningful again, and an
+// unmeasured node must stay out of it - its sentinel latency is ~4295s, far beyond any
+// sane tolerance.
+func TestSlotNodeWithinLatencyExcludesUnmeasuredOnceProbed(t *testing.T) {
+	measured := toleranceTestNode(t, 100*time.Microsecond, false)
+	alsoMeasured := toleranceTestNode(t, 130*time.Microsecond, false)
+
+	unprobed := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})}
+	t.Cleanup(func() { _ = unprobed.Client.Close() })
+	unprobed.loaded.Store(1)
+	unprobed.latency.Store(unmeasuredNodeLatencyMicros)
+
+	state := toleranceTestState(measured, alsoMeasured, unprobed)
+
+	for i := 0; i < 60; i++ {
+		got, err := state.slotNodeWithinLatency(0, 50*time.Microsecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got == unprobed {
+			t.Fatalf("iteration %d: unmeasured node entered the tolerance band", i)
+		}
+	}
+}
+
+// updateLatency publishes the latency and its timestamp as two separate stores, so a probe can
+// land between selection sampling a node's latency and reading its measurement timestamp. That
+// interleaving is observable as "sentinel latency, timestamp set" - and must still count as
+// unmeasured, or the band widens to the whole slot while every sampled latency is a sentinel.
+func TestSlotNodeWithinLatencyIgnoresTimestampWithoutLatency(t *testing.T) {
+	interleaved := func() *clusterNode {
+		n := &clusterNode{Client: NewClient(&Options{Addr: "127.0.0.1:0"})}
+		t.Cleanup(func() { _ = n.Client.Close() })
+		n.loaded.Store(1)
+		n.latency.Store(unmeasuredNodeLatencyMicros)
+		// The half-published state: timestamp visible, latency not yet stored.
+		n.SetLastLatencyMeasurement(time.Now())
+		return n
+	}
+
+	first, second, third := interleaved(), interleaved(), interleaved()
+	state := toleranceTestState(first, second, third)
+
+	for i := 0; i < 50; i++ {
+		got, err := state.slotNodeWithinLatency(0, time.Second)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		if got != first {
+			t.Fatalf("iteration %d: a set timestamp widened the band while latencies were sentinels", i)
+		}
+	}
+}
+
+// maximumNodeLatency (all pings failed) must not be mistaken for the unmeasured sentinel: such a
+// node has a real measurement, so the band applies normally rather than collapsing to strict.
+func TestSlotNodeWithinLatencyTreatsFailedPingsAsMeasured(t *testing.T) {
+	a := toleranceTestNode(t, maximumNodeLatency, false)
+	b := toleranceTestNode(t, maximumNodeLatency, false)
+	state := toleranceTestState(a, b)
+
+	counts := map[*clusterNode]int{}
+	for i := 0; i < 80; i++ {
+		got, err := state.slotNodeWithinLatency(0, time.Millisecond)
+		if err != nil {
+			t.Fatalf("slotNodeWithinLatency: %v", err)
+		}
+		counts[got]++
+	}
+	if counts[a] == 0 || counts[b] == 0 {
+		t.Fatalf("expected rotation across both measured nodes, got a=%d b=%d", counts[a], counts[b])
+	}
+}
--- a/osscluster_test.go
+++ b/osscluster_test.go
@@ -1784,6 +1784,93 @@ var _ = Describe("ClusterClient", func() {
 		assertClusterClient()
 	})

+	Describe("ClusterClient with RouteByLatencyTolerance", func() {
+		BeforeEach(func() {
+			opt = redisClusterOptions()
+			opt.RouteByLatency = true
+			// Far wider than any intra-cluster spread, so every healthy node for a slot is a
+			// candidate and the per-slot rotation is the thing under test.
+			opt.RouteByLatencyTolerance = time.Second
+			client = cluster.newClusterClient(ctx, opt)
+
+			err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
+				return master.FlushDB(ctx).Err()
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			err = client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
+				Eventually(func() int64 {
+					return client.DBSize(ctx).Val()
+				}, 30*time.Second).Should(Equal(int64(0)))
+				return nil
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			err := client.ForEachSlave(ctx, func(ctx context.Context, slave *redis.Client) error {
+				return slave.ReadWrite(ctx).Err()
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			err = client.Close()
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("should read keys from distinct shards", func() {
+			// Discovered rather than hardcoded: keyN does not spread over the three masters
+			// (key2 and key3 both hash into 0-5460), and hardcoded keys would silently stop
+			// covering multiple shards if the slot ranges ever moved.
+			keys := make([]string, 0, 3)
+			slots := map[int64]string{}
+			masters := map[string]string{}
+
+			for i := 0; len(masters) < 3 && i < 100; i++ {
+				key := fmt.Sprintf("tolerance-key-%d", i)
+
+				nodes, err := client.Nodes(ctx, key)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(nodes).NotTo(BeEmpty())
+
+				master := nodes[0].Client.Options().Addr
+				if _, seen := masters[master]; seen {
+					continue
+				}
+				masters[master] = key
+
+				slot := client.ClusterKeySlot(ctx, key).Val()
+				Expect(slots).NotTo(HaveKey(slot))
+				slots[slot] = key
+				keys = append(keys, key)
+			}
+			Expect(masters).To(HaveLen(3), "could not find keys on three distinct masters: %v", masters)
+
+			for _, key := range keys {
+				Expect(client.Set(ctx, key, "VALUE-"+key, 0).Err()).NotTo(HaveOccurred())
+			}
+
+			// Reads go to whichever in-band node the rotation picks, including replicas that
+			// may not have caught up yet, so poll rather than failing on the first attempt.
+			for _, key := range keys {
+				Eventually(func() string {
+					return client.Get(ctx, key).Val()
+				}, 30*time.Second).Should(Equal("VALUE-" + key))
+			}
+
+			// Repeated reads must keep resolving to a usable node for every slot. This is what
+			// regressed when the rotation cursor was shared across slots.
+			for i := 0; i < 50; i++ {
+				for _, key := range keys {
+					Eventually(func() string {
+						return client.Get(ctx, key).Val()
+					}, 30*time.Second).Should(Equal("VALUE-" + key))
+				}
+			}
+		})
+
+		assertClusterClient()
+	})
+
 	Describe("ClusterClient with ClusterSlots", func() {
 		BeforeEach(func() {
 			failover = true
--- a/sentinel.go
+++ b/sentinel.go
@@ -47,6 +47,8 @@ type FailoverOptions struct {
 	// Allows routing read-only commands to the closest master or replica node.
 	// This option only works with NewFailoverClusterClient.
 	RouteByLatency bool
+	// RouteByLatencyTolerance is passed through to ClusterOptions; see its documentation.
+	RouteByLatencyTolerance time.Duration
 	// Allows routing read-only commands to the random master or replica node.
 	// This option only works with NewFailoverClusterClient.
 	RouteRandomly bool
@@ -328,9 +330,10 @@ func (opt *FailoverOptions) clusterOptions() *ClusterOptions {

 		MaxRedirects: opt.MaxRetries,

-		ReadOnly:       opt.ReplicaOnly,
-		RouteByLatency: opt.RouteByLatency,
-		RouteRandomly:  opt.RouteRandomly,
+		ReadOnly:                opt.ReplicaOnly,
+		RouteByLatency:          opt.RouteByLatency,
+		RouteByLatencyTolerance: opt.RouteByLatencyTolerance,
+		RouteRandomly:           opt.RouteRandomly,

 		MinRetryBackoff: opt.MinRetryBackoff,
 		MaxRetryBackoff: opt.MaxRetryBackoff,
@@ -462,6 +465,7 @@ func setupFailoverConnParams(u *url.URL, o *FailoverOptions) (*FailoverOptions,
 	o.MasterName = q.string("master_name")
 	o.ClientName = q.string("client_name")
 	o.RouteByLatency = q.bool("route_by_latency")
+	o.RouteByLatencyTolerance = q.duration("route_by_latency_tolerance")
 	o.RouteRandomly = q.bool("route_randomly")
 	o.ReplicaOnly = q.bool("replica_only")
 	o.UseDisconnectedReplicas = q.bool("use_disconnected_replicas")
--- a/universal.go
+++ b/universal.go
@@ -120,7 +120,10 @@ type UniversalOptions struct {
 	MaxRedirects   int
 	ReadOnly       bool
 	RouteByLatency bool
-	RouteRandomly  bool
+	// RouteByLatencyTolerance is passed through to ClusterOptions and FailoverOptions;
+	// see ClusterOptions.RouteByLatencyTolerance.
+	RouteByLatencyTolerance time.Duration
+	RouteRandomly           bool

 	// MasterName is the sentinel master name.
 	// Only for failover clients.
@@ -203,10 +206,11 @@ func (o *UniversalOptions) Cluster() *ClusterOptions {
 		CredentialsProviderContext:   o.CredentialsProviderContext,
 		StreamingCredentialsProvider: o.StreamingCredentialsProvider,

-		MaxRedirects:   o.MaxRedirects,
-		ReadOnly:       o.ReadOnly,
-		RouteByLatency: o.RouteByLatency,
-		RouteRandomly:  o.RouteRandomly,
+		MaxRedirects:            o.MaxRedirects,
+		ReadOnly:                o.ReadOnly,
+		RouteByLatency:          o.RouteByLatency,
+		RouteByLatencyTolerance: o.RouteByLatencyTolerance,
+		RouteRandomly:           o.RouteRandomly,

 		MaxRetries:      o.MaxRetries,
 		MinRetryBackoff: o.MinRetryBackoff,
@@ -276,8 +280,9 @@ func (o *UniversalOptions) Failover() *FailoverOptions {
 		SentinelUsername: o.SentinelUsername,
 		SentinelPassword: o.SentinelPassword,

-		RouteByLatency: o.RouteByLatency,
-		RouteRandomly:  o.RouteRandomly,
+		RouteByLatency:          o.RouteByLatency,
+		RouteByLatencyTolerance: o.RouteByLatencyTolerance,
+		RouteRandomly:           o.RouteRandomly,

 		MaxRetries:      o.MaxRetries,
 		MinRetryBackoff: o.MinRetryBackoff,