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.into_iter().map(|(version, _)| version).collect(),
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                    .map(|(version, _)| *version)
251                    .filter(|v| !before.contains(v))
252                    .collect();
253                let checksums = after;
254                Ok(persistant_relational::MigrateReport {
255                    applied,
256                    already_applied: before,
257                    checksums,
258                })
259            }
260            Some(e) => {
261                let _ = sqlx::Connection::close(conn.detach()).await;
262                Err(e)
263            }
264        }
265    }};
266}
267#[cfg(any(feature = "mysql", feature = "sqlite"))]
268pub(crate) use run_on_own_connection;
269
270/// The versions the ledger says succeeded, read on `conn`; empty if there is no ledger. `exists` is a
271/// statement that returns 1 when the ledger table is there and 0 when it is not.
272pub(crate) async fn ledger<C>(
273    conn: &mut C,
274    exists: &str,
275) -> Result<Vec<(i64, Vec<u8>)>, MigrateError>
276where
277    C: sqlx::Connection,
278    for<'c> &'c mut C: sqlx::Executor<'c, Database = C::Database>,
279    for<'q> i64: sqlx::Decode<'q, C::Database> + sqlx::Type<C::Database>,
280    for<'q> Vec<u8>: sqlx::Decode<'q, C::Database> + sqlx::Type<C::Database>,
281    <C::Database as sqlx::Database>::Arguments: sqlx::IntoArguments<C::Database>,
282    usize: sqlx::ColumnIndex<<C::Database as sqlx::Database>::Row>,
283{
284    let there: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(exists.to_string()))
285        .fetch_one(&mut *conn)
286        .await
287        .map_err(backend_error)?;
288    if there == 0 {
289        return Ok(Vec::new());
290    }
291    sqlx::query_as(sqlx::AssertSqlSafe(
292        "SELECT version, checksum FROM _sqlx_migrations WHERE success ORDER BY version".to_string(),
293    ))
294    .fetch_all(&mut *conn)
295    .await
296    .map_err(backend_error)
297}
298
299impl Migrates for PgBackend {
300    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
301        run_on_own_connection!(
302            self.pool,
303            migrations,
304            classify,
305            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END",
306            None
307        )
308    }
309
310    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
311        let exists: bool = sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
312            .fetch_one(&self.pool)
313            .await
314            .map_err(backend_error)?;
315        if !exists {
316            return Ok(Vec::new());
317        }
318        sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE success ORDER BY version")
319            .fetch_all(&self.pool)
320            .await
321            .map_err(backend_error)
322    }
323}
324
325impl PgBackend {
326    /// Apply `migrations` in version order, skipping those already applied, in the ledger
327    /// `_sqlx_migrations` of the connection's current schema. See [`Migrates::migrate`].
328    pub async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
329        Migrates::migrate(self, migrations).await
330    }
331
332    /// Prove that `migrations` install on an **empty** schema, in order, and leave nothing behind.
333    ///
334    /// A scratch schema with a unique name is created, the migrations run in it on a pool of one
335    /// connection whose `search_path` is that schema (so the ledger is the scratch schema's own), and
336    /// the schema is dropped afterwards, whether the run passed or failed. The report lists every
337    /// version as applied. The database being installed to is never touched: not its ledger, not its
338    /// tables.
339    ///
340    /// Limits: a migration that names another schema explicitly, or creates an extension, role or
341    /// other cluster-wide object, escapes the scratch schema; that is not detected. This is the
342    /// PostgreSQL driver only.
343    pub async fn verify_fresh(
344        &self,
345        migrations: &[Migration],
346    ) -> Result<MigrateReport, MigrateError> {
347        self.fresh_run(migrations, false).await.map(|(r, _)| r)
348    }
349
350    /// As [`verify_fresh`](Self::verify_fresh), and read the scratch schema's catalog before it is
351    /// dropped: the schema `migrations` produce, from migrations and a server, without touching the
352    /// target. The snapshot is stamped with the versions and checksums it was derived from.
353    pub async fn fresh_catalog(
354        &self,
355        migrations: &[Migration],
356    ) -> Result<persistant_relational::catalog::Snapshot, MigrateError> {
357        let (_, catalog) = self.fresh_run(migrations, true).await?;
358        Ok(catalog.expect("a catalog was asked for"))
359    }
360
361    async fn fresh_run(
362        &self,
363        migrations: &[Migration],
364        want_catalog: bool,
365    ) -> Result<
366        (
367            MigrateReport,
368            Option<persistant_relational::catalog::Snapshot>,
369        ),
370        MigrateError,
371    > {
372        use std::sync::atomic::{AtomicU64, Ordering};
373        static N: AtomicU64 = AtomicU64::new(0);
374        let nanos = std::time::SystemTime::now()
375            .duration_since(std::time::UNIX_EPOCH)
376            .map_or(0, |d| d.as_nanos());
377        let schema = format!(
378            "persistant_fresh_{}_{}_{}",
379            std::process::id(),
380            nanos,
381            N.fetch_add(1, Ordering::Relaxed)
382        );
383        sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
384            .execute(&self.pool)
385            .await
386            .map_err(backend_error)?;
387        let search = format!("SET search_path TO {schema}");
388        let scratch = sqlx::postgres::PgPoolOptions::new()
389            .max_connections(1)
390            .after_connect(move |conn, _| {
391                let search = search.clone();
392                Box::pin(async move {
393                    sqlx::query(sqlx::AssertSqlSafe(search))
394                        .execute(&mut *conn)
395                        .await
396                        .map(|_| ())
397                })
398            })
399            .connect_with((*self.pool.connect_options()).clone())
400            .await;
401        let result = match scratch {
402            Ok(pool) => {
403                let r = match PgBackend::from_pool(pool.clone()).await {
404                    Ok(b) => scratch_run(&b, migrations, want_catalog, b.applied_checksums()).await,
405                    Err(e) => Err(MigrateError::Backend(e.to_string())),
406                };
407                pool.close().await;
408                r
409            }
410            Err(e) => Err(backend_error(e)),
411        };
412        let dropped = sqlx::query(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
413            .execute(&self.pool)
414            .await
415            .map_err(backend_error);
416        let report = result?;
417        dropped?;
418        Ok(report)
419    }
420
421    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
422    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
423        Migrates::applied_versions(self).await
424    }
425
426    /// The ledger's rows that succeeded, `(version, checksum)` ascending; empty if there is no ledger
427    /// yet. Reads the same rows [`Migrates::migrate`] checks against, so a caller can compare a
428    /// directory of migrations with a database without applying anything.
429    pub async fn applied_checksums(&self) -> Result<Vec<(i64, Vec<u8>)>, MigrateError> {
430        let mut conn = self.pool.acquire().await.map_err(backend_error)?;
431        ledger(
432            &mut *conn,
433            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END",
434        )
435        .await
436    }
437}
438
439/// Run `migrations` on the scratch backend `b` and, if asked, read its catalog stamped with what was
440/// applied. `ledger` reads the scratch backend's own ledger (it needs the backend's concrete type).
441pub(crate) async fn scratch_run<B, L>(
442    b: &B,
443    migrations: &[Migration],
444    want_catalog: bool,
445    ledger: L,
446) -> Result<
447    (
448        MigrateReport,
449        Option<persistant_relational::catalog::Snapshot>,
450    ),
451    MigrateError,
452>
453where
454    B: persistant_relational::RelationalBackend + Migrates,
455    L: std::future::Future<Output = Result<Vec<(i64, Vec<u8>)>, MigrateError>>,
456{
457    let report = Migrates::migrate(b, migrations).await?;
458    if !want_catalog {
459        return Ok((report, None));
460    }
461    let meta = persistant_relational::catalog::migration_meta(&ledger.await?);
462    let catalog = persistant_relational::catalog::read_catalog(b, meta)
463        .await
464        .map_err(|e| MigrateError::Backend(e.to_string()))?;
465    Ok((report, Some(catalog)))
466}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
