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 arrive in
20 /// subsequent v3-cluster tasks (T1.12+); enabling without those
21 /// landed means the backlog fills and frames are dropped per the
22 /// source's eviction policy, but writes proceed normally.
23 #[must_use]
24 pub fn with_replication(mut self, enabled: bool, buffer_size: u64) -> Self {
25 self.enable_replication = enabled;
26 if buffer_size > 0 {
27 self.replication_buffer_size = buffer_size;
28 }
29 self
30 }
31
32 /// v2.3: enable the FEED.* consumer surface. Keeps a per-shard
33 /// backlog (even with no replicas) and persists the (generation,
34 /// offset) cursor via the feed sidecars. `buffer_size` = 0 keeps
35 /// the default (64 MB/shard); effective budget is
36 /// `max(replication_buffer_size, feed_buffer_size)` when both
37 /// features are on.
38 #[must_use]
39 pub fn with_feed(mut self, enabled: bool, buffer_size: u64) -> Self {
40 self.feed_enabled = enabled;
41 if buffer_size > 0 {
42 self.feed_buffer_size = buffer_size;
43 }
44 self
45 }
46
47 /// Bring up a replication listener per shard at
48 /// `port_base + shard_id` (per Issue Ledger I2 — mirrors the
49 /// cluster listener pattern). Replica clients connect to each
50 /// per-shard port to mirror the full keyspace. This is independent
51 /// of [`Self::with_replication`]: a primary that runs the producer
52 /// backlog without a listener (benchmarks, embed-only) is
53 /// supported.
54 #[must_use]
55 pub fn with_replication_listener(mut self, port_base: u16) -> Self {
56 self.replication_port_base = Some(port_base);
57 self
58 }
59
60 /// Per-shard SlotTable reconnect window in milliseconds — the
61 /// grace period a disconnected replica's slot is retained for so
62 /// a reconnect within the window can be correlated against its
63 /// prior `sent_offset`. Default `60_000` (60 s); pass `0` to drop
64 /// slots immediately on disconnect.
65 #[must_use]
66 pub fn with_replication_reconnect_window(mut self, ms: u32) -> Self {
67 self.replication_reconnect_window_ms = ms;
68 self
69 }
70
71 /// Install per-shard replica inboxes (T1.29). The embedder pre-
72 /// constructs `nshards` inbox pairs via
73 /// [`crate::replica_inbox_pair`], keeps the senders to hand to
74 /// the per-shard replica runner threads, and passes the receivers
75 /// here. The order of `receivers` is shard-major: index `i` ↔
76 /// shard `i`. Length must equal `nshards`. When this builder
77 /// isn't called, no shard has an inbox (the standalone /
78 /// primary-only behaviour pre-T1.29).
79 #[must_use]
80 pub fn with_replica_inboxes(
81 mut self,
82 receivers: Vec<crate::replica_inbox::ReplicaInboxReceiver>,
83 ) -> Self {
84 self.replica_inboxes = receivers.into_iter().map(Some).collect();
85 self
86 }
87
88 /// Enable single-node cluster mode: keys route by Redis-cluster slot
89 /// (CRC16 `{hashtag}` & 16383, contiguous even ranges) and every shard
90 /// `i` binds a second, deterministic listener at `port_base + i` that
91 /// answers wrong-shard keys with `-MOVED` instead of forwarding. The
92 /// SO_REUSEPORT listener on the main port keeps today's full
93 /// forward-anywhere behaviour for non-cluster clients.
94 #[must_use]
95 pub fn with_cluster(mut self, port_base: u16) -> Self {
96 self.cluster_port_base = Some(port_base);
97 self
98 }
99
100 /// SLOWLOG tuning (`[slowlog]` config section). Default
101 /// `slower_than_micros = -1` (OFF) so the hot path never reads the
102 /// clock — every enabled command otherwise pays an `Instant::now()`
103 /// pair around dispatch, ~30 ns/op (≈9 % at 3 M ops/s). To match
104 /// Redis's 10 ms default, pass `10_000`; `0` records all; `-1`
105 /// disables. `max_len` is the per-shard ring cap (default 128).
106 #[must_use]
107 pub fn with_slowlog(mut self, slower_than_micros: i64, max_len: u32) -> Self {
108 self.slowlog_slower_than_micros = slower_than_micros;
109 self.slowlog_max_len = max_len;
110 self
111 }
112
113 /// Reactor tuning knobs (`[advanced]` config section). Defaults
114 /// match the pre-v1.4 hardcoded constants. `ring_capacity` is
115 /// applied at SPSC ring construction (startup only); the other
116 /// three are read at each iteration of the reactor loop, so
117 /// values applied here take effect from the next shard.run() call.
118 #[must_use]
119 pub fn with_advanced(
120 mut self,
121 spin_limit: u32,
122 park_timeout_ms: u32,
123 tick_check_every: u32,
124 ring_capacity: usize,
125 ) -> Self {
126 self.spin_limit = spin_limit;
127 self.park_timeout_ms = park_timeout_ms;
128 self.tick_check_every = tick_check_every;
129 self.ring_capacity = ring_capacity;
130 self
131 }
132
133 /// Set the directory where shards snapshot to / load from. Default: `.`.
134 #[must_use]
135 pub fn with_data_dir(mut self, dir: impl Into<PathBuf>) -> Self {
136 self.data_dir = dir.into();
137 self
138 }
139
140 /// **v1.30** — Only shards `0..N` arm accept SQE; rest stay compute-only.
141 /// `None` = every shard accepts (v1.29 byte-identical, the default).
142 #[must_use]
143 pub fn with_accept_shards(mut self, n: Option<usize>) -> Self {
144 self.accept_shards = n;
145 self
146 }
147
148 /// **v1.37** — total cap on active client connections. `0` = unlimited.
149 /// Default `10_000`. Per-shard slice is `ceil(N / nshards)`.
150 #[must_use]
151 pub fn with_max_clients(mut self, n: usize) -> Self {
152 self.max_clients = n;
153 self
154 }
155
156 /// Enable/disable the append-only log. Default: enabled.
157 #[must_use]
158 pub fn with_aof(mut self, on: bool) -> Self {
159 self.enable_aof = on;
160 self
161 }
162
163 /// fsync policy for the AOF. Default `EverySec` matches Redis (lose at
164 /// most ~1 s of writes on a crash). `Always` is zero-loss but ~50 %
165 /// throughput; `No` defers everything to the OS pagecache.
166 #[must_use]
167 pub fn with_appendfsync(mut self, fsync: Fsync) -> Self {
168 self.appendfsync = fsync;
169 self
170 }
171
172 /// Auto-trigger BGREWRITEAOF when the live AOF has grown by at least
173 /// `pct` percent above its size at the previous rewrite, AND is at
174 /// least `min_size` bytes. `pct=0` disables auto-rewrite (clients can
175 /// still run BGREWRITEAOF manually). Defaults: 100 % / 64 MiB.
176 #[must_use]
177 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
178 self.auto_aof_rewrite_pct = pct;
179 self.auto_aof_rewrite_min_size = min_size;
180 self
181 }
182}