# Connectors — runtime guarantees (restart, checkpoints, anomaly guard, alerts)

One page per connector lives next to this file (`clinvar.md`, `clinicaltrials.md`, …). This page
documents what the SDK (`packages/connectors/src/sdk/run.ts`) guarantees for **every** connector,
so a connector author only has to respect three rules.

## Rules for connector authors

1. **Mutate `ctx.cursor` only at a safe point** — after a page / batch is fully persisted
   (`upsertSourceRecord`, canonical rows, provenance). The SDK may persist the cursor at any moment
   after that; a cursor must never be ahead of the data.
2. **Call `ctx.shouldStop()` between pages** and return when it is true (time budget, record cap or
   abort request). Do not catch and swallow the return — the run status is decided by `runConnector`.
3. **Use the SDK writers** (`ctx.upsertSourceRecord`, `ctx.upsertSourceRecordsBatch`) or, when you
   keep your own batch writer (ClinVar), call `await ctx.checkpoint(n)` after each persisted batch.

## What the SDK guarantees

### Cursor persistence (CLAUDE.md §90)

| Moment | What is written | Where |
|---|---|---|
| Run start | `cursor_before` (the cursor the run starts from), `last_attempt_at` | `ingest_runs`, `connector_cursors` |
| 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` |
| `await ctx.saveCursor()` | immediately, serialised (concurrent calls share one write) | same |
| SIGTERM / SIGINT | cursor saved, run marked `aborted` (provisionally, then finalised when `sync()` returns), raw-lake gzip streams flushed | same + `system_alerts` |
| End of run (`succeeded`, `partial`, `failed`, `aborted`) | final cursor, counters, log, schema drift, anomaly | `ingest_runs`, `connector_cursors` |

`dry_run` and `probe` modes never write the cursor (`ctx.persistsCursor === false`), so a smoke run
cannot move a production cursor. `--mode full` and `--reset-cursor` start from `{}`.

### Restart semantics

- **Killed run (SIGKILL, OOM, power loss)**: the next run resumes from the last checkpoint —
  at most `checkpointEvery` records (or 60 s of work) are re-fetched; records are hashed
  (`source_records.payload_hash`), so re-processing is idempotent (`unchanged`, no duplicates).
- **SIGTERM / SIGINT (PM2 stop/restart, Ctrl-C)**: the first signal makes `shouldStop()` return
  true, saves the cursor and marks the run `aborted`; the connector returns at its next page
  boundary and `runConnector` finalises the run. A second signal exits the process immediately
  (status and cursor are already persisted). PM2's default `kill_timeout` is 1.6 s; the provisional
  bookkeeping is done first precisely so that a hard kill after it still leaves a consistent state.
  Handlers are installed for the duration of the run only (`handleSignals: false` disables them).
- **Time budget** (`--max-minutes`, `CI_MAX_RUN_MINUTES`): status `partial`, cursor saved; the
  worker's next scheduled run (or `first-run.sh`'s retry loop) continues.
- **Failure**: status `failed`, cursor saved as it was at the last safe point, health `failing`.
- **Anomaly** (see below): status `failed` with `ingest_runs.anomaly` set; nothing destructive
  happened, the previous data stays published.

### Anomaly guard (CLAUDE.md §171)

`await ctx.guardCount(entity, declaredTotal)` compares a total the source declares (or the number
of parsed records) with the previous **successful** run's `records_fetched` for the connector. If
`declaredTotal < previous × manifest.anomalyGuard.minRatioOfPrevious` (default 0.5) it throws
`AnomalyError`, sets `ctx.anomaly`, and `runConnector` records the run as `failed` with
`ingest_runs.anomaly`, health `failing`, and a **critical** `anomaly` alert. The first run of a
connector is accepted as the baseline. Wired in: OncoTree (tumour types), NCIt (concepts +
qualified states + retired), ClinicalTrials.gov (`totalCount` of a full crawl); connectors with a
fixed floor (HGNC ≥ 30 k rows, GDC ≥ 40 projects, USCS ≥ 20 k rows…) keep their own checks — any
error message starting with `anomaly:` is also recorded in `ingest_runs.anomaly`.

### Schema drift (CLAUDE.md §25)

Observed field names/types per entity are stored in `connector_field_stats`; new fields or type
changes are listed in `ingest_runs.schema_drift`, set health to `degraded` and raise a `schema_drift`
alert (resolved automatically on the next drift-free run).

### Alerts (CLAUDE.md §170)

`system_alerts` rows (`packages/database/src/alerts.ts`: `raiseAlert`, `resolveAlerts`,
`listAlerts`) are deduplicated on kind + connector + message while open/acknowledged (`count`
increments). Raised by the SDK: `connector_failure` (warn), `anomaly` (critical),
`connector_aborted` (info, or warn when credentials are missing), `schema_drift` (warn). Raised by
the worker's hourly health probe: `source_failing`, `source_stale` (no success within 2× the
schedule interval). A successful run resolves the connector's failure/aborted/anomaly/stale/failing
alerts. Read them with `pnpm cix alerts`, `pnpm cix doctor`, or `GET /v1/admin/alerts`.

## Operator cheat-sheet

```bash
pnpm cix run clinvar --max-minutes 30        # partial → cursor saved → run again to continue
pnpm cix run clinvar --mode backfill         # reprocess a dataset version already marked complete
pnpm cix run oncotree --reset-cursor         # start from scratch (idempotent thanks to payload hashes)
pnpm cix doctor                              # readiness: env, DB, migrations, connectors, stale, alerts
pnpm cix alerts [--status resolved] ; pnpm cix alerts ack 12 ; pnpm cix alerts resolve 12
psql -c "SELECT connector_id, cursor, last_success_at, health FROM connector_cursors"
psql -c "SELECT id, status, records_fetched, anomaly, cursor_after FROM ingest_runs WHERE connector_id='clinvar' ORDER BY started_at DESC LIMIT 5"
```

Tests: `packages/connectors/src/sdk/run.test.ts` (fake paged connector: checkpoint every 5 records,
SIGTERM at page 3 → `aborted` + resume from page 3, anomaly guard, alert dedupe, doctor) and
`lake.test.ts` (5 000 concurrent raw writes, single drain listener). The database-backed tests run
only when `CI_TEST_DATABASE_URL` (or `DATABASE_URL`) is set.