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::{Engine, ReadTx, Table, WriteTx};
8use super::format_version::{self, StorageEngine};
9
10/// Every kernel persistence port on one local store.
11///
12/// The engine behind it is chosen when the data directory is created and
13/// hidden behind the storage seam
14/// ([historical ADR-018](https://github.com/underpass-ai/kmp/blob/v0.5.0/archive/docs/adr/ADR-018-multi-process-embedded-store.md)):
15/// SQLite for every store this binary can open. Cloning is cheap (shared
16/// engine handle). Commits are fsync-durable, so
17/// each successful port write survives `kill -9`; a crash mid-transaction
18/// loses only the in-flight transaction.
19#[derive(Debug, Clone)]
20pub struct EmbeddedKernelStore {
21    engine: Arc<dyn Engine>,
22}
23
24impl EmbeddedKernelStore {
25    /// Opens (or initializes) the store inside `data_dir`, applying the
26    /// ADR-012 fail-fast rules before touching the engine. A fresh directory
27    /// gets the default engine; an existing one opens with the engine it was
28    /// created with.
29    pub fn open(data_dir: &Path) -> Result<Self, PortError> {
30        Self::open_as(data_dir, None)
31    }
32
33    /// [`open`](Self::open) with the engine chosen: a fresh directory is
34    /// created for SQLite, and an existing one must already be `engine` — a
35    /// store is never reinterpreted as another engine's.
36    pub fn open_with_engine(data_dir: &Path, engine: StorageEngine) -> Result<Self, PortError> {
37        Self::open_as(data_dir, Some(engine))
38    }
39
40    /// The engine a data directory was created with, without opening it.
41    pub fn engine_of(data_dir: &Path) -> Result<StorageEngine, PortError> {
42        format_version::check_or_stamp_as(data_dir, None)
43    }
44
45    fn open_as(data_dir: &Path, wanted: Option<StorageEngine>) -> Result<Self, PortError> {
46        fs::create_dir_all(data_dir).map_err(|error| {
47            PortError::Unavailable(format!(
48                "embedded store could not create data dir `{}`: {error}",
49                data_dir.display()
50            ))
51        })?;
52        let engine = format_version::check_or_stamp_as(data_dir, wanted)?;
53
54        let store_file = format_version::store_file_path_for(data_dir, engine);
55        fs::create_dir_all(store_file.parent().expect("store file has a parent")).map_err(
56            |error| {
57                PortError::Unavailable(format!(
58                    "embedded store could not create store dir under `{}`: {error}",
59                    data_dir.display()
60                ))
61            },
62        )?;
63
64        let engine: Arc<dyn Engine> =
65            Arc::new(super::engine::sqlite::SqliteEngine::open_file(&store_file)?);
66        Ok(Self { engine })
67    }
68
69    pub(crate) fn begin_write(&self) -> Result<Box<dyn WriteTx + '_>, PortError> {
70        self.engine.begin_write()
71    }
72
73    pub(crate) fn begin_read(&self) -> Result<Box<dyn ReadTx + '_>, PortError> {
74        self.engine.begin_read()
75    }
76
77    /// Runs blocking engine work on the blocking thread pool so port calls
78    /// never stall the async executor on fsync.
79    pub(crate) async fn run<T, F>(&self, task: F) -> Result<T, PortError>
80    where
81        T: Send + 'static,
82        F: FnOnce(&EmbeddedKernelStore) -> Result<T, PortError> + Send + 'static,
83    {
84        let store = self.clone();
85        tokio::task::spawn_blocking(move || task(&store))
86            .await
87            .map_err(|error| {
88                PortError::Unavailable(format!("embedded store worker failed: {error}"))
89            })?
90    }
91
92    /// Number of events in the append-only log and the highest sequence —
93    /// audit surface used by recovery checks and operational tooling.
94    pub async fn event_log_stats(&self) -> Result<(u64, u64), PortError> {
95        self.run(|store| {
96            let tx = store.begin_read()?;
97            let count = tx.count(Table::EventLog)?;
98            let last_sequence = tx.last_u64(Table::EventLog)?.map_or(0, |(key, _)| key);
99            Ok((count, last_sequence))
100        })
101        .await
102    }
103
104    /// Compacts the store file in place, reclaiming free pages left by
105    /// past transactions (e.g. after a projection rebuild). Requires
106    /// exclusive access: call it with no other store handle open on the
107    /// same data directory.
108    pub fn compact_data_dir(data_dir: &Path) -> Result<bool, PortError> {
109        let engine = format_version::check_or_stamp(data_dir)?;
110        let store_file = format_version::store_file_path_for(data_dir, engine);
111        super::engine::sqlite::SqliteEngine::compact_file(&store_file)
112    }
113}
114
115pub(crate) fn aggregate_key(root_node_id: &str, role: &str) -> String {
116    format!("{root_node_id}\u{1f}{role}")
117}
kmp_adapter_embedded/adapter/store.rs

kmp_adapter_embedded/adapter/
store.rs
