async_postgres/pg_connection/buffer_io

Search:
Group by:

Transport-layer buffering and message I/O.

  • recvBuf/sendBuf management (compact, fill, send)
  • Synchronous backend-message parsing (nextMessage) and the async wrapper recvMessage
  • Notification/Notice dispatch (called from nextMessage)
  • Transport teardown (closeTransport)
  • TCP keepalive / TCP_NODELAY socket options
  • Host helpers (isUnixSocket, unixSocketPath, getHosts)
  • makeCopyOutCallback / makeCopyInCallback cross-backend templates

The host helpers and makeCopy* templates are re-exported through pg_connection.nim; the transport buffering 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; what it re-exports is the supported surface (see tests/api_surface.golden).

Types

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 compactRecvBuf(conn: PgConnection) {.inline, ...raises: [], tags: [],
    forbids: [].}
Compact recvBuf (caller checks csClosed). Only safe before reading new data from the socket: it moves bytes an in-flight read still points at. 
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 dispatchNotice(conn: PgConnection; msg: BackendMessage) {....raises: [],
    tags: [RootEffect], forbids: [].}
proc dispatchNotification(conn: PgConnection; msg: BackendMessage) {....raises: [],
    tags: [RootEffect], forbids: [].}
proc enqueueNotification(conn: PgConnection; notif: Notification) {....raises: [],
    tags: [RootEffect], forbids: [].}
Enqueue under notifyMaxQueue (<=0 = unbounded); drop oldest on overflow. 
proc fillRecvBuf(conn: PgConnection; timeout: Duration = ZeroDuration): Future[
    void] {....stackTrace: false, raises: [Exception, PgStateError,
                                        PgConnectionError, AsyncTimeoutError,
                                        CancelledError, CatchableError],
            tags: [RootEffect, TimeEffect], forbids: [].}
Read into recvBuf. AsyncTimeoutError: caller handles state; other errors → csClosed + raiseTransportFailure. 
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. 
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).

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 reclaimHandoff(conn: PgConnection) {....raises: [], tags: [], forbids: [].}
Requeue a handoff whose waiter will never claim it, so an abandoned frame cannot make the notification unreachable. 
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 requeueHandoff(conn: PgConnection; notif: Notification) {....raises: [],
    tags: [], forbids: [].}
Requeue an unconsumed handoff at the front. 
proc resetWireState(conn: PgConnection) {....raises: [], tags: [], forbids: [].}

Forget what the wire's previous life left behind: the buffered bytes on both sides and the replies the old backend owed.

Sole owner of that reset: a stale count carried onto a fresh backend would dial a CancelRequest at an unrelated PID and retire a healthy connection.

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 socketHasFin(conn: PgConnection): bool {....raises: [], tags: [], forbids: [].}
POSIX half-open probe (MSG_PEEK): true if FIN/RST observed; false otherwise or unavailable. 
proc socketHasPendingData(conn: PgConnection): bool {....raises: [], tags: [],
    forbids: [].}
True if kernel has readable bytes (pre-TLS injection check; kernel buffer only). 
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 pumpUntilReady(conn: PgConnection; body: untyped; readyBody: untyped) {.
    dirty.}
Bare pump (skipDataRow=true; for callers that discard rows). 
template pumpUntilReady(conn: PgConnection; resultData: untyped; onRow: untyped;
                        onRowErr: untyped; body: untyped; readyBody: untyped) {.
    dirty.}
Streaming pump (onRow per row; first error in onRowErr). 
template pumpUntilReady(conn: PgConnection; resultData: untyped;
                        rowCountPtr: untyped; body: untyped; readyBody: untyped) {.
    dirty.}
Pump until ReadyForQuery; pumpMsg/queryError injected into body/readyBody.

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