kevy_rt/runtime_builders.rs
1//! Runtime builder methods split out of [`crate::runtime`] so that
2//! file stays under the 500-LOC project ceiling. Same `impl Runtime<C>`,
3//! split purely by responsibility: construction + boot live in
4//! `runtime.rs`; the `with_*` configuration setters live here.
5
6use std::path::PathBuf;
7
8use kevy_persist::Fsync;
9
10use crate::Commands;
11use crate::runtime::Runtime;
12
13impl<C: Commands> Runtime<C> {
14 /// v3-cluster replication producer side: when `enabled`, each shard
15 /// runs a per-shard `ReplicationSource` with `buffer_size` byte
16 /// budget. Every applied mutation is pushed to the backlog for
17 /// connected replicas to consume. `enabled = false` (default) is
18 /// zero hot-path cost — each write checks `Option::is_some()` and
19 /// skips. The replication TCP listener / streaming loop are gated
20 /// separately by [`Self::with_replication_listener`]; enabling the
21 /// producer without a listener means the backlog fills and frames
22 /// are dropped per the source's eviction policy, but writes
23 /// proceed normally.
24 #[must_use]
25 pub fn with_replication(mut self, enabled: bool, buffer_size: u64) -> Self {
26 self.enable_replication = enabled;
27 if buffer_size > 0 {
28 self.replication_buffer_size = buffer_size;
29 }
30 self
31 }
32
33 /// Enable the FEED.* consumer surface. Keeps a per-shard
34 /// backlog (even with no replicas) and persists the (generation,
35 /// offset) cursor via the feed sidecars. `buffer_size` = 0 keeps
36 /// the default (64 MB/shard); effective budget is
37 /// `max(replication_buffer_size, feed_buffer_size)` when both
38 /// features are on.
39 #[must_use]
40 pub fn with_feed(mut self, enabled: bool, buffer_size: u64) -> Self {
41 self.feed_enabled = enabled;
42 if buffer_size > 0 {
43 self.feed_buffer_size = buffer_size;
44 }
45 self
46 }
47
48 /// Bring up a replication listener per shard at
49 /// `port_base + shard_id` (per Issue Ledger I2 — mirrors the
50 /// cluster listener pattern). Replica clients connect to each
51 /// per-shard port to mirror the full keyspace. This is independent
52 /// of [`Self::with_replication`]: a primary that runs the producer
53 /// backlog without a listener (benchmarks, embed-only) is
54 /// supported.
55 #[must_use]
56 pub fn with_replication_listener(mut self, port_base: u16) -> Self {
57 self.replication_port_base = Some(port_base);
58 self
59 }
60
61 /// Per-shard SlotTable reconnect window in milliseconds — the
62 /// grace period a disconnected replica's slot is retained for so
63 /// a reconnect within the window can be correlated against its
64 /// prior `sent_offset`. Default `60_000` (60 s); pass `0` to drop
65 /// slots immediately on disconnect.
66 #[must_use]
67 pub fn with_replication_reconnect_window(mut self, ms: u32) -> Self {
68 self.replication_reconnect_window_ms = ms;
69 self
70 }
71
72 /// Install per-shard replica inboxes. The embedder pre-
73 /// constructs `nshards` inbox pairs via
74 /// [`crate::replica_inbox_pair`], keeps the senders to hand to
75 /// the per-shard replica runner threads, and passes the receivers
76 /// here. The order of `receivers` is shard-major: index `i` ↔
77 /// shard `i`. Length must equal `nshards`. When this builder
78 /// isn't called, no shard has an inbox (standalone /
79 /// primary-only behaviour).
80 #[must_use]
81 pub fn with_replica_inboxes(
82 mut self,
83 receivers: Vec<crate::replica_inbox::ReplicaInboxReceiver>,
84 ) -> Self {
85 self.replica_inboxes = receivers.into_iter().map(Some).collect();
86 self
87 }
88
89 /// Enable single-node cluster mode: keys route by Redis-cluster slot
90 /// (CRC16 `{hashtag}` & 16383, contiguous even ranges) and every shard
91 /// `i` binds a second, deterministic listener at `port_base + i` that
92 /// answers wrong-shard keys with `-MOVED` instead of forwarding. The
93 /// SO_REUSEPORT listener on the main port keeps today's full
94 /// forward-anywhere behaviour for non-cluster clients.
95 #[must_use]
96 pub fn with_cluster(mut self, port_base: u16) -> Self {
97 self.cluster_port_base = Some(port_base);
98 self
99 }
100
101 /// SLOWLOG tuning (`[slowlog]` config section). Default
102 /// `slower_than_micros = -1` (OFF) so the hot path never reads the
103 /// clock — every enabled command otherwise pays an `Instant::now()`
104 /// pair around dispatch, ~30 ns/op (≈9 % at 3 M ops/s). To match
105 /// Redis's 10 ms default, pass `10_000`; `0` records all; `-1`
106 /// disables. `max_len` is the per-shard ring cap (default 128).
107 #[must_use]
108 pub fn with_slowlog(mut self, slower_than_micros: i64, max_len: u32) -> Self {
109 self.slowlog_slower_than_micros = slower_than_micros;
110 self.slowlog_max_len = max_len;
111 self
112 }
113
114 /// Reactor tuning knobs (`[advanced]` config section). Defaults
115 /// match the original hardcoded constants. `ring_capacity` is
116 /// applied at SPSC ring construction (startup only); the other
117 /// three are read at each iteration of the reactor loop, so
118 /// values applied here take effect from the next shard.run() call.
119 #[must_use]
120 pub fn with_advanced(
121 mut self,
122 spin_limit: u32,
123 park_timeout_ms: u32,
124 tick_check_every: u32,
125 ring_capacity: usize,
126 ) -> Self {
127 self.spin_limit = spin_limit;
128 self.park_timeout_ms = park_timeout_ms;
129 self.tick_check_every = tick_check_every;
130 self.ring_capacity = ring_capacity;
131 self
132 }
133
134 /// Set the directory where shards snapshot to / load from. Default: `.`.
135 ///
136 /// This sets the RUNTIME's directory. It does not reach a
137 /// [`Commands`](crate::Commands) implementation's own configuration,
138 /// which is a separate object — so a server built this way answers
139 /// `CONFIG GET dir` from that config rather than from here, and the
140 /// two disagree unless the embedder sets both.
141 ///
142 /// `kevy::serve` builds both from one `Config`, so the shipped
143 /// binary never sees the gap; a programmatic build can, and a test
144 /// that used `CONFIG GET dir` to identify its own server found `.`
145 /// where it had passed a temp directory.
146 #[must_use]
147 pub fn with_data_dir(mut self, dir: impl Into<PathBuf>) -> Self {
148 self.data_dir = dir.into();
149 self
150 }
151
152 /// Only shards `0..N` arm accept SQE; rest stay compute-only.
153 /// `None` = every shard accepts (the default; byte-identical to
154 /// the pre-flag behaviour).
155 #[must_use]
156 pub fn with_accept_shards(mut self, n: Option<usize>) -> Self {
157 self.accept_shards = n;
158 self
159 }
160
161 /// Total cap on active client connections. `0` = unlimited.
162 /// Default `10_000`. Per-shard slice is `ceil(N / nshards)`.
163 #[must_use]
164 pub fn with_max_clients(mut self, n: usize) -> Self {
165 self.max_clients = n;
166 self
167 }
168
169 /// Transparent-tiering RAM budget for the whole process, in
170 /// resolved bytes (the caller resolves `auto` / percent forms
171 /// against `kevy_sys::detected_memory_bound` first). Split evenly
172 /// across shards. `None` (default) = tiering off unless the
173 /// minimal `KEVY_TIER_BUDGET` plain-bytes env knob is set.
174 #[must_use]
175 pub fn with_tier_budget(mut self, bytes: Option<u64>) -> Self {
176 self.tier_budget = bytes;
177 self
178 }
179
180 /// Cold-tier spill dir override (`[tiering] spill_dir`). `None`
181 /// (default) = `<data_dir>/tier/`.
182 #[must_use]
183 pub fn with_tier_spill_dir(mut self, dir: Option<PathBuf>) -> Self {
184 self.tier_dir = dir;
185 self
186 }
187
188 /// Enable/disable the append-only log. Default: enabled.
189 #[must_use]
190 pub fn with_aof(mut self, on: bool) -> Self {
191 self.enable_aof = on;
192 self
193 }
194
195 /// fsync policy for the AOF. Default `EverySec` matches Redis (lose at
196 /// most ~1 s of writes on a crash). `Always` is zero-loss but ~50 %
197 /// throughput; `No` defers everything to the OS pagecache.
198 #[must_use]
199 pub fn with_appendfsync(mut self, fsync: Fsync) -> Self {
200 self.appendfsync = fsync;
201 self
202 }
203
204 /// Auto-trigger BGREWRITEAOF when the live AOF has grown by at least
205 /// `pct` percent above its size at the previous rewrite, AND is at
206 /// least `min_size` bytes. `pct=0` disables auto-rewrite (clients can
207 /// still run BGREWRITEAOF manually). Defaults: 100 % / 64 MiB.
208 #[must_use]
209 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
210 self.auto_aof_rewrite_pct = pct;
211 self.auto_aof_rewrite_min_size = min_size;
212 self
213 }
214
215 /// Absolute-size auto-rewrite trigger: compact whenever the AOF
216 /// reaches `bytes`, regardless of growth ratio (0 = rule off).
217 /// Complements [`Self::with_auto_aof_rewrite`], whose growth rule
218 /// lets a large log double before compacting.
219 #[must_use]
220 pub fn with_auto_rewrite_bytes(mut self, bytes: u64) -> Self {
221 self.auto_aof_rewrite_bytes = bytes;
222 self
223 }
224
225 /// Time-based auto-rewrite trigger: compact at least every
226 /// `interval_secs` seconds while the log grows (0 = rule off).
227 #[must_use]
228 pub fn with_auto_rewrite_interval_secs(mut self, interval_secs: u64) -> Self {
229 self.auto_aof_rewrite_interval_secs = interval_secs;
230 self
231 }
232
233 /// Best-effort boot replay: recover the good records behind a corrupt
234 /// v2 AOF record instead of dropping them. Default false (strict).
235 #[must_use]
236 pub fn with_replay_resync(mut self, resync: bool) -> Self {
237 self.replay_resync = resync;
238 self
239 }
240}