在資料同步、CDC（Change Data Capture）、事件驅動系統中，「如何即時取得 PostgreSQL 的資料變更」是一個非常核心的問題。pg-walstream 是一個使用 Rust 實作的 PostgreSQL WAL（Write-Ahead Log）Streaming 函式庫，透過 PostgreSQL Logical / Physical Replication Protocol，讓開發者可以直接串流並解析資料庫變更事件，作為自訂 CDC 或資料同步系統的底層元件。

專案背景與用途

PostgreSQL 提供 Logical Replication 機制，能將 WAL 中的資料變更以結構化事件的方式輸出，pg-walstream 的目標是：

• 提供 Rust 原生 的 WAL streaming library
• 直接實作 PostgreSQL Logical / Physical Replication protocol
• 可嵌入到任何 async Rust 應用中
• 作為 CDC、資料同步、事件系統的基礎 building block

適合以下場景：

• 自建 CDC pipeline（不使用 Debezium）
• PostgreSQL → Kafka / Redis / EventBus
• 即時資料同步、資料湖（data lake）ingestion
• 高效能、低延遲的資料變更監聽服務
• 建立 physical standby、Point-in-Time Recovery（PITR）、災難復原用的 base backup

核心特色

pg-walstream 專注於「穩定、高效、可控」，v0.8.0 的完整特色如下：

• 完整 Logical Replication 支援：實作 PostgreSQL logical replication protocol v1 ~ v4
• Physical Replication 支援：可串流原始 WAL，用於 standby 與 PITR
• Base Backup 支援：完整的 BASE_BACKUP 指令，含 progress、compression、manifest 選項
• 純 Rust 後端（預設）：rustls-tls 不需 libpq、不需 OpenSSL，使用 aws-lc-rs 做硬體加速 TLS
• TLS/SSL 支援：所有 PostgreSQL SSL 模式（disable、allow、prefer、require、verify-ca、verify-full）
• 認證方式：Cleartext、MD5、SCRAM-SHA-256
• Streaming Transactions：支援串流大型交易（protocol v2+）
• Two-Phase Commit：prepared transaction 支援（protocol v3+）
• Parallel Streaming：多串流平行複寫（protocol v4+）
• Zero-Copy 操作：基於 bytes crate、搭配 drain-loop 批次佇列最佳化的高效 buffer 管理
• 執行緒安全的 LSN 追蹤：以 atomic 進行 LSN feedback，適合 producer-consumer 模式
• 連線管理：內建連線處理與 exponential backoff 重試邏輯
• 型別安全 API：強型別訊息解析與完整錯誤處理
• Typed Row Deserialization：內建 serde deserializer，可將 WAL 列直接映射進自訂 struct（數值、bool、String、Option<T>、enum、bytes）
• 高階消費體驗：ReplicationStreamConfig::builder()、會自動 ack 的 EventStream::for_each_event，以及 typed 的 by-table WalRouter（可搭配 opt-in 的 #[derive(WalTable)]）
• Bounded Replay：with_stop_at_lsn 串流到目標 LSN 後乾淨結束
• Raw XLogData 存取：next_raw_event 回傳未解碼的 pgoutput payload 加上 WAL 位置
• Replication Slot 管理：建立、修改、讀取、刪除 slot，並支援完整選項
• Hot Standby Feedback：可為 physical replication 送出 hot standby feedback

架構概念簡介

pg-walstream 並不是一個「完整 CDC 平台」，而是專注於 WAL streaming 的核心邏輯：

我另一個專案 pg2any 專注完成 CDC 平台，也是使用這個當作 replication protocol library。

Architecture

v0.8.0 的分層架構，特別的是多了一層 PgReplicationConnection——用來抽象化 libpq 與 rustls-tls 兩種後端，並在編譯期依 feature flag 選擇：

┌──────────────────────────────────────────┐
│          Application Layer               │
│  (Your CDC / Replication Logic)          │
└──────────────┬───────────────────────────┘
               │
┌──────────────▼───────────────────────────┐
│    LogicalReplicationStream              │
│  - Connection management & retry         │
│  - Event processing & LSN feedback       │
│  - Snapshot export support               │
└──────────────┬───────────────────────────┘
               │
┌──────────────▼───────────────────────────┐
│  LogicalReplicationParser                │
│  - Protocol v1-v4 parsing                │
│  - Zero-copy message deserialization     │
│  - Streaming transaction support         │
└──────────────┬───────────────────────────┘
               │
┌──────────────▼───────────────────────────┐
│     PgReplicationConnection              │
│  ┌─────────────────┬──────────────────┐  │
│  │  libpq backend  │ rustls-tls       │  │
│  │  (C FFI)        │ (pure Rust)      │  │
│  │                 │                  │  │
│  │  pq-sys         │ rustls +         │  │
│  │                 │ aws-lc-rs +      │  │
│  │                 │ postgres-protocol│  │
│  └─────────────────┴──────────────────┘  │
│  Compile-time feature flag selection     │
└──────────────┬───────────────────────────┘
               │
┌──────────────▼───────────────────────────┐
│     BufferReader / BufferWriter          │
│  - Zero-copy operations (bytes crate)    │
│  - Binary protocol handling              │
│  - Drain-loop batch queue optimization   │
└──────────────────────────────────────────┘

你可以在「Application Layer」這一層：

• 將事件轉成 JSON
• 發送到 Kafka / Pulsar
• 同步到另一個資料庫
• 寫入 data lake

安裝方式

在 Cargo.toml 中加入：

[dependencies]
pg_walstream = "0.8"

這裡有個重要改動要提醒：預設情況下，pg-walstream 用的是純 Rust 的 rustls-tls 後端，不需要 libpq、也不需要 OpenSSL，只要 build 時有 cmake 跟 C compiler（給 aws-lc-rs 用）就好。這跟早期版本「一定要先裝 libpq-dev」的情況已經不一樣了，如果你照舊教學裝了一堆系統依賴，其實現在多半用不到。

如果你想改用 C 的 libpq 後端（透過 pq-sys 綁定，需要系統的 libpq）：

[dependencies]
pg_walstream = { version = "0.8", default-features = false, features = ["libpq"] }

若兩個後端同時啟用，rustls-tls 會自動優先。

Feature Flags

pg-walstream 提供兩種連線後端，外加一個 std 開關，全部都是編譯期決定。rustls-tls 是預設，libpq 為 opt-in；兩者同時啟用時 rustls-tls 優先：

註：protocol parser、encoder 與 types 本身不需要任何後端。以 default-features = false 建置可得到只有這些部分的 no_std + alloc 版本，適用 wasm 與嵌入式。只有在需要即時串流與連線 API 時，才需要連線後端（libpq 或 rustls-tls），此時會拉入 std。

系統依賴

系統層級的依賴只有在 opt-in 的 libpq 功能時才需要。預設的 rustls-tls 後端在 build 時只需要 cmake 與 C compiler（給 aws-lc-rs 用），runtime 沒有任何依賴。

rustls-tls（預設）

build 時需要 cmake 與 C compiler，供 aws-lc-rs（硬體加速密碼學）使用：

# Ubuntu / Debian
sudo apt-get install cmake gcc

TLS 信任根（trust store）

當 sslmode 為 verify-ca 或 verify-full 時，rustls-tls 後端會這樣建立根憑證庫：

1. 若有設定 sslrootcert，則只從該 PEM 檔載入這些 CA（排他）。
2. 否則載入透過 webpki-roots 內建的 Mozilla CA bundle。

注意：不會查詢作業系統的 trust store。如果你的 PostgreSQL server 是由只存在於 OS trust store（例如 /etc/ssl/certs）的企業/內部 CA 簽發，就必須明確把 sslrootcert 指向該 CA，例如：

postgresql://user:pass@host/db?sslmode=verify-full&sslrootcert=/etc/ssl/certs/corporate-ca.pem

libpq（opt-in）

# Ubuntu / Debian
sudo apt-get install libpq-dev libssl-dev

# RHEL / CentOS / Fedora
sudo dnf install postgresql-devel

PostgreSQL 前置設定

在使用這個函式庫之前，你需要為複寫設定 PostgreSQL。

1. 設定 PostgreSQL

編輯 postgresql.conf：

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

修改後需重啟 PostgreSQL。

2. 建立 Publication

-- 針對特定資料表建立 publication
CREATE PUBLICATION my_publication FOR TABLE users, orders;

-- 或發布所有資料表
CREATE PUBLICATION my_publication FOR ALL TABLES;

3. 建立 Replication User

-- 建立具備 replication 權限的使用者
CREATE USER replication_user WITH REPLICATION PASSWORD 'secure_password';

-- 授予必要權限
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user;
GRANT USAGE ON SCHEMA public TO replication_user;

4. Replication Slot 選項

pg-walstream 對 replication slot 的建立提供完整控制，並會依連線的 PostgreSQL 版本自動選擇正確的 SQL 語法：

• PG14：舊式的位置關鍵字語法（EXPORT_SNAPSHOT、NOEXPORT_SNAPSHOT、USE_SNAPSHOT、TWO_PHASE、RESERVE_WAL）
• PG15+：新式的括號選項語法（(SNAPSHOT 'export', TWO_PHASE true, ...)）

註：若同時設定 two_phase 與 snapshot，以 two_phase 優先。failover 選項在 PG14 不可用，會回傳錯誤。

使用範例

以下範例皆對齊 v0.8.0 的現行 API。更多完整、可執行的範例可參考 examples/ 目錄。

範例一：基礎 WAL Streaming Client（futures::Stream）

以下是一個最小可運作的 async 範例，示範如何使用 pg-walstream 連線並接收 WAL 事件。它把 EventStream 用 futures::stream::unfold 包成標準的 futures::Stream，這樣就能使用 stream combinators（filter、take_while…）。

相較舊版，這裡有兩個關鍵改變：ReplicationStreamConfig::new 現在多了一個 StreamingMode 參數（取代舊版的 bool），而取事件的方法是 next_event()（舊版是 next()）。

use futures::stream::{self, StreamExt};
use pg_walstream::{
    CancellationToken, LogicalReplicationStream, ReplicationStreamConfig, RetryConfig,
    StreamingMode,
};
use std::env;
use std::time::Duration;
use tracing::{error, info, Level};
use tracing_subscriber;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化 logging
    tracing_subscriber::fmt()
        .with_max_level(Level::INFO)
        .with_target(false)
        .init();

    info!("Starting PostgreSQL WAL streaming example");

    // 從環境變數取得連線字串，否則使用預設值
    let connection_string = env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://postgres:password@localhost:5432/postgres?replication=database".to_string()
    });

    // 設定 replication stream
    let config = ReplicationStreamConfig::new(
        "example_slot".to_string(),   // Replication slot 名稱
        "my_publication".to_string(), // Publication 名稱（必須先存在）
        2,                            // Protocol 版本（2 支援 streaming）
        StreamingMode::On,            // Streaming 模式
        Duration::from_secs(10),      // 每 10 秒送一次 feedback
        Duration::from_secs(30),      // 連線逾時
        Duration::from_secs(60),      // 健康檢查間隔
        RetryConfig::default(),       // 使用預設重試策略
    );

    // 建立並初始化 stream
    let mut stream = LogicalReplicationStream::new(&connection_string, config).await?;

    // 從最新位置開始複寫（None = latest）
    stream.start(None).await?;

    info!("Replication started successfully");
    info!("Listening for changes... (Press Ctrl+C to stop)");

    // 建立 cancellation token 以便優雅關閉
    let cancel_token = CancellationToken::new();
    let cancel_token_clone = cancel_token.clone();

    // 設定 Ctrl+C handler
    tokio::spawn(async move {
        tokio::signal::ctrl_c()
            .await
            .expect("Failed to listen for ctrl-c");
        info!("Received shutdown signal, cleaning up...");
        cancel_token_clone.cancel();
    });

    // 轉換為 EventStream
    let event_stream = stream.into_stream(cancel_token);

    // 用 futures::stream::unfold 包成標準的 futures::Stream
    // 這樣就能使用 stream combinators！
    // 用 Box::pin 把 stream 釘在 heap 上，方便重複使用
    let mut pg_stream = Box::pin(stream::unfold(
        event_stream,
        |mut event_stream| async move {
            match event_stream.next_event().await {
                Ok(event) => {
                    // 成功取得事件後更新 applied LSN
                    event_stream.update_applied_lsn(event.lsn.value());
                    Some((Ok(event), event_stream))
                }
                Err(e) => {
                    // 回傳錯誤並停止 stream
                    Some((Err(e), event_stream))
                }
            }
        },
    ));

    // 現在可以使用 stream combinators 了！
    while let Some(result) = pg_stream.as_mut().next().await {
        match result {
            Ok(event) => {
                info!("Received event: {:?}", event);
            }
            Err(e) => {
                error!("Error: {}", e);
                break;
            }
        }
    }

    info!("Graceful shutdown complete");
    Ok(())
}

範例二：用 builder 精簡設定

ReplicationStreamConfig::new(...) 參數很多；v0.8.0 提供 builder()，只需給 slot 與 publication 名稱，其餘走預設值，之後再視需要鏈式覆寫：

use pg_walstream::{CancellationToken, LogicalReplicationStream, ReplicationStreamConfig};
use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let connection_string = env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://postgres:password@localhost:5432/postgres?replication=database".to_string()
    });

    // 最精簡的設定：只給 slot 與 publication 名稱
    let config = ReplicationStreamConfig::builder("raw_slot", "my_publication");

    let mut stream = LogicalReplicationStream::new(&connection_string, config).await?;
    stream.start(None).await?;

    let cancel = CancellationToken::new();
    // ... 之後便可依前述方式消費事件
    let _ = cancel;
    Ok(())
}

若需要「有界重播」，可用 ReplicationStreamConfig::with_stop_at_lsn(...)：它會串流到目標 LSN、完整送出跨越邊界的那筆交易後，以 ReplicationError::StreamStopped 乾淨結束——很適合做「補資料到某個時間點」的一次性任務。

範例三：傳統 polling loop（含自動重試）

如果你不想用 futures::Stream，而想要對 loop 有更多掌控，或要整合到不適合 async stream 的系統，可以用傳統 polling。next_event_with_retry 會在連線出問題時自動重試：

use pg_walstream::{
    LogicalReplicationStream, ReplicationStreamConfig, RetryConfig, StreamingMode,
    CancellationToken,
};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = ReplicationStreamConfig::new(
        "my_slot".to_string(),
        "my_publication".to_string(),
        2, StreamingMode::On,
        Duration::from_secs(10),
        Duration::from_secs(30),
        Duration::from_secs(60),
        RetryConfig::default(),
    );

    let mut stream = LogicalReplicationStream::new(
        "postgresql://postgres:password@localhost:5432/mydb?replication=database",
        config,
    ).await?;

    stream.start(None).await?;

    let cancel_token = CancellationToken::new();

    // 傳統 polling loop，含自動重試
    loop {
        match stream.next_event_with_retry(&cancel_token).await {
            Ok(event) => {
                println!("Received event: {:?}", event);
                stream.shared_lsn_feedback.update_applied_lsn(event.lsn.value());
            }
            Err(e) if matches!(e, pg_walstream::ReplicationError::Cancelled(_)) => {
                println!("Cancelled, shutting down gracefully");
                break;
            }
            Err(e) => {
                eprintln!("Error: {}", e);
                break;
            }
        }
    }

    Ok(())
}

若你要更底層、不做自動重試的 polling，也可以直接用 stream.next_event(&cancel_token)。

範例四：Typed 反序列化 + WalRouter

這是 v0.8.0 最實用的新體驗之一。你只要在自訂 struct 上 #[derive(Deserialize)]，就能用 WalRouter 註冊「某張表的 INSERT/UPDATE/DELETE」對應的 typed handler，函式庫會自動把 WAL 列反序列化成你的型別。

use pg_walstream::{
    CancellationToken, LogicalReplicationStream, ReplicationSlotOptions, ReplicationStreamConfig,
    RetryConfig, StreamingMode, WalRouter,
};
use serde::Deserialize;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

// 1. 定義你的 model struct —— 只要 derive Deserialize 即可
//    對應資料表：
//    CREATE TABLE typed_deser_users (
//        id       BIGSERIAL PRIMARY KEY,
//        username VARCHAR(50) NOT NULL,
//        email    TEXT,
//        score    DOUBLE PRECISION NOT NULL DEFAULT 0.0,
//        active   BOOLEAN NOT NULL DEFAULT true
//    );
#[derive(Debug, Deserialize)]
struct User {
    id: i64,
    username: String,
    email: Option<String>, // 可為 NULL 的欄位 → Option
    score: f64,
    active: bool,
}

const TABLE_NAME: &str = "typed_deser_users";
const PUBLICATION: &str = "typed_deser_pub";
const SLOT_NAME: &str = "typed_deser_slot";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 使用 rustls-tls 後端時，需安裝預設的 crypto provider
    #[cfg(feature = "rustls-tls")]
    {
        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
    }

    let conn = std::env::var("DATABASE_URL")?;

    // 2. 啟動 WAL replication stream（這裡用 protocol v4 + 平行 streaming）
    let stream_config = ReplicationStreamConfig::new(
        SLOT_NAME.to_string(),
        PUBLICATION.to_string(),
        4,
        StreamingMode::Parallel,
        Duration::from_secs(5),
        Duration::from_secs(30),
        Duration::from_secs(60),
        RetryConfig::default(),
    )
    .with_slot_options(ReplicationSlotOptions {
        temporary: true,
        ..Default::default()
    });

    let mut stream = LogicalReplicationStream::new(&conn, stream_config).await?;
    stream.start(None).await?;

    let cancel_token = CancellationToken::new();
    let mut event_stream = stream.into_stream(cancel_token.clone());

    // 3. 用 WalRouter 註冊「每張表 → typed handler」
    let count = Arc::new(AtomicU64::new(0));
    let mut router = WalRouter::new();

    {
        let c = count.clone();
        router.on_insert::<User, _, _>(TABLE_NAME, move |u| {
            let c = c.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                println!("[INSERT] {:?}", u);
                Ok(())
            }
        });
    }
    {
        let c = count.clone();
        // 注意：old 是 Option<User>——舊列只有在資料表設定
        // REPLICA IDENTITY FULL 時才會出現，否則為 None
        router.on_update::<User, _, _>(TABLE_NAME, move |old, new| {
            let c = c.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                println!("[UPDATE] old={:?} => new={:?}", old.as_ref(), new);
                Ok(())
            }
        });
    }
    {
        let c = count.clone();
        router.on_delete::<User, _, _>(TABLE_NAME, move |u| {
            let c = c.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                println!("[DELETE] {:?}", u);
                Ok(())
            }
        });
    }

    // 4. 執行 router，直到 stream 結束或被取消
    router.run(&mut event_stream).await?;

    let _ = event_stream.shutdown().await;
    println!("Total DML events deserialized: {}", count.load(Ordering::SeqCst));

    Ok(())
}

範例五：Derive Router（#[wal_table("...")] + 型別推導）

啟用 opt-in 的 derive 功能後，可以在 struct 上加 #[wal_table("...")]，把型別綁定到資料表。這樣註冊 handler 時就不用再重複傳表名，直接用 on_insert_of::<T>() 這類型別推導寫法：

use pg_walstream::{
    CancellationToken, LogicalReplicationStream, ReplicationStreamConfig,
    RetryConfig, StreamingMode, WalRouter, wal_table,
};
use serde::Deserialize;

// 綁定型別到資料表——注意 #[wal_table(...)] 要放在最外層
#[wal_table("typed_deser_users")]
#[derive(Debug, Deserialize)]
struct User {
    id: i64,
    username: String,
    email: Option<String>,
    score: f64,
    active: bool,
}

async fn run(mut event_stream: pg_walstream::EventStream) -> Result<(), Box<dyn std::error::Error>> {
    let mut router = WalRouter::new();

    // 不需要再傳表名——型別本身已知道自己對應哪張表
    router.on_insert_of::<User, _>(|u| async move {
        println!("[INSERT] {:?}", u);
        Ok(())
    });
    // old 是 Option<User>（只有 REPLICA IDENTITY FULL 時才有舊列）
    router.on_update_of::<User, _>(|old, new| async move {
        println!("[UPDATE] {:?} => {:?}", old.as_ref(), new);
        Ok(())
    });
    router.on_delete_of::<User, _>(|u| async move {
        println!("[DELETE] {:?}", u);
        Ok(())
    });

    router.run(&mut event_stream).await?;
    let _ = event_stream.shutdown().await;
    Ok(())
}

對應的 Cargo.toml：pg_walstream = { version = "0.8", features = ["derive"] }。#[wal_table("users")] 是單行寫法，等同於 #[derive(WalTable)] #[wal(table = "users")]。

範例六：交易安全的消費者（commit 後才回報 LSN）

在正式的 CDC 場景裡，最重要的一件事是：只有在整筆交易被下游成功套用之後，才把 LSN 回報給 PostgreSQL。否則一旦中途崩潰，就可能遺失資料。下面示範以交易邊界（BEGIN / COMMIT）緩衝變更、依序套用，套用成功後才推進 feedback LSN：

use pg_walstream::{
    CancellationToken, ChangeEvent, EventType, LogicalReplicationStream, Lsn,
    ReplicationStreamConfig, RetryConfig, SharedLsnFeedback, StreamingMode,
};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

/// 緩衝中的交易
struct TxBuffer {
    xid: u32,
    changes: Vec<ChangeEvent>,
    commit_lsn: Option<Lsn>,
    applied: bool,
}

struct SafeConsumer {
    feedback: Arc<SharedLsnFeedback>,
    active: BTreeMap<u32, TxBuffer>,
    committed: Vec<TxBuffer>,
}

impl SafeConsumer {
    async fn process(&mut self, event: ChangeEvent) {
        match &event.event_type {
            EventType::Begin { transaction_id, .. } => {
                self.active.insert(*transaction_id, TxBuffer {
                    xid: *transaction_id,
                    changes: vec![],
                    commit_lsn: None,
                    applied: false,
                });
            }
            EventType::Insert { .. } | EventType::Update { .. } | EventType::Delete { .. } => {
                if let Some(tx) = self.active.values_mut().last() {
                    tx.changes.push(event);
                }
            }
            EventType::Commit { commit_lsn, .. } => {
                if let Some(xid) = self.active.keys().next().copied() {
                    if let Some(mut tx) = self.active.remove(&xid) {
                        tx.commit_lsn = Some(*commit_lsn);
                        // === 在這裡把 tx.changes 套用到你的下游系統 ===
                        tx.applied = true;
                        self.committed.push(tx);
                        // 只有套用成功後才推進 feedback LSN
                        if let Some(lsn) = self.committed.last().and_then(|t| t.commit_lsn) {
                            self.feedback.update_applied_lsn(lsn.value());
                        }
                    }
                }
            }
            _ => {}
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://postgres:password@localhost:5432/postgres?replication=database".to_string()
    });

    let config = ReplicationStreamConfig::new(
        "safe_consumer_slot".to_string(),
        "my_publication".to_string(),
        2, StreamingMode::On,
        Duration::from_secs(10),
        Duration::from_secs(30),
        Duration::from_secs(60),
        RetryConfig::default(),
    );

    let mut stream = LogicalReplicationStream::new(&conn, config).await?;
    // 從 stream 取得共享的 LSN feedback
    let feedback = stream.shared_lsn_feedback.clone();
    let mut consumer = SafeConsumer { feedback, active: BTreeMap::new(), committed: Vec::new() };

    stream.start(None).await?;

    let token = CancellationToken::new();
    let mut event_stream = stream.into_stream(token.clone());

    loop {
        match event_stream.next_event().await {
            Ok(event) => consumer.process(event).await,
            Err(e) if matches!(e, pg_walstream::ReplicationError::Cancelled(_)) => break,
            Err(e) => { eprintln!("Error: {}", e); break; }
        }
    }

    Ok(())
}

範例七：限流（Rate Limiting）

當下游 API 有速率上限、或你想平滑地攤平尖峰負載時，可以直接利用 stream combinator。包成 futures::Stream 後，用 tokio_stream 的 throttle 就能控制每秒事件數：

use futures::stream;
use pg_walstream::{
    CancellationToken, EventType, LogicalReplicationStream, ReplicationStreamConfig, RetryConfig,
    StreamingMode,
};
use std::time::Duration;
use tokio_stream::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://postgres:password@localhost:5432/postgres?replication=database".to_string()
    });

    let config = ReplicationStreamConfig::new(
        "rate_limited_slot".to_string(),
        "my_publication".to_string(),
        2, StreamingMode::On,
        Duration::from_secs(10),
        Duration::from_secs(30),
        Duration::from_secs(60),
        RetryConfig::default(),
    );

    let mut stream = LogicalReplicationStream::new(&conn, config).await?;
    stream.start(None).await?;

    let cancel_token = CancellationToken::new();
    let event_stream = stream.into_stream(cancel_token);

    // 包成 futures::Stream
    let pg_stream = stream::unfold(event_stream, |mut es| async move {
        match es.next_event().await {
            Ok(event) => {
                es.update_applied_lsn(event.lsn.value());
                Some((Ok(event), es))
            }
            Err(e) => Some((Err(e), es)),
        }
    });

    // 限流：每秒 10 筆
    let max_events_per_second = 10;
    let delay = Duration::from_secs(1) / max_events_per_second;
    let mut rate_limited = std::pin::pin!(pg_stream.throttle(delay));

    while let Some(result) = rate_limited.next().await {
        match result {
            Ok(event) => match &event.event_type {
                EventType::Insert { schema, table, .. } =>
                    println!("INSERT into {}.{} at LSN {}", schema, table, event.lsn),
                EventType::Update { schema, table, .. } =>
                    println!("UPDATE {}.{} at LSN {}", schema, table, event.lsn),
                EventType::Delete { schema, table, .. } =>
                    println!("DELETE from {}.{} at LSN {}", schema, table, event.lsn),
                EventType::Truncate(tables) =>
                    println!("TRUNCATE tables: {:?}", tables),
                _ => println!("Event at LSN {}", event.lsn),
            },
            Err(e) => { eprintln!("Error: {}", e); break; }
        }
    }

    Ok(())
}

範例八：Raw XLogData（自帶 decoder）

如果你想自己解碼 pgoutput（例如接自訂的 decoder，或只想轉發原始 bytes），可以用 next_raw_event。它回傳未解碼的 pgoutput payload 加上 WAL 位置（RawXLogData）；keepalive、feedback、cancellation 仍由函式庫處理，只是不做 pgoutput 解碼、也不會自動 ack（restart 語意由你自己掌握）：

use pg_walstream::{
    CancellationToken, LogicalReplicationStream, ReplicationError, ReplicationStreamConfig,
};
use std::env;
use tracing::info;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let connection_string = env::var("DATABASE_URL").unwrap_or_else(|_| {
        "postgresql://postgres:password@localhost:5432/postgres?replication=database".to_string()
    });

    let config = ReplicationStreamConfig::builder("raw_slot", "my_publication");

    let mut stream = LogicalReplicationStream::new(&connection_string, config).await?;
    stream.start(None).await?;
    info!("Streaming raw XLogData... (Press Ctrl+C to stop)");

    let cancel = CancellationToken::new();
    let cancel_clone = cancel.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.ok();
        cancel_clone.cancel();
    });

    loop {
        match stream.next_raw_event(&cancel).await {
            Ok(raw) => {
                // raw.data 是未解碼的 pgoutput bytes——自帶你的 decoder
                info!(
                    wal_start = %raw.wal_start,
                    wal_end = %raw.wal_end,
                    payload_len = raw.data.len(),
                    "raw XLogData",
                );

                // 自行 ack：在 payload 被「持久化處理」之後，才推進 applied LSN。
                // 用 wal_end（本訊息之後的下一個 byte）回報。
                stream.shared_lsn_feedback.update_applied_lsn(raw.wal_end.value());
            }
            Err(ReplicationError::Cancelled(_)) => break,
            Err(e) => { eprintln!("Error: {}", e); break; }
        }
    }

    Ok(())
}

欄位對應：struct 欄位名 ↔ 資料表欄位名

WAL 的每一列是以真實的 PostgreSQL 欄位名為 key。當 struct 欄位名和欄位名不同時，直接用 serde 的 #[serde(rename = "...")] 即可，不需要額外屬性：

use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct User {
    id: i64,
    #[serde(rename = "user_name")] // 欄位 `username` ← 資料表欄位 `user_name`
    username: String,
    #[serde(rename = "mail")]      // 欄位 `email`    ← 資料表欄位 `mail`
    email: Option<String>,         // 可為 NULL 的欄位 → Option
}

// let user: User = event.deserialize_insert()?;   // 或 row.deserialize_into()?

搭配 derive 功能時，加上 #[wal_table("...")]（放最外層）即可把型別綁定到資料表供 WalRouter 使用；它會和上面的 rename 組合在一起：

#[wal_table("typed_deser_users")]
#[derive(Debug, Deserialize)]
struct User { /* 欄位同上 */ }

LSN 追蹤

執行緒安全的 LSN 追蹤，用於向 PostgreSQL 回報進度：

use pg_walstream::SharedLsnFeedback;

let feedback = SharedLsnFeedback::new_shared();

// Producer 端：讀取要回報的 LSN
let (flushed_lsn, applied_lsn) = feedback.get_feedback_lsn();

// Consumer 端：處理完後更新 LSN
feedback.update_applied_lsn(commit_lsn);

訊息型別（Message Types）

pg-walstream 支援所有 PostgreSQL logical replication 訊息型別：

Protocol Version 1 訊息

• BEGIN：交易開始
• COMMIT：交易提交
• ORIGIN：複寫來源
• RELATION：資料表 schema 定義
• TYPE：資料型別定義
• INSERT：新增列
• UPDATE：更新列
• DELETE：刪除列
• TRUNCATE：清空資料表
• MESSAGE：通用訊息

Protocol Version 2+ 訊息（Streaming）

• STREAM_START：串流交易開始
• STREAM_STOP：串流交易段落結束
• STREAM_COMMIT：串流交易提交
• STREAM_ABORT：串流交易中止

Protocol Version 3+ 訊息（Two-Phase Commit）

• BEGIN_PREPARE：prepared 交易開始
• PREPARE：交易 prepare
• COMMIT_PREPARED：提交 prepared 交易
• ROLLBACK_PREPARED：回滾 prepared 交易
• STREAM_PREPARE：stream prepare 訊息

更多可執行範例

倉庫的 examples/ 目錄提供多個獨立、可直接執行的範例：

效能與壓力測試

官方以漸進式的寫入併發（16 ~ 192 個 writer）測試函式庫的 CPU 飽和點與吞吐上限，並比較兩種後端：

• Backend A：rustls-tls
• Backend B：libpq

測試環境：8 vCPU 的 Linux VM（已做 TCP 調校：64 MB buffer、BBR），透過 TLS（sslmode=require）從遠端的 Azure PostgreSQL Flexible Server 18.4 串流。每個情境跑 10 秒暖機 + 30 秒量測，每個組態每個後端量測 3 次取中位數。CPU/RSS 只反映 pg-walstream consumer 本身（寫入產生器是另一個獨立 OS process）。

如何解讀這些數字：在真實網路連線上，consumer 大部分時間都停在 epoll/TLS I/O，而不是在解析，因此兩種後端都是 network-I/O-bound，在 CPU 效率、CPU%、RSS、延遲上收斂到跑與跑之間的雜訊範圍內。rustls-tls 在飽和的單串流 ingest 上維持約 8% 的吞吐優勢。若你在 loopback/localhost 上量測，可能會看到較大的 CPU 差距，但那並不代表真實的遠端資料庫部署。

峰值數字

CPU 效率（DML events/sec / 1% CPU，越高越好）

兩種後端在每個情境都落在 ±5% 以內——統計上打平。

吞吐（節選）

rustls-tls 在飽和的單串流 ingest（Baseline、Batch-5000、Payload-2KB）約快 8–9%，其餘皆在雜訊範圍內。

生產環境的 Linux VM TCP 調校

跨區域（例如 cross-region 的 Azure PostgreSQL）串流 WAL 時，Linux 預設的 TCP buffer 大小可能成為吞吐瓶頸。核心預設的 rmem_max（208 KB）會限制 TCP 接收視窗，再結合往返延遲，透過 Bandwidth-Delay Product（BDP） 卡住吞吐。

建議的 sysctl 設定：

# --- TCP buffer 大小 ---
# 允許每個 socket 最大 64 MB 收/送 buffer（核心會在此上限內自動調整）
net.core.rmem_max = 67108864
net.core.wmem_max = 67108864

# TCP 自動調整範圍：min / default / max（bytes）
net.ipv4.tcp_rmem = 4096 262144 67108864
net.ipv4.tcp_wmem = 4096 262144 67108864

# --- 壅塞控制 ---
# 在高延遲連線上，BBR 的吞吐明顯優於 cubic
net.ipv4.tcp_congestion_control = bbr

# --- 封包 backlog ---
# 增加 NIC 接收佇列（在高封包速率下有幫助）
net.core.netdev_max_backlog = 5000

套用：

sudo sysctl --system

註：這些設定影響 VM 上所有 TCP 連線，不只 pg-walstream。核心會在設定的上限內自動調整實際 buffer 用量，所以閒置連線不會各自吃掉 64 MB。

限制

• 需要 PostgreSQL 14 或以上，才能完整支援協定
• 必須先建立 logical replication slot 才能開始串流
• 只支援 binary protocol（不支援 text-based protocol）
• 資料庫使用者需具備 replication 權限

資源

• PostgreSQL Logical Replication Documentation
• Logical Replication Message Formats
• Replication Protocol

總結

pg-walstream 提供了一個乾淨、可擴展的 Rust API，讓開發者可以直接使用 PostgreSQL Logical / Physical Replication。到了 v0.8.0，它更進一步：預設純 Rust（不再綁 libpq）、typed 反序列化、WalRouter 型別路由、derive 巨集、bounded replay、raw XLogData、two-phase commit、parallel streaming、base backup——同時在真實網路環境下維持約 17 萬 DML events/sec 的吞吐。

如果你正在：

• 用 Rust 開發資料基礎設施
• 想深入理解 PostgreSQL WAL / replication protocol
• 或需要一個可嵌入的 CDC 核心元件

pg-walstream 會是一個非常適合的起點。

專案連結：
https://github.com/isdaniel/pg-walstream