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

kmp_adapter_embedded/adapter/
store.rs
