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        use std::sync::atomic::{AtomicU64, Ordering};
348        static N: AtomicU64 = AtomicU64::new(0);
349        let nanos = std::time::SystemTime::now()
350            .duration_since(std::time::UNIX_EPOCH)
351            .map_or(0, |d| d.as_nanos());
352        let schema = format!(
353            "persistant_fresh_{}_{}_{}",
354            std::process::id(),
355            nanos,
356            N.fetch_add(1, Ordering::Relaxed)
357        );
358        sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
359            .execute(&self.pool)
360            .await
361            .map_err(backend_error)?;
362        let search = format!("SET search_path TO {schema}");
363        let scratch = sqlx::postgres::PgPoolOptions::new()
364            .max_connections(1)
365            .after_connect(move |conn, _| {
366                let search = search.clone();
367                Box::pin(async move {
368                    sqlx::query(sqlx::AssertSqlSafe(search))
369                        .execute(&mut *conn)
370                        .await
371                        .map(|_| ())
372                })
373            })
374            .connect_with((*self.pool.connect_options()).clone())
375            .await;
376        let result = match scratch {
377            Ok(pool) => {
378                let r = match PgBackend::from_pool(pool.clone()).await {
379                    Ok(b) => Migrates::migrate(&b, migrations).await,
380                    Err(e) => Err(MigrateError::Backend(e.to_string())),
381                };
382                pool.close().await;
383                r
384            }
385            Err(e) => Err(backend_error(e)),
386        };
387        let dropped = sqlx::query(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
388            .execute(&self.pool)
389            .await
390            .map_err(backend_error);
391        let report = result?;
392        dropped?;
393        Ok(report)
394    }
395
396    /// The versions the ledger says succeeded, ascending; empty if there is no ledger yet.
397    pub async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
398        Migrates::applied_versions(self).await
399    }
400
401    /// The ledger's rows that succeeded, `(version, checksum)` ascending; empty if there is no ledger
402    /// yet. Reads the same rows [`Migrates::migrate`] checks against, so a caller can compare a
403    /// directory of migrations with a database without applying anything.
404    pub async fn applied_checksums(&self) -> Result<Vec<(i64, Vec<u8>)>, MigrateError> {
405        let mut conn = self.pool.acquire().await.map_err(backend_error)?;
406        ledger(
407            &mut *conn,
408            "SELECT CASE WHEN to_regclass('_sqlx_migrations') IS NULL THEN 0::int8 ELSE 1::int8 END",
409        )
410        .await
411    }
412}
persistant_sqlx/migrate.rs

persistant_sqlx/
migrate.rs
