Scalable outbox: dedicated connection pool, keyset pagination, dedup

[strobus]

Problem

At scale, the outbox consumer was experiencing context deadline exceeded errors and blocking behavior. Root cause analysis identified three compounding issues:

1. Connection pool starvation — The outbox consumer shared a *sql.DB pool with the API resolvers. Long-running mutations (e.g. createServiceableAddresses) would exhaust the pool, starving the outbox consumer's SELECT and FOR UPDATE queries until the mutation completed or timed out.

2. Unbounded table scan — SELECT id FROM outbox loaded every row into memory in a single query, causing context deadline timeouts as the table grew.

Changes

Dedicated consumer connection pool (store/pg/pg.go)

• Creates a separate *sql.DB from connStr with configurable MaxOpenConns (default 8)
• All consumer operations (queryPage, ProcessTx, GetWithLock, Delete) run on this isolated pool
• The caller's db parameter is accepted for backward compatibility but no longer stored
• Added Close() method and done channel for clean shutdown and goroutine lifecycle
• Added WithMaxConsumerConns(n) option

Keyset pagination (store/pg/pg.go)

• Replaced SELECT id FROM outbox with SELECT id FROM outbox WHERE id > $1 ORDER BY id LIMIT N
• O(1) per page via primary key index, stable under concurrent inserts/deletes
• Split into queryPage (DB-scoped context) and fetchPage (channel sends) so backpressure cannot trigger context timeouts
• Added WithPageSize(n) and WithChannelBufferSize(n) options
• Defaults tuned to chanBufferSize=5, pageSize=6 — producer blocks on the 6th item, keeping the consumer saturated while minimizing duplicate-read windows across pods

In-flight deduplication (outbox.go)

• Added sync.Map to track IDs currently being processed
• Duplicate IDs from overlapping scans are skipped (LoadOrStore / Delete around process)

ProcessTx rollback safety (store/pg/pg.go)

• Added defer tx.Rollback() to handle context cancellation gracefully (no-op after commit)
• Eliminated noisy sql: transaction has already been committed or rolled back errors

Listener lifecycle (store/pg/pg.go)

• Added defer l.Close() in Listen goroutine
• Moved l.Ping() back into the time.After select case (only needed when idle)
• Removed redundant pre-check select on done channel
• Added rows.Err() check after row iteration

Tests (store/pg/pg_test.go, store/pg/pg_suite_test.go)

• Added tests for dedicated consumer pool creation, WithMaxConsumerConns, and Close()
• Added defer subject.Close() to all test cases
• Fixed connStr initialization in test suite setup

Test plan

• go test ./store/pg/... passes
• Verified under load — context deadline errors resolved
• Confirmed outbox processes messages concurrently with long-running API mutations
• Keyset pagination handles concurrent inserts/deletes correctly

🤖 Generated with Claude Code

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.