coven_database/
migration.rs

1//! The host's synced-schema ladder, tracked in `PRAGMA user_version`.
2//!
3//! `user_version` is a SQLite header field, so it travels inside a snapshot's
4//! byte-for-byte DB image (`VACUUM INTO`/`VACUUM` preserve it): a device that
5//! bootstraps from a snapshot inherits the writer's applied version directly, and
6//! that same number is the wire `schema_version` every changeset is stamped with.
7//! Bumping the schema is therefore adding a migration — a device cannot stamp a
8//! version it has not migrated to.
9//!
10//! This is the host's *synced* schema only. On a fresh database, the complete
11//! host ladder, sync-routing validation, and Coven bookkeeping initialization
12//! commit in one transaction. On an initialized database, pending host steps
13//! commit only when their final routing contract exactly matches the pinned
14//! contract. Coven's bookkeeping tables are not part of this ladder.
15
16use rusqlite::Connection;
17
18use crate::DbError;
19
20/// Host SQL inside one schema-migration transaction.
21///
22/// The database retains the connection and transaction boundary; migration
23/// code can transform the host schema without opening, retaining, or passing a
24/// SQLite connection.
25pub struct MigrationContext<'connection> {
26    connection: &'connection Connection,
27}
28
29impl MigrationContext<'_> {
30    fn new(connection: &Connection) -> MigrationContext<'_> {
31        MigrationContext { connection }
32    }
33
34    pub fn execute<P>(&self, sql: &str, params: P) -> rusqlite::Result<usize>
35    where
36        P: rusqlite::Params,
37    {
38        self.connection.execute(sql, params)
39    }
40
41    pub fn execute_batch(&self, sql: &str) -> rusqlite::Result<()> {
42        self.connection.execute_batch(sql)
43    }
44
45    pub fn query_row<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<T>
46    where
47        P: rusqlite::Params,
48        F: FnOnce(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
49    {
50        self.connection.query_row(sql, params, map)
51    }
52
53    pub fn query<T, P, F>(&self, sql: &str, params: P, map: F) -> rusqlite::Result<Vec<T>>
54    where
55        P: rusqlite::Params,
56        F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
57    {
58        let mut statement = self.connection.prepare(sql)?;
59        let values = statement.query_map(params, map)?.collect();
60        values
61    }
62}
63
64/// One ordered step in the host's synced-schema ladder.
65pub struct Migration {
66    /// 1-based, contiguous across the registered set. A gap, duplicate, or a set
67    /// that does not start at 1 is a startup error, not a silent skip.
68    pub version: u32,
69    /// Recorded in logs, e.g. "initial", "add_album_disc_count".
70    pub name: &'static str,
71    pub up: MigrationStep,
72}
73
74/// The boxed closure a [`MigrationStep::Run`] holds. `Fn` (not `FnOnce`) because
75/// the engine runs migrations through a `&[Migration]`, and a step runs at most
76/// once anyway. `Send + Sync` keeps `Migration` `Sync`, so `&[Migration]` is
77/// `Send`: the bootstrap paths (`join`/`restore`) hold the slice across an
78/// `.await`, and a host that spawns those futures on a multi-threaded runtime
79/// needs them to stay `Send`.
80type MigrationFn = Box<
81    dyn for<'connection> Fn(&MigrationContext<'connection>) -> Result<(), DbError> + Send + Sync,
82>;
83
84/// How a migration applies its change to the synced schema.
85pub enum MigrationStep {
86    /// A DDL batch (`CREATE` / `ALTER` / `CREATE INDEX`), e.g. `include_str!` of a
87    /// `.sql` file.
88    Sql(&'static str),
89    /// Table rebuilds and backfills that DDL alone cannot express. Invoked at most
90    /// once (only when its version is above the on-disk one). All pending steps
91    /// share the open transaction, so a failed step or routing validation rolls
92    /// the full ladder back with `user_version`.
93    Run(MigrationFn),
94}
95
96impl MigrationStep {
97    fn apply(&self, conn: &Connection) -> Result<(), DbError> {
98        match self {
99            Self::Sql(sql) => conn.execute_batch(sql.as_ref()).map_err(DbError::from),
100            Self::Run(run) => run(&MigrationContext::new(conn)),
101        }
102    }
103}
104
105impl Migration {
106    /// A migration whose `up` is a static DDL batch.
107    pub fn sql(version: u32, name: &'static str, sql: &'static str) -> Self {
108        Migration {
109            version,
110            name,
111            up: MigrationStep::Sql(sql),
112        }
113    }
114
115    /// A migration whose `up` runs a closure for rebuilds/backfills DDL cannot
116    /// express.
117    pub fn run<F>(version: u32, name: &'static str, f: F) -> Self
118    where
119        F: for<'connection> Fn(&MigrationContext<'connection>) -> Result<(), DbError>
120            + Send
121            + Sync
122            + 'static,
123    {
124        Migration {
125            version,
126            name,
127            up: MigrationStep::Run(Box::new(f)),
128        }
129    }
130}
131
132/// The top synced-schema version this binary supports: the count of registered
133/// migrations. `run_migrations` validates the set is `1..=N` contiguous, so the
134/// count is the highest version (and `0` for an empty ladder — no synced schema).
135/// The snapshot bootstrap gate compares an incoming snapshot's version against
136/// this before adopting the image, so it lives beside the ladder rather than being
137/// re-derived at each bootstrap call site.
138pub fn supported_version(migrations: &[Migration]) -> u32 {
139    migrations.len() as u32
140}
141
142/// Why running the synced-schema ladder failed. Carried as its own arm of
143/// `OpenError` at the `Database::open` boundary — not
144/// flattened into a [`DbError`] string — so the variants stay typed for the
145/// engine's own tests, the snapshot bootstrap gate, and hosts matching
146/// [`MigrationError::SchemaTooNew`] to prompt an app update.
147#[derive(Debug, thiserror::Error)]
148pub enum MigrationError {
149    /// The registered set is not exactly `1..=N` ascending with no gaps or
150    /// duplicates — a host wiring bug, surfaced at startup before any DDL runs.
151    #[error(
152        "migration at position {position} has version {found}, expected {expected}: \
153         the registered set must be contiguous 1..=N, strictly ascending"
154    )]
155    NotContiguous {
156        position: usize,
157        found: u32,
158        expected: u32,
159    },
160    /// The on-disk schema version is newer than this binary's top migration, so
161    /// this binary cannot apply a changeset (or open a snapshot image) at that
162    /// version. Covers opening a snapshot written by a newer device and an older
163    /// binary reopening a db a newer binary already migrated.
164    #[error(
165        "on-disk schema version {current} is newer than this binary supports \
166         ({supported}); update the app"
167    )]
168    SchemaTooNew { current: u32, supported: u32 },
169    /// A migration's DDL or backfill failed. Its transaction rolled back, so
170    /// `user_version` did not advance over the half-applied step.
171    #[error("migration {version} ({name}) failed: {source}")]
172    Failed {
173        version: u32,
174        name: &'static str,
175        source: Box<DbError>,
176    },
177    /// Reading or writing `user_version`, or the `BEGIN`/`COMMIT` around a step,
178    /// failed.
179    #[error("migration ledger access failed: {0}")]
180    Ledger(Box<DbError>),
181}
182
183/// Apply every pending migration on a transaction the caller already owns.
184///
185/// The database initializer uses this form so the whole pending ladder, its
186/// `user_version` advances, final routing-contract validation, and fresh Coven
187/// metadata either commit together or roll back together. This function never
188/// begins or commits a transaction; returning an error requires its caller to
189/// roll back the enclosing transaction.
190pub(crate) fn run_migrations_in_transaction(
191    conn: &Connection,
192    migrations: &[Migration],
193) -> Result<u32, MigrationError> {
194    let current = validate_registered_migrations(conn, migrations)?;
195    for migration in migrations
196        .iter()
197        .filter(|migration| migration.version > current)
198    {
199        if let Err(source) = migration.up.apply(conn) {
200            return Err(MigrationError::Failed {
201                version: migration.version,
202                name: migration.name,
203                source: Box::new(source),
204            });
205        }
206        conn.pragma_update(None, "user_version", migration.version)
207            .map_err(|error| MigrationError::Ledger(Box::new(DbError::from(error))))?;
208    }
209    read_user_version(conn)
210}
211
212fn validate_registered_migrations(
213    conn: &Connection,
214    migrations: &[Migration],
215) -> Result<u32, MigrationError> {
216    // Validate the registered set before touching the db: versions must be exactly
217    // 1, 2, …, N in order. This one check rejects a gap, a duplicate, a non-ascending
218    // pair, and a set that does not start at 1, all at once.
219    for (position, migration) in migrations.iter().enumerate() {
220        let expected = position as u32 + 1;
221        if migration.version != expected {
222            return Err(MigrationError::NotContiguous {
223                position,
224                found: migration.version,
225                expected,
226            });
227        }
228    }
229    let current = read_user_version(conn)?;
230    let top = supported_version(migrations);
231    if current > top {
232        return Err(MigrationError::SchemaTooNew {
233            current,
234            supported: top,
235        });
236    }
237    Ok(current)
238}
239
240/// Validate the ladder and check the on-disk schema is one this binary supports,
241/// **without applying anything** — the read-only counterpart of
242/// [`run_migrations_in_transaction`]
243/// for [`Database::open_read_only`](crate::Database::open_read_only).
244///
245/// Runs the same contiguity validation and the same `SchemaTooNew` refusal as
246/// [`run_migrations_in_transaction`], then returns the current
247/// `PRAGMA user_version` unchanged. A reader cannot migrate (its connection is
248/// read-only), so an on-disk version below this binary's top is left as-is: the
249/// reader reads the schema the writer left, and a writer that opens the same db
250/// migrates it forward.
251pub(crate) fn ensure_schema_supported(
252    conn: &Connection,
253    migrations: &[Migration],
254) -> Result<u32, MigrationError> {
255    validate_registered_migrations(conn, migrations)
256}
257
258/// Read the db's applied synced-schema version from `PRAGMA user_version`.
259fn read_user_version(conn: &Connection) -> Result<u32, MigrationError> {
260    conn.pragma_query_value(None, "user_version", |r| r.get::<_, i64>(0))
261        .map(|v| v as u32)
262        .map_err(|e| MigrationError::Ledger(Box::new(DbError::from(e))))
263}
264
265#[cfg(test)]
266mod tests {
267    use super::*;
268
269    /// `Migration` must stay `Send + Sync` so `&[Migration]` is `Send`. The
270    /// bootstrap paths (`join`/`restore`) hold the slice across an `.await`, and a
271    /// host that spawns those futures on a multi-threaded runtime requires them to
272    /// be `Send` — dropping `Sync` from the `Run` closure regresses that silently
273    /// (coven builds fine; the host fails to compile). This fails to compile if it
274    /// regresses.
275    #[test]
276    fn migration_types_are_send_and_sync() {
277        fn assert_send_sync<T: Send + Sync>() {}
278        assert_send_sync::<Migration>();
279        assert_send_sync::<MigrationStep>();
280    }
281
282    fn user_version(conn: &Connection) -> u32 {
283        read_user_version(conn).expect("read user_version")
284    }
285
286    fn table_exists(conn: &Connection, name: &str) -> bool {
287        conn.query_row(
288            "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
289            [name],
290            |r| r.get::<_, i64>(0),
291        )
292        .expect("query sqlite_master")
293            > 0
294    }
295
296    fn run_migrations(conn: &Connection, migrations: &[Migration]) -> Result<u32, MigrationError> {
297        let transaction = conn
298            .unchecked_transaction()
299            .expect("begin migration transaction");
300        let version = run_migrations_in_transaction(&transaction, migrations)?;
301        transaction
302            .commit()
303            .map_err(|error| MigrationError::Ledger(Box::new(DbError::from(error))))?;
304        Ok(version)
305    }
306
307    #[test]
308    fn fresh_db_applies_every_migration_and_lands_at_top() {
309        let conn = Connection::open_in_memory().expect("open");
310        let migrations = vec![
311            Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
312            Migration::sql(2, "b", "CREATE TABLE b (id TEXT PRIMARY KEY)"),
313            Migration::sql(3, "c", "CREATE TABLE c (id TEXT PRIMARY KEY)"),
314        ];
315        let version = run_migrations(&conn, &migrations).expect("run migrations");
316        assert_eq!(version, 3);
317        assert_eq!(user_version(&conn), 3);
318        for t in ["a", "b", "c"] {
319            assert!(table_exists(&conn, t), "table {t} should exist");
320        }
321    }
322
323    #[test]
324    fn reopen_with_same_list_is_a_noop() {
325        let conn = Connection::open_in_memory().expect("open");
326        let migrations = || {
327            vec![
328                Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
329                Migration::sql(2, "b", "CREATE TABLE b (id TEXT PRIMARY KEY)"),
330                Migration::sql(3, "c", "CREATE TABLE c (id TEXT PRIMARY KEY)"),
331            ]
332        };
333        assert_eq!(run_migrations(&conn, &migrations()).expect("first"), 3);
334        // The second run finds current == N: every DDL above is a `CREATE TABLE`
335        // that would fail if re-run, so a no-op proves nothing re-executed.
336        assert_eq!(run_migrations(&conn, &migrations()).expect("second"), 3);
337        assert_eq!(user_version(&conn), 3);
338    }
339
340    #[test]
341    fn run_backfill_mutates_rows_and_bumps_version_together() {
342        let conn = Connection::open_in_memory().expect("open");
343        let migrations = vec![
344            Migration::sql(
345                1,
346                "create",
347                "CREATE TABLE t (id TEXT PRIMARY KEY, n INTEGER NOT NULL)",
348            ),
349            Migration::run(2, "backfill", |conn| {
350                conn.execute("INSERT INTO t (id, n) VALUES ('row', 1)", [])
351                    .map_err(DbError::from)?;
352                conn.execute("UPDATE t SET n = 42 WHERE id = 'row'", [])
353                    .map_err(DbError::from)?;
354                Ok(())
355            }),
356        ];
357        let version = run_migrations(&conn, &migrations).expect("run migrations");
358        assert_eq!(version, 2);
359        let n: i64 = conn
360            .query_row("SELECT n FROM t WHERE id = 'row'", [], |r| r.get(0))
361            .expect("read backfilled row");
362        assert_eq!(n, 42);
363    }
364
365    #[test]
366    fn failing_run_rolls_back_the_pending_ladder() {
367        let conn = Connection::open_in_memory().expect("open");
368        // Migration 2 fails after creating a table. The caller-owned transaction
369        // rolls back both pending steps, their DDL, and both version advances.
370        let migrations = vec![
371            Migration::sql(1, "create", "CREATE TABLE t (id TEXT PRIMARY KEY)"),
372            Migration::run(2, "boom", |conn| {
373                conn.execute("CREATE TABLE late (id TEXT PRIMARY KEY)", [])
374                    .map_err(DbError::from)?;
375                Err(DbError::Message("backfill failed".to_string()))
376            }),
377        ];
378        let err = run_migrations(&conn, &migrations).expect_err("must fail");
379        assert!(matches!(
380            err,
381            MigrationError::Failed {
382                version: 2,
383                name,
384                ..
385            } if name == "boom"
386        ));
387        assert_eq!(
388            user_version(&conn),
389            0,
390            "version must not advance over a failed pending ladder"
391        );
392        assert!(
393            !table_exists(&conn, "t"),
394            "the earlier pending step must roll back with the ladder",
395        );
396        assert!(
397            !table_exists(&conn, "late"),
398            "the failed step's DDL must have rolled back",
399        );
400    }
401
402    #[test]
403    fn malformed_sets_are_rejected_before_any_ddl() {
404        // Gap: [1, 3].
405        let conn = Connection::open_in_memory().expect("open");
406        let gap = vec![
407            Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
408            Migration::sql(3, "c", "CREATE TABLE c (id TEXT PRIMARY KEY)"),
409        ];
410        assert!(matches!(
411            run_migrations(&conn, &gap),
412            Err(MigrationError::NotContiguous {
413                position: 1,
414                found: 3,
415                expected: 2
416            })
417        ));
418        assert!(!table_exists(&conn, "a"), "no DDL runs on a malformed set");
419
420        // Duplicate: [1, 1].
421        let dup = vec![
422            Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
423            Migration::sql(1, "a2", "CREATE TABLE a2 (id TEXT PRIMARY KEY)"),
424        ];
425        assert!(matches!(
426            run_migrations(&conn, &dup),
427            Err(MigrationError::NotContiguous {
428                position: 1,
429                found: 1,
430                expected: 2
431            })
432        ));
433
434        // Not from 1: [2].
435        let not_from_one = vec![Migration::sql(
436            2,
437            "b",
438            "CREATE TABLE b (id TEXT PRIMARY KEY)",
439        )];
440        assert!(matches!(
441            run_migrations(&conn, &not_from_one),
442            Err(MigrationError::NotContiguous {
443                position: 0,
444                found: 2,
445                expected: 1
446            })
447        ));
448    }
449
450    #[test]
451    fn ensure_schema_supported_checks_without_migrating() {
452        let migrations = vec![
453            Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
454            Migration::sql(2, "b", "CREATE TABLE b (id TEXT PRIMARY KEY)"),
455        ];
456
457        // A db at version 1 (an older schema than this 2-step binary) is supported and
458        // read as-is: no migration runs, no table is created, the version is unchanged.
459        let conn = Connection::open_in_memory().expect("open");
460        conn.pragma_update(None, "user_version", 1u32)
461            .expect("set user_version");
462        assert_eq!(
463            ensure_schema_supported(&conn, &migrations).expect("v1 is supported"),
464            1
465        );
466        assert!(
467            !table_exists(&conn, "b"),
468            "ensure_schema_supported must not apply any migration",
469        );
470        assert_eq!(user_version(&conn), 1, "the version is left untouched");
471
472        // A db a newer binary migrated past this one is refused with the matchable
473        // variant — same policy as run_migrations, so a reader can prompt "update the app".
474        let ahead = Connection::open_in_memory().expect("open");
475        ahead
476            .pragma_update(None, "user_version", 5u32)
477            .expect("set user_version");
478        assert!(matches!(
479            ensure_schema_supported(&ahead, &migrations),
480            Err(MigrationError::SchemaTooNew {
481                current: 5,
482                supported: 2
483            })
484        ));
485    }
486
487    #[test]
488    fn schema_newer_than_binary_is_refused() {
489        let conn = Connection::open_in_memory().expect("open");
490        // A db migrated to version 5 by a newer binary, reopened with a 3-migration
491        // list: the engine refuses rather than apply a schema it does not know.
492        conn.pragma_update(None, "user_version", 5u32)
493            .expect("set user_version");
494        let migrations = vec![
495            Migration::sql(1, "a", "CREATE TABLE a (id TEXT PRIMARY KEY)"),
496            Migration::sql(2, "b", "CREATE TABLE b (id TEXT PRIMARY KEY)"),
497            Migration::sql(3, "c", "CREATE TABLE c (id TEXT PRIMARY KEY)"),
498        ];
499        assert!(matches!(
500            run_migrations(&conn, &migrations),
501            Err(MigrationError::SchemaTooNew {
502                current: 5,
503                supported: 3
504            })
505        ));
506    }
507}
coven_database/migration.rs

coven_database/
migration.rs
