Add opt-in latency-aware cluster read balancing

[lh0156]

Fixes #4185

Summary

• Add opt-in LoadBalancingStrategy.LATENCY_BASED for cluster reads.
• Select two eligible nodes with power-of-two choices and prefer the lower peak-sensitive EWMA score:
  peak_ewma * (in_flight + 1).
• Keep existing round-robin and random strategies unchanged.
• Record per-node latency and in-flight state for sync and asyncio commands.
• Record MOVED/ASK attempts against the node that handled each attempt.
• Record one node-level sample for grouped pipeline execution when the batch contains reads.
• Document the strategy and add a small cluster benchmark.

Implementation notes

The latency state is implemented once in the shared LoadBalancer, so sync and asyncio clients use the same selection and update logic. The selection path is protected by the existing bounded lock; no await is introduced into asyncio selection.

Latency instrumentation is enabled only for the new strategy. Existing strategies do not pay for request timing or state updates.

Verification

• TDD RED: node-manager tests initially failed because the candidate node list was not supplied to LoadBalancer.
• TDD GREEN: focused sync tests: 8 passed.
• Focused asyncio tests: 3 passed.
• invoke linters: passed.
• python -m compileall: passed.
• git diff --check: passed.

The full cluster test files require local Redis cluster services. They were attempted, but this environment has no Redis cluster listening on the test ports, so connection-dependent tests could not complete.

The benchmark is available at benchmarks/cluster_latency_load_balancing.py; it requires a Redis cluster with at least one replica for the selected key slot and reports delayed-replica selection share and p99 latency against round-robin.

[petyaslavova]

This needs changes in three areas before it can be reviewed as a candidate for merge.

Sampling. record_latency runs in a finally block, so failure durations become "this node is fast"
evidence. MOVED/ASK, TRYAGAIN, CLUSTERDOWN and auth errors all return in well under a millisecond, so
a node with a stale slot view scores near zero and keeps winning selection — the redirect loop
reinforces itself. On the async side the timer also covers connection acquisition, so
MaxConnectionsError (raised synchronously by acquire_connection) and a refused TCP connect both record
near-zero latency: the most saturated node, and a dead one, become the preferred targets. In the other
direction, a TimeoutError records the full socket_timeout — client configuration rather than node
service time — and task cancellation records the caller's wait. Keep the in-flight release in finally,
but record a sample only on a successful response, and skip redirects, cluster-state and auth errors,
pool exhaustion, connect failures and cancellation. If timeouts should penalize a node, make that an
explicit capped penalty.

Sync/async parity. The two stacks measure different intervals: sync starts timing after
get_connection() — so pool wait, connect and the health-check reconnect are excluded, which is exactly
where a degraded node burns time — while async starts before it. Settle on one definition of a node
latency sample and apply it identically in both stacks.

Algorithm. The peak score only decays when the node completes another request, but a penalized node is
never selected again once a healthier peer exists. A single transient outlier therefore excludes a node
from reads permanently, until the next NodesManager.initialize(), which also clears all state. That is
worse than round-robin in the recovery case. Decay by elapsed time since the last observation, give
unmeasured nodes a genuinely neutral score rather than 0.0 (currently the minimum, so cold nodes always
win), and decide explicitly how the state should survive initialize(). Related: the primary is an
eligible candidate but only READ_COMMANDS are instrumented, so a write-saturated primary looks idle and
attracts reads.

Tests. None of the current cases reach the power-of-two-choices path — all four use two nodes and take
the len(nodes) <= 2 shortcut, so random.sample is never called and the patch is inert. Add a case with
three or more nodes, one asserting in_flight returns to zero after success, failure and a MOVED retry,
and one covering decay with no traffic (the current decay test drives record_latency directly, which is
precisely what cannot happen for a starved node). Note also that patch("redis.cluster.random.sample")
patches the stdlib random module globally rather than a local alias. Separately, the sync pipeline
finally block rebinds start_time, shadowing the value later passed to _raise_first_error.

Benchmark. #4185 asked for measured numbers from a redis-py cluster; the PR states the script was never
run. invoke devenv provides a cluster. The script also injects a client-side asyncio.sleep into a
globally monkeypatched ClusterNode.execute_command and then times its own injected delay, so it does not
measure server-side degradation — use DEBUG SLEEP on the replica or netem instead.

Rebase on master as well; the hunks in _execute_command and _send_cluster_commands conflict.

[petyaslavova]

This block will record inaccurate latencies. Executed in finally clause it will record the fast failing error situation as best execution times.

Rewarded (fast failure → lowest score → node becomes preferred):

• MovedError / AskError — sub-millisecond ResponseErrors. A node with a stale slot view scores ≈0 and keeps winning selection, so the redirect loop is self-reinforcing until the slot cache rebuilds. The PR body presents per-attempt MOVED attribution as a feature; the effect is the opposite.
• TryAgainError, ClusterDownError, SlotNotCoveredError, AuthenticationError, and ConnectionError raised from send_command/parse_response on an already-broken socket — all immediate, all rewarded.

[petyaslavova]

Newly created or existing touched methods should use PEP604 standard for type annotations --> Optional[Sequence["ClusterNode"]] should become Sequence["ClusterNode"] | None

[petyaslavova]

Situations that will produce fake good and prefered latency records:

• MaxConnectionsError. Raised synchronously by acquire_connection, which is inside the tracked window because async starts the timer before target_node.execute_command(). So the most saturated node scores ≈0, attracts more reads, and produces more MaxConnectionsError. That's a strict inversion of the feature's purpose.

• connect failure. A refused TCP connect fails in microseconds, so a dead node becomes top-ranked until the slot map refreshes.