kevy_embedded/config_builders.rs
1//! The builder methods of [`Config`] (child module via `#[path]`,
2//! split from `config.rs` for the 500-LOC house rule; behaviour
3//! unchanged).
4
5use super::*;
6
7impl Config {
8 /// Enable the read-only RESP listener on `addr`
9 /// (e.g. `"127.0.0.1:6009".parse().unwrap()`).
10 #[cfg(feature = "listener")]
11 #[must_use]
12 pub fn with_resp_listener(mut self, addr: std::net::SocketAddr) -> Self {
13 self.resp_listener = Some(addr);
14 self
15 }
16
17 /// Enable persistence under `dir` — snapshot file + AOF land inside.
18 /// AOF defaults on; turn it off with [`Self::without_aof`] for pure
19 /// snapshot-only durability.
20 #[cfg(feature = "persist")]
21 #[must_use]
22 pub fn with_persist(mut self, dir: impl Into<PathBuf>) -> Self {
23 self.data_dir = Some(dir.into());
24 self
25 }
26
27 /// Disable the AOF (snapshot-only persistence — explicit `save_snapshot`
28 /// calls are the only way data survives restart).
29 #[cfg(feature = "persist")]
30 #[must_use]
31 pub fn without_aof(mut self) -> Self {
32 self.aof = false;
33 self
34 }
35
36 /// Soft memory ceiling in bytes. `0` keeps the default (unlimited).
37 #[must_use]
38 pub fn with_max_memory(mut self, bytes: u64) -> Self {
39 self.maxmemory = bytes;
40 self
41 }
42
43 /// Eviction policy when over [`Self::with_max_memory`].
44 #[must_use]
45 pub fn with_eviction(mut self, policy: EvictionPolicy) -> Self {
46 self.eviction_policy = policy;
47 self
48 }
49
50 /// AOF fsync policy. Default [`AppendFsync::EverySec`].
51 #[cfg(feature = "persist")]
52 #[must_use]
53 pub fn with_appendfsync(mut self, fsync: AppendFsync) -> Self {
54 self.appendfsync = fsync;
55 self
56 }
57
58 /// Absolute-size auto-rewrite trigger: compact whenever the AOF reaches
59 /// `bytes`, regardless of growth ratio (0 = off). Complements
60 /// [`Self::with_auto_aof_rewrite`], whose growth rule is too sluggish
61 /// for long-lived instances with a large baseline.
62 #[cfg(feature = "persist")]
63 #[must_use]
64 pub fn with_auto_rewrite_bytes(mut self, bytes: u64) -> Self {
65 self.auto_aof_rewrite_bytes = bytes;
66 self
67 }
68
69 /// Time-based auto-rewrite trigger: compact at least every `interval`
70 /// while the log grows (zero duration = off).
71 #[cfg(feature = "persist")]
72 #[must_use]
73 pub fn with_auto_rewrite_interval(mut self, interval: std::time::Duration) -> Self {
74 self.auto_aof_rewrite_interval_secs = interval.as_secs();
75 self
76 }
77
78 /// Auto-`BGREWRITEAOF` thresholds: rewrite once the AOF has grown `pct`
79 /// percent past its size at the last rewrite AND is at least `min_size`
80 /// bytes. In `Background` reaper mode the check runs on the reaper tick;
81 /// in `Manual` mode it runs when you call [`crate::Store::tick`]. Pass
82 /// `pct = 0` to disable auto-rewrite (you can still call
83 /// [`crate::Store::rewrite_aof`] yourself). Defaults: 100 % / 64 MiB.
84 #[cfg(feature = "persist")]
85 #[must_use]
86 pub fn with_auto_aof_rewrite(mut self, pct: u32, min_size: u64) -> Self {
87 self.auto_aof_rewrite_pct = pct;
88 self.auto_aof_rewrite_min_size = min_size;
89 self
90 }
91
92 /// Disable every automatic rewrite trigger — growth, absolute size
93 /// and interval — in one named call This is
94 /// the canary-window switch: the first rewrite is the documented
95 /// one-way step that upgrades a 3.x-era AOF to v2
96 /// ([`crate::Store::downgradeable_to_v3`] reads the window), so an
97 /// embedder keeping a binary-swap escape hatch open turns the
98 /// automatics off rather than remembering which of three knobs
99 /// zeroes which rule. Explicit [`crate::Store::rewrite_aof`] calls
100 /// still work — and still close the window.
101 #[cfg(feature = "persist")]
102 #[must_use]
103 pub fn with_auto_aof_rewrite_disabled(mut self) -> Self {
104 self.auto_aof_rewrite_pct = 0;
105 self.auto_aof_rewrite_bytes = 0;
106 self.auto_aof_rewrite_interval_secs = 0;
107 self
108 }
109
110 /// Shard the keyspace into `n` shared-nothing partitions (`hash(key) % n`),
111 /// each with its own lock + keyspace + AOF, so concurrent access scales
112 /// across cores. `n` clamps to ≥ 1; `1` (default) is the original
113 /// single-shard layout. Going from a single-AOF store to `n > 1`
114 /// re-shards the existing `aof-0.aof` into `aof-0..aof-{n-1}` on the next
115 /// open (the old file is backed up to `aof-0.aof.premigration.<ts>` first).
116 /// Pub/sub is process-wide (handled on shard 0), not sharded.
117 #[must_use]
118 pub fn with_shards(mut self, n: usize) -> Self {
119 self.shards = n.max(1);
120 self
121 }
122
123 /// Register a push-style metric callback. It receives a [`crate::KevyMetric`] for
124 /// each AOF replay (startup) and AOF rewrite (compaction) — wire it to
125 /// Prometheus / a log line / a counter. The callback runs synchronously on
126 /// the emitting thread (reaper thread for background rewrites), so keep it
127 /// fast and non-blocking. Replaces any previously-set sink.
128 #[cfg(feature = "persist")]
129 #[must_use]
130 pub fn with_metric_sink(
131 mut self,
132 sink: impl Fn(crate::KevyMetric) + Send + Sync + 'static,
133 ) -> Self {
134 self.metric_sink = Some(crate::metric::MetricSink::new(sink));
135 self
136 }
137
138 /// Caller-driven TTL reaping — disables the background thread.
139 /// Required for WASM (no threads available). Call
140 /// [`crate::Store::tick`] yourself from your event loop.
141 #[must_use]
142 pub fn with_ttl_reaper_manual(mut self) -> Self {
143 self.ttl_reaper = TtlReaperMode::Manual;
144 self
145 }
146
147 /// Configure this store as a replication replica of `upstream`
148 /// (`"host:port"` of a kevy server's replication listener). A
149 /// background thread streams writes from the primary and applies
150 /// them locally; this store rejects local writes with a
151 /// `READONLY` error. See [`crate::Store::open_replica`] for the
152 /// convenience constructor.
153 #[cfg(feature = "replicate")]
154 #[must_use]
155 pub fn with_replica_upstream(mut self, upstream: impl Into<String>) -> Self {
156 self.replica_upstream = Some(upstream.into());
157 self
158 }
159
160 /// Override the replica identity sent to the primary at handshake.
161 /// Useful when multiple embed replicas share one primary —
162 /// otherwise they'd share the slot and stomp each other's session
163 /// state.
164 #[cfg(feature = "replicate")]
165 #[must_use]
166 pub fn with_replica_id(mut self, id: impl Into<String>) -> Self {
167 self.replica_id = id.into();
168 self
169 }
170
171 /// Override the replica reconnect backoff bounds.
172 #[cfg(feature = "replicate")]
173 #[must_use]
174 pub fn with_replica_reconnect(mut self, min: Duration, max: Duration) -> Self {
175 self.replica_reconnect_min = min;
176 self.replica_reconnect_max = max.max(min);
177 self
178 }
179
180 /// Run this store as an embed-as-writer: bind a replication
181 /// source listener on `bind_addr` so replicas can subscribe to
182 /// the writes applied here.
183 #[cfg(feature = "replicate")]
184 #[must_use]
185 pub fn with_embed_writer(mut self, bind_addr: impl Into<String>) -> Self {
186 self.embed_writer_listen_addr = Some(bind_addr.into());
187 self
188 }
189
190 /// Enable the CDC feed (`changes_since` / `changes_tail`).
191 /// `buffer_size` = 0 keeps the 64 MB default; values cap at 1 GB.
192 #[cfg(feature = "replicate")]
193 #[must_use]
194 pub fn with_feed(mut self, buffer_size: u64) -> Self {
195 self.feed_enabled = true;
196 if buffer_size > 0 {
197 self.feed_buffer_size = buffer_size.min(1024 * 1024 * 1024);
198 }
199 self
200 }
201
202 /// Override the embed-as-writer backlog byte budget.
203 #[cfg(feature = "replicate")]
204 #[must_use]
205 pub fn with_embed_writer_backlog(mut self, bytes: usize) -> Self {
206 self.embed_writer_backlog_bytes = bytes.max(64 * 1024);
207 self
208 }
209
210 /// Override the background reaper interval. Default 100 ms.
211 #[must_use]
212 pub fn with_reaper_interval(mut self, iv: Duration) -> Self {
213 self.reaper_interval = iv;
214 self
215 }
216}
217
218#[cfg(test)]
219mod tests {
220 use super::*;
221
222 #[test]
223 fn default_is_pure_in_memory() {
224 let c = Config::default();
225 assert_eq!(c.maxmemory, 0);
226 assert!(c.data_dir.is_none());
227 assert_eq!(c.ttl_reaper, TtlReaperMode::Background);
228 assert!(c.aof);
229 }
230
231 #[test]
232 fn builder_chains() {
233 let c = Config::default()
234 .with_persist("/tmp/foo")
235 .with_max_memory(1024)
236 .with_eviction(EvictionPolicy::AllKeysLru)
237 .with_ttl_reaper_manual()
238 .with_appendfsync(AppendFsync::Always);
239 assert_eq!(c.data_dir.as_deref(), Some(std::path::Path::new("/tmp/foo")));
240 assert_eq!(c.maxmemory, 1024);
241 assert_eq!(c.eviction_policy, EvictionPolicy::AllKeysLru);
242 assert_eq!(c.ttl_reaper, TtlReaperMode::Manual);
243 }
244
245 #[test]
246 fn without_aof_disables_logging_path() {
247 let c = Config::default().with_persist("/tmp/foo").without_aof();
248 assert!(c.data_dir.is_some());
249 assert!(!c.aof);
250 }
251}