kevy_embedded/store.rs
1//! [`Store`] — the embedded entry point. Wraps `kevy_store::Store` with
2//! per-shard locks (for cross-thread access), optional AOF auto-logging, an
3//! optional background TTL reaper, and an in-process pub/sub bus.
4
5use std::io;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard, Weak};
8use std::thread::JoinHandle;
9
10use kevy_persist::{Aof, Argv};
11use kevy_store::ExpireStats;
12
13use crate::config::Config;
14use crate::pubsub::PubsubBus;
15use crate::shard::{build_shards, shard_idx};
16
17/// The keyspace shards (`hash(key) % n`), each a fully independent
18/// `kevy_store::Store` + AOF behind its own lock. `n == 1` (the default) is a
19/// one-element vec = the original single-lock store.
20pub(crate) type Shards = Arc<Vec<Arc<RwLock<Inner>>>>;
21
22/// The embedded keyspace.
23///
24/// **`Store` is `Clone`** (since v1.1.0). A clone is a cheap `Arc` bump:
25/// every clone reaches the same underlying shards + AOF + reaper + pub/sub
26/// bus. The reaper thread is joined and each shard's AOF is flushed exactly
27/// once, when the **last** clone is dropped.
28///
29/// ```
30/// use kevy_embedded::{Config, Store};
31///
32/// # fn main() -> std::io::Result<()> {
33/// let s = Store::open(Config::default().with_ttl_reaper_manual())?;
34/// let s2 = s.clone();
35/// std::thread::spawn(move || {
36/// s2.set(b"from-thread", b"v").unwrap();
37/// }).join().unwrap();
38/// assert_eq!(s.get(b"from-thread")?, Some(b"v".to_vec()));
39/// # Ok(())
40/// # }
41/// ```
42///
43/// Every method takes `&self`. Sharding (see [`Config::with_shards`]) lets a
44/// multi-threaded consumer scale across cores; pub/sub is process-wide
45/// (handled on shard 0).
46#[derive(Clone)]
47pub struct Store {
48 pub(crate) shards: Shards,
49 /// Shared drop guard: signals + joins reaper and flushes AOFs when the
50 /// LAST `Store` clone (or `Subscription`) holding a strong ref drops.
51 pub(crate) guard: Arc<DropGuard>,
52 pub(crate) config: Config,
53 /// v2.3 CDC feed handle (read API side); shards carry clones for
54 /// the write side. `None` = feed off (or wasm).
55 #[cfg(not(target_arch = "wasm32"))]
56 pub(crate) feed: Option<std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>>,
57 /// v2.4 blocking-pop wake channel (always present; writers pay one
58 /// Relaxed load while nobody blocks).
59 pub(crate) blocker: Arc<crate::ops_blocking::Blocker>,
60 /// v2.5 index registry (catalog + version).
61 pub(crate) indexes: Arc<crate::ops_index::IndexReg>,
62 /// v2.6 view registry.
63 pub(crate) views: Arc<crate::ops_view::ViewReg>,
64}
65
66/// Weak handle to a `Store` — does not keep the underlying keyspace alive.
67///
68/// Used by the URL-keyed registry in `kevy-client` so that multiple
69/// `Connection::open("mem://name")` calls share the same backing store
70/// without leaking it when all strong handles go away.
71#[derive(Clone)]
72pub struct WeakStore {
73 shards: Weak<Vec<Arc<RwLock<Inner>>>>,
74 guard: Weak<DropGuard>,
75 config: Config,
76 #[cfg(not(target_arch = "wasm32"))]
77 feed_weak: Option<std::sync::Weak<Mutex<kevy_replicate::feed::FeedSource>>>,
78 blocker_weak: Weak<crate::ops_blocking::Blocker>,
79 indexes_weak: Weak<crate::ops_index::IndexReg>,
80 views_weak: Weak<crate::ops_view::ViewReg>,
81}
82
83impl WeakStore {
84 /// Try to upgrade back to a `Store`. Returns `None` if the last strong
85 /// reference has already been dropped.
86 pub fn upgrade(&self) -> Option<Store> {
87 Some(Store {
88 shards: self.shards.upgrade()?,
89 guard: self.guard.upgrade()?,
90 config: self.config.clone(),
91 #[cfg(not(target_arch = "wasm32"))]
92 feed: self.feed_weak.as_ref().and_then(std::sync::Weak::upgrade),
93 blocker: self.blocker_weak.upgrade()?,
94 indexes: self.indexes_weak.upgrade()?,
95 views: self.views_weak.upgrade()?,
96 })
97 }
98}
99
100pub(crate) struct Inner {
101 pub(crate) store: kevy_store::Store,
102 pub(crate) aof: Option<Aof>,
103 /// Pub/sub bus. Only shard 0's is ever used (pub/sub is process-wide);
104 /// other shards carry an idle one (cheap).
105 pub(crate) bus: PubsubBus,
106 /// Shared replication source if this store is an embed-as-writer.
107 /// Every shard holds a clone of the same `Arc<Mutex<...>>` so
108 /// `commit_write` can push mutations without reaching back up
109 /// through the `DropGuard`.
110 #[cfg(not(target_arch = "wasm32"))]
111 pub(crate) writer_source:
112 Option<std::sync::Arc<Mutex<kevy_replicate::source::ReplicationSource>>>,
113 /// v2.3 CDC feed (one stream per store); every shard holds a clone
114 /// so `commit_write` pushes effects inline. `None` = feed off.
115 #[cfg(not(target_arch = "wasm32"))]
116 pub(crate) feed: Option<std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>>,
117 /// v2.4 blocking-pop wake channel clone (see `Store::blocker`).
118 pub(crate) blocker: Option<Arc<crate::ops_blocking::Blocker>>,
119 /// v2.5: this shard's index segments + the store-level registry
120 /// handle (for the commit_write hook).
121 pub(crate) idx_segs: crate::ops_index::ShardSegs,
122 pub(crate) idx_reg: Option<Arc<crate::ops_index::IndexReg>>,
123 /// v2.6: this shard's view states + registry handle.
124 pub(crate) view_segs: crate::ops_view::ShardViews,
125 pub(crate) view_reg: Option<Arc<crate::ops_view::ViewReg>>,
126}
127
128impl Inner {
129 pub(crate) fn new(store: kevy_store::Store, aof: Option<Aof>) -> Self {
130 Inner {
131 store,
132 aof,
133 bus: PubsubBus::new(),
134 #[cfg(not(target_arch = "wasm32"))]
135 writer_source: None,
136 #[cfg(not(target_arch = "wasm32"))]
137 feed: None,
138 blocker: None,
139 idx_segs: crate::ops_index::ShardSegs::default(),
140 idx_reg: None,
141 view_segs: crate::ops_view::ShardViews::default(),
142 view_reg: None,
143 }
144 }
145}
146
147/// Owns the reaper-thread handle + the shards for the final AOF flush. Lives
148/// in an `Arc<DropGuard>` shared across every `Store` clone; the drop logic
149/// fires only when the last clone goes away.
150pub(crate) struct DropGuard {
151 reaper_stop: Option<Arc<AtomicBool>>,
152 reaper_join: Mutex<Option<JoinHandle<()>>>,
153 shards_for_flush: Shards,
154 /// Replica runner thread + reconnect machinery, present iff this
155 /// store was opened with `Config::replica_upstream = Some(...)`.
156 /// Joined here so the runner stops cleanly when the last `Store`
157 /// clone goes away.
158 #[cfg(not(target_arch = "wasm32"))]
159 pub(crate) replica_runner: Option<crate::replica_runner::ReplicaRunner>,
160 /// v2.3: feed close-marker inputs — the feed handle + data dir,
161 /// present iff feed enabled AND persistent. Written after the AOF
162 /// flush so the marker's cursor describes durable state.
163 #[cfg(not(target_arch = "wasm32"))]
164 pub(crate) feed_close: Option<(
165 std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>,
166 std::path::PathBuf,
167 )>,
168 /// Replica-source listener + accepted connection threads, present
169 /// iff this store is an embed-as-writer
170 /// (`Config::embed_writer_listen_addr = Some(...)`). Joined on
171 /// last-clone drop.
172 #[cfg(not(target_arch = "wasm32"))]
173 pub(crate) replica_source: Option<crate::replica_source::ReplicaSource>,
174}
175
176impl Store {
177 /// Open an embedded keyspace per `config`.
178 ///
179 /// - Pure in-memory when `config.data_dir` is `None`.
180 /// - With persistence: each shard loads its snapshot then replays its AOF
181 /// (`config.shards > 1` re-shards a legacy single AOF on first open).
182 /// - Spawns a background TTL reaper thread when
183 /// `config.ttl_reaper == Background` (the default).
184 /// - When `config.replica_upstream = Some("host:port")`, spawns a
185 /// background thread that streams replication frames from the
186 /// named primary and applies them to this store; local writes are
187 /// rejected with `READONLY` (see [`Self::open_replica`]).
188 pub fn open(config: Config) -> io::Result<Self> {
189 Self::open_inner(config)
190 }
191
192 /// Answer one RESP request against this store using the SAME
193 /// read-only verb whitelist the embedded RESP listener serves
194 /// (`Config::with_resp_listener`). The reply is appended to `out`
195 /// as raw RESP bytes; write verbs answer `-ERR` like the listener
196 /// does. This is the programmatic face of the listener — tooling
197 /// (e.g. `kevy-cli --embed`) inspects a store without a socket.
198 #[cfg(not(target_arch = "wasm32"))]
199 pub fn dispatch_readonly(&self, argv: &[Vec<u8>], out: &mut Vec<u8>) {
200 crate::listener::verbs_dispatch(self, argv, out);
201 }
202
203 fn open_inner(config: Config) -> io::Result<Self> {
204 let shards: Shards = Arc::new(build_shards(&config)?);
205 let (reaper_stop, reaper_join) = crate::reaper::spawn_reaper(&config, &shards)?;
206 #[cfg(not(target_arch = "wasm32"))]
207 let replica_runner = crate::replica_glue::spawn_replica_runner(&config, &shards);
208 #[cfg(not(target_arch = "wasm32"))]
209 let replica_source = match config.embed_writer_listen_addr.as_ref() {
210 Some(addr) => {
211 // Snapshot provider (v3.2): freeze every shard's COW
212 // view under the source lock (so ack_offset and the
213 // frozen keyspace are one point in time — writes
214 // between the two would replay twice otherwise), then
215 // serialize outside the locks via the persist writer.
216 let shards_for_snap: Shards = Arc::clone(&shards);
217 let snapshot: crate::replica_source::SnapshotProvider =
218 Arc::new(move || crate::replica_source::freeze_and_serialize(&shards_for_snap));
219 let rs = crate::replica_source::ReplicaSource::spawn(
220 addr,
221 config.embed_writer_backlog_bytes,
222 snapshot,
223 )?;
224 // Inject the shared source Arc into every shard's
225 // Inner so `commit_write` pushes mutations into the
226 // backlog inline. Done once at open under the
227 // shard's write lock; reads of `Inner::writer_source`
228 // afterwards are uncontended.
229 let shared = rs.shared_source();
230 for shard in shards.iter() {
231 let mut g = lock_write(shard);
232 g.writer_source = Some(shared.clone());
233 }
234 Some(rs)
235 }
236 None => None,
237 };
238 #[cfg(not(target_arch = "wasm32"))]
239 let feed = Store::feed_open(&config)?;
240 #[cfg(not(target_arch = "wasm32"))]
241 if let Some(f) = &feed {
242 for shard in shards.iter() {
243 let mut g = lock_write(shard);
244 g.feed = Some(f.clone());
245 }
246 }
247 let blocker = Arc::new(crate::ops_blocking::Blocker::new());
248 let indexes = Arc::new(crate::ops_index::IndexReg::default());
249 let views = Arc::new(crate::ops_view::ViewReg::default());
250 for shard in shards.iter() {
251 let mut g = lock_write(shard);
252 g.blocker = Some(blocker.clone());
253 g.idx_reg = Some(indexes.clone());
254 g.view_reg = Some(views.clone());
255 }
256 let guard = Arc::new(DropGuard {
257 reaper_stop,
258 reaper_join: Mutex::new(reaper_join),
259 shards_for_flush: shards.clone(),
260 #[cfg(not(target_arch = "wasm32"))]
261 replica_runner,
262 #[cfg(not(target_arch = "wasm32"))]
263 feed_close: match (&feed, &config.data_dir) {
264 (Some(f), Some(d)) => Some((f.clone(), d.clone())),
265 _ => None,
266 },
267 #[cfg(not(target_arch = "wasm32"))]
268 replica_source,
269 });
270 let store = Store {
271 shards,
272 guard,
273 config,
274 #[cfg(not(target_arch = "wasm32"))]
275 feed,
276 blocker,
277 indexes,
278 views,
279 };
280 store.idx_boot();
281 store.view_boot();
282 #[cfg(not(target_arch = "wasm32"))]
283 if let Some(addr) = store.config.resp_listener {
284 crate::listener::spawn(addr, store.downgrade())?;
285 }
286 Ok(store)
287 }
288
289 /// Convenience constructor for an embed-as-read-replica store
290 /// streaming writes from `upstream` (`"host:port"` of a kevy
291 /// server's replication listener).
292 ///
293 /// The replica:
294 /// - has its local AOF force-disabled (the upstream stream is the
295 /// source of truth; replica AOF would diverge and double-apply
296 /// on restart);
297 /// - rejects every local write with a `READONLY` `io::Error`
298 /// (you can still call read APIs concurrently);
299 /// - reconnects with exponential backoff on disconnect, resuming
300 /// from the last applied offset;
301 /// - gets a process-unique `replica_id` so an open / drop / reopen
302 /// cycle within the primary's reconnect window does not look like
303 /// the same slot from the primary's POV (which would evict
304 /// backlog frames the new embed still needs from offset 0).
305 /// Override via [`Config::with_replica_id`] when you specifically
306 /// want the slot to be re-claimed across restarts.
307 ///
308 /// For full builder control (custom replica id, backoff bounds,
309 /// snapshot dir, etc.) use [`Self::open`] with
310 /// [`Config::with_replica_upstream`] + the related setters
311 /// instead.
312 #[cfg(not(target_arch = "wasm32"))]
313 pub fn open_replica(upstream: impl Into<String>) -> io::Result<Self> {
314 let cfg = Config::default()
315 .without_aof()
316 .with_replica_id(crate::replica_glue::fresh_replica_id())
317 .with_replica_upstream(upstream);
318 Self::open(cfg)
319 }
320
321 /// `true` when this store was opened against a replication
322 /// upstream — local writes are rejected with `READONLY`.
323 pub fn is_replica(&self) -> bool {
324 self.config.replica_upstream.is_some()
325 }
326
327 /// Retarget this replica at a new primary URL (`host:port`). The
328 /// runner picks up the change on its next connect — which is
329 /// forced now by `shutdown`ing the current socket clone, so the
330 /// retarget lands within `Config::replica_reconnect_min` (default
331 /// 100 ms) of this call.
332 ///
333 /// Returns `Err` with `ErrorKind::InvalidInput` when this store is
334 /// not a replica (no upstream was configured at open). Application
335 /// code typically drives this from a `kevy-elect` failover signal —
336 /// see [`docs/cluster.md`](https://github.com/goliajp/kevy/blob/develop/docs/cluster.md).
337 /// `kevy-embedded` itself stays elect-protocol-agnostic; the
338 /// integration glue lives in the application.
339 #[cfg(not(target_arch = "wasm32"))]
340 pub fn set_replica_upstream(&self, new_upstream: impl Into<String>) -> io::Result<()> {
341 if !self.is_replica() {
342 return Err(io::Error::new(
343 io::ErrorKind::InvalidInput,
344 "set_replica_upstream called on a non-replica store",
345 ));
346 }
347 let Some(runner) = self.guard.replica_runner.as_ref() else {
348 return Err(io::Error::new(
349 io::ErrorKind::InvalidInput,
350 "replica runner is not active (open was racy?)",
351 ));
352 };
353 runner.set_upstream(new_upstream.into());
354 Ok(())
355 }
356
357 /// Get a weak handle that does not keep the keyspace alive.
358 pub fn downgrade(&self) -> WeakStore {
359 WeakStore {
360 shards: Arc::downgrade(&self.shards),
361 guard: Arc::downgrade(&self.guard),
362 config: self.config.clone(),
363 #[cfg(not(target_arch = "wasm32"))]
364 feed_weak: self.feed.as_ref().map(Arc::downgrade),
365 blocker_weak: Arc::downgrade(&self.blocker),
366 indexes_weak: Arc::downgrade(&self.indexes),
367 views_weak: Arc::downgrade(&self.views),
368 }
369 }
370
371 /// The active config (a clone — modifying it has no effect on the
372 /// running store). Useful for introspection / `INFO`-style telemetry.
373 pub fn config(&self) -> &Config {
374 &self.config
375 }
376
377 // ---- escape hatches -------------------------------------------------
378
379 /// Run `f` against the underlying `kevy_store::Store` under its lock. Use
380 /// for direct access to methods this crate hasn't wrapped. The closure can
381 /// mutate, but *does not auto-log to the AOF* — call [`Self::log`] yourself
382 /// if the mutation must survive a crash.
383 ///
384 /// **Sharded stores:** this targets shard 0 only. Use [`Self::with_key`]
385 /// to reach the shard owning a specific key.
386 pub fn with<F, R>(&self, f: F) -> R
387 where
388 F: FnOnce(&mut kevy_store::Store) -> R,
389 {
390 let mut g = self.lock();
391 f(&mut g.store)
392 }
393
394 /// Like [`Self::with`] but targets the shard that owns `key`.
395 pub fn with_key<F, R>(&self, key: &[u8], f: F) -> R
396 where
397 F: FnOnce(&mut kevy_store::Store) -> R,
398 {
399 let mut g = self.wshard(key);
400 f(&mut g.store)
401 }
402
403 /// `KEYS` / `SCAN`-glob across **every shard** — the cross-shard
404 /// replacement for `with(|s| s.collect_keys(pat, lim))`, which only sees
405 /// shard 0 once sharding is on. Behaves identically to `with(...)` when
406 /// `shard_count() == 1`. `limit` bounds the *total* returned across shards.
407 /// Takes a read lock per shard (concurrent-safe).
408 pub fn collect_keys(&self, pattern: Option<&[u8]>, limit: Option<usize>) -> Vec<Vec<u8>> {
409 let mut out = Vec::new();
410 for shard in self.shards.iter() {
411 if limit.is_some_and(|l| out.len() >= l) {
412 break;
413 }
414 let remaining = limit.map(|l| l - out.len());
415 out.extend(lock_read(shard).store.collect_keys(pattern, remaining));
416 }
417 out
418 }
419
420 /// Run `f` against **each shard's** underlying `kevy_store::Store` (in
421 /// shard-index order) — the cross-shard escape hatch. The caller assembles
422 /// the merged result. Pairs with [`Self::shard_count`]. For a single key,
423 /// prefer [`Self::with_key`]; for a glob scan, prefer [`Self::collect_keys`].
424 pub fn for_each_shard<F: FnMut(&mut kevy_store::Store)>(&self, mut f: F) {
425 for shard in self.shards.iter() {
426 f(&mut lock_write(shard).store);
427 }
428 }
429
430 /// Number of keyspace shards (`== Config::shards`).
431 #[inline]
432 pub fn shard_count(&self) -> usize {
433 self.shards.len()
434 }
435
436 /// Append a raw RESP-frame argument list to the shard owning its key's
437 /// AOF. No-op when persistence is disabled.
438 pub fn log(&self, parts: &[&[u8]]) -> io::Result<()> {
439 let mut g = match parts.get(1) {
440 Some(key) => self.wshard(key),
441 None => self.lock(),
442 };
443 if let Some(aof) = &mut g.aof {
444 let argv = Argv::from(parts.iter().map(|p| p.to_vec()).collect::<Vec<_>>());
445 aof.append(&argv)?;
446 }
447 Ok(())
448 }
449
450 // ---- maintenance ----------------------------------------------------
451
452 /// Run one TTL-reaper tick across every shard. Required call cadence in
453 /// `Manual` mode (~10×/s to match Redis `hz=10`). Returns the summed stats.
454 pub fn tick(&self) -> ExpireStats {
455 let mut total = ExpireStats::default();
456 for shard in self.shards.iter() {
457 let stats = {
458 let mut g = lock_write(shard);
459 g.store.tick_expire(self.config.reaper_samples, self.config.reaper_max_rounds)
460 };
461 total.sampled += stats.sampled;
462 total.expired += stats.expired;
463 // Auto-rewrite rides the caller-driven tick in Manual mode; the
464 // non-blocking path releases the lock for the disk spill.
465 crate::reaper::concurrent_auto_rewrite(
466 shard,
467 self.config.auto_aof_rewrite_pct,
468 self.config.auto_aof_rewrite_min_size,
469 self.config.metric_sink.as_ref(),
470 );
471 }
472 total
473 }
474
475 // Durability methods (`rewrite_aof`, `save_snapshot`) live in
476 // `crate::store_persist` to keep this file under the 500-LOC
477 // project ceiling.
478 // Data-type methods live in `crate::ops` / `crate::info`.
479
480 /// Crate-internal: clone shard 0's handle for a `Subscription`'s bus.
481 pub(crate) fn inner_handle(&self) -> Arc<RwLock<Inner>> {
482 self.shards[0].clone()
483 }
484
485 /// Crate-internal: clone the shared `Arc<DropGuard>`.
486 pub(crate) fn guard_handle(&self) -> Arc<DropGuard> {
487 self.guard.clone()
488 }
489
490 fn shard_for(&self, key: &[u8]) -> &Arc<RwLock<Inner>> {
491 &self.shards[shard_idx(key, self.shards.len())]
492 }
493
494 /// Write-lock the shard owning `key`.
495 pub(crate) fn wshard(&self, key: &[u8]) -> RwLockWriteGuard<'_, Inner> {
496 lock_write(self.shard_for(key))
497 }
498
499 /// Read-lock the shard owning `key` (GET fast path — concurrent readers
500 /// across shards run in parallel).
501 pub(crate) fn rshard(&self, key: &[u8]) -> RwLockReadGuard<'_, Inner> {
502 lock_read(self.shard_for(key))
503 }
504
505 /// Write-lock shard 0 — pub/sub bus + keyless escape hatches.
506 pub(crate) fn lock(&self) -> RwLockWriteGuard<'_, Inner> {
507 lock_write(&self.shards[0])
508 }
509
510 /// Run `f` over every shard's write guard, summing a `usize` (DBSIZE etc.).
511 pub(crate) fn sum_shards<F: Fn(&mut Inner) -> usize>(&self, f: F) -> usize {
512 self.shards.iter().map(|s| f(&mut lock_write(s))).sum()
513 }
514
515 /// Run `f` over every shard's write guard, summing a `u64`.
516 pub(crate) fn sum_shards_u64<F: Fn(&mut Inner) -> u64>(&self, f: F) -> u64 {
517 self.shards.iter().map(|s| f(&mut lock_write(s))).sum()
518 }
519
520 /// Run a fallible `f` over every shard (mutating, e.g. FLUSHALL).
521 pub(crate) fn try_for_each_shard<F: FnMut(&mut Inner) -> io::Result<()>>(
522 &self,
523 mut f: F,
524 ) -> io::Result<()> {
525 for s in self.shards.iter() {
526 f(&mut lock_write(s))?;
527 }
528 Ok(())
529 }
530}
531
532
533pub(crate) use crate::store_glue::{commit_write, lock_read, lock_write, store_err};
534
535impl Drop for DropGuard {
536 fn drop(&mut self) {
537 // Stop the replica runner FIRST so no more frames arrive while
538 // we're shutting down + flushing the AOF (frames would race
539 // with the shutdown path).
540 #[cfg(not(target_arch = "wasm32"))]
541 if let Some(r) = &self.replica_runner {
542 r.shutdown();
543 }
544 // Stop the writer-source accept + connection threads next, so
545 // no new replica picks up bytes mid-flush.
546 #[cfg(not(target_arch = "wasm32"))]
547 if let Some(rs) = &self.replica_source {
548 rs.shutdown();
549 }
550 // Stop + join the reaper, then flush every shard's AOF so EverySec
551 // users don't lose the last sub-second of writes.
552 if let Some(stop) = &self.reaper_stop {
553 stop.store(true, Ordering::Relaxed);
554 }
555 if let Some(j) = self
556 .reaper_join
557 .lock()
558 .unwrap_or_else(std::sync::PoisonError::into_inner)
559 .take()
560 {
561 let _ = j.join();
562 }
563 for shard in self.shards_for_flush.iter() {
564 let mut g = lock_write(shard);
565 if let Some(aof) = &mut g.aof {
566 let _ = aof.maybe_sync();
567 }
568 }
569 // v2.3: with the AOF durable, record the feed continuity
570 // marker — the cursor now exactly describes on-disk state.
571 #[cfg(not(target_arch = "wasm32"))]
572 if let Some((feed, dir)) = &self.feed_close {
573 Store::feed_write_close_marker(feed, dir);
574 }
575 }
576}
577
578
579#[cfg(test)]
580#[path = "store_tests.rs"]
581mod tests;
582#[cfg(test)]
583#[path = "store_tests_shard.rs"]
584mod tests_shard;
585#[cfg(test)]
586#[path = "store_tests_p2.rs"]
587mod tests_p2;
588#[cfg(test)]
589#[path = "store_tests_p3.rs"]
590mod tests_p3;
591#[cfg(test)]
592#[path = "store_tests_bitmap.rs"]
593mod tests_bitmap;
594#[cfg(test)]
595#[path = "store_tests_bonus.rs"]
596mod tests_bonus;
597#[cfg(test)]
598#[path = "store_tests_scan.rs"]
599mod tests_scan;
600#[cfg(test)]
601#[path = "store_tests_atomic.rs"]
602mod tests_atomic;
603#[cfg(test)]
604#[path = "store_tests_more.rs"]
605mod tests_more;
606#[cfg(test)]
607#[path = "store_tests_keyspace.rs"]
608mod tests_keyspace;
609#[cfg(test)]
610#[path = "store_tests_atomic_all.rs"]
611mod tests_atomic_all;
612#[cfg(test)]
613#[path = "store_tests_replay_all.rs"]
614mod tests_replay_all;
615#[cfg(test)]
616#[path = "store_tests_op_table.rs"]
617mod tests_op_table;