persistant_sqlx/
migrate.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Checksummed, ordered, resumable schema migrations that **keep sqlx's ledger**.
3//!
4//! The ledger is sqlx's own `_sqlx_migrations` table, written by sqlx's own `Migrator`, so its
5//! shape and its checksums (SHA-384 of the migration text) are exactly what `sqlx::migrate!` and
6//! `sqlx migrate run` write and read. That is a constraint, not a preference: Humboldt's transfer
7//! tool compares versions *and* checksums from that table (`docs/DESIGN-KINDS.md` §5), and a
8//! database can be handled by either this crate or sqlx's CLI without a conversion.
9//!
10//! What this layer adds is a typed outcome instead of an sqlx error string, and a report of what it
11//! applied. The types and the [`Migrates`] trait live in `persistant-relational`; this module
12//! implements it for each backend, and `persistant_relational::migrate_conformance` decides what the
13//! implementation actually promises.
14
15use crate::{PgBackend, classify};
16use persistant_relational::{MigrateError, MigrateReport, Migrates, Migration, RelError};
17use sqlx::SqlSafeStr;
18use sqlx::migrate::{Migration as SqlxMigration, MigrationType, Migrator};
19
20/// The program's migrations as sqlx's own `Migrator`.
21pub(crate) fn migrator(migrations: &[Migration]) -> Migrator {
22    let sqlx_migrations: Vec<SqlxMigration> = migrations
23        .iter()
24        .map(|m| {
25            SqlxMigration::new(
26                m.version,
27                m.description.clone().into(),
28                MigrationType::Simple,
29                sqlx::AssertSqlSafe(m.sql.clone()).into_sql_str(),
30                false,
31            )
32        })
33        .collect();
34    Migrator::with_migrations(sqlx_migrations)
35}
36
37/// sqlx's migration error as a typed one; `classify` is the backend's own error classification.
38pub(crate) fn map_error(
39    e: sqlx::migrate::MigrateError,
40    classify: fn(sqlx::Error) -> RelError,
41) -> MigrateError {
42    match e {
43        sqlx::migrate::MigrateError::VersionMismatch(version) => {
44            MigrateError::ChecksumMismatch { version }
45        }
46        sqlx::migrate::MigrateError::VersionMissing(version) => {
47            MigrateError::NewerThanProgram { version }
48        }
49        sqlx::migrate::MigrateError::Dirty(version) => MigrateError::Dirty { version },
50        sqlx::migrate::MigrateError::ExecuteMigration(e, version) => MigrateError::Failed {
51            version,
52            error: classify(e),
53        },
54        other => MigrateError::Backend(other.to_string()),
55    }
56}
57
58pub(crate) fn backend_error(e: sqlx::Error) -> MigrateError {
59    MigrateError::Backend(e.to_string())
60}
61
62/// Run the migrations on one connection taken from the pool. sqlx holds a session advisory lock for the
63/// length of a run and does **not** release it when a migration fails, so the pooled connection would keep
64/// it and the next run, on another connection, would wait for it forever (measured on MySQL 8.4: a hang
65/// after a failed migration). The lock belongs to the session, so after a failure the session is ended
66/// instead of returned to the pool.
67macro_rules! run_on_own_connection {
68    ($pool:expr, $migrations:expr, $classify:expr) => {{
69        let mut conn = $pool
70            .acquire()
71            .await
72            .map_err(crate::migrate::backend_error)?;
73        // `run_direct` is sqlx own way round "implementation of `Acquire` is not general enough".
74        match crate::migrate::migrator($migrations)
75            .run_direct(None, &mut *conn, false)
76            .await
77        {
78            Ok(()) => Ok(()),
79            Err(e) => {
80                let _ = sqlx::Connection::close(conn.detach()).await;
81                Err(crate::migrate::map_error(e, $classify))
82            }
83        }
84    }};
85}
86#[cfg(any(feature = "mysql", feature = "sqlite"))]
87pub(crate) use run_on_own_connection;
88
89impl Migrates for PgBackend {
90    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
91        let before = Migrates::applied_versions(self).await?;
92        run_on_own_connection!(self.pool, migrations, classify)?;
93        let after = Migrates::applied_versions(self).await?;
94        let applied = after
95            .iter()
96            .copied()
97            .filter(|v| !before.contains(v))
98            .collect();
99        Ok(MigrateReport {
100            applied,
101            already_applied: before,
102        })
103    }
104
105    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
106        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
107            .fetch_one(&self.pool)
108            .await
109            .map_err(backend_error)?;
110        if !exists {
111            return Ok(Vec::new());
112        }
113        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
114            .fetch_all(&self.pool)
115            .await
116            .map_err(backend_error)
117    }
118}
119
120impl PgBackend {
121    /// Apply `migrations` in version order, skipping those already applied, in the ledger
122    /// `_sqlx_migrations` of the connection's current schema. See [`Migrates::migrate`].
123    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
124        Migrates::migrate(self, migrations).await
125    }
126
127    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
128    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
129        Migrates::applied_versions(self).await
130    }
131}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
