kmp_adapter_embedded/adapter/
store.rs

1use std::fs;
2use std::path::Path;
3use std::sync::Arc;
4
5use kmp_domain::PortError;
6use redb::{Database, TableDefinition};
7
8use super::format_version;
9
10/// Graph nodes: `node_id -> NodeRecord` (JSON).
11pub(crate) const NODES: TableDefinition<&str, &[u8]> = TableDefinition::new("nodes");
12/// Outgoing adjacency: `(source, target, relation_type) -> explanation properties` (JSON).
13pub(crate) const RELATIONS: TableDefinition<(&str, &str, &str), &[u8]> =
14    TableDefinition::new("relations_by_source");
15/// Incoming adjacency index: `(target, source, relation_type) -> ()`.
16pub(crate) const RELATIONS_BY_TARGET: TableDefinition<(&str, &str, &str), ()> =
17    TableDefinition::new("relations_by_target");
18/// Node details: `node_id -> DetailRecord` (JSON).
19pub(crate) const DETAILS: TableDefinition<&str, &[u8]> = TableDefinition::new("details");
20/// Memory anchor index: `node_id -> ()` for nodes with kind `memory_anchor`.
21pub(crate) const ANCHORS: TableDefinition<&str, ()> = TableDefinition::new("memory_anchors");
22/// Append-only context event log: `sequence -> ContextUpdatedEvent` (JSON).
23pub(crate) const EVENT_LOG: TableDefinition<u64, &[u8]> = TableDefinition::new("event_log");
24/// Aggregate heads: `"root\u{1f}role" -> AggregateRecord` (JSON).
25pub(crate) const AGGREGATES: TableDefinition<&str, &[u8]> = TableDefinition::new("aggregates");
26/// Idempotency outcomes: `key -> IdempotentOutcome` (JSON).
27pub(crate) const IDEMPOTENCY: TableDefinition<&str, &[u8]> = TableDefinition::new("idempotency");
28/// Projection-consumer dedup: `(consumer, event_id) -> ()`.
29pub(crate) const PROCESSED: TableDefinition<(&str, &str), ()> =
30    TableDefinition::new("processed_events");
31/// Projection checkpoints: `(consumer, stream) -> CheckpointRecord` (JSON).
32pub(crate) const CHECKPOINTS: TableDefinition<(&str, &str), &[u8]> =
33    TableDefinition::new("projection_checkpoints");
34/// Snapshot audit records: `(root, role) -> snapshot summary` (JSON).
35pub(crate) const SNAPSHOTS: TableDefinition<(&str, &str), &[u8]> =
36    TableDefinition::new("snapshots");
37
38/// Every kernel persistence port on one local redb file.
39///
40/// Cloning is cheap (shared database handle). Commits are fsync-durable, so
41/// each successful port write survives `kill -9`; a crash mid-transaction
42/// loses only the in-flight transaction.
43#[derive(Debug, Clone)]
44pub struct EmbeddedKernelStore {
45    database: Arc<Database>,
46}
47
48impl EmbeddedKernelStore {
49    /// Opens (or initializes) the store inside `data_dir`, applying the
50    /// ADR-012 fail-fast rules before touching the engine.
51    pub fn open(data_dir: &Path) -> Result<Self, PortError> {
52        fs::create_dir_all(data_dir).map_err(|error| {
53            PortError::Unavailable(format!(
54                "embedded store could not create data dir `{}`: {error}",
55                data_dir.display()
56            ))
57        })?;
58        format_version::check_or_stamp(data_dir)?;
59
60        let store_file = format_version::store_file_path(data_dir);
61        fs::create_dir_all(store_file.parent().expect("store file has a parent")).map_err(
62            |error| {
63                PortError::Unavailable(format!(
64                    "embedded store could not create store dir under `{}`: {error}",
65                    data_dir.display()
66                ))
67            },
68        )?;
69
70        Self::open_store_file(&store_file)
71    }
72
73    /// Opens a bare redb file, without the data-directory layout or its
74    /// format gate. Only two callers may want this: `open`, which has just
75    /// applied the gate itself, and the migration, which reads a *copy* of a
76    /// store whose format this binary refuses to open in place.
77    pub(crate) fn open_store_file(store_file: &Path) -> Result<Self, PortError> {
78        let database = Database::create(store_file).map_err(|error| {
79            PortError::Unavailable(format!(
80                "embedded store could not open `{}`: {error}",
81                store_file.display()
82            ))
83        })?;
84
85        let store = Self {
86            database: Arc::new(database),
87        };
88        store.initialize_tables()?;
89        Ok(store)
90    }
91
92    /// Creates every table up front so read transactions never race table
93    /// existence.
94    fn initialize_tables(&self) -> Result<(), PortError> {
95        let tx = self.begin_write()?;
96        {
97            tx.open_table(NODES).map_err(table_error)?;
98            tx.open_table(RELATIONS).map_err(table_error)?;
99            tx.open_table(RELATIONS_BY_TARGET).map_err(table_error)?;
100            tx.open_table(DETAILS).map_err(table_error)?;
101            tx.open_table(ANCHORS).map_err(table_error)?;
102            tx.open_table(EVENT_LOG).map_err(table_error)?;
103            tx.open_table(AGGREGATES).map_err(table_error)?;
104            tx.open_table(IDEMPOTENCY).map_err(table_error)?;
105            tx.open_table(PROCESSED).map_err(table_error)?;
106            tx.open_table(CHECKPOINTS).map_err(table_error)?;
107            tx.open_table(SNAPSHOTS).map_err(table_error)?;
108        }
109        tx.commit().map_err(commit_error)
110    }
111
112    pub(crate) fn begin_write(&self) -> Result<redb::WriteTransaction, PortError> {
113        self.database.begin_write().map_err(|error| {
114            PortError::Unavailable(format!("embedded store write transaction failed: {error}"))
115        })
116    }
117
118    pub(crate) fn begin_read(&self) -> Result<redb::ReadTransaction, PortError> {
119        use redb::ReadableDatabase;
120        self.database.begin_read().map_err(|error| {
121            PortError::Unavailable(format!("embedded store read transaction failed: {error}"))
122        })
123    }
124
125    /// Runs blocking engine work on the blocking thread pool so port calls
126    /// never stall the async executor on fsync.
127    pub(crate) async fn run<T, F>(&self, task: F) -> Result<T, PortError>
128    where
129        T: Send + 'static,
130        F: FnOnce(&EmbeddedKernelStore) -> Result<T, PortError> + Send + 'static,
131    {
132        let store = self.clone();
133        tokio::task::spawn_blocking(move || task(&store))
134            .await
135            .map_err(|error| {
136                PortError::Unavailable(format!("embedded store worker failed: {error}"))
137            })?
138    }
139
140    /// Number of events in the append-only log and the highest sequence —
141    /// audit surface used by recovery checks and operational tooling.
142    pub async fn event_log_stats(&self) -> Result<(u64, u64), PortError> {
143        self.run(|store| {
144            let tx = store.begin_read()?;
145            let log = tx.open_table(EVENT_LOG).map_err(table_error)?;
146            let mut count = 0u64;
147            let mut last_sequence = 0u64;
148            for row in redb::ReadableTable::iter(&log).map_err(range_error)? {
149                let (key, _) = row.map_err(range_error)?;
150                count += 1;
151                last_sequence = key.value();
152            }
153            Ok((count, last_sequence))
154        })
155        .await
156    }
157}
158
159pub(crate) fn aggregate_key(root_node_id: &str, role: &str) -> String {
160    format!("{root_node_id}\u{1f}{role}")
161}
162
163pub(crate) fn table_error(error: redb::TableError) -> PortError {
164    PortError::Unavailable(format!("embedded store table access failed: {error}"))
165}
166
167pub(crate) fn storage_error(error: redb::StorageError) -> PortError {
168    PortError::Unavailable(format!("embedded store storage access failed: {error}"))
169}
170
171pub(crate) fn range_error(error: impl std::fmt::Display) -> PortError {
172    PortError::Unavailable(format!("embedded store range read failed: {error}"))
173}
174
175pub(crate) fn commit_error(error: redb::CommitError) -> PortError {
176    PortError::Unavailable(format!("embedded store commit failed: {error}"))
177}
178
179impl EmbeddedKernelStore {
180    /// Compacts the store file in place, reclaiming free pages left by
181    /// past transactions (e.g. after a projection rebuild). Requires
182    /// exclusive access: call it with no other store handle open on the
183    /// same data directory.
184    pub fn compact_data_dir(data_dir: &Path) -> Result<bool, PortError> {
185        format_version::check_or_stamp(data_dir)?;
186        let store_file = format_version::store_file_path(data_dir);
187        let mut database = Database::create(&store_file).map_err(|error| {
188            PortError::Unavailable(format!(
189                "embedded store could not open `{}` for compaction: {error}",
190                store_file.display()
191            ))
192        })?;
193        database.compact().map_err(|error| {
194            PortError::Unavailable(format!(
195                "embedded store compaction failed for `{}`: {error}",
196                store_file.display()
197            ))
198        })
199    }
200}
kmp_adapter_embedded/adapter/store.rs

kmp_adapter_embedded/adapter/
store.rs
