kevy_rt/types.rs
1//! Public command-classification + live-config types for the [`Commands`]
2//! trait (`ResolvedCmd`, `NotifyClass`, `TxnKind`, `LiveRuntimeConfig`).
3//! Split out of `lib.rs` (500-LOC house rule); all re-exported from the
4//! crate root, so the public paths (`kevy_rt::TxnKind`, …) are unchanged.
5//!
6//! [`Commands`]: crate::Commands
7
8use crate::blocked::BlockHint;
9use crate::route::Route;
10use kevy_config::NotificationFlags;
11use kevy_persist::Fsync;
12
13/// Per-command verb-resolution result. Produced once by [`Commands::resolve`]
14/// in the reactor's parse-then-dispatch loop, reused for routing decisions,
15/// AOF logging, and the QUIT branch — so the per-cmd `upper_verb` cost goes
16/// from 4× down to 1×.
17///
18/// [`Commands::resolve`]: crate::Commands::resolve
19pub struct ResolvedCmd {
20 /// MULTI/EXEC/DISCARD/WATCH classification, so the transaction layer
21 /// does not re-parse the verb.
22 pub txn_kind: TxnKind,
23 /// Where this command goes: one shard, all of them, or a local answer.
24 pub route: Route,
25 /// `QUIT`, which the reactor answers and then closes on rather than
26 /// dispatching.
27 pub is_quit: bool,
28 /// Whether the command mutates — the AOF and replication gate. Set
29 /// from the verb table, not inferred from the route.
30 pub is_write: bool,
31 /// Blocking-command classification (see [`Commands::block_hint`]).
32 /// `BlockHint::None` for every non-blocking verb.
33 ///
34 /// [`Commands::block_hint`]: crate::Commands::block_hint
35 pub block_hint: BlockHint,
36 /// Index into `args` whose write may wake a `BLPOP` / `XREAD BLOCK`
37 /// waiter parked on that key — `Some(1)` for `LPUSH` / `RPUSH` /
38 /// `XADD`, `None` for every other command (including reads). The
39 /// dispatcher's wake hook is gated on both this being `Some` *and*
40 /// the per-shard `BlockedClients` registry being non-empty, so the
41 /// steady-state cost when nobody is parked is one `is_empty()` check.
42 pub wake_idx: Option<u8>,
43}
44
45/// Keyspace-notification event class — what category a write command
46/// belongs to, so the runtime can match it against the per-conn
47/// notify_keyspace_events flags before publishing.
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum NotifyClass {
50 /// `g` — generic key commands (DEL / EXPIRE / PERSIST / RENAME / TYPE).
51 Generic,
52 /// `$` — string commands (SET / GETSET / INCR / APPEND / MSET).
53 String,
54 /// `l` — list commands (LPUSH / RPUSH / LPOP / LREM / LTRIM / …).
55 List,
56 /// `s` — set commands (SADD / SREM / SPOP / …).
57 Set,
58 /// `h` — hash commands (HSET / HDEL / HINCRBY / …).
59 Hash,
60 /// `z` — sorted-set commands (ZADD / ZREM / ZINCRBY / …).
61 Zset,
62 /// `t` — stream commands (XADD / XDEL / XTRIM / XGROUP / XACK /
63 /// XCLAIM / XREADGROUP / …). Matches Redis's `t` class.
64 Stream,
65}
66
67impl NotifyClass {
68 /// Whether `flags` enables this event class.
69 #[inline]
70 pub fn enabled_in(self, flags: &NotificationFlags) -> bool {
71 match self {
72 NotifyClass::Generic => flags.generic,
73 NotifyClass::String => flags.string,
74 NotifyClass::List => flags.list,
75 NotifyClass::Set => flags.set,
76 NotifyClass::Hash => flags.hash,
77 NotifyClass::Zset => flags.zset,
78 NotifyClass::Stream => flags.stream,
79 }
80 }
81}
82
83/// Outcome of an extension fan-out reduce ([`Commands::extension_reduce`]).
84///
85/// [`Commands::extension_reduce`]: crate::Commands::extension_reduce
86#[derive(Debug, PartialEq, Eq)]
87pub enum ExtensionReduced {
88 /// The final RESP reply bytes for the client.
89 Reply(Vec<u8>),
90 /// Not final yet: fan `argv` out to every shard as a follow-up
91 /// extension phase and reduce again when its chunks land. Phase
92 /// state rides inside the argv itself, so the runtime holds no
93 /// per-phase bookkeeping.
94 Continue(Vec<Vec<u8>>),
95}
96
97/// Transaction-control classification for a command.
98pub enum TxnKind {
99 /// `MULTI` — opens a queue on this connection.
100 Multi,
101 /// `EXEC` — runs the queue, or replies nil if a WATCH was broken.
102 Exec,
103 /// `DISCARD` — drops the queue and any WATCH set.
104 Discard,
105 /// `WATCH` — outside MULTI runs the fan-out; inside MULTI is rejected
106 /// with an error (Redis semantics: `WATCH inside MULTI is not allowed`).
107 /// `UNWATCH` is plain [`Self::Other`] — outside MULTI it routes to
108 /// [`Route::Unwatch`] (clear + OK); inside MULTI it queues as a no-op
109 /// that dispatch resolves to +OK at EXEC time.
110 Watch,
111 /// Everything else: queued inside MULTI, dispatched outside it.
112 Other,
113}
114
115/// Live snapshot of the runtime-owned knobs that may have been changed
116/// since this shard's last tick. Built by the [`Commands`] impl from
117/// its own config source (e.g. kevy reads `config_global`). Each
118/// `Some(_)` is applied to the shard; each `None` leaves the existing
119/// setting alone.
120///
121/// One snapshot is built per tick (every 100 ms by default), so its
122/// cost is amortised across thousands of commands.
123///
124/// [`Commands`]: crate::Commands
125#[derive(Debug, Default, Clone, Copy)]
126pub struct LiveRuntimeConfig {
127 /// AOF fsync policy. Applied via `Aof::set_fsync` — switching to
128 /// `Always` mid-flight also flushes any buffered bytes so the new
129 /// "every write is on disk before reply" contract is honoured from
130 /// the next append onward.
131 pub appendfsync: Option<Fsync>,
132 /// `auto_aof_rewrite_percentage`. `0` disables the auto-trigger.
133 pub auto_aof_rewrite_pct: Option<u32>,
134 /// Absolute-size auto-rewrite trigger in bytes (0 = rule off).
135 pub auto_aof_rewrite_bytes: Option<u64>,
136 /// Time-based auto-rewrite trigger in seconds (0 = rule off).
137 pub auto_aof_rewrite_interval_secs: Option<u64>,
138 /// `auto_aof_rewrite_min_size` in bytes.
139 pub auto_aof_rewrite_min_size: Option<u64>,
140 /// New tick interval in ms (`1000/hz`). `0` disables ticking
141 /// entirely — note that disabling also turns off active TTL
142 /// expiry and the auto-rewrite tick path. Lazy expiry on access
143 /// always still works.
144 pub tick_interval_ms: Option<u64>,
145 /// `notify_keyspace_events` flags. Parsed by the [`Commands`]
146 /// impl from its config source (e.g. kevy reads
147 /// `config_global` + [`kevy_config::parse_notification_flags`]).
148 /// Default-empty flags mean OFF — writes pay one bool-OR check
149 /// and skip every per-key keyspace notification publish.
150 ///
151 /// [`Commands`]: crate::Commands
152 pub notify_flags: Option<NotificationFlags>,
153 /// `[slowlog].slower_than_micros` — `-1` disables, `0` records all,
154 /// `>0` is the strict micros threshold. `None` keeps the existing
155 /// shard setting (set by the [`Runtime`] builder at startup).
156 ///
157 /// [`Runtime`]: crate::Runtime
158 pub slowlog_slower_than_micros: Option<i64>,
159 /// `[slowlog].max_len` — ring cap per shard. Shrinking trims the
160 /// oldest entries on the next tick application.
161 pub slowlog_max_len: Option<u32>,
162 /// Monotonic promotion counter. The command layer bumps
163 /// it every time this process is PROMOTED (replica → primary:
164 /// `REPLICAOF NO ONE` on a following replica, or an election win).
165 /// Each shard tracks the last value it saw; an increase makes the
166 /// shard bump its feed generation (offsets restart at 0, persisted
167 /// via the feed-gen sidecar) — so a REPL.TOKEN minted before the
168 /// failover can never falsely satisfy a REPL.WAIT against the new
169 /// primary's unrelated offset space. Not an Option: `0` (the
170 /// default) means "never promoted" and embedders pay nothing.
171 pub promotion_epoch: u64,
172}
173
174/// A replica's acknowledged state, published per shard tick via
175/// [`Commands::on_replication_view`]: the offset from its latest
176/// `REPLCONF ACK` plus that ACK's age at publication time. `None` in
177/// the view tuple means the replica has never ACKed.
178///
179/// [`Commands::on_replication_view`]: crate::Commands::on_replication_view
180#[derive(Debug, Clone, Copy, PartialEq, Eq)]
181pub struct ReplicaAck {
182 /// Offset from the latest `REPLCONF ACK` (`0` is a real heartbeat
183 /// ACK from an empty replica, not a placeholder).
184 pub acked_offset: u64,
185 /// Milliseconds since that ACK was received, measured when the
186 /// view was published. Feeds the `min_replicas_max_lag_ms` gate.
187 pub ack_age_ms: u64,
188}
189
190/// One replica conn's row in the per-tick replication view:
191/// `(replica_id, peer_ipv4, peer_port, sent_offset, ack)`. The id is
192/// the identity string the replica presented at handshake — command
193/// layers group per-shard rows by it to render one aggregate entry
194/// per replica process.
195pub type ReplicaViewRow = (String, std::net::Ipv4Addr, u16, u64, Option<ReplicaAck>);