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
5#[cfg(feature = "persist")]
6use std::path::PathBuf;
7use std::time::Duration;
8
9#[cfg(feature = "persist")]
10pub use kevy_persist::Fsync as AppendFsync;
11pub use kevy_store::EvictionPolicy;
12
13#[cfg(feature = "tier")]
14pub use crate::config_tier::TierBudgetSpec;
15
16/// How the active TTL reaper runs.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub enum TtlReaperMode {
19 /// Spawn a background thread that ticks at the configured interval
20 /// (default 100 ms / 10 Hz, matching Redis's `hz=10`). Default.
21 Background,
22 /// Caller-driven via [`crate::Store::tick`]. Required for WASM
23 /// targets (no threads) and single-threaded apps that don't want a
24 /// background worker.
25 Manual,
26}
27
28/// Embedded-store config. Build by chaining `with_*` methods on
29/// [`Config::default`].
30#[derive(Debug, Clone)]
31pub struct Config {
32 /// Optional READ-ONLY RESP listener address (ops tooling —
33 /// redis-cli against a live embedded store). `None` (default) =
34 /// no listener thread, no socket, zero tax.
35 #[cfg(feature = "listener")]
36 pub resp_listener: Option<std::net::SocketAddr>,
37 /// Soft memory ceiling in bytes. `0` (default) = unlimited.
38 pub maxmemory: u64,
39 /// Eviction policy when over `maxmemory`. Default `NoEviction`.
40 pub eviction_policy: EvictionPolicy,
41 /// Persistence directory. `None` = pure in-memory (no AOF, no snapshot).
42 #[cfg(feature = "persist")]
43 pub data_dir: Option<PathBuf>,
44 /// AOF on/off when `data_dir` is set. Defaults to `true` (on) when
45 /// `with_persist` was called; ignored if `data_dir` is `None`.
46 #[cfg(feature = "persist")]
47 pub aof: bool,
48 /// AOF fsync policy. Default `EverySec` (matches Redis: ≤ 1 s loss).
49 #[cfg(feature = "persist")]
50 pub appendfsync: AppendFsync,
51 /// Transparent-tiering RAM budget (capacity arc). `None` (default)
52 /// = tiering off — today's paths byte-identical. `Some` requires a
53 /// disk `data_dir` (the cold value log lives at `<data_dir>/tier/`);
54 /// a mem-only store rejects the combo at open. The budget is the
55 /// WHOLE store's (split evenly across shards); auto/percent forms
56 /// resolve against the detected memory bound at open and re-resolve
57 /// on every reaper tick.
58 #[cfg(feature = "tier")]
59 pub tier_budget: Option<TierBudgetSpec>,
60 /// Largest value the tier may spill (bytes; 0 = unlimited). Default
61 /// 256 KiB (RFC §7): an embedded cold read holds the shard lock for
62 /// the pread, so the cap bounds that hold time. Over-cap values
63 /// simply stay hot.
64 pub max_spill_value: u64,
65 /// TTL reaper mode. Default `Background`.
66 pub ttl_reaper: TtlReaperMode,
67 /// Reaper tick interval. Default 100 ms (10 Hz).
68 pub reaper_interval: Duration,
69 /// `tick_expire` samples per round. Default 20 (matches Redis).
70 pub reaper_samples: usize,
71 /// Max sample rounds per tick. Default 16.
72 pub reaper_max_rounds: u32,
73 /// Auto-`BGREWRITEAOF` trigger: rewrite when the live AOF has grown by at
74 /// least this percent over its size at the previous rewrite. `0` disables
75 /// (call [`crate::Store::rewrite_aof`] manually). Default `100` (Redis).
76 #[cfg(feature = "persist")]
77 pub auto_aof_rewrite_pct: u32,
78 /// Floor below which auto-rewrite is skipped. Default `64 MiB` (Redis).
79 #[cfg(feature = "persist")]
80 pub auto_aof_rewrite_min_size: u64,
81 /// Absolute-size auto-rewrite trigger in bytes (0 = off). The growth
82 /// rule alone lets a large log double before compacting — a 2.2 GB AOF
83 /// waits for 4.4 GB; this caps it outright.
84 pub auto_aof_rewrite_bytes: u64,
85 /// Time-based auto-rewrite trigger in seconds (0 = off): compact at
86 /// least this often while the log grows.
87 pub auto_aof_rewrite_interval_secs: u64,
88 /// Best-effort replay: on a corrupt v2 record, hop to the next valid
89 /// record (length + CRC + parse all agree) instead of dropping the
90 /// good tail behind it. Default false (strict).
91 pub replay_resync: bool,
92 /// Optional push-style metric callback (replay / rewrite events). Default
93 /// `None`. Set via [`Self::with_metric_sink`]; not part of `Debug` output.
94 #[cfg(feature = "persist")]
95 pub(crate) metric_sink: Option<crate::metric::MetricSink>,
96 /// Keyspace shard count (`hash(key) % shards`), each a fully independent
97 /// lock + keyspace + AOF (shared-nothing) — concurrent access scales across
98 /// cores. **Default `1`** (single shard = the original single-lock /
99 /// single-`aof-0.aof` layout, zero migration). Set `> 1` via
100 /// [`Self::with_shards`]; the first open with `> 1` re-shards an existing
101 /// single AOF into per-shard files.
102 pub shards: usize,
103 /// Replication upstream. `Some("host:port")` makes this store a
104 /// read-replica that streams writes from the named primary; `None`
105 /// (default) is a normal primary store. Configured via
106 /// [`Self::with_replica_upstream`] or the convenience constructor
107 /// [`crate::Store::open_replica`].
108 #[cfg(feature = "replicate")]
109 pub replica_upstream: Option<String>,
110 /// Replica identity string sent to the primary at handshake
111 /// (`REPLICATE FROM <offset> ID <replica_id>`). Default
112 /// `"kevy-embedded-replica"`. Override per-process when multiple
113 /// embed replicas connect to the same primary (they'd otherwise
114 /// share the slot and clobber each other's session state on the
115 /// primary side).
116 #[cfg(feature = "replicate")]
117 pub replica_id: String,
118 /// Replica reconnect backoff: lower bound. Default 100 ms. The
119 /// runner sleeps this long after the first connection failure;
120 /// each subsequent failure doubles the wait up to
121 /// [`Self::replica_reconnect_max`].
122 #[cfg(feature = "replicate")]
123 pub replica_reconnect_min: Duration,
124 /// Replica reconnect backoff: upper bound. Default 5 s — matches
125 /// the server-side replica reconnect default so embed replicas and
126 /// server replicas behave identically when the same primary
127 /// disappears.
128 #[cfg(feature = "replicate")]
129 pub replica_reconnect_max: Duration,
130 /// Embed-as-writer bind address (`"host:port"` or
131 /// `"0.0.0.0:port"`) for the replication source listener. When
132 /// `Some`, every commit on this store pushes its argv into a
133 /// process-local `ReplicationSource` backlog, and replicas (other
134 /// embeds, server-as-replicas) connect to this port to stream the
135 /// writes. `None` (default) keeps the embed in pure-local mode.
136 /// Mutually exclusive in spirit with `replica_upstream` (a single
137 /// store should be either a writer source or a reader sink, not
138 /// both); the builder does not reject the combo so tests can
139 /// exercise the guard rails.
140 #[cfg(feature = "replicate")]
141 pub embed_writer_listen_addr: Option<String>,
142 /// CDC feed (changes_since / changes_tail). Default off.
143 #[cfg(feature = "replicate")]
144 pub feed_enabled: bool,
145 /// Feed backlog byte budget. Default 64 MB, capped at 1 GB.
146 #[cfg(feature = "replicate")]
147 pub feed_buffer_size: u64,
148 /// Backlog byte budget for the embed-as-writer source. Default
149 /// `1 MiB` (matches the server replication default).
150 /// Set higher when consumers may disconnect for longer than
151 /// the backlog can buffer (otherwise a reconnect falls past the
152 /// backlog and is re-seeded with a full snapshot ship instead of
153 /// an incremental stream).
154 #[cfg(feature = "replicate")]
155 pub embed_writer_backlog_bytes: usize,
156}
157
158impl Default for Config {
159 fn default() -> Self {
160 Self {
161 #[cfg(feature = "listener")]
162 resp_listener: None,
163 maxmemory: 0,
164 eviction_policy: EvictionPolicy::NoEviction,
165 #[cfg(feature = "persist")]
166 data_dir: None,
167 #[cfg(feature = "persist")]
168 aof: true,
169 #[cfg(feature = "persist")]
170 appendfsync: AppendFsync::EverySec,
171 #[cfg(feature = "tier")]
172 tier_budget: None,
173 max_spill_value: 256 << 10,
174 ttl_reaper: TtlReaperMode::Background,
175 reaper_interval: Duration::from_millis(100),
176 reaper_samples: 20,
177 reaper_max_rounds: 16,
178 #[cfg(feature = "persist")]
179 auto_aof_rewrite_pct: 100,
180 #[cfg(feature = "persist")]
181 auto_aof_rewrite_min_size: 64 * 1024 * 1024,
182 auto_aof_rewrite_bytes: 0,
183 auto_aof_rewrite_interval_secs: 0,
184 replay_resync: false,
185 #[cfg(feature = "persist")]
186 metric_sink: None,
187 shards: 1,
188 #[cfg(feature = "replicate")]
189 replica_upstream: None,
190 #[cfg(feature = "replicate")]
191 replica_id: String::from("kevy-embedded-replica"),
192 #[cfg(feature = "replicate")]
193 replica_reconnect_min: Duration::from_millis(100),
194 #[cfg(feature = "replicate")]
195 replica_reconnect_max: Duration::from_secs(5),
196 #[cfg(feature = "replicate")]
197 embed_writer_listen_addr: None,
198 #[cfg(feature = "replicate")]
199 feed_enabled: false,
200 #[cfg(feature = "replicate")]
201 feed_buffer_size: 64 * 1024 * 1024,
202 #[cfg(feature = "replicate")]
203 embed_writer_backlog_bytes: 1024 * 1024,
204 }
205 }
206}
207
208impl Config {
209 /// Enable the read-only RESP listener on `addr`
210 /// (e.g. `"127.0.0.1:6009".parse().unwrap()`).
211 #[cfg(feature = "listener")]
212 #[must_use]
213 pub fn with_resp_listener(mut self, addr: std::net::SocketAddr) -> Self {
214 self.resp_listener = Some(addr);
215 self
216 }
217
218 /// Enable persistence under `dir` — snapshot file + AOF land inside.
219 /// AOF defaults on; turn it off with [`Self::without_aof`] for pure
220 /// snapshot-only durability.
221 #[cfg(feature = "persist")]
222 #[must_use]
223 pub fn with_persist(mut self, dir: impl Into<PathBuf>) -> Self {
224 self.data_dir = Some(dir.into());
225 self
226 }
227
228 /// Disable the AOF (snapshot-only persistence — explicit `save_snapshot`
229 /// calls are the only way data survives restart).
230 #[cfg(feature = "persist")]
231 #[must_use]
232 pub fn without_aof(mut self) -> Self {
233 self.aof = false;
234 self
235 }
236
237 /// Soft memory ceiling in bytes. `0` keeps the default (unlimited).
238 #[must_use]
239 pub fn with_max_memory(mut self, bytes: u64) -> Self {
240 self.maxmemory = bytes;
241 self
242 }
243
244 /// Eviction policy when over [`Self::with_max_memory`].
245 #[must_use]
246 pub fn with_eviction(mut self, policy: EvictionPolicy) -> Self {
247 self.eviction_policy = policy;
248 self
249 }
250
251 /// AOF fsync policy. Default [`AppendFsync::EverySec`].
252 #[cfg(feature = "persist")]
253 #[must_use]
254 pub fn with_appendfsync(mut self, fsync: AppendFsync) -> Self {
255 self.appendfsync = fsync;
256 self
257 }
258
259 /// Absolute-size auto-rewrite trigger: compact whenever the AOF reaches
260 /// `bytes`, regardless of growth ratio (0 = off). Complements
261 /// [`Self::with_auto_aof_rewrite`], whose growth rule is too sluggish
262 /// for long-lived instances with a large baseline.
263 #[cfg(feature = "persist")]
264 #[must_use]
265 pub fn with_auto_rewrite_bytes(mut self, bytes: u64) -> Self {
266 self.auto_aof_rewrite_bytes = bytes;
267 self
268 }
269
270 /// Time-based auto-rewrite trigger: compact at least every `interval`
271 /// while the log grows (zero duration = off).
272 #[cfg(feature = "persist")]
273 #[must_use]
274 pub fn with_auto_rewrite_interval(mut self, interval: std::time::Duration) -> Self {
275 self.auto_aof_rewrite_interval_secs = interval.as_secs();
276 self
277 }
278
279 /// Best-effort replay: recover the good records BEHIND a corrupt v2
280 /// record instead of dropping them (a production incident lost a
281 /// 231 MB well-formed tail over one bad frame). The skip is
282 /// deterministic — length + CRC32C + an exactly-one-command parse must
283 /// all agree before a record is trusted — and the open still reports
284 /// `corrupt` so hosts still alert. Strict (default false) remains the
285 /// conservative choice: nothing after the first bad byte is trusted.
286 #[cfg(feature = "persist")]
287 #[must_use]
288 pub fn with_replay_resync(mut self, resync: bool) -> Self {
289 self.replay_resync = resync;
290 self
291 }
292
293 /// Auto-`BGREWRITEAOF` thresholds: rewrite once the AOF has grown `pct`
294 /// percent past its size at the last rewrite AND is at least `min_size`
295 /// bytes. In `Background` reaper mode the check runs on the reaper tick;
296 /// in `Manual` mode it runs when you call [`crate::Store::tick`]. Pass
297 /// `pct = 0` to disable auto-rewrite (you can still call
298 /// [`crate::Store::rewrite_aof`] yourself). Defaults: 100 % / 64 MiB.
299 #[cfg(feature = "persist")]
300 #[must_use]
301 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
302 self.auto_aof_rewrite_pct = pct;
303 self.auto_aof_rewrite_min_size = min_size;
304 self
305 }
306
307 /// Shard the keyspace into `n` shared-nothing partitions (`hash(key) % n`),
308 /// each with its own lock + keyspace + AOF, so concurrent access scales
309 /// across cores. `n` clamps to ≥ 1; `1` (default) is the original
310 /// single-shard layout. Going from a single-AOF store to `n > 1`
311 /// re-shards the existing `aof-0.aof` into `aof-0..aof-{n-1}` on the next
312 /// open (the old file is backed up to `aof-0.aof.premigration.<ts>` first).
313 /// Pub/sub is process-wide (handled on shard 0), not sharded.
314 #[must_use]
315 pub fn with_shards(mut self, n: usize) -> Self {
316 self.shards = n.max(1);
317 self
318 }
319
320 /// Register a push-style metric callback. It receives a [`crate::KevyMetric`] for
321 /// each AOF replay (startup) and AOF rewrite (compaction) — wire it to
322 /// Prometheus / a log line / a counter. The callback runs synchronously on
323 /// the emitting thread (reaper thread for background rewrites), so keep it
324 /// fast and non-blocking. Replaces any previously-set sink.
325 #[cfg(feature = "persist")]
326 #[must_use]
327 pub fn with_metric_sink(
328 mut self,
329 sink: impl Fn(crate::KevyMetric) + Send + Sync + 'static,
330 ) -> Self {
331 self.metric_sink = Some(crate::metric::MetricSink::new(sink));
332 self
333 }
334
335 /// Enable transparent tiering with a RAM budget of `bytes`: values
336 /// past the demote watermark spill to the cold tier on disk
337 /// (`<data_dir>/tier/`) and page back in on access — every command
338 /// keeps its exact semantics. Requires [`Self::with_persist`]; a
339 /// memory-only store fails at open with a named error.
340 #[cfg(feature = "tier")]
341 #[must_use]
342 pub fn with_tier_budget(mut self, bytes: u64) -> Self {
343 self.tier_budget = Some(TierBudgetSpec::Bytes(bytes));
344 self
345 }
346
347 /// Enable transparent tiering with the `auto` budget: 0.70 × the
348 /// detected memory bound (cgroup v2 limit / `MemAvailable` on
349 /// Linux, `hw.memsize` on macOS), re-probed on every reaper tick.
350 /// Open fails with a named error when no bound is detectable.
351 #[cfg(feature = "tier")]
352 #[must_use]
353 pub fn with_tier_budget_auto(mut self) -> Self {
354 self.tier_budget = Some(TierBudgetSpec::Auto);
355 self
356 }
357
358 /// Cap the largest spillable value (bytes; 0 = unlimited). Bounds
359 /// the embedded cold-read lock-hold time; default 256 KiB.
360 #[must_use]
361 pub fn with_max_spill_value(mut self, bytes: u64) -> Self {
362 self.max_spill_value = bytes;
363 self
364 }
365
366 /// Enable transparent tiering with a percent-of-detected-bound
367 /// budget (`p` in 1..=100 — validated at open with a named error,
368 /// like every other config refusal).
369 #[cfg(feature = "tier")]
370 #[must_use]
371 pub fn with_tier_budget_percent(mut self, p: u8) -> Self {
372 self.tier_budget = Some(TierBudgetSpec::Percent(p));
373 self
374 }
375
376 /// Caller-driven TTL reaping — disables the background thread.
377 /// Required for WASM (no threads available). Call
378 /// [`crate::Store::tick`] yourself from your event loop.
379 #[must_use]
380 pub fn with_ttl_reaper_manual(mut self) -> Self {
381 self.ttl_reaper = TtlReaperMode::Manual;
382 self
383 }
384
385 /// Configure this store as a replication replica of `upstream`
386 /// (`"host:port"` of a kevy server's replication listener). A
387 /// background thread streams writes from the primary and applies
388 /// them locally; this store rejects local writes with a
389 /// `READONLY` error. See [`crate::Store::open_replica`] for the
390 /// convenience constructor.
391 #[cfg(feature = "replicate")]
392 #[must_use]
393 pub fn with_replica_upstream(mut self, upstream: impl Into<String>) -> Self {
394 self.replica_upstream = Some(upstream.into());
395 self
396 }
397
398 /// Override the replica identity sent to the primary at handshake.
399 /// Useful when multiple embed replicas share one primary —
400 /// otherwise they'd share the slot and stomp each other's session
401 /// state.
402 #[cfg(feature = "replicate")]
403 #[must_use]
404 pub fn with_replica_id(mut self, id: impl Into<String>) -> Self {
405 self.replica_id = id.into();
406 self
407 }
408
409 /// Override the replica reconnect backoff bounds.
410 #[cfg(feature = "replicate")]
411 #[must_use]
412 pub fn with_replica_reconnect(mut self, min: Duration, max: Duration) -> Self {
413 self.replica_reconnect_min = min;
414 self.replica_reconnect_max = max.max(min);
415 self
416 }
417
418 /// Run this store as an embed-as-writer: bind a replication
419 /// source listener on `bind_addr` so replicas can subscribe to
420 /// the writes applied here.
421 #[cfg(feature = "replicate")]
422 #[must_use]
423 pub fn with_embed_writer(mut self, bind_addr: impl Into<String>) -> Self {
424 self.embed_writer_listen_addr = Some(bind_addr.into());
425 self
426 }
427
428 /// Enable the CDC feed (`changes_since` / `changes_tail`).
429 /// `buffer_size` = 0 keeps the 64 MB default; values cap at 1 GB.
430 #[cfg(feature = "replicate")]
431 #[must_use]
432 pub fn with_feed(mut self, buffer_size: u64) -> Self {
433 self.feed_enabled = true;
434 if buffer_size > 0 {
435 self.feed_buffer_size = buffer_size.min(1024 * 1024 * 1024);
436 }
437 self
438 }
439
440 /// Override the embed-as-writer backlog byte budget.
441 #[cfg(feature = "replicate")]
442 #[must_use]
443 pub fn with_embed_writer_backlog(mut self, bytes: usize) -> Self {
444 self.embed_writer_backlog_bytes = bytes.max(64 * 1024);
445 self
446 }
447
448 /// Override the background reaper interval. Default 100 ms.
449 #[must_use]
450 pub fn with_reaper_interval(mut self, iv: Duration) -> Self {
451 self.reaper_interval = iv;
452 self
453 }
454
455}
456
457#[cfg(test)]
458mod tests {
459 use super::*;
460
461 #[test]
462 fn default_is_pure_in_memory() {
463 let c = Config::default();
464 assert_eq!(c.maxmemory, 0);
465 assert!(c.data_dir.is_none());
466 assert_eq!(c.ttl_reaper, TtlReaperMode::Background);
467 assert!(c.aof);
468 }
469
470 #[test]
471 fn builder_chains() {
472 let c = Config::default()
473 .with_persist("/tmp/foo")
474 .with_max_memory(1024)
475 .with_eviction(EvictionPolicy::AllKeysLru)
476 .with_ttl_reaper_manual()
477 .with_appendfsync(AppendFsync::Always);
478 assert_eq!(c.data_dir.as_deref(), Some(std::path::Path::new("/tmp/foo")));
479 assert_eq!(c.maxmemory, 1024);
480 assert_eq!(c.eviction_policy, EvictionPolicy::AllKeysLru);
481 assert_eq!(c.ttl_reaper, TtlReaperMode::Manual);
482 }
483
484 #[test]
485 fn without_aof_disables_logging_path() {
486 let c = Config::default().with_persist("/tmp/foo").without_aof();
487 assert!(c.data_dir.is_some());
488 assert!(!c.aof);
489 }
490}