spb/cancerindex

1# Connectors — runtime guarantees (restart, checkpoints, anomaly guard, alerts)23One page per connector lives next to this file (`clinvar.md`, `clinicaltrials.md`, …). This page4documents what the SDK (`packages/connectors/src/sdk/run.ts`) guarantees for **every** connector,5so a connector author only has to respect three rules.67## Rules for connector authors891. **Mutate `ctx.cursor` only at a safe point** — after a page / batch is fully persisted10   (`upsertSourceRecord`, canonical rows, provenance). The SDK may persist the cursor at any moment11   after that; a cursor must never be ahead of the data.122. **Call `ctx.shouldStop()` between pages** and return when it is true (time budget, record cap or13   abort request). Do not catch and swallow the return — the run status is decided by `runConnector`.143. **Use the SDK writers** (`ctx.upsertSourceRecord`, `ctx.upsertSourceRecordsBatch`) or, when you15   keep your own batch writer (ClinVar), call `await ctx.checkpoint(n)` after each persisted batch.1617## What the SDK guarantees1819### Cursor persistence (CLAUDE.md §90)2021| Moment | What is written | Where |22|---|---|---|23| Run start | `cursor_before` (the cursor the run starts from), `last_attempt_at` | `ingest_runs`, `connector_cursors` |24| Automatic checkpoint | current `ctx.cursor` **if it changed** since the last save — every `manifest.checkpointEvery` upserted records (default 2 000) **or** every 60 s, evaluated in `upsertSourceRecord` / `upsertSourceRecordsBatch` / `ctx.checkpoint()` | `connector_cursors.cursor`, `ingest_runs.cursor_after` |25| `await ctx.saveCursor()` | immediately, serialised (concurrent calls share one write) | same |26| SIGTERM / SIGINT | cursor saved, run marked `aborted` (provisionally, then finalised when `sync()` returns), raw-lake gzip streams flushed | same + `system_alerts` |27| End of run (`succeeded`, `partial`, `failed`, `aborted`) | final cursor, counters, log, schema drift, anomaly | `ingest_runs`, `connector_cursors` |2829`dry_run` and `probe` modes never write the cursor (`ctx.persistsCursor === false`), so a smoke run30cannot move a production cursor. `--mode full` and `--reset-cursor` start from `{}`.3132### Restart semantics3334- **Killed run (SIGKILL, OOM, power loss)**: the next run resumes from the last checkpoint —35  at most `checkpointEvery` records (or 60 s of work) are re-fetched; records are hashed36  (`source_records.payload_hash`), so re-processing is idempotent (`unchanged`, no duplicates).37- **SIGTERM / SIGINT (PM2 stop/restart, Ctrl-C)**: the first signal makes `shouldStop()` return38  true, saves the cursor and marks the run `aborted`; the connector returns at its next page39  boundary and `runConnector` finalises the run. A second signal exits the process immediately40  (status and cursor are already persisted). PM2's default `kill_timeout` is 1.6 s; the provisional41  bookkeeping is done first precisely so that a hard kill after it still leaves a consistent state.42  Handlers are installed for the duration of the run only (`handleSignals: false` disables them).43- **Time budget** (`--max-minutes`, `CI_MAX_RUN_MINUTES`): status `partial`, cursor saved; the44  worker's next scheduled run (or `first-run.sh`'s retry loop) continues.45- **Failure**: status `failed`, cursor saved as it was at the last safe point, health `failing`.46- **Anomaly** (see below): status `failed` with `ingest_runs.anomaly` set; nothing destructive47  happened, the previous data stays published.4849### Anomaly guard (CLAUDE.md §171)5051`await ctx.guardCount(entity, declaredTotal)` compares a total the source declares (or the number52of parsed records) with the previous **successful** run's `records_fetched` for the connector. If53`declaredTotal < previous × manifest.anomalyGuard.minRatioOfPrevious` (default 0.5) it throws54`AnomalyError`, sets `ctx.anomaly`, and `runConnector` records the run as `failed` with55`ingest_runs.anomaly`, health `failing`, and a **critical** `anomaly` alert. The first run of a56connector is accepted as the baseline. Wired in: OncoTree (tumour types), NCIt (concepts +57qualified states + retired), ClinicalTrials.gov (`totalCount` of a full crawl); connectors with a58fixed floor (HGNC ≥ 30 k rows, GDC ≥ 40 projects, USCS ≥ 20 k rows…) keep their own checks — any59error message starting with `anomaly:` is also recorded in `ingest_runs.anomaly`.6061### Schema drift (CLAUDE.md §25)6263Observed field names/types per entity are stored in `connector_field_stats`; new fields or type64changes are listed in `ingest_runs.schema_drift`, set health to `degraded` and raise a `schema_drift`65alert (resolved automatically on the next drift-free run).6667### Alerts (CLAUDE.md §170)6869`system_alerts` rows (`packages/database/src/alerts.ts`: `raiseAlert`, `resolveAlerts`,70`listAlerts`) are deduplicated on kind + connector + message while open/acknowledged (`count`71increments). Raised by the SDK: `connector_failure` (warn), `anomaly` (critical),72`connector_aborted` (info, or warn when credentials are missing), `schema_drift` (warn). Raised by73the worker's hourly health probe: `source_failing`, `source_stale` (no success within 2× the74schedule interval). A successful run resolves the connector's failure/aborted/anomaly/stale/failing75alerts. Read them with `pnpm cix alerts`, `pnpm cix doctor`, or `GET /v1/admin/alerts`.7677## Operator cheat-sheet7879```bash80pnpm cix run clinvar --max-minutes 30        # partial → cursor saved → run again to continue81pnpm cix run clinvar --mode backfill         # reprocess a dataset version already marked complete82pnpm cix run oncotree --reset-cursor         # start from scratch (idempotent thanks to payload hashes)83pnpm cix doctor                              # readiness: env, DB, migrations, connectors, stale, alerts84pnpm cix alerts [--status resolved] ; pnpm cix alerts ack 12 ; pnpm cix alerts resolve 1285psql -c "SELECT connector_id, cursor, last_success_at, health FROM connector_cursors"86psql -c "SELECT id, status, records_fetched, anomaly, cursor_after FROM ingest_runs WHERE connector_id='clinvar' ORDER BY started_at DESC LIMIT 5"87```8889Tests: `packages/connectors/src/sdk/run.test.ts` (fake paged connector: checkpoint every 5 records,90SIGTERM at page 3 → `aborted` + resume from page 3, anomaly guard, alert dedupe, doctor) and91`lake.test.ts` (5 000 concurrent raw writes, single drain listener). The database-backed tests run92only when `CI_TEST_DATABASE_URL` (or `DATABASE_URL`) is set.93