Bugfix: don't try to cull connections from an empty pool.

The first step in culling connections was to take a subset of the pool's
connections from index 0 to numToCull. This leads to an error if
numToCull is also zero.

fiber_orm.nimble

@@ -1,6 +1,6 @@
 # Package
 
-version       = "0.3.6"
+version       = "1.0.2"
 author        = "Jonathan Bernard"
 description   = "Lightweight Postgres ORM for Nim."
 license       = "GPL-3.0"

src/fiber_orm.nim

@@ -1,10 +1,13 @@
-import db_postgres, macros, options, sequtils, strutils, uuids
-import namespaced_logging
+import std/db_postgres, std/macros, std/options, std/sequtils, std/strutils
+import namespaced_logging, uuids
 
-from unicode import capitalize
+from std/unicode import capitalize
 
+import ./fiber_orm/pool
 import ./fiber_orm/util
+
 export
+  pool,
   util.columnNamesForModel,
   util.dbFormat,
   util.dbNameToIdent,
@@ -18,7 +21,7 @@ type NotFoundError* = object of CatchableError
 var logNs {.threadvar.}: LoggingNamespace
 
 template log(): untyped =
-  if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm", level = lvlDebug)
+  if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm", level = lvlNotice)
   logNs
 
 proc newMutateClauses(): MutateClauses =
@@ -31,13 +34,11 @@ proc createRecord*[T](db: DbConn, rec: T): T =
   var mc = newMutateClauses()
   populateMutateClauses(rec, true, mc)
 
-  # Confusingly, getRow allows inserts and updates. We use it to get back the ID
-  # we want from the row.
   let sqlStmt =
     "INSERT INTO " & tableName(rec) &
     " (" & mc.columns.join(",") & ") " &
     " VALUES (" & mc.placeholders.join(",") & ") " &
-    " RETURNING *"
+    " RETURNING " & columnNamesForModel(rec).join(",")
 
   log().debug "createRecord: [" & sqlStmt & "]"
   let newRow = db.getRow(sql(sqlStmt), mc.values)
@@ -123,14 +124,27 @@ macro generateProcsForModels*(dbType: type, modelTypes: openarray[type]): untype
     let deleteName = ident("delete" & modelName)
     let idType = typeOfColumn(t, "id")
     result.add quote do:
-      proc `getName`*(db: `dbType`, id: `idType`): `t` = getRecord(db.conn, `t`, id)
-      proc `getAllName`*(db: `dbType`): seq[`t`] = getAllRecords(db.conn, `t`)
+      proc `getName`*(db: `dbType`, id: `idType`): `t` =
+        db.withConn: result = getRecord(conn, `t`, id)
+
+      proc `getAllName`*(db: `dbType`): seq[`t`] =
+        db.withConn: result = getAllRecords(conn, `t`)
+
       proc `findWhereName`*(db: `dbType`, whereClause: string, values: varargs[string, dbFormat]): seq[`t`] =
-        return findRecordsWhere(db.conn, `t`, whereClause, values)
-      proc `createName`*(db: `dbType`, rec: `t`): `t` = createRecord(db.conn, rec)
-      proc `updateName`*(db: `dbType`, rec: `t`): bool = updateRecord(db.conn, rec)
-      proc `deleteName`*(db: `dbType`, rec: `t`): bool = deleteRecord(db.conn, rec)
-      proc `deleteName`*(db: `dbType`, id: `idType`): bool = deleteRecord(db.conn, `t`, id)
+        db.withConn:
+          result = findRecordsWhere(conn, `t`, whereClause, values)
+
+      proc `createName`*(db: `dbType`, rec: `t`): `t` =
+        db.withConn: result = createRecord(conn, rec)
+
+      proc `updateName`*(db: `dbType`, rec: `t`): bool =
+        db.withConn: result = updateRecord(conn, rec)
+
+      proc `deleteName`*(db: `dbType`, rec: `t`): bool =
+        db.withConn: result = deleteRecord(conn, rec)
+
+      proc `deleteName`*(db: `dbType`, id: `idType`): bool =
+        db.withConn: result = deleteRecord(conn, `t`, id)
 
 macro generateLookup*(dbType: type, modelType: type, fields: seq[string]): untyped =
   let fieldNames = fields[1].mapIt($it)
@@ -139,7 +153,7 @@ macro generateLookup*(dbType: type, modelType: type, fields: seq[string]): untyp
   # Create proc skeleton
   result = quote do:
     proc `procName`*(db: `dbType`): seq[`modelType`] =
-      return findRecordsBy(db.conn, `modelType`)
+      db.withConn: result = findRecordsBy(conn, `modelType`)
 
   var callParams = quote do: @[]
 
@@ -149,10 +163,19 @@ macro generateLookup*(dbType: type, modelType: type, fields: seq[string]): untyp
     paramTuple.add(newColonExpr(ident("field"), newLit(identNameToDb(n))))
     paramTuple.add(newColonExpr(ident("value"), ident(n)))
 
+    # Add the parameter to the outer call (the generated proc)
+    # result[3] is ProcDef -> [3]: FormalParams
     result[3].add(newIdentDefs(ident(n), ident("string")))
+
+    # Build up the AST for the inner procedure call
     callParams[1].add(paramTuple)
 
-  result[6][0][0].add(callParams)
+  # Add the call params to the inner procedure call
+  # result[6][0][1][0][1] is
+  #   ProcDef -> [6]: StmtList (body) -> [0]: Call ->
+  #     [1]: StmtList (withConn body) -> [0]: Asgn (result =) ->
+  #     [1]: Call (inner findRecords invocation)
+  result[6][0][1][0][1].add(callParams)
 
 macro generateProcsForFieldLookups*(dbType: type, modelsAndFields: openarray[tuple[t: type, fields: seq[string]]]): untyped =
   result = newStmtList()
@@ -166,7 +189,7 @@ macro generateProcsForFieldLookups*(dbType: type, modelsAndFields: openarray[tup
     # Create proc skeleton
     let procDefAST = quote do:
       proc `procName`*(db: `dbType`): seq[`modelType`] =
-        return findRecordsBy(db.conn, `modelType`)
+        db.withConn: result = findRecordsBy(conn, `modelType`)
 
     var callParams = quote do: @[]
 
@@ -179,6 +202,28 @@ macro generateProcsForFieldLookups*(dbType: type, modelsAndFields: openarray[tup
       procDefAST[3].add(newIdentDefs(ident(n), ident("string")))
       callParams[1].add(paramTuple)
 
-    procDefAST[6][0][0].add(callParams)
+    procDefAST[6][0][1][0][1].add(callParams)
 
     result.add procDefAST
+
+proc initPool*(
+    connect: proc(): DbConn,
+    poolSize = 10,
+    hardCap = false,
+    healthCheckQuery = "SELECT 'true' AS alive"): DbConnPool =
+
+  initDbConnPool(DbConnPoolConfig(
+    connect: connect,
+    poolSize: poolSize,
+    hardCap: hardCap,
+    healthCheckQuery: healthCheckQuery))
+
+template inTransaction*(db: DbConnPool, body: untyped) =
+  pool.withConn(db):
+    conn.exec(sql"BEGIN TRANSACTION")
+    try:
+      body
+      conn.exec(sql"COMMIT")
+    except:
+      conn.exec(sql"ROLLBACK")
+      raise getCurrentException()

src/fiber_orm/pool.nim

@@ -0,0 +1,99 @@
+import std/db_postgres, std/sequtils, std/strutils, std/sugar
+
+import namespaced_logging
+
+
+type
+  DbConnPoolConfig* = object
+    connect*: () -> DbConn
+    poolSize*: int
+    hardCap*: bool
+    healthCheckQuery*: string
+
+  PooledDbConn = ref object
+    conn: DbConn
+    id: int
+    free: bool
+
+  DbConnPool* = ref object
+    conns: seq[PooledDbConn]
+    cfg: DbConnPoolConfig
+    lastId: int
+
+var logNs {.threadvar.}: LoggingNamespace
+
+template log(): untyped =
+  if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm/pool", level = lvlNotice)
+  logNs
+
+proc initDbConnPool*(cfg: DbConnPoolConfig): DbConnPool =
+  log().debug("Initializing new pool (size: " & $cfg.poolSize)
+  result = DbConnPool(
+    conns: @[],
+    cfg: cfg)
+
+proc newConn(pool: DbConnPool): PooledDbConn =
+  log().debug("Creating a new connection to add to the pool.")
+  pool.lastId += 1
+  let conn = pool.cfg.connect()
+  result = PooledDbConn(
+    conn: conn,
+    id: pool.lastId,
+    free: true)
+  pool.conns.add(result)
+
+proc maintain(pool: DbConnPool): void =
+  log().debug("Maintaining pool. $# connections." % [$pool.conns.len])
+  pool.conns.keepIf(proc (pc: PooledDbConn): bool =
+    if not pc.free: return true
+
+    try:
+      discard getRow(pc.conn, sql(pool.cfg.healthCheckQuery), [])
+      return true
+    except:
+      try: pc.conn.close()  # try to close the connection
+      except: discard ""
+      return false
+  )
+  log().debug(
+    "Pruned dead connections. $# connections remaining." %
+    [$pool.conns.len])
+
+  let freeConns = pool.conns.filterIt(it.free)
+  if pool.conns.len > pool.cfg.poolSize and freeConns.len > 0:
+    let numToCull = min(freeConns.len, pool.conns.len - pool.cfg.poolSize)
+
+    if numToCull > 0:
+      let toCull = freeConns[0..numToCull]
+      pool.conns.keepIf((pc) => toCull.allIt(it.id != pc.id))
+      for culled in toCull:
+        try: culled.conn.close()
+        except: discard ""
+      log().debug(
+        "Trimming pool size. Culled $# free connections. $# connections remaining." %
+        [$toCull.len, $pool.conns.len])
+
+proc take*(pool: DbConnPool): tuple[id: int, conn: DbConn] =
+  pool.maintain
+  let freeConns = pool.conns.filterIt(it.free)
+
+  log().debug(
+    "Providing a new connection ($# currently free)." % [$freeConns.len])
+
+  let reserved =
+    if freeConns.len > 0: freeConns[0]
+    else: pool.newConn()
+
+  reserved.free = false
+  log().debug("Reserve connection $#" % [$reserved.id])
+  return (id: reserved.id, conn: reserved.conn)
+
+proc release*(pool: DbConnPool, connId: int): void =
+  log().debug("Reclaiming released connaction $#" % [$connId])
+  let foundConn = pool.conns.filterIt(it.id == connId)
+  if foundConn.len > 0: foundConn[0].free = true
+
+template withConn*(pool: DbConnPool, stmt: untyped): untyped =
+  let (connId, conn {.inject.}) = take(pool)
+  try: stmt
+  finally: release(pool, connId)