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    /// Freeze all cloned read ports at one SQLite snapshot for this operation.
70    /// Dropping the last clone releases it; it cannot be used for writes.
71    pub async fn read_snapshot(&self) -> Result<Self, PortError> {
72        self.run(|store| {
73            Ok(Self {
74                engine: store.engine.read_snapshot()?,
75            })
76        })
77        .await
78    }
79
80    pub(crate) fn read_revision(&self) -> Option<kmp_domain::GraphReadRevision> {
81        self.engine.graph_read_revision()
82    }
83
84    pub(crate) fn begin_write(&self) -> Result<Box<dyn WriteTx + '_>, PortError> {
85        self.engine.begin_write()
86    }
87
88    pub(crate) fn begin_read(&self) -> Result<Box<dyn ReadTx + '_>, PortError> {
89        self.engine.begin_read()
90    }
91
92    /// Runs blocking engine work on the blocking thread pool so port calls
93    /// never stall the async executor on fsync.
94    pub(crate) async fn run<T, F>(&self, task: F) -> Result<T, PortError>
95    where
96        T: Send + 'static,
97        F: FnOnce(&EmbeddedKernelStore) -> Result<T, PortError> + Send + 'static,
98    {
99        let store = self.clone();
100        tokio::task::spawn_blocking(move || task(&store))
101            .await
102            .map_err(|error| {
103                PortError::Unavailable(format!("embedded store worker failed: {error}"))
104            })?
105    }
106
107    /// Number of events in the append-only log and the highest sequence —
108    /// audit surface used by recovery checks and operational tooling.
109    pub async fn event_log_stats(&self) -> Result<(u64, u64), PortError> {
110        self.run(|store| {
111            let tx = store.begin_read()?;
112            let count = tx.count(Table::EventLog)?;
113            let last_sequence = tx.last_u64(Table::EventLog)?.map_or(0, |(key, _)| key);
114            Ok((count, last_sequence))
115        })
116        .await
117    }
118
119    /// Compacts the store file in place, reclaiming free pages left by
120    /// past transactions (e.g. after a projection rebuild). Requires
121    /// exclusive access: call it with no other store handle open on the
122    /// same data directory.
123    pub fn compact_data_dir(data_dir: &Path) -> Result<bool, PortError> {
124        let engine = format_version::check_or_stamp(data_dir)?;
125        let store_file = format_version::store_file_path_for(data_dir, engine);
126        super::engine::sqlite::SqliteEngine::compact_file(&store_file)
127    }
128}
129
130pub(crate) fn aggregate_key(root_node_id: &str, role: &str) -> String {
131    format!("{root_node_id}\u{1f}{role}")
132}
kmp_adapter_embedded/adapter/store.rs

kmp_adapter_embedded/adapter/
store.rs
