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

kmp_adapter_embedded/adapter/
store.rs
