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

persistant_relational/
conformance.rs
