Neon wal日志处理流程

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分发：

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_lsn

timeline.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记录 │