persistant_relational/
conformance.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Contract tests for the relational kind: what [`RelationalBackend`] promises, decided against a real
3//! database. Each leaf produces a matrix row, **on** only when the controls held.
4//!
5//! - `rel.values`: integers, text, bytes and floats written as parameters come back as written, and a
6//!   `NULL` comes back as `Value::Null`.
7//! - `rel.transaction`: a committed write is visible, a rolled-back one is not, and a transaction
8//!   dropped without `commit` rolls back.
9//! - `rel.error.unique`: a duplicate key is reported as [`RelError::UniqueViolation`], not as an
10//!   opaque backend error.
11//! - `rel.foreign_key`: a reference to a missing row is rejected as [`RelError::ForeignKeyViolation`].
12//! - `rel.check_constraint`: a row violating a `CHECK` is rejected as [`RelError::CheckViolation`].
13//! - `rel.isolation.no_dirty_read`: a write inside an open transaction is invisible to another
14//!   connection until it commits, and visible after (read committed or stronger).
15//! - `rel.isolation.repeatable_read`: inside one transaction begun at that level a row read twice reads the same,
16//!   and a table counted twice counts the same, though another connection committed an update and an insert in
17//!   between. SQLite is off: its `begin` is `BEGIN IMMEDIATE`, which holds the write lock, so another connection
18//!   cannot commit in between and the probe cannot run.
19//! - `rel.isolation.serializable`: PostgreSQL: two concurrent serializable transactions that each read both rows and
20//!   then change a different one (write skew) cannot both commit, while at read committed they can (the control).
21//!   MySQL and MariaDB: a plain read inside a serializable transaction takes a shared lock, so another connection's
22//!   `FOR UPDATE NOWAIT` on that row is refused, and is not at read committed or after the transaction ends.
23//!   SQLite is off for the same reason as above.
24//! - `rel.row_lock`: a row locked `FOR UPDATE` by an open transaction cannot be locked by another connection
25//!   (probed with `SKIP LOCKED`, which skips it) until the transaction ends. SQLite has no row locks, so it is off there.
26//! - `rel.upsert_where`: `INSERT .. ON CONFLICT (..) DO UPDATE .. WHERE ..` (MySQL: `ON DUPLICATE KEY UPDATE n = IF(cond, new, n)`) updates only when the condition holds.
27//!   MySQL and MariaDB have no such clause, so it is off there.
28//! - `rel.savepoint`: inside a transaction, `ROLLBACK TO` a savepoint undoes the work after it and keeps the work before it.
29//!
30//! **Controls.** There is no reference relational engine to use as a positive control, so the positive
31//! control is weaker than in the kv and blob suites: it is that the leaf's schema could be created on
32//! the target at all (a failure to create it is "missing", not "wrong"). The **known-bad** controls are
33//! real: a [`Defect`] wrapper that makes the same target misbehave in one way (no transactions, lossy
34//! bytes, unclassified errors), and, for the two constraint leaves, the same test run against a schema
35//! created *without* the constraint, which must be caught accepting the bad row.
36//!
37//! Statements are written per dialect (this is a probe of the dialect itself, not of translation).
38//! Booleans are left out of `rel.values`: how a driver returns one (`BOOL`, `TINYINT(1)`, `0/1`) is a
39//! documented difference, not a contract.
40
41use crate::{Dialect, Isolation, RelError, RelationalBackend, RelationalTx, Row, Value};
42use persistant_core::{CapabilityId, Config, Controls, Reason, Row as MatrixRow, Verdict};
43use std::collections::BTreeMap;
44
45/// A leaf this module can decide.
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47#[non_exhaustive]
48pub enum Leaf {
49    /// `rel.values`
50    Values,
51    /// `rel.transaction`
52    Transaction,
53    /// `rel.error.unique`
54    UniqueClassified,
55    /// `rel.foreign_key`
56    ForeignKey,
57    /// `rel.check_constraint`
58    CheckConstraint,
59    /// `rel.isolation.no_dirty_read`
60    IsolationNoDirtyRead,
61    /// `rel.isolation.repeatable_read`
62    IsolationRepeatableRead,
63    /// `rel.isolation.serializable`
64    IsolationSerializable,
65    /// `rel.savepoint`
66    Savepoint,
67    /// `rel.row_lock`
68    RowLock,
69    /// `rel.upsert_where`
70    UpsertWhere,
71}
72
73impl Leaf {
74    /// Every leaf with a test here.
75    pub const ALL: [Leaf; 11] = [
76        Leaf::Values,
77        Leaf::Transaction,
78        Leaf::UniqueClassified,
79        Leaf::ForeignKey,
80        Leaf::CheckConstraint,
81        Leaf::IsolationNoDirtyRead,
82        Leaf::IsolationRepeatableRead,
83        Leaf::IsolationSerializable,
84        Leaf::Savepoint,
85        Leaf::RowLock,
86        Leaf::UpsertWhere,
87    ];
88
89    /// The capability id.
90    pub fn id(self) -> &'static str {
91        match self {
92            Leaf::Values => "rel.values",
93            Leaf::Transaction => "rel.transaction",
94            Leaf::UniqueClassified => "rel.error.unique",
95            Leaf::ForeignKey => "rel.foreign_key",
96            Leaf::CheckConstraint => "rel.check_constraint",
97            Leaf::IsolationNoDirtyRead => "rel.isolation.no_dirty_read",
98            Leaf::IsolationRepeatableRead => "rel.isolation.repeatable_read",
99            Leaf::IsolationSerializable => "rel.isolation.serializable",
100            Leaf::Savepoint => "rel.savepoint",
101            Leaf::RowLock => "rel.row_lock",
102            Leaf::UpsertWhere => "rel.upsert_where",
103        }
104    }
105
106    fn defects(self) -> &'static [Defect] {
107        match self {
108            Leaf::Values => &[Defect::LossyBytes],
109            Leaf::Transaction
110            | Leaf::IsolationNoDirtyRead
111            | Leaf::IsolationRepeatableRead
112            | Leaf::IsolationSerializable
113            | Leaf::Savepoint
114            | Leaf::RowLock => &[Defect::NoTransactions],
115            Leaf::UniqueClassified => &[Defect::UnclassifiedErrors],
116            Leaf::ForeignKey | Leaf::CheckConstraint | Leaf::UpsertWhere => &[],
117        }
118    }
119
120    /// Whether the leaf has a "schema without the constraint" known-bad run.
121    fn has_unconstrained_control(self) -> bool {
122        matches!(
123            self,
124            Leaf::ForeignKey | Leaf::CheckConstraint | Leaf::UpsertWhere
125        )
126    }
127}
128
129/// One way for a wrapped backend to be wrong, for the known-bad controls.
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131#[non_exhaustive]
132pub enum Defect {
133    /// `begin` returns a transaction whose statements run at once and whose `rollback` and drop do
134    /// nothing.
135    NoTransactions,
136    /// A byte parameter is cut to its first byte.
137    LossyBytes,
138    /// Every error comes back as `RelError::Backend`.
139    UnclassifiedErrors,
140    /// As a migrator ([`Migrates`](crate::Migrates)): no ledger, no checksum, no transaction; every
141    /// migration runs again each time it is asked. Used by [`migrate_conformance`](crate::migrate_conformance).
142    NoMigrationLedger,
143}
144
145/// A backend with one [`Defect`] injected. The wrapped backend does the work.
146#[derive(Debug)]
147pub struct Defective<'a, R> {
148    inner: &'a R,
149    defect: Defect,
150}
151
152impl<'a, R: RelationalBackend> Defective<'a, R> {
153    /// Wrap `inner` so it is wrong in exactly the way `defect` says.
154    pub fn new(inner: &'a R, defect: Defect) -> Self {
155        Defective { inner, defect }
156    }
157
158    pub(crate) fn inner(&self) -> &'a R {
159        self.inner
160    }
161
162    pub(crate) fn defect(&self) -> Defect {
163        self.defect
164    }
165}
166
167/// The transaction of a [`Defective`] backend.
168pub enum DefectiveTx<'a, R: RelationalBackend + 'a> {
169    /// The real transaction (the defect is elsewhere).
170    Real(R::Tx<'a>, Defect),
171    /// No transaction at all: statements run on the backend directly.
172    Direct(&'a R),
173}
174
175impl<R: RelationalBackend> RelationalBackend for Defective<'_, R> {
176    type Tx<'a>
177        = DefectiveTx<'a, R>
178    where
179        Self: 'a;
180
181    fn backend(&self) -> &str {
182        self.inner.backend()
183    }
184
185    fn version(&self) -> &str {
186        self.inner.version()
187    }
188
189    fn dialect(&self) -> Dialect {
190        self.inner.dialect()
191    }
192
193    async fn execute(&self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
194        self.inner
195            .execute(sql, &lossy(self.defect, params))
196            .await
197            .map_err(|e| unclassify(self.defect, e))
198    }
199
200    async fn query(&self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
201        self.inner
202            .query(sql, &lossy(self.defect, params))
203            .await
204            .map_err(|e| unclassify(self.defect, e))
205    }
206
207    async fn begin(&self, isolation: Isolation) -> Result<DefectiveTx<'_, R>, RelError> {
208        if self.defect == Defect::NoTransactions {
209            return Ok(DefectiveTx::Direct(self.inner));
210        }
211        Ok(DefectiveTx::Real(
212            self.inner
213                .begin(isolation)
214                .await
215                .map_err(|e| unclassify(self.defect, e))?,
216            self.defect,
217        ))
218    }
219}
220
221fn lossy(defect: Defect, params: &[Value]) -> Vec<Value> {
222    params
223        .iter()
224        .map(|p| match (defect, p) {
225            (Defect::LossyBytes, Value::Bytes(b)) => {
226                Value::Bytes(b.iter().take(1).copied().collect())
227            }
228            _ => p.clone(),
229        })
230        .collect()
231}
232
233fn unclassify(defect: Defect, e: RelError) -> RelError {
234    if defect == Defect::UnclassifiedErrors {
235        RelError::Backend(e.to_string())
236    } else {
237        e
238    }
239}
240
241impl<R: RelationalBackend> RelationalTx for DefectiveTx<'_, R> {
242    async fn execute(&mut self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
243        match self {
244            DefectiveTx::Real(tx, d) => tx
245                .execute(sql, &lossy(*d, params))
246                .await
247                .map_err(|e| unclassify(*d, e)),
248            // With no transaction there is nothing for a savepoint to be in: the statements are
249            // no-ops (run on a pool they would open a transaction on one connection and leak it).
250            DefectiveTx::Direct(_)
251                if sql.starts_with("SAVEPOINT")
252                    || sql.starts_with("RELEASE")
253                    || sql.starts_with("ROLLBACK TO") =>
254            {
255                Ok(0)
256            }
257            DefectiveTx::Direct(db) => db.execute(sql, params).await,
258        }
259    }
260
261    async fn query(&mut self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
262        match self {
263            DefectiveTx::Real(tx, d) => tx
264                .query(sql, &lossy(*d, params))
265                .await
266                .map_err(|e| unclassify(*d, e)),
267            DefectiveTx::Direct(db) => db.query(sql, params).await,
268        }
269    }
270
271    async fn commit(self) -> Result<(), RelError> {
272        match self {
273            DefectiveTx::Real(tx, d) => tx.commit().await.map_err(|e| unclassify(d, e)),
274            DefectiveTx::Direct(_) => Ok(()),
275        }
276    }
277
278    async fn rollback(self) -> Result<(), RelError> {
279        match self {
280            DefectiveTx::Real(tx, d) => tx.rollback().await.map_err(|e| unclassify(d, e)),
281            DefectiveTx::Direct(_) => Ok(()),
282        }
283    }
284}
285
286/// Why a leaf's test did not pass.
287#[derive(Debug, Clone, PartialEq, Eq)]
288pub enum Failure {
289    /// The backend returned something other than the contract says.
290    Wrong(String),
291    /// The backend could not do something the test needs (including creating the leaf's schema).
292    Backend(RelError),
293}
294
295impl From<RelError> for Failure {
296    fn from(e: RelError) -> Self {
297        Failure::Backend(e)
298    }
299}
300
301fn wrong<T>(msg: impl Into<String>) -> Result<T, Failure> {
302    Err(Failure::Wrong(msg.into()))
303}
304
305// ---- per-dialect statement text -------------------------------------------------------------
306
307#[derive(Clone, Copy)]
308pub(crate) enum Ty {
309    Int,
310    Text,
311    Bytes,
312    Float,
313}
314
315pub(crate) struct Sql(pub(crate) Dialect);
316
317impl Sql {
318    /// The placeholder for parameter `n` (1-based).
319    pub(crate) fn p(&self, n: usize) -> String {
320        match self.0 {
321            Dialect::Postgres => format!("${n}"),
322            Dialect::Sqlite => format!("?{n}"),
323            _ => "?".to_string(),
324        }
325    }
326
327    pub(crate) fn ty(&self, t: Ty) -> &'static str {
328        match (&self.0, t) {
329            (_, Ty::Int) => "BIGINT",
330            (Dialect::Postgres, Ty::Text) => "TEXT",
331            (Dialect::Sqlite, Ty::Text) => "TEXT",
332            (_, Ty::Text) => "VARCHAR(255)",
333            (Dialect::Postgres, Ty::Bytes) => "BYTEA",
334            (Dialect::Sqlite, Ty::Bytes) => "BLOB",
335            (_, Ty::Bytes) => "LONGBLOB",
336            (Dialect::Postgres, Ty::Float) => "DOUBLE PRECISION",
337            (_, Ty::Float) => "DOUBLE",
338        }
339    }
340
341    /// Table options (InnoDB is explicit on MySQL so constraints and transactions exist).
342    pub(crate) fn tail(&self) -> &'static str {
343        match self.0 {
344            Dialect::MySql | Dialect::MariaDb => " ENGINE=InnoDB",
345            _ => "",
346        }
347    }
348
349    /// An integer primary key column definition.
350    pub(crate) fn pk(&self, name: &str) -> String {
351        format!("{name} BIGINT NOT NULL PRIMARY KEY")
352    }
353}
354
355fn supported(d: &Dialect) -> Result<(), Failure> {
356    match d {
357        Dialect::Postgres | Dialect::Sqlite | Dialect::MySql | Dialect::MariaDb => Ok(()),
358        other => Err(Failure::Backend(RelError::Unsupported(format!(
359            "the relational contract tests have no statements for {other}"
360        )))),
361    }
362}
363
364pub(crate) async fn drop_tables<R: RelationalBackend>(
365    b: &R,
366    names: &[&str],
367) -> Result<(), RelError> {
368    for n in names {
369        b.execute(&format!("DROP TABLE IF EXISTS {n}"), &[]).await?;
370    }
371    Ok(())
372}
373
374/// Run one leaf's test against `b`. `constrained = false` creates the constraint leaves' schema without
375/// the constraint (the known-bad control); it makes no difference to the other leaves. Writes only to
376/// tables named `persistant_rel_*`, which it drops first and last.
377pub async fn run_leaf<R: RelationalBackend>(
378    leaf: Leaf,
379    b: &R,
380    constrained: bool,
381) -> Result<(), Failure> {
382    let d = b.dialect();
383    supported(&d)?;
384    let s = Sql(d);
385    match leaf {
386        Leaf::Values => values(b, &s).await,
387        Leaf::Transaction => transaction(b, &s).await,
388        Leaf::UniqueClassified => unique(b, &s).await,
389        Leaf::ForeignKey => foreign_key(b, &s, constrained).await,
390        Leaf::CheckConstraint => check(b, &s, constrained).await,
391        Leaf::IsolationNoDirtyRead => no_dirty_read(b, &s).await,
392        Leaf::IsolationRepeatableRead => repeatable_read(b, &s).await,
393        Leaf::IsolationSerializable => serializable(b, &s).await,
394        Leaf::Savepoint => savepoint(b, &s).await,
395        Leaf::RowLock => row_lock(b, &s).await,
396        Leaf::UpsertWhere => upsert_where(b, &s, constrained).await,
397    }
398}
399
400async fn values<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
401    const T: &str = "persistant_rel_values";
402    drop_tables(b, &[T]).await?;
403    b.execute(
404        &format!(
405            "CREATE TABLE {T} ({}, i {}, t {}, b {}, f {}){}",
406            s.pk("id"),
407            s.ty(Ty::Int),
408            s.ty(Ty::Text),
409            s.ty(Ty::Bytes),
410            s.ty(Ty::Float),
411            s.tail()
412        ),
413        &[],
414    )
415    .await?;
416    let cases: Vec<(i64, Value, Value, Value, Value)> = vec![
417        (
418            1,
419            Value::Int(i64::MIN),
420            Value::Text("ünï ✓ \u{1F980}".to_string()),
421            Value::Bytes(vec![0x00, 0xFF, 0x80, 0x00, 0x7F]),
422            Value::Float(1.5),
423        ),
424        (
425            2,
426            Value::Int(i64::MAX),
427            Value::Text(String::new()),
428            Value::Bytes(vec![]),
429            Value::Float(-0.25),
430        ),
431        (3, Value::Null, Value::Null, Value::Null, Value::Null),
432    ];
433    for (id, i, t, bytes, f) in &cases {
434        b.execute(
435            &format!(
436                "INSERT INTO {T} (id, i, t, b, f) VALUES ({}, {}, {}, {}, {})",
437                s.p(1),
438                s.p(2),
439                s.p(3),
440                s.p(4),
441                s.p(5)
442            ),
443            &[
444                Value::Int(*id),
445                i.clone(),
446                t.clone(),
447                bytes.clone(),
448                f.clone(),
449            ],
450        )
451        .await?;
452    }
453    let rows = b
454        .query(&format!("SELECT id, i, t, b, f FROM {T} ORDER BY id"), &[])
455        .await?;
456    drop_tables(b, &[T]).await?;
457    if rows.len() != cases.len() {
458        return wrong(format!("wrote {} rows, read {}", cases.len(), rows.len()));
459    }
460    for (row, (id, i, t, bytes, f)) in rows.iter().zip(&cases) {
461        let want = [
462            Value::Int(*id),
463            i.clone(),
464            t.clone(),
465            bytes.clone(),
466            f.clone(),
467        ];
468        // An empty byte string and an empty text may come back as NULL on no engine here; they must not.
469        if row.values != want {
470            return wrong(format!("row {id}: wrote {want:?}, read {:?}", row.values));
471        }
472    }
473    Ok(())
474}
475
476async fn count<R: RelationalBackend>(b: &R, table: &str) -> Result<i64, Failure> {
477    let r = b
478        .query(&format!("SELECT COUNT(*) FROM {table}"), &[])
479        .await?;
480    match r.first().and_then(|r| r.int(0)) {
481        Some(n) => Ok(n),
482        None => wrong("COUNT(*) did not come back as an integer"),
483    }
484}
485
486async fn transaction<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
487    const T: &str = "persistant_rel_tx";
488    drop_tables(b, &[T]).await?;
489    b.execute(
490        &format!(
491            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
492            s.pk("id"),
493            s.tail()
494        ),
495        &[],
496    )
497    .await?;
498    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
499    let result = async {
500        // Commit: visible.
501        let mut tx = b.begin(Isolation::ReadCommitted).await?;
502        tx.execute(&ins, &[Value::Int(1), Value::Int(10)]).await?;
503        let own = tx.query(&format!("SELECT COUNT(*) FROM {T}"), &[]).await?;
504        if own.first().and_then(|r| r.int(0)) != Some(1) {
505            return wrong("a transaction cannot see its own write");
506        }
507        tx.commit().await?;
508        if count(b, T).await? != 1 {
509            return wrong("a committed write is not visible");
510        }
511        // Rollback: not visible.
512        let mut tx = b.begin(Isolation::ReadCommitted).await?;
513        tx.execute(&ins, &[Value::Int(2), Value::Int(20)]).await?;
514        tx.rollback().await?;
515        if count(b, T).await? != 1 {
516            return wrong("a rolled-back write is visible");
517        }
518        // Dropped without commit: rolled back.
519        {
520            let mut tx = b.begin(Isolation::ReadCommitted).await?;
521            tx.execute(&ins, &[Value::Int(3), Value::Int(30)]).await?;
522        }
523        if count(b, T).await? != 1 {
524            return wrong("a transaction dropped without commit was not rolled back");
525        }
526        Ok(())
527    }
528    .await;
529    drop_tables(b, &[T]).await?;
530    result
531}
532
533async fn no_dirty_read<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
534    const T: &str = "persistant_rel_isolation";
535    drop_tables(b, &[T]).await?;
536    b.execute(
537        &format!(
538            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
539            s.pk("id"),
540            s.tail()
541        ),
542        &[],
543    )
544    .await?;
545    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
546    let result = async {
547        let mut tx = b.begin(Isolation::ReadCommitted).await?;
548        tx.execute(&ins, &[Value::Int(1), Value::Int(1)]).await?;
549        // `b.query` runs on another connection, outside `tx`.
550        let outside = count(b, T).await?;
551        if outside != 0 {
552            let _ = tx.rollback().await;
553            return wrong(format!(
554                "another connection saw {outside} uncommitted row(s) (dirty read)"
555            ));
556        }
557        tx.commit().await?;
558        if count(b, T).await? != 1 {
559            return wrong("a committed write is not visible to another connection");
560        }
561        Ok(())
562    }
563    .await;
564    drop_tables(b, &[T]).await?;
565    result
566}
567
568/// SQLite's `begin` is `BEGIN IMMEDIATE`: it holds the write lock, so the probes that need another
569/// connection to commit (or a second transaction to run) in the meantime cannot run.
570fn no_concurrent_commit(s: &Sql) -> Result<(), Failure> {
571    if matches!(s.0, Dialect::Sqlite) {
572        return Err(Failure::Backend(RelError::Unsupported(
573            "begin() is BEGIN IMMEDIATE: it holds the write lock, so no other connection can commit during the transaction and the probe cannot run"
574                .to_string(),
575        )));
576    }
577    Ok(())
578}
579
580/// Run one transaction's last two steps. `Ok(true)` committed, `Ok(false)` was refused with a
581/// serialization failure (at the update or at the commit); any other error is the caller's.
582async fn finish<T: RelationalTx>(mut tx: T, sql: &str) -> Result<bool, Failure> {
583    match tx.execute(sql, &[]).await {
584        Ok(_) => {}
585        Err(RelError::Serialization) => {
586            let _ = tx.rollback().await;
587            return Ok(false);
588        }
589        Err(e) => {
590            let _ = tx.rollback().await;
591            return Err(e.into());
592        }
593    }
594    match tx.commit().await {
595        Ok(()) => Ok(true),
596        Err(RelError::Serialization) => Ok(false),
597        Err(e) => Err(e.into()),
598    }
599}
600
601/// Two transactions at `level` each read how many rows have `n = 1` (two), then each sets a different
602/// row to 0 (write skew). Returns whether each committed.
603async fn write_skew<R: RelationalBackend>(
604    b: &R,
605    s: &Sql,
606    table: &str,
607    level: Isolation,
608) -> Result<(bool, bool), Failure> {
609    b.execute(&format!("DELETE FROM {table}"), &[]).await?;
610    let ins = format!("INSERT INTO {table} (id, n) VALUES ({}, 1)", s.p(1));
611    b.execute(&ins, &[Value::Int(1)]).await?;
612    b.execute(&ins, &[Value::Int(2)]).await?;
613    let read = format!("SELECT COUNT(*) FROM {table} WHERE n = 1");
614    let mut t1 = b.begin(level).await?;
615    let mut t2 = b.begin(level).await?;
616    t1.query(&read, &[]).await?;
617    t2.query(&read, &[]).await?;
618    let c1 = finish(t1, &format!("UPDATE {table} SET n = 0 WHERE id = 1")).await?;
619    let c2 = finish(t2, &format!("UPDATE {table} SET n = 0 WHERE id = 2")).await?;
620    Ok((c1, c2))
621}
622
623async fn serializable<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
624    const T: &str = "persistant_rel_serializable";
625    match s.0 {
626        Dialect::Postgres | Dialect::MySql | Dialect::MariaDb => {}
627        _ => {
628            no_concurrent_commit(s)?;
629            return Err(Failure::Backend(RelError::Unsupported(format!(
630                "no probe for {}",
631                s.0
632            ))));
633        }
634    }
635    drop_tables(b, &[T]).await?;
636    b.execute(
637        &format!(
638            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
639            s.pk("id"),
640            s.tail()
641        ),
642        &[],
643    )
644    .await?;
645    let result = if matches!(s.0, Dialect::Postgres) {
646        async {
647            // Control: at read committed the schedule does commit both, so the probe can see the anomaly.
648            let control = write_skew(b, s, T, Isolation::ReadCommitted).await?;
649            if control != (true, true) {
650                return wrong(format!(
651                    "control: at read committed the write-skew schedule committed {control:?}, not both"
652                ));
653            }
654            let (c1, c2) = write_skew(b, s, T, Isolation::Serializable).await?;
655            if c1 && c2 {
656                return wrong("two serializable transactions that each read both rows and wrote a different one both committed (write skew)");
657            }
658            if !c1 && !c2 {
659                return wrong("both serializable transactions were refused: one should have committed");
660            }
661            Ok(())
662        }
663        .await
664    } else {
665        async {
666            b.execute(
667                &format!("INSERT INTO {T} (id, n) VALUES ({}, 1)", s.p(1)),
668                &[Value::Int(1)],
669            )
670            .await?;
671            let read = format!("SELECT n FROM {T} WHERE id = 1");
672            let nowait = format!("SELECT id FROM {T} WHERE id = 1 FOR UPDATE NOWAIT");
673            // Control: after a plain read at read committed the row is not locked.
674            let mut rc = b.begin(Isolation::ReadCommitted).await?;
675            rc.query(&read, &[]).await?;
676            let free = b.query(&nowait, &[]).await;
677            let _ = rc.rollback().await;
678            if !matches!(&free, Ok(rows) if rows.len() == 1) {
679                return wrong(format!(
680                    "control: a plain read at read committed left the row locked ({free:?})"
681                ));
682            }
683            let mut tx = b.begin(Isolation::Serializable).await?;
684            tx.query(&read, &[]).await?;
685            let held = b.query(&nowait, &[]).await;
686            let _ = tx.rollback().await;
687            if held.is_ok() {
688                return wrong(
689                    "a plain read in a serializable transaction did not lock the row it read",
690                );
691            }
692            match b.query(&nowait, &[]).await {
693                Ok(rows) if rows.len() == 1 => Ok(()),
694                other => wrong(format!(
695                    "the row is still locked after the serializable transaction ended ({other:?})"
696                )),
697            }
698        }
699        .await
700    };
701    drop_tables(b, &[T]).await?;
702    result
703}
704
705async fn repeatable_read<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
706    const T: &str = "persistant_rel_repeatable";
707    no_concurrent_commit(s)?;
708    drop_tables(b, &[T]).await?;
709    b.execute(
710        &format!(
711            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
712            s.pk("id"),
713            s.tail()
714        ),
715        &[],
716    )
717    .await?;
718    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
719    let read = format!("SELECT n FROM {T} WHERE id = 1");
720    let result = async {
721        b.execute(&ins, &[Value::Int(1), Value::Int(1)]).await?;
722        let mut tx = b.begin(Isolation::RepeatableRead).await?;
723        let first = tx.query(&read, &[]).await?;
724        if first.first().and_then(|r| r.int(0)) != Some(1) {
725            let _ = tx.rollback().await;
726            return wrong("the first read did not return the row written before the transaction");
727        }
728        // `b.execute` runs on another connection, outside `tx`, and commits at once.
729        b.execute(
730            &format!("UPDATE {T} SET n = 2 WHERE id = 1"),
731            &[],
732        )
733        .await?;
734        b.execute(&ins, &[Value::Int(2), Value::Int(2)]).await?;
735        let again = tx.query(&read, &[]).await?;
736        let seen = again.first().and_then(|r| r.int(0));
737        if seen != Some(1) {
738            let _ = tx.rollback().await;
739            return wrong(format!(
740                "a row read twice in one transaction changed from 1 to {seen:?} (non-repeatable read)"
741            ));
742        }
743        let rows = tx
744            .query(&format!("SELECT COUNT(*) FROM {T}"), &[])
745            .await?;
746        let n = rows.first().and_then(|r| r.int(0));
747        if n != Some(1) {
748            let _ = tx.rollback().await;
749            return wrong(format!(
750                "a table counted inside one transaction went from 1 to {n:?} rows (phantom read)"
751            ));
752        }
753        tx.commit().await?;
754        let after = b.query(&read, &[]).await?;
755        if after.first().and_then(|r| r.int(0)) != Some(2) {
756            return wrong("the other connection's committed update is not visible after the transaction ends");
757        }
758        Ok(())
759    }
760    .await;
761    drop_tables(b, &[T]).await?;
762    result
763}
764
765async fn savepoint<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
766    const T: &str = "persistant_rel_savepoint";
767    drop_tables(b, &[T]).await?;
768    b.execute(
769        &format!(
770            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
771            s.pk("id"),
772            s.tail()
773        ),
774        &[],
775    )
776    .await?;
777    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
778    let result = async {
779        let mut tx = b.begin(Isolation::ReadCommitted).await?;
780        tx.execute(&ins, &[Value::Int(1), Value::Int(1)]).await?;
781        tx.execute("SAVEPOINT persistant_sp", &[]).await?;
782        tx.execute(&ins, &[Value::Int(2), Value::Int(2)]).await?;
783        tx.execute("ROLLBACK TO SAVEPOINT persistant_sp", &[])
784            .await?;
785        tx.commit().await?;
786        let rows = b
787            .query(&format!("SELECT id FROM {T} ORDER BY id"), &[])
788            .await?;
789        let ids: Vec<Option<i64>> = rows.iter().map(|r| r.int(0)).collect();
790        if ids != [Some(1)] {
791            return wrong(format!(
792                "after ROLLBACK TO a savepoint and commit, expected only row 1, found {ids:?}"
793            ));
794        }
795        Ok(())
796    }
797    .await;
798    drop_tables(b, &[T]).await?;
799    result
800}
801
802async fn row_lock<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
803    const T: &str = "persistant_rel_row_lock";
804    if matches!(s.0, Dialect::Sqlite) {
805        return Err(Failure::Backend(RelError::Unsupported(
806            "SQLite has no row locks: a write locks the whole database".to_string(),
807        )));
808    }
809    drop_tables(b, &[T]).await?;
810    b.execute(
811        &format!(
812            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
813            s.pk("id"),
814            s.tail()
815        ),
816        &[],
817    )
818    .await?;
819    b.execute(
820        &format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2)),
821        &[Value::Int(1), Value::Int(1)],
822    )
823    .await?;
824    let lock = format!("SELECT id FROM {T} WHERE id = 1 FOR UPDATE");
825    let probe = format!("SELECT id FROM {T} WHERE id = 1 FOR UPDATE SKIP LOCKED");
826    let result = async {
827        let mut tx = b.begin(Isolation::ReadCommitted).await?;
828        if tx.query(&lock, &[]).await?.len() != 1 {
829            return wrong("FOR UPDATE did not return the row it locks");
830        }
831        // `b.query` runs on another connection, outside `tx`.
832        let held = b.query(&probe, &[]).await;
833        let held = match held {
834            Ok(rows) => rows.len(),
835            Err(e) => {
836                let _ = tx.rollback().await;
837                return Err(e.into());
838            }
839        };
840        if held != 0 {
841            let _ = tx.rollback().await;
842            return wrong("another connection locked a row an open transaction holds FOR UPDATE");
843        }
844        tx.commit().await?;
845        if b.query(&probe, &[]).await?.len() != 1 {
846            return wrong("the row is still locked after the transaction committed");
847        }
848        Ok(())
849    }
850    .await;
851    drop_tables(b, &[T]).await?;
852    result
853}
854
855/// `constrained = false` drops the `WHERE`: the known-bad control, a backend that ignores the condition.
856async fn upsert_where<R: RelationalBackend>(
857    b: &R,
858    s: &Sql,
859    constrained: bool,
860) -> Result<(), Failure> {
861    const T: &str = "persistant_rel_upsert";
862    drop_tables(b, &[T]).await?;
863    b.execute(
864        &format!(
865            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
866            s.pk("id"),
867            s.tail()
868        ),
869        &[],
870    )
871    .await?;
872    // MySQL and MariaDB have no `DO UPDATE .. WHERE`: the condition moves into the assignment.
873    let up = if matches!(s.0, Dialect::MySql | Dialect::MariaDb) {
874        let set = if constrained {
875            format!("IF({T}.n < VALUES(n), VALUES(n), {T}.n)")
876        } else {
877            "VALUES(n)".to_string()
878        };
879        format!(
880            "INSERT INTO {T} (id, n) VALUES ({}, {}) ON DUPLICATE KEY UPDATE n = {set}",
881            s.p(1),
882            s.p(2)
883        )
884    } else {
885        let cond = if constrained {
886            format!(" WHERE {T}.n < EXCLUDED.n")
887        } else {
888            String::new()
889        };
890        format!(
891            "INSERT INTO {T} (id, n) VALUES ({}, {}) ON CONFLICT (id) DO UPDATE SET n = EXCLUDED.n{cond}",
892            s.p(1),
893            s.p(2)
894        )
895    };
896    let result = async {
897        // Absent: inserted. Then a lower value is refused by the WHERE, a higher one stored.
898        for (n, want) in [(10, 10), (5, 10), (20, 20)] {
899            b.execute(&up, &[Value::Int(1), Value::Int(n)]).await?;
900            let rows = b
901                .query(&format!("SELECT n FROM {T} WHERE id = 1"), &[])
902                .await?;
903            let got = rows.first().and_then(|r| r.int(0));
904            if got != Some(want) {
905                return wrong(format!(
906                    "upsert of {n} left {got:?}, expected {want} (the WHERE must hold the row when n is not higher)"
907                ));
908            }
909        }
910        Ok(())
911    }
912    .await;
913    drop_tables(b, &[T]).await?;
914    result
915}
916
917async fn unique<R: RelationalBackend>(b: &R, s: &Sql) -> Result<(), Failure> {
918    const T: &str = "persistant_rel_unique";
919    drop_tables(b, &[T]).await?;
920    b.execute(
921        &format!(
922            "CREATE TABLE {T} ({}, n BIGINT NOT NULL){}",
923            s.pk("id"),
924            s.tail()
925        ),
926        &[],
927    )
928    .await?;
929    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
930    let result = async {
931        b.execute(&ins, &[Value::Int(1), Value::Int(1)]).await?;
932        match b.execute(&ins, &[Value::Int(1), Value::Int(2)]).await {
933            Err(RelError::UniqueViolation { .. }) => {}
934            other => return wrong(format!("a duplicate key came back as {other:?}")),
935        }
936        // Inside a transaction too.
937        let mut tx = b.begin(Isolation::ReadCommitted).await?;
938        let dup = tx.execute(&ins, &[Value::Int(1), Value::Int(3)]).await;
939        let _ = tx.rollback().await;
940        match dup {
941            Err(RelError::UniqueViolation { .. }) => Ok(()),
942            other => wrong(format!(
943                "a duplicate key in a transaction came back as {other:?}"
944            )),
945        }
946    }
947    .await;
948    drop_tables(b, &[T]).await?;
949    result
950}
951
952async fn foreign_key<R: RelationalBackend>(
953    b: &R,
954    s: &Sql,
955    constrained: bool,
956) -> Result<(), Failure> {
957    const P: &str = "persistant_rel_fk_parent";
958    const C: &str = "persistant_rel_fk_child";
959    drop_tables(b, &[C, P]).await?;
960    b.execute(
961        &format!("CREATE TABLE {P} ({}){}", s.pk("id"), s.tail()),
962        &[],
963    )
964    .await?;
965    let refs = if constrained {
966        format!(", FOREIGN KEY (pid) REFERENCES {P}(id)")
967    } else {
968        String::new()
969    };
970    b.execute(
971        &format!(
972            "CREATE TABLE {C} ({}, pid BIGINT NOT NULL{refs}){}",
973            s.pk("id"),
974            s.tail()
975        ),
976        &[],
977    )
978    .await?;
979    let result = async {
980        b.execute(
981            &format!("INSERT INTO {P} (id) VALUES ({})", s.p(1)),
982            &[Value::Int(1)],
983        )
984        .await?;
985        let ins = format!("INSERT INTO {C} (id, pid) VALUES ({}, {})", s.p(1), s.p(2));
986        b.execute(&ins, &[Value::Int(1), Value::Int(1)]).await?;
987        match b.execute(&ins, &[Value::Int(2), Value::Int(99)]).await {
988            Err(RelError::ForeignKeyViolation { .. }) => {}
989            Ok(_) => return wrong("a reference to a missing row was accepted"),
990            other => {
991                return wrong(format!(
992                    "a reference to a missing row came back as {other:?}"
993                ));
994            }
995        }
996        match b
997            .execute(
998                &format!("DELETE FROM {P} WHERE id = {}", s.p(1)),
999                &[Value::Int(1)],
1000            )
1001            .await
1002        {
1003            Err(RelError::ForeignKeyViolation { .. }) => Ok(()),
1004            Ok(_) => wrong("a referenced row was deleted"),
1005            other => wrong(format!("deleting a referenced row came back as {other:?}")),
1006        }
1007    }
1008    .await;
1009    drop_tables(b, &[C, P]).await?;
1010    result
1011}
1012
1013async fn check<R: RelationalBackend>(b: &R, s: &Sql, constrained: bool) -> Result<(), Failure> {
1014    const T: &str = "persistant_rel_check";
1015    drop_tables(b, &[T]).await?;
1016    let ck = if constrained { " CHECK (n >= 0)" } else { "" };
1017    b.execute(
1018        &format!(
1019            "CREATE TABLE {T} ({}, n BIGINT NOT NULL{ck}){}",
1020            s.pk("id"),
1021            s.tail()
1022        ),
1023        &[],
1024    )
1025    .await?;
1026    let ins = format!("INSERT INTO {T} (id, n) VALUES ({}, {})", s.p(1), s.p(2));
1027    let result = async {
1028        b.execute(&ins, &[Value::Int(1), Value::Int(0)]).await?;
1029        match b.execute(&ins, &[Value::Int(2), Value::Int(-1)]).await {
1030            Err(RelError::CheckViolation { .. }) => Ok(()),
1031            Ok(_) => wrong("a row violating a CHECK was accepted"),
1032            other => wrong(format!("a row violating a CHECK came back as {other:?}")),
1033        }
1034    }
1035    .await;
1036    drop_tables(b, &[T]).await?;
1037    result
1038}
1039
1040// ---- certification --------------------------------------------------------------------------
1041
1042/// The outcome of certifying one leaf.
1043#[derive(Debug, Clone, PartialEq, Eq)]
1044pub struct Certified {
1045    /// The leaf.
1046    pub leaf: Leaf,
1047    /// The row to record.
1048    pub row: MatrixRow,
1049}
1050
1051/// Run every leaf here against `target` and return one row per leaf.
1052///
1053/// A leaf is **on** only if the target passed *and* every known-bad control failed (see the module
1054/// docs for the positive control). `config` is the facts a gate can read for this target; `metadata`
1055/// is copied into every row with `test=rel::<leaf>` added.
1056pub async fn certify<R: RelationalBackend>(
1057    target: &R,
1058    config: &Config,
1059    metadata: &Config,
1060) -> Vec<Certified> {
1061    certify_leaves(target, config, metadata, &Leaf::ALL).await
1062}
1063
1064/// As [`certify`], for the given leaves only (the eligibility check of a consumer that needs a few).
1065pub async fn certify_leaves<R: RelationalBackend>(
1066    target: &R,
1067    config: &Config,
1068    metadata: &Config,
1069    leaves: &[Leaf],
1070) -> Vec<Certified> {
1071    let mut out = Vec::new();
1072    for leaf in leaves.iter().copied() {
1073        let result = run_leaf(leaf, target, true).await;
1074        let positive_passed = !matches!(
1075            &result,
1076            Err(Failure::Backend(
1077                RelError::Unsupported(_) | RelError::Unavailable(_)
1078            ))
1079        );
1080        let mut known_bad_failed = true;
1081        for d in leaf.defects() {
1082            let bad = Defective::new(target, *d);
1083            if run_leaf(leaf, &bad, true).await.is_ok() {
1084                known_bad_failed = false;
1085            }
1086        }
1087        if leaf.has_unconstrained_control() && run_leaf(leaf, target, false).await.is_ok() {
1088            known_bad_failed = false;
1089        }
1090        let controls = Controls {
1091            positive_passed,
1092            known_bad_failed,
1093        };
1094        let id = CapabilityId::new(leaf.id()).expect("leaf id is valid");
1095        let md = metadata.clone().with("test", format!("rel::{}", leaf.id()));
1096        let off = |reason: Reason, note: String| MatrixRow {
1097            subject_kind: "capability".to_string(),
1098            subject: id.clone(),
1099            backend: target.backend().to_string(),
1100            version: target.version().to_string(),
1101            config: config.clone(),
1102            enabled: false,
1103            reason: Some(reason),
1104            metadata: md.clone(),
1105            note: note.replace(['\t', '\n'], " "),
1106            extra: BTreeMap::new(),
1107        };
1108        let row = match (
1109            &result,
1110            controls.positive_passed && controls.known_bad_failed,
1111        ) {
1112            (Ok(()), true) => Verdict::Exact
1113                .into_row(
1114                    id.clone(),
1115                    target.backend(),
1116                    target.version(),
1117                    config.clone(),
1118                    md.clone(),
1119                    controls,
1120                )
1121                .expect("both controls hold"),
1122            (Ok(()), false) => off(
1123                Reason::Untested,
1124                format!("controls did not hold: {controls:?}"),
1125            ),
1126            (Err(Failure::Wrong(m)), _) => off(Reason::Wrong, m.clone()),
1127            (Err(Failure::Backend(e)), _) => off(Reason::Missing, e.to_string()),
1128        };
1129        out.push(Certified { leaf, row });
1130    }
1131    out
1132}
persistant_relational/conformance.rs

persistant_relational/
conformance.rs
