SQLx 源码深度解析

第16章 Postgres 驱动：协议、管道、Extended Query

作者杨艺韬 · 12,619 字 · 发布于 2026-04-24 · 更新于 2026-09-24

Postgres 有两套查询协议。sqlx 默认走 Extended Query 的消息序列：Parse、Bind、Execute、Close、Sync——每一条消息都有它固定的位置。

本章要点

sqlx-postgres（19841 行代码）实现了Postgres wire protocol 完整栈——从 TCP 建立到 SCRAM 认证、Startup 消息、Simple Query 协议、Extended Query 协议、COPY 协议、LISTEN/NOTIFY。
Postgres 协议有两大流派：Simple Query（一条 Q 消息发 SQL + 文本结果） vs Extended Query（Parse / Bind / Execute / Sync 四消息 + 二进制结果）。sqlx 默认走 Extended Query——更快、类型精确、支持 prepared。
Extended Query 的五消息序列（sqlx-postgres/src/connection/executor.rs:23-377，Parse 在 prepare、其余在 run）：Parse（编译 SQL）→ Bind（绑定参数到 portal）→ Execute（跑 portal）→ Close（清 portal）→ Sync（边界 + 触发 ReadyForQuery）。
连接建立的三阶段（sqlx-postgres/src/connection/establish.rs:18-154）：TCP/TLS 握手 → Startup + 认证（只支持 Cleartext / MD5 / SASL 三种，其余方式报协议错误——GSSAPI 这类未识别的类型在解码时就报 unknown authentication method）→ 接收 BackendKeyData / ParameterStatus / 第一个 ReadyForQuery。
pending_ready_for_query_count（sqlx-postgres/src/connection/mod.rs:69）—— 跟踪"已发多少 Sync 还没收到 ReadyForQuery"。0.8.6 没有做真正的 pipelining，它的作用是防止状态错位：wait_until_ready 靠它把上一条 query 的响应吃干净，handle_ready_for_query 用 checked_sub(1) 在收到多余的 ReadyForQuery 时直接报协议错。
prepared statement cache（StatementCache<(StatementId, Arc<PgStatementMetadata>)>，底层是 hashlink::LruCache）—— LRU 缓存，默认容量 100（PgConnectOptions 的 statement_cache_capacity）。每条 SQL 第一次执行时发 Parse、后续直接复用 statement ID 跳过 Parse；被挤出去的条目会同步发一条 Close::Statement 释放服务端资源。
PgStream（sqlx-postgres/src/connection/stream.rs）—— 协议层：内部是 BufferedSocket<Box<dyn Socket>>（sqlx-core 自己写的读写缓冲，不是 tokio 的 BufStream），TLS 在建它之前就升级好了。提供 write_msg / recv，flush 靠 Deref 借下层的。
sqlx-postgres 的 message/ 目录有 26 个 .rs 文件、共 2314 行——mod.rs 放 FrontendMessage / BackendMessage 两个 trait 和两张 format 枚举，其余按消息（族）分文件；无载荷的消息（BindComplete、CloseComplete、NoData…）只在枚举里占一项、没有独立文件。

16.1 问题引入：Postgres 协议与 Rust 异步的融合

用户写一行 sqlx::query("SELECT id, name FROM users WHERE id = $1").bind(42).fetch_one(&pool).await?——背后实际和 Postgres 服务端交换了多少消息？答案：cache 命中时是 Bind / Execute / Close / Sync 四条发出去、BindComplete / DataRow / CommandComplete / CloseComplete / ReadyForQuery 五条收回来；第一次执行还要多一轮 Parse / Describe / Sync。

每条消息都有严格的格式：单字节类型标识 + 4 字节长度 + 载荷。消息按 TCP 流顺序发送——client 和 server 各自维护一个状态机、处理消息按既定顺序。

Rust 异步的挑战：TCP 是字节流、消息边界要 Rust 代码自己识别；每条消息的序列化 / 反序列化要走二进制格式；消息相互依赖顺序（例如必须 Parse 后才能 Bind）；错误路径（ErrorResponse）可以在任何消息之间发生。

sqlx-postgres 的19841 行代码大部分就在做这些事——把 Postgres 协议完整地翻译成 Rust async 代码。本章不可能讲完每一处——但会把协议的主干和关键实现技巧拆开、让你读懂任何 sqlx-postgres 的 handler 代码。

16.2 Postgres Wire Protocol 全貌

Postgres wire protocol（https://www.postgresql.org/docs/current/protocol.html）从 PostgreSQL 6.4 的 v2.0 到 7.4 起的 v3.0，已经稳定了 20 多年（PostgreSQL 18 新增了 3.2，见 §16.20）。核心结构：

Frontend → Backend 消息（client 发给 server）：Q、P、B、E、S、X、...
Backend → Frontend 消息（server 发给 client）：R、K、Z、T、D、C、E、...

每个消息格式：

+-----------+------------+----------+
| 类型字节  | 长度 (i32) | 载荷     |
| 1 byte    | 4 bytes    | N bytes  |
+-----------+------------+----------+

类型字节：一个 ASCII 字符（多为大写字母，也有 1、2、n、t 这类），标识消息类型。Q = simple Query、P = Parse、B = Bind、E = Execute、S = Sync、Z = ReadyForQuery 等。
长度：含自身（4 字节）和载荷，不含类型字节。
载荷：消息类型特定格式，包含字符串（null-terminated）、整数（大端）、可变数据。

16.2.1 两大协议流派

Postgres 协议提供两种 query 执行流派：

Simple Query（简单查询）：

Client → Server: Q "SELECT * FROM users WHERE id = 1"
Server → Client: T (RowDescription)
Server → Client: D (DataRow) × N
Server → Client: C "SELECT 1" (CommandComplete)
Server → Client: Z (ReadyForQuery)

一条 Q 消息发 SQL 字符串——服务端按字符串解析、执行、返回结果。不支持参数绑定（SQL 里不能有 $1）、结果全用 Text format。

Extended Query（扩展查询）：

Client → Server: P (Parse, 编译 SQL)
Client → Server: B (Bind, 绑定参数)
Client → Server: E (Execute, 跑 portal)
Client → Server: C (Close Portal)
Client → Server: S (Sync, 边界)
Server → Client: 1 (ParseComplete)
Server → Client: 2 (BindComplete)
Server → Client: D (DataRow) × N
Server → Client: C (CommandComplete)
Server → Client: 3 (CloseComplete)
Server → Client: Z (ReadyForQuery)

五消息发送 + 六消息接收——复杂但好处多：

支持参数——Bind 消息带参数值、SQL 里用 $1、$2。
prepared statement —— Parse 可以命名 statement、后续 Bind 直接用名字跳过 Parse。
Binary format —— 参数和结果可以用二进制编码（协议默认仍是文本，sqlx 在 Bind 里显式要求二进制）、比文本紧凑快。
Portal —— 执行结果的游标、可以部分读取（Execute 带 limit）。

sqlx 的 Extended Query 路径是业务代码的默认路径——性能更好、功能更强。Simple Query 只在不带参数对象的执行里使用：直接把 &str 交给 executor（如 conn.execute("...")）、raw_sql()，以及 PgListener 的 LISTEN、迁移执行 SQL 文件这类内部用法。注意 sqlx::query(...) 即使一个 bind 都没有，也带着一个空的参数对象，照样走 Extended。

16.3 连接建立：七步握手

sqlx-postgres/src/connection/establish.rs 的 PgConnection::establish 方法（sqlx-postgres/src/connection/establish.rs:18-154，约 140 行）实现完整握手。简化步骤：

Step 1：TCP 连接

let mut stream = PgStream::connect(options).await?;

内部走 sqlx-core 的 net::connect_tcp / net::connect_uds（按启用的运行时落到 tokio 或 async-std 的 TCP / Unix socket），Step 2 的 TLS 升级也在这一个调用里完成（MaybeUpgradeTls）。

Step 2：SSL 升级（可选）

如果 ssl_mode 是默认的 Prefer（且编译进了 TLS 支持）或 Require/VerifyCa/VerifyFull——先发 SSLRequest（8 字节特殊消息）、server 返回 S（支持）或 N（不支持）、然后 TLS 握手；Prefer 收到 N 就继续走明文，Require 系则直接报错（sqlx-postgres/src/connection/tls.rs:18-45）。

Step 3：Startup 消息

stream.write(Startup {
    username: Some(&options.username),
    database: options.database.as_deref(),
    params: &[("DateStyle", "ISO, MDY"), ("client_encoding", "UTF8"), ...],
})?;

Startup 不带类型字节（历史遗留）——只有长度 + 载荷。载荷先是 4 字节协议版本号（sqlx 固定写 196608，即 3.0，sqlx-postgres/src/message/startup.rs:32），后面是 null-terminated key-value 对：user / database / 其他参数。

Step 4：循环接收 Authentication

loop {
    let message = stream.recv().await?;
    match message.format {
        BackendMessageFormat::Authentication => match message.decode()? {
            Authentication::Ok => { /* 认证完成，跳出 */ }
            Authentication::CleartextPassword => { /* 发 PasswordMessage */ }
            Authentication::Md5Password { salt } => { /* MD5 hash + 发 PasswordMessage */ }
            Authentication::Sasl { mechanisms } => { /* SCRAM-SHA-256 挑战响应 */ }
            // ...
        }
        // ...
    }
}

认证机制协议里有好几种——cleartext（明文，不安全）、MD5（老 Postgres）、SASL/SCRAM-SHA-256（PG 10 引入、PG 14 起为默认）、GSSAPI / SSPI（企业单点登录）等；LDAP、RADIUS 这类服务端认证方式在线路上走的其实是 cleartext 密码。sqlx 0.8.6 只处理 Cleartext / MD5 / SASL 三种，各自发送不同格式的 PasswordMessage。

SCRAM-SHA-256 的挑战响应（sqlx-postgres/src/connection/sasl.rs，225 行）是最复杂的——Client First、Server First、Client Final、Server Final 四条消息往返：客户端在 Client Final 里用 HMAC-SHA-256 算出 proof，Server Final 里的服务端签名也要用 HMAC-SHA-256 校验。sqlx 直接用 hmac + sha2 crate 实现。

Step 5：接收 ParameterStatus

服务端发一堆 ParameterStatus 消息——告诉 client 服务端的配置：server_version、TimeZone、client_encoding、integer_datetimes 等。注意 establish 的 match 里没有 ParameterStatus 分支——这些消息在下一层的 PgStream::recv 里就被消化掉了，server_version 存进 server_version_num、其余进 parameter_statuses: BTreeMap。

Step 6：接收 BackendKeyData

K + process_id (i32) + secret_key (i32)

协议上这对值用于 CancelRequest——想取消正在跑的 query 时，用 process_id + secret_key 另建一条 TCP 连接发 CancelRequest。sqlx 0.8.6 只把它们存进 PgConnectionInner::process_id / secret_key（第 12 章 §12.11），两个字段都标着 #[allow(dead_code)]——全树没有任何发送 CancelRequest 的代码，0.8.6 并没有实现查询取消。（对照：tokio-postgres 0.7.9 提供了 Client::cancel_token()。）

Step 7：接收 ReadyForQuery

Z + 状态字节 ('I' = idle, 'T' = in tx, 'E' = in failed tx)

Z 'I' 标志认证流程完全结束——连接可以开始执行 query。sqlx 在 PgConnectionInner::transaction_status = r.transaction_status 记录状态。

总共 7 步——具体消息条数取决于认证方式（SCRAM 要多 2 个来回）和服务端发多少条 ParameterStatus。不管几条，握手都比一次普通 query 贵得多——这就是 Pool 要复用连接的原因。

16.3.1 参数协商的含义

Startup 消息里发的 ("DateStyle", "ISO, MDY") 等参数是告诉 server 用什么样的格式：

DateStyle："ISO, MDY" 让 server 按 ISO 格式输出日期（2026-04-24 而不是 04/24/2026）。
client_encoding：UTF8——client 发送 / 接收字符串的编码。
TimeZone：UTC——时间戳的默认时区。
extra_float_digits：浮点数文本输出的精度。sqlx 默认发 2（sqlx-postgres/src/options/mod.rs:90），可用 extra_float_digits(None) 不发；PostgreSQL 12 起任何大于 0 的值都输出最短的精确表示。

这些参数是 per-session 的——建连接时协商好、后续 session 期间生效。SET DateStyle = ... SQL 也能动态改。sqlx 在 Startup 里一次性设好，后续每条 query 都按同一套格式解释。

16.3.2 连接建立的完整序列图

把这七步用 mermaid 画出来：

sequenceDiagram
    participant C as sqlx Client
    participant S as Postgres Server
    C->>S: TCP SYN
    S-->>C: SYN ACK
    C->>S: ACK
    Note over C,S: TCP 握手完成
    opt ssl_mode 为 Prefer / Require 等
        C->>S: SSLRequest (8 字节)
        S-->>C: S 表示支持
        C->>S: TLS ClientHello
        S-->>C: TLS 握手消息
        Note over C,S: TLS 建立
    end
    C->>S: Startup username + database + params
    S-->>C: AuthenticationSASL mechanisms
    C->>S: SASLInitialResponse client-first
    S-->>C: AuthenticationSASLContinue server-first
    C->>S: SASLResponse client-final + proof
    S-->>C: AuthenticationSASLFinal server-signature
    S-->>C: AuthenticationOk
    S-->>C: ParameterStatus server_version
    S-->>C: ParameterStatus TimeZone
    S-->>C: ParameterStatus client_encoding
    S-->>C: BackendKeyData process_id, secret_key
    S-->>C: ReadyForQuery idle
    Note over C,S: 连接就绪

图里 SCRAM 那一段是四条消息两个来回；跨机房时每个来回都要吃一次 RTT——这就是"建连接明显比跑 query 慢"的来源。所以 Pool 复用连接必不可少。

16.4 Extended Query 的五消息序列

建立连接后，真正的 query 执行走 Extended Query。以 SELECT id, name FROM users WHERE id = $1 + bind(42) 为例：

16.4.1 Parse 消息

P + len + statement_name (C-string) + query (C-string) + num_param_types (u16) + param_types (OID × num)

作用：让 server 编译 SQL 成 parse tree + 优化计划。

statement_name：空串（不命名、unnamed statement）或 sqlx_s_1（命名、prepared；前缀见 sqlx-postgres/src/io/mod.rs:33 的 NAME_PREFIX）。
query：SQL 字符串。
param_types：预先指定类型的参数个数 + 对应 OID 数组（协议注释里明确说这不等于 SQL 里的参数个数）。sqlx 会先把 PgArguments::types 里每个 PgTypeInfo 解析成 OID（prepare() 里的 conn.resolve_type_id），再一并发过去——不是传 0 让 server 自己推。

sqlx 代码（sqlx-postgres/src/connection/executor.rs:52-56，自由函数 prepare 里）：

conn.inner.stream.write_msg(Parse {
    param_types: &param_types,
    query: sql,
    statement: id,  // StatementId::UNNAMED 或命名
})?;

16.4.2 Bind 消息

B + len + portal_name (C-string) + statement_name (C-string) + param_format_count (u16)
   + param_formats (u16 × count) + num_params (u16) + params (length-prefixed bytes × num)
   + result_format_count (u16) + result_formats (u16 × count)

作用：给 statement 绑定参数值、创建 portal。

portal_name：空串（unnamed portal）。
statement_name：对应 Parse 里的名字。
param_formats：[PgValueFormat::Binary]——所有参数用二进制。
params：每个参数 len(i32) + N bytes——-1 表示 NULL、正数表示字节数（参见第 6 章 §6.4.1）。
result_formats：[PgValueFormat::Binary]——返回的行值用二进制。

conn.inner.stream.write_msg(Bind {
    portal: PortalId::UNNAMED,
    statement,
    formats: &[PgValueFormat::Binary],
    num_params,
    params: &arguments.buffer,
    result_formats: &[PgValueFormat::Binary],
})?;

16.4.3 Execute 消息

E + len + portal_name (C-string) + max_rows (i32，sqlx 里字段类型是 u32)

作用：执行 portal、返回最多 max_rows 行。max_rows = 0 表示"全部返回"。

conn.inner.stream.write_msg(message::Execute {
    portal: PortalId::UNNAMED,
    limit: 0,  // 0 = all rows
})?;

注意 limit = 0——sqlx 0.8 有一个注释（sqlx-postgres/src/connection/executor.rs:254-255）：

Non-zero limits cause query plan pessimization by disabling parallel workers

非 0 limit 会让 Postgres 关闭并行 worker——为了方便 suspend/resume、不做 parallel execute。所以 sqlx 默认 limit = 0 让 query plan 最优。

16.4.4 Close 消息

C + len + 'P' (portal) or 'S' (statement) + name (C-string)

作用：关 portal 或 statement、释放 server 资源。

conn.inner.stream.write_msg(Close::Portal(PortalId::UNNAMED))?;

只关 unnamed portal——statement 留着让 cache 复用。

16.4.5 Extended Query 的完整消息序列图

一次完整 sqlx::query("SELECT ... WHERE id = $1").bind(42).fetch_one(&pool).await? 的协议序列：

sequenceDiagram
    participant C as sqlx Client
    participant S as Postgres Server
    Note over C: get_or_prepare 命中缓存<br/>(statement id sqlx_s_1 已在 cache)
    C->>S: Bind portal + sqlx_s_1 + binary params
    C->>S: Execute portal, limit 0
    C->>S: Close portal
    C->>S: Sync
    S-->>C: BindComplete
    S-->>C: DataRow binary row
    S-->>C: CommandComplete SELECT 1
    S-->>C: CloseComplete
    S-->>C: ReadyForQuery idle
    Note over C: recv 循环处理到 ReadyForQuery 退出<br/>用户拿到 PgRow

sqlx 的写消息合并：Bind / Execute / Close / Sync 一次 flush 发 4 条消息——这么小的写缓冲通常一次 write syscall 就送完。服务端按序处理、依次返回响应。整个过程 1 个 RTT——非常紧凑。

如果 cache miss（第一次执行）：

sequenceDiagram
    participant C
    participant S
    C->>S: Parse sqlx_s_1, SQL, param_types
    C->>S: Describe Statement sqlx_s_1
    C->>S: Sync
    S-->>C: ParseComplete
    S-->>C: ParameterDescription OIDs
    S-->>C: RowDescription columns
    S-->>C: ReadyForQuery
    Note over C: 缓存好 sqlx_s_1
    C->>S: Bind + Execute + Close + Sync
    S-->>C: BindComplete, DataRow..., CommandComplete, CloseComplete, ReadyForQuery

2 个 RTT——Parse 一次、Bind/Execute 一次。之后同条 SQL 的执行都只要 1 个 RTT（cache 命中）。这是 prepared statement 的性能收益——第一次慢、后续快。

16.4.6 Sync 消息

S + len (4)

作用：协议分割点——告诉 server "之前发的消息处理完、回一个 ReadyForQuery"。

self.write_sync();
// 内部 pending_ready_for_query_count += 1

关键：Sync 是事务 atomic unit 的边界——从上次 Sync 到这次 Sync 之间的所有消息作为一个 implicit transaction 执行（如果不在显式事务中）。ErrorResponse 发生时、server 会丢弃后续消息直到下一个 Sync 才重新处理。

16.5 run 方法的完整流程

sqlx-postgres/src/connection/executor.rs:198-377 的 PgConnection::run 方法把五消息序列组织成 Rust async 代码（Parse 藏在 get_or_prepare 里）：

pub(crate) async fn run<'e, 'c: 'e, 'q: 'e>(
    &'c mut self,
    query: &'q str,
    arguments: Option<PgArguments>,
    persistent: bool,
    metadata_opt: Option<Arc<PgStatementMetadata>>,
) -> Result<impl Stream<...>, Error> {
    self.wait_until_ready().await?;  // 等之前的 ReadyForQuery

    if let Some(mut arguments) = arguments {
        // Extended Query 路径
        let (statement, metadata) = self.get_or_prepare(query, &arguments.types, persistent, metadata_opt).await?;
        arguments.apply_patches(self, &metadata.parameters).await?;  // §6.4.3
        self.wait_until_ready().await?;

        self.inner.stream.write_msg(Bind { ... })?;
        self.inner.stream.write_msg(Execute { ... })?;
        self.inner.stream.write_msg(Close::Portal(...))?;
        self.write_sync();
    } else {
        // Simple Query 路径（无参数）
        self.inner.stream.write_msg(Query(query))?;
        self.inner.pending_ready_for_query_count += 1;
    }

    self.inner.stream.flush().await?;

    // 返回 stream：用 try_stream! 循环接收消息、yield Row 或 QueryResult
    Ok(try_stream! {
        loop {
            let message = self.inner.stream.recv().await?;
            match message.format {
                BackendMessageFormat::BindComplete | ParseComplete | ... => { /* 忽略 */ }
                BackendMessageFormat::CommandComplete => { yield Left(query_result); }
                BackendMessageFormat::RowDescription => { /* 更新 metadata */ }
                BackendMessageFormat::DataRow => { yield Right(row); }
                BackendMessageFormat::ReadyForQuery => {
                    self.handle_ready_for_query(message)?;  // pending 计数 -1
                    break;  // 流结束
                }
                // 注意：ErrorResponse 在 self.stream.recv() 里就已经转成 Err 了，
                // 这里不需要（也不会）匹配到它
                _ => { return Err(err_protocol!("execute: unexpected message: {:?}", message.format)); }
            }
        }
    })
}

核心逻辑：

wait_until_ready：等之前的 query 完成（ReadyForQuery 收到）。
Extended / Simple 分叉：arguments 为 Some 走 Extended（query() 不 bind 也是 Some），为 None（直接执行 &str、raw_sql()）走 Simple。
get_or_prepare：查 statement cache、没有就 Parse、有就复用。
apply_patches：处理 PgArguments 的 type holes（第 6 章 §6.4.3）。
write_msg × 3 + write_sync：Bind / Execute / Close / Sync 四条进写缓冲（Parse 已在第 3 步单独发过或被缓存省掉）。
flush：把 write buffer 的字节送出 TCP。
try_stream! 接收：循环 recv、按 message format 分发。

这段代码把 Postgres 协议状态机和 Rust async stream 融合——对上层 Executor trait 提供 BoxStream<Either<QueryResult, Row>>、对下层用协议消息和服务端对话。

16.6 prepared statement cache

sqlx-postgres/src/connection/executor.rs:168-195 的 get_or_prepare（真正发 Parse 的是它调用的自由函数 prepare，sqlx-postgres/src/connection/executor.rs:23-102）：

async fn get_or_prepare<'a>(
    &mut self,
    sql: &str,
    parameters: &[PgTypeInfo],
    persistent: bool,
    metadata: Option<Arc<PgStatementMetadata>>,
) -> Result<(StatementId, Arc<PgStatementMetadata>), Error> {
    // 1. 查缓存（注意：这一步不看 persistent）
    if let Some(statement) = self.inner.cache_statement.get_mut(sql) {
        return Ok((*statement).clone());
    }

    // 2. 没缓存——走自由函数 prepare()：分配 StatementId、解析参数 OID、
    //    发 Parse (+ Describe) + Sync、收 ParseComplete / ParameterDescription / RowDescription
    let statement = prepare(self, sql, parameters, metadata, persistent).await?;

    // 3. 只有 persistent 且缓存开启时才写入；被 LRU 挤掉的旧 statement 要显式 Close
    if persistent && self.inner.cache_statement.is_enabled() {
        if let Some((id, _)) = self.inner.cache_statement.insert(sql, statement.clone()) {
            self.inner.stream.write_msg(Close::Statement(id))?;
            self.write_sync();
            self.inner.stream.flush().await?;
            self.wait_for_close_complete(1).await?;
            self.recv_ready_for_query().await?;
        }
    }

    Ok(statement)
}

三步：

查缓存——命中直接返回 (StatementId, Arc<PgStatementMetadata>)。注意查缓存不受 persistent 影响——persistent 只决定"要不要往里写"。
prepare()——persistent == true 时从 next_statement_id 取一个命名 ID（sqlx_s_N），否则用 StatementId::UNNAMED；把参数 PgTypeInfo 解析成 OID，发 Parse（用户没提供 metadata 时再补一条 Describe::Statement）+ Sync，然后收 ParseComplete、ParameterDescription、RowDescription。
存缓存——只在 persistent && cache_statement.is_enabled() 时写入；LRU 挤出旧条目时同步发 Close::Statement + Sync 并等 CloseComplete / ReadyForQuery，把 server 端的 prepared statement 也释放掉。

命名规则：StatementId 的名字前缀是 sqlx_s_、序号从 NonZeroU32::MIN（即 1）开始递增（sqlx-postgres/src/io/mod.rs），所以实际发给 server 的是 sqlx_s_1 / sqlx_s_2 / …；portal 的前缀则是 sqlx_p_。每连接独立——不同连接的 statement ID 空间不共享（也不需要）。

LRU 淘汰时释放 server 资源——避免 server 端累积无用 statement 占内存。这条细节体现了 sqlx 对"资源管理"的周到——不只管客户端、也管 server 端资源。

16.7 `pending_ready_for_query_count`：流水线跟踪

PgConnectionInner::pending_ready_for_query_count（sqlx-postgres/src/connection/mod.rs:69）跟踪"已发多少 Sync 还没收到 ReadyForQuery"。

理论上 Postgres 支持 pipelining——可以一次发 N 条 Sync、server 顺序处理、client 按顺序接收 N 条 ReadyForQuery。但 sqlx 0.8.6 没有做真正的 pipelining——run、prepare、prepare_with、describe 开头都先 wait_until_ready()，同一条连接上前一条 query 的 ReadyForQuery 收完才发下一条。pending_ready_for_query_count 主要是防错：

    pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> {
        if !self.inner.stream.write_buffer_mut().is_empty() {
            self.inner.stream.flush().await?;
        }

        while self.inner.pending_ready_for_query_count > 0 {
            let message = self.inner.stream.recv().await?;

            if let BackendMessageFormat::ReadyForQuery = message.format {
                self.handle_ready_for_query(message)?;
            }
        }

        Ok(())
    }

wait_until_ready 循环消耗 pending 的 ReadyForQuery——保证新 query 开始前 pipeline 清空。这是保守的做法——保证消息状态机干净、但放弃了 pipelining 能省下的往返。

pipelining 的潜在收益：

没 pipelining: send(Q1) → recv(R1) → send(Q2) → recv(R2) ——2 次 RTT
有 pipelining: send(Q1) → send(Q2) → recv(R1) → recv(R2) ——1 次 RTT

按这张示意图算，两条独立 query 从 2 个 RTT 降到约 1 个 RTT（服务端执行时间另计），跨机房 RTT 越大越明显。sqlx 0.8.6 没有提供任何 pipeline / batch API（§16.25）。

16.8 `PgStream`：缓冲 I/O 层

sqlx-postgres/src/connection/stream.rs 的 PgStream 是协议和 Tokio I/O 的桥梁。结构（简化）：

pub struct PgStream {
    inner: BufferedSocket<Box<dyn Socket>>,
    pub(crate) notifications: Option<UnboundedSender<Notification>>,
    pub(crate) parameter_statuses: BTreeMap<String, String>,
    pub(crate) server_version_num: Option<u32>,
}

分层是这样的：

Socket：sqlx-core 自己的 trait 对象（Box<dyn Socket>），底下是 TCP、Unix socket，或者被 TLS 包过一层——PgStream::connect 用 net::connect_tcp / net::connect_uds + MaybeUpgradeTls(options) 得到，TLS 升级在建 BufferedSocket 之前完成。
BufferedSocket<S>（sqlx-core/src/net/socket/buffered.rs）：自己实现的读写缓冲（WriteBuffer + ReadBuffer，默认容量 DEFAULT_BUF_SIZE = 8192），不是 tokio::io::BufStream。
PgStream：在 BufferedSocket 之上加 Postgres 语义——消息分帧、ParameterStatus / Notice / Notification 的透明消化。

PgStream 对 BufferedSocket 实现了 Deref / DerefMut，所以 write / flush / write_buffer_mut 这些方法是直接"借"下层的。

核心方法：

write_msg(impl FrontendMessage)——self.write(EncodeMessage(message))，把消息编码进 write buffer。
flush()（来自 BufferedSocket）——把 write buffer 里剩下的字节循环写进 socket。
recv()——取下一条正常消息；ErrorResponse 直接转成 Err，ParameterStatus / NoticeResponse / NotificationResponse 在循环里消化掉不返回给调用方。

缓冲的价值：多条 Extended Query 消息可以先全部 write、最后一次 flush——减少 TCP syscall。prepare 里的 Parse + Describe + Sync 是一批，run 里的 Bind + Execute + Close + Sync 是另一批；如果每条消息都单独 send，一批就是 3-4 次 write syscall，合并后小查询通常一次就写完。

这层缓冲是性能关键——sqlx-postgres/src/connection/executor.rs:292 的 self.inner.stream.flush().await? 只在所有消息写完后才调一次、压缩 syscall 数。

16.9 `message/` 模块：26 个协议消息

sqlx-postgres/src/message/ 目录有 26 个 .rs 文件——mod.rs 放公共 trait 和两张 format 枚举表，另外 25 个按消息（或消息族）分文件：

Frontend 侧：startup.rs、ssl_request.rs、password.rs、sasl.rs、parse.rs、bind.rs、execute.rs、describe.rs、close.rs、sync.rs、query.rs、flush.rs、terminate.rs。
Backend 侧：authentication.rs、backend_key_data.rs、parse_complete.rs、parameter_description.rs、parameter_status.rs、row_description.rs、data_row.rs、command_complete.rs、ready_for_query.rs、response.rs（ErrorResponse / NoticeResponse 共用）、notification.rs。
copy.rs 一个文件装下 COPY 相关的双向消息。

注意不是每种消息都有自己的文件——BindComplete、CloseComplete、EmptyQueryResponse、PortalSuspended、NoData 这些"只有一个 format code、没有载荷"的消息，只在 mod.rs 的 BackendMessageFormat 枚举里占一项，executor.rs 直接按 format 匹配、不需要 decode 结构体。

Frontend 消息实现 FrontendMessage、Backend 消息实现 BackendMessage。模式类似：

// message/parse.rs
pub struct Parse<'a> {
    pub statement: StatementId,
    pub query: &'a str,
    pub param_types: &'a [Oid],
}

impl FrontendMessage for Parse<'_> {
    const FORMAT: FrontendMessageFormat = FrontendMessageFormat::Parse;

    fn body_size_hint(&self) -> Saturating<usize> {
        let mut size = Saturating(0);
        size += self.statement.name_len();
        size += self.query.len();
        size += 1; // NUL terminator
        size += 2; // param_types_len
        size += self.param_types.len().saturating_mul(4);
        size
    }

    fn encode_body(&self, buf: &mut Vec<u8>) -> Result<(), Error> {
        buf.put_statement_name(self.statement);
        buf.put_str_nul(self.query);

        let param_types_len = u16::try_from(self.param_types.len())
            .map_err(|_| err_protocol!("param_types.len() too large for binary protocol: {}",
                                       self.param_types.len()))?;
        buf.extend(param_types_len.to_be_bytes());
        for &oid in self.param_types {
            buf.extend(oid.0.to_be_bytes());
        }
        Ok(())
    }
}

注意 encode_body 不写 format code、也不写长度前缀——这两件事由 EncodeMessage<F>（sqlx-postgres/src/message/mod.rs:210-229）统一做：先按 body_size_hint() 给 buffer try_reserve（分配失败返回 Err 而不是 panic），再 push format 字节，最后 put_length_prefixed(|buf| self.0.encode_body(buf)) 完成"先留 4 字节坑、写完载荷回填长度"（sqlx-postgres/src/io/buf_mut.rs:16-45，第 6 章 §6.4.1 见过同一手法）。回填时 i32::try_from(size) 校验消息没超协议上限。

Backend 侧是对称的一半——sqlx-postgres/src/message/data_row.rs 里 DataRow 的 decode_body 就长这个形状（列解析的具体实现略去）：

// message/data_row.rs (backend)
impl BackendMessage for DataRow {
    const FORMAT: BackendMessageFormat = BackendMessageFormat::DataRow;

    fn decode_body(buf: Bytes) -> Result<Self, Error> {
        // 解析列数 + 按列读长度 + 字节
    }
}

Backend 消息走 decode_body——buf 是 server 发来的载荷字节（不含类型和长度字段）。

这 26 个文件加起来 2314 行——协议编解码的全部实现，其中不少行是文件末尾的 #[test] 编码断言（例如 parse.rs 里直接比对 b"P\0\0\0\x26sqlx_s_1234567890\0SELECT $1\0\0\x01\0\0\0\x19" 这样的字节串）。

16.10 ErrorResponse 的处理

协议中 任何时候 server 都可能发 ErrorResponse——查询语法错、认证失败、连接被 kill、数据库挂掉。sqlx 在 stream.recv() 里统一处理：

// sqlx-postgres/src/connection/stream.rs:123-200 的 recv 简化（省略 ParameterStatus 分支）
pub(crate) async fn recv(&mut self) -> Result<ReceivedMessage, Error> {
    loop {
        let message = self.recv_unchecked().await?;
        match message.format {
            BackendMessageFormat::ErrorResponse => {
                let error: PgDatabaseError = message.decode()?;
                return Err(Error::Database(Box::new(error)));
            }
            BackendMessageFormat::NoticeResponse => {
                // Notice 只是警告——按 PgSeverity 映射日志等级、打到
                // sqlx::postgres::notice target，不返错
                continue;
            }
            BackendMessageFormat::NotificationResponse => {
                // LISTEN/NOTIFY 通知——路由到 PgListener
                if let Some(buffer) = &mut self.notifications {
                    let notification: Notification = message.decode()?;
                    let _ = buffer.send(notification).await;
                    continue;
                }
                // 没设 notifications 时落到下面，当普通消息返回
            }
            _ => return Ok(message),  // 正常消息、返回给调用方
        }
    }
}

四种非正常消息：

ErrorResponse——decode 成 PgDatabaseError 后转 Error::Database、上抛。
NoticeResponse——server 级警告（WARNING: column 'x' is deprecated），按 PgSeverity 映射到 log / tracing 的等级、打到 sqlx::postgres::notice target，不打断流程。
NotificationResponse——LISTEN/NOTIFY 通知；只有 self.notifications 已被设置（即这条连接归 PgListener 用）时才发进 channel 并 continue，否则会当普通消息返回给调用方。
ParameterStatus——server_version 存进 server_version_num（parse_server_version 转成 libpq 风格的数字），其余存进 parameter_statuses: BTreeMap。

这条 recv 里的消息过滤循环让上层代码不用处理 notice / notification——它们被透明消化。上层只收正常消息和错误——业务逻辑清爽。

16.11 Postgres 协议之外的特性

除了 Extended Query 协议栈，sqlx-postgres 还实现了几个协议扩展：

PgListener（listener.rs，517 行）——LISTEN / NOTIFY 订阅。
PgAdvisoryLock（advisory_lock.rs，421 行）——Postgres 的 app-level 锁。
COPY FROM / COPY TO（copy.rs，352 行）——批量数据导入导出的高效协议。

这些都是 Postgres 独有的高级特性——sqlx 把它们纳入驱动、让 Rust 生态拥有 Postgres 全部能力，不只是普通 SELECT/INSERT。

16.11.1 `PgListener` 简介

let mut listener = PgListener::connect_with(&pool).await?;  // 或 PgListener::connect(url)
listener.listen("channel_name").await?;

loop {
    match listener.recv().await {
        Ok(notification) => {
            println!("Got: {}", notification.payload());
        }
        Err(e) => { /* 重连等 */ }
    }
}

内部结构：PgListener 从池里 acquire 一条 PoolConnection 专用，并给它的 PgStream::notifications 装上 channel；listen 发 LISTEN "channel_name"（Simple Query），然后 recv 循环调 try_recv——后者直接走 PgStream::recv_unchecked，自己匹配 NotificationResponse 返回、顺手给 ReadyForQuery 计数减一（sqlx-postgres/src/listener.rs:256-327）。连接断了会按 eager_reconnect 重新 acquire 并重发 LISTEN。

这条特性让 Rust 里实现Postgres 作为消息队列成为可能——很多轻量消息场景不需要额外上 Kafka，Postgres 的 NOTIFY 足够。

16.12 阶段小结

到这里，sqlx-postgres 驱动的核心已经拆开：

Postgres wire protocol 两流派（§16.2）—— Simple Query vs Extended Query。sqlx 默认 Extended——类型精确、支持 prepared、二进制更快。
连接建立七步（§16.3）—— TCP / SSL / Startup / 认证 / ParameterStatus / BackendKeyData / ReadyForQuery。后三步其实是同一个 loop 里按消息类型分派的。
Extended Query 五消息（§16.4）—— Parse / Bind / Execute / Close / Sync。每条的格式和作用。
run 方法整合（§16.5）—— Rust async stream + 协议状态机。wait_until_ready / get_or_prepare / write_msg × 3 + write_sync / flush / try_stream!。
prepared statement cache（§16.6）—— LRU + 淘汰时 Close server-side statement。
pending_ready_for_query_count（§16.7）—— 跟踪未收的 ReadyForQuery、防状态错位。0.8.6 没有做真正的 pipelining。
PgStream 缓冲 I/O（§16.8）—— Box<dyn Socket>（TLS 已在建流前升级好）外面套 sqlx-core 的 BufferedSocket，write_msg 先进 buffer、一次 flush 减少 syscall。
26 个 message 文件、2314 行（§16.9）—— FrontendMessage::encode_body 只写载荷，format code 和长度前缀由 EncodeMessage + put_length_prefixed 统一"留坑回填"。
ErrorResponse 透明处理（§16.10）—— recv 循环里统一过滤 notice / notification、业务只收正常消息。
协议扩展（§16.11）—— LISTEN/NOTIFY、advisory lock、COPY——sqlx 把 Postgres 全能力纳入 Rust 生态。

16.13 Postgres 协议和 Tokio I/O 的融合点

sqlx-postgres 把 Postgres 协议翻译成 Rust async 代码的三个关键融合点值得单独讲：

融合点 1：BufferedSocket::try_read + ControlFlow 做消息分帧

Postgres 每条消息都是"1 字节类型 + 4 字节长度 + N 字节载荷"。sqlx 没有用"两次 read_exact"，而是给 BufferedSocket::try_read 传一个回调，让它在同一块读缓冲上重试（sqlx-postgres/src/connection/stream.rs:79-118 的 recv_unchecked）：

pub(crate) async fn recv_unchecked(&mut self) -> Result<ReceivedMessage, Error> {
    self.inner
        .try_read(|buf| {
            // 头都还没凑齐——告诉上层"我至少还要 5 字节"
            let Some(mut header) = buf.get(..5) else {
                return Ok(ControlFlow::Continue(5));
            };
            let format = BackendMessageFormat::try_from_u8(header.get_u8())?;
            let message_len = header.get_u32() as usize;
            let expected_len = message_len.checked_add(1)
                .ok_or_else(|| err_protocol!("message_len + 1 overflows usize: {message_len}"))?;
            if buf.len() < expected_len {
                return Ok(ControlFlow::Continue(expected_len));  // 载荷不全，继续读
            }
            buf.advance(1);                              // 吃掉 format code
            let mut contents = buf.split_to(message_len).freeze();
            contents.advance(4);                         // 切掉长度前缀
            Ok(ControlFlow::Break(ReceivedMessage { format, contents }))
        })
        .await
}

ControlFlow::Continue(n) 表示"缓冲区总长度还不够 n 字节，请再读"，ControlFlow::Break(v) 表示"已经从缓冲区里切走一条完整消息"。回调在返回 Continue 之前不修改 buffer——这正是 try_read 文档里说的 cancel-safe 前提：future 被丢弃时不会吃掉半条消息。

融合点 2：Pin<&mut Future> 配合 try_stream!

run 方法返回 impl Stream ——内部用 async-stream::try_stream! 宏。每次 recv 是一个 await 点、r#yield!(value) 产生 stream 的下一项。

这让Postgres 的消息流变成 Rust 的 Stream——上层 Executor 能用 try_next / try_collect / for_each 等 combinators 消费。

融合点 3：BufferedSocket 的空刷新短路

BufferedSocket::flush（PgStream 通过 DerefMut 借来）是 while !self.write_buf.is_empty() { ... }——buffer 空就一次循环都不进。所以像 wait_until_ready 里那样先判断 write_buffer_mut().is_empty() 再 flush，或者直接 flush，效果一样：没东西写就不会有多余的写 syscall。

这三个融合点把字节流 I/O、消息协议、Rust async stream三个抽象层串起来——每层各自简单、组合后能力强大。

16.14 Postgres 驱动的性能特征

下面几条是从协议和源码结构推得的定性特征，不带具体数字——真实数字必须在你自己的库、网络和数据形状上压测才有意义：

1. prepared statement 命中缓存时比 simple query 少一次服务端 SQL 解析、且参数和结果都走 binary format——省掉每次的 Parse 与文本/二进制转换。 2. cache hit 比 cache miss 少一个 RTT——miss 时要多一轮 Parse + Describe + Sync（见 §16.4.5 的两张时序图）。 3. pipelining 没有被利用——run 开头无条件 wait_until_ready()，同一条连接上前一条 query 的 ReadyForQuery 收完才发下一条；源码注释里明说"理论上可以在一个 Sync 前发几十条 query"，但没做成 API。 4. 大结果集是逐行解码 + yield——try_stream! 每收到一条 DataRow 就产出一个 PgRow，内存不会一次性膨胀，但每行都要过一次 DataRow::decode_body。 5. COPY 走独立的 CopyIn / CopyData 协议——不需要为每行做 Parse/Bind/Execute，是批量导入的正确工具。

这些特征让 sqlx-postgres 在大多数场景够用——真要压极致性能（比如靠 pipelining 摊平跨机房 RTT）才需要考虑更底层的方案。

16.15 和 tokio-postgres 的对比

sqlx-postgres 和 tokio-postgres（rust-postgres 生态）都实现 Postgres 协议——但定位不同：

维度	sqlx-postgres	tokio-postgres
目标用户	业务开发者	基础设施 / 协议级工具开发者
API 风格	高抽象（Pool, Query, FromRow）	低抽象（`client.query(...)`）
参数绑定	`.bind(x).bind(y)` 链式 + Encode trait	`&[&dyn ToSql]` 切片
行解码	FromRow 派生 + try_get	`row.get<T>(idx)` panic on error
Pipelining	不支持（wait_until_ready 串行）	支持（并发 poll 多个请求 future 时自动 pipelining）
编译期校验	query! 宏	无
LISTEN / NOTIFY	PgListener 高级封装（自动重连、重发 LISTEN）	无专门封装，自己从 `Connection::poll_message` 取 `AsyncMessage::Notification`
COPY 支持	有	有
查询取消	未实现（BackendKeyData 只存不用）	`Client::cancel_token()`
代码量	19841 行（`sqlx-postgres/src`）	8174 行（仅 `tokio-postgres/src`，另有 postgres-protocol / postgres-types 两个 crate）
编译时间	较长（宏系统 + derive）	较短

选择规则：

业务后端：用 sqlx-postgres——抽象 + 类型安全值得。
工具 / 代理：用 tokio-postgres——更贴近协议、更好控制。
极致性能：用 tokio-postgres + 自己的 pipeline 实现。

两者不是"谁更好"——是"针对不同用户"。sqlx-postgres 用更多代码换来了对业务开发者的更多抽象（Pool / FromRow / query! 宏）——这个交换对大多数业务型 Rust 项目合算。

16.16 sqlx-postgres 的可扩展性

sqlx-postgres 有几个扩展点让用户或上层库能定制：

1. 自定义类型的 Encode/Decode/Type——第 5 章讲过、让用户加任意 Rust 类型到 Postgres 映射。 2. PgConnectOptions::options(...)——传自定义 server options（-c statement_timeout=5000）。 3. after_connect 回调（第 13 章 §13.11）——每条新连接初始化 session 参数。 4. PgListener::connect_with(&pool)——用已有的池建 listener（从池里借一条连接专用，断线时也从这个池重新取）——和自己的连接池集成。 5. PgQueryResult 的 rows_affected——来自 CommandComplete 的 tag（INSERT 0 3 里的 3）。注意用 fetch_one / fetch_all 拿 INSERT ... RETURNING 的结果时，stream 里的 Either::Left(PgQueryResult) 会被丢掉、你只拿到行；要同时拿到影响行数得自己用 fetch_many 消费两种 variant。

生产 sqlx-postgres 用户扩展最常用的是 1 和 3——加新类型（UUID / chrono / 业务类型）+ session 初始化。其他扩展点用得少但有时关键。

16.17 本章的核心要义

回看整章、提炼三条核心要义：

1. Extended Query 协议是 Postgres 的现代标准——带参数对象的查询都走 Extended、只有少数场景（直接执行 &str、migration、raw_sql）走 Simple。理解这条协议就理解了 sqlx-postgres 绝大部分的行为。

2. prepared statement cache 是性能关键——第一次 query 慢（Parse）、后续快（直接 Bind/Execute）。persistent(true) 默认开启正是为此。

3. Rust async + Tokio I/O 是协议实现的好载体——字节流读写、缓冲、消息状态机、stream 返回——每层 Rust 生态都有合适工具。sqlx-postgres 的 19841 行代码看起来多、实际每一块都在做具体可定位的事。

读完本章你对 "Postgres server 后面到底怎么运作" 有了完整的心智模型——下次看到"为什么这条 query 这么慢"的问题时能精确分解到协议级原因（Parse 开销 / 数据量 / RTT / pipelining / 等等）。

16.18 延伸阅读

如果本章激发了你对 Postgres 协议的兴趣，推荐几个方向：

官方协议文档：postgresql.org/docs/current/protocol.html —— 消息格式与消息流的 ground truth；establish.rs 顶部的注释直接引的就是它的 protocol-flow 小节。
tokio-postgres 源码——对比 sqlx 的另一种实现风格。
Postgres server 侧——backend/libpq/ 子目录是 server 侧协议实现（C 代码）——看到 server 端怎么响应对客户端更有感觉。
libpq 源码——src/interfaces/libpq/，官方 C 客户端库；psql 就是基于它写的。sqlx 的 parse_server_version（stream.rs）注释里直接给出了 libpq fe-exec.c 的对应代码链接，可以对照着看。

这些资源加起来能让你对 Postgres 协议的理解跳出 Rust 这一层——从"会用 sqlx"走到"知道线上到底在发什么字节"。

16.19 sqlx-postgres 的代码结构导览

sqlx-postgres crate 的目录结构（约 108 个文件、19841 行）：

sqlx-postgres/src/
├── lib.rs                    顶层 re-export
├── arguments.rs              PgArguments（第 6 章）
├── column.rs                 PgColumn
├── connection/
│   ├── mod.rs                PgConnection struct
│   ├── establish.rs          握手流程
│   ├── executor.rs           run 方法 + Executor impl
│   ├── stream.rs             PgStream
│   ├── sasl.rs               SCRAM-SHA-256
│   ├── tls.rs                TLS 升级
│   └── describe.rs           Describe 查询
├── copy.rs                   COPY FROM/TO
├── database.rs               impl Database for Postgres
├── error.rs                  PgDatabaseError
├── io/                       底层 I/O helper
├── listener.rs               PgListener
├── message/                  26 个协议消息文件
├── options/                  PgConnectOptions（5 个文件、共 1391 行）
├── query_result.rs           PgQueryResult
├── row.rs                    PgRow
├── statement.rs              PgStatement
├── transaction.rs            PgTransactionManager
├── type_checking.rs          编译期类型映射
├── type_info.rs              PgTypeInfo
├── types/                    内置类型的 Encode/Decode/Type
└── value.rs                  PgValue / PgValueRef

文件组织的几个观察：

每个公共类型一个文件——PgConnection / PgStream / PgListener 等各自独立——方便 IDE 导航和 review。
connection/ 子目录放协议握手相关——stream / sasl / tls / establish / executor——这些是 Postgres 驱动最复杂部分。
message/ 独立——26 个消息文件清晰分离——符合协议级抽象的清晰度。
types/ 独立子目录——每种类型（int / str / bool / chrono / uuid / json）一个文件——加新类型不影响其他代码。
options/ 共 1391 行——拆成 mod.rs / parse.rs（URL 解析）/ pgpass.rs（~/.pgpass 支持）/ connect.rs / ssl_mode.rs 五个文件，PgConnectOptions 的一大堆配置方法占了相当篇幅。

读 sqlx-postgres 源码时按这个结构导航——先看 connection/ 了解握手、再看 message/ 了解协议、最后 executor.rs 看整合。两万行代码逐个文件读并不轻松，但有明确的阅读路径。

16.20 协议变化与驱动维护

Postgres 的前后端协议 3.0 从 PostgreSQL 7.4 起沿用了二十多年；直到 PostgreSQL 18 才新增 3.2——把查询取消用的 secret key 从 4 字节改成变长，BackendKeyData 与 CancelRequest 随之调整（见官方文档 Protocol Versions）。服务端仍接受 3.0，sqlx 0.8.6 的 Startup 固定按 3.0 握手（sqlx-postgres/src/message/startup.rs:32），又没实现查询取消，所以 3.2 对它没有影响。其余增量能力都是在 3.0 之内加的，比如 SCRAM-SHA-256（PostgreSQL 10 引入，sqlx 在 connection/sasl.rs 里实现）。

协议稳定意味着驱动维护成本低——sqlx 的协议代码几乎不用随 Postgres 新版本变。想验证这一点，看仓库的 CHANGELOG.md 就行：绝大多数 Postgres 相关条目是类型适配（chrono / bigdecimal / uuid / PgHstore 之类），而不是改协议消息。这让新贡献者加一个类型适配容易上手、不涉及协议核心。

16.21 Postgres 驱动对其他 DB 的启示

第 17 章 MySQL 和第 18 章 SQLite 的驱动结构和 sqlx-postgres 一脉相承。观察 sqlx-postgres 你会看到其他驱动的骨架：

connection/ 子目录的组织——MySQL / SQLite 也是 mod.rs + establish.rs + executor.rs + ...
message/（或等价）目录——MySQL 有 protocol/ 目录放消息、SQLite 有 statement/ 目录（SQLite 不是网络协议、statement 是等价抽象）。
types/ 目录——三家完全一致——每类型一文件。
options/ 目录——三家都有 ConnectOptions 的丰富配置。

这条一致性让 sqlx 贡献者学会一个驱动、能改所有驱动——这是 sqlx-core 的 trait 家族驱动层实现标准化的价值兑现。

读懂 sqlx-postgres，你对 sqlx-mysql / sqlx-sqlite 的理解大部分能迁移过去——剩下的是各自的协议特殊性（MySQL 的 COM_STMT_EXECUTE / SQLite 的 worker 线程）。第 17 章 / 第 18 章重点讲这部分。

16.22 实战：Postgres 特有特性的 sqlx 使用

sqlx-postgres 支持 Postgres 独有的几个特性——这些在其他 DB 上没有或形式不同、值得专门用 sqlx-postgres 享受：

1. JSONB 查询 + 索引

#[derive(sqlx::FromRow)]
struct Event {
    id: i32,
    #[sqlx(json)]
    payload: serde_json::Value,
}

// 用 JSONB 操作符查询
let events: Vec<Event> = sqlx::query_as(
    "SELECT id, payload FROM events WHERE payload @> $1"
).bind(serde_json::json!({"user_id": 42})).fetch_all(&pool).await?;

Postgres 的 JSONB 操作符（@>、?、->、->>）里，@>、? 这类能走 GIN 索引、让 JSON 字段查询变高效——sqlx 的 #[sqlx(json)] 完美配合。

2. ARRAY 类型 + UNNEST

let ids: Vec<i32> = vec![1, 2, 3, 4];

// 批量查询
let users: Vec<User> = sqlx::query_as(
    "SELECT * FROM users WHERE id = ANY($1)"
).bind(&ids).fetch_all(&pool).await?;

// 批量插入用 UNNEST（第 10 章 §10.11）
sqlx::query("INSERT INTO users (name, age) SELECT * FROM UNNEST($1::text[], $2::int4[])")
    .bind(&names).bind(&ages).execute(&pool).await?;

3. RETURNING + 批量

let new_ids: Vec<i32> = sqlx::query_scalar(
    "INSERT INTO logs (level, message) SELECT * FROM UNNEST($1::text[], $2::text[]) RETURNING id"
).bind(&levels).bind(&messages).fetch_all(&pool).await?;

一次 INSERT + UNNEST 批插入 N 行、用 RETURNING 拿回新 ID——相比"N 次单 INSERT"，网络往返从 N 次降到 1 次、SQL 只 Parse 一次、事务边界也只有一个；相比 INSERT ... VALUES (...), (...) 的展开写法，SQL 文本长度不随行数增长，所以 prepared statement 也能命中缓存（否则每种行数都是一条新 SQL）。具体快多少请自己压测。

4. 自定义类型 / Enum

#[derive(sqlx::Type)]
#[sqlx(type_name = "user_role", rename_all = "lowercase")]
enum UserRole { Admin, User, Guest }

Postgres CREATE TYPE user_role AS ENUM ('admin', 'user', 'guest') + Rust enum——类型安全贯穿到 DB schema。

5. LISTEN/NOTIFY 做消息总线

let mut listener = PgListener::connect_with(&pool).await?;
listener.listen_all(["channel_a", "channel_b"]).await?;

loop {
    let n = listener.recv().await?;
    match n.channel() {
        "channel_a" => { /* ... */ }
        "channel_b" => { /* ... */ }
        _ => {}
    }
}

Postgres 作为轻量消息队列——不需要 Kafka 的场景够用。

6. Advisory Lock

// 会话级 advisory lock 绑在连接上：加锁和解锁必须用同一条连接
let mut conn = pool.acquire().await?;
sqlx::query("SELECT pg_advisory_lock($1)").bind(1234_i64).execute(&mut *conn).await?;
// ... 独占临界区 ...
sqlx::query("SELECT pg_advisory_unlock($1)").bind(1234_i64).execute(&mut *conn).await?;

App 级分布式锁——跨 Rust 进程的互斥。注意别直接对 &pool 执行这两条：两次 execute 可能落在池里不同的连接上，解锁会失败（返回 false），锁则一直留在前一条连接上、随它回到池里。sqlx 的 PgAdvisoryLock 封装了这个。

这六条 Postgres 独特性是"选 Postgres 而不是 MySQL"的核心理由——也是 sqlx-postgres 花大力气支持的特性。用 sqlx-postgres 就能享受 Rust 类型安全 + Postgres 功能强——两全其美。

16.23 源码精读：wait_until_ready 的精妙

值得单独精读的一段代码是 PgConnection::wait_until_ready（sqlx-postgres/src/connection/mod.rs:85-99）：

    pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> {
        if !self.inner.stream.write_buffer_mut().is_empty() {
            self.inner.stream.flush().await?;
        }

        while self.inner.pending_ready_for_query_count > 0 {
            let message = self.inner.stream.recv().await?;

            if let BackendMessageFormat::ReadyForQuery = message.format {
                self.handle_ready_for_query(message)?;
            }
        }

        Ok(())
    }

三条关键决策：

1. 先 flush 再 recv——保证之前 write 的消息送出去才能收到响应。否则可能死锁（server 等消息、client 等响应）。

2. 只消耗 ReadyForQuery——循环里只处理 Z 消息，其他消息被 recv 内部过滤（NoticeResponse / NotificationResponse 等）或返回但被 while 循环忽略。

3. handle_ready_for_query 减 count + 更新 transaction_status：

fn handle_ready_for_query(&mut self, message: ReceivedMessage) -> Result<(), Error> {
    self.inner.pending_ready_for_query_count = self
        .inner
        .pending_ready_for_query_count
        .checked_sub(1)
        .ok_or_else(|| err_protocol!("received more ReadyForQuery messages than expected"))?;

    self.inner.transaction_status = message.decode::<ReadyForQuery>()?.transaction_status;
    Ok(())
}

注意这里是 checked_sub(1) 而不是 -= 1——收到比预期多的 ReadyForQuery 会变成一条明确的协议错误，而不是 usize 下溢 panic（release 模式下则会 wrap 成天文数字、把后续所有 wait_until_ready 卡死）。

这条方法是连接状态机的核心——每个 Z 消息触发一次、同步 client 和 server 的事务状态视图。

wait_until_ready 是每次 query 的入口——保证"开始新 query 前清理残留状态"。这条保守做法让 sqlx 的连接状态永远一致——不会出现"server 还在处理上条 query、client 就发下一条"的错误。

如果 sqlx 实现 pipelining、这条方法要重写——允许有 pending ReadyForQuery 的情况下发新 query。0.8.6 没做——保守、但正确。

16.24 一条协议级排查案例

下面是一个示意案例（按 0.8.6 源码推演，本机没有 Postgres，未实跑）：某个新接口一调用就报数据库语法错误（Error::Database，SQLSTATE 42601）。

排查路径：

Step 1：查错误上下文——错误来自 stream.recv()：它收到 ErrorResponse 就直接转成 Err 上抛。

Step 2：抓网络包（tcpdump + wireshark）——发现 client 只发了 Parse + Describe + Sync（prepare 那一批），server 回 ErrorResponse + ReadyForQuery；Bind / Execute 根本没发出去，因为 prepare 已经返回错误。

Step 3：检查代码——query("SELECT ... WHERE id = ?")——用了 ? 占位符（MySQL / SQLite 语法）而不是 $1（Postgres）。Parse 阶段 server 就判定 SQL 语法错。

Step 4：修复——改用 $1。

这条 bug 的根源在 SQL 文本本身——sqlx-postgres 不做占位符翻译（不会把 ? 改写成 $N），SQL 原样传给 server，所以错误要等到 Parse 时由 server 报出来；用 query! 宏的话，它会在编译期就暴露。协议知识的用处在于读懂抓包：看到只有 Parse 那一批、没有 Bind，就能断定问题出在 SQL 编译阶段而不是参数绑定。

类似的协议层排查在生产中不常见（大多数 bug 在业务层）——但遇到时你要能用 tcpdump + 协议知识追踪——本章提供的协议理解就是工具。

16.25 Postgres 协议的未来

协议这一层已经很稳，值得关注的反而是客户端怎么用它。一个具体的、已经落地的例子：pipelining——libpq 从 PostgreSQL 14 起提供了 pipeline mode（PQenterPipelineMode 等，见官方文档的 Pipeline Mode），说明协议层面早就允许"连发多条、后收响应"。sqlx 0.8.6 没有做（§16.7），sqlx-postgres/src/connection/executor.rs:271-275 那段注释也承认了这一点：

Theoretically, we could send dozens of queries before a [Sync] and postgres can handle that. Execution on the server is still serial but it would reduce round-trips. Some kind of builder pattern that is termed batching might suit this.

想跟进后续进展，就 watch launchbadge/sqlx 的 CHANGELOG 和 issue——本专栏只对 0.8.6 的现状负责，不预测路线图。

16.26 Postgres 驱动代码品质的七条观察

读完整个 sqlx-postgres 的源码，有七条代码品质观察可以总结：

1. 注释密度适中。协议相关代码（message / 握手）注释较多、引用协议文档链接；普通业务代码较少注释。这让"要理解的地方有解释、不需要的地方不冗余"。

2. 错误类型层次清晰。sqlx::Error::Database(Box<PgDatabaseError>)——顶层 Error enum + DB-specific detail。用户能按需匹配具体错误（比如 e.is_unique_violation()）也能当 generic 错误处理。

3. 内部 API 用 pub(crate)——不对外公开、但 driver 内各文件可用。严格区分内部细节和公开接口。

4. unsafe 极少。grep -rn unsafe sqlx-postgres/src/ 只命中 1 行，而且是 types/citext.rs 里的一行文档注释——实际代码零 unsafe，所有复杂度都在 safe Rust 里完成。复杂的 async 状态机靠类型系统 + RAII 保证、不依赖 unsafe hack。（对照 sqlx-sqlite：那边因为要过 C FFI，unsafe 是常态，见第 18 章 §18.5。）

5. 合理使用 #[cfg(feature = "...")]——chrono / uuid / time 等类型的支持通过 feature gate 可选加入。用户不用的 feature 不拉依赖。

6. 错误消息对生产友好。err_protocol!("PgConnection::run(): too many arguments for query: {}", args.len())——错误里带具体数字、方便排查。不是"something went wrong"这种无用错误。

7. 单元测试覆盖关键 helper。message encode / decode（message/ 下多数文件末尾都有 #[test] 字节断言）、MD5 密码编码、parse_server_version 都有 unit test；SCRAM 的 hi() 只有一个 bench。集成测试用真 Postgres 数据库。

这七条品质观察是 sqlx-postgres 能长期稳定演进的基础——代码质量是长期维护的基石。你自己写 Rust 库时这七条都值得遵守。

16.27 Postgres 驱动和业务代码的接口

本章末尾的一个哲学观察——sqlx-postgres 和业务代码的接口极简：

业务代码只看到 PgPool / PgConnection / PgRow / PgTypeInfo——几个顶层类型。
内部的 19841 行代码全部隐藏——业务开发者不需要知道 Parse / Bind / Execute 消息、不需要知道 pending_ready_for_query_count、不需要知道 SCRAM 挑战响应。

这种"复杂实现 + 简洁接口"就是好的抽象的标志——用户享受功能、不被细节淹没。sqlx-postgres 做到了这一点——即使内部结构复杂、对外使用非常轻松。

从 API 设计角度看、这条原则值得你在自己项目里应用——接口的简洁程度和实现的复杂度成反比才是好的抽象。简陋实现给简陋接口、复杂实现也给复杂接口——都不如 sqlx-postgres 这种"内部繁复、外部清爽"。

16.28 BackendMessageFormat：21 个变体撑起 sqlx 认得的全部响应

sqlx 认得 21 种后端消息（协议里还有 CopyBothResponse、FunctionCallResponse、NegotiateProtocolVersion 等，sqlx 不用，收到会报 unknown message type）——用一个 enum 承载（sqlx-postgres/src/message/mod.rs:92-114；对应的前端侧是 FrontendMessageFormat，13 个变体，:68-88）：

pub enum BackendMessageFormat {
    Authentication,
    BackendKeyData,
    BindComplete,
    CloseComplete,
    CommandComplete,
    CopyData,
    CopyDone,
    CopyInResponse,
    CopyOutResponse,
    DataRow,
    EmptyQueryResponse,
    ErrorResponse,
    NoData,
    NoticeResponse,
    NotificationResponse,
    ParameterDescription,
    ParameterStatus,
    ParseComplete,
    PortalSuspended,
    ReadyForQuery,
    RowDescription,
}

tag 字节与 enum 的映射（try_from_u8）—— tag 字节是 ASCII 字符：'R' → Authentication、'K' → BackendKeyData、'1' → ParseComplete、'2' → BindComplete、'D' → DataRow、'Z' → ReadyForQuery、'E' → ErrorResponse……

Postgres 协议选 ASCII 可见字符做 tag 的好处——wireshark 抓包直接读得懂——D 看就知道是 DataRow——不需要查表——诊断友好。

类型安全的 decode（ReceivedMessage::decode，:124-143）：

// sqlx-postgres/src/message/mod.rs:124
pub fn decode<T>(self) -> Result<T, Error>
where
    T: BackendMessage,
{
    if T::FORMAT != self.format {
        return Err(err_protocol!(
            "Postgres protocol error: expected {:?}, got {:?}",
            T::FORMAT,
            self.format
        ));
    }

    T::decode_body(self.contents).map_err(|e| match e {
        Error::Protocol(s) => {
            err_protocol!("Postgres protocol error (reading {:?}): {s}", self.format)
        }
        other => other,
    })
}

T::FORMAT 是每个消息类型的关联常量—— 编译期确定"我这个 struct 对应哪个 tag"—— decode 时 runtime 核对 tag 和类型是否匹配—— 不匹配直接 protocol error 而不是 panic—— 错误路径优雅。

这个 FORMAT 常量 + decode_body 方法 的 pattern 把 Rust 类型系统和 wire 协议串起来——每个 Rust struct 声明"我是 ParseComplete 消息"、FORMAT 常量固定、decode 逻辑独立。注意不是 21 个消息就有 21 个 struct：无载荷的消息（BindComplete、CloseComplete、NoData、PortalSuspended、EmptyQueryResponse）只在枚举里占一项，executor.rs 直接按 format 匹配、不需要 decode 结构体。

16.29 startup 握手的 SSL 协商

Postgres 连接建立时可选 SSL/TLS 加密——sqlx 的 establish.rs 有一个特殊的pre-startup 阶段：

客户端发 SslRequest（sqlx-postgres/src/message/ssl_request.rs）—— 一个 8 字节特殊包（长度 + magic number 80877103）。
服务器回单字节—— 'S'（支持 SSL）或 'N'（不支持）。
客户端按回答决定是否 upgrade 到 TLS。
之后才进入正常的 StartupMessage 发送。

为什么 SSL 协商走这种怪异单字节协议？

历史遗留—— SSL 支持是在已有的启动流程上后加的—— Startup 消息里没有预留 TLS negotiation 字段—— 只能在 Startup 的"协议版本号"位置放一个特殊值（高 16 位 1234、低 16 位 5679，官方文档特意说明它不能与任何真实协议版本号相同）—— 在 startup 前插入一个"你支持 TLS 吗"的询问。

sqlx-postgres 在代码里小心处理这个 pre-startup 阶段—— 如果服务器回 'S' 就用 rustls/native-tls 包装 socket、再发 StartupMessage—— 整个过程对上层（query API）完全透明。

MySQL 的做法不同—— 服务端先发 Handshake V10、在 capabilities 位图里声明支持 SSL，客户端回一个精简版的握手响应（SSLRequest 包：capabilities、最大包长、字符集，不含用户名密码）就直接开始 TLS 握手（sqlx-mysql/src/connection/tls.rs）—— 协商嵌在握手包里、不需要额外的"魔数"消息。两种设计反映了两家 DB 的协议演进历史。

16.30 Parse 消息的源码结构：body_size_hint 的意义

Parse 消息的 Rust 实现（sqlx-postgres/src/message/parse.rs:8-20）：

pub struct Parse<'a> {
    pub statement: StatementId,
    pub query: &'a str,
    pub param_types: &'a [Oid],
}

三个字段——statement name、SQL string、参数类型 OID 数组——对应 Postgres 协议 Parse 消息的三段 body。

FrontendMessage trait 实现（:22-63）—— 除了 const FORMAT: FrontendMessageFormat 外、有一个关键方法 body_size_hint(&self) -> Saturating<usize>：

// sqlx-postgres/src/message/parse.rs:25
fn body_size_hint(&self) -> Saturating<usize> {
    let mut size = Saturating(0);
    size += self.statement.name_len();
    size += self.query.len();
    size += 1; // NUL terminator
    size += 2; // param_types_len
    size += self.param_types.len().saturating_mul(4);
    size
}

它管什么——Postgres 协议每个 message 以 tag(1) + length(4) + body 开头—— length 是长度字段自身 4 字节 + body 字节数——sqlx 用 body_size_hint 预估 body 大小、预分配 buf，避免编码过程中 Vec 反复扩容；length 本身仍是"先留坑、写完 body 再回填"。

Saturating<usize> 是标准库的饱和整数—— 加法到 usize::MAX 就停住、不会溢出 panic。这只是个"提示"、算歪了也不影响正确性：真正的长度是 put_length_prefixed 写完 body 后回填的，body_size_hint 只用来给 EncodeMessage 做一次 try_reserve（见 §16.9）。

参数类型 OID 的 u16 限制（:46-53）：

let param_types_len = u16::try_from(self.param_types.len()).map_err(|_| {
    err_protocol!("param_types.len() too large for binary protocol: ...")
})?;

Postgres 协议限制单 query 最多 65535 个参数—— sqlx 在 encode 前校验、超限直接 protocol error——这条显式限制 是生产代码的老练之处：不要等服务器回 ErrorResponse（错误信息可能含糊）、自己先查清楚并给出明确错误。

这个 body_size_hint + encode_body 两段式 encode 是 sqlx-postgres 所有 FrontendMessage 的统一模式——Parse / Bind / Execute / Sync / Close / Describe 都遵循（grep -c "impl FrontendMessage for" sqlx-postgres/src/message/ 一共 14 处）—— 代码一致性让维护者从单个消息的代码 pattern 推断其他消息。

16.31 Execute 消息的 limit 字段：默认为 0 的深意

源码（sqlx-postgres/src/message/execute.rs:8-15）：

pub struct Execute {
    pub portal: PortalId,
    pub limit: u32,
}

仅 两字段—— portal name + 整数 limit。但 limit 字段的文档注释（:12-13）值得注意：

Maximum number of rows to return, if portal contains a query that returns rows (ignored otherwise). Zero denotes "no limit".

Postgres 协议允许 Execute 带行数限制—— 客户端可以执行一个 portal、只拿前 N 行—— N 行后portal 暂停、下一次 Execute(same portal) 继续—— 类似数据库 server-side cursor。

但 sqlx 永远用 limit = 0（sqlx-postgres/src/connection/executor.rs:251-257，注释里给了 issue 链接）—— 为什么？