Neon WAL 处理全链路解析：从 Compute 到 Pageserver 的深度解码与存储
摘要：本文深入剖析 Neon 数据库架构中 WAL（Write-Ahead Log）从 Compute 节点产生，经 Safekeeper 接收、解码、转发，最终被 Pageserver 摄入并存储的完整流程。重点解析 Safekeeper 的"流解码"与"深度解码"两层机制，以及 Pageserver 如何利用 InterpretedWalRecord 实现计算存储分离，避免传统 WAL Redo 的架构设计。
阶段一：Compute → Safekeeper（WAL Push）
Compute 节点 (PostgreSQL)
│ TCP连接, 发送 START_WAL_PUSH 命令
▼
1. 网络接入层
wal_service.rs:61handle_socket()→ 创建SafekeeperPostgresHandler，启动 PG 协议主循环
handler.rs:306process_query()→ 匹配START_WAL_PUSH→ 调用handle_start_wal_push()
2. WAL 流读取
receive_wal.rs:191handle_start_wal_push_guts()：
- 建立 mpsc channel（
msg_tx/msg_rx），将网络读取和 WAL 处理解耦 - 启动
NetworkReader（读网络消息 → 发到 channel） - 启动
WalAcceptor::run()（从 channel 收消息 → 处理）
receive_wal.rs:412read_network_loop()→ 循环读取CopyData消息 →ProposerAcceptorMessage::parse()反序列化 →msg_tx.send()
3. WAL 写入磁盘 + 流解码
receive_wal.rs:600WalAcceptor::run()→ 收到AppendRequest→tli.process_msg(NoFlushAppendRequest)
timeline.rs:1106→safekeeper.rs:937SafeKeeper::process_msg()→safekeeper.rs:1293handle_append_request()
wal_storage.rs:437PhysicalStorage::write_wal()—— 这是第一个关键点：
self.write_exact(startpos, buf)?; // ① 写入磁盘 self.decoder.feed_bytes(buf); // ② 喂入解码器 while let Some((lsn, _rec)) = self.decoder.poll_decode()? { // ③ 找记录边界 self.write_record_lsn = lsn; }这里只做WAL page header 校验 + 记录边界识别 + CRC 校验，不做内容解析。对应：
lib.rs:409feed_bytes()→ 追加字节到内部 buffer
lib.rs:413poll_decode()→ 版本分派到waldecoder_handler.rs:100poll_decode_internal()：
- 验证 page header（magic、LSN、continuation record 标志）
- 状态机：
WaitingForRecord→ 读xl_tot_len→ 跨页重组 →ReassemblingRecord complete_record()→ CRC 校验 → 处理XLOG_SWITCH
此时 safekeeper 得到的是原始 WAL 字节（Bytes），只确认了记录边界，未解析内容。
阶段二：Safekeeper → Pageserver（WAL 转发 + 深度解码）
Pageserver (WAL consumer)
│ TCP连接, 发送 START_REPLICATION
▼
4. Safekeeper 侧：解读 WAL 并发送
send_wal.rs:443handle_start_replication()→send_wal.rs:559解读模式：
// send_wal.rs:601 let reader_handle = InterpretedWalReader::spawn( wal_stream, start_pos, tx, shard, pg_version, ... ); // send_wal.rs:634 let sender_handle = InterpretedWalSender::spawn(rx, pgb_writer, ...);send_interpreted_wal.rs:254InterpretedWalReader::spawn()→send_interpreted_wal.rs:393run_impl()：
let mut wal_decoder = WalStreamDecoder::new(start_pos, self.pg_version); loop { let wal = self.wal_stream.read().await?; // 从磁盘读 WAL wal_decoder.feed_bytes(&wal); // 喂入流解码器 while let Ok(Some((lsn, recdata))) = wal_decoder.poll_decode() { // ← 你选中的行 // 这里是深度解码！ let records = InterpretedWalRecord::from_bytes_filtered( recdata, &shard_ids, next_record_lsn, self.pg_version )?; tx.send(Batch { records, ... }).await?; // 发送给 sender } }5. 深度解码：decode_wal_record()（核心！）
decoder.rs:23InterpretedWalRecord::from_bytes_filtered()：
// ① 调用 postgres_ffi 的 decode_wal_record 解析 WAL 记录内容 decode_wal_record(buf, &mut decoded, pg_version)?; // ② 按 shard 过滤并提取元数据 + 页面数据 MetadataRecord::from_decoded_filtered(&decoded, pg_version)? // 元数据(事务/DDL等) SerializedValueBatch::from_decoded_filtered(&decoded, ...) // 页面KV数据walrecord.rs:246decode_wal_record()—— 深度解析：
- 解析 XLogRecord header（rmid, info, xid）
- 遍历 block headers：提取 RelFileNode、forknum、blkno、full-page image、压缩信息
- 处理
XLR_BLOCK_ID_ORIGIN、XLR_BLOCK_ID_TOPLEVEL_XID等特殊标志
decoder.rs:77MetadataRecord::from_decoded_filtered()→ 按xl_rmid分发：
| rmgr ID | 处理函数 | 产物 |
|---|---|---|
RM_XACT_ID | decode_xact_record() | XACT commit/abort 信息 |
RM_HEAP_ID | decode_heapam_record() | VM bit 清除 |
RM_DBASE_ID | decode_dbase_record() | DB create/drop |
RM_SMGR_ID | decode_smgr_record() | SMGR create/truncate |
RM_CLOG_ID | decode_clog_record() | CLOG zero/truncate |
RM_LOGICALMSG_ID | decode_logical_message_record() | LogicalMessage(Put/Failpoint) |
6. 网络发送给 Pageserver
send_interpreted_wal.rs:650InterpretedWalSender::run()→ 从 channel 收Batch→ 序列化为InterpretedWalRecordswire format → 通过 PG 协议发送
阶段三：Pageserver 接收 + 存储
Safekeeper 发送 InterpretedWalRecords (wire format)
│ TCP
▼
7. Pageserver 接收 WAL
walreceiver_connection.rs:116handle_walreceiver_connection()：
- 连接 safekeeper（物理复制协议）
IDENTIFY_SYSTEM获取 WAL 位点START_REPLICATION PHYSICAL <startpoint>开始流复制- 创建
WalIngest实例
walreceiver_connection.rs:337接收循环：
ReplicationMessage::RawInterpretedWalRecords(raw) => { let batch = InterpretedWalRecords::from_wire(raw.data(), format, compression)?; for record in batch.records { walingest.ingest_record(interpreted, &mut modification, &ctx)?; } modification.commit(&ctx).await?; }8. WAL 摄入引擎
walingest.rs:233WalIngest::ingest_record()—— 中央分发：
ingest_record() ├── MetadataRecord 分发 │ ├── Xact(commit/abort) → ingest_xact_record() → 写 CLOG、删关系 │ ├── Dbase(create) → ingest_xlog_dbase_create() → 复制源库关系 │ ├── Dbase(drop) → ingest_xlog_dbase_drop() → 删除库关系 │ ├── Smgr(create) → ingest_xlog_smgr_create() → 创建关系元数据 │ ├── Smgr(truncate) → ingest_xlog_smgr_truncate() │ ├── Clog(truncate) → ingest_clog_truncate() │ ├── Heap(ClearVM) → ingest_clear_vm_bits() → WAL record 写入 │ ├── LogicalMessage(put) → ingest_logical_message_put() │ └── ... │ ├── modification.ingest_batch(interpreted.batch) → 页面数据累积 │ └── [pgdatadir_mapping.rs:1929] → pending_data_batch 累积 KV │ └── modification.put_checkpoint() → 更新 checkpoint (nextXid等)9. 提交到存储层
pgdatadir_mapping.rs:2867DatadirModification::commit()：
let mut writer = self.tline.writer().await; // 获取 TimelineWriter writer.put_batch(batch, ctx)?; // 写入 InMemoryLayer writer.finish_write(pending_lsn)?; // 推进 last_record_lsntimeline.rs:7862→inmemory_layer.rs:571InMemoryLayer::put_batch()：
file.write_raw(raw)→ 追加序列化字节到文件index.write()→ 更新内存 BTree 索引（key → LSN/offset/length）
完整调用栈总览
┌─────────────────────────────────────────────────────────────────────────┐ │ Compute 节点 │ │ PostgreSQL WAL writer → TCP → START_WAL_PUSH │ └────────────────────────────┬────────────────────────────────────────────┘ ▼ ┌─ Safekeeper ───────────────────────────────────────────────────────────┐ │ wal_service.rs:61 handle_socket() TCP accept │ │ handler.rs:306 process_query() PG协议分发 │ │ receive_wal.rs:263 read_first_message() 读首条消息 │ │ receive_wal.rs:412 read_network_loop() CopyData循环读取 │ │ receive_wal.rs:600 WalAcceptor::run() 消息处理循环 │ │ safekeeper.rs:1293 handle_append_request() WAL追加处理 │ │ wal_storage.rs:437 write_wal() ─────────────────────┐ │ │ ├── write_exact() 写入磁盘 │ │ │ ├── feed_bytes() 喂入WalStreamDecoder │ ← 阶段一 │ │ └── poll_decode() 找记录边界+CRC校验 ← 你选中的行 │ (写入) │ │ (waldecoder_handler.rs:100) │ │ │ │ │ │ ─────────────── safekeeper 两条路径的分界线 ──────────── │ │ │ │ │ │ send_wal.rs:443 handle_start_replication() │ │ │ send_interpreted_wal:254 InterpretedWalReader::spawn()│ │ │ send_interpreted_wal:393 run_impl() ─────────────────────┐ │ │ │ ├── feed_bytes() 喂入WalStreamDecoder │ │ │ │ ├── poll_decode() 找记录边界 ← 又是你选中的行 │ │ ← 阶段二 │ │ ├── from_bytes_filtered() ★ 深度解码入口 │ │ (转发) │ │ │ ├── decode_wal_record() 解析WAL内容(walrecord:246)│ │ │ │ │ ├── MetadataRecord解码 (decoder.rs:77) │ │ │ │ │ └── SerializedValueBatch 页面数据序列化 │ │ │ │ └── tx.send(Batch) 发送给InterpretedWalSender │ │ │ │ │ │ │ send_interpreted_wal:650 InterpretedWalSender::run()│ │ │ └── BeMessage::InterpretedWalRecords 序列化+网络发送│ │ └────────────────────────────┬────────────────────────────────────────────┘ │ TCP (InterpretedWalRecords wire format) ▼ ┌─ Pageserver ───────────────────────────────────────────────────────────┐ │ walreceiver_connection:116 handle_walreceiver_connection() 连接+握手 │ │ walreceiver_connection:346 from_wire() 反序列化 │ │ walreceiver_connection:446 ingest_record() ───────────────┐ │ │ │ │ │ walingest.rs:233 WalIngest::ingest_record() 中央分发 │ │ │ ├── ingest_xact_record() → CLOG写入 + 关系删除 │ │ │ ├── ingest_xlog_dbase_create() → 复制数据库关系 │ ← 阶段三 │ │ ├── ingest_xlog_smgr_create() → 创建关系元数据 │ (存储) │ │ ├── ingest_clear_vm_bits() → VM页面WAL记录 │
