FIFO

#[field(group, fifo = true)] keeps each group in order: one row of a group in work at a time across every broker, the head first.

A FIFO group has at most one row in work, however many brokers and processes read it. Its rows go out in claim order: a smaller priority first, then an earlier retry_after, then the smaller id, each where the table has the column. The group's first unfinished row in that order is its head, and a claim takes the head alone. The rows behind the head wait while it is in work or not yet due. retry() keeps a row at the head, so a change that failed goes again before the ones behind it. retry_after(d) gives the row a later retry_after, and it moves behind the rows due before it; on a table with priority it moves only among the rows of its own priority. A head that keeps failing holds its group back, so a FIFO table pairs with a retry cap that moves such a row out of the group. In the lease form a crash with a row in work holds its group until that row's lease runs out. workers(n) gives no parallelism inside a group, so one worker serves a FIFO subscription, and a batch of a FIFO group holds one row. Groups run side by side, each through a subscription of its own. FIFO groups serve the row lock and the lease forms. In the advisory lock form a key on the group's field keeps one row of the group in work, and fifo = true beside advisory_lock does not compile. By hand, .fifo_group(..) names the group's column, and Fifo goes into the description's type.

Example

The CRM receives its changes one at a time, in the order they were queued, and a change that timed out goes again before the next one.

Macros
Manual

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

/// One row of `sync_jobs`; each group holds the changes bound for one system, in order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "sync_jobs")]
struct SyncChange {
    #[field(id, generated)]
    id: i64,
    #[field(group, fifo = true)]
    target: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The CRM receives its changes one at a time, in the order they were queued.
#[subscriber(InboxQueue::<SyncChange>::new("crm"))]
async fn sync(change: &Change, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if push(change, attempt).await {
        return HandlerOutcome::ack();
    }
    // At once, so the change keeps its place at the head of the group.
    HandlerOutcome::retry()
}

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

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

/// One row of `sync_jobs`; each group holds the changes bound for one system, in order.
#[derive(sqlx::FromRow)]
struct SyncChange {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

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

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

impl AttemptRow for SyncChange {
    type Attempt = i16;

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

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

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

/// The CRM receives its changes one at a time, in the order they were queued.
struct PushChange;

impl Handle<Change, (), (), InboxContext> for PushChange {
    async fn handle(
        &self,
        change: &Change,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if push(change, attempt).await {
            return Ok(());
        }
        // At once, so the change keeps its place at the head of the group.
        Err(HandlerOutcome::retry())
    }
}

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

Example​

Example