kevy_embedded/config.rs
1//! Embedded-store configuration. Builder-style — every knob has a sane
2//! default so `Config::default()` works for the simplest use case
3//! (in-memory, no persistence, background TTL reaper).
4
5use std::path::PathBuf;
6use std::time::Duration;
7
8pub use kevy_persist::Fsync as AppendFsync;
9pub use kevy_store::EvictionPolicy;
10
11/// How the active TTL reaper runs.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum TtlReaperMode {
14 /// Spawn a background thread that ticks at the configured interval
15 /// (default 100 ms / 10 Hz, matching Redis's `hz=10`). Default.
16 Background,
17 /// Caller-driven via [`crate::Store::tick`]. Required for WASM
18 /// targets (no threads) and single-threaded apps that don't want a
19 /// background worker.
20 Manual,
21}
22
23/// Embedded-store config. Build by chaining `with_*` methods on
24/// [`Config::default`].
25#[derive(Debug, Clone)]
26pub struct Config {
27 /// Soft memory ceiling in bytes. `0` (default) = unlimited.
28 pub maxmemory: u64,
29 /// Eviction policy when over `maxmemory`. Default `NoEviction`.
30 pub eviction_policy: EvictionPolicy,
31 /// Persistence directory. `None` = pure in-memory (no AOF, no snapshot).
32 pub data_dir: Option<PathBuf>,
33 /// AOF on/off when `data_dir` is set. Defaults to `true` (on) when
34 /// `with_persist` was called; ignored if `data_dir` is `None`.
35 pub aof: bool,
36 /// AOF fsync policy. Default `EverySec` (matches Redis: ≤ 1 s loss).
37 pub appendfsync: AppendFsync,
38 /// Snapshot file name inside `data_dir` (single-shard only; `n > 1`
39 /// always uses `dump-{i}.rdb`). Default `"dump-0.rdb"`. A custom name
40 /// opts the dir out of server interop: no `shards.meta` is recorded,
41 /// and a `kevy` server opening the same dir won't find the files.
42 pub snapshot_filename: String,
43 /// AOF file name inside `data_dir` (single-shard only; `n > 1` always
44 /// uses `aof-{i}.aof`). Default `"aof-0.aof"`. Same interop opt-out as
45 /// [`Self::snapshot_filename`].
46 pub aof_filename: String,
47 /// TTL reaper mode. Default `Background`.
48 pub ttl_reaper: TtlReaperMode,
49 /// Reaper tick interval. Default 100 ms (10 Hz).
50 pub reaper_interval: Duration,
51 /// `tick_expire` samples per round. Default 20 (matches Redis).
52 pub reaper_samples: usize,
53 /// Max sample rounds per tick. Default 16.
54 pub reaper_max_rounds: u32,
55 /// Auto-`BGREWRITEAOF` trigger: rewrite when the live AOF has grown by at
56 /// least this percent over its size at the previous rewrite. `0` disables
57 /// (call [`crate::Store::rewrite_aof`] manually). Default `100` (Redis).
58 pub auto_aof_rewrite_pct: u32,
59 /// Floor below which auto-rewrite is skipped. Default `64 MiB` (Redis).
60 pub auto_aof_rewrite_min_size: u64,
61 /// Optional push-style metric callback (replay / rewrite events). Default
62 /// `None`. Set via [`Self::with_metric_sink`]; not part of `Debug` output.
63 pub(crate) metric_sink: Option<crate::metric::MetricSink>,
64 /// Keyspace shard count (`hash(key) % shards`), each a fully independent
65 /// lock + keyspace + AOF (shared-nothing) — concurrent access scales across
66 /// cores. **Default `1`** (single shard = the original single-lock /
67 /// single-`aof-0.aof` layout, zero migration). Set `> 1` via
68 /// [`Self::with_shards`]; the first open with `> 1` re-shards an existing
69 /// single AOF into per-shard files.
70 pub shards: usize,
71 /// Replication upstream. `Some("host:port")` makes this store a
72 /// read-replica that streams writes from the named primary; `None`
73 /// (default) is a normal primary store. Configured via
74 /// [`Self::with_replica_upstream`] or the convenience constructor
75 /// [`crate::Store::open_replica`]. See Phase 2 of the v3-cluster RFC.
76 pub replica_upstream: Option<String>,
77 /// Replica identity string sent to the primary at handshake
78 /// (`REPLICATE FROM <offset> ID <replica_id>`). Default
79 /// `"kevy-embedded-replica"`. Override per-process when multiple
80 /// embed replicas connect to the same primary (they'd otherwise
81 /// share the slot and clobber each other's session state on the
82 /// primary side).
83 pub replica_id: String,
84 /// Replica reconnect backoff: lower bound. Default 100 ms. The
85 /// runner sleeps this long after the first connection failure;
86 /// each subsequent failure doubles the wait up to
87 /// [`Self::replica_reconnect_max`].
88 pub replica_reconnect_min: Duration,
89 /// Replica reconnect backoff: upper bound. Default 5 s — picked to
90 /// match the v1.18 server's `reconnect_window_ms` default so embed
91 /// replicas and server replicas behave identically when the same
92 /// primary disappears.
93 pub replica_reconnect_max: Duration,
94 /// Phase 3 / v1.21 **embed-as-writer**: bind address
95 /// (`"host:port"` or `"0.0.0.0:port"`) for the replication
96 /// source listener. When `Some`, every commit on this store
97 /// pushes its argv into a process-local `ReplicationSource`
98 /// backlog, and replicas (other embeds, server-as-replicas)
99 /// connect to this port to stream the writes. `None` (default)
100 /// keeps the embed in pure-local mode. Mutually exclusive in
101 /// spirit with `replica_upstream` (a single store should be
102 /// either a writer source OR a reader sink, not both); the
103 /// builder doesn't reject the combo so tests can exercise the
104 /// guard rails.
105 pub embed_writer_listen_addr: Option<String>,
106 /// Backlog byte budget for the embed-as-writer source. Default
107 /// `1 MiB` (matches the v1.18 server replication default).
108 /// Set higher when consumers may disconnect for longer than
109 /// the backlog can buffer (otherwise reconnects hit `TooOld`
110 /// and v1.21 closes the link — snapshot-ship from embed is a
111 /// follow-up).
112 pub embed_writer_backlog_bytes: usize,
113}
114
115impl Default for Config {
116 fn default() -> Self {
117 Self {
118 maxmemory: 0,
119 eviction_policy: EvictionPolicy::NoEviction,
120 data_dir: None,
121 aof: true,
122 appendfsync: AppendFsync::EverySec,
123 snapshot_filename: String::from("dump-0.rdb"),
124 aof_filename: String::from("aof-0.aof"),
125 ttl_reaper: TtlReaperMode::Background,
126 reaper_interval: Duration::from_millis(100),
127 reaper_samples: 20,
128 reaper_max_rounds: 16,
129 auto_aof_rewrite_pct: 100,
130 auto_aof_rewrite_min_size: 64 * 1024 * 1024,
131 metric_sink: None,
132 shards: 1,
133 replica_upstream: None,
134 replica_id: String::from("kevy-embedded-replica"),
135 replica_reconnect_min: Duration::from_millis(100),
136 replica_reconnect_max: Duration::from_secs(5),
137 embed_writer_listen_addr: None,
138 embed_writer_backlog_bytes: 1024 * 1024,
139 }
140 }
141}
142
143impl Config {
144 /// Enable persistence under `dir` — snapshot file + AOF land inside.
145 /// AOF defaults on; turn it off with [`Self::without_aof`] for pure
146 /// snapshot-only durability.
147 #[must_use]
148 pub fn with_persist(mut self, dir: impl Into<PathBuf>) -> Self {
149 self.data_dir = Some(dir.into());
150 self
151 }
152
153 /// Disable the AOF (snapshot-only persistence — explicit `save_snapshot`
154 /// calls are the only way data survives restart).
155 #[must_use]
156 pub fn without_aof(mut self) -> Self {
157 self.aof = false;
158 self
159 }
160
161 /// Soft memory ceiling in bytes. `0` keeps the default (unlimited).
162 #[must_use]
163 pub fn with_max_memory(mut self, bytes: u64) -> Self {
164 self.maxmemory = bytes;
165 self
166 }
167
168 /// Eviction policy when over [`Self::with_max_memory`].
169 #[must_use]
170 pub fn with_eviction(mut self, policy: EvictionPolicy) -> Self {
171 self.eviction_policy = policy;
172 self
173 }
174
175 /// AOF fsync policy. Default [`AppendFsync::EverySec`].
176 #[must_use]
177 pub fn with_appendfsync(mut self, fsync: AppendFsync) -> Self {
178 self.appendfsync = fsync;
179 self
180 }
181
182 /// Auto-`BGREWRITEAOF` thresholds: rewrite once the AOF has grown `pct`
183 /// percent past its size at the last rewrite AND is at least `min_size`
184 /// bytes. In `Background` reaper mode the check runs on the reaper tick;
185 /// in `Manual` mode it runs when you call [`crate::Store::tick`]. Pass
186 /// `pct = 0` to disable auto-rewrite (you can still call
187 /// [`crate::Store::rewrite_aof`] yourself). Defaults: 100 % / 64 MiB.
188 #[must_use]
189 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
190 self.auto_aof_rewrite_pct = pct;
191 self.auto_aof_rewrite_min_size = min_size;
192 self
193 }
194
195 /// Shard the keyspace into `n` shared-nothing partitions (`hash(key) % n`),
196 /// each with its own lock + keyspace + AOF, so concurrent access scales
197 /// across cores. `n` clamps to ≥ 1; `1` (default) is the original
198 /// single-shard layout. Going from a single-AOF store to `n > 1`
199 /// re-shards the existing `aof-0.aof` into `aof-0..aof-{n-1}` on the next
200 /// open (the old file is backed up to `aof-0.aof.premigration.<ts>` first).
201 /// Pub/sub is process-wide (handled on shard 0), not sharded.
202 #[must_use]
203 pub fn with_shards(mut self, n: usize) -> Self {
204 self.shards = n.max(1);
205 self
206 }
207
208 /// Register a push-style metric callback. It receives a [`crate::KevyMetric`] for
209 /// each AOF replay (startup) and AOF rewrite (compaction) — wire it to
210 /// Prometheus / a log line / a counter. The callback runs synchronously on
211 /// the emitting thread (reaper thread for background rewrites), so keep it
212 /// fast and non-blocking. Replaces any previously-set sink.
213 #[must_use]
214 pub fn with_metric_sink(
215 mut self,
216 sink: impl Fn(crate::KevyMetric) + Send + Sync + 'static,
217 ) -> Self {
218 self.metric_sink = Some(crate::metric::MetricSink::new(sink));
219 self
220 }
221
222 /// Caller-driven TTL reaping — disables the background thread.
223 /// Required for WASM (no threads available). Call
224 /// [`crate::Store::tick`] yourself from your event loop.
225 #[must_use]
226 pub fn with_ttl_reaper_manual(mut self) -> Self {
227 self.ttl_reaper = TtlReaperMode::Manual;
228 self
229 }
230
231 /// Configure this store as a replication replica of `upstream`
232 /// (`"host:port"` of a kevy server's replication listener). A
233 /// background thread streams writes from the primary and applies
234 /// them locally; this store rejects local writes with a
235 /// `READONLY` error. See [`crate::Store::open_replica`] for the
236 /// convenience constructor. Phase 2 of the v3-cluster RFC.
237 #[must_use]
238 pub fn with_replica_upstream(mut self, upstream: impl Into<String>) -> Self {
239 self.replica_upstream = Some(upstream.into());
240 self
241 }
242
243 /// Override the replica identity sent to the primary at handshake.
244 /// Useful when multiple embed replicas share one primary —
245 /// otherwise they'd share the slot and stomp each other's session
246 /// state.
247 #[must_use]
248 pub fn with_replica_id(mut self, id: impl Into<String>) -> Self {
249 self.replica_id = id.into();
250 self
251 }
252
253 /// Override the replica reconnect backoff bounds.
254 #[must_use]
255 pub fn with_replica_reconnect(mut self, min: Duration, max: Duration) -> Self {
256 self.replica_reconnect_min = min;
257 self.replica_reconnect_max = max.max(min);
258 self
259 }
260
261 /// Run this store as a Phase 3 embed-as-writer: bind a
262 /// replication source listener on `bind_addr` so replicas can
263 /// subscribe to the writes applied here.
264 #[must_use]
265 pub fn with_embed_writer(mut self, bind_addr: impl Into<String>) -> Self {
266 self.embed_writer_listen_addr = Some(bind_addr.into());
267 self
268 }
269
270 /// Override the embed-as-writer backlog byte budget.
271 #[must_use]
272 pub fn with_embed_writer_backlog(mut self, bytes: usize) -> Self {
273 self.embed_writer_backlog_bytes = bytes.max(64 * 1024);
274 self
275 }
276
277 /// Override the background reaper interval. Default 100 ms.
278 #[must_use]
279 pub fn with_reaper_interval(mut self, iv: Duration) -> Self {
280 self.reaper_interval = iv;
281 self
282 }
283
284 /// Override the snapshot file name inside `data_dir`.
285 #[must_use]
286 pub fn with_snapshot_filename(mut self, name: impl Into<String>) -> Self {
287 self.snapshot_filename = name.into();
288 self
289 }
290
291 /// Override the AOF file name inside `data_dir`.
292 #[must_use]
293 pub fn with_aof_filename(mut self, name: impl Into<String>) -> Self {
294 self.aof_filename = name.into();
295 self
296 }
297}
298
299#[cfg(test)]
300mod tests {
301 use super::*;
302
303 #[test]
304 fn default_is_pure_in_memory() {
305 let c = Config::default();
306 assert_eq!(c.maxmemory, 0);
307 assert!(c.data_dir.is_none());
308 assert_eq!(c.ttl_reaper, TtlReaperMode::Background);
309 assert!(c.aof);
310 }
311
312 #[test]
313 fn builder_chains() {
314 let c = Config::default()
315 .with_persist("/tmp/foo")
316 .with_max_memory(1024)
317 .with_eviction(EvictionPolicy::AllKeysLru)
318 .with_ttl_reaper_manual()
319 .with_appendfsync(AppendFsync::Always);
320 assert_eq!(c.data_dir.as_deref(), Some(std::path::Path::new("/tmp/foo")));
321 assert_eq!(c.maxmemory, 1024);
322 assert_eq!(c.eviction_policy, EvictionPolicy::AllKeysLru);
323 assert_eq!(c.ttl_reaper, TtlReaperMode::Manual);
324 }
325
326 #[test]
327 fn without_aof_disables_logging_path() {
328 let c = Config::default().with_persist("/tmp/foo").without_aof();
329 assert!(c.data_dir.is_some());
330 assert!(!c.aof);
331 }
332}