Worker HTTP API

The worker (karet-worker) is a Rust/Axum service that runs pipeline jobs. It binds to :8080 inside the compose network.

Jobs do not arrive over HTTP. The worker consumes them from the Valkey stream karet:jobs:stream via a consumer group; see Architecture. The HTTP surface is three endpoints:

Endpoints

GET /health

Liveness and readiness. Reports Redis connectivity, stream length, and in-flight job count; returns 503 when Valkey is unreachable.

{ "redis": "ok", "queue_depth": 0, "in_flight": 1 }

GET / also answers 200. RustFS probes the webhook origin's root before delivering events.

POST /config/validate

Type-check a candidate pipeline.json body without running anything. Used by the graph editor's "Validate" button.

Auth: Authorization: Bearer <KARET_WORKER_TOKEN>.

Request body: a PipelineConfig JSON (see Pipeline config).

Response: always 200, with { "ok": true } or { "ok": false, "errors": [{ "kind", "message", "path" }] } where path is a JSON Pointer to the offending config node.

POST /events/s3

RustFS object-created notifications; drives auto-runs.

Auth: the webhook secret, as X-Karet-Webhook-Secret: <secret> or an Authorization header, bare or Bearer-prefixed. RustFS's WEBHOOK_AUTH_TOKEN sends whichever form it's configured with.

Accepts the standard S3 event envelope (Records[].eventName, Records[].s3.bucket.name, Records[].s3.object.key). Only s3:ObjectCreated:* events for the lake bucket are acted on. Each key extends the debounce window of every pipeline with a source path_prefix that the key starts with. The worker reads those prefixes from each pipeline's live config in Postgres, and caches them for 30 seconds.

Job lifecycle (queue, not HTTP)

A stream message carries { job_id, pipeline, prefix, clean_run, trigger, enqueued_at }. For each claimed job the worker:

1. Takes the per-pipeline run lock (at most one run per pipeline, cluster-wide; lock-busy jobs defer and retry).
2. Validates the config; an invalid config fails the job without retry.
3. Lists each source prefix and decides per mapping whether to run in full or incrementally, then downloads only the files that needs. clean_run: true runs every mapping in full and starts the new table version from an empty file list, so removed source files don't leave stale partitions.
4. Streams progress into karet:jobs:live:<id> (stage, file and mapping counters, partitions written).
5. Publishes the table version, updates the job's row in Postgres with its outcome, then acks.

Transient failures (e.g. S3 unreachable) retry with exponential backoff up to MAX_ATTEMPTS. Jobs whose worker died are reclaimed by another worker after ~2 minutes of pending-idle. On SIGTERM the worker stops claiming, finishes in-flight jobs, and exits.

Required env vars

All required unless noted; the worker fails fast if any is missing.

Variable	Purpose
S3_BUCKET_LAKE	Bucket for raw source files (default karet-lake).
S3_BUCKET_WAREHOUSE	Bucket for Parquet output (default karet-warehouse).
AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION	S3 credentials.
AWS_ENDPOINT_URL	S3 endpoint URL (e.g. http://rustfs:9000 for local dev).
REDIS_URL	Valkey/Redis connection string (e.g. redis://valkey:6379).
DATABASE_URL	Postgres connection string. The worker reads config versions and writes job rows.
DATABASE_POOL_MAX	Optional. Postgres connections the worker opens (default 4).
KARET_WORKER_TOKEN	Bearer token for POST /config/validate; the web service sends the same value.
KARET_WEBHOOK_SECRET	Shared secret for POST /events/s3.
WORKER_CONCURRENCY	Optional. Jobs processed concurrently per worker (default 1).
MAX_ATTEMPTS	Optional. Delivery attempts before a job is terminally failed (default 3).
JOB_LOCK_TTL_MS / HEARTBEAT_MS	Optional. Per-pipeline lock TTL and heartbeat interval (defaults 90000 / 30000).
PORT	Optional HTTP listen port (default 8080).

Scaling

Run more workers (docker compose up --scale worker=N or more replicas): the consumer group distributes jobs, and per-pipeline locks keep each pipeline serialized. A full run holds every source file of the pipeline in memory at once, and the parsed rows on top of that, so size worker memory to several times your largest pipeline's raw data. An incremental run holds only the new files and the partitions they touch. Polars uses every CPU unless POLARS_MAX_THREADS limits it.