Lease

#[field(locked_until)] selects the lease form: the claim writes an expiry and commits, the handler holds no connection, and the lease is extended.

A claim writes the lease's expiry into locked_until, counts the attempt and commits at once, so no transaction stays open while the handler runs. The claim reads "now" once: the rows it finds due, the leases it finds ended and the expiry it writes start from that instant. The expiry is the delivery's ownership token: a settlement takes effect only while the row still holds it. While the handler runs, the subscription extends the lease of every delivery in work each half lease, and each extension moves the token forward, so a handler may run longer than its lease. A lease is how long a crashed process's row stays out of the queue: thirty seconds unless SqlxBroker::lease sets another for the broker or InboxQueue::lease for one subscription. It is whole seconds, at least one, and a shorter one is rounded up. A lease runs out under a running handler only when its extensions stop: the database out of reach, the subscription closed, the broker shut down. The next claim then takes the row, and the late holder's settlement fails with SqlxBrokerError::LeaseLost. A delivery dropped unsettled releases its row at once; a settlement whose statement fails keeps the row until its lease runs out, as a crash does. The claim counts the attempt, so a crash spends one too, and a delivery reports the attempt as it stood before its claim. After the broker shuts down, a delivery in work keeps its lease and settles as before, and the lease is no longer extended. The crate computes the expiry on the host, so the form reads the host's clock: locked_until beside clock = DatabaseClock does not compile. The column holds a time type, Option<DateTime<Utc>> here. By hand, .lease(..) names the column, and Lease<Time> in the description's type gives the lease's time type (LeaseRow).

Example

A report takes minutes to render under a one-minute lease, which the subscription extends while the handler runs.

Macros
Manual

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

/// One row of `report_jobs`: `locked_until` makes the table a lease table.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
struct RenderReport {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A report takes minutes to render. The subscription extends the minute's lease every thirty
/// seconds while the handler runs; after a crash the row waits out its minute, then returns.
#[subscriber(InboxQueue::<RenderReport>::new("reports").lease(Duration::from_mins(1)))]
async fn render(report: &Report) -> HandlerOutcome {
    draw(report).await;
    tracing::info!(report.id, "report rendered");
    HandlerOutcome::ack()
}

/// 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("reports", "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::<RenderReport>("reports"), |b| {
            b.include(render);
        })
}

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

/// One row of `report_jobs`: `.lease(..)` makes the table a lease table.
#[derive(sqlx::FromRow)]
struct RenderReport {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

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

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

impl AttemptRow for RenderReport {
    type Attempt = i16;

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

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

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

/// A report takes minutes to render. The subscription extends the minute's lease every thirty
/// seconds while the handler runs; after a crash the row waits out its minute, then returns.
struct Render;

impl Handle<Report> for Render {
    async fn handle(
        &self,
        report: &Report,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        draw(report).await;
        tracing::info!(report.id, "report rendered");
        Ok(())
    }
}

/// 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("reports", "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::<RenderReport>("reports"), |b| {
            let render = subscriber("reports", Render).map_source(|_| {
                InboxQueue::<RenderReport>::new("reports").lease(Duration::from_mins(1))
            });
            b.include(render.build());
        })
}

Example​

Example