kevy_embedded/
store_inner.rs

1//! The [`Store`]'s shared internals — per-shard [`Inner`], the
2//! last-clone [`DropGuard`], and the [`WeakStore`] handle (split out
3//! of `store.rs` to keep it under the 500-LOC project ceiling;
4//! behaviour unchanged).
5
6// A discarded fsync. A transient failure self-heals — `dirty` stays
7// set and the next tick retries — but a persistent one (full disk,
8// read-only remount, EIO) means `appendfsync everysec` has quietly
9// become "never" with nothing saying so. Open question §2.
10#![expect(
11    clippy::let_underscore_must_use,
12    reason = "a persistent fsync failure is invisible; see .claude/OPEN-QUESTIONS-6.4.md"
13)]
14
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::{Arc, Mutex, RwLock, Weak};
17use std::thread::JoinHandle;
18
19#[cfg(feature = "persist")]
20use kevy_persist::Aof;
21
22use crate::config::Config;
23use crate::pubsub::PubsubBus;
24#[cfg(feature = "persist")]
25use crate::store::lock_write;
26use crate::store::{Shards, Store};
27
28/// Weak handle to a `Store` — does not keep the underlying keyspace alive.
29///
30/// Used by the URL-keyed registry in `kevy-client` so that multiple
31/// `Connection::connect("mem://name")` calls share the same backing store
32/// without leaking it when all strong handles go away.
33#[derive(Debug, Clone)]
34pub struct WeakStore {
35    shards: Weak<Vec<Arc<RwLock<Inner>>>>,
36    guard: Weak<DropGuard>,
37    config: Config,
38    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
39    feed_weak: Option<std::sync::Weak<Mutex<kevy_replicate::feed::FeedSource>>>,
40    blocker_weak: Weak<crate::ops_blocking::Blocker>,
41    #[cfg(feature = "index")]
42    indexes_weak: Weak<crate::ops_index::IndexReg>,
43    #[cfg(feature = "index")]
44    views_weak: Weak<crate::ops_view::ViewReg>,
45}
46
47impl WeakStore {
48    /// Try to upgrade back to a `Store`. Returns `None` if the last strong
49    /// reference has already been dropped.
50    pub fn upgrade(&self) -> Option<Store> {
51        let guard = self.guard.upgrade()?;
52        Some(Store {
53            shards: self.shards.upgrade()?,
54            config: self.config.clone(),
55            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
56            feed: self.feed_weak.as_ref().and_then(std::sync::Weak::upgrade),
57            blocker: self.blocker_weak.upgrade()?,
58            #[cfg(feature = "index")]
59            indexes: self.indexes_weak.upgrade()?,
60            #[cfg(feature = "index")]
61            views: self.views_weak.upgrade()?,
62            #[cfg(feature = "index")]
63            tables: guard.tables.clone(),
64            // The report rides the DropGuard (engine lifetime), so a
65            // resurrection that outlives every full Store handle
66            // still reports the ORIGINAL boot's replay verdict.
67            open_report: guard.open_report.clone(),
68            guard,
69        })
70    }
71}
72
73impl Store {
74    /// Get a weak handle that does not keep the keyspace alive.
75    pub fn downgrade(&self) -> WeakStore {
76        WeakStore {
77            shards: Arc::downgrade(&self.shards),
78            guard: Arc::downgrade(&self.guard),
79            config: self.config.clone(),
80            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
81            feed_weak: self.feed.as_ref().map(Arc::downgrade),
82            blocker_weak: Arc::downgrade(&self.blocker),
83            #[cfg(feature = "index")]
84            indexes_weak: Arc::downgrade(&self.indexes),
85            #[cfg(feature = "index")]
86            views_weak: Arc::downgrade(&self.views),
87        }
88    }
89}
90
91#[derive(Debug)]
92pub(crate) struct Inner {
93    pub(crate) store: kevy_store::Store,
94    #[cfg(feature = "persist")]
95    pub(crate) aof: Option<Aof>,
96    /// Pub/sub bus. Only shard 0's is ever used (pub/sub is process-wide);
97    /// other shards carry an idle one (cheap).
98    pub(crate) bus: PubsubBus,
99    /// Shared replication source if this store is an embed-as-writer.
100    /// Every shard holds a clone of the same `Arc<Mutex<...>>` so
101    /// `commit_write` can push mutations without reaching back up
102    /// through the `DropGuard`.
103    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
104    pub(crate) writer_source:
105        Option<std::sync::Arc<Mutex<kevy_replicate::source::ReplicationSource>>>,
106    /// CDC feed (one stream per store); every shard holds a clone
107    /// so `commit_write` pushes effects inline. `None` = feed off.
108    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
109    pub(crate) feed: Option<std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>>,
110    /// Blocking-pop wake channel clone (see `Store::blocker`).
111    pub(crate) blocker: Option<Arc<crate::ops_blocking::Blocker>>,
112    /// This shard's index segments + the store-level registry
113    /// handle (for the commit_write hook).
114    #[cfg(feature = "index")]
115    pub(crate) idx_segs: crate::ops_index::ShardSegs,
116    #[cfg(feature = "index")]
117    pub(crate) idx_reg: Option<Arc<crate::ops_index::IndexReg>>,
118    /// This shard's view states + registry handle.
119    #[cfg(feature = "index")]
120    pub(crate) view_segs: crate::ops_view::ShardViews,
121    #[cfg(feature = "index")]
122    pub(crate) view_reg: Option<Arc<crate::ops_view::ViewReg>>,
123}
124
125impl Inner {
126    pub(crate) fn new(
127        store: kevy_store::Store,
128        #[cfg(feature = "persist")] aof: Option<Aof>,
129    ) -> Self {
130        Inner {
131            store,
132            #[cfg(feature = "persist")]
133            aof,
134            bus: PubsubBus::new(),
135            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
136            writer_source: None,
137            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
138            feed: None,
139            blocker: None,
140            #[cfg(feature = "index")]
141            idx_segs: crate::ops_index::ShardSegs::default(),
142            #[cfg(feature = "index")]
143            idx_reg: None,
144            #[cfg(feature = "index")]
145            view_segs: crate::ops_view::ShardViews::default(),
146            #[cfg(feature = "index")]
147            view_reg: None,
148        }
149    }
150}
151
152/// Owns the reaper-thread handle + the shards for the final AOF flush. Lives
153/// in an `Arc<DropGuard>` shared across every `Store` clone; the drop logic
154/// fires only when the last clone goes away.
155#[derive(Debug)]
156pub(crate) struct DropGuard {
157    /// Set by [`Store::shutdown`]: every later write fails with
158    /// `KevyError::Closed`. Shared across clones (it lives here so ANY
159    /// clone's shutdown gates ALL clones' writes).
160    pub(crate) shutdown: AtomicBool,
161    /// The boot replay verdict. Owned by the guard (engine lifetime),
162    /// so `WeakStore::upgrade` can rebuild a full `Store` — with the
163    /// original boot's report — even after every full handle dropped
164    /// while a subscription kept the engine alive.
165    pub(crate) open_report: Arc<crate::metric::OpenReport>,
166    /// The table registry — owned by the guard (engine lifetime) for
167    /// the same reason as `open_report`: a `WeakStore::upgrade` after
168    /// every full handle dropped must still see the declared tables.
169    #[cfg(feature = "index")]
170    pub(crate) tables: Arc<crate::ops_table::TableReg>,
171    pub(crate) reaper_stop: Option<Arc<AtomicBool>>,
172    pub(crate) reaper_join: Mutex<Option<JoinHandle<()>>>,
173    // Read by the persist flush; without it the strong ref still
174    // pins the shards until the LAST clone (incl. subscriptions) drops.
175    #[cfg_attr(not(feature = "persist"), allow(dead_code))]
176    pub(crate) shards_for_flush: Shards,
177    /// Replica runner thread + reconnect machinery, present iff this
178    /// store was opened with `Config::replica_upstream = Some(...)`.
179    /// Joined here so the runner stops cleanly when the last `Store`
180    /// clone goes away.
181    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
182    pub(crate) replica_runner: Option<crate::replica_runner::ReplicaRunner>,
183    /// Feed close-marker inputs — the feed handle + data dir,
184    /// present iff feed enabled AND persistent. Written after the AOF
185    /// flush so the marker's cursor describes durable state.
186    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
187    pub(crate) feed_close:
188        Option<(std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>, std::path::PathBuf)>,
189    /// Replica-source listener + accepted connection threads, present
190    /// iff this store is an embed-as-writer
191    /// (`Config::embed_writer_listen_addr = Some(...)`). Joined on
192    /// last-clone drop.
193    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
194    pub(crate) replica_source: Option<crate::replica_source::ReplicaSource>,
195    /// Exclusive claim on the persist dir (`with_persist`) — one engine
196    /// per directory; a second `Store::open` on the same dir errors
197    /// instead of interleaving appends into this engine's AOF. Held for
198    /// the engine lifetime; released here when the last clone drops
199    /// (the drop body's final AOF flush runs before fields drop).
200    #[cfg(feature = "persist")]
201    pub(crate) _dir_lock: Option<kevy_persist::DirLock>,
202}
203
204impl Drop for DropGuard {
205    fn drop(&mut self) {
206        // Stop the replica runner FIRST so no more frames arrive while
207        // we're shutting down + flushing the AOF (frames would race
208        // with the shutdown path).
209        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
210        if let Some(r) = &self.replica_runner {
211            r.shutdown();
212        }
213        // Stop the writer-source accept + connection threads next, so
214        // no new replica picks up bytes mid-flush.
215        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
216        if let Some(rs) = &self.replica_source {
217            rs.shutdown();
218        }
219        // Stop + join the reaper, then flush every shard's AOF so EverySec
220        // users don't lose the last sub-second of writes.
221        if let Some(stop) = &self.reaper_stop {
222            stop.store(true, Ordering::Relaxed);
223        }
224        if let Some(j) =
225            self.reaper_join.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take()
226        {
227            let _ = j.join();
228        }
229        #[cfg(feature = "persist")]
230        for shard in self.shards_for_flush.iter() {
231            let mut g = lock_write(shard);
232            if let Some(aof) = &mut g.aof {
233                // Unconditional: `maybe_sync` is a no-op inside the EverySec
234                // window, which let the fsynced close marker below claim
235                // durability the AOF tail didn't have yet — a power loss in
236                // that gap resumed the cursor over a rolled-back store, the
237                // one phantom the generation fence cannot detect. Same
238                // discipline as the server's shutdown_drain.
239                let _ = aof.sync_now();
240            }
241        }
242        // With the AOF durable, record the feed continuity
243        // marker — the cursor now exactly describes on-disk state.
244        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
245        if let Some((feed, dir)) = &self.feed_close {
246            Store::feed_write_close_marker(feed, dir);
247        }
248    }
249}
kevy_embedded/store_inner.rs

kevy_embedded/
store_inner.rs
