kevy_rt/route.rs
1//! [`Route`] — how each command maps onto shards. Returned by
2//! [`crate::Commands::route`] / carried in [`crate::ResolvedCmd`]; the
3//! runtime's `start_command` matches on it to pick a dispatch shape.
4
5use crate::exec_slowlog::SlowlogSub;
6
7/// How a command maps onto shards.
8#[derive(Debug, PartialEq)]
9pub enum Route {
10 /// Keyless; execute on the connection's own shard (e.g. PING).
11 Local,
12 /// Single-key; route by `args[idx]`.
13 Single(usize),
14 /// `args[1..]` are keys; delete each on its shard, sum the counts.
15 DelKeys,
16 /// `args[1..]` are keys; count existing across shards.
17 ExistsKeys,
18 /// Sum every shard's key count.
19 Dbsize,
20 /// Flush every shard.
21 Flush,
22 /// Snapshot every shard's store to disk, synchronously (`SAVE` —
23 /// blocks until durable, the Redis contract for the explicit form).
24 Save,
25 /// `BGSAVE` — collect a COW view per shard and persist in the
26 /// background; the command returns once the views are frozen.
27 BgSave,
28 /// `BGREWRITEAOF` — rebuild every shard's AOF from in-memory state.
29 /// Each shard freezes a COW view and hands the dump to its persist
30 /// worker, so the reply returns before the rewrite is durable.
31 RewriteAof,
32 /// `MSET` — `args[1..]` are key/value pairs, routed per key's shard.
33 MSet,
34 /// Cross-shard multi-key gather (`MGET` / `SINTER` / `SUNION` /
35 /// `SDIFF` / `ZINTERCARD`): each key's payload is fetched on its
36 /// owning shard and the origin reduces them per [`crate::MultiOp`].
37 Gather(crate::MultiOp),
38 /// zset/set algebra `*STORE` family: gather sources, combine
39 /// per [`crate::message::ZCombine`], materialize at `args[1]`.
40 ZAlgebraStore(crate::ZCombine),
41 /// Geo `*STORE` family — `GEOSEARCHSTORE dst src …` and
42 /// `GEORADIUS[BYMEMBER] src … STORE|STOREDIST dst`.
43 ///
44 /// These MUST be routed, not left to the catch-all `Route::Single(1)`:
45 /// GEOSEARCHSTORE puts the DESTINATION at argv[1] (so the search then
46 /// read the source off the wrong shard — `:0`, or "could not decode
47 /// requested zset member" for FROMMEMBER) while GEORADIUS puts the
48 /// SOURCE there (so the destination was written into the source's
49 /// shard, invisible to every later read of it). Both keys are carried
50 /// here because neither sits at a fixed argv index — the legacy forms
51 /// hide `dst` behind an option-soup scan.
52 ///
53 /// The search runs on `src`'s shard ([`crate::Commands::geo_search`]),
54 /// the write lands on `dst`'s (`Op::ZStoreResult`) — see
55 /// [`crate::exec_geostore`].
56 GeoStore { src: Vec<u8>, dst: Vec<u8> },
57 /// `FEED.READ <shard> <gen> <offset> …` — shard-index routed.
58 FeedRead,
59 /// `FEED.TAIL <shard>`.
60 FeedTail,
61 /// `FEED.SHARDS` — answered locally.
62 FeedShards,
63 /// `PREFIX.STATS <prefix>` — all-shard fanout, summed.
64 PrefixStats,
65 /// `CLIENT LIST` — all-shard fanout; each shard renders its conn
66 /// table rows, the origin concatenates into one bulk reply.
67 ClientList,
68 /// `CLIENT KILL …` — all-shard fanout; each shard closes its
69 /// matching conns, the origin sums (or maps the legacy positional
70 /// form to `+OK` / `-ERR`).
71 ClientKill,
72 /// Extension fan-out (IDX.* reads): every shard runs
73 /// `Commands::extension_op`, the origin reduces.
74 Extension,
75 /// `WAIT numreplicas timeout` — all-shard barrier: each
76 /// shard answers (possibly deferred until its replicas ACK or the
77 /// deadline) with how many of its replicas acked its
78 /// `master_repl_offset` at arm time; the origin replies the MIN.
79 /// `timeout_ms == 0` = the Redis "wait forever" form (the runtime
80 /// hard-caps it — see `exec_replwait::WAIT_HARD_CAP_MS`).
81 ReplWait { numreplicas: u32, timeout_ms: u64 },
82 /// `REPL.TOKEN` on a primary — gather every shard's
83 /// `(feed generation, next_offset)` pair into one flat array.
84 ReplToken,
85 /// `REPL.WAIT` on a replica — all-shard applied barrier:
86 /// shard `i` answers once its replication-apply position reaches
87 /// `offsets[i]` (or the deadline passes). All met → `+OK`; any
88 /// timeout → the pre-built `miss` reply (kevy sends
89 /// `-MISDIRECTED writer is <primary>`). The command layer builds
90 /// `miss` because the upstream address is its knowledge, not the
91 /// runtime's.
92 ReplBarrier {
93 offsets: Vec<u64>,
94 timeout_ms: u64,
95 miss: Vec<u8>,
96 },
97 /// `KEYS pattern` — every shard returns its matching keys.
98 Keys(Option<Vec<u8>>),
99 /// `SCAN cursor [MATCH pattern] [COUNT count] [TYPE type]` — a real
100 /// cursor iterator: each call visits ~COUNT buckets of ONE shard
101 /// (chaining into the next shard only while the work budget lasts)
102 /// and replies `[next-cursor, keys]`. `Err` carries the pre-parsed
103 /// error message the command layer wants on the wire (invalid
104 /// cursor / syntax error) — the runtime replies it verbatim.
105 Scan(Result<ScanArgs, &'static str>),
106 /// `RANDOMKEY` — one arbitrary key across all shards.
107 RandomKey,
108 /// `SUBSCRIBE` / `UNSUBSCRIBE` — connection-level (modifies this conn).
109 Subscribe,
110 Unsubscribe,
111 /// `PSUBSCRIBE pattern [pattern ...]` / `PUNSUBSCRIBE [pattern ...]` —
112 /// like Subscribe/Unsubscribe but the conn registers Redis-glob
113 /// patterns; `PUBLISH` to a matching channel delivers a `pmessage`
114 /// frame. Connection-level (modifies this conn + shared pattern
115 /// registry).
116 Psubscribe,
117 Punsubscribe,
118 /// `PUBLISH channel message` — delivered to subscribers on every core.
119 Publish,
120 /// `WATCH key [key ...]` — fan-out to record per-shard versions, then
121 /// stash the (key, version) pairs in the conn's `watched` set so the
122 /// next `EXEC` can validate them. Connection-level.
123 Watch,
124 /// `UNWATCH` — clear the conn's `watched` set. Connection-level, local.
125 Unwatch,
126 /// `HELLO [protover [AUTH user pass] [SETNAME name]]` — server
127 /// handshake; on `HELLO 3` flips the conn into RESP3 mode (per-conn
128 /// `proto` field). Reply shape itself is proto-aware (V2: array of
129 /// pairs; V3: Map). Connection-level, dispatch via the
130 /// [`crate::Commands::hello_reply`] hook so embedders set their own server
131 /// metadata.
132 Hello,
133 /// `RENAME source destination` / `RENAMENX source destination`. The
134 /// runtime handles the two-shard decision: same-shard renames go
135 /// through one atomic [`crate::Store::rename`] on the owning shard; cross-
136 /// shard renames use the Take→Put orchestrator (lands in v2-3b;
137 /// v2-3a emits `-CROSSSHARD ...` for that case).
138 Rename {
139 /// `true` for `RENAMENX` (no overwrite — reply `:0` if dst exists).
140 nx: bool,
141 },
142 /// `RPOPLPUSH src dst` / `LMOVE src dst LEFT|RIGHT LEFT|RIGHT` /
143 /// `BRPOPLPUSH src dst timeout`, once the blocking form has an element
144 /// to serve.
145 ///
146 /// These MUST be routed, not left to `Route::Single(1)`. The source and
147 /// the destination are different keys and can live on different shards;
148 /// the catch-all route hashes args[1] (the source), so the destination
149 /// push executed on the SOURCE's shard and the element was written into
150 /// a keyspace nobody would ever read it from. It returned the moved
151 /// value, so the caller believed it had worked. Measured on an 8-shard
152 /// server: 11 of 12 moves silently lost the element.
153 ///
154 /// Same-shard pairs are one atomic Op on the owning shard. Cross-shard
155 /// pairs run the Take→Push orchestrator (mirroring [`Self::Rename`]),
156 /// which is NOT atomic — see `exec_listmove`.
157 ListMove {
158 /// Pop from the head of the source (`LMOVE ... LEFT ...`) rather
159 /// than the tail (`RPOPLPUSH`).
160 from_left: bool,
161 /// Push onto the head of the destination (`RPOPLPUSH`, `LMOVE ...
162 /// LEFT`) rather than the tail.
163 to_left: bool,
164 },
165 /// `SLOWLOG GET / LEN / RESET / HELP`. The sub-command + parsed
166 /// args are pre-decoded at routing time so the runtime knows
167 /// whether to short-circuit (HELP / error) or fan out across
168 /// shards (GET / LEN / RESET). See [`crate::parse_slowlog_sub`].
169 Slowlog(SlowlogSub),
170 /// Non-blocking `XREAD` / `XREADGROUP` over **multiple** streams — fan
171 /// each stream out to its owning shard and merge the per-stream replies
172 /// in request order (single-stream forms still route via
173 /// [`Self::Single`]). Each element is `(stream key, last-seen id)`;
174 /// `count` is the optional `COUNT` cap applied per stream; `group`
175 /// `Some` makes each per-shard sub-query an `XREADGROUP` (a write —
176 /// PEL / last-delivered updates happen on each stream's owning shard
177 /// and are AOF-logged there as the rewritten single-stream command).
178 /// The command set builds this only for the non-blocking, ≥2-stream
179 /// forms; blocking reads park on the origin shard instead (see the
180 /// cross-shard BLOCK arbiter).
181 XReadGather {
182 streams: Vec<(Vec<u8>, Vec<u8>)>,
183 count: Option<usize>,
184 group: Option<XGroupCtx>,
185 },
186}
187
188/// Parsed `SCAN` arguments carried by [`Route::Scan`].
189///
190/// `cursor` is the raw wire cursor: the runtime splits it into
191/// `(shard, in-shard position)` — shard index in the top 10 bits,
192/// reverse-binary bucket cursor in the low 54 (see `exec_scan` for the
193/// documented limits). Cursors are therefore only meaningful on the
194/// server (and shard count) that issued them, like Redis Cluster
195/// cursors are per-node.
196#[derive(Debug, Clone, PartialEq, Eq)]
197pub struct ScanArgs {
198 /// Raw wire cursor (`0` starts a sweep).
199 pub cursor: u64,
200 /// `COUNT` — buckets-visited work bound per call (default 10).
201 pub count: usize,
202 /// `MATCH` glob, applied to each visited key.
203 pub pattern: Option<Vec<u8>>,
204 /// `TYPE` — keep only keys whose value type name matches
205 /// (case-insensitive; unknown names match nothing).
206 pub type_filter: Option<Vec<u8>>,
207}
208
209/// The `GROUP <name> <consumer>` (+ `NOACK`) context an `XREADGROUP`
210/// gather carries to each per-stream sub-query.
211#[derive(Debug, PartialEq)]
212pub struct XGroupCtx {
213 /// Consumer-group name.
214 pub group: Vec<u8>,
215 /// Consumer name within the group.
216 pub consumer: Vec<u8>,
217 /// `NOACK` — deliver without adding to the PEL.
218 pub noack: bool,
219}