kmp_adapter_embedded/adapter/
migration.rs

1//! Compatibility API for store-layout migration receipts.
2//!
3//! Current KMP creates SQLite format 4 and upgrades format 3 in place on open.
4//! The cross-directory migration API remains a compatibility surface for
5//! downstream Rust callers: a current source reports that migration is
6//! unnecessary, and an unsupported source is preserved with the same generic
7//! external export/import recovery contract as the open gate.
8
9use std::fs;
10use std::path::Path;
11
12use kmp_domain::{ContextUpdatedEvent, PortError, ProjectionMutation};
13use serde::{Deserialize, Serialize};
14
15use super::engine::{Key, Table};
16use super::format_version::{self, StorageEngine};
17use super::store::EmbeddedKernelStore;
18
19/// What a completed historical store migration recorded in its destination.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct StoreMigrationReceipt {
22    pub source_format: u32,
23    pub source_sha256: String,
24    pub destination_format: u32,
25    pub events_migrated: u64,
26    pub mutations_applied: u64,
27    pub kernel_version: String,
28}
29
30impl StoreMigrationReceipt {
31    /// Single key used by historical migration receipts.
32    pub const MIGRATION_ID: &'static str = "store-format-migration";
33}
34
35impl EmbeddedKernelStore {
36    /// Compatibility entry point. Format 3 upgrades in place when opened.
37    pub async fn migrate_data_dir<F>(
38        source_dir: &Path,
39        destination_dir: &Path,
40        derive: F,
41    ) -> Result<(Self, StoreMigrationReceipt), PortError>
42    where
43        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
44    {
45        Self::migrate_data_dir_to(source_dir, destination_dir, StorageEngine::Sqlite, derive).await
46    }
47
48    /// Compatibility entry point with an explicit destination engine.
49    pub async fn migrate_data_dir_to<F>(
50        source_dir: &Path,
51        destination_dir: &Path,
52        destination_engine: StorageEngine,
53        _derive: F,
54    ) -> Result<(Self, StoreMigrationReceipt), PortError>
55    where
56        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
57    {
58        if same_file(source_dir, destination_dir) {
59            return Err(PortError::InvalidState(
60                "migration source and destination are the same data directory".to_string(),
61            ));
62        }
63        let source_format = format_version::read_stamped_version(source_dir)?;
64        if source_format > StorageEngine::NEWEST_KNOWN_FORMAT_VERSION {
65            return Err(PortError::InvalidState(format!(
66                "migration source `{}` uses format version {source_format}, newer than this \
67                 binary supports ({}); upgrade the binary",
68                source_dir.display(),
69                StorageEngine::NEWEST_KNOWN_FORMAT_VERSION
70            )));
71        }
72        if matches!(source_format, 3 | 4) {
73            return Err(PortError::Unavailable(format!(
74                "cross-directory migration from SQLite format {source_format} is unnecessary and unsupported; open \
75                 with the current binary to upgrade format 3 in place. The \
76                 source at `{}` is left untouched",
77                source_dir.display()
78            )));
79        }
80        let _ = destination_engine;
81        Err(PortError::InvalidState(format!(
82            "migration source `{}` uses unsupported format version {source_format}; current \
83             KMP left it untouched. Old contracts are not migrated by this redesign; \
84             use a compatible binary to inspect the source and start a fresh current store",
85            source_dir.display()
86        )))
87    }
88
89    /// Reopen a completed destination or apply the compatibility migration.
90    pub async fn open_or_migrate_data_dir<F>(
91        source_dir: &Path,
92        destination_dir: &Path,
93        derive: F,
94    ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>
95    where
96        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
97    {
98        Self::open_or_migrate_data_dir_to(
99            source_dir,
100            destination_dir,
101            StorageEngine::Sqlite,
102            derive,
103        )
104        .await
105    }
106
107    /// Reopen a completed destination or apply the compatibility migration.
108    pub async fn open_or_migrate_data_dir_to<F>(
109        source_dir: &Path,
110        destination_dir: &Path,
111        destination_engine: StorageEngine,
112        derive: F,
113    ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>
114    where
115        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
116    {
117        if format_version::existing_store_file(destination_dir).is_some() {
118            let store = Self::open(destination_dir)?;
119            let receipt = store.migration_receipt().await?;
120            return Ok((store, receipt));
121        }
122        let (store, receipt) =
123            Self::migrate_data_dir_to(source_dir, destination_dir, destination_engine, derive)
124                .await?;
125        Ok((store, Some(receipt)))
126    }
127
128    /// A historical receipt stored by the migration that produced this store.
129    pub async fn migration_receipt(&self) -> Result<Option<StoreMigrationReceipt>, PortError> {
130        self.run(|store| {
131            let tx = store.begin_read()?;
132            let Some(raw) = tx.get(
133                Table::Migrations,
134                Key::Str(StoreMigrationReceipt::MIGRATION_ID),
135            )?
136            else {
137                return Ok(None);
138            };
139            let receipt = serde_json::from_slice(&raw).map_err(|error| {
140                PortError::InvalidState(format!("migration receipt is unreadable: {error}"))
141            })?;
142            Ok(Some(receipt))
143        })
144        .await
145    }
146}
147
148fn same_file(left: &Path, right: &Path) -> bool {
149    match (fs::canonicalize(left), fs::canonicalize(right)) {
150        (Ok(left), Ok(right)) => left == right,
151        _ => left == right,
152    }
153}
kmp_adapter_embedded/adapter/migration.rs

kmp_adapter_embedded/adapter/
migration.rs
