gpool

README

Gpool: A Go Connection Pooling & CDC Library for PostgreSQL

Gpool is a Go 1.26-native library for PostgreSQL connection pooling and Change Data Capture (CDC). It is designed to be embedded in your application or composed into another library — there is no daemon, no CLI, no config file, no logger, and no process-global state.

📦 Installation

go get github.com/gsoultan/gpool@v0.1.0

While the major version is 0 the API may change in a minor release, so pin an exact version. See CHANGELOG.md for the versioning policy and what v1.0.0 will require.

🧱 Architecture

Gpool separates interfaces from vendor implementations, so consumers depend only on behaviour:

Package	Contents
pkg/gpool	Pool, Conn, Tx, Rows, Row, Result, Stat, Engine, vendor registry, and the optional capabilities BulkCopier, Batcher, Notifier
pkg/gpool/cdc	Subscriber = Stream + TableManager + ReplicationManager, plus EventStream and Event
pkg/vendors/postgres/pool	PostgreSQL pool implementation (pgx/v5)
pkg/vendors/postgres/cdc	PostgreSQL logical-replication implementation (pglogrepl)

Vendors self-register through init(), in the style of database/sql drivers. Importing the vendor package is what makes gpool.Postgres resolvable:

import (
    "github.com/gsoultan/gpool/pkg/gpool"

    // Registers the "postgres" pool factory.
    postgrespool "github.com/gsoultan/gpool/pkg/vendors/postgres/pool"
)

If you only need the interfaces — for example to accept a gpool.Pool in your own library's API — import pkg/gpool alone and let the caller choose the vendor. The CDC interfaces are segregated, so a consumer that only reads changes can depend on cdc.Stream rather than the full cdc.Subscriber.

🛠️ Pool Usage

import (
    "github.com/gsoultan/gpool/pkg/gpool"
    postgrespool "github.com/gsoultan/gpool/pkg/vendors/postgres/pool"
)

pool, err := gpool.NewPool(gpool.Postgres, postgrespool.Config{
    ConnString: "postgres://user:pass@localhost:5432/dbname",
    MaxConns:   50,
    MinConns:   5,
})
if err != nil {
    return err
}
defer pool.Close()

rows, err := pool.Query(ctx, "SELECT id, name FROM users WHERE active = $1", true)
if err != nil {
    return err
}
defer rows.Close() // safe: All() also closes, and Close is idempotent

for row := range rows.All() {
    var id int
    var name string
    if err := row.Scan(&id, &name); err != nil {
        return err
    }
}
return rows.Err()

Explicit acquisition and transactions:

conn, err := pool.Acquire(ctx)
if err != nil {
    return err
}
defer conn.Release()

tx, err := conn.Begin(ctx)
if err != nil {
    return err
}
defer func() { _ = tx.Rollback(ctx) }() // returns ErrTxClosed after a commit; safe to ignore

if _, err := tx.Exec(ctx, "UPDATE users SET active = $1 WHERE id = $2", false, id); err != nil {
    return err
}
return tx.Commit(ctx)

Lifecycle guarantees

Every teardown method is idempotent, because the idiomatic Go patterns call them more than once:

Call	Repeat behaviour
Pool.Close	No-op
Conn.Release	No-op; later use returns ErrConnReleased
Rows.Close	No-op; later use returns ErrRowsClosed
Row.Scan / Row.Release	Either settles the row; the other becomes a no-op
Tx.Commit / Tx.Rollback	Whichever runs second returns ErrTxClosed
Engine.Close	Replays the first result

Rows from Pool.Query and Row from Pool.QueryRow own the connection and return it when they are closed. For QueryRow that means calling either Scan or Release — if you decide not to read the result, Release still closes the query and frees the connection.

The Row yielded by Rows.All() is a cursor over the current row, valid only until the loop advances. Scan what you need inside the loop body.

Pool configuration

Field	Default	Notes
ConnString	required	Parsed once at construction, then cloned per connection
MaxConns	max(4, GOMAXPROCS)	Hard cap; Acquire blocks at the limit
MinConns	0	Kept warm by the background reaper
MaxConnLifetime	1h	What lets the pool recover from a failover; negative disables
MaxConnIdleTime	30m	Negative disables
HealthCheckPeriod	1m	Reaper interval; negative disables
ResetQuery	""	See below
ResetQueryTimeout	5s
ConnectTimeout	10s	Used when the caller's context has no earlier deadline
StatementCacheCapacity	64	Per-connection prepared-statement cache; DisableCache turns it off
DescriptionCacheCapacity	64	Per-connection statement-description cache
BeforeConnect	nil	Adjust the config per connection, e.g. credential rotation
AfterConnect	nil	Register types or run SET; an error discards the connection

A connection that is closed, has outlived its bounds, or fails its reset query is destroyed rather than returned to the pool.

The two cache capacities are the largest per-connection memory cost. pgx defaults both to 512 and preallocates each one's map at that size, which measured as 57% of the pool's entire heap. gpool defaults them to 64 — within 10% of the floor while still caching a typical service's query set. Raise it if your query variety is wide; lower it if connection count matters more. See Footprint at scale.

Pooling modes and prepared statements

gpool has no pooling-mode switch. The mode is decided by how long you hold a connection, which makes the PgBouncer equivalents a matter of usage rather than configuration:

PgBouncer mode	gpool equivalent	Prepared statements
statement	pool.Query / pool.Exec / pool.QueryRow — acquired and released per call	✅ cached per connection
transaction	Acquire → Begin → Commit → Release	✅ cached per connection
session	Acquire once, hold it, Release at the end — plus ResetQuery: "DISCARD ALL"	❌ traded away, see below

Prepared statements work by default. gpool caches them per connection (pgx.QueryExecModeCacheStatement), and because a connection is only ever handed to one caller at a time, the cache stays valid across reuse. Nothing about pooling by itself invalidates it — this is the advantage of pooling in-process over a wire proxy, where a client's statement names follow it onto whichever backend it lands on.

What returns a connection to a clean state. On release gpool refuses to recycle a connection that is closed or mid-protocol with an unread result, and cleans up two kinds of leftover state:

Leftover	What gpool does	Why it matters
Open or failed transaction	ROLLBACK	The next caller would inherit the locks and snapshot, or a session rejecting everything with SQLSTATE 25P02
Active LISTEN	UNLISTEN *	The next caller would silently receive notifications it never subscribed to

Both cost one round trip and only on connections that actually need them — versus a full reconnect if the connection were simply discarded. A cleanup that fails means the state is unknown, so that connection is destroyed rather than reused.

ResetQuery is the session-mode option, and it costs you prepared statements. DISCARD ALL deallocates the server's prepared statements, so gpool switches the connection to pgx.QueryExecModeExec — parameters still bound server-side, but nothing cached by name. Without that switch the driver's cache goes stale and the next query fails with SQLSTATE 26000. It also costs a full extra round trip on every release, roughly doubling per-query latency.

Set ResetQuery only when callers hold connections long enough to accumulate session state (SET, LISTEN, temp tables, WITH HOLD cursors). For statement- and transaction-scoped usage the rollback gate above is already sufficient, and leaving ResetQuery empty is both faster and keeps prepared statements. BeforeConnect can override the execution mode if your reset query preserves them.

⚡ CDC Streaming

import (
    "github.com/gsoultan/gpool/pkg/gpool"
    postgrescdc "github.com/gsoultan/gpool/pkg/vendors/postgres/cdc"
)

subscriber, err := gpool.NewSubscriber(gpool.Postgres, postgrescdc.Config{
    ConnString:        "postgres://user:pass@localhost:5432/dbname",
    SlotName:          "gpool_slot",
    PublicationName:   "gpool_pub",
    Tables:            []string{"public.users"},
    CreateSlot:        true,
    CreatePublication: true,
})
if err != nil {
    return err
}
defer subscriber.Close()

stream, err := subscriber.Subscribe(ctx)
if err != nil {
    return err
}
defer stream.Close()

for event := range stream.All() {
    fmt.Printf("%s %s.%s lsn=%d after=%v\n",
        event.Op, event.Schema, event.Table, event.LSN, event.After)
}
return stream.Err()

Requires wal_level=logical and a role with the REPLICATION attribute. Creating the publication additionally needs ownership of the tables.

CDC cannot be pooled, and does not use prepared statements. A replication stream opens with replication=database, enters CopyBoth after START_REPLICATION, and stays there for the life of the subscription — there is no transaction boundary at which a connection could be handed back, and a slot admits one active walsender at a time. It is inherently session-scoped, and PgBouncer does not proxy replication connections at all. Point the subscriber at the database directly, not at a pooler.

The subscriber's control connection — publication DDL and pg_publication_tables lookups — is an ordinary connection issuing discrete statements with unnamed parameter binding, so it holds no server-side prepared statements and is safe through a pooler in transaction mode.

Delivery semantics

Delivery is at-least-once. The stream confirms a WAL position to the server only after the iterator body for that event has returned, so a crash mid-processing replays the event rather than losing it.

The corollary: a consumer that hands work to another goroutine and returns immediately has confirmed work it has not done. Either finish processing before returning from the loop body, or persist Event.LSN yourself and pass it back as Config.StartLSN on the next run.

Config.StartLSN defaults to 0, which resumes from the slot's confirmed_flush_lsn — that is, it replays everything the slot retained while the consumer was away.

Replication slots retain WAL. A consumer that stops draining grows the primary's disk usage until it fills. Slots are never dropped automatically; use DropSlot deliberately when you are done with one. During idle periods the stream advances its confirmed position from the server's keepalives, so a quiet publication does not pin WAL.

Event shape

type Event struct {
    Op     Op                // OpInsert, OpUpdate, OpDelete
    Schema string
    Table  string
    LSN    uint64            // WAL position immediately after this record
    Before map[string]any    // Update and Delete, subject to REPLICA IDENTITY
    After  map[string]any    // Insert and Update
}

• Before and After are allocated per event and owned by you. They are safe to retain and to send to another goroutine.
• Values are string, exactly as pgoutput transmits them in text format. The replication stream does not carry the destination Go type, so decoding is left to the consumer.
• A column absent from the map was not transmitted: either REPLICA IDENTITY does not cover it, or it is an unchanged TOASTed value. That is distinct from a key present with a nil value, which is a real SQL NULL. Writing an absent column as NULL would blank it on replay.
• Before is nil under REPLICA IDENTITY DEFAULT unless the key changed. Use ALTER TABLE ... REPLICA IDENTITY FULL for a complete before-image.

Table management

Table management runs on a separate control connection, so it is safe while a stream is live:

err := subscriber.AddTables(ctx, "public.orders", "public.products")
err  = subscriber.RemoveTables(ctx, "public.products")
err  = subscriber.SyncTables(ctx, "public.users", "public.orders") // reconcile to an exact list

tracking := subscriber.IsTracking("public.users") // local view
tables   := subscriber.GetTables()                // local copy
ok, err  := subscriber.VerifyTable(ctx, "public.users") // queries pg_publication_tables

The local tracking list is updated only after the server accepts the change, so IsTracking never reports a table the publication does not actually have. Already-tracked additions and untracked removals are skipped rather than sent.

Table names may be schema-qualified; the schema and table parts are quoted separately.

Slot and publication lifecycle

err := subscriber.CreateSlot(ctx, "manual_slot")      // no-op if it exists
err  = subscriber.DropSlot(ctx, "manual_slot")        // no-op if it does not
err  = subscriber.CreatePublication(ctx, "manual_pub", "public.t1", "public.t2")
err  = subscriber.DropPublication(ctx, "manual_pub")

CDC configuration

Field	Default	Notes
ConnString	required	The replication parameter is added internally where needed
SlotName	required	Must match [a-z0-9_]{1,63} — PostgreSQL's own rule
PublicationName	required
Tables	required with CreatePublication
StartLSN	0	Resume from the slot's confirmed position
StandbyInterval	10s	Must stay well under the server's wal_sender_timeout
Buffer	256	Read-ahead depth in events

Only one stream may be open per subscriber; a second Subscribe returns ErrAlreadySubscribed.

🧩 Engine

Engine is an optional facade that owns any number of named pools and CDC subscribers, so a host application can shut everything down through a single Close. Both constructor arguments may be nil.

engine := gpool.NewEngine(pool, subscriber) // registered as gpool.DefaultPool / DefaultSubscriber

engine.AddSubscriber("analytics", analyticsSubscriber)
analytics := engine.Subscriber("analytics")
names := engine.Subscribers()
err := engine.RemoveSubscriber("analytics") // closes it

defer engine.Close() // closes every subscriber and every pool, joining errors

Close is idempotent and one component failing to close does not stop the others. AddPool and AddSubscriber replace an existing registration without closing the displaced one — the caller may still be holding it.

🗄️ Multiple Databases

Reaching several databases from one process means one pool per database, selected by name. Nothing is shared between them: separate connection strings, separate capacity, separate lifecycles. A database that saturates or fails cannot starve another.

engine := gpool.NewEngine(nil, nil)

for name, connString := range map[string]string{
    "primary":   "postgres://user:pass@host-a:5432/app",
    "analytics": "postgres://user:pass@host-b:5432/warehouse",
} {
    pool, err := gpool.NewPool(gpool.Postgres, postgrespool.Config{
        ConnString: connString,
        MaxConns:   25, // capacity is per database
    })
    if err != nil {
        return err
    }
    engine.AddPool(name, pool)
}
defer engine.Close() // closes every pool

// Route by name.
var total int
err := engine.Pool("analytics").
    QueryRow(ctx, "SELECT count(*) FROM events").Scan(&total)

names := engine.Pools()        // ["primary", "analytics"]
engine.RemovePool("analytics") // unregisters and closes it

engine.Pool() with no argument returns the pool registered under gpool.DefaultPool, so single-database code needs no changes.

The same works for CDC: register one subscriber per database with AddSubscriber. Each holds its own replication slot, which is the correct topology — sharing a slot between consumers loses data.

🧰 Optional Capabilities

Bulk copy, batching, and LISTEN/NOTIFY are exposed as separate interfaces rather than piled onto Pool and Conn, so neither grows past what every caller needs. Reach them with a type assertion.

BulkCopier — load many rows in one pass over the COPY protocol, far faster than the equivalent INSERTs. Available on both Pool and Conn. COPY is atomic: a source that fails part-way leaves nothing behind.

copier, ok := pool.(gpool.BulkCopier)
if !ok {
    return errors.New("vendor does not support bulk copy")
}

copied, err := copier.CopyFrom(ctx, gpool.CopyRequest{
    Table:   []string{"public", "users"}, // parts, so a dotted name is still addressable
    Columns: []string{"id", "email"},
    Rows: gpool.CopyFromSlice(len(users), func(i int) ([]any, error) {
        return []any{users[i].ID, users[i].Email}, nil
    }),
})

CopyRows is a cursor, not a slice, so you can stream a dataset larger than memory.

Batcher — pipeline statements so N of them cost roughly one round trip instead of N. Available on Conn only: the result reader stays valid just while the connection is held.

batch := &gpool.Batch{}
batch.Queue("INSERT INTO events (kind) VALUES ($1)", "click")
batch.Queue("INSERT INTO events (kind) VALUES ($1)", "view")
batch.Queue("SELECT count(*) FROM events")

results := conn.(gpool.Batcher).SendBatch(ctx, batch)
defer results.Close() // required: unread replies would desynchronise the connection

for range 2 {
    if _, err := results.Exec(); err != nil {
        return err
    }
}
var total int
if err := results.QueryRow().Scan(&total); err != nil {
    return err
}
return results.Close()

Notifier — LISTEN/NOTIFY, on Conn only. A subscription belongs to the session, so it only means anything on a connection you continue to hold; a pool-level Listen would register on whichever connection served the call and then hand it to someone else. gpool clears the subscription on release, so it cannot leak onward.

conn, _ := pool.Acquire(ctx)
defer conn.Release()

listener := conn.(gpool.Notifier)
if err := listener.Listen(ctx, "events"); err != nil {
    return err
}
for {
    notification, err := listener.WaitForNotification(ctx)
    if err != nil {
        return err
    }
    handle(notification.Channel, notification.Payload)
}

📊 Observability

Gpool exports no metrics and starts no HTTP server. Poll Pool.Stat() — it is lock-free — and feed it to whatever collector your application already uses.

Stat composes two interfaces, so a consumer that only graphs utilisation can depend on Occupancy alone:

stat := pool.Stat()

// Occupancy: a point-in-time view.
stat.TotalConnections()
stat.IdleConnections()
stat.ActiveConnections()
stat.MaxConnections()   // so a dashboard can show utilisation without being told the config

// Acquisition: cumulative counters.
stat.AcquireCount()          // successful acquisitions
stat.AcquireDuration()       // total time callers spent waiting
stat.EmptyAcquireCount()     // acquisitions that had to wait
stat.CanceledAcquireCount()  // acquisitions whose context ended first

Occupancy alone cannot answer the question that actually matters when sizing a pool — is MaxConns too low? — because a pool that is permanently full looks identical to one that is merely busy. The acquisition counters separate them:

Signal	Reading
EmptyAcquireCount near zero	Callers never wait; the pool is large enough
EmptyAcquireCount tracking AcquireCount	Every caller queues; raise MaxConns
AcquireDuration / EmptyAcquireCount rising	Waits are getting longer, not just more frequent
CanceledAcquireCount rising	Callers are timing out before they get a connection

Only waits are timed. An acquisition served immediately contributes nothing to AcquireDuration, which keeps the uncontended path free of a clock read it does not need.

🚨 Errors

All sentinels are comparable with errors.Is.

pkg/gpool: ErrVendorNotRegistered, ErrNilFactory, ErrEmptyVendor.

pkg/vendors/postgres/pool: ErrPoolClosed, ErrConnReleased, ErrTxClosed, ErrRowsClosed, ErrInvalidConfig.

pkg/vendors/postgres/cdc: ErrClosed, ErrAlreadySubscribed, ErrNoTables, ErrInvalidConfig.

🔌 Adding a Vendor

Implement gpool.Pool (and optionally cdc.Subscriber) and register a factory from your package's init():

func init() {
    _ = gpool.RegisterPool(gpool.Vendor("mysql"), func(config any) (gpool.Pool, error) {
        cfg, ok := config.(Config)
        if !ok {
            return nil, fmt.Errorf("%w: expected %T, got %T", ErrInvalidConfig, Config{}, config)
        }
        return New(cfg)
    })
}

The registry is mutex-guarded, so registration is safe to interleave with NewPool.

🧪 Testing

Unit tests need no database:

go test -race ./pkg/...

Integration tests and benchmarks need a server with logical replication enabled:

podman run -d --rm --name gpool-test \
  -e POSTGRES_PASSWORD=postgres -p 55432:5432 \
  docker.io/library/postgres:17-alpine \
  -c wal_level=logical -c max_replication_slots=10 -c max_wal_senders=10

export DATABASE_URL='postgres://postgres:postgres@127.0.0.1:55432/postgres'
go test -race ./integration/
go test -bench=. ./benchmarks/

📉 Footprint at Scale

A pooler's job is to absorb far more callers than it holds connections, so those two numbers are measured separately. Apple M5 Pro, PostgreSQL 17 over loopback, GOMAXPROCS=15.

Per connection. Measured by TestMemoryPerPooledConnection, which holds every connection open and reads steady-state heap:

StatementCacheCapacity	KiB / connection	Projected at 5,000 connections
DisableCache	25.8	126 MiB
16	26.1	128 MiB
64 (default)	28.5	139 MiB
128	36.2	177 MiB
512 (pgx's default)	71.2	348 MiB

Per caller: nothing. The pool adds exactly one goroutine — the background maintainer — regardless of connection count or client concurrency, and Close reclaims it. Verified by TestGoroutineCostAtScale.

Note the two readings of "5,000 connections". Five thousand callers over a bounded pool is the normal shape and costs MaxConns × 28.5 KiB — about 3 MiB for a 100-connection pool. Five thousand actual backend connections is the 139 MiB row above, and is usually the wrong design: PostgreSQL itself allocates several MiB of backend memory per connection, so the server is the binding constraint long before gpool is.

📈 Benchmarks

benchmarks/ compares gpool against pgxpool, database/sql, and PgBouncer on the same workload, with pool capacity matched so the comparison measures the driver rather than queueing.

Query path — -benchtime=20000x, default parallelism, capacity max(4, GOMAXPROCS) on both:

Benchmark	ns/op
PgxPool	45,956 – 47,229
Stdlib	46,882 – 48,973
GpoolQueryRow	46,393 – 47,096

A query is dominated by the network round trip, so all three land in the same band across repeated runs. gpool adds no measurable overhead to the query itself.

Pool mechanics — what gpool actually controls, -benchtime=200000x:

Benchmark	ns/op	B/op	allocs/op
GpoolAcquireRelease (default parallelism)	257	48	1
ScaleAcquireRelease (5,000 callers, 16 conns)	574	65	1
ScaleAcquireRelease (5,000 callers, 64 conns)	744	52	1
ScaleAcquireRelease (5,000 callers, 256 conns)	537	52	1
ScaleStatUnderLoad (scraping under traffic)	30	19	1

The single allocation is the per-acquisition handle, which is deliberately not pooled — recycling an object whose lifetime the caller controls is what made double-Release corrupt the pool. Stat is lock-free, so a metrics scrape does not serialise against acquisition.

ResetQuery costs a full extra round trip: GpoolResetQuery measures ~103,000 ns/op against ~47,000 without it. Enable it only when session isolation is required.

Rerun on your own hardware before quoting any of this; loopback flatters every number here.

🛡️ License

Copyright (c) 2026 Gembit Soultan Shirazi. All rights reserved. Licensed under the MIT License.