async_postgres/pg_connection/buffer_io

Search:
Group by:

Transport-layer message I/O.

  • sendBuf management and raw sends
  • Synchronous backend-message parsing (nextMessage) and the async wrapper recvMessage
  • Background read watch for COPY IN early-error detection (RecvWatch)
  • Transport teardown (closeTransport)
  • TCP keepalive / TCP_NODELAY socket options
  • Host helpers (isUnixSocket, unixSocketPath, getHosts) and dialing (resolveTargets, dialTargets, dialServer, socketError, oneLine)
  • makeCopyOutCallback / makeCopyInCallback cross-backend templates

The receive buffer and its fills live in types with the private recvBuf / recvBufStart pair they move.

The host helpers and makeCopy* templates are re-exported through pg_connection.nim; the message-parsing and send machinery stays here for sibling modules and tests. Depends only on types.nim and the protocol/error/backend abstraction modules.

Internal module: not part of the public API. Import the pg_connection hub instead.

Types

Dialed = tuple[stream: DialStream, target: DialTarget]
A connected stream and the address it reached. 
DialStream = AsyncSocket
IdleProbe = enum
  ipQuiet,                  ## nothing arrived since and no FIN/RST observed
  ipActivity,               ## bytes arrived since, unread in `recvBuf` or the kernel
  ipDead                     ## FIN/RST observed
What idleProbe sees of a connection parked at ReadyForQuery. 
RecvWatch = ref object

Background read watch for unsolicited messages during send (at most one).

Settle it (take + await, or cancel) before reusing the normal recv path: an unsettled read shares recvBuf with whatever runs next.

Procs

proc cancel(w: RecvWatch) {....raises: [Exception], tags: [RootEffect], forbids: [].}
Abandon in-flight read (must raise/exit immediately after). 
proc closeTransport(conn: PgConnection): owned(Future[void]) {.
    ...stackTrace: false, raises: [Exception, LibraryError, SslError, ValueError],
    tags: [RootEffect], forbids: [].}

Close transport resources without sending Terminate.

Re-entrant: a racing second teardown awaits the first rather than returning early on the already-detached handles, so a resolved close() still means the fd and the backend session are released.

proc configureKeepalive(fd: posix.SocketHandle; config: ConnConfig) {.
    ...raises: [PgConnectionError], tags: [], forbids: [].}
Set TCP keepalive options on the socket. 
proc configureTcpNoDelay(fd: posix.SocketHandle) {....raises: [], tags: [],
    forbids: [].}
Disable Nagle's algorithm for low-latency sends. 
proc dialServer(host: string; port: int): Future[Dialed] {....raises: [Exception,
    ValueError, LibraryError, SslError, CatchableError, PgConnectionError,
    PgUnavailableError, OSError, IOError], tags: [RootEffect], forbids: [].}
Connect to the first address host resolves to that accepts. 
proc dialTargets(targets: seq[DialTarget]): Future[Dialed] {....stackTrace: false, raises: [
    Exception, ValueError, LibraryError, SslError, CatchableError,
    PgConnectionError], tags: [RootEffect], forbids: [].}
Connect to the first of targets that accepts, as libpq does. 
proc getHosts(config: ConnConfig): seq[HostEntry] {....raises: [], tags: [],
    forbids: [].}
Return the list of hosts to try. If hosts is populated, return it; otherwise synthesize a single entry from host/port. Each host goes through effectiveHost, so errors and traces show the name verified, not a 127.0.0.1 default left behind by a later hostaddr. 
proc idleProbe(conn: PgConnection): IdleProbe {....raises: [], tags: [],
    forbids: [].}
Non-consuming liveness probe for a connection parked at ReadyForQuery. A healthy idle session normally receives nothing, so any byte (a NOTIFY, or the FATAL/WARNING a closing backend sends ahead of its FIN) is activity that only a round trip can judge; the bytes are not interpreted. Bytes a lower layer already read off the socket stay out of sight: those the TLS layer took, and under chronos on Windows all of them, since its posted read takes them as they arrive. 
proc isConnected(conn: PgConnection): bool {....raises: [], tags: [], forbids: [].}
Transport present and no kernel FIN/RST observed (cheap, non-blocking; use ping for full check). 
proc isUnixSocket(host: string): bool {.inline, ...raises: [], tags: [],
                                        forbids: [].}
True if host represents a Unix socket directory (starts with '/'). Compatible with libpq behavior. 
proc nextMessage(conn: PgConnection; rowData: RowData = nil;
                 rowCount: ptr int32 = nil; onRow: RowCallback = nil;
                 onRowError: ptr ref CatchableError = nil;
                 skipDataRow: bool = false): Option[BackendMessage] {.
    ...raises: [PgProtocolError], tags: [RootEffect], forbids: [].}

Parse next message from recvBuf (none = incomplete). Dispatches notify/notice, consumes ParameterStatus/DataRow (streaming via onRow); skipDataRow avoids decode. Error → csClosed.

onRow requires onRowError: callback failures are deferred into that slot so the pump can drain to ReadyForQuery. Without a slot the first DataRow would dereference nil, so the missing slot is rejected up front with PgProtocolError (the connection is left open — unlike wire corruption, this is a caller bug, not a broken peer).

A DataRow with no rowData to land in is a PgProtocolError when the caller asked for rows (onRow or rowCount), skipDataRow or not: a described result always precedes its rows, and a row the caller cannot read must not be counted as if it had been. With no row sink at all the decoded message is returned instead; skipDataRow frames the row without decoding it.

proc oneLine(msg: string): string {....raises: [], tags: [], forbids: [].}
Collapse msg to one line, dropping the asyncdispatch traceback. Cuts at the traceback marker so server DETAIL/HINT lines survive, joined with " | ". 
proc pending(w: RecvWatch): bool {....raises: [], tags: [], forbids: [].}
Whether a background read is currently in flight. 
proc ready(w: RecvWatch): bool {....raises: [], tags: [], forbids: [].}
Whether the in-flight read has settled, so take + await will not block. A read that failed also reports ready; awaiting it then re-raises. 
proc rearm(w: RecvWatch; conn: PgConnection) {....raises: [Exception, PgStateError,
    PgConnectionError, AsyncTimeoutError, CancelledError, CatchableError],
    tags: [RootEffect, TimeEffect], forbids: [].}
Resume watching with a fresh background read. Only call once the previous read has been consumed (take + await), never while one is still in flight. 
proc recvMessage(conn: PgConnection; timeout = ZeroDuration;
                 rowData: RowData = nil; rowCount: ptr int32 = nil): Future[
    BackendMessage] {....stackTrace: false, raises: [Exception, ValueError,
    PgProtocolError, PgStateError, PgConnectionError, CancelledError,
    CatchableError, AsyncTimeoutError], tags: [RootEffect, TimeEffect],
                      forbids: [].}
Receive one message (nextMessage + fillRecvBuf); timeout → csClosed. 
proc resolveTargets(host: string; port: int): seq[DialTarget] {....raises: [
    PgConnectionError, PgUnavailableError, Exception, OSError, IOError],
    tags: [], forbids: [].}
What host names: its Unix socket, or each address it resolves to in the resolver's order. 
proc sendBufMsg(conn: PgConnection): Future[void] {....stackTrace: false, raises: [
    Exception, CancelledError, PgStateError, CatchableError, PgConnectionError],
    tags: [RootEffect], forbids: [].}
Send sendBuf (copied; safe to mutate after call); failure → csClosed. Books what the buffer leaves owed; see sendMsg. 
proc sendMsg(conn: PgConnection; data: seq[byte]): Future[void] {.
    ...stackTrace: false, raises: [Exception, CancelledError, PgStateError,
                                CatchableError, PgConnectionError],
    tags: [RootEffect], forbids: [].}
Send raw bytes; failure → csClosed. Books what they leave owed first; see noteWrite. 
proc sendRawBytes(socket: AsyncSocket; data: seq[byte]): Future[void] {.
    ...raises: [Exception, SslError, ValueError], tags: [RootEffect], forbids: [].}
Send seq[byte] via asyncdispatch socket. 
proc sendRawData(socket: AsyncSocket; p: pointer; len: int): Future[void] {.
    ...raises: [Exception, SslError, ValueError], tags: [RootEffect], forbids: [].}
Send raw bytes via asyncdispatch socket. Copies data into a string once. 
proc sendStagedBufMsg(conn: PgConnection): owned(Future[void]) {.
    ...stackTrace: false, raises: [Exception, CancelledError, PgStateError,
                                CatchableError, PgConnectionError, ValueError],
    tags: [RootEffect], forbids: [].}
sendBufMsg paired with stagePendingStmtCloses: drop the staged statement Closes only once the buffer is on the wire. 
proc sendStagedMsg(conn: PgConnection; data: seq[byte]): owned(Future[void]) {.
    ...stackTrace: false, raises: [Exception, CancelledError, PgStateError,
                                CatchableError, PgConnectionError, ValueError],
    tags: [RootEffect], forbids: [].}
sendMsg counterpart, for builds that assemble their own buffer. 
func shown(t: DialTarget): string {....raises: [], tags: [], forbids: [].}
t as error messages name it. 
proc socketError(e: ref CatchableError): ref PgConnectionError {....raises: [],
    tags: [], forbids: [].}
A failed connect or send, classified as a dial is. 
proc socketHasFin(conn: PgConnection): bool {....raises: [], tags: [], forbids: [].}
Half-open probe (MSG_PEEK): true if FIN/RST observed; false otherwise or unavailable. 
proc socketHasPendingData(conn: PgConnection): bool {....raises: [], tags: [],
    forbids: [].}
True if the kernel has readable bytes (pre-TLS injection check; kernel buffer only). Blind under chronos on Windows once reading has started: its posted read takes the bytes first. 
proc startRecvWatch(conn: PgConnection): RecvWatch {....raises: [Exception,
    PgStateError, PgConnectionError, AsyncTimeoutError, CancelledError,
    CatchableError], tags: [RootEffect, TimeEffect], forbids: [].}
Begin watching for an unsolicited backend message. The bytes are committed to recvBuf when the read completes; poll with ready, then take + await (immediate once ready) and parse with nextMessage. 
proc take(w: RecvWatch): Future[void] {....raises: [], tags: [], forbids: [].}
Surrender the in-flight read for the caller to await (immediate when ready). Clears the watch; the caller owns the returned Future. 
proc unixSocketPath(host: string; port: int): string {....raises: [], tags: [],
    forbids: [].}
Build the libpq-compatible Unix socket file path: {dir}/.s.PGSQL.{port}.

Templates

template makeCopyInCallback(body: untyped): CopyInCallback
Create a CopyInCallback that works with both asyncdispatch and chronos. body must evaluate to seq[byte]. Return an empty seq to signal completion.
var idx = 0
let rows = @["1\tAlice\n".toBytes(), "2\tBob\n".toBytes()]
let cb = makeCopyInCallback:
  if idx < rows.len:
    let chunk = rows[idx]
    inc idx
    chunk
  else:
    newSeq[byte]()
template makeCopyOutCallback(body: untyped): CopyOutCallback
Create a CopyOutCallback that works with both asyncdispatch and chronos. Inside body, the current chunk is available as data: sink seq[byte].
var chunks: seq[seq[byte]]
let cb = makeCopyOutCallback:
  chunks.add(data)
template posixFd(s: AsyncSocket): posix.SocketHandle
s's descriptor as posix's SocketHandle. 
template pumpUntilReady(conn: PgConnection;
                        resultData, onRow, onRowErr, body, readyBody: untyped)
Streaming pump (onRow per row; first error in onRowErr). 
template pumpUntilReady(conn: PgConnection;
                        resultData, rowCountPtr, body, readyBody: untyped)
Pump until ReadyForQuery; pumpMsg/queryError injected into body/readyBody. 
template pumpUntilReady(conn: PgConnection; body, readyBody: untyped)
Bare pump (skipDataRow=true; for callers that discard rows).

Made with Nim. Generated: 2026-10-23 17:09:00 UTC