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)]
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    /// Synchronous in v1.0 (each shard blocks for its own rewrite duration).
30    RewriteAof,
31    /// `MSET` — `args[1..]` are key/value pairs, routed per key's shard.
32    MSet,
33    /// `MGET` — `args[1..]` are keys; values gathered in request order.
34    MGet,
35    /// `SINTER` / `SUNION` / `SDIFF` — `args[1..]` are set keys.
36    SInter,
37    SUnion,
38    SDiff,
39    /// v2.2 zset/set algebra `*STORE` family: gather sources, combine
40    /// per [`crate::message::ZCombine`], materialize at `args[1]`.
41    ZAlgebraStore(crate::ZCombine),
42    /// `ZINTERCARD numkeys key… [LIMIT n]` — read-only gathered count.
43    ZInterCard,
44    /// v2.3 `FEED.READ <shard> <gen> <offset> …` — shard-index routed.
45    FeedRead,
46    /// v2.3 `FEED.TAIL <shard>`.
47    FeedTail,
48    /// v2.3 `FEED.SHARDS` — answered locally.
49    FeedShards,
50    /// v2.3 `PREFIX.STATS <prefix>` — all-shard fanout, summed.
51    PrefixStats,
52    /// v2.5 extension fan-out (IDX.* reads): every shard runs
53    /// `Commands::extension_op`, the origin reduces.
54    Extension,
55    /// v3.16 D1 `WAIT numreplicas timeout` — all-shard barrier: each
56    /// shard answers (possibly deferred until its replicas ACK or the
57    /// deadline) with how many of its replicas acked its
58    /// `master_repl_offset` at arm time; the origin replies the MIN.
59    /// `timeout_ms == 0` = the Redis "wait forever" form (the runtime
60    /// hard-caps it — see `exec_replwait::WAIT_HARD_CAP_MS`).
61    ReplWait { numreplicas: u32, timeout_ms: u64 },
62    /// v3.16 D2 `REPL.TOKEN` on a primary — gather every shard's
63    /// `(feed generation, next_offset)` pair into one flat array.
64    ReplToken,
65    /// v3.16 D2 `REPL.WAIT` on a replica — all-shard applied barrier:
66    /// shard `i` answers once its replication-apply position reaches
67    /// `offsets[i]` (or the deadline passes). All met → `+OK`; any
68    /// timeout → the pre-built `miss` reply (kevy sends
69    /// `-MISDIRECTED writer is <primary>`). The command layer builds
70    /// `miss` because the upstream address is its knowledge, not the
71    /// runtime's.
72    ReplBarrier {
73        offsets: Vec<u64>,
74        timeout_ms: u64,
75        miss: Vec<u8>,
76    },
77    /// `KEYS pattern` — every shard returns its matching keys.
78    Keys(Option<Vec<u8>>),
79    /// `SCAN` (cursor-0 approximation) — like KEYS but replies `[cursor, keys]`.
80    Scan(Option<Vec<u8>>),
81    /// `RANDOMKEY` — one arbitrary key across all shards.
82    RandomKey,
83    /// `SUBSCRIBE` / `UNSUBSCRIBE` — connection-level (modifies this conn).
84    Subscribe,
85    Unsubscribe,
86    /// `PSUBSCRIBE pattern [pattern ...]` / `PUNSUBSCRIBE [pattern ...]` —
87    /// like Subscribe/Unsubscribe but the conn registers Redis-glob
88    /// patterns; `PUBLISH` to a matching channel delivers a `pmessage`
89    /// frame. Connection-level (modifies this conn + shared pattern
90    /// registry).
91    Psubscribe,
92    Punsubscribe,
93    /// `PUBLISH channel message` — delivered to subscribers on every core.
94    Publish,
95    /// `WATCH key [key ...]` — fan-out to record per-shard versions, then
96    /// stash the (key, version) pairs in the conn's `watched` set so the
97    /// next `EXEC` can validate them. Connection-level.
98    Watch,
99    /// `UNWATCH` — clear the conn's `watched` set. Connection-level, local.
100    Unwatch,
101    /// `HELLO [protover [AUTH user pass] [SETNAME name]]` — server
102    /// handshake; on `HELLO 3` flips the conn into RESP3 mode (per-conn
103    /// `proto` field). Reply shape itself is proto-aware (V2: array of
104    /// pairs; V3: Map). Connection-level, dispatch via the
105    /// [`crate::Commands::hello_reply`] hook so embedders set their own server
106    /// metadata.
107    Hello,
108    /// `RENAME source destination` / `RENAMENX source destination`. The
109    /// runtime handles the two-shard decision: same-shard renames go
110    /// through one atomic [`crate::Store::rename`] on the owning shard; cross-
111    /// shard renames use the Take→Put orchestrator (lands in v2-3b;
112    /// v2-3a emits `-CROSSSHARD ...` for that case).
113    Rename {
114        /// `true` for `RENAMENX` (no overwrite — reply `:0` if dst exists).
115        nx: bool,
116    },
117    /// `SLOWLOG GET / LEN / RESET / HELP`. The sub-command + parsed
118    /// args are pre-decoded at routing time so the runtime knows
119    /// whether to short-circuit (HELP / error) or fan out across
120    /// shards (GET / LEN / RESET). See [`crate::parse_slowlog_sub`].
121    Slowlog(SlowlogSub),
122    /// Non-blocking `XREAD` / `XREADGROUP` over **multiple** streams — fan
123    /// each stream out to its owning shard and merge the per-stream replies
124    /// in request order (single-stream forms still route via
125    /// [`Self::Single`]). Each element is `(stream key, last-seen id)`;
126    /// `count` is the optional `COUNT` cap applied per stream; `group`
127    /// `Some` makes each per-shard sub-query an `XREADGROUP` (a write —
128    /// PEL / last-delivered updates happen on each stream's owning shard
129    /// and are AOF-logged there as the rewritten single-stream command).
130    /// The command set builds this only for the non-blocking, ≥2-stream
131    /// forms; blocking reads park on the origin shard instead (see the
132    /// cross-shard BLOCK arbiter).
133    XReadGather {
134        streams: Vec<(Vec<u8>, Vec<u8>)>,
135        count: Option<usize>,
136        group: Option<XGroupCtx>,
137    },
138}
139
140/// The `GROUP <name> <consumer>` (+ `NOACK`) context an `XREADGROUP`
141/// gather carries to each per-stream sub-query.
142#[derive(Debug)]
143pub struct XGroupCtx {
144    /// Consumer-group name.
145    pub group: Vec<u8>,
146    /// Consumer name within the group.
147    pub consumer: Vec<u8>,
148    /// `NOACK` — deliver without adding to the PEL.
149    pub noack: bool,
150}
kevy_rt/route.rs

kevy_rt/
route.rs
