Pooling Connections for Parallel DSR Discovery

Within the broader Database Connector Configuration layer — itself part of the Cross-System Data Discovery & Sync architecture — the hardest scaling problem is not reading one database quickly, it is reading many databases at once without knocking any of them over. A single Data Subject Request (DSR) may need a subject’s rows from a dozen relational stores simultaneously, and the naive fix — open a connection per store per request and fan out with asyncio.gather — quietly exhausts the very production databases you are trying to search. The team that hits this is any privacy or platform engineering group whose discovery worker suddenly competes with live OLTP traffic for a finite max_connections budget, so a burst of concurrent DSRs degrades checkout latency for real users while the statutory clock keeps ticking. This page shows how to bound that concurrency with a connection pool per store and a global fan-out limit, so parallel discovery stays fast, read-only, and provably within the response window GDPR Article 12(3) fixes at one month and CCPA §1798.130(a)(2) fixes at 45 days. It builds directly on the pool primitives introduced when connecting PostgreSQL and Snowflake for DSR discovery, extending them from one bridge to a whole fleet of stores.

A pool does two things at once here: it caps how many sessions any one store ever sees, and — paired with a semaphore — it caps how many stores you touch at the same instant. The diagram shows both limits acting together.

Prerequisites #

This implementation targets Python 3.11+ (for mature asyncio cancellation and tuple[...] generics) and the following libraries:

• psycopg[pool] (psycopg 3.1+) — the modern PostgreSQL driver whose psycopg_pool.AsyncConnectionPool gives a native async pool with checkout timeouts and connection lifecycle hooks. If your codebase standardizes on SQLAlchemy, its QueuePool (with pool_size, max_overflow, and pool_timeout) exposes the same three controls and the sizing logic here transfers unchanged.
• pydantic (v2) — a frozen, typed PoolConfig validated at startup so a misconfigured pool fails fast rather than mid-sweep. Uses v2 idioms (model_config = ConfigDict(...), field_validator); v1 patterns will not work.
• A read-replica endpoint per store — discovery must never point at the write primary. Reading from a replica is what keeps a burst of DSRs from stealing connection slots and I/O from live transactions.
• A dedicated read-only role per store — a least-privilege role scoped to the discovery tables, with no INSERT/UPDATE/DELETE/DDL grant, so you can attest to a regulator that discovery is physically incapable of mutating a subject’s record.

Credentials are resolved at call time from a secret manager, consistent with the key-lifecycle guidance in NIST SP 800-57 Part 1; they are never baked into images or committed to configuration.

Step-by-step implementation #

Step 1 — Model the pool configuration with Pydantic v2 #

Every knob that governs pool behavior is described by a frozen model so it is validated once, at startup, and cannot drift mid-run. Separating max_size (per-store session cap) from max_parallel (global store fan-out) in the type system is deliberate: they bound different resources and are tuned against different limits.

from pydantic import BaseModel, ConfigDict, SecretStr, field_validator


class PoolConfig(BaseModel):
    """Validated, immutable configuration for one store's discovery pool."""

    model_config = ConfigDict(frozen=True)

    store_id: str
    conninfo: SecretStr              # libpq DSN pointing at the READ REPLICA
    min_size: int = 1
    max_size: int = 4                # peak sessions this store will ever see
    pool_timeout_s: float = 5.0      # fail fast if no slot frees within budget
    open_timeout_s: float = 10.0
    max_lifetime_s: float = 600.0    # recycle sessions to shed replica-side state
    statement_timeout_ms: int = 30_000
    application_name: str = "dsr-discovery-pool"

    @field_validator("max_size")
    @classmethod
    def cap_ge_min(cls, v: int, info) -> int:
        if v < 1:
            raise ValueError("max_size must be at least 1")
        return v

SecretStr keeps the DSN out of repr() and logs, and frozen=True prevents a resolved connection string from being mutated during a sweep.

Step 2 — Configure a bounded, read-only pool per store #

Each store gets its own AsyncConnectionPool. The max_size ceiling is the single most important compliance control on this page: it is the maximum number of sessions this store’s replica will ever receive from discovery, no matter how many DSRs arrive at once. Every connection is forced read-only and given a server-side statement_timeout, so a pathological plan cannot outlive its budget.

from psycopg_pool import AsyncConnectionPool


async def make_store_pool(cfg: PoolConfig) -> AsyncConnectionPool:
    """Open one bounded, read-only pool against a store's read replica."""
    pool = AsyncConnectionPool(
        conninfo=cfg.conninfo.get_secret_value(),
        min_size=cfg.min_size,
        max_size=cfg.max_size,
        timeout=cfg.pool_timeout_s,           # PoolTimeout if no slot frees in time
        max_lifetime=cfg.max_lifetime_s,
        kwargs={
            "application_name": cfg.application_name,
            "options": (
                f"-c statement_timeout={cfg.statement_timeout_ms} "
                "-c default_transaction_read_only=on"
            ),
        },
        open=False,
    )
    await pool.open(wait=True, timeout=cfg.open_timeout_s)
    return pool

default_transaction_read_only=on enforces the least-privilege guarantee at the session level even if the role grant is ever loosened by mistake, and application_name tags every session in pg_stat_activity so an auditor can attribute each discovery query to the DSR worker rather than to an anonymous connection.

Step 3 — Run a bounded fan-out across every store #

The per-store max_size caps sessions within a store; an asyncio.Semaphore caps how many stores run at the same instant. Without the semaphore, one thousand concurrent DSRs would try to touch every store simultaneously and the aggregate connection demand across the fleet would still spike — even with each pool bounded — because a thousand tasks would each be queued at checkout. The semaphore keeps the live working set small and predictable.

import asyncio
from psycopg_pool import AsyncConnectionPool


async def discover_store(pool: AsyncConnectionPool, subject_id: str) -> list[dict]:
    """Borrow one pooled session and read a subject's rows read-only."""
    async with pool.connection() as conn:            # returned to pool on exit
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT record_id, payload FROM subject_records "
                "WHERE subject_id = %s",
                (subject_id,),
            )
            cols = [d.name for d in cur.description]
            return [dict(zip(cols, row)) for row in await cur.fetchall()]


async def discover_all_stores(
    pools: dict[str, AsyncConnectionPool], subject_id: str, max_parallel: int = 6
) -> dict[str, list[dict] | Exception]:
    """Fan out discovery across stores under a global concurrency ceiling."""
    sem = asyncio.Semaphore(max_parallel)

    async def one(store_id: str, pool: AsyncConnectionPool):
        async with sem:                              # at most max_parallel live
            return store_id, await discover_store(pool, subject_id)

    tasks = [one(sid, p) for sid, p in pools.items()]
    settled = await asyncio.gather(*tasks, return_exceptions=True)

    results: dict[str, list[dict] | Exception] = {}
    for item in settled:
        if isinstance(item, Exception):
            # A single store failing must not fail the whole manifest.
            results.setdefault("__errors__", [])  # collect for the DLQ / recorded gap
            continue
        store_id, rows = item
        results[store_id] = rows
    return results

Using pool.connection() as an async with context manager is what guarantees the session is returned to the pool on every exit path — including cancellation — so a withdrawn or timed-out request cannot leak a connection. return_exceptions=True keeps one unreachable store from collapsing the entire sweep; the failing slice is recorded as a gap and routed onward, exactly as the SaaS API Sync Strategies layer records a partial-outage gap for API sources.

Step 4 — Size the pools to read-replica capacity, not worker ambition #

Pool sizing is a capacity-planning exercise anchored on the replica, not the worker. The peak sessions any one replica receives is that store’s max_size; the peak stores hit at once is max_parallel. The binding constraint is the replica’s own max_connections, minus the headroom reserved for its other consumers:

def safe_max_size(replica_max_connections: int, reserved_headroom: int, workers: int) -> int:
    """Largest per-store pool that leaves replica headroom for other consumers."""
    budget = replica_max_connections - reserved_headroom
    per_worker = budget // max(workers, 1)
    return max(1, per_worker)


# e.g. a replica with max_connections=200, 120 reserved for BI/ETL, 4 discovery workers:
#   safe_max_size(200, 120, 4) -> 20 sessions per worker, so max_size <= 20.

Start well below the computed ceiling and raise it only after measuring pg_stat_activity on the replica under load. A pool sized from the worker’s appetite rather than the replica’s headroom does not make discovery faster — it converts spare replica capacity into checkout contention for everyone.

Configuration reference #

Verification #

Prove the two ceilings hold before trusting the fan-out with production PII. The critical assertions are that no store ever exceeds its max_size, and that a saturated pool fails fast to a re-queue rather than blocking past the SLA window.

import asyncio
import pytest
from psycopg_pool import PoolTimeout


@pytest.mark.asyncio
async def test_pool_caps_concurrent_sessions(store_pool):
    """max_size sessions can be held; the next checkout times out fast."""
    held = [await store_pool.getconn() for _ in range(store_pool.max_size)]
    try:
        with pytest.raises(PoolTimeout):
            await store_pool.getconn(timeout=0.5)   # no slot free -> fail fast
    finally:
        for conn in held:
            await store_pool.putconn(conn)


@pytest.mark.asyncio
async def test_semaphore_bounds_live_stores(monkeypatch):
    """No more than max_parallel discover_store calls run concurrently."""
    live = 0
    peak = 0

    async def fake_discover(_pool, _subject):
        nonlocal live, peak
        live += 1
        peak = max(peak, live)
        await asyncio.sleep(0.01)
        live -= 1
        return []

    monkeypatch.setattr("mymodule.discover_store", fake_discover)
    await discover_all_stores({f"s{i}": object() for i in range(20)}, "subj-1", max_parallel=4)
    assert peak <= 4

In production, expect one structured log line per sweep carrying a trace ID, the per-store row counts, the observed pool checkout wait, and any PoolTimeout re-queues — this is the auditable evidence that discovery stayed inside both the concurrency budget and the statutory window.

Troubleshooting #

PoolTimeout under a burst of concurrent DSRs : Root cause: aggregate demand exceeds max_size for a hot store, so checkouts queue past pool_timeout_s. Fix: do not raise max_size past the replica’s headroom to paper over it — lower max_parallel so fewer stores compete at once, and let the re-queued tasks drain as slots free. Confirm the sizing with safe_max_size against the replica’s real max_connections.

Leaked connections — the pool drains and never recovers : Root cause: a session was checked out with getconn() (or a raw connection()) and an exception skipped the putconn/context-manager exit. Fix: always borrow via async with pool.connection() so cancellation and errors still return the session; audit any bare getconn call for a matching putconn in a finally.

Replica lag yields rows older than the request : Root cause: reading from a lagging replica can return a snapshot predating the verified request time. Fix: check replication lag before the sweep and, if it exceeds a tolerance, pin discovery to a specific snapshot or fail the store closed to a recorded gap rather than returning stale, incomplete data that would under-report a subject’s records under GDPR Article 15.

Idle discovery sessions hold slots between requests : Root cause: min_size set high, keeping warm connections that squat on replica capacity when no DSR is running. Fix: set min_size to 1 (or 0) so the pool shrinks to near-zero at rest, and rely on max_lifetime to recycle long-lived sessions that accumulate server-side state.

One unreachable store fails the whole sweep : Root cause: asyncio.gather called without return_exceptions=True, so a single connection error cancels sibling tasks. Fix: gather with return_exceptions=True, record the failing store as a gap, and route it onward for retry — a partial manifest with a documented gap is auditable; a collapsed sweep is not.

Frequently Asked Questions #

Related #

• Database Connector Configuration — the parent connector-configuration guide: typed, time-budgeted, credential-isolated relational extraction.
• Connecting PostgreSQL and Snowflake for DSR Discovery — the single-bridge pool primitives this page scales out across a fleet.
• Async Polling & Queue Management — bounded concurrency and backpressure for the discovery sweeps these pools feed.
• Schema Validation Rules — the gate that validates the deduplicated manifest before fulfillment.
• Cross-System Data Discovery & Sync — the parent architecture this pooling layer sits within.