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).
72///
73/// `$guard` is `Some((take, release))` where sqlx's own lock is not a lock. MariaDB rejects the negative
74/// timeout sqlx passes to `GET_LOCK(.., -1)` and returns NULL, which sqlx ignores, so two runs overlap
75/// (measured on MariaDB 11.4: `Duplicate entry '2' for key 'PRIMARY'`). `take` is a statement returning 1
76/// once the lock is held; `release` frees it.
77macro_rules! run_on_own_connection {
78    ($pool:expr, $migrations:expr, $classify:expr, $exists:expr, $guard:expr) => {{
79        let mut conn = $pool
80            .acquire()
81            .await
82            .map_err(crate::migrate::backend_error)?;
83        let mut failure = None;
84        let mut before = Vec::new();
85        let mut after = Vec::new();
86        let guard: Option<(&str, &str)> = $guard;
87        'run: {
88            if let Some((take, _)) = guard {
89                match sqlx::query_scalar::<_, Option<i64>>(sqlx::AssertSqlSafe(take.to_string()))
90                    .fetch_one(&mut *conn)
91                    .await
92                {
93                    Ok(Some(1)) => {}
94                    Ok(other) => {
95                        failure = Some(persistant_relational::MigrateError::Backend(format!(
96                            "the migration lock was not granted: {other:?}"
97                        )));
98                        break 'run;
99                    }
100                    Err(e) => {
101                        failure = Some(crate::migrate::backend_error(e));
102                        break 'run;
103                    }
104                }
105            }
106            if let Err(e) = sqlx::migrate::Migrate::lock(&mut *conn).await {
107                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
108                break 'run;
109            }
110            match crate::migrate::ledger(&mut *conn, $exists).await {
111                Ok(v) => before = v,
112                Err(e) => {
113                    failure = Some(e);
114                    break 'run;
115                }
116            }
117            // `run_direct` is sqlx own way round "implementation of `Acquire` is not general enough".
118            if let Err(e) = crate::migrate::migrator($migrations)
119                .run_direct(None, &mut *conn, false)
120                .await
121            {
122                failure = Some(crate::migrate::map_error(e, $classify));
123                break 'run;
124            }
125            match crate::migrate::ledger(&mut *conn, $exists).await {
126                Ok(v) => after = v,
127                Err(e) => {
128                    failure = Some(e);
129                    break 'run;
130                }
131            }
132            if let Err(e) = sqlx::migrate::Migrate::unlock(&mut *conn).await {
133                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
134            }
135            if let Some((_, release)) = guard {
136                if let Err(e) = sqlx::query(sqlx::AssertSqlSafe(release.to_string()))
137                    .execute(&mut *conn)
138                    .await
139                {
140                    failure = Some(crate::migrate::backend_error(e));
141                }
142            }
143        }
144        match failure {
145            None => {
146                let applied = after
147                    .iter()
148                    .copied()
149                    .filter(|v| !before.contains(v))
150                    .collect();
151                Ok(persistant_relational::MigrateReport {
152                    applied,
153                    already_applied: before,
154                })
155            }
156            Some(e) => {
157                let _ = sqlx::Connection::close(conn.detach()).await;
158                Err(e)
159            }
160        }
161    }};
162}
163#[cfg(any(feature = "mysql", feature = "sqlite"))]
164pub(crate) use run_on_own_connection;
165
166/// The versions the ledger says succeeded, read on `conn`; empty if there is no ledger. `exists` is a
167/// statement that returns 1 when the ledger table is there and 0 when it is not.
168pub(crate) async fn ledger<C>(conn: &mut C, exists: &str) -> Result<Vec<i64>, MigrateError>
169where
170    C: sqlx::Connection,
171    for<'c> &'c mut C: sqlx::Executor<'c, Database = C::Database>,
172    for<'q> i64: sqlx::Decode<'q, C::Database> + sqlx::Type<C::Database>,
173    <C::Database as sqlx::Database>::Arguments: sqlx::IntoArguments<C::Database>,
174    usize: sqlx::ColumnIndex<<C::Database as sqlx::Database>::Row>,
175{
176    let there: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(exists.to_string()))
177        .fetch_one(&mut *conn)
178        .await
179        .map_err(backend_error)?;
180    if there == 0 {
181        return Ok(Vec::new());
182    }
183    sqlx::query_scalar(sqlx::AssertSqlSafe(
184        "SELECT version FROM _sqlx_migrations WHERE success ORDER BY version".to_string(),
185    ))
186    .fetch_all(&mut *conn)
187    .await
188    .map_err(backend_error)
189}
190
191impl Migrates for PgBackend {
192    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
193        run_on_own_connection!(
194            self.pool,
195            migrations,
196            classify,
197            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END",
198            None
199        )
200    }
201
202    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
203        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
204            .fetch_one(&self.pool)
205            .await
206            .map_err(backend_error)?;
207        if !exists {
208            return Ok(Vec::new());
209        }
210        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
211            .fetch_all(&self.pool)
212            .await
213            .map_err(backend_error)
214    }
215}
216
217impl PgBackend {
218    /// Apply `migrations` in version order, skipping those already applied, in the ledger
219    /// `_sqlx_migrations` of the connection's current schema. See [`Migrates::migrate`].
220    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
221        Migrates::migrate(self, migrations).await
222    }
223
224    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
225    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
226        Migrates::applied_versions(self).await
227    }
228}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
