feat(event-engine): enable outbox poller by default

The attempts column migration (20260720120000_add_outbox_table) is now
applied, so the poller's prerequisite is met. Flip the default from opt-in
to opt-out: OUTBOX_POLL_ENABLED now defaults ON and is disabled only with
OUTBOX_POLL_ENABLED=false. This means the reconciliation backstop runs in
every environment where the migration exists without each one having to set
the var — pairing with the process-then-delete reorder so failed/aged rows
are actually drained.

Safe on autoscaled pods: a Postgres advisory lock elects a single sweeping
pod, the grace window keeps it off the live CDC path, and FOR UPDATE SKIP
LOCKED + cursor paging bound each row to one attempt per sweep.

Updates .env.example and MIGRATION.md to document the opt-out default.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
briant
2026-07-21 11:02:41 -06:00
parent ce92443b3c
commit 0499c602dd
3 changed files with 11 additions and 12 deletions
+3 -3
View File
@@ -36,9 +36,9 @@ WORKER_POOL_SIZE=10
HEALTH_CHECK_PORT=3000
# Outbox Reconciliation Poller (backstop that drains Outbox rows the live CDC path missed)
# DEFAULT OFF: requires the "Outbox".attempts column (packages/civitai-db-schema migration
# 20260720120000_add_outbox_table). Apply that migration, THEN set OUTBOX_POLL_ENABLED=true.
# OUTBOX_POLL_ENABLED=true
# DEFAULT ON: requires the "Outbox".attempts column (packages/civitai-db-schema migration
# 20260720120000_add_outbox_table, now applied). Set OUTBOX_POLL_ENABLED=false to disable.
# OUTBOX_POLL_ENABLED=false # opt-out; the poller runs by default
# OUTBOX_POLL_INTERVAL=300 # seconds between sweeps
# OUTBOX_POLL_GRACE=300 # seconds; only claim rows older than this (never races the CDC path)
# OUTBOX_POLL_BATCH_SIZE=100
+4 -4
View File
@@ -47,12 +47,12 @@ start/stop), `OUTBOX_POLL_*`/`OUTBOX_MAX_ATTEMPTS` config, `outboxPollerMetrics`
(`pg_try_advisory_lock(25974, 1)`) — a separate key space from the main app's single-bigint
`pg_advisory_xact_lock(articleId)`, so it can never collide. Self-healing on pod death; `FOR UPDATE SKIP
LOCKED` + cursor paging are additional safety.
- **DEFAULT OFF.** It requires an `attempts` int column on `"Outbox"`, added by
- **DEFAULT ON (opt-out).** It requires an `attempts` int column on `"Outbox"`, added by
`packages/civitai-db-schema/prisma/migrations/20260720120000_add_outbox_table/` — which also **models the
pre-existing Outbox table + `OutboxEntity` enum** in `schema.full.prisma` (they existed in the DB but were
not in the Prisma schema). **Sequence: apply that migration → run `pnpm db:generate` to regenerate the
derived types (`models.ts`, `kysely/*` — NOT hand-edited here) → set `OUTBOX_POLL_ENABLED=true`.** Until the
column exists, leave the poller off (its SQL would error each sweep).
not in the Prisma schema). **That migration has been applied**, so the poller now runs by default; set
`OUTBOX_POLL_ENABLED=false` to disable. (Sequence was: apply migration → `pnpm db:generate` to regenerate
the derived types `models.ts`/`kysely/*` — NOT hand-edited here → poller on.)
## Security
`.env.example` shipped real-looking production credentials in the source repo; they were **scrubbed to
+4 -5
View File
@@ -71,11 +71,10 @@ export const config = {
// processed (created pre-connector, during downtime, or handler failures).
// Only claims rows older than the grace window so it never races the
// real-time Kafka path.
// DEFAULT OFF (opt-in): the poller REQUIRES an `attempts` int column on the
// "Outbox" table (a main-app civitai-db-schema migration that is NOT part of
// this repo). Set OUTBOX_POLL_ENABLED=true only AFTER that column exists, or
// every sweep will error against the missing column. See MIGRATION.md.
outboxPollEnabled: process.env.OUTBOX_POLL_ENABLED === 'true',
// DEFAULT ON (opt-out): the poller REQUIRES an `attempts` int column on the
// "Outbox" table (civitai-db-schema migration 20260720120000_add_outbox_table,
// now applied). Set OUTBOX_POLL_ENABLED=false to disable. See MIGRATION.md.
outboxPollEnabled: process.env.OUTBOX_POLL_ENABLED !== 'false',
outboxPollIntervalMs: parseInt(process.env.OUTBOX_POLL_INTERVAL ?? '300') * 1000,
outboxPollGraceMs: parseInt(process.env.OUTBOX_POLL_GRACE ?? '300') * 1000,
outboxPollBatchSize: parseInt(process.env.OUTBOX_POLL_BATCH_SIZE ?? '100'),