Provider worker contract

Every provider worker (strava, garmin, yours) is a standalone process in its own repo — see Provider workers. The only contract with cairn-core is:

1. the protobuf messages in cairn.worker.v1 (committed stubs in each worker repo), and
2. the NATS subjects + behaviors specified here.

This page is normative: MUST items are required for a worker to be compliant — core's scheduler, import queue, and UI assume them. Run the conformance harness (§10) against a live worker to verify mechanically.

Wire conventions: job payloads (server → worker) are hand-rolled JSON; results (worker → server) are protojson-encoded cairn.worker.v1.JobResult. All job subjects live on the CAIRN_JOBS work-queue stream; all result subjects on the CAIRN_RESULTS interest stream — see the pipeline.

1. Identity and connection

• The worker connects to NATS with an enrollment token (production, auth-callout) or static/anonymous credentials (dev). The admin sets the provider on the enrollment; any well-formed type ([a-z0-9_-], 1–48 chars) is accepted, so a new provider's worker can enroll without a core change.
• Under auth-callout the worker MUST stay within core's standard permission template: the exact job and result subjects of §3–§7 for its provider, its own cairn.workers.<worker_key>.>, cairn.cmd.worker.<worker_key>.> and webhook subjects, and the JetStream API subjects for its pull consumers on CAIRN_JOBS / CAIRN_WEBHOOKS and the presence, manifest and rate-limit KV buckets. The full list is on Self-host → Providers. Anything else is refused with Permissions Violation.
• Identity: provider (e.g. garmin), package (stable build identity, e.g. cairn.workers.garmin) and the parser version (see Versioning). Every published JobResult MUST carry worker_name, worker_version (the parser version N.M) and worker_package.

Versioning

A worker carries three independent versions (from 0.5.0):

Version	Form	Meaning	Bumped when
Release	v0.5.1	the core contract the worker implements	every release; major.minor follows core
Dataset	N	format of the archived raw blob (version in the archive envelope)	a parser can no longer read the previous format
Parser	N.M	parser for dataset N, output revision M	every change to the parsed output

• Release compatibility: before 1.0 a worker is compatible with core when both releases share major and minor (v0.5.3 worker with v0.5.0 core); patch levels are free. From 1.0 on, minor and patch releases are backwards compatible: same major, and the worker's minor not above core's (v1.1.4 worker with v1.3.0 core works; v1.4.0 with v1.3.0 does not). Core publishes its release under info in the cairn_core KV bucket; a worker MUST refuse to run against an incompatible core. Dev builds are exempt.
• Parser N.x MUST read every dataset N archive, including archives written by older importers; fields missing from them are absent, not an error. It does not read other dataset versions.
• Bump M on any change to the parsed output (new field, fixed mapping, changed unit). A refactor with byte-identical output does not bump it. A new dataset version resets M: 1.4 → 2.0.
• Forward only. A source parsed by N.M is only re-parsed or re-fetched by a parser >= N.M. Core rejects results that would lower a source's version; a worker given a parse_blob job with a higher min_parser_version than its own MUST NAK it so a newer pooled instance takes it.
• A legacy bare integer N reads as N.0.
• The archive MUST hold every input the parser uses, so a re-parse reproduces the import. A parser change needing a new input (e.g. a separate gear call) first archives it — additive, still dataset N; older archives lack it and need one re-fetch.

What core offers per source, given the live workers:

Live worker vs. source	Offered
same N.M	re-parse (re-run)
same N, higher M	re-parse from archive (no provider quota)
higher N	re-fetch from provider
older	nothing — would downgrade

2. Presence heartbeat — MUST

Write a JSON presence record to the cairn_worker_presence KV bucket every ~20s under the key <worker_name>.<instance_id>:

{"worker_name": "...", "worker_key": "...", "instance_id": "...",
 "version": "<parser N.M>", "release": "<vX.Y.Z>", "provider": "...", "package": "...",
 "webhooks": false, "capabilities": {...}, "last_seen": "<RFC3339>"}

Core uses presence for the worker-admin UI, connect-flow gating, and manage-view liveness.

3. Job consumption — MUST

Jobs are routed to pools (from 0.5.0): every instance with the same worker_name for a provider is one pool, and core publishes each job to one pool on cairn.jobs.<job>.<provider>.<worker_name>. An external account is assigned to one pool; its imports, re-fetches, gear and metrics jobs go there. Re-parse goes to a pool of the source's own package. Several pools (e.g. an API importer and a file importer) can serve one provider side by side: the user connects the same provider account once per pool they want, each connection with its own sync state, tokens and queue. Rate-limit budgets are per provider app: the token reply carries app_id (a hash of the connection's client id), and the worker keys its buckets <provider>:<app_id>:<window> (stored as <provider>.<app_id>.<window> in the cairn_rate_limits KV). A token without app_id (older core) uses the pool's bucket <provider>:<worker_name>:<window>. Connections of one provider account on different pools MUST use different apps. Core groups the resulting sources into one activity and merges them by its priority list, where <provider>.<worker_name> names one pool. A pool name is permanent: core binds it to the first package it reports and refuses another package under that name, so a different implementation needs a new name.

Create a durable pull consumer on CAIRN_JOBS for every job subject of your pool, filtered on the exact subject. CAIRN_JOBS is a work queue: it refuses two consumers with the same filter, so instances of one pool share one durable per subject. The permission template only allows filtered consumer creation to these subjects:

Subject	Required	Purpose
cairn.jobs.fetch_source.<provider>.<worker_name>	MUST	import one activity
cairn.jobs.reconcile.<provider>.<worker_name>	MUST	catch new activity since watermark
cairn.jobs.parse_blob.<provider>.<worker_name>	MUST if raw blobs are archived	quota-free re-parse
cairn.jobs.import_metrics.<provider>.<worker_name>	MUST if manifest declares METRIC events	health metrics
cairn.jobs.import_gear.<provider>.<worker_name>	MUST if manifest declares Gear	the account's full gear list (0.4.0)
cairn.jobs.backfill.<provider>.<worker_name>	MAY	legacy full-history walk

The 0.4 job subjects without <worker_name> stay subscribable through 0.5 so an upgraded worker drains jobs queued before the upgrade (reuse the old durable name; a work queue lets only one pool drain them). They are removed in 0.6. Results and lookups are pool-scoped only. Workers publish their own fetch_source sub-jobs (reconcile, webhooks) to their own pool subject. Results stay provider-scoped: cairn.results.<job>.<provider>.

Why "every subject" is a MUST

CAIRN_JOBS is a work-queue stream — messages for a subject nobody consumes are retained forever. A worker without a reconcile consumer leaves the scheduler's reconcile jobs queued indefinitely and new activities are never auto-imported.

Ack discipline (the delivery semantics core assumes):

• Success → ack.
• Transient failure (network, 5xx, DB blip) → NAK; provider rate limit → NAK with delay anchored to the limit window.
• Poison (bad payload, activity gone, unmappable data) → publish a failure envelope (§6) then Term.
• Long-running handling MUST extend the ack deadline (in_progress) on a cadence below the AckWait of the job consumer the worker creates (2 min in the Strava worker, 30 s in the Garmin worker; core's CAIRN_NATS_JOB_ACK_WAIT does not apply to worker-created consumers): heartbeat every fetched message from batch arrival until its disposition, otherwise JetStream redelivers mid-processing and the activity is re-imported on every redelivery.
• Bound retries: Term (with failure envelope) after ~5 deliveries of a persistently failing job rather than NAK-looping forever.

4. Job payloads (server → worker, JSON)

fetch_source:

{"job_id": "…", "account_id": "<external_account uuid>", "user_id": "<uuid>",
 "provider": "…", "ext_id": "…", "fetch_streams": true,
 "reason": "backfill|webhook|reconcile"}

reconcile:

{"job_id": "…", "account_id": "…", "user_id": "…", "provider": "…",
 "watermark": "<RFC3339, optional>", "max_enqueue": 500,
 "known_ext_ids": ["…", "…"]}

import_gear (0.4.0): {"job_id", "account_id", "user_id", "provider", "reason"} — answer with one GEAR event per gear on the account, retired included. Core publishes it on account connect, on the user's "Sync gear" and alongside every reconcile, only when a live worker for the provider declares Gear in its manifest.

parse_blob: as fetch_source plus "blob": {"url": "<presigned GET>", "expires_at": "…", "fallback_handle": "<blob key>"} and "min_parser_version": "N.M" (the source's current parser version; NAK when yours is lower or reads another dataset) — on an expired URL, re-presign via cairn.blobs.presign_download.<provider> with the fallback handle.

5. Results — claim-check — MUST

Results are protojson JobResult published to cairn.results.<job_type>.<provider>.<worker_name> (reconcile appends the account id). Core checks that worker_name in the result equals the pool token of the subject, that the name's package never changes, and that an activity result comes from the account's pool (fetch_source) or from the package that imported the source (parse_blob). The pre-0.5 cairn.results.<job_type>.<provider> form is no longer granted to workers.

Event-carrying results MUST be claim-checked (streams routinely exceed the broker's max_payload; inline oversized publishes used to Term and wedge the import queue):

1. Marshal the full JobResult (protojson).
2. Request a presigned PUT via cairn.blobs.presign_upload.<provider> with "kind": "result", content_type, content_length and the hex content_sha256 of the body.
3. PUT the body to the returned URL, sending required_headers verbatim.
4. Publish only the envelope: worker stamp + new_watermark/more_available
   • payload_ref {blob_id, size_bytes, content_type, content_sha256}, no inline events. Core fetches the object, ingests, and deletes it; a bucket lifecycle rule reaps orphans after ~1 day.

Event-free results (reconcile watermark, backfill stamps) publish inline; import_gear results are small and MAY publish inline too.

Gear events (0.4.0) — MAY: any result may carry WORKER_EVENT_TYPE_GEAR events (ImportedGear{ref, payload: GearImport}), keyed by (external_account_id, external_id). They precede the activity that references them, like segments. The activity's ActivitySourcePayload.gear_external_id names the provider's gear id and gear_name is a display hint; the id MAY reference gear core has not seen yet (it is resolved once the gear arrives). Core never creates gear from a GEAR event on its own: an unknown record is stored pending and the user decides. Gear is optional data: a worker MUST NOT fail or delay an activity import because a gear lookup failed, and ref.user_id must be the owner of external_account_id (core rejects a mismatch).

Publish routing — SHOULD: results and fetch_source sub-jobs go to JetStream streams (CAIRN_RESULTS, CAIRN_JOBS) and need a JetStream publish with a Nats-Msg-Id; token, credential, presign and discover subjects are core NATS request/reply. Decide per subject before publishing — look up whether a stream covers it ($JS.API.STREAM.NAMES with a subject filter, granted by the template) and cache the answer — instead of trying a JetStream publish first: a JetStream publish to a core-only subject waits for an ack that never comes and delivers twice on retry.

Presign/upload failures are transient (NAK + retry) — never drop a result body silently.

Message-ids (JetStream dedup): activity results result:<job_id>:d<delivery_attempt> (the attempt suffix lets a deferred retry re-publish past the dedup window; ingest is idempotent), reconcile and import_gear results result:<job_id>, fetch_source sub-jobs enqueued by reconcile reconcile:<provider>:<account_id>:<ext_id>.

Stream hygiene — SHOULD: emit strictly unique, ordered sample timestamps. Core dedupes defensively (last sample wins), but providers do ship duplicate samples around pauses — clean them at the source.

6. Failure envelope — MUST

Before Terming a job that can never succeed, publish a small JobResult carrying only:

• error: WorkerError {class, code, message} (no events), and
• failed_ref: ExternalRef {user_id, provider, external_account_id, external_id} for import-queue correlation.

Core fails the queue item immediately with the true reason. Without it the item sits in_progress until the stale reaper masks the cause as stale: no result after N dispatch attempts.

7. Reconcile behavior — MUST

• Window: list from max(watermark − 1h, now − 30d) — never the full history (that is the user-driven import queue's job).
• Enqueue fetch_source sub-jobs for unknown ext-ids, capped by max_enqueue, with the deterministic msg-ids above.
• Publish the result to the account-suffixed subject cairn.results.reconcile.<provider>.<external_account_id> — and publish it even when nothing new was found. Advancing last_sync_at is what stops the scheduler from re-polling the account every tick; core also applies new_watermark (advance-only) from it.
• The scheduler's target cadence is CAIRN_SCHEDULER_POLLING_INTERVAL (default 60m, per-connection override). Workers SHOULD coalesce reconcile jobs per account (~10 min) so a scheduler backlog drains as cheap acks.

8. Discover — MUST

Subscribe to and respond on plain request/reply cairn.discover.<provider>.<worker_name> (the template allows subscribing to it); core asks the account's pool:

Request: {"account_id": "…", "start_page": 1} Reply: {"items": [{"item_type": "activity", "external_id": "…", "item_time": "<RFC3339>"}], "next_page": 2} — or "complete": true at the end, "rate_limited": true when the budget is exhausted mid-walk, or {"error": "…"}. Never return an empty success when the real cause was an error or rate limit — the UI would show "0 importable activities".

9. Credentials & rate limits

• Fetch credentials fresh per job over request/reply — OAuth providers: cairn.tokens.<provider>.get (+ .store, .needs_reauth); custom-cred providers: cairn.creds.<provider>.get. Never persist credentials in the worker beyond in-memory caches.
• Workers hold zero S3 credentials — all blob I/O goes through presigned URLs (§4/§5).
• Rate-limited providers: reserve from the shared budget KV (cairn_rate_limits) before each API call, sync the limiter from the provider's usage headers on every response, anchor 429 backoff to the real window reset, and report RateLimitSnapshots opportunistically.

9a. Account lookup (webhooks)

A webhook event names the provider's owner id, not a Cairn account. Resolve it with request/reply on your pool's subject cairn.accounts.lookup.<provider>.<worker_name>:

{"external_id": "<owner id>", "subscription_id": "<optional>"}

Reply: {"account_id": "…", "user_id": "…"} or {"error": "not_found" | "auto_import_suspended" | …}. subscription_id is preferred when known. The same provider account may be connected once per pool; core returns the connection of the pool named in the subject (a connection without a pool matches any). The pre-0.5 cairn.accounts.lookup_by_provider_ext is no longer granted to workers.

10. Conformance harness

cairn-core ships a checker that verifies the mechanical parts of this contract against a running worker on a dev/staging NATS:

# from a cairn-core checkout
go run ./cmd/worker-conformance --nats nats://localhost:4222 --provider garmin

It asserts: presence heartbeat exists and is fresh; durable consumers exist on CAIRN_JOBS for every required subject (catches the §3 incident class); discover answers with a valid shape; and (with --active) a synthetic reconcile job round-trips to an account-suffixed result. Non-zero exit on any violation — wire it into the provider repo's CI against its dev stack where possible.