persistant_relational/
migrate_conformance.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Contract tests for the migration ledger ([`Migrates`]): what it promises, decided against a real
3//! database. Each leaf produces a matrix row, **on** only when the controls held.
4//!
5//! - `rel.migrate.apply`: migrations are applied in version order, the ledger lists them, and a second
6//!   run applies nothing and does not run them again.
7//! - `rel.migrate.checksum`: changing the text of an applied migration is refused as
8//!   [`MigrateError::ChecksumMismatch`], not silently re-run or ignored.
9//! - `rel.migrate.newer`: a program that knows fewer migrations than the database has applied is
10//!   refused as [`MigrateError::NewerThanProgram`].
11//! - `rel.migrate.atomic`: a migration whose statements fail part-way leaves none of its row changes,
12//!   keeps the migrations before it, and is reported as [`MigrateError::Failed`] with the database's own
13//!   classification of the failure.
14//! - `rel.migrate.ddl_atomic`: the same, for a table the failed migration created. MySQL and MariaDB
15//!   commit DDL implicitly and cannot undo it, so this leaf is off there.
16//! - `rel.migrate.concurrent`: two migrators started at once against the same database apply each
17//!   migration exactly once between them and both finish cleanly (the second waits or finds the work
18//!   done); neither fails, and no migration runs twice.
19//!
20//! **Where it runs.** The ledger belongs to the connection's schema or database, so these tests must be
21//! pointed at a **disposable** one (a scratch schema, a scratch database, a fresh file): each leaf
22//! drops the ledger and its own tables first and last.
23//!
24//! **Controls.** The known-bad control is a real one: [`Defect::NoMigrationLedger`] wraps the same
25//! target in a migrator that runs every migration's statements each time with no ledger, no checksum
26//! and no transaction. Every leaf must fail it. The positive control is weaker (as for the other
27//! relational leaves): the leaf's own tables could be created on the target at all.
28
29use crate::conformance::{Defect, Defective, Failure};
30use crate::{
31    Dialect, MigrateError, MigrateReport, Migrates, Migration, RelError, RelationalBackend, Value,
32};
33use persistant_core::{CapabilityId, Config, Controls, Reason, Row as MatrixRow, Verdict};
34use std::collections::BTreeMap;
35use std::future::Future;
36use std::task::Poll;
37
38/// A leaf this module can decide.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40#[non_exhaustive]
41pub enum MigrateLeaf {
42    /// `rel.migrate.apply`
43    Apply,
44    /// `rel.migrate.checksum`
45    Checksum,
46    /// `rel.migrate.newer`
47    Newer,
48    /// `rel.migrate.atomic`
49    Atomic,
50    /// `rel.migrate.ddl_atomic`
51    DdlAtomic,
52    /// `rel.migrate.concurrent`
53    Concurrent,
54}
55
56impl MigrateLeaf {
57    /// Every leaf with a test here.
58    pub const ALL: [MigrateLeaf; 6] = [
59        MigrateLeaf::Apply,
60        MigrateLeaf::Checksum,
61        MigrateLeaf::Newer,
62        MigrateLeaf::Atomic,
63        MigrateLeaf::DdlAtomic,
64        MigrateLeaf::Concurrent,
65    ];
66
67    /// The capability id.
68    pub fn id(self) -> &'static str {
69        match self {
70            MigrateLeaf::Apply => "rel.migrate.apply",
71            MigrateLeaf::Checksum => "rel.migrate.checksum",
72            MigrateLeaf::Newer => "rel.migrate.newer",
73            MigrateLeaf::Atomic => "rel.migrate.atomic",
74            MigrateLeaf::DdlAtomic => "rel.migrate.ddl_atomic",
75            MigrateLeaf::Concurrent => "rel.migrate.concurrent",
76        }
77    }
78}
79
80const T: &str = "persistant_mig_t";
81const T2: &str = "persistant_mig_t2";
82
83fn wrong<X>(msg: impl Into<String>) -> Result<X, Failure> {
84    Err(Failure::Wrong(msg.into()))
85}
86
87fn mig(version: i64, sql: &str) -> Migration {
88    Migration::new(version, format!("persistant migrate leaf {version}"), sql)
89}
90
91fn create_t() -> Migration {
92    mig(
93        1,
94        &format!("CREATE TABLE {T} (id BIGINT PRIMARY KEY, n BIGINT NOT NULL)"),
95    )
96}
97
98async fn reset<M: Migrates>(m: &M) -> Result<(), RelError> {
99    for t in [T, T2, "_sqlx_migrations"] {
100        m.execute(&format!("DROP TABLE IF EXISTS {t}"), &[]).await?;
101    }
102    Ok(())
103}
104
105async fn count<M: Migrates>(m: &M, table: &str) -> Result<i64, RelError> {
106    let rows = m
107        .query(&format!("SELECT COUNT(*) FROM {table}"), &[])
108        .await?;
109    Ok(rows.first().and_then(|r| r.int(0)).unwrap_or(-1))
110}
111
112fn migrated<X>(r: Result<X, MigrateError>, what: &str) -> Result<X, Failure> {
113    r.or_else(|e| wrong(format!("{what}: {e}")))
114}
115
116/// Run one leaf's test against `m`. Writes only to tables named `persistant_mig_*` and to the ledger
117/// `_sqlx_migrations`, which it drops first and last.
118pub async fn run_leaf<M: Migrates>(leaf: MigrateLeaf, m: &M) -> Result<(), Failure> {
119    if matches!(leaf, MigrateLeaf::DdlAtomic)
120        && matches!(m.dialect(), Dialect::MySql | Dialect::MariaDb)
121    {
122        return Err(Failure::Backend(RelError::Unsupported(
123            "DDL commits implicitly: a failed migration cannot undo the tables it created"
124                .to_string(),
125        )));
126    }
127    reset(m).await?;
128    let result = match leaf {
129        MigrateLeaf::Apply => apply(m).await,
130        MigrateLeaf::Checksum => checksum(m).await,
131        MigrateLeaf::Newer => newer(m).await,
132        MigrateLeaf::Atomic => atomic(m).await,
133        MigrateLeaf::DdlAtomic => ddl_atomic(m).await,
134        MigrateLeaf::Concurrent => concurrent(m).await,
135    };
136    reset(m).await?;
137    result
138}
139
140async fn apply<M: Migrates>(m: &M) -> Result<(), Failure> {
141    let migs = [
142        create_t(),
143        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
144    ];
145    let first = migrated(m.migrate(&migs).await, "the first run")?;
146    if first.applied != [1, 2] || !first.already_applied.is_empty() {
147        return wrong(format!(
148            "the first run reported {first:?}, not [1, 2] applied"
149        ));
150    }
151    if count(m, T).await? != 1 {
152        return wrong("migration 2 did not leave its row");
153    }
154    let ledger = migrated(m.applied_versions().await, "reading the ledger")?;
155    if ledger != [1, 2] {
156        return wrong(format!("the ledger lists {ledger:?}, not [1, 2]"));
157    }
158    let again = migrated(m.migrate(&migs).await, "the second run")?;
159    let want = MigrateReport {
160        applied: vec![],
161        already_applied: vec![1, 2],
162    };
163    if again != want {
164        return wrong(format!("the second run reported {again:?}, not {want:?}"));
165    }
166    if count(m, T).await? != 1 {
167        return wrong("the second run ran a migration again");
168    }
169    Ok(())
170}
171
172async fn checksum<M: Migrates>(m: &M) -> Result<(), Failure> {
173    let migs = [
174        create_t(),
175        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
176    ];
177    migrated(m.migrate(&migs).await, "the first run")?;
178    let edited = [
179        create_t(),
180        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 2)")),
181    ];
182    match m.migrate(&edited).await {
183        Err(MigrateError::ChecksumMismatch { version: 2 }) => Ok(()),
184        other => wrong(format!(
185            "an applied migration was edited and the run answered {other:?}, not a checksum mismatch on 2"
186        )),
187    }
188}
189
190async fn newer<M: Migrates>(m: &M) -> Result<(), Failure> {
191    let migs = [
192        create_t(),
193        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
194    ];
195    migrated(m.migrate(&migs).await, "the first run")?;
196    match m.migrate(&migs[..1]).await {
197        Err(MigrateError::NewerThanProgram { version: 2 }) => Ok(()),
198        other => wrong(format!(
199            "a program that knows only migration 1 met a database at 2 and the run answered {other:?}"
200        )),
201    }
202}
203
204async fn atomic<M: Migrates>(m: &M) -> Result<(), Failure> {
205    let migs = [
206        create_t(),
207        mig(
208            2,
209            &format!(
210                "INSERT INTO {T} (id, n) VALUES (1, 1); INSERT INTO {T} (id, n) VALUES (1, 2)"
211            ),
212        ),
213    ];
214    match m.migrate(&migs).await {
215        Err(MigrateError::Failed {
216            version: 2,
217            error: RelError::UniqueViolation { .. },
218        }) => {}
219        other => {
220            return wrong(format!(
221                "a migration whose second statement repeats a key answered {other:?}, not a classified failure of 2"
222            ));
223        }
224    }
225    if count(m, T).await? != 0 {
226        return wrong("the failed migration left the rows of its first statement behind");
227    }
228    let ledger = migrated(m.applied_versions().await, "reading the ledger")?;
229    if ledger != [1] {
230        return wrong(format!(
231            "after the failure the ledger lists {ledger:?}, not [1]"
232        ));
233    }
234    Ok(())
235}
236
237async fn ddl_atomic<M: Migrates>(m: &M) -> Result<(), Failure> {
238    let migs = [
239        create_t(),
240        mig(
241            2,
242            &format!(
243                "CREATE TABLE {T2} (id BIGINT PRIMARY KEY); INSERT INTO {T} (id, n) VALUES (1, 1); INSERT INTO {T} (id, n) VALUES (1, 2)"
244            ),
245        ),
246    ];
247    if !matches!(
248        m.migrate(&migs).await,
249        Err(MigrateError::Failed { version: 2, .. })
250    ) {
251        return wrong("the failing migration was not reported as failed");
252    }
253    // Control: the table migration 1 made is there and can be counted; a table that is not there cannot.
254    count(m, T).await?;
255    if count(m, T2).await.is_ok() {
256        return wrong("the failed migration left the table it created behind");
257    }
258    Ok(())
259}
260
261/// Poll two futures together on this task until both are done.
262async fn join2<A: Future, B: Future>(a: A, b: B) -> (A::Output, B::Output) {
263    let mut a = std::pin::pin!(a);
264    let mut b = std::pin::pin!(b);
265    let (mut ra, mut rb) = (None, None);
266    std::future::poll_fn(|cx| {
267        if ra.is_none()
268            && let Poll::Ready(v) = a.as_mut().poll(cx)
269        {
270            ra = Some(v);
271        }
272        if rb.is_none()
273            && let Poll::Ready(v) = b.as_mut().poll(cx)
274        {
275            rb = Some(v);
276        }
277        if ra.is_some() && rb.is_some() {
278            Poll::Ready(())
279        } else {
280            Poll::Pending
281        }
282    })
283    .await;
284    (ra.expect("a is done"), rb.expect("b is done"))
285}
286
287async fn concurrent<M: Migrates>(m: &M) -> Result<(), Failure> {
288    let migs = [
289        create_t(),
290        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
291    ];
292    let (one, two) = join2(m.migrate(&migs), m.migrate(&migs)).await;
293    let one = migrated(one, "the first of two simultaneous runs")?;
294    let two = migrated(two, "the second of two simultaneous runs")?;
295    let mut applied: Vec<i64> = one.applied.iter().chain(&two.applied).copied().collect();
296    applied.sort_unstable();
297    if applied != [1, 2] {
298        return wrong(format!(
299            "two simultaneous runs applied {applied:?} between them, not each of [1, 2] once ({one:?}, {two:?})"
300        ));
301    }
302    if count(m, T).await? != 1 {
303        return wrong("a migration ran twice: its row is not there exactly once");
304    }
305    let ledger = migrated(m.applied_versions().await, "reading the ledger")?;
306    if ledger != [1, 2] {
307        return wrong(format!("the ledger lists {ledger:?}, not [1, 2]"));
308    }
309    Ok(())
310}
311
312/// The outcome of certifying one migration leaf.
313#[derive(Debug, Clone, PartialEq, Eq)]
314pub struct MigrationCertified {
315    /// The leaf.
316    pub leaf: MigrateLeaf,
317    /// The row to record.
318    pub row: MatrixRow,
319}
320
321/// Run every migration leaf against `target` (a **disposable** ledger, see the module docs) and return
322/// one row per leaf. A leaf is **on** only if the target passed *and* the known-bad control failed.
323pub async fn certify<M: Migrates>(
324    target: &M,
325    config: &Config,
326    metadata: &Config,
327) -> Vec<MigrationCertified> {
328    let mut out = Vec::new();
329    for leaf in MigrateLeaf::ALL {
330        let result = run_leaf(leaf, target).await;
331        let positive_passed = !matches!(
332            &result,
333            Err(Failure::Backend(
334                RelError::Unsupported(_) | RelError::Unavailable(_)
335            ))
336        );
337        let bad = Defective::new(target, Defect::NoMigrationLedger);
338        let known_bad_failed = run_leaf(leaf, &bad).await.is_err();
339        let controls = Controls {
340            positive_passed,
341            known_bad_failed,
342        };
343        let id = CapabilityId::new(leaf.id()).expect("leaf id is valid");
344        let md = metadata.clone().with("test", format!("rel::{}", leaf.id()));
345        let off = |reason: Reason, note: String| MatrixRow {
346            subject_kind: "capability".to_string(),
347            subject: id.clone(),
348            backend: target.backend().to_string(),
349            version: target.version().to_string(),
350            config: config.clone(),
351            enabled: false,
352            reason: Some(reason),
353            metadata: md.clone(),
354            note: note.replace(['\t', '\n'], " "),
355            extra: BTreeMap::new(),
356        };
357        let row = match (
358            &result,
359            controls.positive_passed && controls.known_bad_failed,
360        ) {
361            (Ok(()), true) => Verdict::Exact
362                .into_row(
363                    id.clone(),
364                    target.backend(),
365                    target.version(),
366                    config.clone(),
367                    md.clone(),
368                    controls,
369                )
370                .expect("both controls hold"),
371            (Ok(()), false) => off(
372                Reason::Untested,
373                format!("controls did not hold: {controls:?}"),
374            ),
375            (Err(Failure::Wrong(m)), _) => off(Reason::Wrong, m.clone()),
376            (Err(Failure::Backend(e)), _) => off(Reason::Missing, e.to_string()),
377        };
378        out.push(MigrationCertified { leaf, row });
379    }
380    out
381}
382
383/// The known-bad migrator: no ledger, no checksum, no transaction. It runs each migration's
384/// statements (split on `;`) straight on the wrapped backend every time it is asked.
385impl<R: RelationalBackend> Migrates for Defective<'_, R> {
386    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
387        if self.defect() != Defect::NoMigrationLedger {
388            return Err(MigrateError::Backend(
389                "this defect does not wrap a migrator".to_string(),
390            ));
391        }
392        let mut applied = Vec::new();
393        for m in migrations {
394            for stmt in m.sql.split(';').map(str::trim).filter(|s| !s.is_empty()) {
395                self.inner()
396                    .execute(stmt, &[] as &[Value])
397                    .await
398                    .map_err(|error| MigrateError::Failed {
399                        version: m.version,
400                        error,
401                    })?;
402            }
403            applied.push(m.version);
404        }
405        Ok(MigrateReport {
406            applied,
407            already_applied: Vec::new(),
408        })
409    }
410
411    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
412        Ok(Vec::new())
413    }
414}
persistant_relational/migrate_conformance.rs

persistant_relational/
migrate_conformance.rs
