kmp_adapter_embedded/adapter/
format_version.rs

1use std::fs;
2use std::io::ErrorKind;
3use std::path::{Path, PathBuf};
4
5use kmp_domain::PortError;
6
7/// The layout this binary creates for a fresh data directory, and the one it
8/// has always created: redb ([ADR-009](../../../../docs/adr/ADR-009-embedded-storage-engine.md)).
9///
10/// `FORMAT_VERSION` in a data directory names the *layout* — which engine
11/// wrote `store/`, and how. Bumping it is what makes a binary that predates
12/// a layout refuse the directory instead of opening an empty store beside
13/// it, so a new engine is a new number ([ADR-018](../../../../docs/adr/ADR-018-multi-process-embedded-store.md)).
14pub const SUPPORTED_FORMAT_VERSION: u32 = StorageEngine::Redb.format_version();
15
16/// The logical shape of the event log — what a bundle carries and what a
17/// migration translates. Independent of the engine: a redb store and a
18/// SQLite store export byte-identical bundles.
19pub const EVENT_FORMAT_VERSION: u32 = 1;
20
21const FORMAT_VERSION_FILE: &str = "FORMAT_VERSION";
22
23/// The engine behind a data directory's `store/`. Chosen once, when the
24/// directory is created; recorded as its `FORMAT_VERSION`; never guessed.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum StorageEngine {
27    /// Pure Rust, one file, single process. The default.
28    Redb,
29    /// WAL-mode SQLite: several processes may open the same store. Opt-in
30    /// through the `sqlite` cargo feature; a binary built without it still
31    /// recognises the layout, and refuses it by name.
32    Sqlite,
33}
34
35impl StorageEngine {
36    /// The `FORMAT_VERSION` this engine stamps.
37    pub const fn format_version(self) -> u32 {
38        match self {
39            StorageEngine::Redb => 1,
40            StorageEngine::Sqlite => 2,
41        }
42    }
43
44    /// The highest layout number any build of this crate knows about,
45    /// compiled in or not. Above this the binary is simply too old.
46    pub(crate) const NEWEST_KNOWN_FORMAT_VERSION: u32 = StorageEngine::Sqlite.format_version();
47
48    pub(crate) const fn from_format_version(version: u32) -> Option<Self> {
49        match version {
50            1 => Some(StorageEngine::Redb),
51            2 => Some(StorageEngine::Sqlite),
52            _ => None,
53        }
54    }
55
56    /// Whether this build can open the engine, as opposed to merely name it.
57    pub const fn is_compiled(self) -> bool {
58        match self {
59            StorageEngine::Redb => true,
60            StorageEngine::Sqlite => cfg!(feature = "sqlite"),
61        }
62    }
63
64    pub const fn name(self) -> &'static str {
65        match self {
66            StorageEngine::Redb => "redb",
67            StorageEngine::Sqlite => "sqlite",
68        }
69    }
70
71    const fn store_file_name(self) -> &'static str {
72        match self {
73            StorageEngine::Redb => "kernel.redb",
74            StorageEngine::Sqlite => "kernel.sqlite3",
75        }
76    }
77
78    const ALL: [StorageEngine; 2] = [StorageEngine::Redb, StorageEngine::Sqlite];
79}
80
81impl std::fmt::Display for StorageEngine {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.write_str(self.name())
84    }
85}
86
87pub fn format_version_path(data_dir: &Path) -> PathBuf {
88    data_dir.join(FORMAT_VERSION_FILE)
89}
90
91/// The version stamped in `data_dir`, without applying the gate.
92///
93/// The migration needs to know what it is looking at precisely when
94/// [`check_or_stamp`] would refuse to open it.
95pub(crate) fn read_stamped_version(data_dir: &Path) -> Result<u32, PortError> {
96    let version_path = format_version_path(data_dir);
97    let raw = fs::read_to_string(&version_path).map_err(|error| {
98        PortError::Unavailable(format!(
99            "could not read FORMAT_VERSION at `{}`: {error}",
100            version_path.display()
101        ))
102    })?;
103    raw.trim().parse().map_err(|_| {
104        PortError::InvalidState(format!(
105            "FORMAT_VERSION at `{}` is corrupt (`{}`)",
106            version_path.display(),
107            raw.trim()
108        ))
109    })
110}
111
112/// Where `engine` keeps its store inside `data_dir`.
113pub(crate) fn store_file_path_for(data_dir: &Path, engine: StorageEngine) -> PathBuf {
114    data_dir.join("store").join(engine.store_file_name())
115}
116
117/// Whether any engine's store file is present — the "half-initialized
118/// layout" signal used to refuse a directory with a store but no stamp.
119fn any_store_file_exists(data_dir: &Path) -> bool {
120    StorageEngine::ALL
121        .iter()
122        .any(|engine| store_file_path_for(data_dir, *engine).exists())
123}
124
125/// The store file a stamped directory points at, if the stamp names a
126/// layout this crate knows and the file is there. `None` for a fresh
127/// directory. Does not apply the gate: the migration asks this about
128/// directories it may be about to refuse.
129pub(crate) fn existing_store_file(data_dir: &Path) -> Option<(StorageEngine, PathBuf)> {
130    let version = read_stamped_version(data_dir).ok()?;
131    let engine = StorageEngine::from_format_version(version)?;
132    let path = store_file_path_for(data_dir, engine);
133    path.exists().then_some((engine, path))
134}
135
136/// Fail-fast format check per ADR-012: stamp fresh directories with the
137/// default layout, reject version mismatches and half-initialized layouts
138/// explicitly — never open a store that could silently read as empty memory.
139///
140/// Returns the engine the directory is (now) stamped for.
141pub(crate) fn check_or_stamp(data_dir: &Path) -> Result<StorageEngine, PortError> {
142    check_or_stamp_as(data_dir, None)
143}
144
145/// [`check_or_stamp`] with a say in the outcome: a fresh directory is
146/// stamped for `wanted`, and an existing one must already be `wanted` — a
147/// store is never reinterpreted as another engine's.
148pub(crate) fn check_or_stamp_as(
149    data_dir: &Path,
150    wanted: Option<StorageEngine>,
151) -> Result<StorageEngine, PortError> {
152    let version_path = format_version_path(data_dir);
153    match fs::read_to_string(&version_path) {
154        Ok(raw) => {
155            let version: u32 = raw.trim().parse().map_err(|_| {
156                PortError::InvalidState(format!(
157                    "embedded store at `{}` has a corrupt FORMAT_VERSION (`{}`); refusing to open",
158                    data_dir.display(),
159                    raw.trim()
160                ))
161            })?;
162            let stamped = resolve_stamped(data_dir, version)?;
163            if let Some(wanted) = wanted
164                && wanted != stamped
165            {
166                return Err(PortError::InvalidState(format!(
167                    "embedded store at `{}` is a {stamped} store (format version {}), not {wanted}; \
168                     a store is never reopened with another engine — to change engines, migrate it: \
169                     `kmp-mcp migrate <this-dir> <new-dir> --engine {wanted}`, or unset the engine \
170                     to open it as it is",
171                    data_dir.display(),
172                    stamped.format_version()
173                )));
174            }
175            Ok(stamped)
176        }
177        Err(error) if error.kind() == ErrorKind::NotFound => {
178            if any_store_file_exists(data_dir) {
179                return Err(PortError::InvalidState(format!(
180                    "embedded store at `{}` has a store file but no FORMAT_VERSION; the data \
181                     directory layout is corrupt, refusing to open",
182                    data_dir.display()
183                )));
184            }
185            let engine = wanted.unwrap_or(StorageEngine::Redb);
186            require_compiled(data_dir, engine)?;
187            fs::write(&version_path, format!("{}\n", engine.format_version())).map_err(
188                |error| {
189                    PortError::Unavailable(format!(
190                        "embedded store could not stamp FORMAT_VERSION at `{}`: {error}",
191                        version_path.display()
192                    ))
193                },
194            )?;
195            Ok(engine)
196        }
197        Err(error) => Err(PortError::Unavailable(format!(
198            "embedded store could not read FORMAT_VERSION at `{}`: {error}",
199            version_path.display()
200        ))),
201    }
202}
203
204/// Maps a stamped number to an engine this build can open, or says exactly
205/// why not: too old to migrate, too new for the binary, or known but not
206/// compiled in.
207fn resolve_stamped(data_dir: &Path, version: u32) -> Result<StorageEngine, PortError> {
208    if version > StorageEngine::NEWEST_KNOWN_FORMAT_VERSION {
209        return Err(PortError::InvalidState(format!(
210            "embedded store at `{}` uses format version {version}, newer than this \
211             binary supports ({}); upgrade the binary",
212            data_dir.display(),
213            StorageEngine::NEWEST_KNOWN_FORMAT_VERSION
214        )));
215    }
216    let Some(engine) = StorageEngine::from_format_version(version) else {
217        return Err(PortError::InvalidState(format!(
218            "embedded store at `{}` uses format version {version}, older than this \
219             binary supports ({SUPPORTED_FORMAT_VERSION}); migrate it with \
220             `kmp-mcp migrate <this-dir> <new-dir>` — the source is left untouched",
221            data_dir.display()
222        )));
223    };
224    require_compiled(data_dir, engine)?;
225    Ok(engine)
226}
227
228fn require_compiled(data_dir: &Path, engine: StorageEngine) -> Result<(), PortError> {
229    if engine.is_compiled() {
230        return Ok(());
231    }
232    Err(PortError::Unavailable(format!(
233        "embedded store at `{}` uses the {engine} engine (format version {}), which this \
234         binary was built without; rebuild with `--features {engine}`, or open it with a \
235         build that has it",
236        data_dir.display(),
237        engine.format_version()
238    )))
239}
240
241#[cfg(test)]
242mod tests {
243    use super::*;
244
245    #[test]
246    fn fresh_directory_is_stamped_with_supported_version() {
247        let dir = tempfile::tempdir().expect("tempdir");
248
249        let engine = check_or_stamp(dir.path()).expect("fresh directory should stamp");
250
251        assert_eq!(engine, StorageEngine::Redb);
252        let stamped = fs::read_to_string(format_version_path(dir.path())).expect("read stamp");
253        assert_eq!(stamped.trim(), SUPPORTED_FORMAT_VERSION.to_string());
254        check_or_stamp(dir.path()).expect("stamped directory should reopen");
255    }
256
257    #[test]
258    fn newer_format_version_fails_fast() {
259        let dir = tempfile::tempdir().expect("tempdir");
260        fs::write(format_version_path(dir.path()), "999\n").expect("write");
261
262        let error = check_or_stamp(dir.path()).expect_err("newer version must fail");
263        assert!(error.to_string().contains("upgrade the binary"));
264    }
265
266    #[test]
267    fn older_format_version_requires_migration() {
268        let dir = tempfile::tempdir().expect("tempdir");
269        fs::write(format_version_path(dir.path()), "0\n").expect("write");
270
271        let error = check_or_stamp(dir.path()).expect_err("older version must fail");
272        assert!(error.to_string().contains("kmp-mcp migrate"));
273    }
274
275    #[test]
276    fn corrupt_version_content_fails_fast() {
277        let dir = tempfile::tempdir().expect("tempdir");
278        fs::write(format_version_path(dir.path()), "not-a-number\n").expect("write");
279
280        let error = check_or_stamp(dir.path()).expect_err("corrupt version must fail");
281        assert!(error.to_string().contains("corrupt FORMAT_VERSION"));
282    }
283
284    #[test]
285    fn store_without_version_stamp_is_a_corrupt_layout() {
286        let dir = tempfile::tempdir().expect("tempdir");
287        let store = store_file_path_for(dir.path(), StorageEngine::Redb);
288        fs::create_dir_all(store.parent().expect("parent")).expect("mkdir");
289        fs::write(&store, b"stub").expect("write store stub");
290
291        let error = check_or_stamp(dir.path()).expect_err("missing stamp must fail");
292        assert!(error.to_string().contains("corrupt"));
293    }
294
295    #[test]
296    fn a_store_is_never_reopened_as_another_engine() {
297        let dir = tempfile::tempdir().expect("tempdir");
298        check_or_stamp(dir.path()).expect("stamps redb");
299
300        let error = check_or_stamp_as(dir.path(), Some(StorageEngine::Sqlite))
301            .expect_err("redb store must not open as sqlite");
302        assert!(error.to_string().contains("is a redb store"));
303        assert!(error.to_string().contains("not sqlite"));
304    }
305
306    #[test]
307    fn sqlite_layout_is_named_even_when_not_compiled() {
308        let dir = tempfile::tempdir().expect("tempdir");
309        fs::write(format_version_path(dir.path()), "2\n").expect("write");
310
311        let result = check_or_stamp(dir.path());
312        if cfg!(feature = "sqlite") {
313            assert_eq!(result.expect("sqlite compiled in"), StorageEngine::Sqlite);
314        } else {
315            let error = result.expect_err("sqlite not compiled in");
316            assert!(error.to_string().contains("sqlite engine"));
317            assert!(error.to_string().contains("--features sqlite"));
318        }
319    }
320}
kmp_adapter_embedded/adapter/format_version.rs

kmp_adapter_embedded/adapter/
format_version.rs
