Skip to main content

MySQL

The mysql feature serves MySQL 8.0.1 and MariaDB 10.6 or later in every form, READ COMMITTED by default; index the claim order so a claim reads its own group.

Claims pass over locked rows with SKIP LOCKED, which MySQL 8.0.1 and MariaDB 10.6 added, and an advisory table shares that floor. A subscription reads the server's version when it opens, and an older server stops it with SqlxBrokerError::ServerTooOld. A claim's transaction runs at READ COMMITTED unless a row lock table declares another level. At READ COMMITTED a claim held for a handler locks no gaps between rows and holds back no insert into its table. A server whose binary log records statements (binlog_format = STATEMENT) refuses writes at that level; the row and mixed formats accept them.

A lease claim selects its rows, stamps each one with its lease and commits. The expiry is a whole second, so a DATETIME column without fractions holds the token exactly. A dead letter into a table is two statements in one transaction. A claim of the service's own comes with a Fetch of its own, because the crate reads rows by a list of ids on Postgres alone; a subscription without one stops at startup. The advisory lock form takes GET_LOCK, whose names hold 64 characters at most.

Index a table's claim order: (group, priority, retry_after, id), for the columns the table has. A row lock or lease claim locks every row of its group that it reads before sorting them. Without that index a claim held for a handler keeps the rest of its group from other claims, and in the row lock form the rows of a partition key go into work out of order. In a table with FIFO groups a claim first locks the unfinished rows of its group until its transaction ends, and the same index keeps that read to the group's own rows. A row lock table with FIFO groups that declares serializable stops its subscription when it opens, with SqlxBrokerError::Dialect: at that level InnoDB locks every row a read touches, and a claim would wait for the group's row in work.

Example​

Two groups share one table in the lease form, and the index (name, id) keeps each claim to its own group's rows. The tests run the same service on MySQL and on MariaDB.

examples/sqlx_databases_mysql.rs
/// One row of `welcome_jobs`, claimed under a lease. The claim stamps `locked_until` with a whole
/// second, so a `DATETIME` column without fractions holds it.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "welcome_jobs")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The claim order, `(group, id)` for this table's columns, keeps each claim to its group's rows.
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS welcome_jobs (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(64) NOT NULL,
    locked_until DATETIME,
    payload BLOB NOT NULL,
    INDEX claim_order (name, id)
)";

#[subscriber(InboxQueue::<SendEmail>::new("welcome"))]
async fn welcome(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending the welcome");
    HandlerOutcome::ack()
}

#[subscriber(InboxQueue::<SendEmail>::new("receipts"))]
async fn receipt(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending the receipt");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, MySqlPoolOptions::new())
        .route::<SendEmail>("welcome")
        .route::<SendEmail>("receipts");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscriptions check it.
        .on_startup(async move |()| {
            let mut conn = MySqlConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(welcome);
            b.include(receipt);
        })
}