Connections

Each message in work holds pool connections by its form; workers(n) claims in parallel but always leaves the pool one connection for handlers.

A subscription mounted with workers(n) keeps a claim in flight for each free worker. It never holds more than the pool's size less one connection, so the pool always keeps one for the handlers' own queries, their publishes and the settlements that take one. A second claim starts only while the pool has a connection to spare, so subscriptions that share a pool leave each other room, and a pool short of room runs fewer deliveries at once. A subscription keeps one claim in flight on SQLite, which takes one writer at a time, on a table whose groups keep their order, and with a partition_key column under workers_by_key(n). On SQLite that one claim takes a row for every worker in the lease form; in transactional mode each of those rows holds its transaction's connection while it waits for a worker, within the connections the subscription may hold.

What a delivery holds depends on the form:

Form	Connections
row lock	one per delivery in work and one per claim in flight, for the whole handler
lease	one for each claim's and each settlement's statement, and one per subscription each half lease to extend the leases
advisory lock	one per delivery in work until it settles, in a batch too
transactional mode, any form	one per delivery in work, for the transaction its handler writes through

A publish takes a connection of its own for its insert, and a handler that reads keys::Pool takes one for each query. So a pool is sized for every subscription: n for each workers(n), one more for the pool, and the connections the handlers take for their own work. In the lease form a pool of n + 1 lets every free worker claim at once. Under sqlx's default test_before_acquire, each acquire pings the connection first, a round trip a pool built with test_before_acquire(false) skips. Subscriptions on dedicated threads open connections of their own on top of the pool.

Example

Four workers resize images, and each handler records its image through the pool; the pool has room for the four workers, the connection the subscription leaves free and the handlers' writes.

Macros

examples/sqlx_broker_connections.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "resize_jobs")]
struct ResizeImage {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

// The handler takes a connection of the pool for its own write.
#[subscriber(InboxQueue::<ResizeImage>::new("resizes"))]
async fn resize(image: &Image, Ctx(pool): Ctx<keys::Pool<Sqlite>>) -> HandlerOutcome {
    let recorded = sqlx::query("INSERT INTO resized (path) VALUES (?)")
        .bind(&image.path)
        .execute(&pool)
        .await;
    match recorded {
        Ok(_) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "resizes", path = %image.path, %error, "not recorded");
            HandlerOutcome::retry()
        }
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Four workers, one connection the subscription always leaves free, and one for the
    // handlers' own writes and the publishes.
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new().max_connections(6));
    RustStream::new(AppInfo::new("resizer", "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::<ResizeImage>("resizes"), |b| {
            b.include(resize.workers(nonzero!(4)));
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:resizes.db?mode=rwc")
}

Manual

examples/manual/sqlx_broker_connections.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;

struct Resize;

// The handler takes a connection of the pool for its own write.
impl Handle<Image, (), (), PoolContext<Sqlite>> for Resize {
    async fn handle(
        &self,
        image: &Image,
        _outs: &(),
        ctx: &mut Context<'_, PoolContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ok(Ctx(pool)) = Ctx::<keys::Pool<Sqlite>>::from_context(ctx).await;
        let recorded = sqlx::query("INSERT INTO resized (path) VALUES (?)")
            .bind(&image.path)
            .execute(&pool)
            .await;
        recorded.map(drop).map_err(|error| {
            tracing::warn!(subscription = "resizes", path = %image.path, %error, "not recorded");
            HandlerOutcome::retry()
        })
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Four workers, one connection the subscription always leaves free, and one for the
    // handlers' own writes and the publishes.
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new().max_connections(6));
    RustStream::new(AppInfo::new("resizer", "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::<ResizeImage>("resizes"), |b| {
            let resize = subscriber("resizes", Resize)
                .map_source(|_| InboxQueue::<ResizeImage>::new("resizes"))
                .workers(nonzero!(4));
            b.include(resize.build());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:resizes.db?mode=rwc")
}