Connection Pooling for pgvector Ingestion Pipelines

A high-concurrency embedding pipeline and a PostgreSQL server want opposite things from connections. The pipeline runs dozens or hundreds of async consumers, each wanting a connection the instant its embedding call returns; PostgreSQL, by contrast, spends real memory and a backend process per connection and starts to degrade well before max_connections is reached. Between the two sits the pool, and the way it is layered decides whether the pipeline scales smoothly or collapses into too many clients already, idle in transaction leaks, and cryptic prepared-statement errors that only appear behind a proxy. This is the connection-budget problem inside Embedding Ingestion Pipeline Engineering: fan a large, bursty consumer count into a bounded number of physical backends without starving the writers, breaking the vector type adapter, or pinning a connection across a slow embedding-API call. This page compares the three pooling layers, shows how to size them, and codes the one connection-handling rule that matters most.

Up: Embedding Ingestion Pipeline Engineering

Architectural Divergence & Trade-offs #

Pooling happens at three layers, and a production pipeline often stacks them. The choice hinges on where fan-in is absorbed and — for pgvector specifically — whether server-side prepared statements survive.

Driver-native pools (asyncpg.create_pool, psycopg_pool.AsyncConnectionPool) live in the application process. Each holds a fixed set of physical connections and hands one to a coroutine on acquire, returning it on release. Granularity is per-connection for the lifetime of the acquire, so a consumer keeps the same backend for as long as it holds it. This is the simplest layer, keeps prepared-statement caching and session state fully intact, and is the right default for a single writer service. Its ceiling is that each process’s pool consumes real backends; a fleet of ingestion pods multiplies max_size by pod count, and that product must stay under the server budget.

PgBouncer in transaction pooling mode is the fan-in specialist. It multiplexes many client connections onto a small default_pool_size of server connections, assigning a server connection only for the duration of a single transaction and returning it the instant the transaction commits. That granularity is what lets thousands of clients share tens of backends. The cost is severe for a naive pgvector client: because a server connection is not pinned to a client across transactions, server-side prepared statements break — the PS_xxx a client prepared may not exist on the backend it lands on next — and session-level SETs (including anything that configures the session) do not persist. You must disable prepared statements (statement_cache_size=0 in asyncpg) and re-establish per-connection setup, including the vector type adapter, on every physical connection rather than once.

RDS Proxy (or an equivalent cloud proxy) sits outside the application as a managed service, absorbing fan-in like PgBouncer while adding failover: it holds the connection to the database across a failover event so the client sees a brief pause rather than a storm of dropped sockets. It suits serverless and autoscaling ingestion where pod count is elastic and you want connection reuse plus resilience. Like PgBouncer, when it multiplexes it constrains session-scoped features; it can pin a connection to preserve them at the cost of reuse.

A common and sound topology stacks them: driver-native pools inside each pod for vector adaptation and cheap reuse, behind PgBouncer or a cloud proxy that caps the total backend count across the whole fleet. The moment a proxy in transaction mode enters the path, the prepared-statement and per-connection-setup rules below become mandatory.

Parameter Space & Diagnostic Workflow #

Pool sizing is a budgeting exercise with one hard constraint: the sum of every pool’s max_size, across every process, must stay at or below the server’s max_connections (minus a reserve for superuser and maintenance). Oversubscribe it and clients fail to connect; undersize it and consumers block on acquire, turning the pool into the bottleneck.

The primary diagnostic is pg_stat_activity, sliced by application_name and state. It tells you at a glance whether the pool is starved, leaking, or healthy:

SELECT application_name,
       count(*) FILTER (WHERE state = 'active')              AS active,
       count(*) FILTER (WHERE state = 'idle')                AS idle,
       count(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_txn,
       max(extract(epoch FROM now() - state_change))         AS oldest_state_s
FROM   pg_stat_activity
WHERE  application_name = 'embedding_ingest'
GROUP  BY application_name;

A healthy ingest pool shows mostly active connections during a load with a small idle reserve. A non-zero, sustained idle_in_txn is the connection-leak signature described below — a consumer holding a transaction open across an await. active pinned at the pool max_size with consumers still queuing means the pool is the limiter and should be widened (within budget) or matched to the consumer count. Cross-reference against the total backend count to confirm you are within max_connections:

SELECT count(*) AS total_backends,
       current_setting('max_connections')::int AS limit
FROM   pg_stat_activity;

Step-by-Step Implementation #

The following builds an asyncpg pool that adapts the vector type on every physical connection, sizes correctly against the consumer count, and enforces the one rule that prevents connection leaks: acquire only around the write.

Step 1 — Create a driver-native pool with a per-connection vector adapter. The init hook runs once per physical connection, which is exactly where register_vector belongs — so the adapter survives connection recycling and works even behind a transaction-pooling proxy.

import asyncpg
from pgvector.asyncpg import register_vector

async def _init(conn: asyncpg.Connection) -> None:
    await register_vector(conn)          # per physical connection, always

async def make_pool(dsn: str, consumers: int) -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn=dsn,
        min_size=min(10, consumers),
        max_size=consumers,              # one connection per consumer
        max_inactive_connection_lifetime=300.0,
        statement_cache_size=0,          # required behind PgBouncer txn mode
        init=_init,
        server_settings={"application_name": "embedding_ingest"},
    )

statement_cache_size=0 disables asyncpg’s server-side prepared statements so the pool is safe behind PgBouncer transaction pooling; if you run driver-native only, you may leave the cache on for a throughput gain. The psycopg3 equivalent is psycopg_pool.AsyncConnectionPool(..., configure=_setup) where _setup calls register_vector(conn) and the client uses non-prepared execution behind a proxy.

Step 2 — Acquire only around the write, never across the embedding call. This is the rule that keeps the pool from leaking. Compute the vector first — the slow, awaitable network call to the embedding provider — and only then check out a connection for the brief upsert. Holding a connection (or worse, an open transaction) across the embedding await pins a backend for the full round-trip and drives the pool into exhaustion.

UPSERT = """
INSERT INTO doc_embeddings (doc_id, chunk_index, content_hash, embedding)
VALUES ($1, $2, $3, $4)
ON CONFLICT (doc_id, chunk_index) DO UPDATE
   SET embedding = EXCLUDED.embedding,
       content_hash = EXCLUDED.content_hash,
       updated_at = now()
   WHERE doc_embeddings.content_hash IS DISTINCT FROM EXCLUDED.content_hash;
"""

async def handle(item, session, pool):
    vector = await embed(session, item["text"])   # slow: no connection held
    async with pool.acquire() as conn:            # acquire just for the write
        await conn.execute(
            UPSERT, item["doc_id"], item["chunk_index"],
            item["content_hash"], vector,
        )

The transient failures of that embed call — 429s and 5xx — are handled separately by implementing exponential backoff for embedding API calls; note that the backoff waits happen entirely outside the pool.acquire() block, which is the whole point.

Step 3 — Size the pool to the consumer count, within the connection budget. The pool width should equal the number of concurrent consumers in the process so no consumer blocks on acquire after embedding. Then verify the fleet-wide sum fits the server budget.

CONSUMERS = 30
pool = await make_pool(DSN, CONSUMERS)
# fleet budget check: CONSUMERS * pod_count must stay under max_connections

The consumer pool itself and its backpressure model come from async processing with Python asyncio; this pool is the write-side resource those consumers share.

Step 4 — Configure PgBouncer for transaction pooling when fan-in demands it. When many pods must share a small backend count, put PgBouncer in front in transaction mode. default_pool_size — not the application pool size — is the number of real backends per user/database.

[databases]
vectors = host=db.internal port=5432 dbname=vectors

[pgbouncer]
pool_mode = transaction
default_pool_size = 25
max_client_conn = 2000

With this in place the application must use statement_cache_size=0 (Step 1) and rely on the per-connection init hook for register_vector, because a client no longer keeps a fixed backend across transactions.

Validation & Recall Testing #

Pooling defects do not corrupt vectors directly, but they cause dropped or stalled writes that surface later as missing rows and recall gaps. Validate the pool’s behavior first, then the data it produced.

Connection accounting. Under a steady load, active connections for the ingest role should hover near the pool size and idle in transaction should be zero. Run the pg_stat_activity query from the diagnostic section during a load; any sustained idle_in_txn means Step 2’s rule is being violated somewhere in the code path. Confirm the fleet stays within budget by checking total_backends against max_connections.

Adapter correctness. Verify the vector type is actually adapted on pooled connections rather than falling back to a text literal. A quick round-trip through the pool that inserts and reads back a vector, asserting the returned type is a numpy array (asyncpg with pgvector) rather than a string, catches a missing register_vector on the init hook before it becomes a production error.

Completeness after a run. Because pool exhaustion or a leaked connection can silently drop writes into a dead-letter path, reconcile stored chunk counts against the expected manifest:

SELECT doc_id, count(*) AS stored
FROM   doc_embeddings
GROUP  BY doc_id
HAVING count(*) <> (SELECT expected_chunks FROM ingest_manifest m
                    WHERE m.doc_id = doc_embeddings.doc_id);

Rows returned mean chunks were lost — check the connection diagnostics for exhaustion or leaks during that window. For retrieval quality, run a recall-at-k check comparing index answers to an exact numpy ground truth; a gap here points to the ingest data path, not the pool, but the pool is the first thing to rule out when writes went missing.

Failure Modes & Gotchas #

• prepared statement "__asyncpg_stmt_..." does not exist behind PgBouncer. The classic transaction-pooling failure: a client prepared a statement on one backend and the next transaction landed on a different one. Fix by disabling server-side prepared statements (statement_cache_size=0 in asyncpg; use non-prepared execution in psycopg3) whenever a transaction-mode proxy is in the path.
• register_vector missing on a pooled connection. If the adapter is registered once at startup instead of in the pool’s init/configure hook, a recycled or proxy-supplied connection cannot encode the vector type and raises an adaptation error — or worse, sends a text literal that fails to parse. Always register per physical connection.
• Connection leak → idle in transaction. Opening a transaction (or holding pool.acquire()) across the embedding-API await pins a backend for the whole round-trip. Under load the pool drains into idle in transaction, consumers block on acquire, and throughput collapses. Acquire only around the write; keep the embedding call and its backoff outside the connection scope.
• Pool exhaustion → latency cliff. A pool narrower than the consumer count means consumers queue on acquire after embedding, so latency climbs sharply while the database looks idle. Match max_size to the consumer count, and set an acquire timeout so a genuinely stuck pool fails fast instead of hanging.
• Oversubscribing max_connections. Summing pool max_size across an autoscaled fleet quietly exceeds the server budget as pods scale out, producing FATAL: sorry, too many clients already. Cap total backends with a proxy’s default_pool_size rather than trusting each pod to self-limit.
• Session SETs that silently vanish. Any per-session configuration set once after connecting is lost under transaction pooling because the next transaction may use a different backend. Move such settings into the per-connection init hook or into the connection string’s options.

Monitoring & Alerting Hooks #

Monitor the pool from both ends: the in-process pool’s own counters and the server’s view of the backends it produces. On the application side, export the pool’s free/used sizes and the acquire wait time as gauges — a rising acquire-wait with a fully used pool is the earliest, most actionable saturation signal, well before latency visibly degrades.

On the database side, alert on the three states that map directly to the failure modes above:

SELECT count(*) FILTER (WHERE state = 'active')              AS active,
       count(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_txn,
       count(*)                                              AS total,
       current_setting('max_connections')::int              AS max_conn
FROM   pg_stat_activity
WHERE  application_name = 'embedding_ingest';

Page on idle_in_txn > 0 sustained beyond a few seconds (the connection-leak signature), on active approaching the pool max_size (the pool is the limiter — widen it or the consumer count, within budget), and on total approaching max_conn fleet-wide (backend oversubscription — cap it at the proxy). A Prometheus rule expressing the saturation condition reads naturally as pg_stat_activity_count{state="active"} / pg_settings_max_connections > 0.8. Track write progress with the delta of pg_stat_user_tables.n_tup_ins + n_tup_upd; a flat line while the application pool reports high acquire-wait means writes are queued on connections, not slow on the server.

FAQ #

Related #

• Async Processing with Python AsyncIO — the consumer pool that shares this connection pool as its write resource
• Implementing exponential backoff for embedding API calls — retry waits that must stay outside the connection scope
• Batch Upsert Patterns for pgvector — the write strategies these pooled connections execute
• Metadata Mapping & Schema Design — the schema whose conflict keys the write path depends on
• Up: Embedding Ingestion Pipeline Engineering