# How pgrust Handles Logical Replication: A Deep Dive into the PostgreSQL Rust Client

> Discover how pgrust handles logical replication. Learn about its CopyBoth streaming API and PgClientConn for seamless Rust client to PostgreSQL server communication.

- Repository: [Michael Malis/pgrust](https://github.com/malisper/pgrust)
- Tags: deep-dive
- Published: 2026-07-13

---

**pgrust implements logical replication by exposing a CopyBoth streaming API that mirrors PostgreSQL's libpq logical-replication protocol, using `PgClientConn` to manage bidirectional streams between Rust clients and PostgreSQL servers.**

The pgrust project (`malisper/pgrust`) provides a Rust-native implementation of the PostgreSQL wire protocol and client libraries. When handling pgrust logical replication, the codebase exposes a type-safe abstraction over PostgreSQL's libpq logical-replication protocol, enabling Rust applications to act as replication consumers through the `CopyBoth` streaming mechanism.

## The CopyBoth Replication Protocol

PostgreSQL logical replication relies on the `COPY BOTH` protocol mode, which creates a bidirectional stream between client and server. In pgrust, this is implemented through the `PgClientConn` struct in [`crates/interfaces/libpq/fe/src/client.rs`](https://github.com/malisper/pgrust/blob/main/crates/interfaces/libpq/fe/src/client.rs), which manages the stateful connection required for replication slots and logical decoding.

## Establishing a Replication Connection

### Startup Parameters and Connection Flags

Replication connections require special handling during the startup phase. In [`crates/interfaces/libpq/fe/src/protocol3.rs`](https://github.com/malisper/pgrust/blob/main/crates/interfaces/libpq/fe/src/protocol3.rs), the `StartupParams` struct provides a `replication()` method that sets the `replication` parameter to `"true"`, signaling the server to enter replication mode.

To negotiate a replication connection, configure the startup parameters before calling `PgClientConn::connect`:

```rust
let params = pgrust::interfaces::libpq::fe::protocol3::StartupParams::new()
    .user("replicator")
    .replication("true");   // Enables logical replication mode

```

## Starting and Managing the Replication Stream

### Initiating START_REPLICATION

Once connected, the client initiates logical replication by calling `PgClientConn::start_replication` (defined at line 425 in [`client.rs`](https://github.com/malisper/pgrust/blob/main/client.rs)). This sends the `START_REPLICATION` command as part of a `CopyBoth` transaction, opening the bidirectional stream.

The method returns a result with `ExecStatusType::CopyBoth`, confirming the connection has entered streaming mode.

### Receiving XLogData Messages

The `PgClientConn::copy_receive` method (lines 39-50 in [`client.rs`](https://github.com/malisper/pgrust/blob/main/client.rs)) wraps the underlying `PQgetCopyData` call and returns a `CopyRecv` enum with variants:

- **`Data`** – Contains the logical decoding payload (`XLogData` messages)
- **`Done`** – Signals the stream has completed
- **`WouldBlock`** – Indicates no data is currently available

### Sending Standby Feedback

Replication slots require periodic feedback to maintain progress and prevent WAL accumulation. The `PgClientConn::copy_send` method allows the client to push standby progress messages back to the server, while `copy_done` and `end_copy` properly terminate the `CopyBoth` lifecycle.

## Logical Decoding Plugin Architecture

### The Test Decoding Plugin Implementation

The repository includes a reference implementation in [`crates/contrib/test_decoding/src/lib.rs`](https://github.com/malisper/pgrust/blob/main/crates/contrib/test_decoding/src/lib.rs), demonstrating how logical decoding plugins receive transactions through `LogicalDecodingContext`, `TxnHandle`, `RelationHandle`, and `ChangeHandle`. This plugin uses the `types_logical` crate types that underpin the entire replication data path, processing changes into textual output.

## Server-Side Infrastructure

### Replication Slot Management

Server-side functions for slot management, such as `pg_create_logical_replication_slot`, are declared in [`crates/backend/utils/fmgr/builtin_canonical.rs`](https://github.com/malisper/pgrust/blob/main/crates/backend/utils/fmgr/builtin_canonical.rs) (function IDs 3786, 3787). These provide the server-side counterpart to the client's replication commands, enabling Rust applications to interact with PostgreSQL's replication slot infrastructure.

### Configuration GUCs

Runtime configuration for logical replication workers is controlled through GUCs defined in [`crates/backend/utils/misc/guc_tables/src/vars.rs`](https://github.com/malisper/pgrust/blob/main/crates/backend/utils/misc/guc_tables/src/vars.rs), including `max_logical_replication_workers` and `wal_receiver_create_temp_slot`. These variables expose runtime knobs for tuning replication behavior.

## Complete Implementation Example

The following example demonstrates the complete flow for establishing a pgrust logical replication connection, receiving `XLogData` messages, and sending feedback:

```rust
use pgrust::interfaces::libpq::fe::client::PgClientConn;
use pgrust::interfaces::libpq::fe::transport::TcpTransport;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1️⃣ Open a TCP transport to the server
    let transport = TcpTransport::connect("localhost:5432")?;

    // 2️⃣ Build startup parameters with replication=true
    let params = pgrust::interfaces::libpq::fe::protocol3::StartupParams::new()
        .user("replicator")
        .replication("true");   // <‑‑ logical‑replication flag

    // 3️⃣ Connect and authenticate
    let mut conn = PgClientConn::connect(transport, &params, None)?;

    // 4️⃣ Start the replication stream
    let result = conn.start_replication("START_REPLICATION 0/0")?;
    assert!(result.status == pgrust::interfaces::libpq::fe::result::ExecStatusType::CopyBoth);

    // 5️⃣ Receive logical‑decoding messages
    loop {
        match conn.copy_receive()? {
            pgrust::interfaces::libpq::fe::client::CopyRecv::Data(buf) => {
                // `buf` contains an XLogData record from the output plugin
                println!("Received {} bytes", buf.len());
                // …process the logical change…
                // (optionally send feedback)
                conn.copy_send(b"standby feedback ...")?;
            }
            pgrust::interfaces::libpq::fe::client::CopyRecv::Done => break,
            pgrust::interfaces::libpq::fe::client::CopyRecv::WouldBlock => {
                // In a real app you would poll / await the socket here
                std::thread::sleep(std::time::Duration::from_millis(10));
            }
        }
    }

    // 6️⃣ End the COPY‑Both transaction
    conn.copy_done()?;
    conn.end_copy()?;
    Ok(())
}

```

## Summary

- **CopyBoth Protocol**: pgrust uses `PgClientConn` to manage bidirectional `CopyBoth` streams that mirror libpq's logical replication behavior.
- **Startup Configuration**: Set `replication("true")` in `StartupParams` to negotiate a replication connection during protocol negotiation in [`protocol3.rs`](https://github.com/malisper/pgrust/blob/main/protocol3.rs).
- **Stream Management**: Use `start_replication`, `copy_receive`, and `copy_send` to handle the lifecycle of logical decoding messages.
- **Type Safety**: The `CopyRecv` enum provides safe handling of `XLogData` messages with `Data`, `Done`, and `WouldBlock` variants.
- **Plugin Support**: The `test_decoding` plugin in [`crates/contrib/test_decoding/src/lib.rs`](https://github.com/malisper/pgrust/blob/main/crates/contrib/test_decoding/src/lib.rs) demonstrates the server-side logical decoding API using `types_logical` primitives.
- **Server Integration**: Replication slots and GUCs in [`builtin_canonical.rs`](https://github.com/malisper/pgrust/blob/main/builtin_canonical.rs) and [`guc_tables/src/vars.rs`](https://github.com/malisper/pgrust/blob/main/guc_tables/src/vars.rs) complete the implementation stack.

## Frequently Asked Questions

### What is the `CopyBoth` protocol in pgrust?

The `CopyBoth` protocol is PostgreSQL's bidirectional streaming mode used for logical replication. In pgrust, it is implemented through the `PgClientConn` struct in [`client.rs`](https://github.com/malisper/pgrust/blob/main/client.rs), which sends `START_REPLICATION` commands and manages the bidirectional flow of `XLogData` messages and standby feedback between Rust clients and PostgreSQL servers.

### How do I initiate a logical replication connection in pgrust?

You initiate a logical replication connection by calling `PgClientConn::connect` with `StartupParams` that include `.replication("true")`, as defined in [`protocol3.rs`](https://github.com/malisper/pgrust/blob/main/protocol3.rs). This sets the replication flag in the startup packet, causing the server to negotiate a replication connection rather than a standard query session.

### What does the `CopyRecv` enum represent in pgrust logical replication?

The `CopyRecv` enum in [`client.rs`](https://github.com/malisper/pgrust/blob/main/client.rs) represents the possible states of data reception during logical replication. It has three variants: `Data` containing the payload buffer, `Done` indicating the stream has ended, and `WouldBlock` signaling that no data is currently available and the client should poll or wait.

### Does pgrust provide server-side logical decoding capabilities?

Yes, pgrust includes server-side infrastructure for logical decoding, demonstrated by the `test_decoding` output plugin in [`crates/contrib/test_decoding/src/lib.rs`](https://github.com/malisper/pgrust/blob/main/crates/contrib/test_decoding/src/lib.rs). This plugin implements the `LogicalDecodingContext` trait and handles transactions through `TxnHandle`, `RelationHandle`, and `ChangeHandle` types, processing changes into textual output using the `types_logical` crate.