Crate ruststream_sqlx_dialect 

SQL text generation for ruststream-sqlx, the SQL database crate of the RustStream messaging framework.

The procedural macros of ruststream-sqlx and its runtime build their statements here, so both produce the same SQL for a table. The crate generates plain text, free of any database driver, so a procedural macro can run it at compile time.

A TableSpec describes a queue table: its name, the Column that identifies a row, one slot per Role, the columns of the message’s data, and the Form in which rows are claimed. The id is part of the constructor, every role has one slot, and the form carries its own columns, so a table without an id, with a role played twice or with two forms cannot be described. A dialect turns the description into the Statement each queue event runs; a statement carries the Params its placeholders bind, in order.

A dialect is a set of traits. Dialect quotes names, numbers placeholders and builds the statements every form runs to settle a row. Each form of claiming is a trait over it: RowLock builds the claim that locks rows for the transaction a handler settles in, Lease the claim that writes a lease into each row and commits, with the lease’s extension, and Advisory the claim of candidates with their lock keys, the lock and the unlock of a key, and the take of a row whose key the session holds. A dialect implements the traits of the forms its database serves, and a table in a form its dialect lacks does not compile.

In the lease form a claim writes the lease’s expiry (Param::Lease) into each row it takes and commits, and skips every row whose lease has not ended by Param::LeaseNow. The expiry it wrote is the delivery’s ownership token (Param::Held): a settlement, or an extension that moves the expiry forward (Lease::extend), passes only while the row still holds it. A dialect whose claim only selects the rows says so (Lease::claim_writes_lease), and each claimed row is then stamped with its lease (Lease::stamp).

In the advisory lock form a session lock on each row’s key holds the row. The key’s parts (KeyPart) name literal text and columns, and the database renders each row’s key from them as text. A claim selects candidates with their keys (Advisory::advisory_claim), takes the lock on a key (Advisory::lock, bound as Param::Key), then the row while it is still claimable (Advisory::take); the unlock (Advisory::unlock) follows the settlement. The settlements name the row alone, and a retry needs no statement: the take counted the attempt, and the unlock frees the row.

A table with FIFO groups (TableSpec::fifo_group) keeps one row of a group in work. Its claim takes the group’s head, the first unfinished row of the group in claim order, or nothing. The claim’s transaction first takes the group with the statement Dialect::fifo_guard gives, so a row that enters the group ahead of the head in work waits for it; a lease claim also takes nothing while a row of the group holds a lease.

A table may open its transactions at an isolation level (Isolation) or, on SQLite, in a mode (Mode): its Opening, set with TableSpec::isolation and TableSpec::mode. Dialect::begin gives the statement that opens a transaction at it, and refuses an opening the database lacks. A dialect also states the openings it serves as types, one Opens implementation per level type, so a table that names a level its dialect lacks does not compile.

A transactional outbox table records what a service publishes until a consumer has processed it. OutboxDialect builds its statements: the take of a record by its id, the mark of a processed record, and the recovery of the unprocessed records of one name, which the table’s group column holds. Every built-in dialect implements it.

Postgres, MySql and Sqlite are built in, behind the postgres, mysql and sqlite features. Postgres and MySql implement RowLock, Lease and Advisory, and MySql serves MariaDB too; Sqlite implements Lease and Advisory. Their inserts are written by the const fns of insert, so a service that describes a table by hand renders its insert in a const, as the derive does. A database without a built-in dialect takes a type of the service’s own: it implements Dialect, the trait of each form it builds and Opens for each level it opens, and writes every statement in its database’s SQL. A service on a built-in dialect that writes one statement its own way writes it as an event of its inbox: #[inbox(custom(ack))] and an Ack implementation of its own, in ruststream-sqlx.

Examples

A dialect for SQL Server, a database without a built-in one, in the row lock form. The service mounts it with SqlxBroker::with_dialect, as the ruststream-sqlx overview shows.

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Opening, Param, RowLock, Statement, StatementError, TableSpec,
};

/// SQL Server: bracket-quoted names, `@p1` placeholders, rows claimed under an update lock that
/// skips the rows another claim holds.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn quote_into(&self, ident: &str, out: &mut String) {
        out.push('[');
        out.push_str(&ident.replace(']', "]]"));
        out.push(']');
    }

    fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
        out.push_str("@p");
        out.push_str(&index.to_string());
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        let mut sql = String::from("DELETE FROM ");
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = ");
        self.placeholder_into(NonZeroUsize::MIN, &mut sql);
        Ok(Statement::new(sql, [Param::Id]))
    }

    // SQL Server starts a transaction with `BEGIN TRANSACTION`.
    fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
        match opening {
            Opening::Default => Ok(Some("BEGIN TRANSACTION")),
            other => Err(StatementError::UnsupportedOpening {
                dialect: self.name(),
                opening: other.name(),
            }),
        }
    }

    // The statements this service never runs refuse the table.
}

// The row lock form, which the service's `email_jobs` table takes.
impl RowLock for Mssql {
    fn lock_claim(
        &self,
        _spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        Ok(Statement::new(
            "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
             WITH (UPDLOCK, READPAST, ROWLOCK) ORDER BY [job_id]",
            [Param::Limit],
        ))
    }
}

Modules

insert

The insert of a row into a queue table, for each built-in dialect, written by a const fn.

level

The isolation levels and SQLite modes as types: a table names one, and its dialect opens it where it implements Opens for it.

Structs

Column

One column of a queue table: its name in the database, and whether the database fills it in.

MySql

MySQL and MariaDB: backtick-quoted names, ? placeholders, rows claimed with FOR UPDATE SKIP LOCKED.

Postgres

Postgres: double-quoted names, $1 placeholders, rows claimed with FOR UPDATE SKIP LOCKED.

Sqlite

SQLite: backtick-quoted names, ? placeholders, rows claimed by lease.

Statement

One SQL statement: its text and the parameters its placeholders bind, in placeholder order.

TableName

A table outside the queue’s own description, such as the table dead-lettered rows move to.

TableSpec

The description of a queue table: its name, the column that identifies a row, one slot per role, the message’s data columns, the form its rows are claimed in, and what its transactions open at.

Enums

ClaimShape

What a claim selects: whole rows, only their ids for a service’s own fetch, or the columns that run the queue, each named by its role, for a reader that knows no struct.

Form

How a subscription claims the rows of a table; a table has exactly one form.

Isolation

The isolation level a transaction opens at, on Postgres, MySQL and MariaDB.

KeyPart

One piece of an advisory lock key: literal text, or the value of a column of the row.

Mode

How a SQLite transaction takes the database’s write lock.

NameLimit

The longest name a database keeps, in the unit it measures names by.

Opening

What a table’s transactions open at: the dialect’s default, an isolation level, or a SQLite mode.

Param

A value a statement binds, named by its meaning; the broker supplies it when the statement runs.

ParseTableNameError

Why a string does not name a table.

Role

What a column does for the queue, named as #[field(..)] spells it.

StatementError

Why a dialect cannot build a statement for a table.

Traits

Advisory

The advisory lock form of a dialect: the claim of candidates, the lock and the unlock of a key in the session, and the take of a candidate once its lock is held.

Dialect

A database’s SQL: how it quotes names and numbers placeholders, and the statements every form of claiming runs to settle a row.

Lease

The lease form of a dialect: the claim that writes a lease into each row it takes, the extension of a lease in work, and the stamp that leases a row a claim only selected.

Opens

A dialect that opens transactions at Level: an isolation level or a SQLite mode as one of the level types, or (), which a table that names neither carries.

OutboxDialect

The statements a dialect builds for an outbox table, which records what a service publishes until a consumer has processed it.

RowLock

The row lock form of a dialect: the claim that locks its rows until its transaction ends.