async_postgres/pg_protocol

Search:
Group by:

Types

BackendMessage = object
  case kind*: BackendMessageKind
  of bmkAuthenticationOk, bmkAuthenticationCleartextPassword:
    nil
  of bmkAuthenticationMD5Password:
    md5Salt*: array[4, byte]
  of bmkAuthenticationSASL:
    saslMechanisms*: seq[string]
  of bmkAuthenticationSASLContinue:
    saslData*: seq[byte]
  of bmkAuthenticationSASLFinal:
    saslFinalData*: seq[byte]
  of bmkBackendKeyData:
    backendPid*: int32
    backendSecretKey*: int32
  of bmkBindComplete, bmkCloseComplete, bmkNoData, bmkEmptyQueryResponse,
     bmkParseComplete, bmkPortalSuspended, bmkCopyDone:
    nil
  of bmkCommandComplete:
    commandTag*: string
  of bmkCopyInResponse, bmkCopyOutResponse, bmkCopyBothResponse:
    copyFormat*: CopyFormat
    copyColumnFormats*: seq[int16]
  of bmkCopyData:
    copyData*: seq[byte]
  of bmkDataRow:
    columns*: seq[Option[seq[byte]]]
  of bmkErrorResponse:
    errorFields*: seq[ErrorField]
  of bmkNoticeResponse:
    noticeFields*: seq[ErrorField]
  of bmkNotificationResponse:
    notifPid*: int32
    notifChannel*: string
    notifPayload*: string
  of bmkParameterDescription:
    paramTypeOids*: seq[int32]
  of bmkParameterStatus:
    paramName*: string
    paramValue*: string
  of bmkReadyForQuery:
    txStatus*: TransactionStatus
  of bmkRowDescription:
    fields*: seq[FieldDescription]
Parsed message from the PostgreSQL backend. Variant type keyed by kind. 
BackendMessageKind = enum
  bmkAuthenticationOk, bmkAuthenticationCleartextPassword,
  bmkAuthenticationMD5Password, bmkAuthenticationSASL,
  bmkAuthenticationSASLContinue, bmkAuthenticationSASLFinal, bmkBackendKeyData,
  bmkBindComplete, bmkCloseComplete, bmkCommandComplete, bmkCopyInResponse,
  bmkCopyOutResponse, bmkCopyBothResponse, bmkCopyData, bmkCopyDone, bmkDataRow,
  bmkEmptyQueryResponse, bmkErrorResponse, bmkNoData, bmkNoticeResponse,
  bmkNotificationResponse, bmkParameterDescription, bmkParameterStatus,
  bmkParseComplete, bmkPortalSuspended, bmkReadyForQuery, bmkRowDescription
Message types received from server to client. 
CopyFormat = enum
  cfText = 0, cfBinary = 1
Wire format for COPY operations. 
DescribeKind = enum
  dkPortal = 80, dkStatement = 83
Target of a Describe or Close message. 
FieldDescription = object
  name*: string
  tableOid*: int32
  columnAttrNum*: int16
  typeOid*: int32
  typeSize*: int16
  typeMod*: int32
  formatCode*: int16
Column metadata from a RowDescription message. 
FrontendMessageKind = enum
  fmkStartup, fmkSSLRequest, fmkPassword, fmkSASLInitialResponse,
  fmkSASLResponse, fmkQuery, fmkParse, fmkBind, fmkDescribe, fmkExecute,
  fmkClose, fmkSync, fmkFlush, fmkTerminate, fmkCopyData, fmkCopyDone,
  fmkCopyFail
Message types sent from client to server. 
ParseResult = object
  state*: ParseState
  message*: BackendMessage
Result of parsing bytes from the receive buffer. 
ParseState = enum
  psComplete, psIncomplete, psDataRow ## DataRow parsed in-place into RowData; no BackendMessage constructed
Row = object
Lightweight view into a single row within a RowData buffer. 
RowData = ref object
  buf*: seq[byte]            ## All column data concatenated
  cellIndex*: seq[int32]     ## [off, len, off, len, ...] per cell; len=-1 = NULL
  numCols*: int16
  colFormats*: seq[int16]    ## Per-column format codes (0=text, 1=binary)
  colTypeOids*: seq[int32]   ## Per-column type OIDs for binary→text conversion
  fields*: seq[FieldDescription] ## Column metadata for name-based access
  colMap*: Table[string, int] ## Cached name→index mapping (lazily built)
Flat buffer holding all row data for a query result. 
TransactionStatus = enum
  tsInFailedTransaction = 69, tsIdle = 73, tsInTransaction = 84
Server transaction state reported in ReadyForQuery.

Consts

BinarySafeOids = [16'i32, 17, 20, 21, 23, 25, 114, 142, 143, 600, 601, 602, 603,
                  604, 628, 629, 650, 651, 700, 701, 718, 719, 774, 775, 790,
                  791, 829, 869, 1000, 1001, 1005, 1007, 1009, 1015, 1016, 1017,
                  1018, 1019, 1020, 1021, 1022, 1027, 1040, 1041, 1043, 1082,
                  1083, 1114, 1115, 1182, 1183, 1184, 1185, 1186, 1187, 1231,
                  1266, 1270, 1560, 1561, 1562, 1563, 1700, 2950, 2951, 3614,
                  3615, 3643, 3645, 3802, 3807, 3904, 3905, 3906, 3907, 3908,
                  3909, 3910, 3911, 3912, 3913, 3926, 3927, 4451, 4532, 4533,
                  4534, 4535, 4536, 6150, 6151, 6152, 6153, 6155, 6157]
copyDoneMsg = [99'u, 0'u8, 0'u8, 0'u8, 4'u8]
Pre-built CopyDone message bytes. 
DefaultMaxBackendMessageLen = 1073741824
Default upper bound on a single backend message (header + body), 1 GiB. Prevents a malicious or broken server from causing unbounded recv-buffer growth (and OOM) by advertising an int32-max length. PostgreSQL itself bounds individual values by MaxAllocSize (~1 GiB - 1), so legitimate traffic stays well below this cap. 
flushMsg = [72'u, 0'u8, 0'u8, 0'u8, 4'u8]
Pre-built Flush message bytes. 
pgCopyBinaryHeader: array[19, byte] = [80'u, 71'u, 67'u, 79'u, 80'u, 89'u, 10'u,
                                       0xFF'u8, 13'u, 10'u, 0x00'u8, 0x00'u8,
                                       0x00'u8, 0x00'u8, 0x00'u8, 0x00'u8,
                                       0x00'u8, 0x00'u8, 0x00'u8]
pgCopyBinaryTrailer: array[2, byte] = [0xFF'u8, 0xFF'u8]
PGCOPY binary format trailer (int16(-1) sentinel). 
syncMsg = [83'u, 0'u8, 0'u8, 0'u8, 4'u8]
Pre-built Sync message bytes.

Procs

proc addBind(buf: var seq[byte]; portalName: string; stmtName: string;
             paramFormats: openArray[int16];
             paramValues: openArray[Option[seq[byte]]];
             resultFormats: openArray[int16] = []) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Append a Bind message to the buffer (extended query protocol). 
proc addBindRaw(buf: var seq[byte]; portalName: string; stmtName: string;
                paramFormats: openArray[int16]; paramData: openArray[byte];
                paramRanges: openArray[tuple[off: int32, len: int32]];
                resultFormats: openArray[int16] = []) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}

Append a Bind message built from a raw byte buffer and offset/length ranges. Each parameter is described by (off, len): len == -1 encodes NULL; any other len reads paramData[off ..< off + len]. Lets callers write payloads straight into a single owned buffer without constructing Option[seq[byte]] per parameter.

Each range must satisfy one of: len == -1 (NULL), or len >= 0 with 0 <= off and off + len <= paramData.len. Invalid ranges raise ValueError — the check is always active so callers cannot silently trigger an out-of-bounds copyMem in release builds.

proc addClose(buf: var seq[byte]; kind: DescribeKind; name: string) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Append a Close message to the buffer (portal or statement). 
proc addCopyBinaryHeader(buf: var seq[byte]) {....raises: [], tags: [], forbids: [].}
Append the PostgreSQL binary COPY header (signature + flags + extension area). 
proc addCopyBinaryTrailer(buf: var seq[byte]) {....raises: [], tags: [],
    forbids: [].}
Append the binary COPY trailer (int16 = -1). 
proc addCopyDone(buf: var seq[byte]) {....raises: [], tags: [], forbids: [].}
Append a CopyDone message to the buffer. 
proc addCopyFieldBool(buf: var seq[byte]; val: bool) {....raises: [], tags: [],
    forbids: [].}
Append a boolean field in binary COPY format. 
proc addCopyFieldFloat32(buf: var seq[byte]; val: float32) {....raises: [],
    tags: [], forbids: [].}
Append a float32 field in binary COPY format. 
proc addCopyFieldFloat64(buf: var seq[byte]; val: float64) {....raises: [],
    tags: [], forbids: [].}
Append a float64 field in binary COPY format. 
proc addCopyFieldInt16(buf: var seq[byte]; val: int16) {....raises: [], tags: [],
    forbids: [].}
Append an int16 field in binary COPY format. 
proc addCopyFieldInt32(buf: var seq[byte]; val: int32) {....raises: [], tags: [],
    forbids: [].}
Append an int32 field in binary COPY format. 
proc addCopyFieldInt64(buf: var seq[byte]; val: int64) {....raises: [], tags: [],
    forbids: [].}
Append an int64 field in binary COPY format. 
proc addCopyFieldNull(buf: var seq[byte]) {....raises: [], tags: [], forbids: [].}
Append a NULL field in binary COPY format. 
proc addCopyFieldString(buf: var seq[byte]; val: string) {....raises: [ValueError],
    tags: [], forbids: [].}
Append a string field in binary COPY format. 
proc addCopyFieldText(buf: var seq[byte]; val: openArray[byte]) {.
    ...raises: [ValueError], tags: [], forbids: [].}
Append a raw byte field in binary COPY format. 
proc addCopyTupleStart(buf: var seq[byte]; numCols: int16) {....raises: [],
    tags: [], forbids: [].}
Start a new tuple in binary COPY format with the given column count. 
proc addCount16(buf: var seq[byte]; n: int; what: string) {.
    ...raises: [ValueError], tags: [], forbids: [].}

Append an Int16 count field, rejecting counts that overflow the wire's signed 16-bit range. Without this guard the int16(n) conversion raises an uncatchable RangeDefect on default builds, or silently wraps to a bogus (often negative) count that desyncs the protocol stream under -d:danger. The check is always active so callers get a catchable ValueError instead.

ValueError is used here (and in addLen32) because the count is supplied directly by the caller; message-level length overflows detected after the message has been assembled raise PgProtocolError instead.

proc addCString(buf: var seq[byte]; s: string) {....raises: [ValueError], tags: [],
    forbids: [].}

Append a null-terminated C string to the buffer.

Raises ValueError if s contains an embedded NUL byte. PostgreSQL protocol fields are NUL-terminated, so an embedded \0 would split the value into two fields on the server side — at best causing a desync (invalid message format), at worst allowing startup-parameter injection through the StartupMessage K/V stream.

proc addDescribe(buf: var seq[byte]; kind: DescribeKind; name: string) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Append a Describe message to the buffer (portal or statement). 
proc addExecute(buf: var seq[byte]; portalName: string; maxRows: int32 = 0) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Append an Execute message to the buffer. maxRows of 0 means unlimited. 
proc addFlush(buf: var seq[byte]) {....raises: [], tags: [], forbids: [].}
Append a Flush message to the buffer. 
proc addInt16(buf: var seq[byte]; val: int16) {....raises: [], tags: [],
    forbids: [].}
Append a 16-bit integer in big-endian format to the buffer. 
proc addInt32(buf: var seq[byte]; val: int32) {....raises: [], tags: [],
    forbids: [].}
Append a 32-bit integer in big-endian format to the buffer. 
proc addInt64(buf: var seq[byte]; val: int64) {....raises: [], tags: [],
    forbids: [].}
Append a 64-bit integer in big-endian format to the buffer. 
proc addLen32(buf: var seq[byte]; n: int; what: string) {....raises: [ValueError],
    tags: [], forbids: [].}

Append an Int32 length field, rejecting payloads that overflow the wire's signed 32-bit length. Like addCount16, this turns the otherwise uncatchable RangeDefect (or a wrapped, often-negative length that desyncs the stream under -d:danger) into a catchable ValueError raised before the oversized payload is appended.

ValueError is used here (and in addCount16) because the length is supplied directly by the caller; message-level length overflows detected after the message has been assembled raise PgProtocolError instead.

proc addParse(buf: var seq[byte]; stmtName: string; sql: string;
              paramTypeOids: openArray[int32] = []) {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Append a Parse message to the buffer (extended query protocol). 
proc addSync(buf: var seq[byte]) {....raises: [], tags: [], forbids: [].}
Append a Sync message to the buffer. 
proc buildResultFormats(fields: openArray[FieldDescription]): seq[int16] {.
    ...raises: [], tags: [], forbids: [].}
Build per-column binary format codes: 1 for known safe types, 0 for others. 
proc clone(row: Row): Row {....raises: [], tags: [], forbids: [].}
Return a deep copy of row with an independent RowData backing buffer containing only this single row. Use this to retain rows from a queryEach callback beyond the callback's lifetime — the original buffer is reused for subsequent rows and would otherwise be overwritten. 
func data(row: Row): RowData {....raises: [], tags: [], forbids: [].}
The underlying RowData buffer. 
proc decodeCString(buf: openArray[byte]; offset: int): (string, int) {.
    ...raises: [PgProtocolError], tags: [], forbids: [].}
Decode a null-terminated string at the given offset. Returns (string, bytes consumed). 
proc decodeInt16(buf: openArray[byte]; offset: int): int16 {.inline, ...raises: [],
    tags: [], forbids: [].}
Decode a 16-bit integer from big-endian bytes at the given offset. 
proc decodeInt32(buf: openArray[byte]; offset: int): int32 {.inline, ...raises: [],
    tags: [], forbids: [].}
Decode a 32-bit integer from big-endian bytes at the given offset. 
proc decodeInt64(buf: openArray[byte]; offset: int): int64 {.inline, ...raises: [],
    tags: [], forbids: [].}
Decode a 64-bit integer from big-endian bytes at the given offset. 
proc encodeBind(portalName: string; stmtName: string;
                paramFormats: openArray[int16];
                paramValues: openArray[Option[seq[byte]]];
                resultFormats: openArray[int16] = []): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a standalone Bind message. 
proc encodeCancelRequest(pid: int32; secretKey: int32): seq[byte] {....raises: [],
    tags: [], forbids: [].}
Encode a CancelRequest message to abort a running query. 
proc encodeClose(kind: DescribeKind; name: string): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a standalone Close message. 
proc encodeCopyData(buf: var seq[byte]; data: openArray[byte]) {.
    ...raises: [ValueError], tags: [], forbids: [].}
Encode a CopyData message, appending to buf. Single setLen for header + payload to minimize bounds checks. The Int32 length field covers itself plus the payload, so reject payloads that would overflow it (a wrapped length desyncs the stream) before any allocation, matching the addLen32 guard used by the other encoders. Like addLen32, the payload length comes from the caller, so an overflow raises ValueError rather than PgProtocolError. 
proc encodeCopyDone(): seq[byte] {....raises: [], tags: [], forbids: [].}
Encode a standalone CopyDone message. 
proc encodeCopyFail(errorMsg: string): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a CopyFail message to abort a COPY operation with an error. 
proc encodeDescribe(kind: DescribeKind; name: string): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a standalone Describe message. 
proc encodeExecute(portalName: string; maxRows: int32 = 0): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a standalone Execute message. 
proc encodeFlush(): seq[byte] {....raises: [], tags: [], forbids: [].}
Encode a standalone Flush message. 
proc encodeInt16(val: int16): array[2, byte] {.inline, ...raises: [], tags: [],
    forbids: [].}
Encode a 16-bit integer as big-endian bytes. 
proc encodeInt32(val: int32): array[4, byte] {.inline, ...raises: [], tags: [],
    forbids: [].}
Encode a 32-bit integer as big-endian bytes. 
proc encodeParse(stmtName: string; sql: string;
                 paramTypeOids: openArray[int32] = []): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a standalone Parse message. 
proc encodePassword(password: string): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a PasswordMessage for cleartext or MD5 authentication. Pre-allocates the buffer so internal add calls do not realloc and leave residual copies of the password in freed heap memory. 
proc encodeQuery(sql: string): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a simple Query message. 
proc encodeSASLInitialResponse(mechanism: string; data: seq[byte]): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a SASLInitialResponse message with the chosen mechanism and client-first data. 
proc encodeSASLResponse(data: seq[byte]): seq[byte] {....raises: [PgProtocolError],
    tags: [], forbids: [].}
Encode a SASLResponse message with client-final data. 
proc encodeSSLRequest(): seq[byte] {....raises: [], tags: [], forbids: [].}
Encode an SSLRequest message (magic number 80877103). 
proc encodeStandbyStatusUpdate(receiveLsn, flushLsn, applyLsn, sendTime: int64;
                               reply: byte): seq[byte] {....raises: [], tags: [],
    forbids: [].}
Encode a Standby Status Update as a CopyData message. The inner format is: byte 'r' + receiveLsn(8) + flushLsn(8) + applyLsn(8) + sendTime(8) + reply(1) = 34 bytes payload. 
proc encodeStartup(user: string; database: string;
                   extraParams: openArray[(string, string)] = []): seq[byte] {.
    ...raises: [ValueError, PgProtocolError], tags: [], forbids: [].}
Encode a StartupMessage (protocol v3.0) with user, database, and extra parameters. Raises ValueError for invalid caller-supplied values (e.g. embedded NUL bytes) and PgProtocolError if the assembled message exceeds the Int32 maximum length. 
proc encodeSync(): seq[byte] {....raises: [], tags: [], forbids: [].}
Encode a standalone Sync message. 
proc encodeTerminate(): seq[byte] {....raises: [], tags: [], forbids: [].}
Encode a Terminate message to close the connection. 
proc formatError(fields: seq[ErrorField]): string {....raises: [], tags: [],
    forbids: [].}
Format error fields into a human-readable error message with severity, SQLSTATE, detail, and hint. 
func initRow(data: RowData; rowIdx: int32): Row {....raises: [], tags: [],
    forbids: [].}
Create a Row view into the given RowData at the specified row index. 
func isBinarySafeOid(oid: int32): bool {....raises: [], tags: [], forbids: [].}
Check if a type OID can be safely requested in binary format. 
proc newRowData(numCols: int16; colFormats: seq[int16] = @[];
                colTypeOids: seq[int32] = @[]): RowData {....raises: [], tags: [],
    forbids: [].}
Create a new RowData flat buffer for accumulating DataRow messages. 
proc parseBackendMessage(buf: openArray[byte]; consumed: var int;
                         rowData: RowData = nil;
                         maxLen: int = DefaultMaxBackendMessageLen): ParseResult {.
    ...raises: [PgProtocolError], tags: [], forbids: [].}
Parse a single backend message from buf. On success, sets consumed to the number of bytes used. The caller is responsible for discarding those bytes from the buffer. A message whose declared length exceeds maxLen is rejected with PgProtocolError before any allocation, capping recv-buffer growth. maxLen <= 0 disables the cap (intended for tests); production callers should resolve the default via ConnConfig.maxMessageSize and effectiveMaxMessageSize. 
proc parseDataRowInto(body: openArray[byte]; rd: RowData) {.
    ...raises: [PgProtocolError], tags: [], forbids: [].}
Parse a DataRow message body directly into a RowData flat buffer. Column data is appended to rd.buf and (offset, length) pairs to rd.cellIndex. Uses a single bulk copyMem for the entire row payload, then walks the copied buffer to build cellIndex entries. 
proc patchLen(buf: var seq[byte]; offset: int = 1) {....raises: [PgProtocolError],
    tags: [], forbids: [].}
Patch the length placeholder at offset with buf.len minus the tag byte. Raises PgProtocolError when the assembled message exceeds the Int32 maximum; this is a protocol-level failure for an internally built message, distinct from the ValueError raised by addLen32/addCount16 for caller-supplied field values. 
proc patchMsgLen(buf: var seq[byte]; msgStart: int) {....raises: [PgProtocolError],
    tags: [], forbids: [].}
Patch the length field of a message starting at msgStart. Length = total message size minus the type byte. Raises PgProtocolError when the assembled message exceeds the Int32 maximum; this is a protocol-level failure for an internally built message, distinct from the ValueError raised by addLen32/addCount16 for caller-supplied field values. 
proc reuseRowData(rd: RowData; numCols: int16): RowData {....raises: [], tags: [],
    forbids: [].}
Create a new RowData that takes over the old buffer's capacity via move, without format metadata. 
proc reuseRowData(rd: RowData; numCols: int16; colFormats: sink seq[int16];
                  colTypeOids: sink seq[int32]): RowData {....raises: [], tags: [],
    forbids: [].}
Create a new RowData that takes over the old buffer's capacity via move. The old RowData (and any QueryResult still referencing it) is left intact. 
func rowIdx(row: Row): int32 {....raises: [], tags: [], forbids: [].}
The row index within the RowData buffer.

Made with Nim. Generated: 2026-06-26 11:39:17 UTC