Async CDC Consumer Patterns with asyncpg

An async CDC consumer built on asyncpg fans a single replication stream into a bounded pool of parallel sink writers without breaking per-key commit order or…

An async CDC consumer built on asyncpg fans a single replication stream into a bounded pool of parallel sink writers without breaking per-key commit order or acknowledging an LSN before its data is durable. This page details that concurrency model as an extension of Python CDC parser development: the reader/queue/worker topology, the backpressure that keeps a slow sink from exhausting memory, and the commit-watermark logic that makes out-of-order parallel completion safe to acknowledge back to PostgreSQL.

The reason to go async is throughput without a JVM: one process reading one slot can saturate a warehouse or Redis sink far better when apply work runs concurrently instead of one commit at a time. The reason it is dangerous is that concurrency breaks the two invariants a single-threaded parser gets for free — that changes to the same row apply in commit order, and that confirmed_flush_lsn never advances past an unpersisted change. Violate the first and a stale UPDATE overwrites a newer one; violate the second and a crash discards WAL for rows you never actually wrote. Everything below exists to preserve both while still running N workers in parallel.

Reader, bounded queues, worker pool, and the commit watermark that gates feedback A single stream reader coroutine owns the logical replication connection and hashes each decoded change by its routing key to one of three bounded asyncio queues, so every change for one key always lands on the same queue. Each queue is drained by exactly one worker coroutine, and all workers share one asyncpg pool that performs idempotent ON CONFLICT upserts into the sink. As each worker persists a commit it reports to a commit-watermark tracker, which advances only across a contiguous prefix of fully persisted commit LSNs and hands that safe LSN back to the reader. The reader sends that value as confirmed_flush_lsn feedback so PostgreSQL may reclaim WAL only up to durably applied data. When a queue reaches its maxsize the reader blocks on put, which stops it reading and pauses feedback, applying backpressure instead of dropping changes or growing memory without bound. Stream reader one coroutine owns replication conn hash(key) → shard Queue 0 asyncio.Queue(maxsize) Queue 1 asyncio.Queue(maxsize) Queue 2 asyncio.Queue(maxsize) full queue → reader blocks (backpressure) Worker 0 await queue.get() Worker 1 one per shard Worker 2 preserves per-key order asyncpg pool idempotent upsert ON CONFLICT DO UPDATE → sink report persisted commit LSN Commit-watermark tracker highest LSN whose whole prefix is persisted out-of-order completion never advances it past a gap send_feedback( confirmed_flush_lsn)
One reader hashes changes to bounded queues; one worker per shard preserves per-key order; the commit-watermark tracker only lets confirmed_flush_lsn advance across a fully persisted prefix, and a full queue blocks the reader instead of dropping data.

Concurrency and Ordering Semantics

asyncpg is the sink driver here, not the replication client — it implements no START_REPLICATION sub-protocol, so the single reader owns a psycopg2 LogicalReplicationConnection (driven through asyncio.to_thread) or a pg_recvlogical pipe, decodes each change enough to extract a routing key and the commit LSN, and hands it to the async side. asyncpg then powers the concurrent apply path, where its pooled prepared-statement execution is exactly what a fan-out of workers needs. The division of labor below is what keeps parallelism from corrupting order.

Component Concurrency Ordering guarantee Backpressure / feedback behavior
Stream reader Exactly one coroutine Reads commits in strict LSN order Blocks on queue.put() when a shard is full; this is the throttle that pauses WAL consumption
Shard queue One asyncio.Queue(maxsize=N) per shard FIFO within the shard maxsize bounds in-flight memory; a slow worker stalls only its own shard
Worker One coroutine per queue Same key → same shard → same worker → per-key commit order preserved Applies through the shared asyncpg pool; on failure it retries in place, never skips
asyncpg pool min_size/max_size connections None — connections are interchangeable Pool exhaustion surfaces as apply latency, not reordering
Watermark tracker Single owner (the reader’s ack timer) Advances only across a contiguous persisted prefix Feeds the safe LSN to send_feedback; a gap holds the whole ack back

Two properties are specific to this design and easy to get wrong. First, the shard count is the parallelism ceiling and the ordering unit: hash on the same identifier your sink treats as the primary key (order id, account id) so that co-dependent changes serialize. Second, the reader must decode only enough to route — extracting the key and LSN — while heavyweight coercion happens in the worker, so a single reader never becomes the bottleneck. The byte-level decoding of each pgoutput tag is owned by the parsing pgoutput format with psycopg2 work; this page assumes those decoded records already exist.

Diagnostic Patterns

The consumer’s health is visible on both sides. On the async side, the queue depths tell you whether a shard is hot; on the server side, the slot’s restart_lsn tells you whether feedback is actually flowing. The relationship between restart_lsn and confirmed_flush_lsn — and why a stalled worker pins WAL — is grounded in WAL stream mechanics.

python
# Export per-shard queue depth and the watermark gap as gauges.
# ALERT: any queue at maxsize for > 30 s (a hot key), or watermark_gap_bytes
# climbing while the pipeline is "up" (workers persisting slower than the reader reads).
def snapshot(queues, tracker, current_wal_lsn):
    for i, q in enumerate(queues):
        QUEUE_DEPTH.labels(shard=str(i)).set(q.qsize())
    WATERMARK_LSN.set(tracker.safe_lsn)
    WATERMARK_GAP.set(current_wal_lsn - tracker.safe_lsn)
sql
-- Publisher side: is the async consumer's feedback advancing the slot?
-- ALERT: retained_wal > 25% of max_slot_wal_keep_size, or active = false > 60 s.
SELECT slot_name,
       active,
       confirmed_flush_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots
WHERE slot_name = 'cdc_async_slot';

Read the two together. A queue pinned at maxsize with retained_wal climbing means a single hot shard is throttling the whole reader through backpressure — the correct response is to raise the shard count or shard that hot key further, not to enlarge the queue, which only defers the stall. A watermark gap that grows while every queue is near-empty means the sink itself is slow; add pool capacity. If active = false, the reader died and the slot is now pinning WAL, the same disk-exhaustion precursor described throughout async monitoring integration.

Safe Deployment Sequence

The whole point of the watermark is that a crash is recoverable, so validate recovery before you trust the pipeline with production write volume.

1. Provision the slot and confirm the resume point. Create the slot deliberately, note its confirmed_flush_lsn, and persist it as the consumer’s start LSN so a restart resumes from an acknowledged position rather than an arbitrary one.

sql
SELECT slot_name, confirmed_flush_lsn
FROM pg_create_logical_replication_slot('cdc_async_slot', 'pgoutput');

2. Start with one shard. Run the consumer with a single queue and single worker first. This is behaviorally identical to a synchronous parser and proves the decode, upsert, and feedback loop before concurrency is in play.

3. Scale shards under representative load. Raise the shard count to your target and drive a peak-rate write batch. Watch that per-key ordering holds — the last-write-wins guard below makes reordering detectable rather than silent — and that no single queue saturates.

4. Kill-test the watermark. Send SIGKILL mid-batch, restart, and confirm the consumer resumes from confirmed_flush_lsn and re-applies the in-flight commits idempotently with no duplicate side effects and no missing rows. If recovery is not clean here, it will not be clean in an incident.

5. Enable graceful shutdown. Only after the kill-test passes, wire SIGTERM to the drain path (below) so planned restarts flush cleanly and rolling deploys do not force a resnapshot.

Pipeline Integration

The async apply path is where idempotency, the feedback watermark, and reconnect logic come together. The commit-watermark tracker is the safety-critical piece: with workers completing out of order, it advances the acknowledged LSN only across a contiguous prefix of fully persisted commits, so confirmed_flush_lsn can never jump past a commit still in flight.

python
class AckTracker:
    """Advances the safe ack LSN only over a gapless prefix of persisted commits."""
    def __init__(self, start_lsn: int):
        self._order: list[int] = []       # commit LSNs in commit (dispatch) order
        self._remaining: dict[int, int] = {}  # lsn -> outstanding shard-parts
        self._safe = start_lsn

    def dispatch(self, lsn: int, parts: int) -> None:
        self._order.append(lsn)
        self._remaining[lsn] = parts       # a commit may fan to several shards

    def complete(self, lsn: int) -> None:
        self._remaining[lsn] -= 1
        # Advance across the leading run of fully persisted commits only.
        while self._order and self._remaining.get(self._order[0]) == 0:
            done = self._order.pop(0)
            del self._remaining[done]
            self._safe = done

    @property
    def safe_lsn(self) -> int:
        return self._safe

The worker applies each shard’s changes idempotently and reports completion. The upsert is a last-write-wins merge keyed on the business identifier, so an at-least-once replay after a crash converges instead of clobbering newer state — the same guard used across event routing and Kafka integration.

python
async def worker(queue: asyncio.Queue, pool: asyncpg.Pool, tracker: AckTracker):
    while True:
        item = await queue.get()
        try:
            if item is SHUTDOWN:
                return
            commit_lsn, rows = item
            async with pool.acquire() as conn:
                async with conn.transaction():
                    await conn.executemany(
                        """INSERT INTO orders (id, status, updated_at)
                           VALUES ($1, $2, $3)
                           ON CONFLICT (id) DO UPDATE
                             SET status = EXCLUDED.status,
                                 updated_at = EXCLUDED.updated_at
                             WHERE orders.updated_at < EXCLUDED.updated_at""",
                        rows,
                    )
            tracker.complete(commit_lsn)     # only now is this shard-part durable
        finally:
            queue.task_done()

A dedicated timer sends feedback on an interval, always using tracker.safe_lsn — never the highest dispatched LSN. Feedback also doubles as the keepalive that prevents wal_sender_timeout from dropping an idle slot.

python
async def feedback_loop(reader, tracker, interval: float = 5.0):
    while True:
        await asyncio.sleep(interval)
        reader.send_feedback(flush_lsn=tracker.safe_lsn)   # confirmed_flush_lsn

Reconnect with jittered backoff. When the replication connection drops, resume from the last acknowledged LSN, never from an earlier arbitrary point, and back off with jitter so a publisher blip does not become a reconnect storm:

python
async def run(dsn, start_lsn):
    tracker = AckTracker(start_lsn)
    backoff = 0.5
    while not shutting_down.is_set():
        try:
            await stream(dsn, resume_from=tracker.safe_lsn, tracker=tracker)
            backoff = 0.5
        except (ConnectionError, asyncpg.PostgresError) as exc:
            await asyncio.sleep(backoff + random.uniform(0, 0.25))
            backoff = min(backoff * 2, 30)     # cap at 30 s

Graceful shutdown. On SIGTERM, stop the reader, push a SHUTDOWN sentinel to every queue, await queue.join() to drain in-flight work, send one final send_feedback(tracker.safe_lsn), then close the pool. Draining rather than dropping means a planned restart never forces the resnapshot that a hard kill would — and because the tracker only ever acknowledged a persisted prefix, even the hard kill is recoverable, just more expensive.

Failover. Before PostgreSQL 17 a logical slot does not survive primary failover, so a promotion invalidates cdc_async_slot. Detect the drop, compare your persisted safe_lsn against the new primary’s pg_current_wal_lsn(), and if the consumer applied changes the promoted node never persisted, re-seed the affected relations — the reconciliation mirrors subscription sync procedures. On PG 17+, create the slot with failover = true so it is synchronized to the standby and survives promotion.

Authoritative references