retry_after

retry_after(d) hides the row until its retry_after column comes due; a table without the column returns the row at once and logs a warning.

The settlement writes "now" plus d into the row's #[field(retry_after)] column, and the row's next delivery reads one more attempt. A claim takes only rows whose retry_after has come, so the row waits in the table, with no timer in the service and nothing lost if the process stops. The column holds a time type, and "now" comes from the table's clock. A table without the column returns the row at once, and the runtime logs a warning. The backoff is the handler's: it reads the delivery's attempt and answers with the delay it wants. A delayed retry counts against a retry cap as an immediate one does. In a FIFO group the delayed row moves behind the rows due before it. In process the queue reads "now" off the tokio clock, so a test on a paused clock brings the row due with tb.advance(d).

Example

An endpoint that answered "too many requests" is called again a minute later.

Macros
Manual

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

/// One row of `webhook_jobs`: a claim takes it only once its `retry_after` has come.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "webhook_jobs")]
struct Webhook {
    #[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::<Webhook>::new("webhooks"))]
async fn notify(call: &Call, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    match post(call, attempt).await {
        Answer::Done => HandlerOutcome::ack(),
        // The row stays out of the queue as long as the endpoint asked, then a claim takes it
        // again.
        Answer::TooManyRequests { retry_after } => HandlerOutcome::retry_after(retry_after),
    }
}

/// 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("webhooks", "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::<Webhook>("webhooks"), |b| {
            b.include(notify);
        })
}

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

/// One row of `webhook_jobs`: a claim takes it only once its `retry_after` has come.
#[derive(sqlx::FromRow)]
struct Webhook {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for Webhook {
    type Id = i64;
    type Table = InboxSpec<(
        Attempt,
        RetryAfter<DateTime<Utc>>,
        Lease<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("webhook_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 Webhook {
    type Attempt = i16;

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

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

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

struct Notify;

impl Handle<Call, (), (), InboxContext> for Notify {
    async fn handle(
        &self,
        call: &Call,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        match post(call, attempt).await {
            Answer::Done => Ok(()),
            // The row stays out of the queue as long as the endpoint asked, then a claim takes it
            // again.
            Answer::TooManyRequests { retry_after } => {
                Err(HandlerOutcome::retry_after(retry_after))
            }
        }
    }
}

/// 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("webhooks", "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::<Webhook>("webhooks"), |b| {
            let notify = subscriber("webhooks", Notify)
                .map_source(|_| InboxQueue::<Webhook>::new("webhooks"));
            b.include(notify.build());
        })
}

Example​

Example