Make posting CDC commit-order safe (#6049)

Author: viktor-shcherb

.github/workflows/ci.yml

@@ -495,6 +495,50 @@ jobs:
       - run: uv run pytest tests/e2e/test_typesense_indexing.py -v
         working-directory: apps/crawler
 
+  test-crawler-postgres-cdc-e2e:
+    name: Test Crawler PostgreSQL CDC E2E
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:16-alpine
+        ports:
+          - 5432:5432
+        env:
+          POSTGRES_USER: crawler
+          POSTGRES_PASSWORD: crawler
+          POSTGRES_DB: crawler
+        options: >-
+          --health-cmd "pg_isready -U crawler -d crawler"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 12
+    env:
+      LOCAL_DATABASE_URL: postgresql://crawler:crawler@localhost:5432/crawler
+      REQUIRE_POSTGRES_E2E: "true"
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        with:
+          enable-cache: true
+          prune-cache: true
+          cache-dependency-glob: "apps/crawler/uv.lock"
+
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.13"
+
+      - run: uv sync --group dev
+        working-directory: apps/crawler
+
+      - run: uv run alembic -c src/migrations/alembic.ini upgrade head
+        working-directory: apps/crawler
+
+      - run: uv run pytest tests/e2e/test_postgres_cdc_commit_order.py -v
+        working-directory: apps/crawler
+
   coverage-crawler:
     name: Coverage Crawler
     needs: changes

AGENTS.md

@@ -201,7 +201,7 @@ cd apps/crawler && uv run python ../../scripts/typesense-backfill-local.py [--li
 
 ### Indexing Pipeline
 
-- **Exporter** (CDC): two-cursor design — Supabase and Typesense cursors advance independently. Concurrent upserts via `asyncio.gather`
+- **Exporter** (CDC): database-triggered shared writer barrier + brief exclusive cutoff prevents commit-order skips; Supabase and Typesense cursors advance independently; concurrent upserts via `asyncio.gather`
 - **Sync**: taxonomy collections (location, occupation, seniority, technology) and the `company` collection populated after CSV sync. Company docs include extended fields (logo, website, employee_count_range, founded_year) and per-locale variants (`description_{de,fr,it}`, `industry_name_{de,fr,it}`) for the company detail page reader. Handles taxonomy rename detection
 - **Reconciliation**: daily count check + sample comparison
 - **refresh-typesense**: periodic count refresh for taxonomy/company collections + watchlist reconciliation. Runs inline at every deploy/CSV sync (via `crawler sync`) and every 4h via `.github/workflows/crawler-scheduled-maintenance.yml` out-of-band

apps/crawler/AGENTS.md

@@ -66,7 +66,7 @@ src/
 │   └── lookups.py         # Cached lookup table loaders (locations, technologies, etc.)
 ├── redis_queue.py         # Lua-backed claim/enqueue/reschedule
 ├── lua/                   # claim_work.lua, enqueue_task.lua, reschedule_task.lua
-├── exporter.py            # CDC: local Postgres -> Supabase + Typesense (two-cursor)
+├── exporter.py            # Commit-safe CDC: local Postgres -> Supabase + Typesense (two-cursor)
 ├── typesense_client.py    # Shared Typesense client (lazy init, None when unconfigured)
 ├── sync.py                # CSV -> local Postgres + Supabase + Redis + Typesense taxonomies
 ├── bootstrap.py           # One-time: Supabase -> local Postgres copy

apps/crawler/VERSION

@@ -1 +1 @@
-0.13.180
+0.13.181

apps/crawler/alerts.yaml

@@ -473,6 +473,48 @@ groups:
             Export Lag panel.
           runbook: https://github.com/colophon-group/jobseek/blob/main/docs/03-crawler-architecture.md
 
+      # A timeout means at least one posting writer held its transaction open
+      # beyond the barrier's bounded 120s wait. The exporter deliberately did
+      # not choose a cutoff or advance either cursor, preserving data at the
+      # cost of freshness. Route the root signal to the daily Codex review.
+      - alert: CdcWriterBarrierTimeout
+        expr: increase(crawler_exporter_cdc_barrier_timeouts_total{instance="exporter"}[15m]) > 0
+        for: 1m
+        labels:
+          severity: high
+          service: crawler
+          owner: codex-error-review
+          route: codex-daily
+        annotations:
+          summary: "CDC writer barrier timed out"
+          description: |
+            The exporter could not establish a commit-safe posting cutoff
+            within 120 seconds. It kept both cursors pinned. Inspect
+            `pg_stat_activity` for a long transaction and the
+            `cdc_snapshot_barrier.timeout` exporter event before terminating
+            anything.
+          runbook: https://github.com/colophon-group/jobseek/blob/main/docs/03-crawler-architecture.md#commit-safe-posting-cdc
+
+      # The daily sample reconciliation now has an explicit alert path rather
+      # than only a log line. Any discrepancy is actionable evidence that a
+      # downstream missed or diverged from local authoritative state.
+      - alert: CdcReconciliationDrift
+        expr: increase(crawler_reconciliation_discrepancies_total{instance="exporter"}[25h]) > 0
+        for: 5m
+        labels:
+          severity: high
+          service: crawler
+          owner: codex-error-review
+          route: codex-daily
+        annotations:
+          summary: "Posting reconciliation found downstream drift"
+          description: |
+            The daily local PostgreSQL comparison found at least one missing
+            or divergent Supabase/Typesense row. The reconciler touched the
+            sampled rows for idempotent replay; confirm cursor health and run
+            the exact reconciliation procedure before closing the incident.
+          runbook: https://github.com/colophon-group/jobseek/blob/main/docs/03-crawler-architecture.md#commit-safe-posting-cdc
+
       # Catches steady-state degradation: WAF blocking a major board
       # source, Postgres going read-only, etc. >50% failure rate is
       # rare in normal operation; legitimate noisy boards top out

apps/crawler/deploy.sh

@@ -361,6 +361,15 @@ pull_deploy_images
 
 docker compose up -d redis
 
+# ── Quiesce every local-Postgres writer before schema cutover ──────
+# Migrations may introduce a database/runtime protocol (for example the
+# shared-writer/exclusive-exporter CDC barrier). Stop both sides before
+# Alembic so no old process can write or advance a cursor in the interval
+# between the schema change and the new containers starting. `--timeout 60`
+# matches the app's 30s bounded drain with headroom before Docker sends
+# SIGKILL. Redis and Alloy remain available throughout.
+docker compose stop --timeout 60 worker-1 worker-2 worker-3 browser-1 exporter drain
+
 # ── Run Alembic migrations on local Postgres ─────────────────────────
 docker run --rm --env-file "$ENV_FILE" --network host \
   "ghcr.io/${OWNER}/jobseek-crawler:${IMAGE_TAG}" \
@@ -373,13 +382,6 @@ docker run --rm --env-file "$ENV_FILE" --network host \
   "ghcr.io/${OWNER}/jobseek-crawler:${IMAGE_TAG}" \
   uv run --no-sync crawler setup-typesense
 
-# ── Quiesce processors before reseeding Redis-backed schedules ───────
-# Keep Redis and alloy up, but stop processors so deploy-time
-# `crawler sync` does not race with live workers claiming work while we
-# reseed board monitors. `--timeout 60` matches the app's 30s bounded
-# drain with headroom before Docker sends SIGKILL.
-docker compose stop --timeout 60 worker-1 worker-2 worker-3 browser-1 exporter drain
-
 # ── Sync board config from CSV → local Postgres + Redis + Typesense ──
 docker run --rm --env-file "$ENV_FILE" --network host \
   "ghcr.io/${OWNER}/jobseek-crawler:${IMAGE_TAG}" \

apps/crawler/src/export_cursor_fence.py

@@ -12,12 +12,15 @@
 
 import asyncio
 import time
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from datetime import datetime
 
 import asyncpg
 import structlog
 
+from src.metrics import exporter_cdc_barrier_timeouts, exporter_cdc_barrier_wait
+
 log = structlog.get_logger()
 
 # Stable, repository-owned bigint (ASCII-ish ``JOBSEEK``) shared by the
@@ -28,7 +31,19 @@
 _TRY_LOCK_SQL = "SELECT pg_try_advisory_lock($1::bigint)"
 _UNLOCK_SQL = "SELECT pg_advisory_unlock($1::bigint)"
 
+# Distinct from the long operator-repair fence above. Every transaction that
+# changes an exported job_posting field takes the shared side from a database
+# trigger; the exporter briefly takes the exclusive side before choosing its
+# cutoff. ASCII-ish ``CDCLOCK`` encoded as a positive bigint.
+CDC_WRITER_BARRIER_ID = 0x4344434C4F434B
+_CDC_TRY_LOCK_SQL = "SELECT pg_try_advisory_lock($1::bigint)"
+_CDC_CUTOFF_SQL = "SELECT clock_timestamp()"
+_CDC_LOCK_TIMEOUT_SECONDS = 120.0
+_CDC_LOCK_RETRY_SECONDS = 0.1
+_CDC_SLOW_WAIT_SECONDS = 1.0
+
 CursorFenceFactory = Callable[[asyncpg.Pool], AbstractAsyncContextManager[None]]
+CutoffFactory = Callable[[asyncpg.Pool], Awaitable[datetime]]
 
 
 @asynccontextmanager
@@ -86,3 +101,82 @@ async def export_cursor_fence(pool: asyncpg.Pool) -> AsyncIterator[None]:
                 if unlocked is not True:
                     conn.terminate()
                     raise RuntimeError("export cursor advisory fence was not held")
+
+
+async def capture_cdc_snapshot_cutoff(pool: asyncpg.Pool) -> datetime:
+    """Return a commit-safe upper bound for one posting export tick.
+
+    The matching database trigger holds a shared transaction advisory lock
+    from the first relevant statement until commit or rollback. Non-blocking
+    exclusive probes avoid queuing healthy writers behind one long transaction.
+    Once a probe succeeds:
+
+    * writers that stamped before this call must commit before it returns;
+    * writers that reach the trigger while it is held stamp only after the
+      returned cutoff; and
+    * the exporter can safely query ``updated_at < cutoff`` without holding
+      the lock during downstream network I/O.
+
+    A finite polling window bounds exporter staleness if a writer is wedged,
+    without advancing a cursor or imposing that wait on new writers. Any
+    acquisition/cutoff/release uncertainty terminates the dedicated pooled
+    session so PostgreSQL releases a possibly-held lock.
+    """
+
+    async with pool.acquire() as conn:
+        acquired = False
+        started = time.monotonic()
+        try:
+            while not acquired:
+                try:
+                    attempt = await conn.fetchval(_CDC_TRY_LOCK_SQL, CDC_WRITER_BARRIER_ID)
+                except BaseException:
+                    conn.terminate()
+                    raise
+
+                if attempt is True:
+                    acquired = True
+                    break
+                if attempt is not False:
+                    conn.terminate()
+                    raise RuntimeError("PostgreSQL returned an invalid CDC barrier result")
+
+                elapsed = time.monotonic() - started
+                if elapsed >= _CDC_LOCK_TIMEOUT_SECONDS:
+                    exporter_cdc_barrier_timeouts.inc()
+                    log.error(
+                        "cdc_snapshot_barrier.timeout",
+                        timeout_s=_CDC_LOCK_TIMEOUT_SECONDS,
+                    )
+                    raise TimeoutError("timed out waiting for a commit-safe CDC cutoff")
+                await asyncio.sleep(
+                    min(_CDC_LOCK_RETRY_SECONDS, _CDC_LOCK_TIMEOUT_SECONDS - elapsed)
+                )
+
+            wait_s = time.monotonic() - started
+            exporter_cdc_barrier_wait.observe(wait_s)
+            if wait_s >= _CDC_SLOW_WAIT_SECONDS:
+                log.warning(
+                    "cdc_snapshot_barrier.acquired_after_wait",
+                    wait_s=round(wait_s, 3),
+                )
+
+            cutoff = await conn.fetchval(_CDC_CUTOFF_SQL)
+            if not isinstance(cutoff, datetime):
+                conn.terminate()
+                raise RuntimeError("PostgreSQL returned an invalid CDC cutoff")
+            return cutoff
+        except BaseException:
+            if acquired:
+                conn.terminate()
+            raise
+        finally:
+            if acquired and not conn.is_closed():
+                try:
+                    unlocked = await conn.fetchval(_UNLOCK_SQL, CDC_WRITER_BARRIER_ID)
+                except BaseException:
+                    conn.terminate()
+                    raise
+                if unlocked is not True:
+                    conn.terminate()
+                    raise RuntimeError("CDC writer advisory barrier was not held")

apps/crawler/src/exporter.py

@@ -14,7 +14,12 @@
 from prometheus_client import Counter, Gauge
 
 from src.config import settings
-from src.export_cursor_fence import CursorFenceFactory, export_cursor_fence
+from src.export_cursor_fence import (
+    CursorFenceFactory,
+    CutoffFactory,
+    capture_cdc_snapshot_cutoff,
+    export_cursor_fence,
+)
 from src.metrics import (
     export_errors_total,
     exporter_export_lag,
@@ -24,6 +29,7 @@
     local_db_pool_idle,
     local_db_pool_size,
     r2_pending_gauge,
+    reconciliation_discrepancies,
     redis_queue_depth,
     supa_db_pool_idle,
     supa_db_pool_size,
@@ -73,6 +79,7 @@
 # ---------------------------------------------------------------------------
 
 _EPOCH = datetime.min.replace(tzinfo=UTC)
+_MAX_CDC_CUTOFF = datetime.max.replace(tzinfo=UTC)
 _ZERO_UUID = uuid.UUID(int=0)
 
 # Sentinel stamped on Typesense `experience_max` for rows the extractor
@@ -995,6 +1002,8 @@ async def _export_postings_dual(
     maps: TaxonomyMaps,
     supa_backoff: _DownstreamBackoff | None = None,
     ts_backoff: _DownstreamBackoff | None = None,
+    *,
+    cutoff: datetime = _MAX_CDC_CUTOFF,
 ) -> tuple[int, Cursor, Cursor]:
     """Fetch changed postings and upsert to both Supabase and Typesense concurrently.
 
@@ -1027,6 +1036,7 @@ async def _export_postings_dual(
         fetch_ts,
         fetch_id,
         settings.export_batch_limit,
+        cutoff,
     )
     if not rows:
         return 0, supa_cursor, ts_cursor
@@ -1256,7 +1266,8 @@ def select_changed_sql(cls, *extras: str) -> str:
             + cls.select_list(*extras)
             + " FROM "
             + cls.table
-            + " WHERE (updated_at, id) > ($1, $2) ORDER BY updated_at, id LIMIT $3"
+            + " WHERE (updated_at, id) > ($1, $2)"
+            + " AND updated_at < $4 ORDER BY updated_at, id LIMIT $3"
         )
 
 
@@ -1269,6 +1280,8 @@ async def _export_changed_postings(
     local_pool: asyncpg.Pool,
     supa_pool: asyncpg.Pool,
     cursor: Cursor,
+    *,
+    cutoff: datetime = _MAX_CDC_CUTOFF,
 ) -> tuple[int, Cursor]:
     """Export job_posting rows changed since cursor to Supabase.
 
@@ -1286,6 +1299,7 @@ async def _export_changed_postings(
         last_ts,
         last_id,
         settings.export_batch_limit,
+        cutoff,
     )
     if not rows:
         return 0, cursor
@@ -1442,6 +1456,7 @@ async def run_exporter(
     shutdown_event: asyncio.Event,
     *,
     cursor_fence_factory: CursorFenceFactory = export_cursor_fence,
+    cutoff_factory: CutoffFactory = capture_cdc_snapshot_cutoff,
 ) -> None:
     """Main exporter loop.
 
@@ -1486,6 +1501,12 @@ async def run_exporter(
                     if maps.stale:
                         await maps.refresh(local_pool, supa_pool)
 
+                    # Probe until transactions that changed exported posting
+                    # fields have committed, capture a clock cutoff, then
+                    # immediately release the writer barrier. Rows stamped by
+                    # later transactions stay above this strict upper bound.
+                    cutoff = await cutoff_factory(local_pool)
+
                     # Two-cursor dual export
                     exported, posting_cursor, ts_cursor = await _export_postings_dual(
                         local_pool,
@@ -1495,6 +1516,7 @@ async def run_exporter(
                         maps,
                         supa_backoff,
                         ts_backoff,
+                        cutoff=cutoff,
                     )
                     # Save both cursors in a single transaction so a crash
                     # between writes cannot leave one cursor advanced while
@@ -1513,8 +1535,12 @@ async def run_exporter(
                         exported = 0
                     else:
                         try:
+                            cutoff = await cutoff_factory(local_pool)
                             exported, posting_cursor = await _export_changed_postings(
-                                local_pool, supa_pool, posting_cursor
+                                local_pool,
+                                supa_pool,
+                                posting_cursor,
+                                cutoff=cutoff,
                             )
                         except Exception as exc:
                             if not _is_downstream_unavailable(exc):
@@ -1727,6 +1753,7 @@ async def run_reconciliation(
         ts_discrepancies = await _reconcile_typesense(local_pool)
         discrepancies += ts_discrepancies
 
+    reconciliation_discrepancies.inc(discrepancies)
     log.info("reconciliation.completed", discrepancies=discrepancies)
     return discrepancies