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 /// Disable every automatic rewrite trigger — growth, absolute size
308 /// and interval — in one named call This is
309 /// the canary-window switch: the first rewrite is the documented
310 /// one-way step that upgrades a 3.x-era AOF to v2
311 /// ([`crate::Store::downgradeable_to_v3`] reads the window), so an
312 /// embedder keeping a binary-swap escape hatch open turns the
313 /// automatics off rather than remembering which of three knobs
314 /// zeroes which rule. Explicit [`crate::Store::rewrite_aof`] calls
315 /// still work — and still close the window.
316 #[cfg(feature = "persist")]
317 #[must_use]
318 pub fn with_auto_aof_rewrite_disabled(mut self) -> Self {
319 self.auto_aof_rewrite_pct = 0;
320 self.auto_aof_rewrite_bytes = 0;
321 self.auto_aof_rewrite_interval_secs = 0;
322 self
323 }
324
325 /// Shard the keyspace into `n` shared-nothing partitions (`hash(key) % n`),
326 /// each with its own lock + keyspace + AOF, so concurrent access scales
327 /// across cores. `n` clamps to ≥ 1; `1` (default) is the original
328 /// single-shard layout. Going from a single-AOF store to `n > 1`
329 /// re-shards the existing `aof-0.aof` into `aof-0..aof-{n-1}` on the next
330 /// open (the old file is backed up to `aof-0.aof.premigration.<ts>` first).
331 /// Pub/sub is process-wide (handled on shard 0), not sharded.
332 #[must_use]
333 pub fn with_shards(mut self, n: usize) -> Self {
334 self.shards = n.max(1);
335 self
336 }
337
338 /// Register a push-style metric callback. It receives a [`crate::KevyMetric`] for
339 /// each AOF replay (startup) and AOF rewrite (compaction) — wire it to
340 /// Prometheus / a log line / a counter. The callback runs synchronously on
341 /// the emitting thread (reaper thread for background rewrites), so keep it
342 /// fast and non-blocking. Replaces any previously-set sink.
343 #[cfg(feature = "persist")]
344 #[must_use]
345 pub fn with_metric_sink(
346 mut self,
347 sink: impl Fn(crate::KevyMetric) + Send + Sync + 'static,
348 ) -> Self {
349 self.metric_sink = Some(crate::metric::MetricSink::new(sink));
350 self
351 }
352
353 /// Caller-driven TTL reaping — disables the background thread.
354 /// Required for WASM (no threads available). Call
355 /// [`crate::Store::tick`] yourself from your event loop.
356 #[must_use]
357 pub fn with_ttl_reaper_manual(mut self) -> Self {
358 self.ttl_reaper = TtlReaperMode::Manual;
359 self
360 }
361
362 /// Configure this store as a replication replica of `upstream`
363 /// (`"host:port"` of a kevy server's replication listener). A
364 /// background thread streams writes from the primary and applies
365 /// them locally; this store rejects local writes with a
366 /// `READONLY` error. See [`crate::Store::open_replica`] for the
367 /// convenience constructor.
368 #[cfg(feature = "replicate")]
369 #[must_use]
370 pub fn with_replica_upstream(mut self, upstream: impl Into<String>) -> Self {
371 self.replica_upstream = Some(upstream.into());
372 self
373 }
374
375 /// Override the replica identity sent to the primary at handshake.
376 /// Useful when multiple embed replicas share one primary —
377 /// otherwise they'd share the slot and stomp each other's session
378 /// state.
379 #[cfg(feature = "replicate")]
380 #[must_use]
381 pub fn with_replica_id(mut self, id: impl Into<String>) -> Self {
382 self.replica_id = id.into();
383 self
384 }
385
386 /// Override the replica reconnect backoff bounds.
387 #[cfg(feature = "replicate")]
388 #[must_use]
389 pub fn with_replica_reconnect(mut self, min: Duration, max: Duration) -> Self {
390 self.replica_reconnect_min = min;
391 self.replica_reconnect_max = max.max(min);
392 self
393 }
394
395 /// Run this store as an embed-as-writer: bind a replication
396 /// source listener on `bind_addr` so replicas can subscribe to
397 /// the writes applied here.
398 #[cfg(feature = "replicate")]
399 #[must_use]
400 pub fn with_embed_writer(mut self, bind_addr: impl Into<String>) -> Self {
401 self.embed_writer_listen_addr = Some(bind_addr.into());
402 self
403 }
404
405 /// Enable the CDC feed (`changes_since` / `changes_tail`).
406 /// `buffer_size` = 0 keeps the 64 MB default; values cap at 1 GB.
407 #[cfg(feature = "replicate")]
408 #[must_use]
409 pub fn with_feed(mut self, buffer_size: u64) -> Self {
410 self.feed_enabled = true;
411 if buffer_size > 0 {
412 self.feed_buffer_size = buffer_size.min(1024 * 1024 * 1024);
413 }
414 self
415 }
416
417 /// Override the embed-as-writer backlog byte budget.
418 #[cfg(feature = "replicate")]
419 #[must_use]
420 pub fn with_embed_writer_backlog(mut self, bytes: usize) -> Self {
421 self.embed_writer_backlog_bytes = bytes.max(64 * 1024);
422 self
423 }
424
425 /// Override the background reaper interval. Default 100 ms.
426 #[must_use]
427 pub fn with_reaper_interval(mut self, iv: Duration) -> Self {
428 self.reaper_interval = iv;
429 self
430 }
431
432}
433
434#[cfg(test)]
435mod tests {
436 use super::*;
437
438 #[test]
439 fn default_is_pure_in_memory() {
440 let c = Config::default();
441 assert_eq!(c.maxmemory, 0);
442 assert!(c.data_dir.is_none());
443 assert_eq!(c.ttl_reaper, TtlReaperMode::Background);
444 assert!(c.aof);
445 }
446
447 #[test]
448 fn builder_chains() {
449 let c = Config::default()
450 .with_persist("/tmp/foo")
451 .with_max_memory(1024)
452 .with_eviction(EvictionPolicy::AllKeysLru)
453 .with_ttl_reaper_manual()
454 .with_appendfsync(AppendFsync::Always);
455 assert_eq!(c.data_dir.as_deref(), Some(std::path::Path::new("/tmp/foo")));
456 assert_eq!(c.maxmemory, 1024);
457 assert_eq!(c.eviction_policy, EvictionPolicy::AllKeysLru);
458 assert_eq!(c.ttl_reaper, TtlReaperMode::Manual);
459 }
460
461 #[test]
462 fn without_aof_disables_logging_path() {
463 let c = Config::default().with_persist("/tmp/foo").without_aof();
464 assert!(c.data_dir.is_some());
465 assert!(!c.aof);
466 }
467}
468
469#[path = "config_tier_builders.rs"]
470mod config_tier_builders;