kmp_adapter_embedded/adapter/
store.rs

1use std::fs;
2use std::path::Path;
3use std::sync::Arc;
4
5use kmp_domain::PortError;
6
7use super::engine::redb::RedbEngine;
8use super::engine::{Engine, ReadTx, Table, WriteTx};
9use super::format_version::{self, StorageEngine};
10
11/// Every kernel persistence port on one local store.
12///
13/// The engine behind it is chosen when the data directory is created and
14/// hidden behind the storage seam
15/// ([ADR-018](../../../../docs/adr/ADR-018-multi-process-embedded-store.md)):
16/// redb by default, SQLite when asked for and compiled in. Cloning is cheap
17/// (shared engine handle). Commits are fsync-durable on both engines, so
18/// each successful port write survives `kill -9`; a crash mid-transaction
19/// loses only the in-flight transaction.
20#[derive(Debug, Clone)]
21pub struct EmbeddedKernelStore {
22    engine: Arc<dyn Engine>,
23}
24
25impl EmbeddedKernelStore {
26    /// Opens (or initializes) the store inside `data_dir`, applying the
27    /// ADR-012 fail-fast rules before touching the engine. A fresh directory
28    /// gets the default engine; an existing one opens with the engine it was
29    /// created with.
30    pub fn open(data_dir: &Path) -> Result<Self, PortError> {
31        Self::open_as(data_dir, None)
32    }
33
34    /// [`open`](Self::open) with the engine chosen: a fresh directory is
35    /// created for `engine`, and an existing one must already be `engine` —
36    /// a store is never reinterpreted as another engine's.
37    pub fn open_with_engine(data_dir: &Path, engine: StorageEngine) -> Result<Self, PortError> {
38        Self::open_as(data_dir, Some(engine))
39    }
40
41    /// The engine a data directory was created with, without opening it.
42    pub fn engine_of(data_dir: &Path) -> Result<StorageEngine, PortError> {
43        format_version::check_or_stamp_as(data_dir, None)
44    }
45
46    fn open_as(data_dir: &Path, wanted: Option<StorageEngine>) -> Result<Self, PortError> {
47        fs::create_dir_all(data_dir).map_err(|error| {
48            PortError::Unavailable(format!(
49                "embedded store could not create data dir `{}`: {error}",
50                data_dir.display()
51            ))
52        })?;
53        let engine = format_version::check_or_stamp_as(data_dir, wanted)?;
54
55        let store_file = format_version::store_file_path_for(data_dir, engine);
56        fs::create_dir_all(store_file.parent().expect("store file has a parent")).map_err(
57            |error| {
58                PortError::Unavailable(format!(
59                    "embedded store could not create store dir under `{}`: {error}",
60                    data_dir.display()
61                ))
62            },
63        )?;
64
65        Self::open_store_file(&store_file, engine)
66    }
67
68    /// Opens a bare store file, without the data-directory layout or its
69    /// format gate. Only two callers may want this: `open`, which has just
70    /// applied the gate itself, and the migration, which reads a *copy* of a
71    /// store whose format this binary refuses to open in place.
72    pub(crate) fn open_store_file(
73        store_file: &Path,
74        engine: StorageEngine,
75    ) -> Result<Self, PortError> {
76        let engine: Arc<dyn Engine> = match engine {
77            StorageEngine::Redb => Arc::new(RedbEngine::open_file(store_file)?),
78            #[cfg(feature = "sqlite")]
79            StorageEngine::Sqlite => {
80                Arc::new(super::engine::sqlite::SqliteEngine::open_file(store_file)?)
81            }
82            #[cfg(not(feature = "sqlite"))]
83            StorageEngine::Sqlite => {
84                // The format gate names this case with the feature to enable;
85                // reaching here means a caller bypassed it.
86                return Err(PortError::Unavailable(format!(
87                    "embedded store `{}` needs the sqlite engine, which this binary was built \
88                     without",
89                    store_file.display()
90                )));
91            }
92        };
93        Ok(Self { engine })
94    }
95
96    pub(crate) fn begin_write(&self) -> Result<Box<dyn WriteTx + '_>, PortError> {
97        self.engine.begin_write()
98    }
99
100    pub(crate) fn begin_read(&self) -> Result<Box<dyn ReadTx + '_>, PortError> {
101        self.engine.begin_read()
102    }
103
104    /// Runs blocking engine work on the blocking thread pool so port calls
105    /// never stall the async executor on fsync.
106    pub(crate) async fn run<T, F>(&self, task: F) -> Result<T, PortError>
107    where
108        T: Send + 'static,
109        F: FnOnce(&EmbeddedKernelStore) -> Result<T, PortError> + Send + 'static,
110    {
111        let store = self.clone();
112        tokio::task::spawn_blocking(move || task(&store))
113            .await
114            .map_err(|error| {
115                PortError::Unavailable(format!("embedded store worker failed: {error}"))
116            })?
117    }
118
119    /// Number of events in the append-only log and the highest sequence —
120    /// audit surface used by recovery checks and operational tooling.
121    pub async fn event_log_stats(&self) -> Result<(u64, u64), PortError> {
122        self.run(|store| {
123            let tx = store.begin_read()?;
124            let count = tx.count(Table::EventLog)?;
125            let last_sequence = tx.last_u64(Table::EventLog)?.map_or(0, |(key, _)| key);
126            Ok((count, last_sequence))
127        })
128        .await
129    }
130
131    /// Compacts the store file in place, reclaiming free pages left by
132    /// past transactions (e.g. after a projection rebuild). Requires
133    /// exclusive access: call it with no other store handle open on the
134    /// same data directory.
135    pub fn compact_data_dir(data_dir: &Path) -> Result<bool, PortError> {
136        let engine = format_version::check_or_stamp(data_dir)?;
137        let store_file = format_version::store_file_path_for(data_dir, engine);
138        match engine {
139            StorageEngine::Redb => RedbEngine::compact_file(&store_file),
140            #[cfg(feature = "sqlite")]
141            StorageEngine::Sqlite => super::engine::sqlite::SqliteEngine::compact_file(&store_file),
142            #[cfg(not(feature = "sqlite"))]
143            StorageEngine::Sqlite => unreachable!("the format gate refuses uncompiled engines"),
144        }
145    }
146}
147
148pub(crate) fn aggregate_key(root_node_id: &str, role: &str) -> String {
149    format!("{root_node_id}\u{1f}{role}")
150}
kmp_adapter_embedded/adapter/store.rs

kmp_adapter_embedded/adapter/
store.rs
