Time

A queue reads "now" to claim, delay and mark rows; by default the host's SystemClock, bound in the column's own type as the service's inserts do.

A claim compares "now" with retry_after and with the leases in work, a delayed retry writes "now" plus its delay, and an acknowledgement writes "now" into processed_at. A table that names no clock reads the host's system clock. The crate binds the time in the column's own type, so sqlx encodes it exactly as the service's own statements write times. The clocks of the hosts that serve one table are kept in step, well within a lease; keeping them in step is the hosts' concern, as NTP does. Where the hosts' clocks cannot be trusted, DatabaseClock makes the statements read the database's clock instead. A clock of the service's own takes the same place (Clock). Both are TimeSources: a host clock binds "now" into the statement, and the database's clock lets the statement read its own. In process a table on the system clock reads the tokio clock, anchored to the system clock when the broker connected. So a test on a paused clock moves the queue's time with tb.advance(d): rows come due and marks move with it.

Example

An invoice the mail server cannot take returns in an hour by the host's clock, and a sent one is marked with the host's time.

Macros
Manual

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

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

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

#[subscriber(InboxQueue::<SendInvoice>::new("invoices"))]
async fn send(invoice: &Invoice) -> HandlerOutcome {
    if deliver(invoice).await {
        // `processed_at` takes the host's time.
        return HandlerOutcome::ack();
    }
    // An hour by the host's clock: `retry_after` takes now plus an hour.
    HandlerOutcome::retry_after(Duration::from_hours(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("billing", "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::<SendInvoice>("invoices"), |b| {
            b.include(send);
        })
}

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

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `invoice_jobs`. No `.clock::<..>()`: the table reads `SystemClock`.
#[derive(sqlx::FromRow)]
struct SendInvoice {
    id: i64,
    payload: Vec<u8>,
}

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

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

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

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

struct SendOne;

impl Handle<Invoice> for SendOne {
    async fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if deliver(invoice).await {
            // `processed_at` takes the host's time.
            return Ok(());
        }
        // An hour by the host's clock: `retry_after` takes now plus an hour.
        Err(HandlerOutcome::retry_after(Duration::from_hours(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("billing", "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::<SendInvoice>("invoices"), |b| {
            let send = subscriber("invoices", SendOne)
                .map_source(|_| InboxQueue::<SendInvoice>::new("invoices"));
            b.include(send.build());
        })
}

Example​

Example