DatabaseClock

clock = DatabaseClock makes the statements read the database's clock, so every host serving one table reads one clock; the lease form keeps the host's.

With DatabaseClock the statements bind no time: each reads the database's own. Postgres reads statement_timestamp(), MySQL and MariaDB UTC_TIMESTAMP(6), and SQLite the current time as text. Every host that claims, delays or marks a row of the table then reads one clock, whatever its own says. The lease form reads the host's clock, because each settlement names the expiry its claim wrote, so locked_until beside clock = DatabaseClock does not compile. A table on the database's clock takes the row lock or the advisory lock form; on SQLite, which has no row locks, the advisory lock form. In process too the statements read the database's clock, which a paused tokio clock does not move, so a test of a delayed retry on such a table runs on the real clock with TestApp::start_live. By hand, .clock::<DatabaseClock>() names it, and Clock<DatabaseClock> goes into the description's type.

Example

Statements written by hosts whose clocks drift wait a second by the database's clock when the month's figures are not closed yet.

Macros
Manual

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

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

/// One row of `statement_jobs`, written by hosts whose clocks drift. The advisory lock form,
/// because the lease form reads the host's clock.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(
    table = "statement_jobs",
    clock = DatabaseClock,
    advisory_lock = "statement_jobs-{id}"
)]
struct SendStatement {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendStatement>::new("statements"))]
async fn send(account: &Account, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if render(account, attempt).await {
        return HandlerOutcome::ack();
    }
    // A second by the database's clock, whichever host claimed the row.
    HandlerOutcome::retry_after(Duration::from_secs(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("statements", "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::<SendStatement>("statements"), |b| {
            b.include(send);
        })
}

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

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

/// One row of `statement_jobs`, written by hosts whose clocks drift. The advisory lock form,
/// because the lease form reads the host's clock.
#[derive(sqlx::FromRow)]
struct SendStatement {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendStatement {
    type Id = i64;
    type Table = InboxSpec<(
        Clock<DatabaseClock>,
        Advisory,
        Attempt,
        RetryAfter<DateTime<Utc>>,
        ProcessedAt<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("statement_jobs", Column::new("id").generated())
        .clock::<DatabaseClock>()
        .advisory(&[KeyPart::Literal("statement_jobs-"), KeyPart::Column("id")])
        .attempt(Column::new("attempt").generated())
        .retry_after(Column::new("retry_after"))
        .processed_at(Column::new("processed_at"))
        .payload(Column::new("payload"));

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

impl AttemptRow for SendStatement {
    type Attempt = i16;

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

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

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

struct SendOne;

impl Handle<Account, (), (), InboxContext> for SendOne {
    async fn handle(
        &self,
        account: &Account,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if render(account, attempt).await {
            return Ok(());
        }
        // A second by the database's clock, whichever host claimed the row.
        Err(HandlerOutcome::retry_after(Duration::from_secs(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("statements", "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::<SendStatement>("statements"), |b| {
            let send = subscriber("statements", SendOne)
                .map_source(|_| InboxQueue::<SendStatement>::new("statements"));
            b.include(send.build());
        })
}

Example​

Example