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(store: kevy_store::Store, #[cfg(feature = "persist")] aof: Option<Aof>) -> Self {
117        Inner {
118            store,
119            #[cfg(feature = "persist")]
120            aof,
121            bus: PubsubBus::new(),
122            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
123            writer_source: None,
124            #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
125            feed: None,
126            blocker: None,
127            #[cfg(feature = "index")]
128            idx_segs: crate::ops_index::ShardSegs::default(),
129            #[cfg(feature = "index")]
130            idx_reg: None,
131            #[cfg(feature = "index")]
132            view_segs: crate::ops_view::ShardViews::default(),
133            #[cfg(feature = "index")]
134            view_reg: None,
135        }
136    }
137}
138
139/// Owns the reaper-thread handle + the shards for the final AOF flush. Lives
140/// in an `Arc<DropGuard>` shared across every `Store` clone; the drop logic
141/// fires only when the last clone goes away.
142pub(crate) struct DropGuard {
143    /// Set by [`Store::shutdown`]: every later write fails with
144    /// `KevyError::Closed`. Shared across clones (it lives here so ANY
145    /// clone's shutdown gates ALL clones' writes).
146    pub(crate) shutdown: AtomicBool,
147    /// The boot replay verdict. Owned by the guard (engine lifetime),
148    /// so `WeakStore::upgrade` can rebuild a full `Store` — with the
149    /// original boot's report — even after every full handle dropped
150    /// while a subscription kept the engine alive.
151    pub(crate) open_report: Arc<crate::metric::OpenReport>,
152    /// The table registry — owned by the guard (engine lifetime) for
153    /// the same reason as `open_report`: a `WeakStore::upgrade` after
154    /// every full handle dropped must still see the declared tables.
155    #[cfg(feature = "index")]
156    pub(crate) tables: Arc<crate::ops_table::TableReg>,
157    pub(crate) reaper_stop: Option<Arc<AtomicBool>>,
158    pub(crate) reaper_join: Mutex<Option<JoinHandle<()>>>,
159    // Read by the persist flush; without it the strong ref still
160    // pins the shards until the LAST clone (incl. subscriptions) drops.
161    #[cfg_attr(not(feature = "persist"), allow(dead_code))]
162    pub(crate) shards_for_flush: Shards,
163    /// Replica runner thread + reconnect machinery, present iff this
164    /// store was opened with `Config::replica_upstream = Some(...)`.
165    /// Joined here so the runner stops cleanly when the last `Store`
166    /// clone goes away.
167    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
168    pub(crate) replica_runner: Option<crate::replica_runner::ReplicaRunner>,
169    /// Feed close-marker inputs — the feed handle + data dir,
170    /// present iff feed enabled AND persistent. Written after the AOF
171    /// flush so the marker's cursor describes durable state.
172    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
173    pub(crate) feed_close: Option<(
174        std::sync::Arc<Mutex<kevy_replicate::feed::FeedSource>>,
175        std::path::PathBuf,
176    )>,
177    /// Replica-source listener + accepted connection threads, present
178    /// iff this store is an embed-as-writer
179    /// (`Config::embed_writer_listen_addr = Some(...)`). Joined on
180    /// last-clone drop.
181    #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
182    pub(crate) replica_source: Option<crate::replica_source::ReplicaSource>,
183}
184
185impl Drop for DropGuard {
186    fn drop(&mut self) {
187        // Stop the replica runner FIRST so no more frames arrive while
188        // we're shutting down + flushing the AOF (frames would race
189        // with the shutdown path).
190        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
191        if let Some(r) = &self.replica_runner {
192            r.shutdown();
193        }
194        // Stop the writer-source accept + connection threads next, so
195        // no new replica picks up bytes mid-flush.
196        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
197        if let Some(rs) = &self.replica_source {
198            rs.shutdown();
199        }
200        // Stop + join the reaper, then flush every shard's AOF so EverySec
201        // users don't lose the last sub-second of writes.
202        if let Some(stop) = &self.reaper_stop {
203            stop.store(true, Ordering::Relaxed);
204        }
205        if let Some(j) = self
206            .reaper_join
207            .lock()
208            .unwrap_or_else(std::sync::PoisonError::into_inner)
209            .take()
210        {
211            let _ = j.join();
212        }
213        #[cfg(feature = "persist")]
214        for shard in self.shards_for_flush.iter() {
215            let mut g = lock_write(shard);
216            if let Some(aof) = &mut g.aof {
217                // Unconditional: `maybe_sync` is a no-op inside the EverySec
218                // window, which let the fsynced close marker below claim
219                // durability the AOF tail didn't have yet — a power loss in
220                // that gap resumed the cursor over a rolled-back store, the
221                // one phantom the generation fence cannot detect. Same
222                // discipline as the server's shutdown_drain.
223                let _ = aof.sync_now();
224            }
225        }
226        // With the AOF durable, record the feed continuity
227        // marker — the cursor now exactly describes on-disk state.
228        #[cfg(all(feature = "replicate", not(target_arch = "wasm32")))]
229        if let Some((feed, dir)) = &self.feed_close {
230            Store::feed_write_close_marker(feed, dir);
231        }
232    }
233}
kevy_embedded/store_inner.rs

kevy_embedded/
store_inner.rs
