Shared building blocks for pg_client submodules (transaction opts, inline params, recv loops).
Internal module: not part of the public API. Import the pg_client hub instead.
Types
AccessMode = enum amDefault, amReadWrite, amReadOnly
- PostgreSQL transaction access mode (read-write or read-only).
DeferrableMode = enum dmDefault, dmDeferrable, dmNotDeferrable
- PostgreSQL transaction deferrable mode (for serializable read-only transactions).
IsolationLevel = enum ilDefault, ilReadCommitted, ilRepeatableRead, ilSerializable, ilReadUncommitted ## PostgreSQL runs it as ``READ COMMITTED`` (no dirty reads).
- PostgreSQL transaction isolation level.
OpFacts = object bound*: bool ## ``BindComplete`` arrived: a later error is the statement's own. parsed*: bool ## ``ParseComplete`` arrived: the statement exists. parsedGen*: int ## ``stmtCacheResetGen`` when it arrived. described*: bool ## ``RowDescription`` or ``NoData`` arrived: ``fields``, ``noData`` and ## ``paramOids`` are complete. fields*: seq[FieldDescription] noData*: bool ## ``NoData`` arrived: the statement returns no rows. paramOids*: seq[int32]
- What the server confirmed about an op's Bind and, for a cache miss, its Parse and Describe(Statement). The cache acts on these alone: whether the operation failed says nothing about how far the server got.
RetryOptions = object maxAttempts*: int = 3 ## Total attempts (``<=1`` = no retry). baseDelayMs*: int = 20 ## Initial backoff ms. maxDelayMs*: int = 1000 ## Max backoff ms. multiplier*: float = 2.0 ## Backoff multiplier. jitter*: bool = true ## Full jitter via ``std/random``; call ``randomize()`` for cross-process de-correlation. retryableStates*: seq[string] = ["40001", "40P01"] ## SQLSTATEs that trigger retry.
- Retry config for withTransactionRetry; unset fields keep the defaults below. Fields are not validated: a backoff that computes negative (e.g. a negative baseDelayMs or maxDelayMs) becomes 0ms, an immediate retry, and one longer than about 146 years is cut to that.
StmtCacheStatus = enum scsUncached, ## Caching disabled: unnamed Parse. scsHit, ## Binds a cache entry's statement (same SQL, matching OIDs). scsShare, ## Binds a statement an earlier op of the same pipeline Parses (same SQL, ## compatible OIDs). Skips Parse/Describe(Statement) and Describes the ## portal instead. scsMiss ## Parses a fresh named statement; `settleStmtCache` decides its fate.
- How an operation uses the statement cache, decided in the send phase.
TransactionOptions = object isolation*: IsolationLevel access*: AccessMode deferrable*: DeferrableMode
- Options for BEGIN: isolation level, access mode, and deferrable mode.
Consts
copyBatchSize = 262144
- 256KB batch threshold for COPY IN buffering
generatedPortalNameLen = 27
- Same, for the portal name openCursorImpl generates.
generatedStmtNameLen = 23
- Longest name nextStmtName can produce. Size pre-flights charge it because the real name is only picked at send time.
StmtCacheInvalidatingStates = ["26000", "0A000"]
- SQLSTATEs that invalidate cached prepared statements (requires re-parse). 42P18 is absent on purpose: it is Parse-phase, so no cached statement can hit it.
Procs
proc backoffDelayMs(opts: RetryOptions; attempt: int): int {....raises: [], tags: [], forbids: [].}
- Backoff ms for attempt (1-based). Exponential with jitter.
proc boundRowData(fields: sink seq[FieldDescription]; colFmts: seq[int16]): RowData {. ...raises: [], tags: [], forbids: [].}
- boundRowData with the type OIDs taken from fields.
proc buildBeginSql(opts: TransactionOptions): string {....raises: [], tags: [], forbids: [].}
- Build a BEGIN SQL statement with the specified transaction options (isolation level, access mode, deferrable mode).
func cacheHitColFmts(resultFormats: openArray[int16]; cachedColFmts: seq[int16]; numCols: int): seq[int16] {....raises: [], tags: [], forbids: [].}
- Per-column formats for a cache hit. Prefers the formats this Bind requested: the same SQL may be re-issued with a different resultFormat, and the stale cached one would reinterpret the bytes.
proc cacheHitRowData(resultFormats: openArray[int16]; cached: CachedStmt): RowData {. ...raises: [], tags: [], forbids: [].}
- boundRowData for a cache hit, with the formats this Bind requested rather than the stale cached ones (see cacheHitColFmts). Nil for a statement that returns no rows, as on its miss.
func deriveColFmts(resultFormats: openArray[int16]; numCols: int): seq[int16] {. ...raises: [], tags: [], forbids: [].}
- Expand Bind result-format codes to per-column: one code broadcasts, an array applies positionally, columns past its end default to text (0).
proc describedRowData(fields: var seq[FieldDescription]; portal: bool; resultFormats: openArray[int16]): RowData {....raises: [], tags: [], forbids: [].}
- RowData for a RowDescription. A portal Describe reports the formats Bind negotiated; a statement Describe reports text for every column, so the requested formats are stamped over fields.
proc extractParams(params: openArray[PgParam]): tuple[oids: seq[int32], formats: seq[int16], values: seq[Option[seq[byte]]]] {....raises: [], tags: [], forbids: [].}
proc flattenInline(params: openArray[PgParamInline]; resultFormatsLen: int = 0): tuple[ data: seq[byte], ranges: seq[tuple[off: int32, len: int32]], oids: seq[int32], formats: seq[int16]] {. ...raises: [PgTypeError, PgMessageTooLargeError, PgProtocolError], tags: [], forbids: [].}
func initBoundResult(qr: var QueryResult; data: sink RowData) {....raises: [], tags: [], forbids: [].}
- Start qr from a boundRowData result, or nil for no rows.
proc invalidateIfOidMismatch(conn: PgConnection; sql: string; cached: CachedStmt; currentOids: openArray[int32]; cacheHit: var bool) {....raises: [], tags: [], forbids: [].}
- Evict cached statement if OIDs mismatch; sets cacheHit=false. cached may be nil iff cacheHit == false (only deref'd under it).
proc invalidateIfOidMismatch(conn: PgConnection; sql: string; cached: CachedStmt; params: openArray[PgParam]; cacheHit: var bool) {....raises: [], tags: [], forbids: [].}
- PgParam overload (no seq[int32] alloc). Same nil precondition on cached.
proc isRetryableTxError(e: ref CatchableError; states: openArray[string]): bool {. ...raises: [], tags: [], forbids: [].}
- Whether e is a PgQueryError whose SQLSTATE is in states. The 25P02 a transaction macro raises for a COMMIT answered with ROLLBACK also qualifies when its parent, the error that aborted the transaction, does. Non-PgQueryError failures (connection drops, timeouts) are never retryable here: they leave the connection unusable for a fresh attempt.
proc observe(facts: var OpFacts; conn: PgConnection; msg: BackendMessage; miss: bool) {....raises: [], tags: [], forbids: [].}
- Record a reply to the op's Bind, or to a miss's Parse or Describe(Statement). Only a miss settles on the latter.
proc paramOidsMatch(cachedOids, currentOids: openArray[int32]): bool {. ...raises: [], tags: [], forbids: [].}
- Whether cached param OIDs match current (0 = wildcard).
proc paramOidsMatch(cachedOids: openArray[int32]; params: openArray[PgParam]): bool {. ...raises: [], tags: [], forbids: [].}
- PgParam overload (avoids seq[int32] alloc).
proc preflightResultFormatsLen(cached: CachedStmt; cacheHit: bool; resultFormatsLen: int = 0): int {....raises: [], tags: [], forbids: [].}
- Result-format count the send path will really emit: a cache hit replays cached.resultFormats when the caller passed none. Call after invalidateIfOidMismatch; cached may be nil iff not cacheHit.
proc settleStmtCache(conn: PgConnection; sql, stmtName: string; cache: StmtCacheStatus; facts: var OpFacts; queryError: ref PgQueryError) {....raises: [], tags: [], forbids: [].}
- Statement-cache bookkeeping once an operation's ReadyForQuery arrives. A miss the server Described is cached even when Bind or Execute then failed (a plan gone stale meanwhile fails its next hit, which invalidates it); one it Parsed but did not Describe is Closed; one it never Parsed is left alone, since the name may be someone else's (42P05). A miss Parsed before a later DISCARD ALL / DEALLOCATE ALL (facts.parsedGen behind stmtCacheResetGen) is neither cached nor Closed: the server already dropped the statement. A hit or share that bound is kept: the error is the statement's own.
func shouldRetryStmtCacheInvalidation(conn: PgConnection; cacheHit: bool; sqlState: string; bound: bool): bool {. ...raises: [], tags: [], forbids: [].}
- Whether a failed op may be re-issued once from Parse: a cache-hit Bind refused with 26000 (statements dropped without a reset tag) outside a transaction. Bind precedes Execute, so nothing ran; bound means the error came later, from the statement itself.
func stmtCacheStatus(cacheHit, cacheMiss: bool): StmtCacheStatus {.inline, ...raises: [], tags: [], forbids: [].}
func toFormatCodes(rf: ResultFormat): seq[int16] {....raises: [], tags: [], forbids: [].}
- Convert a high-level ResultFormat to wire-protocol format codes.
proc validateEncodedParams(params: openArray[Option[seq[byte]]]; paramFormatsLen: int; resultFormatsLen: int = 0; stmtNameLen: int = generatedStmtNameLen; portalLen: int = 0) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- validateTypedParams for the already-encoded seq[Option[seq[byte]]] path. Runs before the send templates start filling sendBuf, so a call this rejects fails without having touched the connection.
proc validateExtendedQuery(sql: string; nParams: int; nParamOids: int = nParams; stmtNameLen: int = generatedStmtNameLen) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- Add-time validation shared by every extended-query entry point. nParamOids sizes the Parse, which the *Impl procs take separately from the values. Pass stmtNameLen 0 for an unnamed statement.
proc validateParamCount(n: int; what: string) {....raises: [PgTypeError], tags: [], forbids: [].}
- Reject a count the encoder would reject later: a send-phase failure would take the whole pipelined batch down.
proc validateParseMsg(sql: string; nParams: int; stmtNameLen = generatedStmtNameLen) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- Reject a Parse the encoder would reject later, so a pipelined op fails on its own instead of in buildSendPhase. Best-effort; patchMsgLen stays the authority on message size.
proc validateRawBind(data: openArray[byte]; ranges: openArray[tuple[off: int32, len: int32]]; paramFormats: openArray[int16]; resultFormatsLen: int = 0; stmtNameLen: int = generatedStmtNameLen) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- validateEncodedParams for the raw buffer/ranges path (addBindRaw). The *Impl procs are public, so data/ranges may never have gone through flattenInline.
proc validateTypedParams(params: openArray[PgParam]; resultFormatsLen: int = 0; stmtNameLen: int = generatedStmtNameLen) {. ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
- validateInlineParam's counterpart for the seq[PgParam] path: checks the Int16 count, the payload total and the Bind envelope at add time. resultFormatsLen 0 means unknown, making the check a lower bound.
Macros
macro queryEachRecvLoop(conn: PgConnection; sql: string; resultFormats: openArray[int16]; cacheHit, cacheMiss: bool; stmtName: string; hitData: RowData; callback: RowCallback; rowCount: var int64; facts: var OpFacts): untyped
macro retryStmtCacheInvalidation(conn: PgConnection; cacheHit, facts, body: untyped): untyped
- Run body, one extended-query op that sets cacheHit in its send phase, fills facts in its receive loop and returns on success, and re-issue it once when shouldRetryStmtCacheInvalidation allows. The re-issue passes checkReady first, so a close() during the failed attempt stops it.
Templates
template appendInlineParamUnchecked(data: var seq[byte]; ranges: var seq[ tuple[off: int32, len: int32]]; oids: var seq[int32]; formats: var seq[int16]; p: PgParamInline)
- Encode one PgParamInline into SoA buffers. Raises PgTypeError. The caller must have run validateInlineParam on p: its bounds keep the overflow read in range.
template execRecvLoop(conn: PgConnection; sql: string; cacheHit, cacheMiss: bool; stmtName: string; commandTag: var string; facts: var OpFacts)
- Receive-loop counterpart of queryRecvLoop for the extended-query exec path: DataRows are dropped by the parser (bare pumpUntilReady uses skipDataRow = true); this loop only exposes the CommandComplete tag via the commandTag out-parameter. Shared by execImpl (both overloads), execInlineImpl, and execDirectRunImpl.
template queryRecvLoop(conn: PgConnection; sql: string; resultFormats: openArray[int16]; cacheHit, cacheMiss: bool; stmtName: string; hitData: RowData; qr: var QueryResult; facts: var OpFacts)
template sendExtendedExec(conn: PgConnection; cached: CachedStmt; cacheHit, cacheMiss: var bool; stmtName: var string; parseStep, bindStep: untyped)
- exec variant of sendExtendedQuery (no per-column format tracking). Precondition: cached may be nil iff cacheHit == false.
template sendExtendedQuery(conn: PgConnection; resultFormats: seq[int16]; cached: CachedStmt; cacheHit, cacheMiss: var bool; stmtName: var string; hitData: var RowData; effectiveResultFormats: var seq[int16]; parseStep, bindStep: untyped)
- Emit Parse/Bind/Describe/Execute/Sync sequence (cache hit/miss/disabled). Precondition: cached may be nil iff cacheHit == false; the cache-miss and cache-disabled branches never read it. Only a hit sets hitData (see cacheHitRowData).
template validateInlineParam(p: PgParamInline)
- Reject a bad PgParamInline with PgTypeError, so a hand-built one stays catchable under PgError instead of a fatal RangeDefect.