Advisory

advisory_lock = ".." holds each row by a lock on its key: rows that share a key go into work one at a time, and a crashed session frees its rows.

advisory_lock = "video_jobs-{channel}" in #[inbox(..)] names each row's lock key: a placeholder names a field, the key reads its column, and the database renders the key as text. A key on the row's id holds each row alone; a key on another column keeps the rows that share it in work one at a time. A key belongs to the database, not to its table, so a key that starts with its table's name keeps two tables apart. A claim selects as many due rows as it may take, in claim order, each with its key. It locks each key on a connection of its own, without waiting, and passes over a key held elsewhere. Then it takes the row while the row is still claimable: the take counts the attempt and reads the row, and it commits, so a crash spends an attempt. The connection that holds the delivery keeps the lock, and no transaction stays open while the handler runs. A settlement runs its statement on that connection, where it commits on its own; then the lock is released and the connection goes back to the pool. retry() runs no statement: the claim counted the attempt, and the release returns the row. A delivery dropped unsettled releases its lock and closes its connection within five seconds, and its row returns. shutdown releases the lock of every delivery in work and waits for the settlements in flight (Shutdown). Each delivery in work holds a connection of its own until it settles, and a batch holds one row per key (Connections). Each database keeps the locks in its own way:

Postgres locks a 64-bit hash of the key, and a lock lives in the session that took it, so the form needs a direct connection or PgBouncer in session pooling.
MySQL and MariaDB lock a name with GET_LOCK: the database's name, a dot and the key, or its SHA-256 past 64 characters.
SQLite has no locks a session holds, so the process keeps the keys in work, and one process serves a database file. A claim there reads the first due rows alone and takes nothing while they are in work, so a subscription has one delivery in work at a time, and with workers(n) each next row waits up to a poll interval.

By hand, .advisory(..) takes the key as its parts, KeyPart::Literal text and KeyPart::Column columns, and Advisory goes into the description's type.

Example

A video takes minutes to encode, and the videos of one channel go into work one at a time; an encode that fails returns its row with the release of its key.

Macros
Manual

examples/sqlx_forms_advisory.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `video_jobs`, held by a lock on its channel's key while it is in work.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "video_jobs", advisory_lock = "video_jobs-{channel}")]
struct Transcode {
    #[field(id, generated)]
    id: i64,
    channel: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A video takes minutes to encode. The delivery's connection keeps the channel's key while the
/// handler runs, with no transaction open; a retry runs no statement and releases the key.
#[subscriber(InboxQueue::<Transcode>::new("videos"))]
async fn transcode(video: &Video, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if encode(video, attempt).await {
        return HandlerOutcome::ack();
    }
    HandlerOutcome::retry()
}

/// 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, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("transcoder", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Transcode>("videos"), |b| {
            b.include(transcode);
        })
}

examples/manual/sqlx_forms_advisory.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `video_jobs`, held by a lock on its channel's key while it is in work.
#[derive(sqlx::FromRow)]
struct Transcode {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for Transcode {
    type Id = i64;
    type Table = InboxSpec<(Advisory, Attempt, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("video_jobs", Column::new("id").generated())
        .advisory(&[KeyPart::Literal("video_jobs-"), KeyPart::Column("channel")])
        .attempt(Column::new("attempt").generated())
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl AttemptRow for Transcode {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

impl PayloadRow for Transcode {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// A video takes minutes to encode. The delivery's connection keeps the channel's key while the
/// handler runs, with no transaction open; a retry runs no statement and releases the key.
struct TranscodeVideo;

impl Handle<Video, (), (), InboxContext> for TranscodeVideo {
    async fn handle(
        &self,
        video: &Video,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if encode(video, attempt).await {
            return Ok(());
        }
        Err(HandlerOutcome::retry())
    }
}

/// 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, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("transcoder", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Transcode>("videos"), |b| {
            let transcode = subscriber("videos", TranscodeVideo)
                .map_source(|_| InboxQueue::<Transcode>::new("videos"));
            b.include(transcode.build());
        })
}

Example​

Example