fix(pg_errors, pg_pool): add PoolErrorKind to PgPoolError for program…

async_postgres/pg_client/transaction.nim

@@ -150,13 +150,12 @@ proc buildTxBeginAndTimeout*(
 proc buildRollbackCleanup*(connSym, rollbackTimeout: NimNode): NimNode =
   ## Build the shared `onCleanupSkipped`-wired ROLLBACK cleanup used on a failed
   ## attempt by the conn/pool/cluster transaction macros: skip ROLLBACK on an
-  ## invalidated connection or when the server already ended the transaction
-  ## (reporting both via `onCleanupSkipped`), otherwise ROLLBACK with
-  ## `rollbackTimeout` as the per-call timeout and report a swallowed failure.
+  ## invalidated connection (reported as `csrConnInvalidated` via
+  ## `onCleanupSkipped`) or a server-ended transaction (`tsIdle`, silent),
+  ## otherwise ROLLBACK with `rollbackTimeout` and report a swallowed failure.
   ##
   ## Cancelled and plain-failure ROLLBACKs are both reported and swallowed so
-  ## the enclosing `except` can re-raise the original body error — re-raising
-  ## the cancel would surface a fresh chronos `CancelledError` and mask it.
+  ## the enclosing `except` can re-raise the original body error.
   let cleanupErrSym = genSym(nskLet, "cleanupErr")
   let cleanupCancelSym = genSym(nskLet, "cleanupCancel")
   let cleanupDefectSym = genSym(nskLet, "cleanupDefect")
@@ -195,9 +194,9 @@ proc buildSavepointRollbackCleanup*(
 ): NimNode =
   ## Build the shared `onCleanupSkipped`-wired ROLLBACK TO SAVEPOINT cleanup used
   ## on a failed body by `withSavepoint` / `withSavepointDeadline`: skip on an
-  ## invalidated connection or when the surrounding transaction has already
-  ## ended (reporting both via `onCleanupSkipped`), otherwise ROLLBACK TO
-  ## SAVEPOINT with `rollbackTimeout` as the per-call timeout and report a
+  ## invalidated connection (reported as `csrConnInvalidated` via
+  ## `onCleanupSkipped`) or an ended surrounding transaction (`tsIdle`, silent),
+  ## otherwise ROLLBACK TO SAVEPOINT with `rollbackTimeout` and report a
   ## swallowed failure. The caller binds `spNameSym` (already quoted via
   ## `quoteIdentifier`) in the surrounding scope.
   ##

async_postgres/pg_errors.nim

@@ -97,13 +97,34 @@ type
     ## ``PgTimeoutError`` before any ``PgConnectionError`` clause if you need to
     ## distinguish those.
 
+  PoolErrorKind* = enum
+    ## Machine-readable category of a `PgPoolError`.
+    pekUnknown
+      ## Default: a `PgPoolError` built without an explicit `kind`; do not
+      ## treat as `pekClosed`.
+    pekClosed ## The pool is permanently closed; retrying cannot succeed.
+    pekAcquireTimeout
+      ## An acquire deadline elapsed (`acquireTimeout` or cluster fallback);
+      ## retrying later may succeed.
+    pekQueueFull
+      ## The waiter queue is full (`maxWaiters` bound); retrying later may succeed.
+    pekConnectFailed
+      ## A connect attempt failed during acquire (underlying error in `parent`);
+      ## retrying may succeed.
+    pekBatchFailed ## A pipelined batch was unservable; no connection was acquired.
+    pekDefectWrapped
+      ## A user-code `Defect` (body/release block or session reset) wrapped to
+      ## cross an async boundary; preserved as `parent`.
+
   PgPoolError* = object of PgError
-    ## Pool-level acquire failure: acquire timeout, pool closed, waiter queue
-    ## full, or a failed connect attempt during acquire (the underlying error,
-    ## e.g. ``PgConnectionError``, is preserved as ``parent``).
+    ## Pool-level acquire/operation failure (closed, acquire timeout, queue
+    ## full, connect failed, unservable batch, or wrapped user-code `Defect`;
+    ## the underlying error is preserved as ``parent``).
     ##
-    ## Also raised by the pooled `with*` macros / `runAndRelease` when a body
-    ## `Defect` is wrapped (Defect as ``parent``).
+    ## `kind` classifies the failure programmatically; the message string is
+    ## informational only. Errors built without `newPoolError` have
+    ## `kind == pekUnknown`.
+    kind*: PoolErrorKind ## Failure category (see `PoolErrorKind`).
 
   PgNotifyOverflowError* = object of PgError
     dropped*: int ## Number of notifications dropped due to queue overflow
@@ -114,6 +135,12 @@ type
     reconnectionAttempted*: bool
       ## True if the pump attempted reconnection before giving up.
 
+template newPoolError*(
+    errKind: PoolErrorKind, message: string, parentErr: ref Exception = nil
+): untyped =
+  ## Create a `PgPoolError` with the given `errKind` (see `PoolErrorKind`).
+  (ref PgPoolError)(kind: errKind, msg: message, parent: parentErr)
+
 const
   # Commonly dispatched-on SQLSTATE codes
   SqlStateNotNullViolation* = "23502"

async_postgres/pg_pool.nim

@@ -360,7 +360,7 @@ proc resetSession*(pool: PgPool, conn: PgConnection) {.async.} =
     # Incomplete reset: mark csClosed so the conn is discarded, then re-raise
     # wrapped (a raw Defect cannot cross a chronos async boundary).
     conn.state = csClosed
-    raise newException(PgPoolError, d.msg, d)
+    raise newPoolError(pekDefectWrapped, d.msg, d)
 
 proc computeConnectBackoff*(initial, maxDelay: Duration, failures: int): Duration =
   ## Exponential backoff for repeated connect failures: returns
@@ -503,7 +503,7 @@ proc spawnConnectForWaiter(pool: PgPool) =
       # decrements `waiterCount` a second time (failLastWaiter below already
       # did) and permanently corrupts the FIFO fast-path guard.
       discard pool.failLastWaiter(
-        newException(PgPoolError, "Pool connect for waiter failed", e)
+        newPoolError(pekConnectFailed, "Pool connect for waiter failed", e)
       )
     finally:
       if not consumed and pool.active > 0:
@@ -864,7 +864,7 @@ type AcquireResult = tuple[conn: PgConnection, wasCreated: bool]
 
 proc acquireImpl(pool: PgPool): Future[AcquireResult] {.async.} =
   if pool.closed:
-    raise newException(PgPoolError, "Pool is closed")
+    raise newPoolError(pekClosed, "Pool is closed")
 
   let now = Moment.now()
   let acquireStart = now
@@ -882,10 +882,10 @@ proc acquireImpl(pool: PgPool): Future[AcquireResult] {.async.} =
 
   template raiseAcquireTimeout() =
     pool.metrics.timeoutCount.inc
-    raise newException(PgPoolError, "Pool acquire timeout")
+    raise newPoolError(pekAcquireTimeout, "Pool acquire timeout")
 
   template raisePoolClosed() =
-    raise newException(PgPoolError, "Pool is closed")
+    raise newPoolError(pekClosed, "Pool is closed")
 
   template recordAcquire() =
     pool.metrics.acquireCount.inc
@@ -1028,8 +1028,8 @@ proc acquireImpl(pool: PgPool): Future[AcquireResult] {.async.} =
         # microseconds early).
         if hasDeadline and remainingBudget() <= milliseconds(1):
           pool.metrics.timeoutCount.inc
-          raise newException(PgPoolError, "Pool acquire timeout", e)
-        raise newException(PgPoolError, "Pool connect failed", e)
+          raise newPoolError(pekAcquireTimeout, "Pool acquire timeout", e)
+        raise newPoolError(pekConnectFailed, "Pool connect failed", e)
       pool.metrics.createCount.inc
       # A successful caller-driven connect signals the DB is reachable —
       # let the maintenance loop resume immediate replenishment.
@@ -1046,8 +1046,8 @@ proc acquireImpl(pool: PgPool): Future[AcquireResult] {.async.} =
   # Either max connections are reached or waiters are queued ahead of us;
   # queue up and wait for delivery.
   if pool.config.maxWaiters >= 0 and pool.waiterCount >= pool.config.maxWaiters:
-    raise newException(
-      PgPoolError,
+    raise newPoolError(
+      pekQueueFull,
       "Pool acquire queue full (maxWaiters=" & $pool.config.maxWaiters & ")",
     )
   # Compute the remaining budget before queueing; whatever the idle
@@ -1084,7 +1084,7 @@ proc acquireImpl(pool: PgPool): Future[AcquireResult] {.async.} =
     except AsyncTimeoutError:
       pool.metrics.timeoutCount.inc
       pool.settleAbandonedWaiter(waiter)
-      raise newException(PgPoolError, "Pool acquire timeout")
+      raise newPoolError(pekAcquireTimeout, "Pool acquire timeout")
     except CancelledError as e:
       # External cancellation (e.g. a caller's `wait()`-style deadline such as
       # pool.withTransactionDeadline or a cluster fallback timeout) can land
@@ -1109,7 +1109,8 @@ proc acquire*(pool: PgPool): Future[PgConnection] {.async.} =
   ## release. Raises `PgPoolError` on every failure mode: acquire timeout,
   ## pool closed, waiter queue full, or a failed connect attempt — for
   ## connect failures the underlying error (e.g. `PgConnectionError`) is
-  ## preserved as the `parent` of the raised `PgPoolError`.
+  ## preserved as the `parent` of the raised `PgPoolError`. Use the `kind`
+  ## field (`PoolErrorKind`) to distinguish the failure mode programmatically.
   if pool.config.tracer == nil:
     let ar = await pool.acquireImpl()
     return ar.conn
@@ -1193,7 +1194,7 @@ proc runAndReleaseImpl[T](
   if bodyErr != nil:
     raise bodyErr
   if bodyDefect != nil:
-    raise newException(PgPoolError, bodyDefect.msg, bodyDefect)
+    raise newPoolError(pekDefectWrapped, bodyDefect.msg, bodyDefect)
   when T isnot void:
     return res
 
@@ -1235,7 +1236,7 @@ proc buildReleaseAndReraise*(releaseCall, bodyErrSym, bodyDefectSym: NimNode): N
       # Same-frame Defect from the release path: wrap like the body Defect
       # (see runAndReleaseImpl), unless it would shadow the body error.
       if `bodyErrSym` == nil and `bodyDefectSym` == nil:
-        raise newException(PgPoolError, `releaseDefectSym`.msg, `releaseDefectSym`)
+        raise newPoolError(pekDefectWrapped, `releaseDefectSym`.msg, `releaseDefectSym`)
     except CatchableError as `releaseErrSym`:
       # Never shadow a body error with a release failure.
       if `bodyErrSym` == nil and `bodyDefectSym` == nil:
@@ -1245,7 +1246,7 @@ proc buildReleaseAndReraise*(releaseCall, bodyErrSym, bodyDefectSym: NimNode): N
     if `bodyDefectSym` != nil:
       # Wrap the Defect (see runAndReleaseImpl): chronos re-raises raw
       # Defects eagerly.
-      raise newException(PgPoolError, `bodyDefectSym`.msg, `bodyDefectSym`)
+      raise newPoolError(pekDefectWrapped, `bodyDefectSym`.msg, `bodyDefectSym`)
 
 macro withConnection*(pool: PgPool, conn, body: untyped): untyped =
   ## Acquire a connection, execute `body`, then release it back to the pool.
@@ -1380,7 +1381,7 @@ proc executeBatch(
       failPendingOp(op, e)
   except Defect as d:
     for op in batch:
-      failPendingOp(op, newException(PgPoolError, d.msg, d))
+      failPendingOp(op, newPoolError(pekDefectWrapped, d.msg, d))
   try:
     await pool.resetSessionAndRelease(conn)
   except CancelledError as e:
@@ -1463,21 +1464,24 @@ proc dispatchHomogeneous(
       failPendingOp(op, e)
     except Defect as d:
       # Wrap the Defect so the op's future fails instead of hanging.
-      failPendingOp(op, newException(PgPoolError, d.msg, d))
+      failPendingOp(op, newPoolError(pekDefectWrapped, d.msg, d))
     return
 
   # Multi-op path: acquire connections and distribute.
   var conns: seq[PgConnection]
+  var acquireErr: ref Exception
   let nConns = min(ops.len, max(1, maxConns))
   for i in 0 ..< nConns:
     try:
       let conn = await pool.acquire()
       conns.add(conn)
-    except CatchableError:
+    except CatchableError as e:
+      acquireErr = e
       break
 
   if conns.len == 0:
-    let err = newException(PgPoolError, "Failed to acquire connection for batch")
+    let err =
+      newPoolError(pekBatchFailed, "Failed to acquire connection for batch", acquireErr)
     for op in ops:
       failPendingOp(op, err)
     return
@@ -1591,7 +1595,7 @@ proc exec*(
   ## and stays unlimited.
   if pool.config.pipelined:
     if pool.closed:
-      raise newException(PgPoolError, "Pool is closed")
+      raise newPoolError(pekClosed, "Pool is closed")
     let fut = newFuture[CommandResult]("PgPool.exec.pipelined")
     pool.pendingOps.addLast(
       PendingPoolOp(
@@ -1614,7 +1618,7 @@ proc exec*(
   ## see the `seq[PgParam]` overload for the batch timeout semantics.
   if pool.config.pipelined:
     if pool.closed:
-      raise newException(PgPoolError, "Pool is closed")
+      raise newPoolError(pekClosed, "Pool is closed")
     let fut = newFuture[CommandResult]("PgPool.exec.pipelined")
     pool.pendingOps.addLast(
       PendingPoolOp(
@@ -1648,7 +1652,7 @@ proc query*(
   ## and stays unlimited.
   if pool.config.pipelined:
     if pool.closed:
-      raise newException(PgPoolError, "Pool is closed")
+      raise newPoolError(pekClosed, "Pool is closed")
     let fut = newFuture[QueryResult]("PgPool.query.pipelined")
     pool.pendingOps.addLast(
       PendingPoolOp(
@@ -1679,7 +1683,7 @@ proc query*(
   ## see the `seq[PgParam]` overload for the batch timeout semantics.
   if pool.config.pipelined:
     if pool.closed:
-      raise newException(PgPoolError, "Pool is closed")
+      raise newPoolError(pekClosed, "Pool is closed")
     let fut = newFuture[QueryResult]("PgPool.query.pipelined")
     pool.pendingOps.addLast(
       PendingPoolOp(
@@ -2263,7 +2267,7 @@ macro withTransactionDeadline*(pool: PgPool, args: varargs[untyped]): untyped =
         `releasedSym` = true
         # Wrap the Defect (see runAndReleaseImpl) unless it shadows the body error.
         if `bodyErrSym` == nil and `bodyDefectSym` == nil:
-          raise newException(PgPoolError, `dSym`.msg, `dSym`)
+          raise newPoolError(pekDefectWrapped, `dSym`.msg, `dSym`)
       except CatchableError as `releaseErrSym`:
         # Already released: re-raise only when the body succeeded.
         `releasedSym` = true
@@ -2275,7 +2279,7 @@ macro withTransactionDeadline*(pool: PgPool, args: varargs[untyped]): untyped =
       if `bodyDefectSym` != nil:
         # Wrap the Defect (see runAndReleaseImpl): chronos re-raises raw
         # Defects eagerly.
-        raise newException(PgPoolError, `bodyDefectSym`.msg, `bodyDefectSym`)
+        raise newPoolError(pekDefectWrapped, `bodyDefectSym`.msg, `bodyDefectSym`)
       checkNoBodyEscapePost(
         block:
           `body`,
@@ -2459,7 +2463,7 @@ macro withTransactionRetryDeadline*(
         `releasedSym` = true
         # Wrap the Defect (see runAndReleaseImpl) unless it shadows the body error.
         if `bodyErrSym` == nil and `bodyDefectSym` == nil:
-          raise newException(PgPoolError, `dSym`.msg, `dSym`)
+          raise newPoolError(pekDefectWrapped, `dSym`.msg, `dSym`)
       except CatchableError as `releaseErrSym`:
         # Already released: re-raise only when the body succeeded.
         `releasedSym` = true
@@ -2471,7 +2475,7 @@ macro withTransactionRetryDeadline*(
       if `bodyDefectSym` != nil:
         # Wrap the Defect (see runAndReleaseImpl): chronos re-raises raw
         # Defects eagerly.
-        raise newException(PgPoolError, `bodyDefectSym`.msg, `bodyDefectSym`)
+        raise newPoolError(pekDefectWrapped, `bodyDefectSym`.msg, `bodyDefectSym`)
       checkNoBodyEscapePost(
         block:
           `body`,
@@ -2547,12 +2551,12 @@ proc close*(pool: PgPool, timeout = ZeroDuration): Future[void] {.async.} =
   while pool.waiters.len > 0:
     let waiter = pool.waiters.popFirst()
     if not waiter.isAbandoned:
-      waiter.fut.fail(newException(PgPoolError, "Pool closed"))
+      waiter.fut.fail(newPoolError(pekClosed, "Pool closed"))
   pool.waiterCount = 0
 
   # Fail all pending pipeline ops
   pool.dispatchScheduled = false
-  let closeErr = newException(PgPoolError, "Pool closed")
+  let closeErr = newPoolError(pekClosed, "Pool closed")
   while pool.pendingOps.len > 0:
     let op = pool.pendingOps.popFirst()
     failPendingOp(op, closeErr)

async_postgres/pg_pool_cluster.nim

@@ -229,8 +229,8 @@ proc acquireRead(
       return (conn, cluster.primary)
     except AsyncTimeoutError:
       asyncSpawn drainAbandonedAcquire(primaryFut)
-      raise newException(
-        PgPoolError,
+      raise newPoolError(
+        pekAcquireTimeout,
         "Pool cluster fallback acquire timeout (replica error: " & replicaErr.msg & ")",
         replicaErr,
       )

tests/test_e2e_pool.nim

@@ -454,7 +454,8 @@ suite "E2E: Pool Stress":
         conn2.release()
       except PgError as e:
         raised = true
-        doAssert "timeout" in e.msg.toLowerAscii()
+        doAssert e of PgPoolError
+        doAssert (ref PgPoolError)(e).kind == pekAcquireTimeout
 
       doAssert raised