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    if !again.applied.is_empty() || again.already_applied != [1, 2] {
160        return wrong(format!(
161            "the second run reported {again:?}, not [1, 2] already applied"
162        ));
163    }
164    if count(m, T).await? != 1 {
165        return wrong("the second run ran a migration again");
166    }
167    Ok(())
168}
169
170async fn checksum<M: Migrates>(m: &M) -> Result<(), Failure> {
171    let migs = [
172        create_t(),
173        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
174    ];
175    migrated(m.migrate(&migs).await, "the first run")?;
176    let edited = [
177        create_t(),
178        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 2)")),
179    ];
180    match m.migrate(&edited).await {
181        Err(MigrateError::ChecksumMismatch { version: 2 }) => Ok(()),
182        other => wrong(format!(
183            "an applied migration was edited and the run answered {other:?}, not a checksum mismatch on 2"
184        )),
185    }
186}
187
188async fn newer<M: Migrates>(m: &M) -> Result<(), Failure> {
189    let migs = [
190        create_t(),
191        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
192    ];
193    migrated(m.migrate(&migs).await, "the first run")?;
194    match m.migrate(&migs[..1]).await {
195        Err(MigrateError::NewerThanProgram { version: 2 }) => Ok(()),
196        other => wrong(format!(
197            "a program that knows only migration 1 met a database at 2 and the run answered {other:?}"
198        )),
199    }
200}
201
202async fn atomic<M: Migrates>(m: &M) -> Result<(), Failure> {
203    let migs = [
204        create_t(),
205        mig(
206            2,
207            &format!(
208                "INSERT INTO {T} (id, n) VALUES (1, 1); INSERT INTO {T} (id, n) VALUES (1, 2)"
209            ),
210        ),
211    ];
212    match m.migrate(&migs).await {
213        Err(MigrateError::Failed {
214            version: 2,
215            error: RelError::UniqueViolation { .. },
216        }) => {}
217        other => {
218            return wrong(format!(
219                "a migration whose second statement repeats a key answered {other:?}, not a classified failure of 2"
220            ));
221        }
222    }
223    if count(m, T).await? != 0 {
224        return wrong("the failed migration left the rows of its first statement behind");
225    }
226    let ledger = migrated(m.applied_versions().await, "reading the ledger")?;
227    if ledger != [1] {
228        return wrong(format!(
229            "after the failure the ledger lists {ledger:?}, not [1]"
230        ));
231    }
232    Ok(())
233}
234
235async fn ddl_atomic<M: Migrates>(m: &M) -> Result<(), Failure> {
236    let migs = [
237        create_t(),
238        mig(
239            2,
240            &format!(
241                "CREATE TABLE {T2} (id BIGINT PRIMARY KEY); INSERT INTO {T} (id, n) VALUES (1, 1); INSERT INTO {T} (id, n) VALUES (1, 2)"
242            ),
243        ),
244    ];
245    if !matches!(
246        m.migrate(&migs).await,
247        Err(MigrateError::Failed { version: 2, .. })
248    ) {
249        return wrong("the failing migration was not reported as failed");
250    }
251    // Control: the table migration 1 made is there and can be counted; a table that is not there cannot.
252    count(m, T).await?;
253    if count(m, T2).await.is_ok() {
254        return wrong("the failed migration left the table it created behind");
255    }
256    Ok(())
257}
258
259/// Poll two futures together on this task until both are done.
260async fn join2<A: Future, B: Future>(a: A, b: B) -> (A::Output, B::Output) {
261    let mut a = std::pin::pin!(a);
262    let mut b = std::pin::pin!(b);
263    let (mut ra, mut rb) = (None, None);
264    std::future::poll_fn(|cx| {
265        if ra.is_none()
266            && let Poll::Ready(v) = a.as_mut().poll(cx)
267        {
268            ra = Some(v);
269        }
270        if rb.is_none()
271            && let Poll::Ready(v) = b.as_mut().poll(cx)
272        {
273            rb = Some(v);
274        }
275        if ra.is_some() && rb.is_some() {
276            Poll::Ready(())
277        } else {
278            Poll::Pending
279        }
280    })
281    .await;
282    (ra.expect("a is done"), rb.expect("b is done"))
283}
284
285async fn concurrent<M: Migrates>(m: &M) -> Result<(), Failure> {
286    let migs = [
287        create_t(),
288        mig(2, &format!("INSERT INTO {T} (id, n) VALUES (1, 1)")),
289    ];
290    let (one, two) = join2(m.migrate(&migs), m.migrate(&migs)).await;
291    let one = migrated(one, "the first of two simultaneous runs")?;
292    let two = migrated(two, "the second of two simultaneous runs")?;
293    let mut applied: Vec<i64> = one.applied.iter().chain(&two.applied).copied().collect();
294    applied.sort_unstable();
295    if applied != [1, 2] {
296        return wrong(format!(
297            "two simultaneous runs applied {applied:?} between them, not each of [1, 2] once ({one:?}, {two:?})"
298        ));
299    }
300    if count(m, T).await? != 1 {
301        return wrong("a migration ran twice: its row is not there exactly once");
302    }
303    let ledger = migrated(m.applied_versions().await, "reading the ledger")?;
304    if ledger != [1, 2] {
305        return wrong(format!("the ledger lists {ledger:?}, not [1, 2]"));
306    }
307    Ok(())
308}
309
310/// The outcome of certifying one migration leaf.
311#[derive(Debug, Clone, PartialEq, Eq)]
312pub struct MigrationCertified {
313    /// The leaf.
314    pub leaf: MigrateLeaf,
315    /// The row to record.
316    pub row: MatrixRow,
317}
318
319/// Run every migration leaf against `target` (a **disposable** ledger, see the module docs) and return
320/// one row per leaf. A leaf is **on** only if the target passed *and* the known-bad control failed.
321pub async fn certify<M: Migrates>(
322    target: &M,
323    config: &Config,
324    metadata: &Config,
325) -> Vec<MigrationCertified> {
326    let mut out = Vec::new();
327    for leaf in MigrateLeaf::ALL {
328        let result = run_leaf(leaf, target).await;
329        let positive_passed = !matches!(
330            &result,
331            Err(Failure::Backend(
332                RelError::Unsupported(_) | RelError::Unavailable(_)
333            ))
334        );
335        let bad = Defective::new(target, Defect::NoMigrationLedger);
336        let known_bad_failed = run_leaf(leaf, &bad).await.is_err();
337        let controls = Controls {
338            positive_passed,
339            known_bad_failed,
340        };
341        let id = CapabilityId::new(leaf.id()).expect("leaf id is valid");
342        let md = metadata.clone().with("test", format!("rel::{}", leaf.id()));
343        let off = |reason: Reason, note: String| MatrixRow {
344            subject_kind: "capability".to_string(),
345            subject: id.clone(),
346            backend: target.backend().to_string(),
347            version: target.version().to_string(),
348            config: config.clone(),
349            enabled: false,
350            reason: Some(reason),
351            metadata: md.clone(),
352            note: note.replace(['\t', '\n'], " "),
353            extra: BTreeMap::new(),
354        };
355        let row = match (
356            &result,
357            controls.positive_passed && controls.known_bad_failed,
358        ) {
359            (Ok(()), true) => Verdict::Exact
360                .into_row(
361                    id.clone(),
362                    target.backend(),
363                    target.version(),
364                    config.clone(),
365                    md.clone(),
366                    controls,
367                )
368                .expect("both controls hold"),
369            (Ok(()), false) => off(
370                Reason::Untested,
371                format!("controls did not hold: {controls:?}"),
372            ),
373            (Err(Failure::Wrong(m)), _) => off(Reason::Wrong, m.clone()),
374            (Err(Failure::Backend(e)), _) => off(Reason::Missing, e.to_string()),
375        };
376        out.push(MigrationCertified { leaf, row });
377    }
378    out
379}
380
381/// The known-bad migrator: no ledger, no checksum, no transaction. It runs each migration's
382/// statements (split on `;`) straight on the wrapped backend every time it is asked.
383impl<R: RelationalBackend> Migrates for Defective<'_, R> {
384    async fn migrate(&self, migrations: &[Migration]) -> Result<MigrateReport, MigrateError> {
385        if self.defect() != Defect::NoMigrationLedger {
386            return Err(MigrateError::Backend(
387                "this defect does not wrap a migrator".to_string(),
388            ));
389        }
390        let mut applied = Vec::new();
391        for m in migrations {
392            for stmt in m.sql.split(';').map(str::trim).filter(|s| !s.is_empty()) {
393                self.inner()
394                    .execute(stmt, &[] as &[Value])
395                    .await
396                    .map_err(|error| MigrateError::Failed {
397                        version: m.version,
398                        error,
399                    })?;
400            }
401            applied.push(m.version);
402        }
403        Ok(MigrateReport {
404            applied,
405            already_applied: Vec::new(),
406            checksums: Vec::new(),
407        })
408    }
409
410    async fn applied_versions(&self) -> Result<Vec<i64>, MigrateError> {
411        Ok(Vec::new())
412    }
413}
persistant_relational/migrate_conformance.rs

persistant_relational/
migrate_conformance.rs
