# 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

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.
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.
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.