顺序：FIFO 组与分区键

本指南保持任务的顺序：一个组的全部任务，或者每个键的任务，同时让其他键并行运行。

SKIP LOCKED 让领取很快，因为它跳过处理中的行。它也让下一行在前一行仍在处理或等待重试时就被取走。任务必须按顺序生效的队列，要在它的表中声明这一点。

示例是一个账户服务。它把联系人变更发送到 CRM，CRM 必须按变更发生的顺序接收它们；它还记入账簿分录，分录必须在一个账户内保持顺序。

表

src/main.rs

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS crm_jobs (
    id INTEGER PRIMARY KEY,
    target TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE INDEX IF NOT EXISTS crm_jobs_claim ON crm_jobs (target, id);
CREATE TABLE IF NOT EXISTS ledger_jobs (
    id INTEGER PRIMARY KEY,
    account TEXT NOT NULL,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";

crm_jobs 上的索引遵循领取顺序，组在最前。FIFO 组的领取会读取该组未完成的行，索引让这次读取只涉及这些行（在生产环境中运行）。

FIFO 组

宏

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `crm_jobs`; each group holds the changes bound for one system, in order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "crm_jobs")]
struct CrmChange {
    #[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::<CrmChange>::new("crm"))]
async fn sync(change: &Contact) -> HandlerOutcome {
    if push(change).await {
        return HandlerOutcome::ack();
    }
    // At once: the change keeps its place at the head of the group.
    HandlerOutcome::retry()
}

手写

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

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

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

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

impl PayloadRow for CrmChange {
    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 PushToCrm;

impl Handle<Contact> for PushToCrm {
    async fn handle(
        &self,
        change: &Contact,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if push(change).await {
            return Ok(());
        }
        // At once: the change keeps its place at the head of the group.
        Err(HandlerOutcome::retry())
    }
}

#[field(group, fifo = true)] 让 target 的每个组成为 FIFO 组。无论有多少个 worker、订阅和进程在读取，一个 FIFO 组最多只有一行处于处理中。按领取顺序，组中第一个未完成的行是它的首行，领取只取首行。首行处于处理中时，后面的行都要等待，首行处在另一个实例的租约下时也是如此。retry() 让这次变更保持在首位，所以它会先于后面的变更再次发出。retry_after(d) 则会把它移到比它更早到期的行之后。

一直失败的首行会阻塞整个组。挂载处为它设置上限：

宏

src/main.rs

/// 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());
    let broker = broker
        .route::<CrmChange>("crm")
        .route::<LedgerPosting>("postings");
    RustStream::new(AppInfo::new("accounts", "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, |b| {
            // A change the CRM refuses three times moves out of the group, which then moves on.
            b.include(sync)
                .max_attempts(nonzero!(3u32))
                .dead_letter("crm.parked");
            // Four workers, and the postings of one account on one of them, one after another.
            b.include(post.workers_by_key(nonzero!(4)));
        })
}

手写

src/main.rs

/// 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());
    let broker = broker
        .route::<CrmChange>("crm")
        .route::<LedgerPosting>("postings");
    RustStream::new(AppInfo::new("accounts", "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, |b| {
            // A change the CRM refuses three times moves out of the group, which then moves on.
            let sync =
                subscriber("crm", PushToCrm).map_source(|_| InboxQueue::<CrmChange>::new("crm"));
            b.include(sync.build())
                .max_attempts(nonzero!(3u32))
                .dead_letter("crm.parked");
            // Four workers, and the postings of one account on one of them, one after another.
            let post = subscriber("postings", Post)
                .map_source(|_| InboxQueue::<LedgerPosting>::new("postings"))
                .workers_by_key(nonzero!(4));
            b.include(post.build());
        })
}

被拒绝三次之后，这次变更转入 crm.parked 组，crm 的下一次变更成为首行。被搁置的变更在那里等人处理；它后面的变更已经在没有它的情况下发出，这就是释放这个组的代价。

各个组并行运行，每个组通过自己的订阅处理，所以组适合少数事先已知的流：每个下游系统一个，或每个租户一个。workers(n) 不会让 FIFO 订阅并行，它由一个 worker 服务。

每个账户一个键

账簿有很多账户，多到无法为每个账户建一个订阅。要紧的顺序在单个账户之内，不同账户可以并行运行。

宏

src/main.rs

/// One row of `ledger_jobs`: the postings of one account apply in the order they were written.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "ledger_jobs")]
struct LedgerPosting {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    account: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<LedgerPosting>::new("postings"))]
async fn post(posting: &Posting) -> HandlerOutcome {
    tracing::info!(account = %posting.account, posting.n, "posting");
    HandlerOutcome::ack()
}

手写

src/main.rs

/// One row of `ledger_jobs`: the postings of one account apply in the order they were written.
#[derive(sqlx::FromRow)]
struct LedgerPosting {
    id: i64,
    account: String,
    payload: Vec<u8>,
}

impl InboxTable for LedgerPosting {
    type Id = i64;
    type Table = InboxSpec<(Key, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("ledger_jobs", Column::new("id").generated())
        .partition_key(Column::new("account"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

impl KeyRow for LedgerPosting {
    type Key = String;

    fn partition_key(&self) -> &String {
        &self.account
    }
}

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

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

struct Post;

impl Handle<Posting> for Post {
    fn handle(
        &self,
        posting: &Posting,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(account = %posting.account, posting.n, "posting");
        ready(Ok(()))
    }
}

#[field(partition_key)] 把账户作为每一行的键。应用用 workers_by_key(4) 挂载 post：四个 worker，一个账户的分录都在其中一个 worker 上，按领取的顺序运行。在这样的表上，订阅只保持一次进行中的领取，所以各次领取不会互相超越。这个顺序在一个订阅之内成立。一个账户的分录如果必须在多个实例之间保持顺序，就放进 FIFO 组，或者采用键取账户的咨询锁方式，它让同一个键的行在同一服务器上的所有进程之间逐个进入处理。

测试

示例文件中的测试在内存中的 SQLite 上运行服务。第一个测试发布一次 CRM 拒绝的变更和一次它接受的变更：第一次变更发出三次后被搁置，然后第二次变更发出。第二个测试把 crm 的首行写成由另一个实例租用，发现它后面的变更在等待。第三个测试交错写入两个账户的分录，发现每个账户的分录都按顺序记入。后两个测试的行不唤醒任何订阅，所以它们用 TestApp::start_live 在真实时钟上运行，越过轮询间隔。