Rust 版 River 实战指南：基于 Postgres/SQLite 的 Tokio 后台任务队列

River 是一个由 Postgres 或 SQLite 支撑的快速、可靠的后台任务（background job）系统，其 Rust 实现基于 Tokio，并与 Go、Ruby、JS/TS 实现共享同一套数据库 Schema 与任务协议，因此不同语言的多个服务可以在同一个数据库里互相插入与消费任务。本文以仓库中的 rust/README.md 为主体，结合 riverqueue 客户端指南、riverqueue-cli、riverqueue-macros、riverqueue-migrate、riverqueue-test 等文档及源码实现，系统讲解依赖配置、快速开始、事务语义、任务与队列管理、Worker 生命周期、可靠性特性、领导选举与维护、CLI 迁移与基准测试，以及 Rust 与 Go 混合部署的要点。读完本文，你将具备从零搭建一个可投入生产的 Rust 任务队列服务、并在多语言共享同一数据库的场景下正确运维它的完整能力。

River 与 Rust：项目定位

River 的核心思路是"多语言共用一套队列协议"：数据库表结构（由版本化迁移维护）与任务插入、领取、完成、重试的协议在所有语言实现中保持一致。Rust 实现的客户端 crate 名为 riverqueue，构建在 Tokio 异步运行时之上；任务参数通过 Serde 的 Serialize/Deserialize 派生进行类型化编解码，落库后以 JSON 字节存储。

从工作区清单 rust/Cargo.toml 可以看到，Rust 侧实际由五个版本一致的 crate 组成，且独立于 Go 模块版本发布：

Crate	用途
riverqueue	类型化客户端、Worker、任务与队列管理、维护服务
riverqueue-macros	提供 #[derive(JobArgs)] 派生宏
riverqueue-migrate	River 的数据库迁移（与 Go 侧完全一致）
riverqueue-cli	riverqueue 命令行，用于迁移与基准测试
riverqueue-test	测试夹具与 Worker 测试辅助

环境要求方面，工作区在 rust/Cargo.toml 中声明了 rust-version = "1.95"（即最低支持的 Rust 版本为 1.95）与 edition = "2024"，对应文档中"minimum supported Rust version is 1.95"的说明。

安装与依赖配置

在应用的 Cargo.toml 中添加如下依赖（与官方快速开始完全一致）：

[dependencies]
riverqueue = "0.3.0"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }

River 运行在 Tokio 上，任务参数必须派生 Serde 的 Serialize 与 Deserialize。riverqueue 提供了三个可开关的 feature：

Feature	默认	作用
postgres	是	通过 SQLx 支持 Postgres
sqlite	否	通过 SQLx 支持 SQLite 3.45 或更新版本
chrono-tz	否	在 cron 的 CRON_TZ= 与 TZ= 前缀中解析 IANA 时区名（如 America/New_York）

若只使用 SQLite，请按官方建议关闭默认特性再单独开启 sqlite：

riverqueue = { version = "0.3.0", default-features = false, features = ["sqlite"] }

这个 feature 结构在 riverqueue/Cargo.toml 中有明确对应：default = ["postgres"]，postgres 与 sqlite 分别联动 sqlx/postgres 与 sqlx/sqlite，chrono-tz 则引入 chrono-tz 依赖来解析 IANA 时区。

版本协调与类型重导出

River 的公开 API 直接使用以下外部库的类型：SQLx（连接池与事务）、Chrono（时间戳）、serde_json（metadata、output 及其他 JSON 值）以及 tokio-util（Worker 的 CancellationToken）。为了避免类型不一致，riverqueue 在 lib.rs 中把它们分别重导出为：

• riverqueue::sqlx（SQLx 0.9）
• riverqueue::chrono（Chrono 0.4）
• riverqueue::serde_json（serde_json 1）
• riverqueue::tokio_util（tokio-util 0.7）

最佳实践是直接使用这些重导出，或在自己的依赖中固定与 River 兼容的版本，否则会出现 PgPool 等类型不匹配的编译问题。此外，River 不为 SQLx 选择 TLS 实现：如果数据库连接走 TLS，需要在你自己声明的 SQLx 依赖上开启某个 TLS feature（如 tls-rustls）。

快速开始：一个最小可运行的 Worker

官方快速开始演示了完整闭环：设置 DATABASE_URL、定义可序列化的任务参数、注册 Worker、执行迁移、启动客户端并插入任务。

use riverqueue::migrate::PostgresMigrator;
use riverqueue::sqlx::PgPool;
use riverqueue::{
    BoxError, Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers,
};
use serde::{Deserialize, Serialize};

#[derive(Clone, Debug, Deserialize, JobArgs, Serialize)]
#[river(kind = "send_email")]
struct SendEmail {
    address: String,
}

async fn send_email(
    context: WorkContext,
    job: Job<SendEmail>,
) -> Result<WorkOutcome, BoxError> {
    println!("sending email to {}", job.args.address);
    context.record_output(serde_json::json!({"delivered": true}))?;
    Ok(WorkOutcome::Complete)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    // Create or upgrade River's tables. Applications often run
    // `riverqueue migrate-up` from `riverqueue-cli` at deploy time instead.
    PostgresMigrator::new(pool.clone()).migrate_up().await?;

    let mut workers = Workers::new();
    workers.add_fn(send_email)?;

    let client = Client::builder(pool)
        .workers(workers)
        .queue("default", QueueConfig::new(10))
        .build()?;
    // Work jobs until Ctrl-C, then stop fetching and let running jobs finish.
    let mut run = client.start_with_graceful_stop(async {
        let _ = tokio::signal::ctrl_c().await;
    })?;

    client
        .insert(SendEmail {
            address: "[e-mail]".to_owned(),
        })
        .await?;

    run.wait().await?;
    Ok(())
}

这个示例背后有几点关键语义：

• 迁移必须先行。任何客户端启动之前都要先建好或升级 River 的表；生产环境更推荐把 riverqueue migrate-up（来自 riverqueue-cli）作为独立部署步骤。
• 启动客户端的时机。Client::start 返回 RunHandle，调用方应等待 wait、stop（软停止，让运行中的任务自然结束）或 stop_and_cancel（硬停止，取消任务）。RunHandle::stopper 返回可克隆的 Stopper，可在其他任务（例如信号处理）里触发停止。需要注意的是：丢弃所有 Client 克隆并不会停止运行中的客户端；丢弃 handle 才会请求硬停止；RunHandle::detach 则让客户端无人监管地继续运行。
• 插入专用客户端。不带队列与 Worker 的客户端（如 Client::builder(pool).build()?）无需启动即可插入任务，这在 unique 示例 中直接体现。
• 跨语言一致性。任务 kind 与序列化后的 JSON 字段在共享任务的语言之间必须一致（详见后文"混合部署"一节）。

Workers::add_fn 接收异步函数或闭包，注册失败会返回 Result，可用 ? 传播。需要自定义超时或"下次重试决策"时，则实现 Worker] trait 并通过 Workers::add 注册——对应 Go 侧 AddWorker 与 WorkFunc 的合体。可运行的最小示例见 [examples/basic_worker.rs。

客户端构建器的关键配置

从 builder.rs 的源码看，ClientBuilder 提供的配置包括但不限于：

• queue(name, QueueConfig)：新增或替换一个队列及其并发配置；
• workers(workers)：设置客户端注册的 Worker 集合；
• id(id)：设置稳定的客户端标识（1~100 字节），默认由主机名、创建时间与随机后缀组成，只有在每个进程唯一时才需要显式设置；
• job_timeout(Duration) / without_job_timeout()：单任务超时，默认 1 分钟（常量 JOB_TIMEOUT_DEFAULT），置空则任务无时长限制；
• soft_stop_timeout(Duration)：软停止开始后经过该时长升级为硬停止；
• without_leader_election()：使客户端永不参与领导选举（详见"领导选举与维护"）；
• fetch_only_known_kinds(true)：只领取已注册 Worker 的 kind 的任务（详见混合部署）；
• periodic_job(job)：注册周期任务；
• maintenance(config)：覆盖领导端维护参数；
• error_handler(handler)、hook(hook)、insert_middleware(...)、work_middleware(...)、plugin(...)：挂接错误处理、生命周期钩子与中间件。

QueueConfig::new(max_workers) 用于配置单个队列的最大并发 Worker 数，源码 builder.rs 中的校验逻辑表明 max_workers 合法范围为 1 到 QUEUE_NUM_WORKERS_MAX（10000）。它还可以用 with_fetch_cooldown（队列级领取冷却，默认 100ms）、with_fetch_poll_interval（无插入通知时的轮询间隔，默认 1s）微调拉取节奏。

事务与业务数据一致性

River 让"任务入队"与"业务写入"可以处于同一个数据库事务中。调用方式是把 .tx(&mut transaction) 链到 client.insert(args) 上：

use riverqueue::{Client, InsertOpts, JobArgs};
use serde::{Deserialize, Serialize};

#[derive(Clone, Deserialize, JobArgs, Serialize)]
#[river(kind = "send_email")]
struct SendEmail {
    address: String,
}

async fn enqueue_urgent(client: &Client) -> Result<(), riverqueue::Error> {
    client
        .insert(SendEmail { address: "[e-mail]".to_owned() })
        .opts(
            InsertOpts::default()
                .with_queue("critical")
                .with_priority(1)
                .with_max_attempts(8),
        )
        .await?;
    Ok(())
}

任务只有在事务提交后才会变得可见并触发客户端通知。完整的业务示例（订单记录与确认任务同事务、Worker 内业务更新与任务完成同事务）见 examples/transactions.rs：其中 worker 通过 context.job_complete_tx(&mut transaction) 在调用方事务里完成自己的任务，事务不提交则任务不完成，从而保证"崩溃不会重复发确认邮件"。

为什么用 River 提供的 begin 辅助函数

官方指南明确指出：开启事务应使用 riverqueue::database::begin_postgres(&pool)；SQLite 下用 riverqueue::database::begin_sqlite_write(&pool)。这两个辅助函数（见 database.rs）在独立任务上开启事务，因此中途放弃是安全的（例如在 select! 或超时场景中）；而 SQLx 自带的 pool.begin() 不可放弃——它可能把仍处于事务中的连接归还到连接池。

begin_sqlite_write 使用 BEGIN IMMEDIATE 一开始就抢占写锁，这样"先读后写"的事务不会被中途的写提交撞出 SQLITE_BUSY_SNAPSHOT 错误。

无 savepoint 语义

River 直接在调用方事务中执行这些请求，不会自动包裹 savepoint。若某个请求返回错误，它可能已经在事务中写入了部分内容，因此应当整体回滚事务；若希望从错误中恢复并继续使用该事务，则需在发起请求前自行打开 savepoint。

另外，insert_many 可原子地批量插入同一 kind 的多个任务并按输入顺序返回结果；insert_batch 则用于批量插入不同 kind 的任务。

任务与队列管理

client.jobs() 提供持久化任务的获取、列表、取消、重试、更新与删除；client.queues() 操作所有客户端共享的队列记录（获取、列表、暂停、恢复、更新）；client.local_queues() 则在运行时增删"本客户端正在消费的队列"。这些请求在被 await 时执行，且和插入一样支持 .tx(&mut transaction)：

use riverqueue::sqlx::PgPool;
use riverqueue::{Client, JobListParams, JobState, QueueConfig, QueueSelector};

async fn maintain(client: &Client, pool: &PgPool) -> Result<(), riverqueue::Error> {
    let page = client
        .jobs()
        .list(JobListParams::default().states([JobState::Retryable]).limit(50))
        .await?;
    for job in &page.jobs {
        client.jobs().retry(job.id).await?;
    }

    let mut transaction = riverqueue::database::begin_postgres(pool).await?;
    client.queues().pause(QueueSelector::All).tx(&mut transaction).await?;
    transaction.commit().await?;

    client.local_queues().add("reports", QueueConfig::new(2))?;
    Ok(())
}

QueueSelector 支持 All（作用于所有有持久记录的队列）与 Named(name)（按名称精确匹配，Named("*") 不匹配任何队列），字符串可隐式转换为 Named，见 queue.rs。

插入选项与默认值体系

插入选项存在清晰的优先级：调用点的 opts > 任务类型的默认值 > 客户端默认值 > River 库默认值。InsertOpts（job.rs）支持以下 with_* 方法：

• with_max_attempts(n)：最大尝试次数（含首次），默认 25（常量 MAX_ATTEMPTS_DEFAULT）；
• with_priority(1..=4)：优先级，1 最高、4 最低，默认 1（常量 PRIORITY_DEFAULT）；
• with_queue(name)：任务所在的队列，默认 "default"；
• with_scheduled_at(ts) / without_schedule()：定时执行或立刻可用；
• with_pending(bool)：是否以 pending 状态插入；
• with_tags(...)：可搜索的标签；
• with_unique(...)：唯一性约束（见下文）；
• with_metadata(...)：整体替换任务默认 metadata。

River 的 InsertOpts、UniqueOpts、PeriodicJobOpts、QueueConfig、MaintenanceConfig、PostgresDatabase 等"既有 getter 又有 setter"的配置值统一采用 with_* 返回新值的风格（类似 PathBuf::with_extension），而纯参数对象（ClientBuilder、JobListParams、JobUpdateParams）使用普通 setter 名。

Worker 结果与取消

Worker 的返回值 WorkOutcome 有四种：

• Complete：任务成功完成；
• Snooze：重新调度，且不消耗一次尝试次数；
• Discard：终结任务，不再重试；
• Cancel：以 cancelled 状态终结。

普通错误则按重试策略一直重试到达到最大尝试次数。

WorkContext 的取消 token 会在三种情况下触发：任务超时、远端取消任务、客户端停止。Worker 应在自然的 await 点检查或 select 取消，并在因取消而中止时返回 WorkCancelled：

use riverqueue::{Job, JobArgs, WorkCancelled, WorkContext, WorkOutcome};
use serde::{Deserialize, Serialize};

#[derive(Clone, Deserialize, JobArgs, Serialize)]
#[river(kind = "build_report")]
struct BuildReport {}

async fn build_report(
    context: WorkContext,
    _job: Job<BuildReport>,
) -> Result<WorkOutcome, WorkCancelled> {
    tokio::select! {
        () = context.cancellation_token().cancelled() => Err(WorkCancelled),
        () = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
            Ok(WorkOutcome::Complete)
        }
    }
}

关键语义：在客户端硬停止（RunHandle::stop_and_cancel 或 Stopper::stop_and_cancel）期间，只要 Worker 返回的错误链中任意位置包含 WorkCancelled，该任务就会不消耗尝试次数地重新变为可用；其他错误会被记录并消耗一次尝试；Ok 则完成任务。从 worker.rs 可见 WorkContext 携带 cancellation token、client 引用、job id、metadata 更新与可恢复状态等。

此外，超过配置的 stuck 阈值（默认 10 秒，JOB_STUCK_THRESHOLD_DEFAULT）后，River 可以中止一个主动让出 CPU 的 Tokio 任务并使其该次尝试失败；但它无法停止正在进行的 CPU 密集计算或已经阻塞的调用。context.record_output(...) 将输出写入 River 预留的 output metadata 键（上限 32MB JSON，见 worker.rs）。

事件订阅

client.subscribe(&<a href="https://link.gitcode.com/i/2e971f89c22574a39b02af79119ad407" target="_blank">EventKind::...]) 提供对客户端本地事件的观察（注意：这是本地观察，不是持久化事件流）。订阅应在启动客户端之前完成，才能从第一批任务开始收到事件；接收端有界，慢消费者不会阻塞 Worker，而是通过 EventRecvError::Lagged 报告丢弃的事件数；EventReceiver 同时实现了 Stream。任务事件在结果持久化之后发送，相互独立的任务之间没有全局完成顺序。可运行示例见 [examples/events.rs，它订阅 JobCompleted 与 JobFailed 并读取运行时长与错误信息。

可靠性特性：唯一、周期、可恢复与中间件

唯一任务（Unique jobs）

唯一任务按 kind、编码后的参数或选定的参数字段路径、队列、周期与任务状态进行去重。派生宏会遵循 Serde 的序列化命名并省略缺失的可选字段。宏层面（见 riverqueue-macros/README.md）支持：

#[derive(Deserialize, JobArgs, Serialize)]
#[river(
    kind = "send_invoice",
    queue = "billing",
    tags("billing"),
    unique(by_args, by_period = "24h", by_state(available, pending, running, scheduled)),
    insert_opts = invoice_insert_opts,
)]
struct SendInvoice {
    #[river(unique)]
    invoice_id: i64,
    note: String,
}

unique(...) 接受 by_args、by_args("nested.path", ...)、by_period（至少一秒的时长，如 "90s" 或 "1h30m"）、by_queue、by_state(...) 与 exclude_kind，与 UniqueOpts 的 setter 一一对应。带 #<a href="https://link.gitcode.com/i/cddd0c03759ad6178bb49af334599467" target="_blank">river(unique)] 的字段以及 by_args 指定的路径参与唯一性哈希；两者皆无时哈希所有参数。注意两点约束：在未声明 unique(by_args) 时给字段标记 #[river(unique)] 是编译错误（否则该标记无意义）；唯一性路径遵循 Serde 序列化侧的 rename/rename_all 规则。唯一路径的测试位于 [riverqueue-macros/tests/ui，覆盖了未知字段、非法周期、重复路径等编译期报错场景。

运行期用法见 examples/unique.rs：默认按参数去重的任务，第二次插入会返回已存在的同一任务（second.unique_skipped_as_duplicate == true，first.id() == second.id()）；未声明唯一的任务则可在插入时通过 InsertOpts::default().with_unique(UniqueOpts::new().with_by_args(true).with_by_period(...)) 临时开启去重。

周期任务（Periodic jobs）

周期任务只由当选的 leader 入队，可在构建客户端时配置（也可在运行时通过 client.periodic_jobs() 调整）；稳定的 ID 防止重复注册。CronSchedule 接受标准五字段 cron 语法及 @hourly、@every 90s 这类描述符（见 cron.rs 的 CronSchedule::parse），默认在进程本地时区求值，除非指定了其他 CronTimeZone。CRON_TZ= 与 TZ= 前缀命名 IANA 时区（如 America/New_York）需要 chrono-tz feature（自带时区数据库）；不开启时仅能解析 UTC、Local 与 Etc/GMT±N。

PeriodicJobOpts（periodic/mod.rs）支持 with_id（客户端内唯一）与 with_run_on_start（每次当选 leader 时立即插入一次）。完整示例见 examples/periodic.rs，它组合了 15 分钟间隔的 IntervalSchedule 与"每天 02:30 UTC"的 cron。

可恢复任务（Resumable steps）

可恢复步骤会持久化"最后完成的步骤"与可选游标；当进度与业务数据必须一起提交时，应使用事务性检查点辅助函数。

中间件与钩子

插入中间件包裹插入的 begin 钩子；工作中间件包裹 work 钩子、参数解码与 Worker 本身（对应 WorkMiddleware 与 Hook trait）。

领导选举与维护

任意时刻只有一个客户端持有数据库租约并运行 leader 专属服务：任务调度器（scheduler）、卡住任务救援器（rescuer）、任务与队列清理器、周期任务入队器、Postgres 重建索引器（reindexer）以及 SQLite 通知清理器。失去租约或客户端停止会立即停止这些服务。

默认 MaintenanceConfig（见 builder.rs）的关键参数与默认值：

参数	默认值	含义
scheduler_interval	5s	leader 将到期任务与可重试任务变为可用的频率
elect_interval	5s	参与/续任领导竞选的频率
rescuer_interval	30s	查找并救援卡住任务的频率
rescue_after	1 小时（或"任务超时 + 1 小时"）	任务运行多久后视为卡住；不得短于任务超时
job_cleaner_interval / job_cleaner_timeout	30s / 30s	清理已终结任务
cancelled_job_retention / completed_job_retention	24 小时	取消/完成任务的保留时长（可用 Retention::Keep 关闭删除）
discarded_job_retention	7 天	丢弃任务的保留时长
queue_retention	24 小时	无客户端触碰的队列记录保留时长
queue_cleaner_interval	1 小时	清理过期队列记录

救援器会丢弃"自己 Worker 注册表不认识的 kind"的卡住任务（而不是重试）。因此，共享同一数据库的客户端应注册相同的 kind；无法注册全部 kind 的客户端可以用 ClientBuilder::without_leader_election() 构建：它照常消费自己的队列，但永远不会成为 leader，因此部署中必须至少保留一个可被选举的客户端。该构建选项的实现细节（含不能配置周期任务等限制）见 builder.rs。

周期任务从每个 leader 任期的起点开始调度；with_run_on_start(true) 的任务每获得一次任期插入一次，插入失败的本次发生会被记录并跳过。

数据库支持

Postgres 与 SQLite 两个后端实现相同的任务/队列行为，Client 不针对数据库泛型化，因此后端类型不会泄漏到 Worker、上下文或扩展中。用法上把 SQLx 连接池（或带选项的 PostgresDatabase/SqliteDatabase）传给 Client::builder。

SQLite 需要 sqlite feature 且版本不低于 3.45——River 使用 SQLite 的 JSONB 函数存储 JSON。River 完全按调用方配置使用连接池，不修改连接 pragma；文件型数据库应开启 WAL 与 busy timeout，让短暂的写冲突等待而不是直接失败。私有 :memory: 数据库只属于单个连接，因此要么把连接池限制为 1 个连接，要么使用 shared-cache URI：

use std::{str::FromStr, time::Duration};

use riverqueue::Client;
use riverqueue::sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};

async fn sqlite_client() -> Result<Client, Box<dyn std::error::Error>> {
    let options = SqliteConnectOptions::from_str("sqlite://river.db")?
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal)
        .busy_timeout(Duration::from_secs(5));
    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await?;
    Ok(Client::builder(pool).build()?)
}

可运行的 SQLite 完整示例见 examples/sqlite.rs，它使用 SqliteMigrator 建表、以 WAL 模式打开库文件、工作一个任务后优雅停止。另外，针对不支持 LISTEN/NOTIFY 的服务（如默认配置的 YugabyteDB），ClientBuilder::without_notifications() 可让客户端退化为纯轮询模式。

CLI 迁移与基准测试

riverqueue-cli 安装与迁移

cargo install riverqueue-cli

riverqueue 命令的子命令在 main.rs 中定义：migrate-down、migrate-list、migrate-up、validate 与 bench。迁移命令示例：

riverqueue migrate-up --database-url postgres://localhost/app
riverqueue migrate-up --database-url postgres://localhost/app --schema river
riverqueue migrate-up --database-url sqlite://app.sqlite3

migrate-down、migrate-list、validate 使用相同的连接选项。--target-version N、--max-steps N、--dry-run 可限制或预览一次迁移（参数解析逻辑见 migrate.rs，--schema 仅对 Postgres 有效）。应用也可以在 Rust 代码中直接使用 riverqueue-migrate crate 完成迁移、回滚预览与校验：

use riverqueue_migrate::{Direction, MigrateOpts, PostgresMigrator, SqliteMigrator};
use sqlx::{PgPool, SqlitePool};

async fn migrate(postgres: PgPool, sqlite: SqlitePool) -> Result<(), Box<dyn std::error::Error>> {
    // Apply every outstanding migration.
    let migrator = PostgresMigrator::new(postgres);
    let applied = migrator.migrate_up().await?;
    println!("applied versions {applied:?}");

    // Or preview what migrating down one version would run.
    let preview = migrator
        .migrate(Direction::Down, MigrateOpts::new().with_dry_run(true))
        .await?;
    for version in preview.versions {
        println!("would revert {:03} {}", version.version, version.name);
    }

    // Check that every migration is applied before starting clients.
    let validation = SqliteMigrator::new(sqlite).validate(None).await?;
    if !validation.is_valid() {
        return Err(validation.to_string().into());
    }
    Ok(())
}

River 的 Schema 由版本化迁移统一管理（Postgres 与 SQLite 迁移文件分别位于 rust/riverqueue-migrate/migrations/main 与 rust/riverqueue-migrate/migrations/sqlite），与 Go 侧完全一致，因此两种语言可以互相迁移对方建过的数据库。

基准测试

riverqueue bench 测量 Worker 吞吐与端到端延迟。它会清空所选数据库中 River 的任务表，因此只能指向一次性（disposable）数据库：

riverqueue bench --database-url postgres://localhost/river_bench --duration 30s

运行 riverqueue bench --help 可查看全部选项（支持连续燃烧、固定 --num-total-jobs 燃烧、自定义 schema、可调 worker/连接池/批量大小，输出周期吞吐量与最终 jobs/sec + p95 端到端延迟）。

测试辅助：riverqueue-test

riverqueue-test 提供三类测试能力：

• 断言已插入任务：require_inserted、require_many_inserted、require_not_inserted 按插入顺序列出期望 kind 的任务，不满足时以描述性消息 panic；_with 变体接受 RequireInsertedOpts（可叠加队列、优先级、状态、标签等期望），_tx 变体透过未提交事务读取，用于测试"事务性入队后尚未提交"的代码路径。
• 单次运行 Worker（无数据库）：TestJobBuilder 从参数类型的插入默认值构造真实感 Job<A>，可覆盖持久化 ID、attempt、状态与 metadata；work_once 以 detached WorkContext 调用类型化 Worker，保留具体错误并捕获输出/metadata 更新的不可变快照。它不运行客户端钩子、中间件、事务、重试与完成持久化。
• 带客户端运行 Worker：work_with_client 是数据库回填版本——用客户端插入任务、按领取方式认领、以该客户端作为 WorkContext 运行 Worker，从而让"通过 context.client() 插入后续任务"或"通过 context.job_complete_tx 在自有事务中完成任务"的 Worker 按生产语义执行。客户端无需启动；River 不记录结果，任务会保持 running，除非 Worker 自己完成了它。

与 Go 混合部署

River 的 Rust 与 Go 客户端共享同一数据库 Schema 与协议。一个 Rust 服务可以插入由 Go Worker 运行的任务、消费 Go 服务插入的任务，并与 Go 客户端一起参与领导选举与维护。完整指南见 rust/riverqueue/docs/mixed-deployments.md，要点如下：

• 版本与迁移：Rust crate 与 Go 独立版本化，部署中所有客户端必须理解同一 Schema。迁移只需用任一方执行一次（Go 用 river migrate-up，Rust 用 riverqueue migrate-up 或 PostgresMigrator/SqliteMigrator），双方写入相同的 river_migration 历史。
• 队列与任务 kind：客户端按队列而非 kind 领取任务。如果某客户端领到了自己没有 Worker 的任务，会记录可重试的"job kind is not registered"错误并反复重试直到其他客户端消费或耗尽尝试。因此应让每种语言消费各自专属的队列；若必须共享队列，Rust 侧用 ClientBuilder::fetch_only_known_kinds(true)（Go 侧 Config.FetchOnlyKnownKinds）只领取已知 kind。注意这仅影响领取——leader 的救援器仍会处理所有 kind 的卡住任务，所以要么在可当选 leader 的客户端注册全部 kind，要么用 without_leader_election 把它们排除在选举之外。
• 唯一任务跨语言一致性：唯一键哈希"编码后的 JSON 参数"，因此双方序列化必须一致。Rust 侧由 encoding.rs 保证——encode_args 按 Go encoding/json 的输出规则编码：浮点数用 Go 的最短往返数字、字符串转义 </>/&/U+2028/U+2029、结构体字段保持声明顺序。类型层面的差异需在参数类型中处理：Go 会排序 map 键（Rust 侧用 BTreeMap 对齐）、Go 的 []byte 是 base64 字符串（Rust Vec<u8> 是数字数组）、参与唯一键的时间字段建议用 #[serde(with = "riverqueue::encoding::go_time")] 得到 Go 的 RFC 3339 最短小数秒形式。字段路径用 JSON 名书写，嵌套对象会保留编码字段顺序，因此嵌套的 Go struct 需要在 Rust 中保持相同的字段顺序。
• 周期任务：周期任务只由 leader 入队，凡是可能当选 leader 的客户端（无论哪种语言）都必须配置相同的周期任务、ID 与 schedule；cron 两侧统一使用标准五字段语法。
• SQLite：跨进程共享的 SQLite 数据库必须在每个进程中启用 WAL 与 busy timeout；Go 客户端共享文件时应以 _txlock=immediate 打开，避免读写穿插触发 SQLITE_BUSY。
• 滚动部署与回滚：先确认 Schema/协议兼容并执行迁移，再小规模部署 Rust 客户端，观察队列深度、重试、被救援任务、领导变更与连接数后再扩大规模。回滚不触碰 Schema：用 RunHandle::stop 优雅停止 Rust 客户端即可，Rust 插入的任务对 Go Worker 而言只是普通行，遗留的 running 任务会被救援器恢复；只有不再有任何客户端需要新 Schema 时，才把 migrate-down 作为独立操作执行。

示例与开发

仓库 rust/riverqueue/examples 提供了可直接运行的程序：basic_worker（基础 Worker）、graceful_shutdown（首次 Ctrl-C 软停止、再次 Ctrl-C 取消运行中任务）、cancellation（取消处理）、transactions（事务入队与完成）、unique 与 periodic（唯一与周期任务）、events（事件订阅）、custom_schema（自定义 Postgres schema）、sqlite（SQLite 运行）以及 mixed_go_rust（Rust 与 Go 共享一个数据库）。运行方式示例：

DATABASE_URL=postgres://localhost/river_example cargo run -p riverqueue --example basic_worker
cargo run -p riverqueue --example sqlite --features sqlite

针对 Rust 侧的开发，rust/docs/development.md 提供了 make lint/rust、make test/rust、make doc/rust、make check/rust/package 等检查命令；Postgres 集成测试需要一次性数据库，并以 RIVER_RUST_DATABASE_URL=... make test/rust/postgres 运行。值得注意：Rust 测试会基于 River 的 Go 实现生成的 conformance fixtures（unique key、重试边界、cron schedule、snooze 计数）做跨语言一致性校验，fixture 由 make test/rust 目标先生成，因此跑测试需要 Go 工具链。Rust crate 的版本发布使用独立的 rust/vX.Y.Z 标签，与 Go 模块标签互不影响。

小结

Rust 版 River 把"多语言共享的后台任务协议"封装成了类型安全、Tokio 原生、Serde 驱动的客户端体验：从快速开始的几十行代码，到事务化入队/完成、唯一/周期/可恢复任务、事件订阅、领导选举与维护、CLI 迁移与基准测试，再到与 Go 服务的无缝混布，它覆盖了后台任务系统从开发到生产运维的完整链路。无论你是新建一个纯 Rust 任务队列服务，还是把 Rust Worker 逐步引入已有的 Go 部署，rust/README.md 与上述各 crate 文档、源码与示例都是最直接的权威参考。