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/// Why an sqlx [`Migrator`] cannot be run by [`Migrates::migrate`] as it stands.
38///
39/// Each variant is something sqlx would honour and this crate would not: refusing is the point, so
40/// a program never believes it ran `sqlx::migrate!` and got a different run.
41#[derive(Debug, Clone, PartialEq, Eq)]
42#[non_exhaustive]
43pub enum FromSqlxError {
44    /// The migration asks to run outside a transaction (`-- no-transaction`, or `no_tx` on the
45    /// `Migrator`); [`Migration`] always runs in one.
46    NoTransaction {
47        /// The version.
48        version: i64,
49    },
50    /// The `Migrator` keeps its ledger in another table than `_sqlx_migrations`, so this crate would
51    /// read and write a different ledger than the program asked for.
52    LedgerTable {
53        /// The table the `Migrator` names.
54        table: String,
55    },
56    /// The `Migrator` was set to create a schema or to skip locking; this crate always locks and never
57    /// creates a schema.
58    Setting(&'static str),
59    /// A migration's stored checksum is not the SHA-384 of its text, so a ledger written by this
60    /// crate would differ from the one sqlx writes for the same migration.
61    Checksum {
62        /// The version.
63        version: i64,
64    },
65}
66
67impl std::fmt::Display for FromSqlxError {
68    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69        match self {
70            FromSqlxError::NoTransaction { version } => {
71                write!(f, "migration {version} runs outside a transaction")
72            }
73            FromSqlxError::LedgerTable { table } => {
74                write!(
75                    f,
76                    "the migrator keeps its ledger in `{table}`, not `_sqlx_migrations`"
77                )
78            }
79            FromSqlxError::Setting(s) => write!(f, "the migrator sets {s}, which is not supported"),
80            FromSqlxError::Checksum { version } => {
81                write!(
82                    f,
83                    "migration {version}: checksum is not the SHA-384 of its text"
84                )
85            }
86        }
87    }
88}
89
90impl std::error::Error for FromSqlxError {}
91
92/// The migrations of an sqlx [`Migrator`] (`sqlx::migrate!("./migrations")` or `Migrator::new(path)`)
93/// as [`Migration`]s for [`Migrates::migrate`], in the order sqlx would run them.
94///
95/// The ledger is shared, not converted: [`Migrates::migrate`] writes sqlx's own `_sqlx_migrations`
96/// with the same SHA-384 checksum sqlx computes from the same text, so a database migrated by
97/// `sqlx migrate run` is accepted as it is (and the other way round). Down halves of reversible
98/// migrations are dropped (this crate never undoes). Anything sqlx would do differently from this
99/// crate is an error, not a silent difference: see [`FromSqlxError`].
100pub fn migrations_from_sqlx(migrator: &Migrator) -> Result<Vec<Migration>, FromSqlxError> {
101    if migrator.table_name != "_sqlx_migrations" {
102        return Err(FromSqlxError::LedgerTable {
103            table: migrator.table_name.to_string(),
104        });
105    }
106    if migrator.no_tx {
107        return Err(FromSqlxError::Setting("no_tx"));
108    }
109    if !migrator.locking {
110        return Err(FromSqlxError::Setting("locking = false"));
111    }
112    let mut out = Vec::new();
113    for m in migrator
114        .iter()
115        .filter(|m| !m.migration_type.is_down_migration())
116    {
117        if m.no_tx {
118            return Err(FromSqlxError::NoTransaction { version: m.version });
119        }
120        let sql = sqlx::SqlStr::as_str(&m.sql);
121        let ours = Migration::new(m.version, m.description.to_string(), sql);
122        if migrator_checksum(&ours) != m.checksum.as_ref() {
123            return Err(FromSqlxError::Checksum { version: m.version });
124        }
125        out.push(ours);
126    }
127    Ok(out)
128}
129
130/// The checksum this crate's ledger will hold for `m`: computed by sqlx's own constructor.
131fn migrator_checksum(m: &Migration) -> Vec<u8> {
132    migrator(std::slice::from_ref(m))
133        .iter()
134        .next()
135        .map(|x| x.checksum.to_vec())
136        .unwrap_or_default()
137}
138
139/// sqlx's migration error as a typed one; `classify` is the backend's own error classification.
140pub(crate) fn map_error(
141    e: sqlx::migrate::MigrateError,
142    classify: fn(sqlx::Error) -> RelError,
143) -> MigrateError {
144    match e {
145        sqlx::migrate::MigrateError::VersionMismatch(version) => {
146            MigrateError::ChecksumMismatch { version }
147        }
148        sqlx::migrate::MigrateError::VersionMissing(version) => {
149            MigrateError::NewerThanProgram { version }
150        }
151        sqlx::migrate::MigrateError::Dirty(version) => MigrateError::Dirty { version },
152        sqlx::migrate::MigrateError::ExecuteMigration(e, version) => MigrateError::Failed {
153            version,
154            error: classify(e),
155        },
156        other => MigrateError::Backend(other.to_string()),
157    }
158}
159
160pub(crate) fn backend_error(e: sqlx::Error) -> MigrateError {
161    MigrateError::Backend(e.to_string())
162}
163
164/// Run the migrations on one connection taken from the pool and report what **this run** applied.
165/// sqlx holds a session advisory lock for the length of a run and does **not** release it when a
166/// migration fails, so the pooled connection would keep it and the next run, on another connection, would
167/// wait for it forever (measured on MySQL 8.4: a hang after a failed migration). The lock belongs to the
168/// session, so after a failure the session is ended instead of returned to the pool.
169///
170/// The ledger is read before and after on the same connection, **inside the lock** (the advisory lock
171/// stacks, so sqlx taking it again is harmless). Read outside it, two simultaneous runs both see the
172/// ledger empty at the start and full at the end, and both report every version as theirs. `$exists` is a
173/// statement that counts the ledger table (0 or 1).
174///
175/// `$guard` is `Some((take, release))` where sqlx's own lock is not a lock. MariaDB rejects the negative
176/// timeout sqlx passes to `GET_LOCK(.., -1)` and returns NULL, which sqlx ignores, so two runs overlap
177/// (measured on MariaDB 11.4: `Duplicate entry '2' for key 'PRIMARY'`). `take` is a statement returning 1
178/// once the lock is held; `release` frees it.
179macro_rules! run_on_own_connection {
180    ($pool:expr, $migrations:expr, $classify:expr, $exists:expr, $guard:expr) => {{
181        let mut conn = $pool
182            .acquire()
183            .await
184            .map_err(crate::migrate::backend_error)?;
185        let mut failure = None;
186        let mut before = Vec::new();
187        let mut after = Vec::new();
188        let guard: Option<(&str, &str)> = $guard;
189        'run: {
190            if let Some((take, _)) = guard {
191                match sqlx::query_scalar::<_, Option<i64>>(sqlx::AssertSqlSafe(take.to_string()))
192                    .fetch_one(&mut *conn)
193                    .await
194                {
195                    Ok(Some(1)) => {}
196                    Ok(other) => {
197                        failure = Some(persistant_relational::MigrateError::Backend(format!(
198                            "the migration lock was not granted: {other:?}"
199                        )));
200                        break 'run;
201                    }
202                    Err(e) => {
203                        failure = Some(crate::migrate::backend_error(e));
204                        break 'run;
205                    }
206                }
207            }
208            if let Err(e) = sqlx::migrate::Migrate::lock(&mut *conn).await {
209                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
210                break 'run;
211            }
212            match crate::migrate::ledger(&mut *conn, $exists).await {
213                Ok(v) => before = v,
214                Err(e) => {
215                    failure = Some(e);
216                    break 'run;
217                }
218            }
219            // `run_direct` is sqlx own way round "implementation of `Acquire` is not general enough".
220            if let Err(e) = crate::migrate::migrator($migrations)
221                .run_direct(None, &mut *conn, false)
222                .await
223            {
224                failure = Some(crate::migrate::map_error(e, $classify));
225                break 'run;
226            }
227            match crate::migrate::ledger(&mut *conn, $exists).await {
228                Ok(v) => after = v,
229                Err(e) => {
230                    failure = Some(e);
231                    break 'run;
232                }
233            }
234            if let Err(e) = sqlx::migrate::Migrate::unlock(&mut *conn).await {
235                failure = Some(persistant_relational::MigrateError::Backend(e.to_string()));
236            }
237            if let Some((_, release)) = guard {
238                if let Err(e) = sqlx::query(sqlx::AssertSqlSafe(release.to_string()))
239                    .execute(&mut *conn)
240                    .await
241                {
242                    failure = Some(crate::migrate::backend_error(e));
243                }
244            }
245        }
246        match failure {
247            None => {
248                let applied = after
249                    .iter()
250                    .copied()
251                    .filter(|v| !before.contains(v))
252                    .collect();
253                Ok(persistant_relational::MigrateReport {
254                    applied,
255                    already_applied: before,
256                })
257            }
258            Some(e) => {
259                let _ = sqlx::Connection::close(conn.detach()).await;
260                Err(e)
261            }
262        }
263    }};
264}
265#[cfg(any(feature = "mysql", feature = "sqlite"))]
266pub(crate) use run_on_own_connection;
267
268/// The versions the ledger says succeeded, read on `conn`; empty if there is no ledger. `exists` is a
269/// statement that returns 1 when the ledger table is there and 0 when it is not.
270pub(crate) async fn ledger<C>(conn: &mut C, exists: &str) -> Result<Vec<i64>, MigrateError>
271where
272    C: sqlx::Connection,
273    for<'c> &'c mut C: sqlx::Executor<'c, Database = C::Database>,
274    for<'q> i64: sqlx::Decode<'q, C::Database> + sqlx::Type<C::Database>,
275    <C::Database as sqlx::Database>::Arguments: sqlx::IntoArguments<C::Database>,
276    usize: sqlx::ColumnIndex<<C::Database as sqlx::Database>::Row>,
277{
278    let there: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(exists.to_string()))
279        .fetch_one(&mut *conn)
280        .await
281        .map_err(backend_error)?;
282    if there == 0 {
283        return Ok(Vec::new());
284    }
285    sqlx::query_scalar(sqlx::AssertSqlSafe(
286        "SELECT version FROM _sqlx_migrations WHERE success ORDER BY version".to_string(),
287    ))
288    .fetch_all(&mut *conn)
289    .await
290    .map_err(backend_error)
291}
292
293impl Migrates for PgBackend {
294    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
295        run_on_own_connection!(
296            self.pool,
297            migrations,
298            classify,
299            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END",
300            None
301        )
302    }
303
304    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
305        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
306            .fetch_one(&self.pool)
307            .await
308            .map_err(backend_error)?;
309        if !exists {
310            return Ok(Vec::new());
311        }
312        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
313            .fetch_all(&self.pool)
314            .await
315            .map_err(backend_error)
316    }
317}
318
319impl PgBackend {
320    /// Apply `migrations` in version order, skipping those already applied, in the ledger
321    /// `_sqlx_migrations` of the connection's current schema. See [`Migrates::migrate`].
322    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
323        Migrates::migrate(self, migrations).await
324    }
325
326    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
327    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
328        Migrates::applied_versions(self).await
329    }
330}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
