async_postgres/pg_replication

Search:
Group by:

PostgreSQL Logical Replication support.

Provides types and procedures for consuming a logical replication stream via the PostgreSQL streaming replication protocol. The streaming API is plugin-agnostic (raw WAL bytes are delivered to a callback). A built-in decoder for the pgoutput logical decoding plugin is included.

Quick start

let conn = await connectReplication("postgresql://user:pass@host/db")
defer: await conn.close()
let slot = await conn.createReplicationSlot("my_slot", "pgoutput", temporary = true)
await conn.startReplication("my_slot", slot.consistentPoint,
    options = {"proto_version": "1", "publication_names": "my_pub"},
    callback = myCallback)

Types

BeginMessage = object
  finalLsn*: Lsn             ## LSN of the commit record
  commitTime*: int64         ## Commit timestamp (microseconds since PG epoch)
  xid*: int32                ## Transaction ID
Transaction begin. 
CommitMessage = object
  flags*: byte
  commitLsn*: Lsn
  endLsn*: Lsn
  commitTime*: int64
Transaction commit. 
DeleteMessage = object
  relationId*: int32
  keyKind*: char ## 'K' if oldTuple holds only the replica identity key,
                 ## 'O' if it holds the full old row (REPLICA IDENTITY FULL).
  oldTuple*: seq[TupleField]
Row deletion. 
InsertMessage = object
  relationId*: int32
  newTuple*: seq[TupleField]
Row insertion. 
LogicalMessage = object
  flags*: byte               ## Bit 0: transactional
  lsn*: Lsn
  prefix*: string
  content*: seq[byte]
Generic logical decoding message (via pg_logical_emit_message). 
Lsn = distinct uint64
LSN (Log Sequence Number) PostgreSQL Log Sequence Number. Displayed as "X/Y" where X and Y are hex-encoded upper and lower 32-bit halves. 
OriginMessage = object
  originLsn*: Lsn
  originName*: string
Replication origin. 
PgOutputMessage = object
  case kind*: PgOutputMessageKind
  of pomkBegin:
    begin*: BeginMessage
  of pomkCommit:
    commit*: CommitMessage
  of pomkOrigin:
    origin*: OriginMessage
  of pomkRelation:
    relation*: RelationInfo
  of pomkType:
    typeMsg*: TypeMessage
  of pomkInsert:
    insert*: InsertMessage
  of pomkUpdate:
    update*: UpdateMessage
  of pomkDelete:
    delete*: DeleteMessage
  of pomkTruncate:
    truncate*: TruncateMessage
  of pomkMessage:
    message*: LogicalMessage
A decoded pgoutput plugin message. 
PgOutputMessageKind = enum
  pomkBegin, pomkCommit, pomkOrigin, pomkRelation, pomkType, pomkInsert,
  pomkUpdate, pomkDelete, pomkTruncate, pomkMessage
Message types within the pgoutput logical decoding plugin. 
PrimaryKeepalive = object
  walEnd*: Lsn               ## Current end of WAL on the server
  sendTime*: int64           ## Server send time (microseconds since PG epoch)
  replyRequested*: bool      ## Whether the server wants an immediate status reply
Keepalive message from the server. 
RelationCache = Table[int32, RelationInfo]
Cache of relation metadata received during replication. The server sends a Relation message before the first DML for each table in a transaction; clients must cache them. 
RelationColumn = object
  flags*: byte               ## Bit 0: part of replica identity key
  name*: string
  typeOid*: int32
  typeMod*: int32
A single column in a relation definition. 
RelationInfo = object
  relationId*: int32
  namespace*: string         ## Schema name
  name*: string              ## Table name
  replicaIdentity*: char     ## 'd' (default), 'n' (nothing), 'f' (full), 'i' (index)
  columns*: seq[RelationColumn]
Relation (table) metadata sent by pgoutput before DML events. 
ReplicationCallback = proc (msg: ReplicationMessage): Future[void] {....gcsafe.}
Callback invoked for each replication message during streaming. 
ReplicationMessage = object
  case kind*: ReplicationMessageKind
  of rmkXLogData:
    xlogData*: XLogData
  of rmkPrimaryKeepalive:
    keepalive*: PrimaryKeepalive
A single message received during replication streaming. 
ReplicationMessageKind = enum
  rmkXLogData, rmkPrimaryKeepalive
Replication message types (decoded from CopyData during streaming) 
ReplicationMode = enum
  rmDatabase, rmPhysical
Replication mode selected at connection time. rmDatabase sends replication=database (logical replication + ability to run SQL on the chosen database). rmPhysical sends replication=true (physical replication; no SQL on user databases). 
ReplicationSlotInfo = object
  slotName*: string
  consistentPoint*: Lsn      ## confirmed_flush_lsn (logical) or restart_lsn (physical)
  snapshotName*: string      ## Snapshot name (only available at CREATE time)
  outputPlugin*: string      ## Output plugin (only available at CREATE time)
  slotType*: string ## Slot type as reported by READ_REPLICATION_SLOT ("physical").
                    ## Empty for CREATE_REPLICATION_SLOT results, which do not return it.
  restartTli*: int64         ## Timeline ID associated with restart_lsn.
                             ## Only populated by READ_REPLICATION_SLOT; 0 when NULL or not applicable.
Information about a replication slot. 
SystemInfo = object
  systemId*: string
  timeline*: int32
  xLogPos*: Lsn
  dbName*: string
Result of IDENTIFY_SYSTEM command. 
TimelineHistory = object
  filename*: string          ## Timeline history file name (e.g. "00000002.history").
  content*: seq[byte]        ## Raw history file content.
Result of TIMELINE_HISTORY command. 
TruncateMessage = object
  options*: byte             ## Bit 0: CASCADE, bit 1: RESTART IDENTITY
  relationIds*: seq[int32]
Table truncation. 
TupleDataKind = enum
  tdkNull = 110,            ## NULL value
  tdkText = 116,            ## Text-formatted value
  tdkBinary = 98,           ## Binary-formatted value (protocol_version >= 2)
  tdkUnchanged = 117         ## TOAST value unchanged
Kind of a single field value in a pgoutput tuple. 
TupleField = object
  kind*: TupleDataKind
  data*: seq[byte]           ## Empty for null/unchanged
A single field value in a pgoutput tuple. 
TypeMessage = object
  typeId*: int32
  namespace*: string
  name*: string
Custom type definition. 
UpdateMessage = object
  relationId*: int32
  keyKind*: char ## 'K' if oldTuple holds only the replica identity key,
                 ## 'O' if it holds the full old row (REPLICA IDENTITY FULL),
                 ## '\0' when no old tuple is present.
  oldTuple*: seq[TupleField]
  newTuple*: seq[TupleField]
Row update. 
XLogData = object
  startLsn*: Lsn             ## Start LSN of the WAL data in this message
  walEnd*: Lsn ## Current end of WAL on the server at the time this message was sent.
               ## This is *not* the end of the WAL data contained in this message; it
               ## reflects how far WAL has advanced on the server and is informational.
               ## To acknowledge what was actually received, use ``receivedEndLsn``
               ## (``startLsn + data.len``), never ``walEnd`` — ``walEnd`` may be ahead
               ## of what this message contains.
  sendTime*: int64           ## Server send time (microseconds since PG epoch)
  data*: seq[byte]           ## Raw WAL data (plugin-dependent format)
WAL data payload from the server.

Consts

InvalidLsn = 0'u
Sentinel value representing an invalid or unset LSN.

Procs

proc `$`(lsn: Lsn): string {....raises: [], tags: [], forbids: [].}
Format an LSN as "X/Y" hex string. 
proc `<`(a, b: Lsn): bool {.borrow, ...raises: [], tags: [], forbids: [].}
proc `<=`(a, b: Lsn): bool {.borrow, ...raises: [], tags: [], forbids: [].}
proc `==`(a, b: Lsn): bool {.borrow, ...raises: [], tags: [], forbids: [].}
proc confirmedFlushLsn(conn: PgConnection): Lsn {.inline, ...raises: [], tags: [],
    forbids: [].}
Confirmed flush LSN for current stream, or InvalidLsn outside stream. 
proc confirmFlushed(conn: PgConnection; lsn: Lsn): bool {.
    ...raises: [PgStateError, PgConnectionError], tags: [], forbids: [].}
Confirm WAL up to lsn as durable. Clamped to received WAL, monotonic. Returns true if advanced. Must be in csReplicating. 
proc connectReplication(config: ConnConfig; mode: ReplicationMode = rmDatabase): Future[
    PgConnection] {....raises: [Exception, ValueError, PgConfigError,
                             PgConnectionError, CatchableError],
                    tags: [RootEffect, WriteIOEffect, TimeEffect], forbids: [].}
Connect with replication param. rmPhysical allows only replication cmds. 
proc connectReplication(dsn: string; mode: ReplicationMode = rmDatabase): Future[
    PgConnection] {....raises: [PgConfigError, ValueError, Exception,
                             PgConnectionError, CatchableError], tags: [
    ReadIOEffect, RootEffect, WriteIOEffect, TimeEffect], forbids: [].}
DSN-string variant of connectReplication. See the ConnConfig overload for the meaning of mode. 
proc createReplicationSlot(conn: PgConnection; slotName: string;
                           plugin: string = "pgoutput"; temporary: bool = false;
                           timeout: async_backend.Duration = ZeroDuration): Future[
    ReplicationSlotInfo] {....stackTrace: false, raises: [Exception, ValueError,
    CatchableError, PgConnectionError, PgTypeError, PgError, PgProtocolError],
                           tags: [RootEffect, TimeEffect], forbids: [].}

Create a logical replication slot. Returns slot info including the consistent point LSN.

On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket).

proc currentPgTimestamp(): int64 {....raises: [], tags: [TimeEffect], forbids: [].}
Current time as microseconds since the PostgreSQL epoch (2000-01-01 UTC). 
proc decodePgOutput(msg: XLogData): PgOutputMessage {....raises: [PgProtocolError],
    tags: [], forbids: [].}
Convenience: decode the pgoutput message from an XLogData's data field. 
proc dropReplicationSlot(conn: PgConnection; slotName: string;
                         wait: bool = false;
                         timeout: async_backend.Duration = ZeroDuration): Future[
    void] {....stackTrace: false, raises: [Exception, ValueError, CatchableError],
            tags: [RootEffect, TimeEffect], forbids: [].}

Drop a replication slot.

On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket).

proc hasOldTuple(msg: UpdateMessage): bool {.inline, ...raises: [], tags: [],
    forbids: [].}
True if the update carries an old tuple (replica identity key or full row). 
proc identifySystem(conn: PgConnection;
                    timeout: async_backend.Duration = ZeroDuration): Future[
    SystemInfo] {....stackTrace: false, raises: [Exception, ValueError,
    CatchableError, PgConnectionError, PgTypeError, PgError, PgProtocolError],
                  tags: [RootEffect, TimeEffect], forbids: [].}

Execute IDENTIFY_SYSTEM and return system identification info.

On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket).

proc parseLsn(s: string): Lsn {....raises: [PgTypeError, PgTypeError], tags: [],
                                forbids: [].}
Parse an LSN from "X/Y" hex string. Converts a malformed value (wrong shape, non-hex halves, or a half wider than 32 bits) into PgTypeError so callers stay under the except PgError contract, mirroring parseTimelineId. 
proc parsePgOutputMessage(data: openArray[byte]): PgOutputMessage {.
    ...raises: [PgProtocolError], tags: [], forbids: [].}
Decode a pgoutput logical decoding message from raw WAL bytes. 
proc parseReplicationMessage(copyData: sink seq[byte]): ReplicationMessage {.
    ...raises: [PgProtocolError], tags: [], forbids: [].}
Parse a CopyData payload into a ReplicationMessage. Takes ownership of copyData so the XLogData path can reuse the incoming buffer for xlogData.data instead of slicing into a fresh allocation. 
proc parseTimelineId(s: string): int32 {....raises: [PgTypeError], tags: [],
    forbids: [].}
Parse the timeline id from an IDENTIFY_SYSTEM result row (text format). Converts a non-numeric value and an out-of-int32-range value into PgTypeError so callers stay under the except PgError contract. Range-check before narrowing: a bare parseInt(...).int32 would raise RangeDefect (a Defect, outside PgError) on an out-of-range value. 
proc readReplicationSlot(conn: PgConnection; slotName: string;
                         timeout: async_backend.Duration = ZeroDuration): Future[
    ReplicationSlotInfo] {....stackTrace: false, raises: [Exception, ValueError,
    CatchableError, PgConnectionError, PgTypeError, PgError, PgProtocolError],
                           tags: [RootEffect, TimeEffect], forbids: [].}

Read information about an existing physical replication slot.

Only physical slots are supported: the server rejects a logical slot with PgQueryError ("cannot use READ_REPLICATION_SLOT with logical replication slot") and reports a nonexistent slot as PgConnectionError. consistentPoint carries restart_lsn (InvalidLsn when the slot never reserved WAL) and restartTli its timeline (0 when NULL).

On timeout, the connection is retired (csClosed) unless the wire had settled (asyncdispatch always retires: the timed-out op stays on the socket).

proc receivedEndLsn(msg: XLogData): Lsn {....raises: [PgProtocolError], tags: [],
    forbids: [].}
End LSN of the WAL data actually contained in this message (startLsn + len(data)). Use this when acknowledging received data via sendStandbyStatus; do not use walEnd, which is the server's current WAL position and may point past data this message does not carry. 
proc sendCopyData(conn: PgConnection; data: openArray[byte]): Future[void] {....raises: [
    PgStateError, PgConnectionError, PgTypeError, PgProtocolError, Exception,
    CancelledError, CatchableError], tags: [RootEffect], forbids: [].}
Send CopyData during csReplicating. Raises PgStateError (not replicating) / PgConnectionError (connection lost) / PgTypeError synchronously before first suspension. data is encoded into the frame there too, so the caller's buffer need not outlive the returned Future. 
proc sendStandbyStatus(conn: PgConnection; receiveLsn: Lsn;
                       flushLsn: Lsn = InvalidLsn; applyLsn: Lsn = InvalidLsn;
                       replyRequested: bool = false): Future[void] {.
    ...stackTrace: false, raises: [Exception, PgStateError, PgConnectionError,
                                CancelledError, CatchableError, ValueError],
    tags: [RootEffect, TimeEffect], forbids: [].}
Send Standby Status Update. InvalidLsn defaults up to receiveLsn. Raises PgStateError unless the connection is csReplicating, or PgConnectionError when the connection was lost. 
proc startPhysicalReplication(conn: PgConnection; startLsn: Lsn;
                              slotName: string = ""; timeline: int32 = 0;
                              autoKeepaliveReply: bool = true; statusInterval: async_backend.Duration = ZeroDuration;
                              callback: ReplicationCallback): Future[void] {.
    ...stackTrace: false, raises: [Exception, PgStateError, PgConnectionError,
                                PgTypeError, PgProtocolError, CancelledError,
                                CatchableError, ValueError, PgQueryError,
                                AsyncTimeoutError],
    tags: [RootEffect, TimeEffect], forbids: [].}

Physical replication streaming. Callback per message, raw WAL in XLogData. Raises PgConnectionError (closed) / PgStateError (busy) unless csReady. Error handling matches startReplication: a callback exception or any other mid-stream failure poisons the connection (marked closed) and propagates, so reconnect and resume from the last LSN tracked.

slotName = "" streams without a slot. A non-zero timeline is appended as TIMELINE n, so the server aborts the stream if it advanced past that timeline. statusInterval behaves as on startReplication.

On a timeline switch the server may send a result set describing the next timeline between CopyDone and ReadyForQuery; this proc drains and discards it — re-issue IDENTIFY_SYSTEM if you need that information.

Synchronous standbys: the flush LSN governs how much WAL the primary may recycle, so a standby in synchronous_standby_names that relies on the auto-reply must call confirmFlushed (or reply manually) or the primary's COMMITs block waiting on a flush position that never advances.

proc startReplication(conn: PgConnection; slotName: string;
                      startLsn: Lsn = InvalidLsn;
                      options: seq[(string, string)] = @[];
                      autoKeepaliveReply: bool = true;
                      statusInterval: async_backend.Duration = ZeroDuration;
                      callback: ReplicationCallback): Future[void] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgStateError,
                                PgConnectionError, PgTypeError, PgProtocolError,
                                CancelledError, CatchableError, PgQueryError,
                                AsyncTimeoutError],
    tags: [RootEffect, TimeEffect], forbids: [].}

Begin logical replication. Callback invoked per message. Use confirmFlushed for flush tracking; or set autoKeepaliveReply=false and use sendStandbyStatus manually.

Returns on server CopyDone or connection close. To stop from the client side, call stopReplication from the callback (or a concurrent task).

Errors poison connection. Track LSN for resume. A failing auto-reply propagates too, and the callback is not invoked for that keepalive. Option values are passed unquoted and single-quoted when building the command (keys stay identifier-validated). An empty value means a flag-only option (binary rather than binary ''). Raises PgConnectionError (closed) / PgStateError (busy) unless csReady, and ValueError for a proto_version other than 1 in options (the value must be the unquoted string "1", an empty one included): the bundled pgoutput decoder supports v1 only. Any value already wrapped in quotes raises ValueError too, whatever its key, so the verbatim-options spelling cannot silently name a publication 'my_pub' or send a thrice-quoted binary flag the plugin rejects mid-stream. An empty publication_names and a value containing a NUL byte are rejected the same way. publication_names without an explicit proto_version adds proto_version '1' to the generated command, so a server-side default bump cannot outrun that decoder.

statusInterval (ZeroDuration = off) sends a proactive Standby Status Update at least that often — receive = highest received, flush/apply = confirmFlushed — so the slot advances on a server that never requests a reply (wal_sender_timeout = 0). Honoured only with autoKeepaliveReply; under asyncdispatch it fires only while messages are flowing, so a fully idle stream sends nothing until the next message.

Synchronous standbys: the auto-reply reports receive and flush/apply separately, so a consumer in synchronous_standby_names that never calls confirmFlushed keeps wal_sender_timeout reset via the receive field yet never advances flush — the primary's COMMITs then block indefinitely.

proc stopReplication(conn: PgConnection): Future[void] {....stackTrace: false, raises: [
    Exception, PgStateError, PgConnectionError, CancelledError, CatchableError,
    ValueError], tags: [RootEffect, TimeEffect], forbids: [].}
Terminate replication. Flushes confirmed position before CopyDone. Raises PgStateError unless the connection is csReplicating, or PgConnectionError when the connection was lost. 
proc timelineHistory(conn: PgConnection; timeline: int32;
                     timeout: async_backend.Duration = ZeroDuration): Future[
    TimelineHistory] {....stackTrace: false, raises: [Exception, ValueError,
    CatchableError, PgConnectionError, PgTypeError, PgError, PgProtocolError],
                       tags: [RootEffect, TimeEffect], forbids: [].}
Execute TIMELINE_HISTORY. Raises ValueError if timeline <=0. On timeout the connection is marked csClosed (protocol out of sync), so a caller catching PgTimeoutError must reconnect, not retry in place. 
proc toString(field: TupleField): string {....raises: [PgProtocolError], tags: [],
    forbids: [].}
Convert a TupleField's data to a string by copying the bytes.

Templates

template makeReplicationCallback(body: untyped): ReplicationCallback

Create a ReplicationCallback that works with both asyncdispatch and chronos. Inside body, the current message is available as msg: ReplicationMessage.

Kept module-local: routing this through a shared template with an untyped/typedesc param for the parameter type trips asyncdispatch's {.async.} macro ("cannot use symbol of kind 'func' as a 'param'").

template toInt64(lsn: Lsn): int64
Get the LSN as int64 (for wire protocol encoding). 
template toUInt64(lsn: Lsn): uint64
Get the raw uint64 value of an LSN.

Made with Nim. Generated: 2026-09-11 08:44:52 UTC