Startup

A subscription builds and prepares its statements when it opens, so a missing table or column, a retry the table cannot carry or a second reader stops startup.

A struct that cannot drive a queue does not compile at all; the derive lists those errors. What only the database can tell is checked at startup. Each subscription builds its statements with the broker's dialect and prepares each one on the server. A table or a column the description names and the database lacks fails startup with SqlxBrokerError::Schema, which names the subscription, the table and the statement. Preparing checks names, not column types: a row whose columns do not decode reaches the decode policy instead.

The other refusals at startup:

Error	When
`Declaration`	the mount declares what the table cannot carry: `max_attempts(n)` without `dead_letter(..)` or the reverse, a cap on a table without an `attempt` column, or transactional mode in the lease form at a strict isolation level on Postgres
`Dialect`	the dialect cannot build a statement the table needs
`ServerTooOld`	a MySQL older than 8.0.1 or a MariaDB older than 10.6, which lack `SKIP LOCKED`
`AlreadySubscribed`	a second subscription to one queue on the same broker; a table without a `group` column is one queue
`Connect`, `Backend`	the broker could not take a connection, or an `AnyPool` reached a database whose dialect feature is off

The service runs its migrations before the app starts, and the checks then see the schema the release reads.

Example

A release adds an attempt column to the queue; on_startup applies the schema before the subscriptions open, and a table without the column stops startup with SqlxBrokerError::Schema.

Macros
Manual

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

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "invoice_jobs")]
struct SendInvoice {
    #[field(id, generated)]
    id: i64,
    // The column this release adds; a table without it stops the subscription at startup.
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendInvoice>::new("invoices"))]
async fn send(invoice: &Invoice, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    tracing::info!(invoice.number, ?attempt, "sending an invoice");
    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, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        // The schema this release reads, applied before the subscriptions check 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::<SendInvoice>("invoices"), |b| {
            b.include(send);
        })
}

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

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

#[derive(sqlx::FromRow)]
struct SendInvoice {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendInvoice {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
    // The column this release adds; a table without it stops the subscription at startup.
    const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
        .attempt(Column::new("attempt").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

impl AttemptRow for SendInvoice {
    type Attempt = i16;

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

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

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

struct Mail;

impl Handle<Invoice, (), (), InboxContext> for Mail {
    async fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ok(Ctx(attempt)) = Ctx::<keys::Attempt>::from_context(ctx).await;
        tracing::info!(invoice.number, ?attempt, "sending an invoice");
        Ok(())
    }
}

/// 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("billing", "0.1.0"))
        // The schema this release reads, applied before the subscriptions check 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::<SendInvoice>("invoices"), |b| {
            let send = subscriber("invoices", Mail)
                .map_source(|_| InboxQueue::<SendInvoice>::new("invoices"));
            b.include(send.build());
        })
}

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

Example​

Example