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 and report what **this run** applied.
63/// sqlx holds a session advisory lock for the length of a run and does **not** release it when a
64/// migration fails, so the pooled connection would keep it and the next run, on another connection, would
65/// wait for it forever (measured on MySQL 8.4: a hang after a failed migration). The lock belongs to the
66/// session, so after a failure the session is ended instead of returned to the pool.
67///
68/// The ledger is read before and after on the same connection, **inside the lock** (the advisory lock
69/// stacks, so sqlx taking it again is harmless). Read outside it, two simultaneous runs both see the
70/// ledger empty at the start and full at the end, and both report every version as theirs. `$exists` is a
71/// statement that counts the ledger table (0 or 1).
72macro_rules! run_on_own_connection {
73    ($pool:expr, $migrations:expr, $classify:expr, $exists:expr) => {{
74        let mut conn = $pool
75            .acquire()
76            .await
77            .map_err(crate::migrate::backend_error)?;
78        let mut failure = None;
79        let mut before = Vec::new();
80        let mut after = Vec::new();
81        'run: {
82            if let Err(e) = sqlx::migrate::Migrate::lock(&mut *conn).await {
83                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
84                break 'run;
85            }
86            match crate::migrate::ledger(&mut *conn, $exists).await {
87                Ok(v) => before = v,
88                Err(e) => {
89                    failure = Some(e);
90                    break 'run;
91                }
92            }
93            // `run_direct` is sqlx own way round "implementation of `Acquire` is not general enough".
94            if let Err(e) = crate::migrate::migrator($migrations)
95                .run_direct(None, &mut *conn, false)
96                .await
97            {
98                failure = Some(crate::migrate::map_error(e, $classify));
99                break 'run;
100            }
101            match crate::migrate::ledger(&mut *conn, $exists).await {
102                Ok(v) => after = v,
103                Err(e) => {
104                    failure = Some(e);
105                    break 'run;
106                }
107            }
108            if let Err(e) = sqlx::migrate::Migrate::unlock(&mut *conn).await {
109                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
110            }
111        }
112        match failure {
113            None => {
114                let applied = after
115                    .iter()
116                    .copied()
117                    .filter(|v| !before.contains(v))
118                    .collect();
119                Ok(persistant_relational::MigrateReport {
120                    applied,
121                    already_applied: before,
122                })
123            }
124            Some(e) => {
125                let _ = sqlx::Connection::close(conn.detach()).await;
126                Err(e)
127            }
128        }
129    }};
130}
131#[cfg(any(feature = "mysql", feature = "sqlite"))]
132pub(crate) use run_on_own_connection;
133
134/// The versions the ledger says succeeded, read on `conn`; empty if there is no ledger. `exists` is a
135/// statement that returns 1 when the ledger table is there and 0 when it is not.
136pub(crate) async fn ledger<C>(conn: &mut C, exists: &str) -> Result<Vec<i64>, MigrateError>
137where
138    C: sqlx::Connection,
139    for<'c> &'c mut C: sqlx::Executor<'c, Database = C::Database>,
140    for<'q> i64: sqlx::Decode<'q, C::Database> + sqlx::Type<C::Database>,
141    <C::Database as sqlx::Database>::Arguments: sqlx::IntoArguments<C::Database>,
142    usize: sqlx::ColumnIndex<<C::Database as sqlx::Database>::Row>,
143{
144    let there: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(exists.to_string()))
145        .fetch_one(&mut *conn)
146        .await
147        .map_err(backend_error)?;
148    if there == 0 {
149        return Ok(Vec::new());
150    }
151    sqlx::query_scalar(sqlx::AssertSqlSafe(
152        "SELECT version FROM _sqlx_migrations WHERE success ORDER BY version".to_string(),
153    ))
154    .fetch_all(&mut *conn)
155    .await
156    .map_err(backend_error)
157}
158
159impl Migrates for PgBackend {
160    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
161        run_on_own_connection!(
162            self.pool,
163            migrations,
164            classify,
165            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END"
166        )
167    }
168
169    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
170        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
171            .fetch_one(&self.pool)
172            .await
173            .map_err(backend_error)?;
174        if !exists {
175            return Ok(Vec::new());
176        }
177        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
178            .fetch_all(&self.pool)
179            .await
180            .map_err(backend_error)
181    }
182}
183
184impl PgBackend {
185    /// Apply `migrations` in version order, skipping those already applied, in the ledger
186    /// `_sqlx_migrations` of the connection's current schema. See [`Migrates::migrate`].
187    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
188        Migrates::migrate(self, migrations).await
189    }
190
191    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
192    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
193        Migrates::applied_versions(self).await
194    }
195}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
