07. PostgreSQL Logical CDC Outbox & Cold Partition ArchiverCopy Anchor Link

Production Paradigm: Native PostgreSQL Logical WAL Replication Slot Poller + In-Flight LSN Tracker + Declarative Zero-Lock Partition Detacher + Gzip NDJSON Cold Archival.

💥 The Real-World BottleneckCopy Anchor Link

High-throughput systems that rely on the Transactional Outbox Pattern to broadcast database events to Kafka, RabbitMQ, or Redis often implement naive table polling:

-- The Disaster of Polling the Outbox Table
SELECT * FROM outbox WHERE status = 'PENDING' ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED;

This naive polling strategy creates severe production pain:

1. Database CPU Burn & Lock Contention: Running this query every 100ms keeps DB CPU pegged at 40–80% even when no events are queued.
2. Replication Lag & Slot Bloat: If consumers slow down, unconsumed replication slots retain WAL files, quickly filling up disk space (PANIC: could not locate a valid checkpoint record).
3. Table Bloat & Vacuum Degradation: Millions of event rows accumulate over time, degrading B-Tree indexes and causing massive PostgreSQL autovacuum freezes.

🏛️ High-Concurrency ArchitectureCopy Anchor Link

This blueprint implements an enterprise-grade Change Data Capture (CDC) pipeline reading directly from PostgreSQL's Write-Ahead Log (WAL), paired with an automated cold partition archiver.

Key Architectural PhasesCopy Anchor Link

1. Native WAL Slot Consumer (Tier1NativeWalSlotPoller):
   • Reads directly from PostgreSQL logical replication slots using pg_logical_slot_peek_changes or Debezium format.
   • Eliminates polling SELECT queries entirely, dropping idle CPU overhead to near 0%.
2. Idempotent LSN Tracker (IdempotentLsnTracker):
   • Accurately tracks Log Sequence Numbers (LSN) in 64-bit BigInt.
   • Only commits the flush point (pg_logical_slot_flush_changes) after downstream brokers acknowledge receipt, guaranteeing strict at-least-once delivery with zero message loss.
3. Disk Safety Guard (evaluateCdcSlotDiskSafety):
   • Monitors replication lag in bytes (calculateLsnLagBytes). If downstream lag exceeds safe disk thresholds, triggers automated alerts before disk exhaustion.
4. Declarative Partition Archiver (ColdTierPartitionArchiver):
   • Detaches historical event tables concurrently (ALTER TABLE ... DETACH PARTITION ... CONCURRENTLY) without holding table-exclusive access locks.
   • Streams rows into compressed NDJSON Gzip files and uploads them to object storage, reclaiming 85%+ disk space.

🧩 Composed Functional SlicesCopy Anchor Link

💻 Code ImplementationCopy Anchor Link

mkdir -p src/lib/cdc
cp -r packages/async-sync/src/pg-cdc-stream src/lib/cdc/
cp -r packages/async-sync/src/pg-partition-archiver src/lib/cdc/
cp packages/async-sync/src/uuidv7.ts src/lib/cdc/

import { Tier1NativeWalSlotPoller, IdempotentLsnTracker, evaluateCdcSlotDiskSafety } from '@/lib/cdc/pg-cdc-stream';
import { ColdTierPartitionArchiver } from '@/lib/cdc/pg-partition-archiver';
import { db } from '@/lib/db';

// 1. Initialize CDC WAL Poller
const lsnTracker = new IdempotentLsnTracker();
const cdcPoller = new Tier1NativeWalSlotPoller({
  slotName: 'kido_cdc_slot',
  batchSize: 500,
  async onChanges(events) {
    for (const event of events) {
      // Dispatch to downstream broker
      await broadcastToEventBus(event.table, event.action, event.payload);
      lsnTracker.markProcessed(event.lsn);
    }

    // Confirm flush to WAL slot
    const highestFlushedLsn = lsnTracker.getHighestFlushedLsn();
    if (highestFlushedLsn) {
      await db.$executeRawUnsafe(
        `SELECT pg_logical_slot_flush_changes('kido_cdc_slot', '${highestFlushedLsn}')`
      );
    }
  },
});

// 2. Archival Task for historical tables (>90 days old)
export async function archiveOldEvents() {
  const archiver = new ColdTierPartitionArchiver({
    parentTable: 'audit_events',
    retentionDays: 90,
  });

  await archiver.detachAndCompress({
    onUploadNdjsonGzip: async (partitionName, gzipStream) => {
      await uploadToS3(`archives/${partitionName}.ndjson.gz`, gzipStream);
    },
  });
}

📊 Production Scale Benchmark & VerificationCopy Anchor Link