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 /// `BITOP op dst src [src …]` — N sources gathered, combined, and
42 /// stored at a destination that sits at `args[2]`, not `args[1]`.
43 /// `ZAlgebraStore` is the same shape with a different payload: it
44 /// combines set and zset members, not raw bytes.
45 ///
46 /// Carries nothing. An earlier draft carried the operator so the
47 /// router could pick it, which meant parsing the operator twice and
48 /// needing a fallback route for the argv the router could not parse
49 /// — and that fallback led to a dispatch table with no BITOP arm,
50 /// so a malformed BITOP would have been answered "unknown command".
51 /// The route says only that this is a BITOP; every refusal is
52 /// worded once, in `exec_bitop`.
53 ///
54 /// Why it cannot ride `Single(1)`, in one assertion:
55 ///
56 /// ```
57 /// use kevy_rt::{Route, shard_of_key};
58 /// // `Single(1)` hashes args[1]. For BITOP that is the OPERATOR.
59 /// let operator = b"AND".as_slice();
60 /// let destination = b"dst".as_slice();
61 /// assert_ne!(shard_of_key(operator, 8, false), shard_of_key(destination, 8, false));
62 /// assert!(matches!(Route::BitOpStore, Route::BitOpStore));
63 /// ```
64 BitOpStore,
65 /// `COPY src dst [REPLACE]` — two keys, so the same hazard the
66 /// rename and list-move routes exist for: left to the catch-all
67 /// `Single(1)` the copy lands in the SOURCE's shard, where no later
68 /// read of the destination will ever look. Same-shard pairs take
69 /// one atomic op; cross-shard pairs run Read → Put, and need no
70 /// rollback because the read does not remove anything.
71 ///
72 /// Why it cannot ride `Single(1)`, in one assertion:
73 ///
74 /// ```
75 /// use kevy_rt::{Route, shard_of_key};
76 /// // A pair of ordinary key names on an eight-shard server.
77 /// let (src, dst) = (b"ca".as_slice(), b"cb".as_slice());
78 /// assert_ne!(shard_of_key(src, 8, false), shard_of_key(dst, 8, false));
79 /// // `Single(1)` hashes args[1] — the SOURCE — and runs the whole
80 /// // command there, so the copy would land in a shard no later read
81 /// // of `dst` ever looks at, while the reply said it worked.
82 /// assert!(matches!(Route::Copy, Route::Copy));
83 /// ```
84 Copy,
85 /// Geo `*STORE` family — `GEOSEARCHSTORE dst src …` and
86 /// `GEORADIUS[BYMEMBER] src … STORE|STOREDIST dst`.
87 ///
88 /// These MUST be routed, not left to the catch-all `Route::Single(1)`:
89 /// GEOSEARCHSTORE puts the DESTINATION at argv[1] (so the search then
90 /// read the source off the wrong shard — `:0`, or "could not decode
91 /// requested zset member" for FROMMEMBER) while GEORADIUS puts the
92 /// SOURCE there (so the destination was written into the source's
93 /// shard, invisible to every later read of it). Both keys are carried
94 /// here because neither sits at a fixed argv index — the legacy forms
95 /// hide `dst` behind an option-soup scan.
96 ///
97 /// The search runs on `src`'s shard ([`crate::Commands::geo_search`]),
98 /// the write lands on `dst`'s (`Op::ZStoreResult`) — see
99 /// [`crate::exec_geostore`].
100 GeoStore {
101 /// Key the search reads — its shard runs the query.
102 src: Vec<u8>,
103 /// Key the result is written to — its shard takes the write, which
104 /// is why both keys have to be extracted before routing.
105 dst: Vec<u8>,
106 },
107 /// `FEED.READ <shard> <gen> <offset> …` — shard-index routed.
108 FeedRead,
109 /// `FEED.TAIL <shard>`.
110 FeedTail,
111 /// `FEED.SHARDS` — answered locally.
112 FeedShards,
113 /// `PREFIX.STATS <prefix>` — all-shard fanout, summed.
114 PrefixStats,
115 /// `CLIENT LIST` — all-shard fanout; each shard renders its conn
116 /// table rows, the origin concatenates into one bulk reply.
117 ClientList,
118 /// `CLIENT KILL …` — all-shard fanout; each shard closes its
119 /// matching conns, the origin sums (or maps the legacy positional
120 /// form to `+OK` / `-ERR`).
121 ClientKill,
122 /// Extension fan-out (IDX.* reads): every shard runs
123 /// `Commands::extension_op`, the origin reduces.
124 Extension,
125 /// `WAIT numreplicas timeout` — all-shard barrier: each
126 /// shard answers (possibly deferred until its replicas ACK or the
127 /// deadline) with how many of its replicas acked its
128 /// `master_repl_offset` at arm time; the origin replies the MIN.
129 /// `timeout_ms == 0` = the Redis "wait forever" form (the runtime
130 /// hard-caps it — see `exec_replwait::WAIT_HARD_CAP_MS`).
131 ReplWait {
132 /// How many replicas the caller wants acked. Reported per shard;
133 /// the origin answers the minimum across them.
134 numreplicas: u32,
135 /// Deadline in milliseconds. `0` is Redis's wait-forever form and
136 /// is hard-capped by the runtime rather than honoured literally.
137 timeout_ms: u64,
138 },
139 /// `REPL.TOKEN` on a primary — gather every shard's
140 /// `(feed generation, next_offset)` pair into one flat array.
141 ReplToken,
142 /// `REPL.WAIT` on a replica — all-shard applied barrier:
143 /// shard `i` answers once its replication-apply position reaches
144 /// `offsets[i]` (or the deadline passes). All met → `+OK`; any
145 /// timeout → the pre-built `miss` reply (kevy sends
146 /// `-MISDIRECTED writer is <primary>`). The command layer builds
147 /// `miss` because the upstream address is its knowledge, not the
148 /// runtime's.
149 ReplBarrier {
150 /// One target apply-position per shard, indexed by shard number.
151 offsets: Vec<u64>,
152 /// Deadline in milliseconds for every shard to reach its target.
153 timeout_ms: u64,
154 /// The reply to send if any shard misses its deadline, pre-built by
155 /// the command layer because it names the upstream primary — the
156 /// runtime does not know that address.
157 miss: Vec<u8>,
158 },
159 /// `KEYS pattern` — every shard returns its matching keys.
160 Keys(Option<Vec<u8>>),
161 /// `SCAN cursor [MATCH pattern] [COUNT count] [TYPE type]` — a real
162 /// cursor iterator: each call visits ~COUNT buckets of ONE shard
163 /// (chaining into the next shard only while the work budget lasts)
164 /// and replies `[next-cursor, keys]`. `Err` carries the pre-parsed
165 /// error message the command layer wants on the wire (invalid
166 /// cursor / syntax error) — the runtime replies it verbatim.
167 Scan(Result<ScanArgs, &'static str>),
168 /// `RANDOMKEY` — one arbitrary key across all shards.
169 RandomKey,
170 /// `SUBSCRIBE` / `UNSUBSCRIBE` — connection-level (modifies this conn).
171 Subscribe,
172 /// The other half of the pair above: drops this conn's channel
173 /// subscriptions, all of them when no channel is named.
174 Unsubscribe,
175 /// `PSUBSCRIBE pattern [pattern ...]` / `PUNSUBSCRIBE [pattern ...]` —
176 /// like Subscribe/Unsubscribe but the conn registers Redis-glob
177 /// patterns; `PUBLISH` to a matching channel delivers a `pmessage`
178 /// frame. Connection-level (modifies this conn + shared pattern
179 /// registry).
180 Psubscribe,
181 /// The other half of the pattern pair: drops this conn's pattern
182 /// subscriptions, all of them when no pattern is named, and removes
183 /// them from the shared registry.
184 Punsubscribe,
185 /// `PUBLISH channel message` — delivered to subscribers on every core.
186 Publish,
187 /// `WATCH key [key ...]` — fan-out to record per-shard versions, then
188 /// stash the (key, version) pairs in the conn's `watched` set so the
189 /// next `EXEC` can validate them. Connection-level.
190 Watch,
191 /// `UNWATCH` — clear the conn's `watched` set. Connection-level, local.
192 Unwatch,
193 /// `HELLO [protover [AUTH user pass] [SETNAME name]]` — server
194 /// handshake; on `HELLO 3` flips the conn into RESP3 mode (per-conn
195 /// `proto` field). Reply shape itself is proto-aware (V2: array of
196 /// pairs; V3: Map). Connection-level, dispatch via the
197 /// [`crate::Commands::hello_reply`] hook so embedders set their own server
198 /// metadata.
199 Hello,
200 /// `RENAME source destination` / `RENAMENX source destination`. The
201 /// runtime handles the two-shard decision: same-shard renames go
202 /// through one atomic [`crate::Store::rename`] on the owning shard; cross-
203 /// shard renames use the Take→Put orchestrator (lands in v2-3b;
204 /// v2-3a emits `-CROSSSHARD ...` for that case).
205 Rename {
206 /// `true` for `RENAMENX` (no overwrite — reply `:0` if dst exists).
207 nx: bool,
208 },
209 /// `RPOPLPUSH src dst` / `LMOVE src dst LEFT|RIGHT LEFT|RIGHT` /
210 /// `BRPOPLPUSH src dst timeout`, once the blocking form has an element
211 /// to serve.
212 ///
213 /// These MUST be routed, not left to `Route::Single(1)`. The source and
214 /// the destination are different keys and can live on different shards;
215 /// the catch-all route hashes args[1] (the source), so the destination
216 /// push executed on the SOURCE's shard and the element was written into
217 /// a keyspace nobody would ever read it from. It returned the moved
218 /// value, so the caller believed it had worked. Measured on an 8-shard
219 /// server: 11 of 12 moves silently lost the element.
220 ///
221 /// Same-shard pairs are one atomic Op on the owning shard. Cross-shard
222 /// pairs run the Take→Push orchestrator (mirroring [`Self::Rename`]),
223 /// which is NOT atomic — see `exec_listmove`.
224 ListMove {
225 /// Pop from the head of the source (`LMOVE ... LEFT ...`) rather
226 /// than the tail (`RPOPLPUSH`).
227 from_left: bool,
228 /// Push onto the head of the destination (`RPOPLPUSH`, `LMOVE ...
229 /// LEFT`) rather than the tail.
230 to_left: bool,
231 },
232 /// `SLOWLOG GET / LEN / RESET / HELP`. The sub-command + parsed
233 /// args are pre-decoded at routing time so the runtime knows
234 /// whether to short-circuit (HELP / error) or fan out across
235 /// shards (GET / LEN / RESET). See [`crate::parse_slowlog_sub`].
236 Slowlog(SlowlogSub),
237 /// Non-blocking `XREAD` / `XREADGROUP` over **multiple** streams — fan
238 /// each stream out to its owning shard and merge the per-stream replies
239 /// in request order (single-stream forms still route via
240 /// [`Self::Single`]). Each element is `(stream key, last-seen id)`;
241 /// `count` is the optional `COUNT` cap applied per stream; `group`
242 /// `Some` makes each per-shard sub-query an `XREADGROUP` (a write —
243 /// PEL / last-delivered updates happen on each stream's owning shard
244 /// and are AOF-logged there as the rewritten single-stream command).
245 /// The command set builds this only for the non-blocking, ≥2-stream
246 /// forms; blocking reads park on the origin shard instead (see the
247 /// cross-shard BLOCK arbiter).
248 XReadGather {
249 /// `(stream key, start id)` per stream, already paired — the wire
250 /// form lists all keys and then all ids, which is not routable.
251 streams: Vec<(Vec<u8>, Vec<u8>)>,
252 /// `COUNT`, applied per stream rather than across the gather.
253 count: Option<usize>,
254 /// `Some` turns each per-shard sub-query into an XREADGROUP, which
255 /// makes it a write: the PEL update happens on the stream's own
256 /// shard and is logged there.
257 group: Option<XGroupCtx>,
258 },
259}
260
261/// Parsed `SCAN` arguments carried by [`Route::Scan`].
262///
263/// `cursor` is the raw wire cursor: the runtime splits it into
264/// `(shard, in-shard position)` — shard index in the top 10 bits,
265/// reverse-binary bucket cursor in the low 54 (see `exec_scan` for the
266/// documented limits). Cursors are therefore only meaningful on the
267/// server (and shard count) that issued them, like Redis Cluster
268/// cursors are per-node.
269#[derive(Debug, Clone, PartialEq, Eq)]
270pub struct ScanArgs {
271 /// Raw wire cursor (`0` starts a sweep).
272 pub cursor: u64,
273 /// `COUNT` — buckets-visited work bound per call (default 10).
274 pub count: usize,
275 /// `MATCH` glob, applied to each visited key.
276 pub pattern: Option<Vec<u8>>,
277 /// `TYPE` — keep only keys whose value type name matches
278 /// (case-insensitive; unknown names match nothing).
279 pub type_filter: Option<Vec<u8>>,
280}
281
282/// The `GROUP <name> <consumer>` (+ `NOACK`) context an `XREADGROUP`
283/// gather carries to each per-stream sub-query.
284#[derive(Debug, PartialEq)]
285pub struct XGroupCtx {
286 /// Consumer-group name.
287 pub group: Vec<u8>,
288 /// Consumer name within the group.
289 pub consumer: Vec<u8>,
290 /// `NOACK` — deliver without adding to the PEL.
291 pub noack: bool,
292}