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

coven_core/
migration.rs
