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 #[must_use]
136 pub fn with_data_dir(mut self, dir: impl Into<PathBuf>) -> Self {
137 self.data_dir = dir.into();
138 self
139 }
140
141 /// Only shards `0..N` arm accept SQE; rest stay compute-only.
142 /// `None` = every shard accepts (the default; byte-identical to
143 /// the pre-flag behaviour).
144 #[must_use]
145 pub fn with_accept_shards(mut self, n: Option<usize>) -> Self {
146 self.accept_shards = n;
147 self
148 }
149
150 /// Total cap on active client connections. `0` = unlimited.
151 /// Default `10_000`. Per-shard slice is `ceil(N / nshards)`.
152 #[must_use]
153 pub fn with_max_clients(mut self, n: usize) -> Self {
154 self.max_clients = n;
155 self
156 }
157
158 /// Transparent-tiering RAM budget for the whole process, in
159 /// resolved bytes (the caller resolves `auto` / percent forms
160 /// against `kevy_sys::detected_memory_bound` first). Split evenly
161 /// across shards. `None` (default) = tiering off unless the
162 /// minimal `KEVY_TIER_BUDGET` plain-bytes env knob is set.
163 #[must_use]
164 pub fn with_tier_budget(mut self, bytes: Option<u64>) -> Self {
165 self.tier_budget = bytes;
166 self
167 }
168
169 /// Cold-tier spill dir override (`[tiering] spill_dir`). `None`
170 /// (default) = `<data_dir>/tier/`.
171 #[must_use]
172 pub fn with_tier_spill_dir(mut self, dir: Option<PathBuf>) -> Self {
173 self.tier_dir = dir;
174 self
175 }
176
177 /// Enable/disable the append-only log. Default: enabled.
178 #[must_use]
179 pub fn with_aof(mut self, on: bool) -> Self {
180 self.enable_aof = on;
181 self
182 }
183
184 /// fsync policy for the AOF. Default `EverySec` matches Redis (lose at
185 /// most ~1 s of writes on a crash). `Always` is zero-loss but ~50 %
186 /// throughput; `No` defers everything to the OS pagecache.
187 #[must_use]
188 pub fn with_appendfsync(mut self, fsync: Fsync) -> Self {
189 self.appendfsync = fsync;
190 self
191 }
192
193 /// Auto-trigger BGREWRITEAOF when the live AOF has grown by at least
194 /// `pct` percent above its size at the previous rewrite, AND is at
195 /// least `min_size` bytes. `pct=0` disables auto-rewrite (clients can
196 /// still run BGREWRITEAOF manually). Defaults: 100 % / 64 MiB.
197 #[must_use]
198 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
199 self.auto_aof_rewrite_pct = pct;
200 self.auto_aof_rewrite_min_size = min_size;
201 self
202 }
203
204 /// Absolute-size auto-rewrite trigger: compact whenever the AOF
205 /// reaches `bytes`, regardless of growth ratio (0 = rule off).
206 /// Complements [`Self::with_auto_aof_rewrite`], whose growth rule
207 /// lets a large log double before compacting.
208 #[must_use]
209 pub fn with_auto_rewrite_bytes(mut self, bytes: u64) -> Self {
210 self.auto_aof_rewrite_bytes = bytes;
211 self
212 }
213
214 /// Time-based auto-rewrite trigger: compact at least every
215 /// `interval_secs` seconds while the log grows (0 = rule off).
216 #[must_use]
217 pub fn with_auto_rewrite_interval_secs(mut self, interval_secs: u64) -> Self {
218 self.auto_aof_rewrite_interval_secs = interval_secs;
219 self
220 }
221
222 /// Best-effort boot replay: recover the good records behind a corrupt
223 /// v2 AOF record instead of dropping them. Default false (strict).
224 #[must_use]
225 pub fn with_replay_resync(mut self, resync: bool) -> Self {
226 self.replay_resync = resync;
227 self
228 }
229}