在生产环境中运行

本指南把一个队列带到 Postgres 上的生产环境：表结构及其索引、发现错误表结构的检查、连接池、数据库版本与连接池代理，以及关闭。

示例服务是一个发信服务，有八个 worker、五次尝试的上限和一个死信组。

表结构

src/main.rs

/// The service's first migration; in a service crate, `migrations/0001_email_jobs.sql`, run by
/// `sqlx::migrate!`.
const MIGRATION: &str = "
CREATE TABLE IF NOT EXISTS email_jobs (
    job_id  BIGSERIAL PRIMARY KEY,
    name    TEXT NOT NULL,
    attempt SMALLINT NOT NULL DEFAULT 1,
    payload BYTEA NOT NULL
);
CREATE INDEX IF NOT EXISTS email_jobs_claim ON email_jobs (name, job_id);
";

本 crate 不创建表：表结构属于服务，由服务的迁移创建，examples/outbox 中用 sqlx::migrate! 运行这些迁移。索引遵循领取顺序，即表中已有的列按 (group, priority, retry_after, id) 排列，这里是 (name, job_id)。领取在排序之前就锁住它读到的组中每一行，所以没有这个索引时，为处理器保持的领取会挡住组中其余的行，其他领取取不到它们。在带 FIFO 组的表中，同一个索引让领取对组的读取只涉及组自己的行。

受检语句

宏

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Postgres;

/// One row of `email_jobs`: every statement it determines goes through sqlx's compile-time check.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", checked, db = postgres)]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

手写

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Postgres;

/// One row of `email_jobs`, claimed by row lock: the description names no other form. Each
/// subscription prepares its statements at startup; the compile-time check is the derive's.
#[derive(sqlx::FromRow)]
struct SendEmail {
    job_id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
        .group(Column::new("name"))
        .attempt(Column::new("attempt").generated())
        .payload(Column::new("payload"));

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

impl AttemptRow for SendEmail {
    type Attempt = i16;

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

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

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

checked, db = postgres 把结构体确定的每条语句交给 sqlx 的编译期检查：领取、各种结算、组的死信转移和插入。结构体指名而表中缺少的列会让构建失败。构建对照 DATABASE_URL 检查，或者对照 cargo sqlx prepare 写出、与 crate 一起提交的 .sqlx 数据检查；CI 中的 cargo sqlx prepare --check 会让数据过期的构建失败。订阅执行的语句与没有 checked 时相同，所以每条消息的开销也相同。这项检查属于派生宏：手写描述的表只有启动检查。

每个订阅还会在启动时预备它的语句。数据库缺少的表或列会让服务以 SqlxBrokerError::Schema 停止，错误中写明订阅、表和语句。声明了上限却没有死信的挂载，或者反过来，也会让服务停止。所以 on_startup 在订阅打开之前运行迁移，这些检查看到的就是这个版本所读取的表结构。

连接池

宏

src/main.rs

use std::num::NonZeroUsize;
use std::time::Duration;

use sqlx::Connection;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

/// The emails in work at once; each holds its row's lock and a connection until it settles.
const WORKERS: NonZeroUsize = nonzero!(8);

/// A connection for each worker, and one the subscription leaves to settlements and publishes.
fn pool() -> PgPoolOptions {
    let size = u32::try_from(WORKERS.get() + 1).unwrap_or(u32::MAX);
    PgPoolOptions::new()
        .max_connections(size)
        .acquire_timeout(Duration::from_secs(5))
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    tracing::info!(to = %email.to, ?attempt, "sending");
    HandlerOutcome::ack()
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let migrate = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, pool()).route::<SendEmail>("emails");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The schema the release reads exists before any subscription prepares its statements.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&migrate).await?;
            sqlx::raw_sql(MIGRATION).execute(&mut conn).await?;
            conn.close().await
        })
        // The label names the server in the generated AsyncAPI document.
        .with_broker_labeled("postgres", broker, |b| {
            b.include(send.workers(WORKERS))
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
        })
}

手写

src/main.rs

use std::num::NonZeroUsize;
use std::time::Duration;

use sqlx::Connection;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

/// The emails in work at once; each holds its row's lock and a connection until it settles.
const WORKERS: NonZeroUsize = nonzero!(8);

/// A connection for each worker, and one the subscription leaves to settlements and publishes.
fn pool() -> PgPoolOptions {
    let size = u32::try_from(WORKERS.get() + 1).unwrap_or(u32::MAX);
    PgPoolOptions::new()
        .max_connections(size)
        .acquire_timeout(Duration::from_secs(5))
}

struct Mailer;

impl Handle<Email, (), (), InboxContext> for Mailer {
    async fn handle(
        &self,
        email: &Email,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        tracing::info!(to = %email.to, ?attempt, "sending");
        Ok(())
    }
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let migrate = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, pool()).route::<SendEmail>("emails");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The schema the release reads exists before any subscription prepares its statements.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&migrate).await?;
            sqlx::raw_sql(MIGRATION).execute(&mut conn).await?;
            conn.close().await
        })
        // The label names the server in the generated AsyncAPI document.
        .with_broker_labeled("postgres", broker, |b| {
            let send = subscriber("emails", Mailer)
                .map_source(|_| InboxQueue::<SendEmail>::new("emails"))
                .workers(WORKERS);
            b.include(send.build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
        })
}

这张表没有指定领取方式，所以它在行锁下领取：每封处理中的邮件在结算之前都持有它所在行的锁和一个连接。一个订阅占用的连接从不超过连接池大小减一，留下的那一个给处理器自己的查询、它们的发布和结算。因此连接池要按服务的全部订阅来确定大小：每个 workers(n) 计 n 个，再加一个，再加上处理器为自己的工作取用的连接（连接）。租约表只在执行每条语句时占用连接，适合无法为每个任务留出一个连接的连接池。在 Postgres 上，通知在 Broker 的整个生命周期内再多占用一个连接。

宏

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    // The connection's settings come from the `PG*` environment variables.
    service(PgConnectOptions::new())
}

手写

src/main.rs

fn main() -> ExitCode {
    // The connection's settings come from the `PG*` environment variables.
    run_main(|| service(PgConnectOptions::new()))
}

数据库与连接池代理

数据库	要求
Postgres	SKIP LOCKED 需要 9.5；咨询锁方式和 FIFO 组需要 11，它们锁住 64 位哈希
MySQL、MariaDB	MySQL 8.0.1 或 MariaDB 10.6，在启动时读取；更旧的服务器会让服务以 ServerTooOld 停止
SQLite	租约方式或咨询锁方式；在咨询锁方式下，每个数据库文件一个进程

在 MySQL 和 MariaDB 上，领取以 READ COMMITTED 运行，除非行锁表声明了其他级别。以语句格式记录日志的服务器，即 binlog_format = STATEMENT，会拒绝这个级别下的写入；行格式和混合格式接受这些写入。在 Postgres 上，read_committed 是行锁表实际可用的隔离级别：更严格的级别在领取与结算同时运行时，会让领取以串行化错误失败。咨询锁方式把锁保存在数据库会话中，所以它需要直接连接，或者会话模式的 PgBouncer；事务模式的连接池代理会在两条语句之间把会话交给另一个客户端。启用 listen_notify 的 Broker 同样在一个会话上监听。

关闭

关闭时，运行时停止各订阅，在它的关闭超时内让处理中的处理器完成，然后关闭 Broker。行锁方式的投递在结算之前一直持有它的事务；进程退出时仍未结算的投递随它的连接一起回滚，它的行回到队列。租约方式的投递保留它的租约，能结算就结算，否则它的行在租约到期后回来。Broker 释放处理中投递的咨询锁。Broker 最后关闭它构建的连接池；处理中的投递仍持有的连接在归还时关闭。

AsyncAPI 文档

启用 asyncapi feature 后，生成的 AsyncAPI 文档描述服务器，以及每个订阅的表和组。with_broker_labeled 的标签给出服务器的名称，服务器条目中是数据库的主机和端口，从不包含用户名或密码：这份文档会被发布和分享。

测试

示例文件中的一个测试在一个带密码的 URL 上构建服务的这份文档，不做任何 I/O，发现其中有主机和端口，而任何地方都没有密码。另外两个测试连接它们自己的 Postgres 数据库运行：旧版本的表缺少结构体指名的列，让启动以一个指名该表的错误停止，受检语句则正常服务队列。它们从 POSTGRES_TEST_URL 读取服务器地址，没有这个变量时跳过（在 SQL 上测试服务）。