Attempt

keys::Attempt reads the delivery's attempt, 1 for the first: the attempt column as it stood before the claim, None where the table has none.

A #[field(attempt)] column counts the deliveries of a row, so it starts at 1, usually as the column's default. A retry counts the attempt in the row lock form; the claim counts it in the lease and advisory lock forms, so a crash with the row in work spends one too. Either way a delivery reports the attempt as it stood before its claim, and the first delivery reads 1 in every form. The field is an i16, i32, i64, u16, u32 or u64 (AttemptColumn), and a negative value reads as 0. A column of another type converts into the field through #[sqlx(try_from = "..")], by hand through .attempt_from::<T>(..). A row that does not decode still reports its attempt, read alone, so a retry cap counts it. The backoff is the handler's: it reads the attempt and answers with retry_after of the delay it wants. By hand, .attempt(..) names the column, Attempt goes into the description's type, and the row reads its attempt through AttemptRow.

Example

A supplier's API answers from the third attempt on, and the handler waits twice as long before each new one: one second, then two.

Macros
Manual

examples/sqlx_context_attempt.rs
use std::time::Duration;

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `stock_jobs`: `attempt` counts the deliveries, 1 for the first.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "stock_jobs")]
struct SyncStock {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SyncStock>::new("stock"))]
async fn sync(item: &Item, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    let attempt = attempt.unwrap_or(1);
    if fetch_level(item, attempt).await {
        return HandlerOutcome::ack();
    }
    // Doubling backoff: 1 s, 2 s, 4 s, ...
    let exponent = u32::try_from(attempt.min(16)).unwrap_or(16);
    HandlerOutcome::retry_after(Duration::from_secs(1 << (exponent - 1)))
}

/// 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("stock", "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::<SyncStock>("stock"), |b| {
            b.include(sync);
        })
}

examples/manual/sqlx_context_attempt.rs
use std::time::Duration;

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `stock_jobs`: `attempt` counts the deliveries, 1 for the first.
#[derive(sqlx::FromRow)]
struct SyncStock {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

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

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

impl AttemptRow for SyncStock {
    type Attempt = i16;

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

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

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

struct FetchLevel;

impl Handle<Item, (), (), InboxContext> for FetchLevel {
    async fn handle(
        &self,
        item: &Item,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        let attempt = attempt.unwrap_or(1);
        if fetch_level(item, attempt).await {
            return Ok(());
        }
        // Doubling backoff: 1 s, 2 s, 4 s, ...
        let exponent = u32::try_from(attempt.min(16)).unwrap_or(16);
        Err(HandlerOutcome::retry_after(Duration::from_secs(
            1 << (exponent - 1),
        )))
    }
}

/// 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("stock", "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::<SyncStock>("stock"), |b| {
            let sync = subscriber("stock", FetchLevel)
                .map_source(|_| InboxQueue::<SyncStock>::new("stock"));
            b.include(sync.build());
        })
}

Example​

Example