Outcomes

A handler's outcome is a statement on its row: ack deletes it or sets processed_at, retry returns it at once with one more attempt, drop finishes it.

What a handler answers decides the row's fate:

Outcome	The row
`ack()`	is deleted, or marked with the current time where the table has a `processed_at` column
`retry()`	returns to the queue at once, and its next delivery reads one more attempt
`retry_after(d)`	stays out of the queue until its `retry_after` comes
`drop()`	is finished as an acknowledgement finishes it

A row marked with processed_at stays in the table, out of every claim, for an audit or a later cleanup. The statement runs where the form holds the row: in the claim's transaction, by the delivery's lease, or on the session that holds its key. It takes effect together with the end of that hold, so a row and its outcome never part. A settlement whose statement fails reports itself, and the row returns as it would after a crash. max_attempts(n) with dead_letter(..) caps the retries and moves a spent row aside. A table may run its own SQL for an outcome, an acknowledgement that moves the row to a sent group for one (Events); it runs in the transaction the crate commits, as the built-in statement does. An acknowledgement takes the row out of what the claim selects, so a statement of one's own that leaves the row claimable delivers it again.

Example

The mail server greylists a first delivery, so the handler retries it at once; an address that cannot receive mail is dropped, and both end marked in processed_at.

Macros
Manual

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

/// One row of `mail_jobs`: `processed_at` keeps a finished row in the table, marked.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "mail_jobs")]
struct SendMail {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendMail>::new("mail"))]
async fn send(mail: &Mail, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if !mail.to.contains('@') {
        // Nothing to retry: the row is finished, as an acknowledgement finishes it.
        return HandlerOutcome::drop();
    }
    if deliver(mail, attempt).await {
        // `processed_at` is set, and the row stays.
        return HandlerOutcome::ack();
    }
    // Back in the queue at once; the next delivery reads one more attempt.
    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("mailer", "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::<SendMail>("mail"), |b| {
            b.include(send);
        })
}

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

/// One row of `mail_jobs`: `processed_at` keeps a finished row in the table, marked.
#[derive(sqlx::FromRow)]
struct SendMail {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendMail {
    type Id = i64;
    type Table = InboxSpec<(
        Attempt,
        Lease<DateTime<Utc>>,
        ProcessedAt<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("mail_jobs", Column::new("id").generated())
        .attempt(Column::new("attempt").generated())
        .lease(Column::new("locked_until"))
        .processed_at(Column::new("processed_at"))
        .payload(Column::new("payload"));

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

impl AttemptRow for SendMail {
    type Attempt = i16;

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

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

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

struct SendOne;

impl Handle<Mail, (), (), InboxContext> for SendOne {
    async fn handle(
        &self,
        mail: &Mail,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        if !mail.to.contains('@') {
            // Nothing to retry: the row is finished, as an acknowledgement finishes it.
            return Err(HandlerOutcome::drop());
        }
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if deliver(mail, attempt).await {
            // `processed_at` is set, and the row stays.
            return Ok(());
        }
        // Back in the queue at once; the next delivery reads one more attempt.
        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("mailer", "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::<SendMail>("mail"), |b| {
            let send =
                subscriber("mail", SendOne).map_source(|_| InboxQueue::<SendMail>::new("mail"));
            b.include(send.build());
        })
}

Example​

Example