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