Skip to main content

Several instances and waking workers

This guide runs a queue on several instances of a service and several workers in each, and sets how fast they see a new job.

A queue table is shared by every instance that reads it. Nothing coordinates the instances but the database: each claims rows, and the form of the table keeps any row in one claim at a time. Two claimers never take the same row, in any form and on any database.

The example renders invoices. The shop writes the invoice jobs with its own SQL, and the rendering service runs as many instances as the load asks for.

The service​

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, Sqlite, SqliteConnection};

/// One row of `invoice_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "invoice_jobs")]
struct RenderInvoice {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<RenderInvoice>::new("invoices"))]
async fn render(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.number, "rendering");
    HandlerOutcome::ack()
}

fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // A row another program writes is seen within a quarter of a second.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new())
        .poll_interval(Duration::from_millis(250))
        .route::<RenderInvoice>("invoices");
    RustStream::new(AppInfo::new("invoices", "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, |b| {
            // Up to four invoices in work at once in this instance.
            b.include(render.workers(nonzero!(4)));
        })
}

workers(4) runs up to four invoices at once in each instance. The subscription keeps a claim in flight for each free worker, so its workers take rows at once instead of one after another. On SQLite, which takes one writer at a time, it keeps one claim in flight, and in the lease form that claim takes a row for every worker. A second instance on the same table adds four more workers, and no invoice goes to both.

How fast a new job is seen​

A subscription claims while its claims come back full, so a backlog drains without a pause. After a claim that found fewer rows than it asked for, it waits the poll interval, one second unless set. Three things bound how long a new row waits:

  • A publish through the broker wakes the subscriptions of its table in the same process at once, at the cost of one atomic operation per woken subscription.
  • A row written by other means waits the poll interval: another program's insert, a trigger, an insert of the service, a publish from another instance. poll_interval(250 ms) here bounds that wait to a quarter of a second, at the price of four claim statements a second per subscription on an idle table.
  • On Postgres, a notification ends the wait.

Notifications on Postgres​

src/main.rs
use sqlx::PgConnection;
use sqlx::postgres::PgPoolOptions;

/// The same service on Postgres: notifications wake it, and the poll each minute reaches a row
/// whose notification was lost.
fn service_on_postgres(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    let broker = SqlxBroker::from_url(url, PgPoolOptions::new())
        .poll_interval(Duration::from_mins(1))
        .listen_notify();
    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        // The table and its trigger, created before the subscription checks them.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect(&schema_url).await?;
            sqlx::raw_sql(POSTGRES_SCHEMA).execute(&mut conn).await?;
            sqlx::raw_sql(ANNOUNCE).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(render.workers(nonzero!(4)));
        })
}

/// The trigger announces every row another program inserts: the channel is the table, and an
/// empty payload wakes every subscription of it.
const ANNOUNCE: &str = "
CREATE OR REPLACE FUNCTION announce_invoice() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(TG_TABLE_NAME, '');
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER invoice_announced AFTER INSERT ON invoice_jobs
    FOR EACH ROW EXECUTE FUNCTION announce_invoice();
";

listen_notify() listens on one connection of the pool, and each subscription listens on its table's channel. Each publish of the broker runs pg_notify(channel, group) right after its write, so a subscription in another instance claims the row once it is visible. A writer outside RustStream sends the same notification itself; the trigger above does it for every insert, and its empty payload wakes every subscription of the table. With notifications the poll interval becomes a floor for what a notification missed, a minute here.

Notifications have a price, and the switch is off unless set. The broker keeps one connection of the pool for its whole life. Each publish runs one more statement. A transaction that notifies takes a lock global to the server at commit, so commits of notifying transactions run one after another; a service with many concurrent writers measures that before it turns notifications on (NOTIFY).

Workers and the pool​

A subscription never holds more than the pool's size less one connection, so the pool always keeps one for the handlers, their publishes and the settlements. A pool too small for its workers runs fewer deliveries at once; it does not fail. In the lease form a pool of n + 1 lets every free worker of workers(n) claim at once; the row lock and advisory lock forms also hold a connection for each job in work (Connections). Running in production sizes the pool for a whole service.

Clocks​

Each instance reads "now" from its host's clock to claim, delay and lease rows (Time). The hosts that serve one table keep their clocks in step, well within a lease, as NTP does. Where they cannot, a table on DatabaseClock reads the database's clock in every statement; the lease form keeps the host's clock, so such a table takes the row lock or the advisory lock form.

Testing​

The tests in the example file run the service on SQLite in memory. Two instances, each with a pool of its own, start on one table that holds twenty invoices, and between them render each invoice once. A row another program writes is seen within the quarter second of the poll interval. A publish through the broker reaches its own instance at once. Those rows wake no subscription, so the tests that wait for them run on the real clock with TestApp::start_live. A last test runs the Postgres service against a database of its own and finds an insert claimed long before the minute's poll; it reads the server's address from POSTGRES_TEST_URL and skips without it.