postgres.pool

A ConnectionPool keeps a small set of open sessions and hands them out one at a time. It is built for a long-lived service: one pool for the process lifetime, several threads checking connections out of it, and connections recycled underneath so a failover or a nightly restart is survivable without a restart of your own.

from postgres.pool import ConnectionPool, PoolConfig

var cfg = PoolConfig(max_size=8, acquire_timeout_ms=2000)
var pool = ConnectionPool("postgresql://localhost/app?connect_timeout=5", cfg)

with pool.lease() as lease:
    var res = lease.connection().query("SELECT id FROM t WHERE id = $1",
                                       Params().int64(1))
    for row in res:
        print(row.int64("id"))

This module is not re-exported from postgres. Import it by name, as above: it pulls in the threads tin for its mutex and condition variable, and a single-threaded caller should not have to carry that.

The one invariant

Connection shares its PGconn with every connection.Statement, connection.Transaction, copy.CopyIn and copy.CopyOut made from it, so that a handle can keep the session open after the Connection value is gone. That is deliberate, documented, and exactly what makes pooling delicate: libpq forbids two threads touching one PGconn, and breaking that rule corrupts the protocol stream silently rather than crashing.

So the pool's invariant is:

A connection goes back into the pool only when nothing else is holding it.

Lease enforces it rather than asking for it. When the lease ends it reads Connection._shares; anything above one means a handle outlived the block, and the connection is closed instead of pooled and counted in PoolStats.escaped. The escapee then gets SQLSTATE 08006 from its next call, which is loud, local and immediate.

The pool's promise is therefore the same shape as connection.Transaction's: the failure mode of losing track is "you lost a connection", never "two threads share one".

Scope the lease with with

with pool.lease() as lease:
    _ = lease.connection().execute("...")

Leaving the block ends the lease, however it is left -- off the end, by return, or by a raise. Mojo destroys a value after its last use rather than at the end of its scope, so a lease held in a plain var also ends earlier than the indentation suggests; with is what makes the extent visible, and it is the supported form.

A transaction is expected to begin and end inside one lease. That is not an honour system either: a connection.Transaction still alive when the lease ends is a share, so the connection is discarded rather than handed on with a block open.

What happens on return

In order, and only when nothing escaped:

1. If PQtransactionStatus is not idle, ROLLBACK. A lease that left a block open, or left it failed, does not leak that into the next caller.
2. With PoolConfig.on_return set to ON_RETURN_DISCARD_ALL, DISCARD ALL. Opt-in, because it is safer and throws away the prepared statements a service reissuing the same queries is paying to keep.
3. If the session is older than PoolConfig.max_lifetime_ms, or no longer alive, it is discarded rather than pooled.

What happens on checkout

An idle connection is taken most-recently-returned first, then checked: expired by PoolConfig.max_lifetime_ms or PoolConfig.max_idle_ms, or no longer alive. A connection that fails the check is recycled in place with Connection.reset -- one blocking reconnect on a PGconn that is already allocated -- and only closed outright if that fails too.

The liveness check is Connection.is_alive: a non-blocking read of what the server has already sent, with no round trip. It is not Connection.ping, which is a SELECT 1 per checkout, and it is deliberately not Connection.is_open, which cannot see this at all -- PQstatus reports a cached opinion, and a backend killed by pg_terminate_backend still reads CONNECTION_OK until a command fails on it.

No local check can be certain; a connection can die in the microsecond after it passes. The point is not certainty, it is that a known-dead connection is never handed out, and that a connection which dies in a caller's hands is discarded on return instead of going back in the pool.

Growth, waiting and reaping

The pool grows lazily to PoolConfig.max_size. PoolConfig.min_idle is opened at construction and kept warm; everything above that is opened when somebody actually waits for it, because opening blocks and pre-opening the whole pool would turn a typo in the conninfo into a slow startup failure.

A caller that finds every connection busy blocks on a real pthread_cond_t -- no spin, no poll -- until one comes back or PoolConfig.acquire_timeout_ms elapses, at which point ConnectionPool.lease raises SQLSTATE 53300.

Reaping is lazy, plus an explicit ConnectionPool.reap. Idle connections are aged out when they are next picked up, and ConnectionPool.reap does the same sweep on demand, closing anything past PoolConfig.max_idle_ms (never below PoolConfig.min_idle) and reopening back up to PoolConfig.min_idle. There is deliberately no reaper thread: a thread inside a library owes its caller a lifecycle -- when it starts, how it stops, what happens if the pool is dropped while it sleeps -- and a detached thread waiting on a condition variable that is about to be destroyed is precisely the use-after-free this module exists to rule out. A service that must not pin backends overnight already has a timer; ConnectionPool.reap is one call on it, and it is the same code the checkout path runs.

Threads

ConnectionPool is the owner. ConnectionPool.ref hands out a PoolRef, which is copyable, shares the same state, and is what travels into a worker thread's context. Every PoolRef and every outstanding Lease holds a share of the pool's state, so the state cannot be destroyed underneath a thread that is still using it.

A lease may outlive the pool value, and routinely does: Mojo destroys a value at its last use, so a ConnectionPool whose final mention is pool.lease() is gone before the lease it produced, leaving that lease holding the last share. Returning the connection then means returning it to a pool nobody else can see, which is fine and is what the share is for -- but it also means the pool's mutex is destroyed by the lease that is using it, so every critical section in this module lives in a function that takes the state as a borrowed argument. A borrowed argument cannot be destroyed during the call, which is what keeps the mutex alive across its own unlock; see _put_back.

ConnectionPool.__init__ checks PQisthreadsafe and refuses to build a pool on a libpq that was compiled without thread safety, because nothing here -- or anywhere else -- is safe on one.

struct PoolConfig

var cfg = PoolConfig(
    max_size=16,
    min_idle=2,
    acquire_timeout_ms=2000,
    max_lifetime_ms=900_000,
)

def __init__(out self, max_size: Int = Int(10), min_idle: Int = Int(0), acquire_timeout_ms: Int = Int(5000), max_lifetime_ms: Int = Int(1800000), max_idle_ms: Int = Int(600000), on_return: Int = Int(0), validate_on_checkout: Bool = True)

struct PoolStats

def write_to(self, mut writer: T)

struct Lease

with pool.lease() as lease:
    var res = lease.connection().query("SELECT 1")
    print(res.num_rows())

def __init__(out self, state: ArcPointer[_PoolState], var conn: Connection, opened_at_ms: Int)

def __deinit__(deinit self)

def __enter__(deinit self) -> Self

def connection(mut self) -> ref[self._conn] Connection

def backend_pid(self) -> Int

struct PoolRef

@fieldwise_init
struct Ctx(Copyable, Movable):
    var pool: PoolRef

def worker(i: Int, mut ctx: Ctx) -> None:
    try:
        with ctx.pool.lease() as lease:
            _ = lease.connection().execute("SELECT 1")
    except:
        pass

parallel_for[worker](64, ctx, num_workers=8)

def lease(self) -> Lease

def stats(self) -> PoolStats

def reap(self)

struct ConnectionPool

from postgres.pool import ConnectionPool, PoolConfig

var pool = ConnectionPool(
    "postgresql://localhost/app?connect_timeout=5",
    PoolConfig(max_size=8, min_idle=1, acquire_timeout_ms=2000),
)

with pool.lease() as lease:
    _ = lease.connection().execute("DELETE FROM sessions WHERE expired")

def __init__(out self, conninfo: String, config: PoolConfig = PoolConfig(Int(10), Int(0), Int(5000), Int(1800000), Int(600000), Int(0), True))

def ref(self) -> PoolRef

def lease(self) -> Lease

def stats(self) -> PoolStats

def reap(self)

def close(self)

comptime ON_RETURN_ROLLBACK = Int(0)

comptime ON_RETURN_DISCARD_ALL = Int(1)