fix(shard): route BITOP/COPY/GRAPH/TEMPORAL/TXN-abort/WS/MQ to owning shards#173
Merged
Merged
Conversation
…sistency-dispatch-gaps Implements frozen contract v2 of ADD task `consistency-dispatch-gaps`. All 3 wire_reachability_red tests now pass (cdg1 registry sweep, cdg2 value asserts, cdg3 purity probes). ## What changed ### dispatch_read arms (src/command/mod.rs) Added 25 missing arms to `dispatch_read_inner` so every command in COMMAND_META that is read-only has a working local and cross-shard read path: XRANGE, XREVRANGE, XREAD, XINFO, XPENDING, XLEN, ZDIFF, ZINTER, ZUNION, ZINTERCARD, ZRANDMEMBER, ZMSCORE, GEOPOS, GEODIST, GEOHASH, GEOSEARCH, LCS, RANDOMKEY, EXPIRETIME, PEXPIRETIME, TOUCH, TIME, WAIT, SLOWLOG, CDC.READ ### is_dispatch_read_supported gate (src/command/mod.rs) Fixed 7 wrong bucket lengths (GEOPOS, ZINTER, ZUNION, RANDOMKEY, XREVRANGE, PEXPIRETIME, ZRANDMEMBER all had mismatched len/b0 pairs) and added 18 new buckets for the commands above. ### extract_primary_key routing (src/server/conn/shared.rs) Corrected subcommand-first commands that would route on the wrong argument: - XGROUP/XINFO: route on args[1] (key), not args[0] (subcommand) - ZDIFF: route on args[1] (key), not args[0] (numkeys literal "2") - ZINTER/ZUNION: route on args[1] (key), not args[0] (numkeys) — critical fix: len guard was 5 (ZDIFF length) instead of 6 (ZINTER/ZUNION length), causing cross-shard ZINTER/ZUNION to always read from the wrong shard and return empty - ZINTERCARD: route on args[1] - XREAD: scan args for STREAMS token, route on the key that follows it ### CDC.READ monoio port (src/server/conn/handler_monoio/dispatch.rs + mod.rs) Added `try_handle_cdc_read` function and wired it into the monoio batch loop (cmd_len == 8 path), matching the tokio handler's existing path. ### Geo lat bounds fix (src/command/geo/mod.rs) Changed GEO_LAT_MIN/MAX from WGS84 input-clamp values (±85.05112878) to standard geohash bit-encoding bounds (±90.0), matching Redis exactly. GEOHASH now returns sqc8b* for Palermo as expected. ### EXPIRETIME u64 fix (src/storage/entry.rs) Widened CompactEntry.ttl_delta (u32) to ttl_secs (u64) — timestamps beyond year 2106 (> u32::MAX seconds) were silently truncated. EXPIRETIME now returns 9999999999 for a key with that expiry. Struct size 24→32 bytes; all constructors, methods, and size assertions updated. ### Read-only DB accessors (src/storage/db.rs) Added `get_stream_if_alive`, `get_if_alive_for_lcs` read-only methods used by the new stream/string dispatch_read arms. ### Script and consistency fixes (scripts/test-consistency.sh) - SETRANGE: added the Redis-side command that was missing - FT.CREATE: changed VECTOR FLAT to VECTOR HNSW (FLAT is not supported) ## Root-cause of the critical ZINTER bug The `extract_primary_key` routing block checked `len == 5` for ZDIFF, ZINTER, and ZUNION together. ZDIFF is 5 bytes (correct), but ZINTER and ZUNION are 6 bytes. The condition never matched for ZINTER/ZUNION, so they fell through to `args[0]` = "2" (the numkeys literal), routing to an arbitrary shard that held no data — producing a silent empty result on cross-shard reads. Closes: consistency-dispatch-gaps task (ADD phase-5) author: Tin Dang
Addresses all findings from the specialist perf/correctness review of the consistency-dispatch-gaps build commit (43e76ef): - Encoding completeness (correctness): the new zset/geo read twins used the BPTree-only get_sorted_set_if_alive, which reports listpack/legacy zsets (created by RDB load) as MISSING. All twins now use get_sorted_set_ref_if_alive (SortedSetRef handles every encoding): ZMSCORE, ZRANDMEMBER, ZDIFF/ZINTER/ZUNION/ZINTERCARD (via collect_source_sets_readonly), GEOPOS, GEODIST, GEOHASH, GEOSEARCH. - Zero-copy twins (perf): collect_source_sets_readonly returns Cow<HashMap> — BPTree/Legacy borrow their member map (no O(n) clone per source set); only small listpacks materialize. GEODIST/GEOHASH no longer clone the full member map for 1–2 lookups. - GEOSEARCH readonly rewrite: the throwaway-Database hack (full entry clone + Database::new per call) is replaced by geosearch_core — the parse/filter logic extracted from geosearch_inner and shared by both paths. The write path now fetches once up front (Redis also resolves the key object before validating options) and no longer clones the member map either. - ttl overflow guard (correctness): is_expired_at/expires_at_ms use saturating_mul(1000) — an adversarial EXPIREAT near u64::MAX/1000 now clamps to "far future" instead of panicking (debug) or silently expiring the key (release wrap). - TIME uses itoa::Buffer instead of two to_string() allocations. - SLOWLOG subcommand parse uses eq_ignore_ascii_case on raw bytes instead of an uppercase Vec copy per call. - WAIT read arm documents why it is local-path-only (keyless routing). Verified: wire_reachability_red 3/3, cargo test --lib 3570 passed, clippy clean on default and runtime-tokio,jemalloc feature sets, fmt clean. author: Tin Dang
First execution of wire_reachability_red on the SECOND runtime (tokio CI feature set, Linux VM) surfaced three harness gaps — all sampling/skip logic; every assertion is byte-identical: - cdg1 skips GRAPH.* when the `graph` feature is compiled out: M1 covers commands WITH an arm in dispatch, and a #[cfg(feature = "graph")] arm does not exist in that build, so "unknown command" is the correct reply. - cdg1 skips PSYNC off-monoio: replication is a monoio-handler special case today; the tokio divergence is recorded as an observe-phase delta alongside the 10 backlogged commands. - cdg3's aof_bytes() now counts both AOF layouts (monoio multi-part appendonlydir/, tokio top-level appendonly.aof) and samples via quiesce-polling (stable across consecutive 250ms samples, 10s cap) instead of fixed 300ms sleeps — flush cadence is runtime-dependent (1ms tick vs everysec batching). Part (c) waits for growth before sampling. Result: 3/3 green on macOS monoio AND Linux tokio (was 1/3 on tokio: GRAPH.*/PSYNC false violations + AOF probe reading the wrong path). Also records the build/verify evidence trail in TASK.md (§4 harness corrections, §5 build record, §6 checklist). author: Tin Dang
First wire exposure of the geo read twins (cdg4 consistency sweep)
surfaced four byte-parity diffs vs redis-server 8.x — the A1-class
fixes the frozen contract anticipated. Root causes and fixes:
- Score bounds restored to Redis's ±85.05112878 latitude clamp
(geohash.h); ±90 is only correct for the GEOHASH string re-encode.
Encode is now Redis-exact interleave64 (Morton, lon owns bit 51)
over offset-normalized cells — verified bit-identical for Palermo
(score 3479099956230698).
- Decode replaced bisection with Redis's direct min/max arithmetic
(geohashDecode + center), so cell-center f64s match Redis exactly.
- GEOHASH string: decode score → re-encode center at standard
geohash bounds (lat ±90) → 11 base32 chars, 11th char always '0'
(Redis's exhausted-bit-budget rule). Was "sqc8b49rnys",
Redis emits "sqc8b49rny0".
- haversine_distance rewritten in Redis's geohashGetDistance
operation order (half-angle sines, single asin, u==v==0 early
return) so GEODIST's %.4f output is byte-identical
("166274.1516" / "166.2742").
- GEOPOS/WITHCOORD coordinates now formatted via fmt_geo_coord =
Rust `{}` Display: Redis 8's d2string uses fpconv_dtoa (grisu2)
shortest-roundtrip, which Display matches byte-for-byte
("13.361389338970184" / "38.1155563954963").
New unit pins: test_redis_parity_palermo (GEOPOS, GEOHASH, GEODIST
m+km vs captured redis-server output) and fmt_geo_coord edge cases.
Local gate: 3572 lib tests pass, clippy clean both feature sets.
author: Tin Dang
… shards
The consistency suite's first honest end-to-end run at shards=4/12 (after
the harness fixes below) surfaced four pre-existing cross-shard defect
groups — all latent on main, none introduced by this branch:
1. BITOP routed by args[0] (the literal "AND"): sources were read and the
dest written on an arbitrary shard. New coordinate_bitop gathers
sources per owning shard (local direct reads, remote via single-command
MultiExecute), computes via the extracted pure bitop_compute (NOT arity
validated before any key read, Redis order; zero-padding; all-missing
-> dest deleted + 0), then SET/DELs dest on ITS owning shard.
2. COPY routed by src: dst was written into src's shard and unreadable at
its own. New coordinate_copy forwards same-owner COPYs whole; for
cross-shard string values it copies value + TTL (REPLACE/NX honored);
cross-shard non-string COPY returns an explicit error and leaves dst
untouched (full-fidelity transfer = DUMP/RESTORE territory, already
backlogged). COPY ... DB keeps the handlers' two-db interception.
CROSSSLOT cluster-mode check now slot-checks only COPY's 2 key args.
3. GRAPH.* executed on the CONNECTION's shard against a per-shard
GraphStore — which connection the kernel accepted decided whether the
graph existed ("ERR graph not found" from ~3/4 of connections at
shards=4 under SO_REUSEPORT). Both runtimes' try_handle_graph_command
now route non-local commands to the graph-name-owning shard
(graph_to_shard, {tag}-aware) via the existing — previously unwired —
ShardMessage::GraphCommand; the shard-side handler dispatches and
drains graph WAL on the owner. TXN ADDNODE/ADDEDGE intents are
captured from the routed response id; cypher writes in a cross-store
TXN against a remote graph are rejected with ERR_TXN_CROSS_SHARD
(MOVE/COPY..DB precedent). GRAPH.LIST stays connection-local
(observe delta).
Two graph-adjacent surfaces had the same defect and are routed too:
- TEMPORAL.INVALIDATE: mutation extracted to temporal::apply_invalidate;
dispatch_graph_command (the GraphCommand entry) now serves it on the
owner; both runtimes' handlers route non-local graphs through the same
hop and share apply_invalidate locally (slice-aware).
- TXN.ABORT graph rollback applied undo ops on the conn's shard. New
apply_graph_rollback (extracted verbatim from abort.rs sections 2a/2b)
+ ShardMessage::GraphRollback (boxed payload) + shard-side handler;
abort_cross_store_txn_routed partitions graph_undo/graph_intents by
owning shard, runs the full local abort, then ships each remote group
and awaits acks. All 4 abort call sites (TXN ABORT + conn teardown,
both runtimes) route. Side fix: the local graph-rollback leg is now
slice-aware (was lock-store-only — latent wrong-store access once
ShardSlice initializes).
4. WS.* and MQ.* were keyed to the CONNECTION's shard (contract v4,
"Fix WS/MQ here too"):
- The WorkspaceRegistry is per-shard, so WS AUTH from a connection the
kernel accepted on a non-creating shard answered "ERR workspace not
found" and WS LIST diverged per connection. The registry accessor is
now global (slot 0 of the per-shard array — workspaces are rare
control-plane objects, not a hot path); WAL replay and the uring
intercept converge on the same slot; WorkspaceCreate/Drop WAL records
pin to shard 0 so replay sees one totally-ordered stream. WS DROP's
prefix cleanup now targets the shard owning the {wsid} hash tag
instead of the connection's.
- MQ CREATE built the durable stream in the connection-shard's db and
registered it in that shard's DurableQueueRegistry; PUSH/POP/ACK/
DLQLEN/TRIGGER from any other connection answered "queue is not
durable" and DLQLEN read a db that never saw the queue. All six arms
now compute owner = key_to_shard(effective_key) and target the
owner's db/registry/trigger-registry/WAL (direct cross-shard access
is legal on the lock-based ShardDatabases path; the dead ShardSlice
branches keep conn-local semantics for the shardslice-migration task
to route). DLQLEN keys off the QUEUE key's owner — POP creates the
DLQ stream in the queue's db. Trigger entries land on the owner's
registry, which timers.rs already documents as the authoritative
home shard. TXN.COMMIT's MQ.PUBLISH materialization acquires each
queue's owner db per intent instead of assuming the conn's shard.
Consistency-suite harness fixes (assertions only added, none weakened):
the WS isolation probe previously ran SET on a fresh UNBOUND one-shot
redis-cli connection — it never tested workspace scoping (unbound GET
returning the value was correct behavior). AUTH/SET/GET now pipe through
ONE redis-cli process so the SET is genuinely workspace-bound and the
unbound-GET nil assertion is meaningful. Other fixes:
fresh --dir per server start (legacy TopLevel-manifest guard refused
multi-shard starts from the shared default dir), null-safe vector blobs
piped via redis-cli -x (command substitution strips \0), discriminative
phase-152 hybrid fixture (old one was all-tied on both cosine and BM25 —
"top-5" was arbitrary tie-breaking), 8 grep substitutions guarded
against set -e silent death, and txn-abort scenario 3's HSET-in-TXN
wrapped in try/except: the pre-existing "TXN does not support
cross-shard writes" rejection (reproduced on the v0.3.0 release binary;
fires whenever Linux SO_REUSEPORT lands the connection off {t}'s shard)
killed the whole suite via set -e; the count=0 / 1-shard-oracle
assertion is untouched.
Review (senior-rust-engineer subagents, one per delta): v3
APPROVE-WITH-FIXES, both applied — num_shards==1 early-exit in
coordinate_bitop/coordinate_copy (zero coordinator overhead on
single-shard servers) and debug_assert_ne in run_remote (ChannelMesh has
no self-send slot). v4 APPROVE-WITH-FIXES (lock discipline, runtime-twin
equivalence, WAL/replay coherence all confirmed), fix applied: the
io_uring batch path's WS CREATE/DROP mutated the registry with no WAL
record (pre-existing gap, load-bearing once the registry went global) —
WorkspaceCreate/Drop records now append to shard 0 there too.
New red suite tests/cross_shard_consistency_red.rs (cdg6a-g): keys are
picked via moon's own key_to_shard so sources/dest provably cross
shards from one connection; graph/WS/MQ cases use fresh connections so
Linux SO_REUSEPORT spreads them. Red established on main (macOS a/b at
GET-dest, VM c "graph not found"; WS/MQ red = the four suite failure
signatures at shards=4/12: AUTH "workspace not found", MQ "OK|no|no|0",
DLQLEN 0); 7/7 green after.
Evidence: cargo test --lib 3572 passed; wire_reachability_red 3/3;
cross_shard_consistency_red 7/7 on macOS AND the Linux VM; fmt + clippy
clean (default and runtime-tokio,jemalloc); all three feature sets
compile. test-consistency.sh: 197 PASS / 0 FAIL at shards=1, 4, AND 12
(exit 0 each, "STATUS: ALL PASSED"). bench-swf.sh on an idle VM: no
regression in any cell vs the spsc-wake-floor baseline (all configs at
or above; e.g. s4 P16 SET 2.22M vs 1.47M, GET 3.33M vs 2.13M). Full
record in TASK.md §6.
Refs: ADD task consistency-dispatch-gaps, contract v3 ("Fix all three
here", 2026-06-11) + contract v4 ("Fix WS/MQ here too", 2026-06-12).
author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
More reviews will be available in 33 minutes and 23 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThis PR closes consistency-suite gaps by expanding the dispatch_read command reach to ~25 new commands, refactoring workspace/MQ ownership to owner-shard models, enabling graph multi-shard routing with temporal.invalidate, fixing cross-shard routing for BITOP/COPY, and updating TTL representation to absolute seconds. Comprehensive failing-first integration tests validate wire reachability, correctness across shards, and read-arm purity. ChangesConsistency-Dispatch-Gaps Implementation
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
|
CI's CHANGELOG check requires every PR to touch CHANGELOG.md or carry the skip-changelog label. PR #173 is user-facing (four cross-shard command families now behave identically from every connection), so it gets a real entry: a Fixed section under [Unreleased] summarizing the BITOP, COPY, GRAPH/TEMPORAL/TXN-abort, and WS/MQ groups and the 197/197 suite result at 1/4/12 shards. Refs: PR #173, ADD task consistency-dispatch-gaps. author: Tin Dang
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
scripts/test-consistency.sh (1)
1250-1280:⚠️ Potential issue | 🟠 Major | ⚡ Quick winIncrease the WS/MQ fresh-connection fanout in the harness.
These probes still only sample one bound WS session and a couple of MQ pushes, but the frozen v4 task and PR objectives rely on many fresh connections to prove the routing is connection-independent. As written, a regression can still pass when
redis-clihappens to land on the owner shard on 4/12-shard runs. Turn these into small loops (for example, repeated freshWS AUTH/WS LISTattempts and several freshMQ PUSHes beforePOP/DLQLEN) so the script reliably exercises non-owner landings instead of sampling them opportunistically.Also applies to: 1357-1378
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scripts/test-consistency.sh` around lines 1250 - 1280, The harness currently only performs a single bound WS session and a couple MQ operations (see WS CREATE/WS AUTH/WS LIST handling around WS_ID, BOUND_OUT, WS_RESULT and the MQ PUSH/POP/DLQLEN probes), which can miss non-owner shard behavior; change those single-shot redis-cli calls into small loops that repeat fresh connections: e.g., loop N times to run fresh "WS AUTH <wsid>" + SET/GET (using new redis-cli invocations each iteration) and aggregate results into WS_RESULT_{1,4,12}, and similarly perform multiple fresh "WS LIST" and several fresh "MQ PUSH" attempts (before the POP/DLQLEN checks) to ensure some invocations land on non-owner shards; ensure you update the variables (WS_LIST, LIST_HAS_WS, UNBOUND_GET, MQ push/pop results) to collect or short-circuit appropriately so the test reliably exercises many fresh connections rather than a single opportunistic sample.src/shard/shared_databases.rs (1)
297-323:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftGlobal workspace replay no longer preserves cross-shard history order.
These records now all mutate the single slot-0 registry, but replay still walks each shard WAL independently and only preserves order within a shard. On upgrade, a workspace CREATE and later DROP that were logged on different shards can replay in shard-index order instead of real append order, which can resurrect deleted workspaces or drop live ones. The global replay needs a merged ordering source, or it needs to ignore non-slot-0 workspace WAL after an explicit migration boundary.
Also applies to: 330-344
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/shard/shared_databases.rs` around lines 297 - 323, The replay handler closure on_command that processes WalRecordType::WorkspaceCreate and WalRecordType::WorkspaceDrop mutates only workspace_registries[0] but is invoked per-shard WAL which preserves order only within a shard; fix by making global workspace replay use a merged ordering across shards (e.g., merge by a global LSN/timestamp field on WalRecord) before applying to workspace_registries[0], or alternatively skip non-slot-0 workspace records when replaying per-shard and only apply slot-0 records until an explicit migration boundary; update logic in on_command (and the loop that supplies records to it) to either perform a multi-way merge on WalRecord (use record.lsn/record.timestamp) across shards or filter out workspace create/drop records from non-slot-0 shard replays so that decode_workspace_create(), decode_workspace_drop(), WorkspaceId::from_bytes(), and WorkspaceRegistry::insert()/remove() are only applied in true global order.
🧹 Nitpick comments (3)
src/command/string/string_read.rs (1)
516-537: ⚡ Quick winAvoid cloning both inputs on the shared-read
LCSpath.This new twin eagerly copies both strings with
to_vec()and allocatesVec::new()for missing keys before building the DP table, so it pays for two extra heap buffers on top of the O(n*m) matrix. Keep the operands borrowed as&[u8](and use&[]for missing keys) until the final response buffer to stay within the command-path allocation rules and reduce peak memory. As per coding guidelines, "NoBox::new(),Vec::new(),String::new(),Arc::new(),clone(),format!(), orto_string()in command dispatch (src/command/)... UseSmallVec,itoa,write!to pre-allocated buffers, or borrow instead."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/command/string/string_read.rs` around lines 516 - 537, The current code eagerly clones both operands into Vecs (s1/s2 via to_vec() in the db.get_if_alive branches), allocating memory unnecessarily for the LCS path; change these branches to return borrowed byte slices (&[u8]) instead (use e.value.as_bytes().unwrap() -> & [u8] or &[] for missing keys) so s1/s2 are &[u8] and all DP matrix work uses borrowed slices, only allocating the final response buffer when building the Frame; update uses of s1/s2 in the LCS logic to accept &[u8] and keep the existing Frame::Error handling in the None->wrong-type branches.Source: Coding guidelines
src/server/conn/handler_monoio/txn.rs (1)
262-287: 💤 Low valueConsider passing an
Arc<Frame>from the caller to avoid allocation in the cross-shard path.
Arc::new(frame.clone())at line 267 allocates on every cross-shardTEMPORAL.INVALIDATE. While this path is only hit whenowner != ctx.shard_id(rare relative to local execution), if the caller already has the frame in anArc, propagating it here would eliminate the clone + allocation.Not blocking — the current implementation is correct and the cross-shard path is infrequent.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_monoio/txn.rs` around lines 262 - 287, The cross-shard path allocates with Arc::new(frame.clone()) inside txn.rs when building ShardMessage::GraphCommand; change the API so callers supply an Arc<Frame> (or an Option<Arc<Frame>>) instead of a plain Frame so you can forward the Arc directly without cloning the Frame, or alternatively accept both and convert only if the caller gives a raw Frame; update the call sites that invoke this code to pass the existing Arc<Frame> (or wrap once at the caller) and adjust the ShardMessage::GraphCommand construction to use the provided Arc; ensure symbols affected include the frame variable in this function, crate::shard::dispatch::graph_to_shard, ShardMessage::GraphCommand, and ctx.dispatch_tx/ctx.spsc_notifiers so the cross-shard send uses the Arc without per-send allocation.src/transaction/abort.rs (1)
280-304: 💤 Low valueConsider logging when edge properties are unexpectedly None during restore.
The module's error discipline (documented in the header) specifies logging at
tracing::warn!for unexpected states. When restoring an edge property, ifedge.propertiesisNone, the code silently skips the restoration. This situation should theoretically never occur (a SET that triggered this undo would have ensurededge.propertiesexists), so a warning would help surface any unexpected state—consistent with how missing graphs are handled at lines 248-255, 311-318, 334-340, and 364-371.🔧 Suggested improvement
let ek = EdgeKey::from(slotmap::KeyData::from_ffi(*entity_id)); if let Some(edge) = graph.write_buf.get_edge_mut(ek) { if let Some(ref mut props) = edge.properties { match old_value { // ... existing logic ... } + } else if old_value.is_some() { + tracing::warn!( + txn_id, + entity_id, + "txn abort: edge.properties is None during RestoreProperty with old_value, skipping", + ); } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/transaction/abort.rs` around lines 280 - 304, When restoring edge properties in the abort routine, add a tracing::warn! if edge.properties is None so unexpected missing properties are logged; locate the branch that constructs ek via EdgeKey::from(slotmap::KeyData::from_ffi(*entity_id)) and calls graph.write_buf.get_edge_mut(ek), and inside the Some(edge) arm, when the if let Some(ref mut props) = edge.properties fails, emit a warning that includes the EdgeKey/entity_id and prop_key to aid debugging (matching the module's error discipline used elsewhere).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.add/state.json:
- Around line 27-35: The "consistency-dispatch-gaps" task record in state.json
currently has "gate": "none" but must persist the real gate decision; update the
"consistency-dispatch-gaps" object's "gate" value to the actual outcome (e.g.,
"PASS" or the specific recorded decision) so tasks[slug].gate reflects the
authoritative result surfaced by cmd_guide, and also ensure the "updated"
timestamp is set when making this change to keep metadata consistent with
TASK.md §6.
In @.add/tasks/consistency-dispatch-gaps/TASK.md:
- Around line 55-72: Update this task document to consistently use "25" new
dispatch_read arms (plus the monoio CDC.READ exception) everywhere the old "20"
appears so the frozen contract, red suite, and all acceptance criteria match;
search for occurrences of "20 read arms", "20 commands", and related phrasing
near references to dispatch_read, frozen contract, red suite, and CDC.READ and
replace them with the normalized "25" wording (also verify the other listed
regions around lines 120-146 and 341-368 reflect the same change) so the spec is
unambiguous for verify/gate runs.
In `@src/command/geo/mod.rs`:
- Around line 107-130: Both geohash_to_string and fmt_geo_coord allocate Strings
on the hot command path; change them to write into a caller-provided buffer
instead. Replace geohash_to_string (which currently calls
String::from_utf8_lossy on result) with a version that accepts a mutable byte
buffer (e.g., &mut Vec<u8> or &mut BytesMut) and pushes the 11 geohash bytes
directly into it; change fmt_geo_coord (which currently uses format!("{v}") to
allocate) to accept the same mutable buffer and format the f64 into it using a
non-allocating formatter (e.g., write! to the buffer or ryu::raw::format64 into
the buffer). Update the GEO callers to pass a pre-allocated buffer and construct
Bytes from that buffer (avoiding to_string/format!/String allocations) before
sending the reply.
- Around line 74-78: In geohash_encode_raw, the computed lat_offset and
lon_offset can equal (1 << GEO_STEP_MAX) for top-edge coordinates which then
loses the high bit when passed to interleave64; clamp both offsets to the
inclusive upper index ((1u64 << GEO_STEP_MAX) - 1) before casting to u32 so that
lon == 180.0 and lat == 85.05112878 map to the final cell rather than wrapping
to the first cell; update the logic around lat_offset and lon_offset (and their
casts) in geohash_encode_raw to enforce this clamp using GEO_STEP_MAX and then
call interleave64 with the clamped u32 values.
In `@src/command/key.rs`:
- Around line 1248-1252: randomkey_readonly currently ignores the read-path
timestamp and calls Database::random_key(), which uses self.cached_now_ms;
update the read path to use the caller's now_ms by threading it into the picker:
add or use a Database method that accepts the snapshot timestamp (e.g.
random_key_if_alive(now_ms) or change random_key to take now_ms) and replace the
call in randomkey_readonly to call that timestamped helper (randomkey_readonly
-> Database::random_key_if_alive(now_ms)) so expiry filtering uses the provided
now_ms consistent with other readonly/twin read semantics.
In `@src/command/stream/stream_read.rs`:
- Around line 888-904: The code recalculates now with
crate::storage::entry::current_time_ms() causing timestamp drift; instead use
the cached timestamp passed in (now_ms) when computing consumer idle time. In
the mapping that builds the consumers Vec (the closure iterating group.consumers
and producing Frame::Array entries), replace the local now usage with the now_ms
parameter and compute idle as now_ms.saturating_sub(c.seen_time) so idle is
consistent with dispatch_read's cached time.
- Around line 969-1049: The XPENDING parser extracts an IDLE value into
_min_idle but never applies it; update the readonly path to enforce the idle
threshold by filtering the results from stream.xpending_detail (or by passing
_min_idle into xpending_detail if you prefer API change). Concretely, keep
extracting _min_idle as currently done, then after obtaining details from
stream.xpending_detail(&group, start, end, count, consumer_filter) filter the
iterator by idle >= _min_idle before mapping to Frames (use the existing tuple
(id, consumer, idle, delivery_count) to compare). Ensure the behavior is
consistent whether _min_idle is zero (no filtering) and preserve error handling
when xpending_detail returns Err.
- Around line 759-779: The current loop using db.get_stream_if_alive(...) always
pushes a [key, []] entry for misses, causing mixed hit/miss XREAD replies to
have the wrong RESP shape; change the logic in the block that builds results
(the code using db.get_stream_if_alive, Stream::range, format_entry, and pushing
into results) so that you first determine per-stream whether any entries exist
(set has_entries when entries is non-empty) and, if any stream in the entire
request has entries, only push streams that have entries; otherwise (no streams
had entries) keep the current behavior of returning empty arrays for each
requested key. Ensure you still use next_id and StreamId::MAX for the range and
preserve format_entry for creating entry frames.
In `@src/server/conn/handler_sharded/write.rs`:
- Around line 129-137: The WS owner routing (compute prefix and owner via
key_to_shard using ws_id.as_hex()) is being ignored in the
ShardSlice::is_initialized() branch because that branch calls with_shard_db(...)
which uses the current accepting shard; change those branches to use the
computed owner shard instead—i.e., replace or route the
with_shard_db(...)/shard_db access so it obtains the DB for owner (use
ctx.shard_databases.write_db(owner, 0) or equivalent read_db) and perform
WS.DROP and MQ CREATE/PUSH/POP/ACK/DLQLEN/TRIGGER operations against that owner
DB; ensure every occurrence (including the other ranges called out) uses the
same owner calculation (prefix, key_to_shard) and uses owner-based DB access
instead of the accepting-shard with_shard_db path.
In `@src/server/conn/shared.rs`:
- Around line 377-390: The COPY single-key routing arm is too permissive (uses
args.len() >= 2) so COPY src dst DB n still matches the single-key route; update
the COPY branch in shared.rs to require at least two key arguments (change
args.len() >= 2 to args.len() >= 3) while keeping the existing exclusion of the
"DB" form (the cmd/args check and the matches!(a, Frame::BulkString(o) if
o.eq_ignore_ascii_case(b"DB")) logic) so that "COPY src dst DB n" does not get
treated as a single-key operation and will be handled by the multi-key/two-db
path.
In `@src/shard/coordinator.rs`:
- Around line 115-585: The file is too large; extract the new coordinator logic
into submodules: move coordinate_bitop (and its helpers bulk, bulk_static) into
a new shard submodule (e.g., shard::coordinator::bitop) and move coordinate_copy
into shard::coordinator::copy (or a single shard::coordinator::multi_key
module), keeping run_local/run_remote/run_on_owner in the original
coordinator.rs and calling into the new modules. Create mod declarations (mod
bitop; mod copy;) and re-export or use super:: items as needed, update imports
so the moved functions reference run_local, run_on_owner, key_to_shard,
extract_key, shard_databases, dispatch_tx, spsc_notifiers, cached_clock, etc.,
and set visibility (pub(crate) or pub) to match callers; run cargo build/tests
to fix any borrow/visibility issues. Ensure no logic changes—only code moved
into new files and references adjusted.
- Around line 492-583: The current cross-shard COPY path issues separate
run_on_owner calls (run_on_owner + bulk_static("GET"), run_on_owner +
bulk_static("PTTL"), and later separate SET and PEXPIRE calls) which is not
snapshot-safe; instead fetch and restore atomically on the owners: replace the
separate GET/PTTL calls with a single atomic fetch on src (either run DUMP on
src and capture the serialized payload and TTL, or execute a single EVAL/LUA on
the src owner that returns value+ttl together), and replace the separate SET
then PEXPIRE on dst with a single atomic restore on the dst owner (RESTORE with
TTL or a single EVAL that performs SET + PEXPIRE atomically, honoring
replace/NX). Update code paths that build set_parts, the value/ttl_ms handling,
and the run_on_owner invocations so run_on_owner is used to perform a single
src-side call (DUMP or EVAL returning payload+ttl) and a single dst-side call
(RESTORE or EVAL performing atomic write+ttl), preserving replace semantics and
returning errors through Frame the same way.
In `@src/storage/entry.rs`:
- Around line 434-442: set_expires_at_ms currently converts milliseconds to
seconds losing sub-second precision; change the stored field from ttl_secs
(u32/u64 seconds) to a millisecond-based field (e.g., ttl_ms: u64) and store the
exact ms value (0 to mean no expiry) to preserve semantics and avoid year-2106
overflow. Update set_expires_at_ms to assign ms directly (keeping the
0-as-no-expiry convention), update expires_at_ms() to return the u64 millisecond
value, and update is_expired_at() to compare in milliseconds; also adjust any
serialization/deserialization, constructors, and tests that reference ttl_secs
so they use ttl_ms. Ensure types and names (ttl_secs, set_expires_at_ms,
expires_at_ms, is_expired_at) are updated consistently throughout the module.
---
Outside diff comments:
In `@scripts/test-consistency.sh`:
- Around line 1250-1280: The harness currently only performs a single bound WS
session and a couple MQ operations (see WS CREATE/WS AUTH/WS LIST handling
around WS_ID, BOUND_OUT, WS_RESULT and the MQ PUSH/POP/DLQLEN probes), which can
miss non-owner shard behavior; change those single-shot redis-cli calls into
small loops that repeat fresh connections: e.g., loop N times to run fresh "WS
AUTH <wsid>" + SET/GET (using new redis-cli invocations each iteration) and
aggregate results into WS_RESULT_{1,4,12}, and similarly perform multiple fresh
"WS LIST" and several fresh "MQ PUSH" attempts (before the POP/DLQLEN checks) to
ensure some invocations land on non-owner shards; ensure you update the
variables (WS_LIST, LIST_HAS_WS, UNBOUND_GET, MQ push/pop results) to collect or
short-circuit appropriately so the test reliably exercises many fresh
connections rather than a single opportunistic sample.
In `@src/shard/shared_databases.rs`:
- Around line 297-323: The replay handler closure on_command that processes
WalRecordType::WorkspaceCreate and WalRecordType::WorkspaceDrop mutates only
workspace_registries[0] but is invoked per-shard WAL which preserves order only
within a shard; fix by making global workspace replay use a merged ordering
across shards (e.g., merge by a global LSN/timestamp field on WalRecord) before
applying to workspace_registries[0], or alternatively skip non-slot-0 workspace
records when replaying per-shard and only apply slot-0 records until an explicit
migration boundary; update logic in on_command (and the loop that supplies
records to it) to either perform a multi-way merge on WalRecord (use
record.lsn/record.timestamp) across shards or filter out workspace create/drop
records from non-slot-0 shard replays so that decode_workspace_create(),
decode_workspace_drop(), WorkspaceId::from_bytes(), and
WorkspaceRegistry::insert()/remove() are only applied in true global order.
---
Nitpick comments:
In `@src/command/string/string_read.rs`:
- Around line 516-537: The current code eagerly clones both operands into Vecs
(s1/s2 via to_vec() in the db.get_if_alive branches), allocating memory
unnecessarily for the LCS path; change these branches to return borrowed byte
slices (&[u8]) instead (use e.value.as_bytes().unwrap() -> & [u8] or &[] for
missing keys) so s1/s2 are &[u8] and all DP matrix work uses borrowed slices,
only allocating the final response buffer when building the Frame; update uses
of s1/s2 in the LCS logic to accept &[u8] and keep the existing Frame::Error
handling in the None->wrong-type branches.
In `@src/server/conn/handler_monoio/txn.rs`:
- Around line 262-287: The cross-shard path allocates with
Arc::new(frame.clone()) inside txn.rs when building ShardMessage::GraphCommand;
change the API so callers supply an Arc<Frame> (or an Option<Arc<Frame>>)
instead of a plain Frame so you can forward the Arc directly without cloning the
Frame, or alternatively accept both and convert only if the caller gives a raw
Frame; update the call sites that invoke this code to pass the existing
Arc<Frame> (or wrap once at the caller) and adjust the
ShardMessage::GraphCommand construction to use the provided Arc; ensure symbols
affected include the frame variable in this function,
crate::shard::dispatch::graph_to_shard, ShardMessage::GraphCommand, and
ctx.dispatch_tx/ctx.spsc_notifiers so the cross-shard send uses the Arc without
per-send allocation.
In `@src/transaction/abort.rs`:
- Around line 280-304: When restoring edge properties in the abort routine, add
a tracing::warn! if edge.properties is None so unexpected missing properties are
logged; locate the branch that constructs ek via
EdgeKey::from(slotmap::KeyData::from_ffi(*entity_id)) and calls
graph.write_buf.get_edge_mut(ek), and inside the Some(edge) arm, when the if let
Some(ref mut props) = edge.properties fails, emit a warning that includes the
EdgeKey/entity_id and prop_key to aid debugging (matching the module's error
discipline used elsewhere).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: aa35247d-79df-475b-bb94-da3d9a9e6167
📒 Files selected for processing (33)
.add/state.json.add/tasks/consistency-dispatch-gaps/BUILD-CONTEXT.md.add/tasks/consistency-dispatch-gaps/TASK.mdscripts/test-consistency.shsrc/admin/slowlog.rssrc/command/geo/geo_cmd.rssrc/command/geo/mod.rssrc/command/graph/mod.rssrc/command/key.rssrc/command/mod.rssrc/command/sorted_set/sorted_set_read.rssrc/command/stream/stream_read.rssrc/command/string/string_bit.rssrc/command/string/string_read.rssrc/command/temporal.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_sharded/txn.rssrc/server/conn/handler_sharded/write.rssrc/server/conn/shared.rssrc/shard/coordinator.rssrc/shard/dispatch.rssrc/shard/shared_databases.rssrc/shard/spsc_handler.rssrc/shard/uring_handler.rssrc/storage/db.rssrc/storage/entry.rssrc/transaction/abort.rstests/cross_shard_consistency_red.rstests/wire_reachability_red.rs
Comment on lines
+55
to
+72
| - **Add 20 read arms to dispatch_read (chosen — user decision at freeze)** — each of | ||
| the 20 commands gains a proper arm in `dispatch_read` built on the read-only | ||
| accessor pattern the existing 62 arms already use (`&Database`, lazy-expired keys | ||
| invisible but NOT deleted under the shared lock). Reads stay on the shared-lock | ||
| read path — full read parallelism, no write-lock detour, and the | ||
| `is_dispatch_read_supported` false positives (GEOPOS et al.) are fixed by the arms | ||
| existing. Feasibility verified up front: TOUCH = read-only `exists` per key | ||
| (dispatch_read already has an EXISTS arm); XREAD's BLOCK option is parsed-but- | ||
| ignored in today's dispatch impl, so a non-blocking arm mirrors current behavior; | ||
| the remainder are pure reads. | ||
| - *NotSupported fallback through the write path* — offered, declined by the user: | ||
| one mechanism for all gaps + future-proofing, but routes these reads through the | ||
| exclusive write path (slower, and A2-class side-effect review burden). The | ||
| future-drift protection it offered is provided instead by M5's registry-sweep test | ||
| (CI fails on the next stranded command). | ||
| - *Exact-set gate (sync the predicate to the arms)* — subsumed: the gate is updated | ||
| to include the new commands' (len, first-byte) buckets so the cross-shard fast | ||
| path serves them too; the sweep test pins gate/arm agreement. |
There was a problem hiding this comment.
Normalize the frozen reachability count throughout this task file.
These sections still use the old “20 commands” wording, but the frozen contract below and the red suite both target 25 new dispatch_read arms plus the monoio CDC.READ exception. Leaving both numbers in the same task artifact makes the acceptance criteria ambiguous for future verify/gate runs and weakens this file’s role as the authoritative spec.
Also applies to: 120-146, 341-368
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/tasks/consistency-dispatch-gaps/TASK.md around lines 55 - 72, Update
this task document to consistently use "25" new dispatch_read arms (plus the
monoio CDC.READ exception) everywhere the old "20" appears so the frozen
contract, red suite, and all acceptance criteria match; search for occurrences
of "20 read arms", "20 commands", and related phrasing near references to
dispatch_read, frozen contract, red suite, and CDC.READ and replace them with
the normalized "25" wording (also verify the other listed regions around lines
120-146 and 341-368 reflect the same change) so the spec is unambiguous for
verify/gate runs.
Comment on lines
+74
to
+78
| fn geohash_encode_raw(lon: f64, lat: f64, lat_min: f64, lat_max: f64) -> u64 { | ||
| let lat_offset = (lat - lat_min) / (lat_max - lat_min) * (1u64 << GEO_STEP_MAX) as f64; | ||
| let lon_offset = | ||
| (lon - GEO_LON_MIN) / (GEO_LON_MAX - GEO_LON_MIN) * (1u64 << GEO_STEP_MAX) as f64; | ||
| interleave64(lat_offset as u32, lon_offset as u32) |
There was a problem hiding this comment.
Clamp the encoded GEO cell index at the inclusive upper bounds.
lon == 180.0 or lat == 85.05112878 currently scales to 1 << 26, but interleave64 only preserves the low 26 bits. At the max bound that extra bit gets masked away, so the top-edge coordinate round-trips as the first cell instead of the last one.
Suggested fix
fn geohash_encode_raw(lon: f64, lat: f64, lat_min: f64, lat_max: f64) -> u64 {
- let lat_offset = (lat - lat_min) / (lat_max - lat_min) * (1u64 << GEO_STEP_MAX) as f64;
- let lon_offset =
- (lon - GEO_LON_MIN) / (GEO_LON_MAX - GEO_LON_MIN) * (1u64 << GEO_STEP_MAX) as f64;
+ let scale = (1u64 << GEO_STEP_MAX) as f64;
+ let max_idx = scale - 1.0;
+ let lat_offset = ((lat - lat_min) / (lat_max - lat_min) * scale).clamp(0.0, max_idx);
+ let lon_offset =
+ ((lon - GEO_LON_MIN) / (GEO_LON_MAX - GEO_LON_MIN) * scale).clamp(0.0, max_idx);
interleave64(lat_offset as u32, lon_offset as u32)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| fn geohash_encode_raw(lon: f64, lat: f64, lat_min: f64, lat_max: f64) -> u64 { | |
| let lat_offset = (lat - lat_min) / (lat_max - lat_min) * (1u64 << GEO_STEP_MAX) as f64; | |
| let lon_offset = | |
| (lon - GEO_LON_MIN) / (GEO_LON_MAX - GEO_LON_MIN) * (1u64 << GEO_STEP_MAX) as f64; | |
| interleave64(lat_offset as u32, lon_offset as u32) | |
| fn geohash_encode_raw(lon: f64, lat: f64, lat_min: f64, lat_max: f64) -> u64 { | |
| let scale = (1u64 << GEO_STEP_MAX) as f64; | |
| let max_idx = scale - 1.0; | |
| let lat_offset = ((lat - lat_min) / (lat_max - lat_min) * scale).clamp(0.0, max_idx); | |
| let lon_offset = | |
| ((lon - GEO_LON_MIN) / (GEO_LON_MAX - GEO_LON_MIN) * scale).clamp(0.0, max_idx); | |
| interleave64(lat_offset as u32, lon_offset as u32) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/command/geo/mod.rs` around lines 74 - 78, In geohash_encode_raw, the
computed lat_offset and lon_offset can equal (1 << GEO_STEP_MAX) for top-edge
coordinates which then loses the high bit when passed to interleave64; clamp
both offsets to the inclusive upper index ((1u64 << GEO_STEP_MAX) - 1) before
casting to u32 so that lon == 180.0 and lat == 85.05112878 map to the final cell
rather than wrapping to the first cell; update the logic around lat_offset and
lon_offset (and their casts) in geohash_encode_raw to enforce this clamp using
GEO_STEP_MAX and then call interleave64 with the clamped u32 values.
Comment on lines
107
to
+130
| pub(crate) fn geohash_to_string(score: f64) -> String { | ||
| const ALPHABET: &[u8] = b"0123456789bcdefghjkmnpqrstuvwxyz"; | ||
| let hash = score as u64; | ||
| // Redis uses 11 characters (55 bits, but we only have 52 so pad with 0) | ||
| let padded = hash << 3; // shift left 3 to fill 55 bits | ||
| let (lon, lat) = geohash_decode(score); | ||
| let bits = geohash_encode_raw(lon, lat, -90.0, 90.0); | ||
| let mut result = [0u8; 11]; | ||
| for i in 0..11 { | ||
| let idx = ((padded >> (50 - i * 5)) & 0x1F) as usize; | ||
| result[i] = ALPHABET[idx]; | ||
| for (i, slot) in result.iter_mut().enumerate() { | ||
| let used = (i + 1) * 5; | ||
| let idx = if used <= 52 { | ||
| ((bits >> (52 - used)) & 0x1F) as usize | ||
| } else { | ||
| 0 | ||
| }; | ||
| *slot = ALPHABET[idx]; | ||
| } | ||
| String::from_utf8_lossy(&result).to_string() | ||
| } | ||
|
|
||
| /// Format a coordinate the way Redis 8 replies to GEOPOS / WITHCOORD. | ||
| /// Redis's d2string uses fpconv_dtoa (grisu2) — the SHORTEST decimal that | ||
| /// round-trips to the same f64 — which is exactly Rust's `{}` Display for | ||
| /// f64 (verified byte-identical against redis-server 8.x for the geohash | ||
| /// cell centers the consistency suite compares). | ||
| pub(crate) fn fmt_geo_coord(v: f64) -> String { | ||
| format!("{v}") |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Keep GEO reply formatting off format!() / to_string() in src/command/.
These helpers now allocate reply strings on the command path, which is exactly the hot path the repo policy forbids. Please switch this to a buffer-based formatter and have the GEO callers build Bytes from that buffer directly.
As per coding guidelines, "No Box::new(), Vec::new(), String::new(), Arc::new(), clone(), format!(), or to_string() in command dispatch (src/command/), protocol parsing (src/protocol/), shard event loops (src/shard/event_loop.rs), or I/O drivers (src/io/)."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/command/geo/mod.rs` around lines 107 - 130, Both geohash_to_string and
fmt_geo_coord allocate Strings on the hot command path; change them to write
into a caller-provided buffer instead. Replace geohash_to_string (which
currently calls String::from_utf8_lossy on result) with a version that accepts a
mutable byte buffer (e.g., &mut Vec<u8> or &mut BytesMut) and pushes the 11
geohash bytes directly into it; change fmt_geo_coord (which currently uses
format!("{v}") to allocate) to accept the same mutable buffer and format the f64
into it using a non-allocating formatter (e.g., write! to the buffer or
ryu::raw::format64 into the buffer). Update the GEO callers to pass a
pre-allocated buffer and construct Bytes from that buffer (avoiding
to_string/format!/String allocations) before sending the reply.
Source: Coding guidelines
Comment on lines
+1248
to
+1252
| pub fn randomkey_readonly(db: &Database, _args: &[Frame], _now_ms: u64) -> Frame { | ||
| match db.random_key() { | ||
| Some(key) => Frame::BulkString(key), | ||
| None => Frame::Null, | ||
| } |
There was a problem hiding this comment.
Use the read-path timestamp when picking RANDOMKEY.
randomkey_readonly ignores now_ms and calls Database::random_key(), but that helper filters expiry with self.cached_now_ms instead of the shared-read snapshot. Since dispatch_read now exposes RANDOMKEY on the fast read path, this can return a key that is already expired at the caller’s now_ms. Thread now_ms into the picker (for example via a random_key_if_alive(now_ms) helper) so readonly expiry semantics stay consistent with the rest of the new twins.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/command/key.rs` around lines 1248 - 1252, randomkey_readonly currently
ignores the read-path timestamp and calls Database::random_key(), which uses
self.cached_now_ms; update the read path to use the caller's now_ms by threading
it into the picker: add or use a Database method that accepts the snapshot
timestamp (e.g. random_key_if_alive(now_ms) or change random_key to take now_ms)
and replace the call in randomkey_readonly to call that timestamped helper
(randomkey_readonly -> Database::random_key_if_alive(now_ms)) so expiry
filtering uses the provided now_ms consistent with other readonly/twin read
semantics.
Comment on lines
129
to
+137
| let prefix = format!("{{{}}}:", ws_id.as_hex()); | ||
| let mut db_guard = ctx.shard_databases.write_db(ctx.shard_id, 0); | ||
| // The {wsid} hash tag co-locates every workspace | ||
| // key on ONE shard — clean up there, not on the | ||
| // connection's shard. | ||
| let owner = crate::shard::dispatch::key_to_shard( | ||
| prefix.as_bytes(), | ||
| ctx.num_shards, | ||
| ); | ||
| let mut db_guard = ctx.shard_databases.write_db(owner, 0); |
There was a problem hiding this comment.
Owner routing stops at the fallback path.
Each of these blocks computes an owner, but the ShardSlice::is_initialized() branch still executes against the current shard via with_shard_db(...). That means the WS.DROP cleanup and MQ CREATE/PUSH/POP/ACK/DLQLEN/TRIGGER fixes only work in the lock-based fallback; as soon as this handler runs with ShardSlice active, those operations keep reading/writing the accepting shard instead of the owning shard.
Also applies to: 279-305, 356-379, 442-547, 567-588, 620-640, 666-677
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/handler_sharded/write.rs` around lines 129 - 137, The WS
owner routing (compute prefix and owner via key_to_shard using ws_id.as_hex())
is being ignored in the ShardSlice::is_initialized() branch because that branch
calls with_shard_db(...) which uses the current accepting shard; change those
branches to use the computed owner shard instead—i.e., replace or route the
with_shard_db(...)/shard_db access so it obtains the DB for owner (use
ctx.shard_databases.write_db(owner, 0) or equivalent read_db) and perform
WS.DROP and MQ CREATE/PUSH/POP/ACK/DLQLEN/TRIGGER operations against that owner
DB; ensure every occurrence (including the other ranges called out) uses the
same owner calculation (prefix, key_to_shard) and uses owner-based DB access
instead of the accepting-shard with_shard_db path.
Comment on lines
+377
to
+390
| // BITOP <op> dest src...: dest + sources can live on different shards. | ||
| (5, b'b') => args.len() >= 3 && cmd.eq_ignore_ascii_case(b"BITOP"), | ||
| // COPY src dst [REPLACE]: src and dst can live on different shards. | ||
| // The `COPY ... DB n` form is EXCLUDED: it needs two databases and is | ||
| // owned by the handlers' two-db interception (cross-db + cross-shard | ||
| // simultaneously is unsupported, as before). | ||
| (4, b'c') => { | ||
| args.len() >= 2 | ||
| && cmd.eq_ignore_ascii_case(b"COPY") | ||
| && !args | ||
| .iter() | ||
| .skip(2) | ||
| .any(|a| matches!(a, Frame::BulkString(o) if o.eq_ignore_ascii_case(b"DB"))) | ||
| } |
There was a problem hiding this comment.
COPY ... DB can still be dispatched to the wrong shard.
Excluding the DB form here leaves it on the single-key route, and the downstream routers still hash only args[0] for COPY. A COPY src dst DB n whose source and destination live on different shards will therefore run on the source shard's two-DB fast path and create dst on the wrong owner instead of rejecting or coordinating the cross-shard case.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/shared.rs` around lines 377 - 390, The COPY single-key
routing arm is too permissive (uses args.len() >= 2) so COPY src dst DB n still
matches the single-key route; update the COPY branch in shared.rs to require at
least two key arguments (change args.len() >= 2 to args.len() >= 3) while
keeping the existing exclusion of the "DB" form (the cmd/args check and the
matches!(a, Frame::BulkString(o) if o.eq_ignore_ascii_case(b"DB")) logic) so
that "COPY src dst DB n" does not get treated as a single-key operation and will
be handled by the multi-key/two-db path.
Comment on lines
+115
to
+585
| // --------------------------------------------------------------------------- | ||
| // Shared legs for the BITOP / COPY coordinators | ||
| // --------------------------------------------------------------------------- | ||
|
|
||
| /// Run one full command on the LOCAL shard through the real dispatcher | ||
| /// (identical semantics to a remote MultiExecute leg, minus the hop). | ||
| fn run_local( | ||
| shard_databases: &Arc<ShardDatabases>, | ||
| my_shard: usize, | ||
| db_index: usize, | ||
| cached_clock: &CachedClock, | ||
| cmd: &[u8], | ||
| args: &[Frame], | ||
| ) -> Frame { | ||
| let db_count = shard_databases.db_count(); | ||
| let mut selected = db_index; | ||
| let mut run = |db: &mut crate::storage::Database| { | ||
| db.refresh_now_from_cache(cached_clock); | ||
| match cmd_dispatch(db, cmd, args, &mut selected, db_count) { | ||
| DispatchResult::Response(f) | DispatchResult::Quit(f) => f, | ||
| } | ||
| }; | ||
| // Phase 2c: gate on is_initialized(); new path uses ShardSlice directly. | ||
| if crate::shard::slice::is_initialized() { | ||
| crate::shard::slice::with_shard_db(db_index, run) | ||
| } else { | ||
| let mut guard = shard_databases.write_db(my_shard, db_index); | ||
| run(&mut guard) | ||
| } | ||
| } | ||
|
|
||
| /// Send one full command to a REMOTE shard and await its reply. | ||
| async fn run_remote( | ||
| target_shard: usize, | ||
| routing_key: &Bytes, | ||
| command: Frame, | ||
| my_shard: usize, | ||
| db_index: usize, | ||
| dispatch_tx: &Rc<RefCell<Vec<HeapProd<ShardMessage>>>>, | ||
| spsc_notifiers: &[Arc<channel::Notify>], | ||
| ) -> Frame { | ||
| // ChannelMesh has no self-send slot — local legs must go via run_local. | ||
| debug_assert_ne!(target_shard, my_shard, "run_remote called for own shard"); | ||
| let (reply_tx, reply_rx) = channel::oneshot(); | ||
| let msg = ShardMessage::MultiExecute { | ||
| db_index, | ||
| commands: vec![(routing_key.clone(), command)], | ||
| reply_tx, | ||
| }; | ||
| spsc_send(dispatch_tx, my_shard, target_shard, msg, spsc_notifiers).await; | ||
| match reply_rx.recv().await { | ||
| Ok(mut frames) if !frames.is_empty() => frames.swap_remove(0), | ||
| _ => Frame::Error(Bytes::from_static(b"ERR cross-shard reply channel closed")), | ||
| } | ||
| } | ||
|
|
||
| /// Run one full command on whichever shard owns `routing_key`. | ||
| #[allow(clippy::too_many_arguments)] | ||
| async fn run_on_owner( | ||
| routing_key: &Bytes, | ||
| command_parts: &[Frame], | ||
| my_shard: usize, | ||
| num_shards: usize, | ||
| db_index: usize, | ||
| shard_databases: &Arc<ShardDatabases>, | ||
| dispatch_tx: &Rc<RefCell<Vec<HeapProd<ShardMessage>>>>, | ||
| spsc_notifiers: &[Arc<channel::Notify>], | ||
| cached_clock: &CachedClock, | ||
| ) -> Frame { | ||
| let owner = key_to_shard(routing_key, num_shards); | ||
| if owner == my_shard { | ||
| let (cmd, args) = match command_parts.split_first() { | ||
| Some((Frame::BulkString(c), rest)) => (c.clone(), rest), | ||
| _ => return Frame::Error(Bytes::from_static(b"ERR invalid command format")), | ||
| }; | ||
| run_local( | ||
| shard_databases, | ||
| my_shard, | ||
| db_index, | ||
| cached_clock, | ||
| &cmd, | ||
| args, | ||
| ) | ||
| } else { | ||
| let command = Frame::Array(command_parts.to_vec().into()); | ||
| run_remote( | ||
| owner, | ||
| routing_key, | ||
| command, | ||
| my_shard, | ||
| db_index, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| ) | ||
| .await | ||
| } | ||
| } | ||
|
|
||
| fn bulk(b: &Bytes) -> Frame { | ||
| Frame::BulkString(b.clone()) | ||
| } | ||
|
|
||
| fn bulk_static(s: &'static [u8]) -> Frame { | ||
| Frame::BulkString(Bytes::from_static(s)) | ||
| } | ||
|
|
||
| /// Coordinate BITOP across shards. | ||
| /// | ||
| /// `BITOP <op> dest src [src ...]` — sources are gathered (local read or | ||
| /// remote GET), combined via the same `bitop_compute` the local path uses, | ||
| /// and the result is written on DEST's owning shard (SET, or DEL when the | ||
| /// combine is empty). Full Redis semantics: BITOP is string-only by spec, | ||
| /// so value transfer is exact; WRONGTYPE from any source propagates. | ||
| #[allow(clippy::too_many_arguments)] | ||
| async fn coordinate_bitop( | ||
| args: &[Frame], | ||
| my_shard: usize, | ||
| num_shards: usize, | ||
| db_index: usize, | ||
| shard_databases: &Arc<ShardDatabases>, | ||
| dispatch_tx: &Rc<RefCell<Vec<HeapProd<ShardMessage>>>>, | ||
| spsc_notifiers: &[Arc<channel::Notify>], | ||
| cached_clock: &CachedClock, | ||
| _response_pool: &(), | ||
| ) -> Frame { | ||
| // Single-shard server: straight to local dispatch — zero coordinator | ||
| // overhead (no key vec, no owner hashing) on the 1-shard hot path. | ||
| if num_shards == 1 { | ||
| return run_local( | ||
| shard_databases, | ||
| my_shard, | ||
| db_index, | ||
| cached_clock, | ||
| b"BITOP", | ||
| args, | ||
| ); | ||
| } | ||
| if args.len() < 3 { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR wrong number of arguments for 'BITOP' command", | ||
| )); | ||
| } | ||
| let (Some(op), Some(dest)) = (extract_key(&args[0]), extract_key(&args[1])) else { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR wrong number of arguments for 'BITOP' command", | ||
| )); | ||
| }; | ||
| // Redis validation order: NOT arity errors before any key is touched. | ||
| if op.eq_ignore_ascii_case(b"NOT") && args.len() != 3 { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR BITOP NOT requires one and only one key", | ||
| )); | ||
| } | ||
| let mut src_keys: Vec<Bytes> = Vec::with_capacity(args.len() - 2); | ||
| for arg in &args[2..] { | ||
| match extract_key(arg) { | ||
| Some(k) => src_keys.push(k), | ||
| None => { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR wrong number of arguments for 'BITOP' command", | ||
| )); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Single-owner fast path: every key (dest included) on one shard — | ||
| // forward the whole command for byte-identical local semantics. | ||
| let dest_shard = key_to_shard(&dest, num_shards); | ||
| if src_keys | ||
| .iter() | ||
| .all(|k| key_to_shard(k, num_shards) == dest_shard) | ||
| { | ||
| let mut parts: Vec<Frame> = Vec::with_capacity(args.len() + 1); | ||
| parts.push(bulk_static(b"BITOP")); | ||
| parts.extend_from_slice(args); | ||
| return run_on_owner( | ||
| &dest, | ||
| &parts, | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| } | ||
|
|
||
| // Gather sources: group by shard ascending (VLL), local legs direct, | ||
| // remote legs as GET batches. | ||
| let mut groups: BTreeMap<usize, Vec<(usize, Bytes)>> = BTreeMap::new(); | ||
| for (i, k) in src_keys.iter().enumerate() { | ||
| groups | ||
| .entry(key_to_shard(k, num_shards)) | ||
| .or_default() | ||
| .push((i, k.clone())); | ||
| } | ||
| let mut sources: Vec<Option<Vec<u8>>> = vec![None; src_keys.len()]; | ||
| let mut pending: Vec<(Vec<usize>, channel::OneshotReceiver<Vec<Frame>>)> = Vec::new(); | ||
| for (shard_id, indexed) in &groups { | ||
| if *shard_id == my_shard { | ||
| for (idx, key) in indexed { | ||
| let reply = run_local( | ||
| shard_databases, | ||
| my_shard, | ||
| db_index, | ||
| cached_clock, | ||
| b"GET", | ||
| &[bulk(key)], | ||
| ); | ||
| match reply { | ||
| Frame::BulkString(v) => sources[*idx] = Some(v.to_vec()), | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| _ => sources[*idx] = Some(Vec::new()), | ||
| } | ||
| } | ||
| } else { | ||
| let (reply_tx, reply_rx) = channel::oneshot(); | ||
| let commands: Vec<(Bytes, Frame)> = indexed | ||
| .iter() | ||
| .map(|(_, k)| { | ||
| ( | ||
| k.clone(), | ||
| Frame::Array(framevec![bulk_static(b"GET"), bulk(k)]), | ||
| ) | ||
| }) | ||
| .collect(); | ||
| let indices: Vec<usize> = indexed.iter().map(|(i, _)| *i).collect(); | ||
| let msg = ShardMessage::MultiExecute { | ||
| db_index, | ||
| commands, | ||
| reply_tx, | ||
| }; | ||
| spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; | ||
| pending.push((indices, reply_rx)); | ||
| } | ||
| } | ||
| for (indices, reply_rx) in pending { | ||
| match reply_rx.recv().await { | ||
| Ok(frames) => { | ||
| for (idx, frame) in indices.into_iter().zip(frames) { | ||
| match frame { | ||
| Frame::BulkString(v) => sources[idx] = Some(v.to_vec()), | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| _ => sources[idx] = Some(Vec::new()), | ||
| } | ||
| } | ||
| } | ||
| Err(_) => { | ||
| return Frame::Error(Bytes::from_static(b"ERR cross-shard reply channel closed")); | ||
| } | ||
| } | ||
| } | ||
| let gathered: Vec<Vec<u8>> = sources.into_iter().map(Option::unwrap_or_default).collect(); | ||
|
|
||
| match crate::command::string::bitop_compute(&op, &gathered) { | ||
| Err(e) => e, | ||
| Ok(None) => { | ||
| // All sources empty/missing — dest is deleted, reply 0. | ||
| let reply = run_on_owner( | ||
| &dest, | ||
| &[bulk_static(b"DEL"), bulk(&dest)], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| if let Frame::Error(e) = reply { | ||
| return Frame::Error(e); | ||
| } | ||
| Frame::Integer(0) | ||
| } | ||
| Ok(Some(result)) => { | ||
| let len = result.len() as i64; | ||
| let reply = run_on_owner( | ||
| &dest, | ||
| &[ | ||
| bulk_static(b"SET"), | ||
| bulk(&dest), | ||
| Frame::BulkString(Bytes::from(result)), | ||
| ], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| if let Frame::Error(e) = reply { | ||
| return Frame::Error(e); | ||
| } | ||
| Frame::Integer(len) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// Coordinate COPY across shards. | ||
| /// | ||
| /// `COPY src dst [REPLACE]` — same-shard pairs (hash tags) forward to the | ||
| /// owning shard and keep full any-type fidelity via the local copy path. | ||
| /// Cross-shard pairs transfer STRING values exactly (value + TTL, NX unless | ||
| /// REPLACE); cross-shard non-string values return an explicit error instead | ||
| /// of silently corrupting (full-fidelity transfer is DUMP/RESTORE territory, | ||
| /// tracked in the task backlog). `COPY ... DB n` never reaches this path | ||
| /// (excluded in `is_multi_key_command`; the handlers' two-db interception | ||
| /// keeps owning it). | ||
| #[allow(clippy::too_many_arguments)] | ||
| async fn coordinate_copy( | ||
| args: &[Frame], | ||
| my_shard: usize, | ||
| num_shards: usize, | ||
| db_index: usize, | ||
| shard_databases: &Arc<ShardDatabases>, | ||
| dispatch_tx: &Rc<RefCell<Vec<HeapProd<ShardMessage>>>>, | ||
| spsc_notifiers: &[Arc<channel::Notify>], | ||
| cached_clock: &CachedClock, | ||
| _response_pool: &(), | ||
| ) -> Frame { | ||
| // Single-shard server: straight to local dispatch — zero coordinator | ||
| // overhead on the 1-shard hot path. | ||
| if num_shards == 1 { | ||
| return run_local( | ||
| shard_databases, | ||
| my_shard, | ||
| db_index, | ||
| cached_clock, | ||
| b"COPY", | ||
| args, | ||
| ); | ||
| } | ||
| if args.len() < 2 { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR wrong number of arguments for 'COPY' command", | ||
| )); | ||
| } | ||
| let (Some(src), Some(dst)) = (extract_key(&args[0]), extract_key(&args[1])) else { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR wrong number of arguments for 'COPY' command", | ||
| )); | ||
| }; | ||
| let mut replace = false; | ||
| for opt in &args[2..] { | ||
| match extract_key(opt) { | ||
| Some(o) if o.eq_ignore_ascii_case(b"REPLACE") => replace = true, | ||
| _ => return Frame::Error(Bytes::from_static(b"ERR syntax error")), | ||
| } | ||
| } | ||
|
|
||
| let src_shard = key_to_shard(&src, num_shards); | ||
| let dst_shard = key_to_shard(&dst, num_shards); | ||
|
|
||
| // Same owner: forward the whole COPY — full any-type fidelity. | ||
| if src_shard == dst_shard { | ||
| let mut parts: Vec<Frame> = Vec::with_capacity(args.len() + 1); | ||
| parts.push(bulk_static(b"COPY")); | ||
| parts.extend_from_slice(args); | ||
| return run_on_owner( | ||
| &src, | ||
| &parts, | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| } | ||
|
|
||
| // Cross-shard: read value + TTL from src's shard. | ||
| let value = match run_on_owner( | ||
| &src, | ||
| &[bulk_static(b"GET"), bulk(&src)], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await | ||
| { | ||
| Frame::BulkString(v) => v, | ||
| Frame::Error(e) if e.starts_with(b"WRONGTYPE") => { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR COPY across shards supports only string values; co-locate the keys with {hash} tags for other types", | ||
| )); | ||
| } | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| _ => return Frame::Integer(0), // src missing | ||
| }; | ||
| let ttl_ms = match run_on_owner( | ||
| &src, | ||
| &[bulk_static(b"PTTL"), bulk(&src)], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await | ||
| { | ||
| Frame::Integer(t) if t > 0 => Some(t), | ||
| Frame::Integer(-2) => return Frame::Integer(0), // expired between reads | ||
| _ => None, | ||
| }; | ||
|
|
||
| // Write to dst's shard: NX unless REPLACE, then restore TTL. | ||
| let set_parts: Vec<Frame> = if replace { | ||
| vec![bulk_static(b"SET"), bulk(&dst), Frame::BulkString(value)] | ||
| } else { | ||
| vec![ | ||
| bulk_static(b"SET"), | ||
| bulk(&dst), | ||
| Frame::BulkString(value), | ||
| bulk_static(b"NX"), | ||
| ] | ||
| }; | ||
| let set_reply = run_on_owner( | ||
| &dst, | ||
| &set_parts, | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| match set_reply { | ||
| Frame::SimpleString(_) => {} | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| // Null reply = NX refused (dst exists); anything else is unexpected. | ||
| _ => return Frame::Integer(0), | ||
| } | ||
| if let Some(t) = ttl_ms { | ||
| let mut ttl_buf = itoa::Buffer::new(); | ||
| let reply = run_on_owner( | ||
| &dst, | ||
| &[ | ||
| bulk_static(b"PEXPIRE"), | ||
| bulk(&dst), | ||
| Frame::BulkString(Bytes::copy_from_slice(ttl_buf.format(t).as_bytes())), | ||
| ], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| if let Frame::Error(e) = reply { | ||
| return Frame::Error(e); | ||
| } | ||
| } | ||
| Frame::Integer(1) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift
Please split the new coordinators out of coordinator.rs.
This file is already 2617 lines long, and adding BITOP/COPY orchestration here pushes it even further past the repo’s size cap. These helpers need to move into submodules before this becomes harder to reason about or safely change.
As per coding guidelines, "No single .rs file should exceed 1500 lines. Split into submodules if approaching this limit."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/coordinator.rs` around lines 115 - 585, The file is too large;
extract the new coordinator logic into submodules: move coordinate_bitop (and
its helpers bulk, bulk_static) into a new shard submodule (e.g.,
shard::coordinator::bitop) and move coordinate_copy into
shard::coordinator::copy (or a single shard::coordinator::multi_key module),
keeping run_local/run_remote/run_on_owner in the original coordinator.rs and
calling into the new modules. Create mod declarations (mod bitop; mod copy;) and
re-export or use super:: items as needed, update imports so the moved functions
reference run_local, run_on_owner, key_to_shard, extract_key, shard_databases,
dispatch_tx, spsc_notifiers, cached_clock, etc., and set visibility (pub(crate)
or pub) to match callers; run cargo build/tests to fix any borrow/visibility
issues. Ensure no logic changes—only code moved into new files and references
adjusted.
Source: Coding guidelines
Comment on lines
+492
to
+583
| // Cross-shard: read value + TTL from src's shard. | ||
| let value = match run_on_owner( | ||
| &src, | ||
| &[bulk_static(b"GET"), bulk(&src)], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await | ||
| { | ||
| Frame::BulkString(v) => v, | ||
| Frame::Error(e) if e.starts_with(b"WRONGTYPE") => { | ||
| return Frame::Error(Bytes::from_static( | ||
| b"ERR COPY across shards supports only string values; co-locate the keys with {hash} tags for other types", | ||
| )); | ||
| } | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| _ => return Frame::Integer(0), // src missing | ||
| }; | ||
| let ttl_ms = match run_on_owner( | ||
| &src, | ||
| &[bulk_static(b"PTTL"), bulk(&src)], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await | ||
| { | ||
| Frame::Integer(t) if t > 0 => Some(t), | ||
| Frame::Integer(-2) => return Frame::Integer(0), // expired between reads | ||
| _ => None, | ||
| }; | ||
|
|
||
| // Write to dst's shard: NX unless REPLACE, then restore TTL. | ||
| let set_parts: Vec<Frame> = if replace { | ||
| vec![bulk_static(b"SET"), bulk(&dst), Frame::BulkString(value)] | ||
| } else { | ||
| vec![ | ||
| bulk_static(b"SET"), | ||
| bulk(&dst), | ||
| Frame::BulkString(value), | ||
| bulk_static(b"NX"), | ||
| ] | ||
| }; | ||
| let set_reply = run_on_owner( | ||
| &dst, | ||
| &set_parts, | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| match set_reply { | ||
| Frame::SimpleString(_) => {} | ||
| Frame::Error(e) => return Frame::Error(e), | ||
| // Null reply = NX refused (dst exists); anything else is unexpected. | ||
| _ => return Frame::Integer(0), | ||
| } | ||
| if let Some(t) = ttl_ms { | ||
| let mut ttl_buf = itoa::Buffer::new(); | ||
| let reply = run_on_owner( | ||
| &dst, | ||
| &[ | ||
| bulk_static(b"PEXPIRE"), | ||
| bulk(&dst), | ||
| Frame::BulkString(Bytes::copy_from_slice(ttl_buf.format(t).as_bytes())), | ||
| ], | ||
| my_shard, | ||
| num_shards, | ||
| db_index, | ||
| shard_databases, | ||
| dispatch_tx, | ||
| spsc_notifiers, | ||
| cached_clock, | ||
| ) | ||
| .await; | ||
| if let Frame::Error(e) = reply { | ||
| return Frame::Error(e); | ||
| } | ||
| } |
There was a problem hiding this comment.
Cross-shard COPY is not snapshot-safe.
This path reads GET and PTTL in separate owner calls, then writes SET and PEXPIRE in separate destination calls. A concurrent writer can change the source between the two reads, or replace the destination between SET and PEXPIRE, which lets this coordinator copy a mixed snapshot or attach the copied TTL to someone else’s value.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/coordinator.rs` around lines 492 - 583, The current cross-shard
COPY path issues separate run_on_owner calls (run_on_owner + bulk_static("GET"),
run_on_owner + bulk_static("PTTL"), and later separate SET and PEXPIRE calls)
which is not snapshot-safe; instead fetch and restore atomically on the owners:
replace the separate GET/PTTL calls with a single atomic fetch on src (either
run DUMP on src and capture the serialized payload and TTL, or execute a single
EVAL/LUA on the src owner that returns value+ttl together), and replace the
separate SET then PEXPIRE on dst with a single atomic restore on the dst owner
(RESTORE with TTL or a single EVAL that performs SET + PEXPIRE atomically,
honoring replace/NX). Update code paths that build set_parts, the value/ttl_ms
handling, and the run_on_owner invocations so run_on_owner is used to perform a
single src-side call (DUMP or EVAL returning payload+ttl) and a single dst-side
call (RESTORE or EVAL performing atomic write+ttl), preserving replace semantics
and returning errors through Frame the same way.
Comment on lines
434
to
442
| /// Set the expiry from absolute milliseconds. Pass 0 to remove expiry. | ||
| /// `base_ts` is accepted for API consistency but not used (TTL stored as absolute seconds). | ||
| #[inline] | ||
| pub fn set_expires_at_ms(&mut self, _base_ts: u32, ms: u64) { | ||
| if ms == 0 { | ||
| self.ttl_delta = 0; | ||
| self.ttl_secs = 0; | ||
| } else { | ||
| // Store as absolute seconds (u32 is good until year 2106) | ||
| let abs_secs = ms / 1000; | ||
| // Ensure non-zero delta for non-zero input | ||
| self.ttl_delta = (abs_secs.max(1)).min(u32::MAX as u64) as u32; | ||
| // Store as absolute seconds; ensure non-zero for non-zero input. | ||
| self.ttl_secs = (ms / 1000).max(1); | ||
| } |
There was a problem hiding this comment.
Storing expiry as whole seconds breaks millisecond TTL semantics.
set_expires_at_ms() now floors ms to seconds, so a key set to expire at 1500ms is stored as 1000ms. That makes expires_at_ms() report the wrong deadline and is_expired_at() evict keys up to 999ms early. Widening the field to u64 fixes the year-2106 overflow, but the stored unit still needs to remain milliseconds if the surrounding API stays millisecond-based.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/storage/entry.rs` around lines 434 - 442, set_expires_at_ms currently
converts milliseconds to seconds losing sub-second precision; change the stored
field from ttl_secs (u32/u64 seconds) to a millisecond-based field (e.g.,
ttl_ms: u64) and store the exact ms value (0 to mean no expiry) to preserve
semantics and avoid year-2106 overflow. Update set_expires_at_ms to assign ms
directly (keeping the 0-as-no-expiry convention), update expires_at_ms() to
return the u64 millisecond value, and update is_expired_at() to compare in
milliseconds; also adjust any serialization/deserialization, constructors, and
tests that reference ttl_secs so they use ttl_ms. Ensure types and names
(ttl_secs, set_expires_at_ms, expires_at_ms, is_expired_at) are updated
consistently throughout the module.
…ispatch-gaps The gate ran AFTER the main commit, so the committed state.json still said "gate: none" (caught by CodeRabbit on PR #173). This commit lands the engine-synced state (phase: done, gate: PASS) and TASK.md's §6 gate record plus §7 observe deltas — including the PR-review triage: both-path XREAD/ XPENDING parity deltas, the COPY..DB pre-existing interception, second- granularity CompactEntry TTL, the coordinator.rs size-cap follow-up, and the ShardSlice owner-routing handoff to the shardslice-migration task. Refs: PR #173, ADD task consistency-dispatch-gaps. author: Tin Dang
Contributor
Author
CodeRabbit triage (all 13 findings verified against code)Fixed:
Skipped, with verification:
|
This was referenced Jul 4, 2026
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
The consistency suite's first honest end-to-end run at shards=4/12 surfaced four pre-existing cross-shard defect groups — all latent on main, all caused by state being keyed to the connection's shard while SO_REUSEPORT lands connections on arbitrary shards:
args[0](the literalAND) — newcoordinate_bitopgathers sources per owning shard, computes via extracted purebitop_compute, writes dest on its owner.coordinate_copyhandles cross-shard string copies (value + TTL, REPLACE/NX honored); cross-shard non-string COPY returns an explicit error.ShardMessage::GraphCommand/ newGraphRollback; TXN intents captured from routed responses;abort_cross_store_txn_routedpartitions rollback by owner and awaits acks.key_to_shard(effective_key)'s db/registry/trigger-registry/WAL in both runtimes; TXN-commit MQ materialization routes per intent.Harness fixes (assertions added, none weakened): WS isolation probe now pipes AUTH/SET/GET through one redis-cli conn (the old probe's SET ran unbound — it never tested scoping), plus server-lifecycle and scenario-3 fixes documented in the task file.
Evidence
scripts/test-consistency.sh: 197 PASS / 0 FAIL at shards=1, 4, and 12 (exit 0 each) on moon-dev VMtests/cross_shard_consistency_red.rs(cdg6 a–g): 7/7 on macOS and Linux VM (red established on main first)bench-swf.sh, idle VM): no regression in any cell vs the spsc-wake-floor baseline (e.g. s4 P16 SET 2.22M vs 1.47M, GET 3.33M vs 2.13M)Refs: ADD task
consistency-dispatch-gaps, contracts v3 + v4.Summary by CodeRabbit
Bug Fixes & Improvements
Tests
Documentation