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

persistant_relational/
conformance.rs
