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

kmp_adapter_embedded/adapter/
format_version.rs
