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

kevy_embedded/
store_inner.rs
