pg2any

A high-performance PostgreSQL Change Data Capture (CDC) tool that streams database changes in real-time to multiple destination databases via logical replication.

Supported Destinations

Key Features

• Crash-safe persistence - File-based producer-consumer with automatic crash recovery
• Resumable processing - Restart from the exact position within large transaction files
• Streaming transactions - Handles PostgreSQL protocol v2+ in-progress transaction streaming
• SQL compression - Optional gzip compression for transaction files with streaming decompression
• Schema mapping - Configurable PostgreSQL schema to destination database/schema translation
• Prometheus metrics - Built-in HTTP metrics endpoint (feature: metrics)
• Graceful shutdown - Coordinated producer-consumer shutdown with LSN persistence

Quick Start

Prerequisites

PostgreSQL with logical replication enabled:

ALTER SYSTEM SET wal_level = logical;
-- Restart PostgreSQL after this change

CREATE PUBLICATION my_publication FOR ALL TABLES;
CREATE USER replicator WITH REPLICATION LOGIN PASSWORD 'password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replicator;

Docker

git clone https://github.com/isdaniel/pg2any
cd pg2any
docker-compose up -d
make build && RUST_LOG=info make run

As a Library

use pg2any_lib::{load_config_from_env, run_cdc_app};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = load_config_from_env()?;
    run_cdc_app(config, None).await?;
    Ok(())
}

Programmatic Configuration

use pg2any_lib::{Config, DestinationType};

let config = Config::builder()
    .source_connection_string("postgresql://user:pass@localhost:5432/db?replication=database")
    .destination_type(DestinationType::MySQL)
    .destination_connection_string("mysql://root:pass@localhost:3306/replica_db")
    .replication_slot_name("cdc_slot")
    .publication_name("cdc_pub")
    .streaming(true)
    .build()?;

Architecture

pg2any uses a file-based producer-consumer pattern for reliable, crash-safe transaction processing:

PostgreSQL WAL Stream
        |
        v
  +-----------+       +------------------+       +-----------+
  | Producer  | ----> | File System      | ----> | Consumer  |
  | (WAL      |       |                  |       | (SQL      |
  |  Reader)  |       | sql_data_tx/     |       |  Executor)|
  |           |       | sql_received_tx/ |       |           |
  +-----------+       | sql_pending_tx/  |       +-----------+
                      +------------------+             |
                                                       v
                                                 Destination DB

Transaction Lifecycle

1. BEGIN - Create .sql file in sql_data_tx/ and .meta in sql_received_tx/
2. Events - Append SQL commands to .sql file via buffered writes (8MB buffer)
3. COMMIT - Move .meta from sql_received_tx/ to sql_pending_tx/; notify consumer
4. Execute - Consumer reads pending .meta, executes SQL from .sql in batches
5. Cleanup - Delete both .meta and .sql files on success; update flush LSN

Crash Recovery

On restart, pg2any:

• Cleans up incomplete transactions from sql_received_tx/ (uncommitted)
• Replays committed transactions from sql_pending_tx/ (committed but not yet applied)
• Resumes from last_executed_command_index within partially-executed transaction files
• Starts replication from the persisted flush_lsn

Workflow Diagrams

Transaction Processing Detail Flow

sequenceDiagram
    participant PG as PostgreSQL WAL
    participant Producer as Producer Task
    participant FS as File System
    participant Channel as Notification Channel
    participant Consumer as Consumer Task
    participant Dest as Destination DB
    participant LSN as LSN Tracker

    Note over PG,LSN: Transaction Processing Flow

    PG->>Producer: BEGIN (tx_id: 12345)
    Producer->>FS: Create 12345.sql in sql_data_tx/
    Producer->>FS: Create 12345.meta in sql_received_tx/

    PG->>Producer: INSERT event
    Producer->>Producer: Generate SQL
    Producer->>FS: Append to buffer (8MB)

    PG->>Producer: UPDATE event
    Producer->>Producer: Generate SQL
    Producer->>FS: Append to buffer

    PG->>Producer: DELETE event
    Producer->>Producer: Generate SQL
    Producer->>FS: Append to buffer

    Note over Producer,FS: Buffer reaches 8MB
    Producer->>FS: Flush buffer to 12345.sql

    PG->>Producer: COMMIT (LSN: 0/1A2B3C4D)
    Producer->>FS: Flush remaining buffer
    Producer->>FS: Move 12345.meta to sql_pending_tx/
    Producer->>Channel: Send notification

    Channel->>Consumer: Transaction ready
    Consumer->>FS: Read 12345.meta from sql_pending_tx/
    Consumer->>FS: Read SQL commands from 12345.sql

    loop For each SQL batch
        Consumer->>Dest: BEGIN TRANSACTION
        Consumer->>Dest: Execute SQL batch
        Consumer->>Dest: COMMIT TRANSACTION
        Consumer->>FS: Update pending .meta progress
    end

    Consumer->>FS: Delete 12345.meta and 12345.sql
    Consumer->>LSN: Update and persist flush_lsn

Crash Recovery Workflow

graph TB
    Crash([System Crash]) --> Restart[Restart pg2any]
    Restart --> LoadMeta[Load LSN Metadata]
    LoadMeta --> CheckMeta{Metadata Exists?}

    CheckMeta -->|No| StartFresh[Start from Latest]
    CheckMeta -->|Yes| GetLSN[Extract flush_lsn]

    GetLSN --> CleanReceived[Cleanup sql_received_tx/]
    CleanReceived --> ProcessPending[Process sql_pending_tx/]

    ProcessPending --> SortByTimestamp[Sort by commit_timestamp]
    SortByTimestamp --> ForEachPending{For Each .meta}

    ForEachPending -->|More files| CheckResume{Has resume index?}
    ForEachPending -->|Done| SetStartLSN[Set start_lsn = flush_lsn]

    CheckResume -->|Yes| ResumeFromIndex[Resume from last_executed_command_index + 1]
    CheckResume -->|No| ReadAll[Read All SQL Commands]

    ResumeFromIndex --> Execute[Execute Remaining Commands]
    ReadAll --> Execute

    Execute --> UpdateLSN[Update LSN Metadata]
    UpdateLSN --> DeleteFiles[Delete .meta and .sql]
    DeleteFiles --> ForEachPending

    SetStartLSN --> StartReplication[Start Replication]
    StartFresh --> StartReplication
    StartReplication --> NormalOperation([Normal Operation])

    style Crash fill:#ff6b6b
    style NormalOperation fill:#51cf66
    style ResumeFromIndex fill:#ffd43b

Configuration

All configuration is via environment variables (ideal for containers) or the ConfigBuilder API.

Required

Destination URI Formats

Optional

Monitoring

Enable with feature flag metrics. Exposes Prometheus-compatible metrics on port 8080.

# Key metrics
pg2any_events_processed_total
pg2any_transactions_processed_total
pg2any_replication_lag_seconds
pg2any_events_per_second
pg2any_errors_total
pg2any_source_connection_status
pg2any_destination_connection_status

The Docker Compose setup includes a full observability stack: Prometheus (:9090), Node Exporter, PostgreSQL Exporter, and MySQL Exporter with predefined alert rules.

Development

make build              # Build the application
make test               # Run full test suite
make check              # Cargo check + validation
make format             # Format code with rustfmt
make before-git-push    # Pre-commit validation

# Docker environment
make docker-start       # Start databases + monitoring
make docker-stop        # Stop all services

# Chaos & integration tests
make chaos-test-mysql-full
make chaos-test-sqlserver-full
make chaos-test-sqlite-full
make chaos-test-kafka-full
make pgbench-test-mysql-full

Feature Flags

[features]
default = ["mysql", "sqlserver", "sqlite"]
mysql = ["sqlx/mysql"]
sqlserver = ["tiberius"]
sqlite = ["sqlx/sqlite"]
kafka = ["rdkafka", "futures-util", "base64"]
metrics = ["hyper", "hyper-util", "http-body-util", "prometheus"]

Dependencies

License

Apache-2.0

References

• PostgreSQL Logical Replication Protocol
• PostgreSQL WAL Internals
• pg_walstream - Underlying replication library