BlueTusk
GitHubGet Started
Provider · docs/replication/README.md

Replication

Physical and logical replication, pgoutput, and WAL feedback.

7 min read1,517 wordsSource synchronized
View source on GitHub
ON THIS PAGE

Replication

BlueTusk.Replication exposes PostgreSQL’s physical and logical streaming replication protocol directly. It uses a dedicated replication session and COPY BOTH; it does not borrow an ADO.NET pooled connection.

Build one long-lived data source for the application configuration, then derive a fresh dedicated-session option snapshot for replication:

await using var dataSource = new BlueTuskDataSourceBuilder(connectionString).Build();
await using var replication = await BlueTuskLogicalReplicationConnection.OpenAsync(
    dataSource.CreateDedicatedSessionOptions(),
    cancellationToken);

The data source remains the configuration root, but it does not own the replication connection. The replication object owns one unpooled physical session and must be disposed independently. Its lifetime may be much longer than an ADO.NET command or pooled checkout. Configured credentials, application name, timeout, TLS mode, and channel binding are copied into the snapshot; ADO.NET codecs and runtime catalogue state are intentionally irrelevant to raw replication payloads.

The stream is pull-based and does not maintain a background prefetch queue. At most the current CopyData payload and decoded message are owned by the consumer path; PostgreSQL/socket flow control supplies backpressure until the consumer requests the next element. Process or hand off each payload promptly, and put an application-owned bounded queue in front of slower durable work only when that queue’s capacity and failure semantics are deliberate.

The BlueTusk.Diagnostics meter records WAL-sender clock lag and byte-position lag for XLogData and keepalive messages. These are receive-side observations, not claims that a downstream consumer has applied a transaction. See Diagnostics and observability for metric names and dimensions.

For a multi-host data source, select a configured endpoint explicitly:

var endpoint = new BlueTuskHostEndpoint("primary.example.test", 5432);
var options = dataSource.CreateDedicatedSessionOptions(endpoint);

BlueTusk does not silently fail a replication stream over to another host. The application must establish that the replacement server and slot are safe for its persisted resume position.

BlueTusk supports PostgreSQL 15 through 19 and provides:

  • physical and logical replication connections;
  • system identification, settings, replication-slot discovery, and slot lifecycle commands;
  • publication and publication-table discovery;
  • WAL data, primary keepalives, standby status updates, and hot-standby feedback;
  • transaction streaming and two-phase startup options;
  • raw payloads for any logical decoding output plugin; and
  • complete pgoutput decoding in BlueTusk.Replication.PgOutput.

The replication subsystem has passed its production-readiness gate: the live version matrix, compatibility baselines, durability and feedback checks, failure-recovery cases, allocation/backpressure benchmarks, cancellation stress, and scheduled endurance are all executable. The packages are published at stable 1.0.0. Later releases still require the exact immutable candidate and protected release evidence.

Server setup

The server must admit a role with the REPLICATION attribute in pg_hba.conf. Logical replication also requires wal_level = logical; replication capacity is controlled by max_wal_senders and max_replication_slots.

Use a dedicated least-privilege role in production. Keep credentials out of logs and source control. BlueTusk verifies the server certificate and hostname by default; local environments without TLS must explicitly select SSL Mode=Disable.

Logical replication

Create the publication and slot once through deployment tooling or through the connection:

await using var replication =
    await BlueTuskLogicalReplicationConnection.OpenAsync(
        dataSource.CreateDedicatedSessionOptions());

var slot = await replication.CreateReplicationSlotAsync(
    slotName: "app_slot",
    outputPlugin: "pgoutput");

The convenience overload exactly configures pgoutput protocol version 1 for one publication:

await foreach (var message in replication.StartReplicationAsync(
    slotName: "app_slot",
    publicationName: "app_publication",
    cancellationToken))
{
    if (message is BlueTuskXLogData data)
    {
        // data.Data contains one output-plugin message.
    }
}

Use typed startup options for binary tuples, logical messages, large in-progress transactions, multiple publications, two-phase transactions, or origin filtering:

var stream = replication.StartReplicationAsync(
    new BlueTuskPgOutputReplicationOptions
    {
        SlotName = "app_slot",
        PublicationNames = ["app_publication", "audit_publication"],
        ProtocolVersion = 3,
        StreamingMode = BlueTuskLogicalStreamingMode.On,
        TwoPhase = true,
        Messages = true,
    },
    cancellationToken);

The selected protocol features must be supported by the server. Protocol version 2 adds streamed transactions, version 3 adds two-phase messages, and version 4 adds parallel-stream abort metadata.

pgoutput decoding

Reference BlueTusk.Replication.PgOutput and apply the decoder extension:

var decoderOptions = new BlueTuskPgOutputDecoderOptions
{
    ProtocolVersion = 3,
    StreamingMode = BlueTuskPgOutputStreamingMode.On,
    TwoPhase = true,
};

await foreach (var envelope in stream.DecodePgOutputAsync(
    decoderOptions,
    cancellationToken))
{
    switch (envelope.Message)
    {
        case BlueTuskPgOutputRelation relation:
            // Cache relation.Columns by relation.RelationId.
            break;
        case BlueTuskPgOutputInsert insert:
            // Tuple values retain null, unchanged-TOAST, text, or binary form.
            break;
        case BlueTuskPgOutputStreamStart streamStart:
            // A segment of a large in-progress transaction has started.
            break;
        case BlueTuskPgOutputPrepare prepare:
            // A two-phase transaction reached PREPARE TRANSACTION.
            break;
    }
}

BlueTuskPgOutputEnvelope retains the enclosing BlueTuskXLogData, including its WAL start, end, server end, and server clock. The decoder validates message lengths, flags, tuple markers, protocol-version capabilities, and stream segment state.

Feedback and durability

BlueTusk automatically answers primary keepalives that request an immediate reply. Applications control acknowledged positions. For logical pgoutput, checkpoint the transaction-end LSN from a terminal message, not BlueTuskXLogData.WalEnd: logical payload byte length is not a WAL byte count.

await ApplyAndCommitAsync(envelope.Message, cancellationToken);

if (envelope.TryGetTransactionEndPosition(out var applied))
{
    await checkpoints.StoreAppliedPositionAsync(applied, cancellationToken);
    await replication.SendStandbyStatusUpdateAsync(
        new BlueTuskStandbyStatus(
            Written: applied,
            Flushed: applied,
            Applied: applied),
        cancellationToken);
}

TryGetTransactionEndPosition recognizes commit, streamed commit, prepared, commit-prepared, rollback-prepared, and streamed-prepare terminal messages. Persist a prepare position only when the consumer also durably retains the prepared transaction’s state.

For physical replication, payload bytes correspond to WAL bytes, so BlueTuskXLogData.WalEnd is the receiver position:

var applied = wal.WalEnd;
await replication.SendStandbyStatusUpdateAsync(
    new BlueTuskStandbyStatus(
        Written: applied,
        Flushed: applied,
        Applied: applied),
    cancellationToken);

Advance Flushed or Applied only after the corresponding data is durable. PostgreSQL can reclaim WAL based on slot progress; acknowledging data that can still be lost breaks recovery guarantees. BlueTusk rejects feedback where Applied > Flushed > Written or where any position moves backwards, and it updates its local status only after the wire write succeeds. Concurrent manual updates and automatic keepalive replies are serialized. Physical standbys can also call SendHotStandbyFeedbackAsync with their xmin horizons.

Physical replication

Identify the system and begin at a retained WAL position:

await using var replication =
    await BlueTuskPhysicalReplicationConnection.OpenAsync(
        dataSource.CreateDedicatedSessionOptions());
var identity = await replication.IdentifySystemAsync(cancellationToken);

await foreach (var message in replication.StartReplicationAsync(
    identity.WalPosition,
    cancellationToken: cancellationToken))
{
    switch (message)
    {
        case BlueTuskXLogData wal:
            await PersistWalAsync(wal.Data, cancellationToken);
            await replication.SendStandbyStatusUpdateAsync(
                new BlueTuskStandbyStatus(wal.WalEnd, wal.WalEnd, wal.WalEnd),
                cancellationToken);
            break;
        case BlueTuskPrimaryKeepalive keepalive:
            Console.WriteLine($"Primary WAL end: {keepalive.ServerWalEnd}");
            break;
    }
}

For a physical slot, prefer the retained restart position returned by ReadReplicationSlotAsync over an older caller-cached position.

Discovery and custom plugins

GetReplicationSlotsAsync lists slot activity and progress. GetPublicationsAsync and GetPublicationTablesAsync expose publication ownership, operations, columns, and row filters.

For another logical decoding plugin, create the slot with its plugin name and pass arbitrary plugin options:

var stream = replication.StartReplicationAsync(
    new BlueTuskLogicalReplicationRequest
    {
        SlotName = "audit_slot",
        PluginOptions = new Dictionary<string, string?>
        {
            ["include-xids"] = "true",
        },
    },
    cancellationToken);

Each BlueTuskXLogData.Data value is the plugin’s raw payload. BlueTusk does not interpret custom formats.

Cancellation or asynchronous enumerator disposal sends CopyDone, drains the server back to ReadyForQuery, and releases the replication operation. Forced connection disposal instead closes the transport to interrupt a pending read. The protocol layer retains its rented receive buffer until that read has unwound, so cancellation or teardown cannot return storage to the shared pool while an asynchronous continuation still references it. A deterministic unit test and the live replication-disposal stress case enforce this ownership invariant.

Reconnect and resume

Persist a BlueTuskLogicalReplicationCheckpoint outside the replication process. It binds the last durably applied transaction-end position to the PostgreSQL system identifier, database, persistent slot, and output plug-in. On a transient disconnect, create a new dedicated replication connection from the data source, validate that checkpoint, and request its applied position. Do not resume from the largest position merely received in memory.

var checkpoint = await checkpoints.LoadAsync(cancellationToken);

while (!cancellationToken.IsCancellationRequested)
{
    try
    {
        await using var replication =
            await BlueTuskLogicalReplicationConnection.OpenAsync(
                dataSource.CreateDedicatedSessionOptions(),
                cancellationToken);

        await replication.ValidateResumeCheckpointAsync(
            checkpoint,
            cancellationToken);

        var request = new BlueTuskPgOutputReplicationOptions
        {
            SlotName = checkpoint.SlotName,
            PublicationNames = ["app_publication"],
            StartPosition = checkpoint.AppliedPosition,
        };

        await foreach (var envelope in replication
            .StartReplicationAsync(request, cancellationToken)
            .DecodePgOutputAsync(cancellationToken: cancellationToken))
        {
            await ApplyAndCommitAsync(envelope.Message, cancellationToken);
            if (envelope.TryGetTransactionEndPosition(out var applied))
            {
                checkpoint = checkpoint with { AppliedPosition = applied };
                await checkpoints.StoreAsync(checkpoint, cancellationToken);
                await replication.SendStandbyStatusUpdateAsync(
                    new BlueTuskStandbyStatus(applied, applied, applied),
                    cancellationToken);
            }
        }
    }
    catch (Exception exception) when (
        IsTransientReplicationFailure(exception) &&
        !cancellationToken.IsCancellationRequested)
    {
        await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
    }
}

Validation rejects another cluster/database, a missing, temporary, active, or wrong-plug-in slot, lost/unreserved WAL, a checkpoint older than restart_lsn, a server confirmed_flush_lsn ahead of the application checkpoint, and a checkpoint ahead of the server. This implements PostgreSQL’s documented advice to compare confirmed_flush_lsn before START_REPLICATION, which otherwise starts at the greater of the requested and confirmed positions. If validation fails, stop and repair from an application-specific snapshot rather than skipping data. Recreate the decoder after reconnect so relation and streamed-transaction state cannot leak across sessions.

The live version-matrix test uses a persistent slot over multiple independent sessions, rejects wrong-system, missing, active, and stale checkpoints, and verifies every transaction exactly once in the durable consumer log. Set BLUETUSK_REPLICATION_DURABILITY_EPOCHS to increase reconnect epochs for a bounded soak.

The scheduled/manual PostgreSQL 19 endurance job runs 1,000 epochs (4,000 replicated rows) and the same test remains part of the ordinary PostgreSQL 15–19 matrix at its fast default. The repository gate covers feedback ordering, wrong-system, missing, active, stale, and retained-WAL failures plus repeated connection teardown/recreation. This closes the replication-specific durability matrix; it does not by itself make the experimental provider a production-ready 1.0 release.