第18章 SQLite 驱动：每条连接一个 worker 线程

SQLite 没有 wire protocol，只有一套同步的 C API。sqlx-sqlite 的办法是给每条连接配一个专用 OS 线程，用 channel 把同步调用包装成 async。

18.1 问题引入：同步 C 库如何变 async

SQLite 是嵌入式数据库（1.0 发布于 2000 年，官方年表）——和 Postgres / MySQL 完全不同的世界：

• 无网络：所有操作走 libsqlite3 的 C API。
• 全同步：sqlite3_step 直接阻塞当前线程直到一步完成（读到行 / 语句结束 / 错误）。
• 单文件：整个数据库是一个 .sqlite 文件——多进程访问靠文件锁。
• 内嵌运行：没有独立 server——直接在应用进程里跑 SQL 执行引擎。

这些特点让 SQLite 在嵌入场景（手机 App、桌面工具、CLI、测试 DB）极好用——开箱即用、零运维、快——但让 sqlx 这种 async 库头疼。在 async 任务里直接调 sqlite3_step，阻塞的是 Tokio 的 worker 线程——慢查询期间这个线程上的其他 task 全部停摆，current_thread runtime 下整个 runtime 都卡住。

sqlx-sqlite 的解法：把 sqlite3 handle 放到专用 OS 线程里、主线程通过 channel 和它通信、每次"同步"调用 C API 只阻塞 worker 线程——Tokio reactor 照常工作。

这条架构让 sqlx::query("SELECT ...").fetch_one(&sqlite_pool).await? 在用户视角和 Postgres 一样——但背后是完全不同的机器。本章拆开这台机器。

18.2 ConnectionWorker 架构

sqlx-sqlite/src/connection/worker.rs:32-42：

pub(crate) struct ConnectionWorker {
    command_tx: flume::Sender<(Command, tracing::Span)>,
    pub(crate) shared: Arc<WorkerSharedState>,
}

pub(crate) struct WorkerSharedState {
    transaction_depth: AtomicUsize,
    cached_statements_size: AtomicUsize,
    pub(crate) conn: Mutex<ConnectionState>,  // futures_intrusive::Mutex
}

ConnectionWorker 的两条通信通道：

1. command_tx: flume::Sender —— 主线程→worker，发命令。
2. shared: Arc<WorkerSharedState> —— 主线程+worker 共享，atomic 字段（depth / cache size）+ conn Mutex。

worker 线程由 ConnectionWorker::establish 通过 thread::Builder::new().spawn(...) 启动——每个 SqliteConnection 一个独立线程。线程里持有 sqlite3 handle、循环接收 command、执行 C API 调用、回复结果。

18.2.1 线程启动代码

sqlx-sqlite/src/connection/worker.rs:103+（简化）：

pub(crate) async fn establish(params: EstablishParams) -> Result<Self, Error> {
    let (establish_tx, establish_rx) = oneshot::channel();

    thread::Builder::new()
        .name(params.thread_name.clone())
        .spawn(move || {
            // 1. flume channel 是在 worker 线程内部建的
            let (command_tx, command_rx) = flume::bounded(params.command_channel_size);

            // 2. 在线程里建立 sqlite3 handle
            let conn = match params.establish() {
                Ok(conn) => conn,
                Err(e) => { establish_tx.send(Err(e)).ok(); return; }
            };

            // 3. 构造 WorkerSharedState，并且**立刻**把 conn 锁在手里
            let shared = Arc::new(WorkerSharedState {
                transaction_depth: AtomicUsize::new(0),
                cached_statements_size: AtomicUsize::new(0),
                conn: Mutex::new(conn, true),  // fair
            });
            let mut conn = shared.conn.try_lock().unwrap();

            // 4. 把 ConnectionWorker 返回给主线程
            if establish_tx.send(Ok(Self { command_tx, shared: Arc::clone(&shared) })).is_err() {
                return;  // 主线程已经不要这条连接了
            }

            // 5. 消息泵循环
            for (cmd, span) in command_rx {
                let _guard = span.enter();
                match cmd {
                    Command::Execute { .. } => { /* ... */ }
                    Command::Prepare { .. } => { /* ... */ }
                    // ... 共 12 种 command
                }
            }
        })?;

    establish_rx.await.map_err(|_| Error::WorkerCrashed)?
}

几个容易看错的点：

1. flume channel 在 worker 线程里创建——不是先建好再 spawn。所以 command_tx 是随 establish_tx 一起送回主线程的。
2. worker 一开始就 try_lock() 住了 shared.conn——正常情况下这把锁一直在 worker 手里，只有 Command::UnlockDb 才会临时放开（见 §18.6）。
3. 消息泵循环跑到 channel 关闭（主线程 drop 掉最后一个 command_tx）或收到 Command::Shutdown（close() 走这条路）为止。

thread::Builder::new() 给线程设名字——默认是 sqlx-sqlite-worker-{id}（sqlx-sqlite/src/options/mod.rs:209），可以用 SqliteConnectOptions::thread_name 换成自己的生成函数——方便 top / profiler 识别。

flume::bounded（不是 unbounded）——容量来自 command_channel_size，默认 50（sqlx-sqlite/src/options/mod.rs:210；返回行的 channel 另有 row_channel_size，默认也是 50）——防止主线程发 command 太快撑爆内存。超容量时发送会 async 等待——自然背压。

18.3 Command 枚举

sqlx-sqlite/src/connection/worker.rs:54-100 共 12 个 variant：

enum Command {
    Prepare { query: Box<str>, tx: oneshot::Sender<Result<SqliteStatement<'static>, Error>> },
    Describe { query: Box<str>, tx: oneshot::Sender<Result<Describe<Sqlite>, Error>> },
    Execute {
        query: Box<str>,
        arguments: Option<SqliteArguments<'static>>,
        persistent: bool,
        tx: flume::Sender<Result<Either<SqliteQueryResult, SqliteRow>, Error>>,
        limit: Option<usize>,
    },
    Serialize { schema: Option<SchemaName>, tx: oneshot::Sender<Result<SqliteOwnedBuf, Error>> },
    Deserialize { schema: Option<SchemaName>, data: SqliteOwnedBuf, read_only: bool,
                  tx: oneshot::Sender<Result<(), Error>> },
    // 注意：事务三兄弟用的是 rendezvous_oneshot，不是普通 oneshot
    Begin { tx: rendezvous_oneshot::Sender<Result<(), Error>>, statement: Option<Cow<'static, str>> },
    Commit { tx: rendezvous_oneshot::Sender<Result<(), Error>> },
    Rollback { tx: Option<rendezvous_oneshot::Sender<Result<(), Error>>> },
    UnlockDb,
    ClearCache { tx: oneshot::Sender<()> },
    Ping { tx: oneshot::Sender<()> },
    Shutdown { tx: oneshot::Sender<()> },
}

除了 UnlockDb（以及 tx 为 None 的 Rollback），每个 variant 都带一个回复 channel——三种：oneshot::Sender（单次回复）、flume::Sender（流式回复，Execute 独有）、以及模块内自定义的 rendezvous_oneshot::Sender（Begin / Commit / Rollback 用）。

为什么事务三兄弟要 rendezvous？——普通 oneshot 是"发完就走"，worker 无法知道对方收没收到。而 BEGIN/COMMIT/ROLLBACK 一旦在 SQLite 里执行成功、消息却没送达（比如那一侧的 future 被取消），事务状态就和 Transaction 对象对不上了。rendezvous 版本的 send 会等到接收方真正取走才算成功，worker 于是能在"执行了但没确认"时立刻补一次回滚——Command::Begin 分支里那段"BEGIN was processed but not acknowledged"的注释讲的就是这件事。COMMIT / ROLLBACK 已执行却没送达时则反过来：worker 置位 ignore_next_start_rollback，跳过随后 Transaction Drop 发来的那次回滚。

Rollback { tx: None } 正是 Drop 用的形态：Transaction 被 drop 时调同步的 start_rollback（sqlx-sqlite/src/connection/worker.rs:399-403），用 flume 的同步 send 把这条命令塞进有界命令通道（满了会阻塞当前线程），不等回复；worker 轮到它就在自己线程里执行 ROLLBACK，不需要 async 一侧再碰这条连接（细节见第 15 章 §15.6.2）。

Execute 的回复为什么是 flume::Sender 而不是 oneshot？—— 因为 Execute 的结果是流式（Either<QueryResult, Row> 可能多次）——oneshot 只能发一次。用 flume channel 让主线程能 stream 从 worker 收到结果。

Command::Execute {
    query: "SELECT * FROM large_table".into(),
    arguments: None,
    persistent: false, // Executor 只在带参数时才置 true（executor.rs:31）
    tx: row_tx,   // 有界 channel，容量 row_channel_size（默认 50）
    limit: None,
}

worker 处理 Execute 时每 yield 一行就 send 一次到 tx——主线程把 rx 转成 stream 逐个 recv。行通道是有界的：主线程消费慢时，worker 的同步 send 会阻塞在这里，形成背压。

18.4 消息泵循环

sqlx-sqlite/src/connection/worker.rs:148+ 的消息泵（简化）：

for (cmd, span) in command_rx {
    let _guard = span.enter();
    match cmd {
        Command::Prepare { query, tx } => {
            tx.send(prepare(&mut conn, &query).map(|prepared| {
                update_cached_statements_size(&conn, &shared.cached_statements_size);
                prepared
            })).ok();
        }
        Command::Execute { query, arguments, persistent, tx, limit } => {
            let iter = match execute::iter(&mut conn, &query, arguments, persistent) {
                Ok(iter) => iter,
                Err(e) => { tx.send(Err(e)).ok(); continue; }
            };

            match limit {
                None => {
                    for res in iter {
                        let has_error = res.is_err();
                        if tx.send(res).is_err() || has_error {
                            break;  // channel 关了或出错，停
                        }
                    }
                }
                Some(limit) => { /* 带 limit 版本 */ }
            }
        }
        Command::Begin { tx, statement } => { /* BEGIN */ }
        Command::Commit { tx } => { /* COMMIT */ }
        // ... 其他 variant
    }
}

worker 线程的性格：

• 单线程——所有 sqlite3_* 调用都在这个线程、串行执行。不会有并发 race。
• 无 yield——每个 command 处理完才接下一个。Tokio 阻塞 schedule 不在这线程。
• 同步 C API——每次 sqlite3_step 阻塞 worker 多久完全取决于查询——其他 task 无影响、但 这条 connection 的其他 command 排队等。

worker 的瓶颈：单条连接并发 limited——如果你想并行跑两条 query，必须两条连接（两个 worker 线程）。Pool 的 max_connections 对应 N 个 worker 线程。

18.5 ConnectionHandle：C FFI 边界

sqlx-sqlite/src/connection/handle.rs 定义了 ConnectionHandle——直接包装 *mut sqlite3 的类型（节选，省略了 #[derive(Debug)]、构造函数和其他方法）：

pub(crate) struct ConnectionHandle(NonNull<sqlite3>);

// A SQLite3 handle is safe to send between threads, provided not more than
// one is accessing it at the same time. This is upheld as long as [SQLITE_CONFIG_MULTITHREAD]
// is enabled and [SQLITE_THREADSAFE] was enabled when sqlite was compiled. We refuse to work
// if these conditions are not upheld.
unsafe impl Send for ConnectionHandle {}

impl ConnectionHandle {
    pub(crate) fn as_ptr(&self) -> *mut sqlite3 { self.0.as_ptr() }
}

impl Drop for ConnectionHandle {
    fn drop(&mut self) {
        unsafe {
            // https://sqlite.org/c3ref/close.html
            let status = sqlite3_close(self.0.as_ptr());
            if status != SQLITE_OK {
                // this should *only* happen due to an internal bug in SQLite where we left
                // SQLite handles open
                panic!("{}", SqliteError::new(self.0.as_ptr()));
            }
        }
    }
}

两处细节值得盯住：

• unsafe impl Send——手动告诉 Rust "这个类型可以跨线程转移"。源码注释把契约写得很明确：前提是同一时间只有一个线程访问，且 SQLite 编译时开了 SQLITE_THREADSAFE、运行时是 SQLITE_CONFIG_MULTITHREAD——条件不满足 sqlx 拒绝工作（打开时带 SQLITE_OPEN_NOMUTEX，非线程安全编译的 SQLite 会直接报错，sqlx-sqlite/src/connection/establish.rs:75-81）。平时只有 worker 线程碰 handle；lock_handle() 把它借给调用方线程时，靠 §18.6 那把 Mutex 保证同一时刻只有一方在用。
• Drop 里用的是 sqlite3_close（不是 _v2），而且失败会 panic。sqlite3_close 在还有未 finalize 的 statement 时会返回 SQLITE_BUSY；sqlx 选择在这里 panic 而不是静默忽略——因为那只可能是 sqlx 自己漏了 sqlite3_finalize，是内部 bug，早炸早好。

ConnectionHandle 是 safe Rust 和 unsafe C API 的边界 ——上层代码（sqlx-sqlite 的 Executor 实现只往 worker 发命令，worker 的消息泵也是普通 Rust）不直接写 FFI，unsafe 调用收在 ConnectionHandle、StatementHandle 这类包装类型里。

取回的行也要过 Send 关。每取一行，SqliteRow::current 对每一列调 sqlite3_column_value 再 sqlite3_value_dup 复制成独立的"受保护值"（sqlx-sqlite/src/value.rs:200-205），包进 Arc<ValueHandle>；SqliteRow 于是自带全部数据，SqliteValueRef::to_owned（sqlx-sqlite/src/value.rs:76-78）只克隆 Arc。真正的跨线程担保写在 ValueHandle 上：// SAFE: only protected value objects are stored in SqliteValue（sqlx-sqlite/src/value.rs:110-112）；SqliteRow 上那对 unsafe impl Send/Sync（sqlx-sqlite/src/row.rs:27-28）是 0.5.10（#1551）之前行直接引用语句句柄时留下的，注释还在讲"不调 sqlite3_step 就安全"。

18.5.1 C API 的关键函数

sqlx-sqlite 调的 libsqlite3 函数（libsqlite3_sys crate）：

use libsqlite3_sys::{
    sqlite3_open_v2,          // 打开数据库
    sqlite3_close,            // 关闭（sqlx 用的是 close 不是 close_v2）
    sqlite3_prepare_v3,       // 编译 SQL 成 statement（缓存语句带 SQLITE_PREPARE_PERSISTENT）
    sqlite3_finalize,         // 释放 statement
    sqlite3_step,             // 执行一步（next row or done）
    sqlite3_reset,            // 重置 statement 复用
    sqlite3_clear_bindings,
    sqlite3_bind_int,         // 绑定参数
    sqlite3_bind_int64,
    sqlite3_bind_double,
    sqlite3_bind_text64,      // text / blob 用 64 位长度版本，析构参数传 SQLITE_TRANSIENT
    sqlite3_bind_blob64,
    sqlite3_bind_null,
    sqlite3_column_type,      // 读列类型
    sqlite3_column_value,     // 取列值，随后 sqlite3_value_dup 复制出来
    sqlite3_value_dup,
    sqlite3_changes,           // 最近操作影响的行数
    sqlite3_last_insert_rowid, // 最后插入的 rowid
    sqlite3_busy_timeout,      // 设置忙碌超时
    sqlite3_exec,              // 执行无参 SQL（快捷版）
    // ... 其余几十个（hook、collation、serialize、元数据等）
};

sqlx 不直接包装所有 SQLite C API——只挑需要的（按各文件从 libsqlite3_sys 导入的函数统计，七十来个）。其他通过 SQL 操作间接触达（PRAGMA 设置等）。

绑定参数时有两次拷贝：ConnectionWorker::execute 发命令前先调 SqliteArguments::into_static（sqlx-sqlite/src/connection/worker.rs:368），把借用的 text / blob 全部复制成自有数据，好让命令能跨线程送进 worker；worker 里 sqlite3_bind_text64 / sqlite3_bind_blob64 传的是 SQLITE_TRANSIENT（sqlx-sqlite/src/statement/handle.rs:218-245），SQLite 会在绑定时再复制一份。

18.6 Mutex<ConnectionState> 和 fair 锁

WorkerSharedState::conn: Mutex<ConnectionState>——futures-intrusive 的异步 Mutex。

为什么需要 Mutex？ 理论上 worker 线程独占 conn——不需要 Mutex。但 sqlx 提供了LockedSqliteHandle 给高级用户——可以在主线程临时借出 worker 的 conn、执行用户自定义 C API：

let mut conn = pool.acquire().await?;
let mut handle = conn.lock_handle().await?;
// handle 里可以直接访问底层 sqlite3 指针
// 用户可以调 libsqlite3 扩展函数
drop(handle);

lock_handle 内部：

1. 调用方用 join 同时做两件事：先排进 shared.conn.lock() 的等待队列，再往命令通道发 Command::UnlockDb（sqlx-sqlite/src/connection/worker.rs:464-476，注释写明"先入队、再发消息"）。
2. worker 处理到 UnlockDb 时 drop 掉自己一直持有的 guard，紧接着 futures_executor::block_on(shared.conn.lock()) 重新排队——排在调用方后面。
3. 调用方拿到 MutexGuard，包成 LockedSqliteHandle 返回；在它被 drop 之前 worker 一直阻塞在重新加锁上，期间发来的命令在通道里排队。

Mutex::new(conn, true) 的 true 是 fair 参数——必须 fair，因为 Command::UnlockDb 让 worker 释放后立即尝试 re-lock——非 fair 锁会让 worker 立刻抢回、正在等待的 lock_handle 拿不到锁。

源码注释把原因写得很直白（sqlx-sqlite/src/connection/worker.rs:122-124）：

note: must be fair because in Command::UnlockDb we unlock the mutex and then immediately try to relock it; an unfair mutex would immediately grant us the lock even if another task is waiting.

18.7 WAL 模式

SQLite 有两种 journal 模式：

• Rollback journal（默认）—— 传统模式。写操作用 journal 文件做 atomicity。读和写互斥——写者把改动落进数据库文件时要拿 EXCLUSIVE 锁，这段时间所有读被挡住。
• WAL (Write-Ahead Logging)——SQLite 3.7+ 的新模式。写到 WAL 文件、读直接读主 DB——读和写可以并发。

差异：

维度	Rollback journal	WAL
读写并发	提交时互斥	读写分离（可并发）
写吞吐	较低	较高（具体倍数取决于负载）
读时延	写者提交时被阻塞	读不被写阻塞
fsync	每事务	也需要（可调 synchronous）
崩溃恢复	用 journal 回滚未完成事务	由 WAL 文件恢复（未提交帧忽略）
跨 FS	几乎任何 FS	需要 shared memory (mmap)

WAL 模式强烈推荐生产使用——在代码里写 SqliteConnectOptions::journal_mode(SqliteJournalMode::Wal)。连接 URL 设不了它：0.8.6 的 URL 解析只认 mode / cache / immutable / vfs 四个参数（sqlx-sqlite/src/options/parse.rs），实测 sqlite:x.db?mode=rwc&journal_mode=wal 直接报 unknown query parameter `journal_mode` while parsing connection URL。

配合 synchronous = Normal（不是 Full）能进一步提升吞吐——WAL + synchronous=Normal 是生产 SQLite 的标准组合。

18.8 busy_timeout 和并发 locking

SQLite 的 locking 层级（文件级 + 数据库级）：

• NO_LOCK → SHARED → RESERVED → PENDING → EXCLUSIVE —— 按严格顺序升级。
• 多读可以并发（SHARED）。
• 把改动写进数据库文件必须 EXCLUSIVE——和所有读互斥（这是 rollback journal 模式的锁；WAL 模式另有一套）。

当一个连接已经在写（持有 RESERVED 或更高的锁）、另一个连接也尝试写——后者拿不到锁，返回 SQLITE_BUSY。

busy_timeout 让 SQLITE_BUSY 返回前 SQLite 先等一段时间：

SqliteConnectOptions::new()
    .busy_timeout(Duration::from_secs(5))
    // 等锁 5 秒、超时 SQLITE_BUSY

sqlx 的 SqliteConnectOptions 默认 busy_timeout = 5s（sqlx-sqlite/src/options/mod.rs:201）——比 SQLite C API 原生默认（0 = 立即 BUSY）好得多。多数场景直接用默认即可；少数需要调整的场景：

• 测试 / CLI：5 秒够。
• web backend：3-10 秒。
• 高写入并发：考虑 WAL 模式；busy_timeout 按业务能接受的最长等锁时间设——设得越短，写冲突时越早报 SQLITE_BUSY。

配合 WAL 模式——busy_timeout 更少触发（因为读不再阻塞写）。

18.9 SQLite 的类型系统

SQLite 的动态类型系统——值有类型而列类型只是建议：

CREATE TABLE t (x INTEGER);
INSERT INTO t VALUES ('hello');   -- 合法！字符串存进 INTEGER 列
SELECT typeof(x) FROM t;          -- 返回 'text'

5 种 storage class：

Class	Rust 对应
NULL	Option::None
INTEGER	i64（最大 8 字节）
REAL	f64
TEXT	String（UTF-8）
BLOB	Vec<u8>

没有更细分——没有 INT4 / INT8 / VARCHAR / DATE / TIMESTAMP。所有日期时间在 SQLite 里实际是 TEXT（ISO 8601 字符串）、REAL（儒略日）或 INTEGER（Unix timestamp）——sqlx 通过 chrono / time feature 的类型适配做编解码。

类型系统简单的好处：

• 解码简单——取列一律 sqlite3_column_value + sqlite3_value_dup，再按 storage class 用 sqlite3_value_int64 / _double / _text / _blob 读。
• 编码简单——sqlite3_bind_int / _int64 / _double / _text64 / _blob64 / _null 几个函数对应几种 storage class。

类型系统简单的代价：

• nullability 只能靠推断——列声明约束不了实际存进去的值，sqlx 解析 EXPLAIN 出来的字节码推断可空性，推不出来时才退回表列的 NOT NULL 元数据（第 11 章 §11.7；实测 LEFT JOIN 内侧列推断为 Option<String>）。
• 跨 DB 迁移需小心——SQLite 的 INTEGER 列能存字符串、Postgres 的 INT4 不能——迁移时可能暴露脏数据。

18.10 SQLite 独有特性

几条 SQLite 用户会用到的独特能力：

18.10.1 in-memory 数据库

let pool = SqlitePool::connect(":memory:").await?;

内存数据库——不落盘，最后一条连接关闭时随之消失。测试、脚本、临时计算的完美选择：没有磁盘 I/O、也不用清理文件。顺带一提，SqliteConnectOptions 的默认 filename 就是 :memory:（sqlx-sqlite/src/options/mod.rs:195）。

从 URL 解析出来的 :memory:（":memory:"、"sqlite::memory:" 都算）会被改写成共享缓存的具名内存库 file:sqlx-in-memory-{seqno}（sqlx-sqlite/src/options/parse.rs:20-24），seqno 每解析一次加一。所以同一个 SqlitePool 的多条连接看到的是同一个库（实测一条连接建表、另一条连接能查到）；但每次重新解析 URL 都会得到一个新库——§18.23 的 Any 驱动就踩在这一点上。

用 sqlx 写测试时常用 :memory:——CI 不需要真 DB。

18.10.2 ATTACH 多库

ATTACH DATABASE 'other.sqlite' AS other;
SELECT * FROM other.users;

一次连接操作多个 DB 文件——SQLite 独有。sqlx 支持——在 SQL 层跑 ATTACH 就行、driver 不需要特殊适配。注意 ATTACH 只对执行它的那条连接生效，用 Pool 时要放进 SqlitePoolOptions::after_connect，否则别的连接看不到 other。

18.10.3 Serialize / Deserialize

sqlite3_serialize / sqlite3_deserialize（SQLite 3.23.0 引入，官方文档）——把整个 DB 导出成一个 byte buffer、或者从 buffer 加载。sqlx-sqlite 通过 Command::Serialize / Deserialize 暴露这条能力：

let mut conn = pool.acquire().await?;
let bytes: SqliteOwnedBuf = conn.serialize(None).await?;  // 导出（sqlx-sqlite/src/connection/serialize.rs:29）

用途：备份（把整个 DB 存到 blob 字段 / 文件）、克隆（从 buffer 恢复一个内存 DB）。

18.10.4 自定义排序规则（collation）

SQLite 允许把 Rust 比较函数注册成排序规则——运行时的公开入口是 LockedSqliteHandle::create_collation（sqlx-sqlite/src/connection/mod.rs:385），底下那层 sqlite3_create_collation_v2 的 FFI 包装在 sqlx-sqlite/src/connection/collation.rs:82：

let mut conn = pool.acquire().await?;
let mut h = conn.lock_handle().await?;
h.create_collation("chinese", |a, b| a.cmp(b))?;  // 自定义排序

之后 SQL 里用 ORDER BY name COLLATE chinese——按自定义规则排。

18.10.5 Update hook

sqlx-sqlite 支持注册update hook——每次 UPDATE/INSERT/DELETE 时触发 Rust 回调：

handle.set_update_hook(|result: UpdateHookResult| {
    println!("{:?} on {}:{} rowid {}", result.operation, result.database, result.table, result.rowid);
});

handle 同样来自 conn.lock_handle().await?。回调挂在这一条连接上、在它的 worker 线程里执行（实测线程名是 sqlx-sqlite-worker-N），只看得到经这条连接做的改动——别的连接、别的进程改库不会触发（实测另一条连接 INSERT，回调不响）。所以它适合做本连接的写后缓存失效之类的事，替代不了 Postgres 的 trigger + NOTIFY 那种跨连接通知。

18.11 sqlx-sqlite 的 run 路径

总结一下 SQLite 的 run 等价路径——从用户代码到 C API：

// 用户代码
let mut conn = pool.acquire().await?;
sqlx::query("SELECT ...").bind(id).fetch_one(&mut *conn).await?;

// ↓ Executor::fetch_one 的默认实现调 fetch_optional
// ↓ sqlx-sqlite 的 impl Executor for &mut SqliteConnection::fetch_optional（limit = Some(1)）
// ↓ ConnectionWorker::execute：参数 into_static 全量拷贝，建有界行通道，
//   send_async 发 Command::Execute 到 worker
// ↓ await 从行通道 recv Either<QueryResult, Row>

// Worker 线程（不 async）：
// ↓ execute::iter(&mut conn, query, args, persistent)
//   （persistent 只在带参数时为 true，这时才进语句缓存）
// ↓ 缓存未命中：sqlite3_prepare_v3；命中：先 reset
// ↓ sqlite3_reset + sqlite3_clear_bindings，再 sqlite3_bind_* × N（text/blob 用 *64 + SQLITE_TRANSIENT）
// ↓ loop { sqlite3_step → SQLITE_ROW 或 SQLITE_DONE }
// ↓ 每行对各列 sqlite3_value_dup 构造 SqliteRow，flume 同步 send
// ↓ 迭代器 drop 时 sqlite3_reset

核心 async/sync 边界：主线程 async 等 flume channel；worker 线程 sync 调 C API。主线程一侧只用 send_async / recv_async，不做阻塞调用（Drop 事务时 start_rollback 的同步 send 是例外，§18.3）。

18.12 SQLite 驱动的整体架构图

用 mermaid 画 sqlx-sqlite 的主线程 + worker 线程互动：

flowchart LR
    subgraph 主线程 async
        U["用户代码 pool.fetch_one"]
        SC["SqliteConnection"]
        CTX["flume Sender Command"]
        FRX["flume Receiver Result"]
    end
    subgraph worker OS thread 同步
        WR["消息泵 for cmd in command_rx"]
        H["ConnectionHandle sqlite3 指针"]
        C["libsqlite3 C 函数"]
    end
    U --> SC
    SC --> CTX
    CTX ==command==> WR
    WR --> H
    H --> C
    C --> H
    H --> WR
    WR ==result==> FRX
    FRX --> SC
    SC --> U
    style WR fill:#fef3e8
    style C fill:#f6f7f0

关键观察：

• 主线程走 async channel 做消息传递——永远不 block Tokio runtime。
• worker 线程走 sync C API ——阻塞自己但只自己。
• 两个世界通过 channel 隔离——共享的只有 WorkerSharedState：两个 atomic 计数，以及那把平时由 worker 持有、只在 lock_handle() 时借出的 conn Mutex。

这条架构让 sqlx-sqlite 能在 Tokio 里自然地用——虽然底层 SQLite 是同步库。代价是每条连接一个 OS 线程——max_connections(N) 就是 N 个 OS 线程、每个要一份栈。worker.rs 顶部的 TODO 注释也承认了这点，说未来想改成"每个 pool 共用一个线程池以降低 OS 资源占用"，但标为低优先级，理由是 "a high concurrent load for SQLite3 is very unlikely"。SQLite 写并发本来就低、连接数一般不大——这个代价可接受。

18.13 阶段小结

前 12 节讲清楚 sqlx-sqlite 的独特架构：

1. SQLite 的本质（§18.1）—— 无网络、C API 同步、单文件、内嵌。和 Postgres/MySQL 完全不同。
2. ConnectionWorker 架构（§18.2）—— 专用 OS 线程 + flume channel，让同步 C API 变 async。
3. Command 枚举 + 回复 channel（§18.3）—— Execute 用有界 flume（流式）、Begin / Commit / Rollback 用 rendezvous oneshot、其余用普通 oneshot；UnlockDb 和 Drop 发的 Rollback 不要回复。
4. 消息泵循环（§18.4）—— worker 线程串行处理 command、内部全同步 C API。
5. ConnectionHandle + unsafe Send（§18.5）—— safe/unsafe 边界、手动标注 Send（同一时刻只有一方访问）；取行时 sqlite3_value_dup 复制，跨线程担保落在 ValueHandle 上。
6. Mutex + fair 锁（§18.6）—— LockedSqliteHandle 让高级用户借出 conn；fair 语义防止 worker 抢回。
7. WAL 模式（§18.7）—— 读写并发、生产必开。
8. busy_timeout（§18.8）—— 多进程共享时的等锁时限、sqlx 默认 5s 已合理（覆盖 C API 原生 0 的默认）。
9. SQLite 类型系统（§18.9）—— 5 种 storage class、极简但 nullability 弱。
10. SQLite 独有特性（§18.10）—— in-memory、ATTACH、Serialize、collation、update hook。
11. run 路径（§18.11）—— 主线程 async channel + worker sync C API 的混合模式。

18.14 SQLite 配置的生产建议

生产用 SQLite 的典型配置：

use std::time::Duration;
use log::LevelFilter;
use sqlx::ConnectOptions; // log_statements / log_slow_statements 是这个 trait 的方法
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};

let pool = SqlitePoolOptions::new()
    .max_connections(10)              // 写并发低、读多可以高些
    .connect_with(
        SqliteConnectOptions::new()
            .filename("app.sqlite")
            .create_if_missing(true)
            .journal_mode(SqliteJournalMode::Wal)          // 关键
            .synchronous(SqliteSynchronous::Normal)         // 性能 + 安全平衡
            .busy_timeout(Duration::from_secs(5))           // 等锁 5s
            .foreign_keys(true)                              // 开启外键约束
            .pragma("cache_size", "-20000")                  // 20MB cache
            .pragma("temp_store", "memory")                  // 临时表用内存
            .log_statements(LevelFilter::Debug)
            .log_slow_statements(LevelFilter::Warn, Duration::from_millis(100))
    ).await?;

八条配置的意义：

1. create_if_missing(true) —— 文件不存在自动建。
2. journal_mode(Wal) —— sqlx 默认不设 journal_mode（源码注释：WAL 是数据库的永久属性，进出 WAL 需要一把等不了 busy_timeout 的独占锁，见 PR #1930 的讨论）——要 WAL 就得自己写这一行。
3. synchronous(Normal) —— WAL 模式下提交时不 fsync，只在 checkpoint 时 fsync——数据库不会损坏、应用进程崩溃也不丢数据，但断电或操作系统崩溃可能丢掉最近提交的事务（官方 PRAGMA 文档）。换来的是 fsync 次数大幅减少。
4. busy_timeout(5s) —— 多进程 / 多连接时等锁。
5. foreign_keys(true) —— SQLite 自己默认关外键检查，但 sqlx 已经替你打开了（默认 pragma 表里 foreign_keys 就是 Some("ON")，sqlx-sqlite/src/options/mod.rs:184）——写出来只是让配置更显式。
6. cache_size -20000 —— 负数表示 KiB、正数表示页数。-20000 约 20MB。默认 -2000（约 2MB，实测 PRAGMA cache_size 返回 -2000），生产可以调大。
7. temp_store memory —— 临时表用内存而不是磁盘——SORT / DISTINCT 等操作加速。
8. 日志配置——和 Postgres/MySQL 类似。

这八条是 SQLite 生产部署的常用清单——不配时 journal 是 DELETE 模式（实测默认 PRAGMA journal_mode 为 delete）、synchronous 是 FULL、cache 只有约 2MB；外键 sqlx 已默认替你打开。

18.15 SQLite 的使用场景

SQLite 在哪些场景下是更好的选择？

1. CLI 工具 / 桌面应用——零运维、打包方便、无 server 开销。
2. 测试数据库——用 sqlx 写测试时常用 :memory:——启动快、测完消失。
3. 嵌入 / 边缘设备——移动 App、IoT、Edge computing。
4. 小流量 web——WAL 模式下读并发不互斥，写不密集的服务用 SQLite 完全够用。
5. 只读数据——文档数据、配置数据——不需要 Postgres 的级别。
6. 原型 / MVP——早期产品用 SQLite 起步、验证后再决定是否换 Postgres。

不适合 SQLite 的场景：

• 高写并发——同一时刻只有一个写事务（§18.21），WAL 模式有帮助但终究有瓶颈。
• 多节点集群——SQLite 是单文件、不能水平扩展。
• 数据量很大——索引构建、备份都会变慢（界线取决于硬件和负载，需自己压测）。
• 复杂查询优化——SQLite query planner 不如 Postgres。
• 需要复制——SQLite 没有原生 replication（有第三方如 Litestream 补）。

一句话——SQLite 的适用场景比大多数人想的更广，关键是开 WAL + 给够 cache。要判断它顶不顶得住你的负载，只有一个办法：拿你自己的数据和查询去压测。

18.16 sqlx-sqlite 和 rusqlite 的对比

Rust 生态里 SQLite 的主要库：

• rusqlite —— 经典选择、同步 API、包装 libsqlite3。
• sqlx-sqlite —— async 包装、和 sqlx 生态集成。

对比：

维度	rusqlite	sqlx-sqlite
同步/异步	同步（用户自己 spawn）	异步（sqlx 封装 worker）
API 风格	直接 SQLite API	sqlx Executor 统一
跨 DB 代码	不支持	支持（sqlx-core）
query! 宏	无	有
类型系统	自己的 trait	sqlx Type/Encode/Decode
性能	直接同步调用 libsqlite3	每次调用多一次 channel 往返
适合场景	CLI / 底层工具	async web / 统一 sqlx 生态

选择原则：

• web 后端用 sqlx——和 Pool / Transaction 统一管理。
• 命令行 / 工具用 rusqlite——直接 API 更简洁、无 async 开销。
• 混合用也合理——业务主路径 sqlx、离线脚本 rusqlite。

sqlx-sqlite 的 worker 线程模式让它在async web 场景里更顺手——rusqlite 在 async 场景下要自己 spawn_blocking 每个 query 或自己维护专用线程、不如 sqlx 内建的 worker 省事。

18.17 sqlx-sqlite 代码品质观察

读 sqlx-sqlite 的 10061 行源码、几条品质观察：

1. 合理使用 unsafe——C FFI 交互必然有 unsafe、sqlx 把它严格封装在 ConnectionHandle 等边界类型里——上层代码 safe。

2. worker 模式优雅——对比"每 query spawn_blocking"的替代方案，worker 持续运行 + channel 通信节省 spawn 开销 + 保持 statement cache。

3. Command enum 扩展性——加新 Command variant 是加新功能的标准路径（Ping / Serialize / Deserialize 都这么加）——低耦合。

4. LockedSqliteHandle 提供逃生舱——as_raw_handle() 本身是 safe 方法，交出的是裸指针 NonNull<sqlite3>，用户拿它自己调 libsqlite3 API 时必须写 unsafe；源码文档还提醒这层 sqlite3 类型不受 semver 保护。这条"给出逃生舱、把 unsafe 留给调用方显式承担"是好设计。

5. PRAGMA 配置的灵活性——SqliteConnectOptions::pragma 让用户设任意 PRAGMA——不用 sqlx 代码穷举所有 PRAGMA 名字。

6. in-memory DB 测试 friendly——:memory: URL 让 CI 不需要任何设置——用 sqlx 写的测试直接受益。

7. 不替用户擅自改库的持久属性——sqlx 默认不发 journal_mode / synchronous 这两条 PRAGMA（§18.14），只在文档里推荐 WAL + synchronous=Normal。理由是 WAL 是数据库文件的永久属性、切换需要一把等不了 busy_timeout 的独占锁——把这个决定留给用户，比默认替他改掉更稳妥。

18.18 本章的一个哲学观察

sqlx-sqlite 有一条值得思考的设计哲学——它承认 SQLite 的限制、不强求对齐 Postgres/MySQL。具体体现：

• SQLite 的同步 C API——sqlx 不试图"改成 async"、而是接受专用线程 + channel的代价。
• SQLite 的弱类型系统——sqlx 不假装列类型可靠，可空性靠解析 EXPLAIN 字节码推断（§18.9），推不出来就退回表列元数据。
• SQLite 的单写者——sqlx 不用复杂 lock 协调、让 busy_timeout 处理。

这条"尊重底层库的性格" 的哲学让 sqlx-sqlite 的代码可维护——不和 SQLite 斗争、而是配合 SQLite。结果是 10061 行代码（比 MySQL 的 9195 行还多一点），但结构清晰。

相反——如果 sqlx 试图"让 SQLite 像 Postgres"（在 SQLite 前面做协议翻译、模拟服务端语义等）——代码会膨胀、维护成本飙升。懂得何时不做抽象是优秀库作者的标志。

这对你设计自己的 Rust 库有借鉴——包装底层 API 时、尊重它的形状、不勉强对齐别的库。

18.19 本章的技术 takeaway

读完本章你应该能：

1. 解释为什么 SQLite 不能像 Postgres 那样走 async 协议。
2. 描述 ConnectionWorker 架构——worker 线程 + channel 通信。
3. 说出 Command enum 每种 variant 的大致作用。
4. 理解 Mutex fair 锁的必要性（UnlockDb 场景）。
5. 配置生产级 SQLite（WAL + busy_timeout + cache_size）。
6. 选择何时用 SQLite 何时用其他 DB。
7. 对比 sqlx-sqlite 和 rusqlite 的适用场景。
8. 用 :memory: 写 sqlx 测试。
9. 理解"包装同步库成 async"的通用模式、迁移到自己项目。

18.20 SQLite 的 WAL 细节

WAL 模式是 SQLite 生产配置的核心、值得展开讲：

WAL 的工作机制：

1. 写事务 append 到 *-wal 文件（DB 文件不变）。
2. 读事务直接读 DB 文件 + 读 WAL（没被 checkpoint 的部分）。
3. 定期 checkpoint——把 WAL 内容写回 DB 文件，之后 WAL 从头复用。

两个配套文件：

• app.sqlite-wal——Write-Ahead Log。
• app.sqlite-shm——Shared Memory（多进程协调用）。

关键配置：

• wal_autocheckpoint —— 自动 checkpoint 的阈值（默认 1000 页，~4MB）。
• journal_size_limit —— WAL 被重置复用时，文件超过这个大小就截断到这个大小。
• wal_checkpoint(PASSIVE|FULL|RESTART|TRUNCATE) —— 手动 checkpoint。

WAL 的权衡：

• 优点：读写并发、写性能高、崩溃恢复好。
• 缺点：需要各进程共享内存（-shm 文件经 mmap），网络文件系统（NFS 等）上不能用、backup 更复杂（要同时复制 DB + WAL）、两种模式切换需独占访问。

生产选 WAL 几乎一定对——除非你在 NFS 之类奇葩存储上。

18.21 SQLite 和并发写

SQLite 最大的短板是并发写——即使 WAL 模式下，同一时间只有一个写事务。

这不是 sqlx 的限制、是 SQLite 本身的语义——官方 WAL 文档写得很直白：WAL 允许多个读者和一个写者同时进行。加连接数不会提高写吞吐：

• 写密集 workload：写事务在文件层面排队；max_connections 调大只是多开 worker 线程，写这一侧仍然串行。
• 读密集 workload：WAL 下读之间不互斥，多开连接确实有收益。
• 混合 workload：看读写比例——读多于写时 SQLite 友好。

具体能扛多少 TPS 完全取决于机器、文件系统、synchronous 设置和事务大小——本专栏不给数字，请在你自己的环境上压测。真顶不住了，方向是 shard（多个 SQLite 文件）或换 Postgres——别硬撑。

18.22 sqlx-sqlite 为什么不基于 rusqlite

sqlx-sqlite 不是直接用 rusqlite——而是直接用 libsqlite3-sys（libsqlite3 的 Rust FFI 绑定）。为什么？

• rusqlite 是同步 API——sqlx 要 async、直接用同步库会"多一层"。
• rusqlite 有自己的抽象——Connection / Statement / Rows——sqlx 需要的是自己的抽象、不是嵌套一层。
• 灵活性——直接用 libsqlite3-sys 让 sqlx 能精细控制每个 C 调用、不受 rusqlite 的限制。

sqlx-sqlite 的 10061 行代码就是"自己那一套 libsqlite3 包装 + async worker 层"。这让它不依赖 rusqlite 的版本节奏、独立演进。

这条"直接对最底层接口编程、不包装更高层库"的选择是 sqlx 的标准做法——其他 driver 也类似：sqlx-postgres 不依赖 tokio-postgres、sqlx-mysql 不依赖 mysql_async，而是自己实现线协议。独立实现 + 统一 sqlx 风格——代价是更多代码、收益是完全控制。

18.23 一个实战例子：用 SQLite 测试 Postgres 代码

一条实用技巧——用 SQLite :memory: 跑 Postgres 代码的集成测试：

#[cfg(test)]
mod tests {
    // 注意 re-export 的位置：`AnyPool` 在 crate 根（`sqlx/src/any/mod.rs` 的
    // `pub(crate) mod reexports` 再由 `sqlx-0.8.6/src/lib.rs:68` 抬到根），
    // `AnyPoolOptions` 则在 `sqlx::any` 下。
    use sqlx::AnyPool;
    use sqlx::any::AnyPoolOptions;

    #[tokio::test]
    async fn test_create_user() {
        sqlx::any::install_default_drivers();

        // CI 里用 SQLite。max_connections(1) 不能省：Any 驱动每开一条连接都重新解析一次 URL，
        // 每条连接各得一个新的内存库（§18.10.1），多连接时建表和插入可能落在不同的库上
        let pool: AnyPool = AnyPoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:").await.unwrap();

        // 建 schema
        sqlx::raw_sql("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
            .execute(&pool).await.unwrap();

        // 业务代码走 AnyPool——跨 DB 兼容
        let id: i64 = sqlx::query_scalar("INSERT INTO users (name) VALUES ($1) RETURNING id")
            .bind("Alice").fetch_one(&pool).await.unwrap();

        assert_eq!(id, 1);
    }
}

用 Any 驱动 + SQLite :memory:——CI 零配置跑测试、不需要真 Postgres（sqlx 0.8.6 实测：加 max_connections(1) 后返回 id = 1；去掉它，默认多连接下 INSERT 报 no such table: users）。SQLite 能认 $1 这种占位符，所以这条 SQL 两边都能跑。但SQL 方言必须是 cross-compatible 的子集——不用 Postgres 特有语法（数组、JSONB、RETURNING 复杂形式等）。

这是 sqlx 跨 DB 能力的实用落点——测试用 SQLite 加速、生产用 Postgres。前提是代码仔细处理方言。第 19 章 Any 驱动会详细讲。

18.24 sqlite-unbundled vs sqlite 的区别

sqlx 提供两个 SQLite feature：

• sqlite —— 展开成 ["_sqlite", "sqlx-sqlite/bundled", "sqlx-macros?/sqlite"]，最终打开 libsqlite3-sys/bundled：编译时自动编一份 libsqlite3 进来、不依赖系统 SQLite。
• sqlite-unbundled —— 展开成 ["_sqlite", "sqlx-sqlite/unbundled", ...]，最终打开 libsqlite3-sys/buildtime_bindgen：构建时用 bindgen 对着系统的 libsqlite3 生成绑定（所以还需要本机有 libclang）。

选哪个？

• sqlite (bundled) —— 优点：构建稳定、版本统一；缺点：要编一份 C 源码、SQLite 静态链接进二进制，升级 SQLite 得跟着升 sqlx / libsqlite3-sys。
• sqlite-unbundled —— 优点：可以独立于 sqlx 升级 SQLite、用系统或自己 fork 的版本；缺点：依赖系统 libsqlite3 版本（README 建议 3.20.0 以上，太老会链接失败），而且 bindgen 可能让构建更慢（README 原话 "Can increase build time due to the use of bindgen"）。

生产建议：

• 通用 Rust 服务：用 sqlite (bundled)——可控、跨平台一致。
• 已有系统 libsqlite3 的环境：用 sqlite-unbundled——跟随系统 SQLite 的安全更新。
• 发行版 Linux 打包：用 sqlite-unbundled——符合发行版的"共享库"哲学。

两个 feature 只开一个：它们指向 libsqlite3-sys 的不同构建方式，同时打开会把 bundled 和 buildtime_bindgen 一起传给 libsqlite3-sys。sqlx 自己没有用 compile_error! 拦这种组合（src/lib.rs 里只拦了"开 sqlite-preupdate-hook 却两个都没开"），最终用哪份 SQLite 由 libsqlite3-sys 的构建脚本决定。默认哪个都不开——SQLite 支持本来就是可选的。

18.25 SQLite 版本兼容性

sqlx 0.8.6 的 sqlx-sqlite 依赖 libsqlite3-sys 0.30.1（sqlx-sqlite/Cargo.toml:58-59；本机 bundled 构建实测 SELECT sqlite_version() 为 3.46.0）。SQLite 本身的版本兼容性：

• 3.x 之间基本兼容——老版本建的 DB 文件新版本能读。
• 向后兼容——新 SQLite 能 open 老 DB。
• 某些新功能（比如 UPSERT ... EXCLUDED、JSON_* 函数）只在新版本——用要求 min SQLite version。

bundled 模式下用的是 libsqlite3-sys 那个版本自带的 SQLite 源码。unbundled 模式下看系统装的是哪个版本——发行版自带的可能比 bundled 的旧。

生产遇到"sqlx 报 syntax near xxx"——先检查 SQLite 版本。SELECT sqlite_version() 能看 runtime 版本。

18.26 SQLite 的 backup 和迁移

SQLite 是单文件 DB——backup 方式多样：

1. 文件级 backup——cp app.sqlite backup.sqlite + 同时复制 app.sqlite-wal（WAL 模式）。必须同一时刻快照否则不一致。

2. sqlite3_backup_init API——SQLite 提供的 online backup。sqlx 没有封装它（sqlx-sqlite 里没有任何 sqlite3_backup_* 调用）；要用得经 lock_handle() 拿裸指针自己调，或者退而用 serialize() 导出整库（§18.10.3）。

3. .dump CLI——sqlite3 app.sqlite .dump > backup.sql——SQL 文本 backup，可读。

4. Litestream——第三方工具、WAL-based 持续复制到 S3 / FS。

生产推荐——Litestream + 定期 .dump——双保险。sqlx 代码里不直接涉及 backup、但知道这些工具对运维很重要。

18.27 SQLite worker 模式的通用启示

本章讨论的"worker 线程 + channel 包装同步库"模式可以迁移到任何同步库的 async 包装项目——几条通用经验：

1. 一个资源一个 worker——sqlx-sqlite 是每连接一个 worker、不是全局一个 worker。这让每连接的操作串行（符合 sqlite3 的线程要求）、不同连接并发（多 worker 并发）。

2. channel capacity 要有界——flume::bounded 而不是 unbounded。防止主线程发送太快撑爆内存（主线程 async、可以快发；worker 同步、处理慢）。有界 channel 自动背压。

3. command 要带 reply channel——send command 后 await reply——这是请求/响应模式的标准。单次 reply 用 oneshot、流式 reply 用 flume/mpsc。

4. worker drop 时 graceful shutdown——SqliteConnection 没有手写 Drop：它被 drop 时字段 ConnectionWorker 跟着 drop，唯一的 command_tx 随之释放、channel 关闭、worker 的 for-loop 退出、最后一个 Arc<WorkerSharedState> 释放，ConnectionState 先清掉缓存语句和各类 hook，ConnectionHandle 再析构调 sqlite3_close、线程结束。SqliteConnection 的文档（sqlx-sqlite/src/connection/mod.rs:53-57）写明：这条路上出错没法把错误交还用户；想拿到错误就调 close()：它发 Command::Shutdown 并等 worker 确认。注意 ConnectionHandle::drop 里 close 失败是 panic，不是静默忽略。

5. Arc + atomic 共享少量状态——Arc<WorkerSharedState> 里的 transaction_depth / cached_statements_size 是 AtomicUsize——worker 写、主线程读。只用于不需要和 channel 同步的状态——和 channel 互补。

6. Mutex 给高级用户借出——LockedSqliteHandle 让用户直接访问 worker 的 conn——但通过 lock/unlock 协议维护并发安全。给逃生舱同时保证不变量。

18.28 ConnectionWorker：flume + oneshot 的协作

§18.2 已经给出 ConnectionWorker / WorkerSharedState 的定义（sqlx-sqlite/src/connection/worker.rs:32-42），这里换个角度看它的两条通道和三种回复。

两条通道：

• command_tx: flume::Sender<(Command, Span)>—— 异步 task 发命令给 worker 线程—— flume 本身是 mpmc channel（这里只有一个发送端、一个接收端）、支持 sync/async 两端混用——sqlx 异步端发、worker 同步端收。
• shared: Arc<WorkerSharedState>—— 不走 channel 的共享状态—— transaction_depth（可由异步端直接原子读）+ cached_statements_size（统计用）+ conn: Mutex<ConnectionState>（平时由 worker 持有，lock_handle() 时借给调用方，§18.6）。

Command enum 是 worker 的 RPC 接口（:54-100，完整定义见 §18.3）。

响应通道分三种：

• oneshot::Sender—— 单次响应（Prepare / Describe / Serialize / Deserialize / ClearCache / Ping / Shutdown）—— futures_channel::oneshot 提供—— 一次性 channel。
• rendezvous_oneshot::Sender—— 事务三兄弟（Begin / Commit / Rollback）专用、模块内自定义：发送要等接收方确认取走才算完成，见 §18.3。
• flume::Sender—— 多次响应（Execute 流式返回多行）—— 异步端把 Receiver 转成 stream 读—— worker 端 send() 发每一行—— 天然 stream。

设计巧妙在哪里—— Command 枚举把所有可能的 worker 操作枚举出来—— worker loop 只需 match 分派—— 一个 loop、一份状态、串行执行所有操作—— 没有锁粒度问题。

为什么用 flume 而不是 tokio::sync::mpsc—— 关键不在快多少，而在 flume 的同一个 channel 同时提供同步 API（send / recv / for ... in rx）和异步 API（send_async / recv_async）。worker 是普通 OS 线程、只能用同步端；调用方在 async 上下文、要用异步端。tokio::sync::mpsc 的 blocking_recv 需要在 runtime 之外的线程上用且限制更多，flume 这种"两端各取所需"的形态刚好贴合 sync/async 桥接场景，也让 sqlx-sqlite 不必绑死 Tokio。

这套"Command enum + 几种 response channel + 一个 worker loop"模式是"async 包装同步库" 的标准姿势—— 值得你在类似项目里复制。

18.29 SqliteConnectOptions 的关键字段对应 PRAGMA

SQLite 的 URI 文件名只定义了 mode / cache / immutable / vfs 等少数参数—— 其余配置 sqlx 通过 SqliteConnectOptions 的 builder 方法暴露：大部分 setter 写进一张有序的 pragmas 表（IndexMap），连接建立后按表里的顺序发 PRAGMA 语句。几个关键设置（源码 sqlx-sqlite/src/options/mod.rs）：

• journal_mode(SqliteJournalMode)（Delete / Truncate / Persist / Memory / Wal / Off）—— 生产推荐 WAL；sqlx 默认不发这条 PRAGMA，只有你显式设了才发（原因见 §18.14 第 2 条）。
• synchronous(SqliteSynchronous)（Off / Normal / Full / Extra）—— sqlx 同样默认不发，于是走 SQLite 自己的默认 FULL—— WAL 模式下 Normal 通常够用、fsync 次数明显减少。
• busy_timeout(Duration)—— 默认 5s—— 但它不是通过 PRAGMA 下发的，而是建连接时直接调 C API sqlite3_busy_timeout(handle, ms)（sqlx-sqlite/src/connection/establish.rs:282-285）—— 遇到锁竞争时 SQLite 内部自动 retry 这么久。
• foreign_keys(bool)—— 默认开—— 落到默认 pragma 表里的 foreign_keys = ON—— SQLite 自己默认关外键约束、sqlx 反其道开启—— 避免新手踩坑。
• locking_mode(SqliteLockingMode)（Normal / Exclusive）—— Exclusive 让连接一直持有文件锁、不在每个事务结束时释放，代价是别的进程完全打不开这个库。默认不设。
• auto_vacuum(SqliteAutoVacuum)（None / Full / Incremental）—— 空间回收策略—— 长寿命 DB 重要。
• pragma(key, value)—— 任意 PRAGMA 兜底，写进同一张 pragmas 表—— 覆盖 sqlx 没直接封装的。

sqlx 做对的一件事—— foreign_keys 默认开、和 SQLite 原生 C API 行为相反—— 用户开心。这种"好的默认值压倒库的历史惯性"是 sqlx 整体风格的缩影。

URL 和 builder 并不等价—— sqlx 0.8.6 的 URL 解析（sqlx-sqlite/src/options/parse.rs）只认 mode（ro / rw / rwc / memory）、cache（private / shared）、immutable、vfs 四个参数，其他参数一律报错：实测 sqlite:?journal_mode=WAL&busy_timeout=2000 解析失败，报 unknown query parameter `journal_mode` while parsing connection URL。所以上面这些设置只能走 builder；常见做法是 URL 放外部配置、解析成 SqliteConnectOptions 后再链式调 setter。

18.30 StatementHandle::step：unsafe + retry 的双重处理

SQLite 的核心执行循环是 sqlite3_step() C API——sqlx 包装在 StatementHandle::step（sqlx-sqlite/src/statement/handle.rs:331-351，下面是逐字摘录，只删掉了几行注释）：

pub(crate) fn step(&mut self) -> Result<bool, SqliteError> {
    // SAFETY: we have exclusive access to the handle
    unsafe {
        loop {
            match sqlite3_step(self.0.as_ptr()) {
                SQLITE_ROW => return Ok(true),
                SQLITE_DONE => return Ok(false),
                SQLITE_MISUSE => panic!("misuse!"),
                SQLITE_LOCKED_SHAREDCACHE => {
                    unlock_notify::wait(self.db_handle())?;
                    sqlite3_reset(self.0.as_ptr());
                }
                _ => return Err(SqliteError::new(self.db_handle())),
            }
        }
    }
}

五条返回路径：

• SQLITE_ROW → 有行可读、返回 Ok(true)—— 调用方 column_* 取值。
• SQLITE_DONE → 执行完毕、返回 Ok(false)—— 调用方退出循环。
• SQLITE_MISUSE → panic！—— API 使用错误、说明调用方（也就是 sqlx 自己）用错了 SQLite API—— 继续跑只会错上加错、直接 panic。
• SQLITE_LOCKED_SHAREDCACHE → shared cache 被另一连接锁住—— 调 unlock_notify::wait 等通知、sqlite3_reset 重置语句、循环重试。
• 其他—— 转 SqliteError 返回。

unlock_notify::wait 是 sqlite 特有机制—— sqlite3_unlock_notify API 注册回调、在另一连接 release 锁时唤醒—— 避免 busy-wait 轮询（实现见 sqlx-sqlite/src/statement/unlock_notify.rs，用条件变量让 worker 线程阻塞等待）。注意 SQLITE_LOCKED_SHAREDCACHE 只在开了 shared cache 时才会出现——SqliteConnectOptions::new() 的 shared_cache 默认是 false，但从 URL 解析出的内存库（sqlite::memory:、mode=memory）会自动打开 shared cache（§18.10.1），同一个 Pool 的多条连接并发读写内存库时就可能走到这条分支。

// SAFETY: we have exclusive access to the handle—— 一句 comment 解释为什么 unsafe 安全—— StatementHandle 所在的 worker 线程是唯一持有者、没有并发访问—— 这是 unsafe 使用的正确姿势—— 永远写 SAFETY 注释、说明为什么契约成立。

两条 Rust unsafe 经验从这 20 行代码里能学到：

1. SAFETY 注释是契约—— 作者对 reviewer 的承诺，写清楚为什么这里的调用是安全的。
2. unsafe 块本身没有运行时开销，块的大小只影响审查范围—— 这里整个 retry loop 都包在一个 unsafe 里，审查时就得把整段都当 unsafe 代码看；一般情况下只包住 FFI 调用、其余逻辑留在 safe Rust 里更好审。

18.31 和《Tokio 源码深度解析》专栏第 16 章的关联

sqlx-sqlite 用专用 OS 线程 + channel 通信包装同步 C API——这个决定回答的正是《Tokio 源码深度解析》专栏第 16 章讨论的核心问题：阻塞调用放哪里。

对照两种路径：

方案	谁用	代价
tokio::task::spawn_blocking	零散、无状态、一次性的阻塞调用	线程不固定、与其他阻塞任务共享池（默认上限 512）、返回值只能有一个
专用 worker 线程 + channel	需要长期持有、独占访问的状态，需要流式返回	每个资源常驻一个 OS 线程

为什么 sqlx-sqlite 不用 spawn_blocking？三条都不是"快多少"的问题：

1. spawn_blocking 的线程是共享池。Tokio 1.40.0 的 max_blocking_threads 默认 512（tokio-1.40.0/tokio/src/runtime/builder.rs:289）、空闲线程 KEEP_ALIVE 是 10 秒（tokio-1.40.0/tokio/src/runtime/blocking/pool.rs:171）——也就是说你的 SQLite 调用会和 tokio::fs、Command::output 之类的阻塞任务抢同一批线程，而且每次调用落到哪个线程是不确定的。
2. SQLite 的连接是要求独占访问的长寿状态。prepared statement cache（ConnectionState 里的 statements）、sqlite3_busy_timeout（sqlx-0.8.6/sqlx-sqlite/src/connection/establish.rs:285）等都跟着那个 *mut sqlite3 走；它不绑定某个线程（lock_handle() 就能把它借给别的线程），但必须独占：ConnectionHandle 是个 NonNull<sqlite3> 的 newtype（sqlx-0.8.6/sqlx-sqlite/src/connection/handle.rs:16），靠一句 unsafe impl Send for ConnectionHandle {}（sqlx-0.8.6/sqlx-sqlite/src/connection/handle.rs:27）过 Send 关——这个契约要求"同一时间只有一个线程访问"。专属线程天然满足，池化线程要额外加锁才行。
3. spawn_blocking 没法表达流式返回。一次 SELECT 要吐 N 行，spawn_blocking 的返回值只有一个；sqlx-sqlite 的 Command::Execute 带一个 flume::Sender<Result<Either<SqliteQueryResult, SqliteRow>, Error>>（sqlx-0.8.6/sqlx-sqlite/src/connection/worker.rs:67），worker 每 step 出一行就 send 一行。

专用 worker 的形态是：一条 SQLite 连接 = 一个专属 OS 线程（thread::Builder::new().name(params.thread_name.clone()).spawn(..)，sqlx-0.8.6/sqlx-sqlite/src/connection/worker.rs:106-108）+ 一个 flume::bounded(params.command_channel_size) 命令 channel（sqlx-0.8.6/sqlx-sqlite/src/connection/worker.rs:109；发送端类型见 sqlx-0.8.6/sqlx-sqlite/src/connection/worker.rs:33 的 command_tx: flume::Sender<(Command, tracing::Span)>）——调用方只做 send_async + recv_async，不碰 Tokio 的 blocking pool。代价是每连接一个线程（§18.12 已讨论）。

两个专栏对照着读的收益——《Tokio 源码深度解析》专栏第 16 章让你分辨两种方案的 trade-off；本章给你 sqlx-sqlite 的具体落地——一个阻塞 C 库接入异步运行时的完整案例，模式可复制到 librdkafka、librocksdb 等其他同步 C 库。

下一章讨论 Any 驱动——运行时多态、类型擦除的代价。