backend/perf: pipeline cluster MGET and add runPipelined helpers

[piyushKumar-1]

What

• mGetClusterRaw: still groups keys by hash slot, since Cluster needs single-slot MGETs. The difference is that all per-slot MGETs now go out in one hedis pipeline. hedis queues the commands, groups them by node and sends each node's batch in one write, so a call costs about one round trip per cluster node. Previously it cost one round trip per slot, or waves of 32 forked MGETs. mGetCluster and mGetClusterWithKeys benefit automatically.
• New zAddIfPossibleMany: the same check-and-add script as zAddIfPossible, run for many keys in one pipeline. Results come back aligned with the input: 1 means added, 0 means the key was at capacity, −1 means a reply error or failed pipeline. The script and its arguments are shared with zAddIfPossible, whose behaviour is unchanged.
• New runPipelined / runPipelinedByKey: a generic way to run many single-key commands (ZCOUNT, EVAL, ZREM, …) in one pipeline. Results come back aligned with the input. A per-key reply error, or a failure of the whole pipeline, gives Nothing. Replies are forced inside the call, so the network I/O is covered by the withTryCatch and the withTimeRedis latency timing. Batches run through runHedis, so a connection-level failure surfaces as an error instead of an empty result.
• euler-hs bump, for parallel nodes: flake.lock moves euler-hs to main (a97ac313, chore: bump hedis for optional parallel per-node cluster pipelines euler-hs#76), which relocks hedis to 773cb39c with perf: optional parallel per-node execution for cluster pipelines hedis#6 merged. With REDIS_CLUSTER_PARALLEL_PIPELINE=true, hedis sends a pipeline's per-node batches concurrently, so a pipeline costs about as much as the slowest node. The flag is off by default and read once per process. The bump also brings the node-pool reset commits that were already on that hedis branch. flake.nix is unchanged from main.
• Removed: the CLUSTER_MGET_ASYNC_ENABLED env flag (getClusterMGetAsyncEnabled), and the Forkable/L.MonadFlow constraints on the three MGET functions. Fewer constraints means existing callers still compile.

Why

Driver pooling in nammayatri reads about 30 keys per driver, each in its own slot. With the sequential path that was about 30 round trips per driver, around 16–19 ms per driver in production; with async fork waves it was around 5.5 ms per driver. Pipelining makes the whole read about one round trip per node, and the hedis flag lets those node round trips overlap.

How it was verified

cabal build mobility-core:lib:mobility-core passes with -Wall -Werror. HLint reports nothing on the changed lines.

Behaviour was checked against a local 3-master Redis Cluster, using the same hedis fork, with 600 keys in 600 distinct slots:

read path	time
pipelined per-slot MGETs (this PR)	2.9 ms
sequential per-slot MGETs (old non-async path)	15.5 ms
600 separate GETs	30.4 ms

• Values are correct and index-aligned, and each master received exactly its own keys, with no retries or duplicates.
• 2,500 keys work correctly past hedis's 1,000-command auto-flush.
• A WRONGTYPE key only affects its own result.
• A pipelined EVAL (single KEYS[1] per call) routes correctly and keeps script semantics.
• hedis flag (perf: optional parallel per-node execution for cluster pipelines hedis#6, GHC 9.2.7, same cluster): with the flag unset, a CPU-bound EVAL on all 3 masters in one pipeline takes 2.6–3.0× as long as on one master. With the flag on it takes 1.02–1.04×. Ordering, auto-flush and per-key error checks pass in both modes.

Tests

test/src/HedisClusterPipeline.hs adds a Redis cluster pipelining group to mobility-core-tests. It runs the real mGetClusterRaw, mGetClusterWithKeys and runPipelinedByKey against a live cluster and checks that each value lands at its key's index:

• 600 keys across slots and nodes
• missing keys
• duplicate keys
• several keys sharing one slot, where one MGET returns many values
• a non-string key
• an empty list
• 2,500 keys, past the pipeline's 1,000-command auto-flush
• zAddIfPossibleMany: keys at capacity return 0 at their own index, and every other key is reserved exactly once

The group runs only when MOBILITY_CORE_TEST_REDIS_CLUSTER_PORT is set; otherwise it reports as skipped, so CI without Redis is unaffected.

MOBILITY_CORE_TEST_REDIS_CLUSTER_PORT=30004 cabal test mobility-core:test:mobility-core-tests --test-show-details=direct --test-options='-p "Redis cluster"'

All 10 pass against a local 3-master cluster, and the full suite passes without the env var.

Notes for reviewers

• Failure granularity: a node timeout now fails the whole call, and every key comes back as a miss (logged). Before, a timeout failed only the slots on that one MGET.
• Latency metric: the per-MGET RedisCluster/mget datastore latency sample is replaced by one RedisCluster/mGetCluster sample per call.
• hedis: comes through euler-hs main; perf: optional parallel per-node execution for cluster pipelines hedis#6 and chore: bump hedis for optional parallel per-node cluster pipelines euler-hs#76 are both merged.
• Consumer PR: backend/perf: pipelined parallel-slot reservation and same-slot driver counters in pooling nammayatri#16738.

Summary by CodeRabbit

• Improvements

  • Improved Redis cluster data retrieval with more consistent handling of missing keys and command failures.
  • Improved large multi-key operations through grouped and pipelined processing.
  • Added support for efficient multi-key sorted-set updates with reservation limits.
  • Redis pipeline errors are now surfaced for more reliable failure handling.

• Bug Fixes

  • Preserved input ordering and aligned results when processing keys across multiple Redis cluster slots.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Advanced

Run ID: 49d58ca4-c4d4-44e5-ad69-f2b76cde8434

📥 Commits

Reviewing files that changed from the base of the PR and between 59ef38e and e65c0f1.

⛔ Files ignored due to path filters (1)

• flake.lock is excluded by !**/*.lock

📒 Files selected for processing (1)

• flake.nix

Included review availability: Your plan provides up to 1 included review per hour; 0 remain after this review.

---

Walkthrough

Changes

The change adds reusable Redis pipeline helpers. Cluster MGET now executes grouped slot commands and maps failures to aligned Nothing values. The change also adds batched sorted-set insertion, integration tests, test-suite wiring, and a pinned euler-hs revision.

Redis pipeline operations

Layer / File(s)	Summary
Redis pipeline helpers lib/mobility-core/src/Kernel/Storage/Hedis/Queries.hs	Adds timed, exception-capturing pipeline execution and keyed reply conversion helpers.
Cluster MGET integration lib/mobility-core/src/Kernel/Storage/Hedis/Queries.hs, lib/mobility-core/src/Kernel/Beam/Connection/EnvVars.hs	Uses grouped slot pipelines, aligns failures as Nothing, logs errors, removes fork-related constraints, and removes the asynchronous MGET environment lookup.
Sorted-set batching and integration validation lib/mobility-core/src/Kernel/Storage/Hedis/Queries.hs, lib/mobility-core/test/src/HedisClusterPipeline.hs, lib/mobility-core/test/app/Main.hs, lib/mobility-core/mobility-core.cabal	Adds zAddIfPossibleMany and reusable Lua argument helpers. Adds Redis cluster tests and wires them into the test suite when a cluster port is configured.
Dependency pinning flake.nix	Pins the euler-hs input to a specific commit.

Priority: ⬇️ Low

Estimated code review effort: 3 (Moderate) | ~30 minutes

Change: Feature

Sequence Diagram(s)

sequenceDiagram
  participant Caller
  participant Queries
  participant RedisCluster
  participant TestSuite
  Caller->>Queries: request cluster MGET or batched sorted-set insertion
  Queries->>RedisCluster: execute grouped pipeline commands
  RedisCluster-->>Queries: return aligned replies or captured errors
  Queries-->>Caller: return values or reservation results
  TestSuite->>Queries: run integration cases
  Queries->>RedisCluster: seed and validate cluster keys

Loading

Merge Risk: ⚪ Minimal · up to e65c0

The pipeline changes preserve aligned results and error handling under the supplied Redis contract and tests; no merge-blocking risk is established.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check. Docstring coverage is scoped to functions touched by this diff. Analyzed 0 functions across 0…
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely summarizes the primary changes: pipelined Redis Cluster MGET and new pipelining helpers.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch backend/perf/pipelined-cluster-redis

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{A rabbit queues keys in a neat little line
Pipelines return their replies in good time
Missing slots settle as Nothing in place
Sorted sets reserve with a measured pace
Cluster tests hop through each Redis lane
And pinned inputs keep builds the same}

_{Comment @coderabbitai help to get the list of available commands.}