diff --git a/.github/workflows/coordinator.yml b/.github/workflows/coordinator.yml new file mode 100644 index 0000000..11a5607 --- /dev/null +++ b/.github/workflows/coordinator.yml @@ -0,0 +1,66 @@ +name: coordinator + +on: + push: + paths: + - "coordinator/**" + - ".github/workflows/coordinator.yml" + pull_request: + paths: + - "coordinator/**" + - ".github/workflows/coordinator.yml" + +defaults: + run: + working-directory: coordinator + +jobs: + test: + runs-on: ubuntu-latest + + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: scimesh + POSTGRES_PASSWORD: scimesh + POSTGRES_DB: scimesh + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U scimesh" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + + env: + TEST_DATABASE_URL: postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: coordinator/go.mod + cache-dependency-path: coordinator/go.sum + + - name: go vet + run: go vet ./... + + - name: gofmt + run: test -z "$(gofmt -l .)" || (gofmt -l . && exit 1) + + - name: unit tests (race) + run: go test -race ./... + + - name: lint + run: go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 run --build-tags=integration ./... + + - name: install migrate CLI + run: go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.17.1 + + - name: apply migrations + run: migrate -path migrations -database "$TEST_DATABASE_URL" up + + - name: integration tests + run: go test -tags=integration ./internal/storage/postgres/ -v diff --git a/coordinator/README.md b/coordinator/README.md index 8a10bac..c190806 100644 --- a/coordinator/README.md +++ b/coordinator/README.md @@ -101,15 +101,24 @@ See `.env.example`; only `DATABASE_URL` is required. ## Endpoints -| Method | Path | Purpose | -| ------ | ----------------------------- | ------------------------------------------ | -| POST | `/jobs` | Create job + pending tasks transactionally | -| POST | `/tasks/claim` | Atomically lease one task (`204` if none) | -| POST | `/tasks/{task_id}/heartbeat` | Renew the caller's lease | -| POST | `/tasks/{task_id}/result` | Record a completed result (idempotent) | -| POST | `/tasks/{task_id}/failure` | Record failure / retryable state | -| GET | `/jobs/{job_id}` | Aggregate job progress | -| GET | `/health` | Liveness (unauthenticated) | +| Method | Path | Purpose | +| ------ | ---------------------------------- | --------------------------------------------- | +| POST | `/workers/register` | Register a worker, get its id | +| POST | `/jobs` | Create job + tasks from chunk URIs | +| POST | `/jobs/upload` | Upload a dataset; coordinator chunks it | +| GET | `/jobs/{job_id}` | Aggregate job progress | +| POST | `/tasks/claim` | Atomically lease one task (`204` if none) | +| GET | `/tasks/{task_id}/input` | Download the task's input shard | +| POST | `/tasks/{task_id}/heartbeat` | Renew the caller's lease (→ `running`) | +| PUT | `/tasks/{task_id}/artifacts/{name}`| Upload a partial-result artifact | +| POST | `/tasks/{task_id}/result` | Complete with an artifact id (idempotent) | +| POST | `/tasks/{task_id}/failure` | Record failure / retryable state | +| GET | `/artifacts/{artifact_id}/download`| Download an artifact by id | +| GET | `/health` | Readiness incl. database (unauthenticated) | + +The full contract is in [`docs/api-contract.md`](../docs/api-contract.md) and +[`docs/openapi.yaml`](../docs/openapi.yaml); a worker-author guide is in +[`docs/building-workers.md`](../docs/building-workers.md). ## Poking the API @@ -125,32 +134,38 @@ reuse ids captured from earlier responses, so it doubles as API documentation. ## Status -The queue works end to end: a job can be submitted, split into tasks, leased to -workers one at a time, heartbeated, completed, and reflected in job progress. +Works end to end: a worker registers, a dataset is uploaded and chunked into +shard tasks (or a job is created from chunk URIs), tasks are leased one at a +time, downloaded, heartbeated (`leased → running`), completed via uploaded +result artifacts, and reflected in job progress. A reaper reclaims expired +leases and marks silent workers offline. -Roadmap: +Done: schema + migrations, atomic claim (`FOR UPDATE SKIP LOCKED`), optimistic +concurrency, result/failure paths, lease expiry, worker registry + liveness, +artifact storage, dataset upload + chunking, request-size limits. -1. schema + migrations ✅ -2. `ClaimNext`, `InsertBatch` — atomic claim via `FOR UPDATE SKIP LOCKED` ✅ -3. `GetForUpdate`, `Update`, `CountByStatus` — result/failure paths ✅ -4. file upload / chunk download — **next** -5. `ExpireLeases` ✅ (reaper + a sweep before every claim) -6. stitcher: merge per-chunk top-k into the final CSV -7. more integration coverage as features land - -Still stubbed: `StitchJob.Execute`, and there is no `POST /upload` or -`GET /download_chunk` yet — so chunk files must be referenced by URI for now. +Still stubbed: `StitchJob.Execute` — merging per-chunk top-k into the final CSV +is workload semantics that belongs to the Python side (reducer). ## Tests -`internal/domain` is covered by unit tests that need **no database** — lease -ownership, stale attempts, idempotent replays, retry budgets, and expiry are all -pure functions of entity state: +Unit tests need **no database** — domain rules, use-case orchestration (over +in-memory `internal/memstore`), and HTTP handlers (via `httptest`): ```sh -go test ./... -go vet ./... +make test # go test ./... +make vet +make lint +go test -race ./... ``` -Integration tests (concurrent claiming, migrations) come in phase 7 and require -a real PostgreSQL instance supplied through `TEST_DATABASE_URL`. +Integration tests run against a **real PostgreSQL** (the spec forbids mocks +here — they verify `FOR UPDATE SKIP LOCKED`, optimistic concurrency, rollback): + +```sh +docker compose up -d +make test-integration TEST_DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable' +``` + +CI (`.github/workflows/coordinator.yml`) runs vet, gofmt, race tests, lint, and +the integration suite against a Postgres service on every push and PR. diff --git a/docs/building-workers.md b/docs/building-workers.md new file mode 100644 index 0000000..f55d034 --- /dev/null +++ b/docs/building-workers.md @@ -0,0 +1,221 @@ +# Building a SciMesh worker + +A worker is a process that pulls tasks from the coordinator, runs them, and +returns results. It talks to the coordinator **only over HTTP** — it never sees +the database, and it needs no inbound port (all requests are outbound). This +guide is what you need to implement one (the reference is a Python daemon, but +nothing here is Python-specific). + +**Read alongside:** +[`api-contract.md`](api-contract.md) (the contract in prose) and +[`openapi.yaml`](openapi.yaml) (machine-readable — generate a typed client from +it, see the bottom). + +--- + +## The one loop + +A worker is essentially this loop: + +```text +register once +loop forever: + task = POST /tasks/claim + if no task (204): sleep, continue + download the task's input, verify its checksum + run the workload ── while running, POST heartbeat before the lease expires + upload the result artifact (PUT) + POST /tasks/{id}/result with the artifact id + on any failure: POST /tasks/{id}/failure +``` + +Everything below fills in the details. + +## 0. Auth + +Every request except `GET /health` carries a shared bearer token: + +``` +Authorization: Bearer +``` + +The token is handed to you out of band (env var / secret) — the same string the +coordinator was started with. Never log it, never send it in an error body. + +## 1. Register (once, at startup) + +```http +POST /workers/register +{ "name": "lab-worker-01", "capabilities": ["similarity_search"] } +``` + +Response: `{ "worker_id": "", "heartbeat_interval_seconds": 15 }`. + +- `capabilities` are the workload names you can run — the coordinator only hands + you matching tasks. +- **Keep `worker_id`**. Use it as your identity in every later call. Using the + registered UUID is what lets the coordinator track your liveness (it marks + workers offline after they go silent). + +## 2. Claim a task + +```http +POST /tasks/claim +{ "worker_id": "", "capabilities": ["similarity_search"] } +``` + +- `200` → a leased task (below). +- `204` → nothing to do; back off a little and poll again. + +```json +{ + "task_id": "", + "attempt": 1, + "lease_expires_at": "2026-07-22T12:05:00Z", + "workload": "similarity_search", + "input": { "uri": "/tasks//input", "sha256": "" }, + "parameters": { "query_id": "CHEMBL939", "top_k": 20 } +} +``` + +**`attempt` matters.** Every later call for this task must echo the exact +`attempt` you were handed. A task requeued after a lost lease comes back with a +higher attempt; an old attempt is rejected with `409`. + +## 3. Download the input, verify it + +```http +GET {input.uri} # e.g. GET /tasks//input +``` + +Stream it to disk and **check the SHA-256 against `input.sha256`** before +running. A mismatch means a corrupt shard — fail the task with a clear code, +don't process garbage. + +> If `input.uri` ever redirects to another host (object storage), **strip the +> `Authorization` header** on the redirect — never send the coordinator token to +> a third party. + +## 4. Run — and heartbeat while you run + +Long tasks must prove they are alive, or the coordinator's reaper reclaims the +lease and hands the task to someone else. + +```http +POST /tasks/{task_id}/heartbeat +{ "worker_id": "", "attempt": 1 } +``` + +Response: `{ "lease_expires_at": "" }`. + +- Schedule the next heartbeat at **less than half** the remaining TTL — don't + rely on a fixed interval. If `lease_expires_at` is 2 minutes out, heartbeat + every ~45s. +- The first heartbeat also moves the task from `leased` to `running` on the + server; you don't have to do anything special for that. + +If you miss the deadline, your lease expires: a later `heartbeat`/`result` will +come back `409`, and the task is already back in the queue. + +## 5. Upload the result artifact + +The coordinator owns results — you upload the bytes, it stores them and computes +the checksum. Identity travels in **headers** here, not the body: + +```http +PUT /tasks/{task_id}/artifacts/result.csv +Content-Type: text/csv +X-Worker-ID: +X-Task-Attempt: 1 + + +``` + +Response: `{ "artifact_id": "", "uri": "...", "sha256": "", "size_bytes": 1234 }`. + +Keep the returned `artifact_id`. + +## 6. Complete the task + +```http +POST /tasks/{task_id}/result +{ "worker_id": "", "attempt": 1, + "result": { "artifact_id": "" }, + "metrics": { "elapsed_seconds": 12.4, "processed_rows": 10000 } } +``` + +- Reference the `artifact_id` you just uploaded **for this task**. The + coordinator verifies it belongs to this task; another task's artifact → `409`. +- **Idempotent:** if your network dropped and you retry the same `artifact_id`, + you get `200` again, not a conflict. Safe to retry. + +## 7. …or fail it + +```http +POST /tasks/{task_id}/failure +{ "worker_id": "", "attempt": 1, + "error_code": "download_failed", "error_message": "checksum mismatch", + "retryable": true } +``` + +- `retryable: true` → the task returns to the queue while attempts remain (a new + worker gets it at a higher `attempt`). +- `retryable: false` → it fails terminally. +- Send only a short, sanitized `error_code`/`error_message`. **Never** a Python + traceback, a token, or an absolute local path. + +--- + +## Status-code cheat sheet + +| Code | Meaning for the worker | +| --- | --- | +| `204` | claim: queue empty — back off and retry | +| `400` | your request is malformed (bad UUID, unknown field) | +| `401` | bad/missing token | +| `404` | task/job/artifact doesn't exist | +| `409` | you don't hold the lease, or your `attempt` is stale, or a different result was already recorded — **stop working on this task**, it's no longer yours | + +A `409` is normal, not a crash: it means the coordinator gave the task to +someone else (usually because your lease expired). Log it and move on to the +next claim. + +## Config the worker should expose + +Per the worker contract, at minimum: + +- `SCIMESH_COORDINATOR_URL` (e.g. `http://coordinator:8080`) +- `SCIMESH_WORKER_ID` (or derive from hostname) +- the bearer token +- poll interval and request timeout +- a working directory for downloaded inputs and generated outputs + +## Generate a client from the spec + +Instead of hand-writing request code, generate it: + +```sh +# typed async client +openapi-python-client generate --path docs/openapi.yaml + +# or just the Pydantic models +datamodel-codegen --input docs/openapi.yaml --output scimesh_models.py +``` + +## Try the endpoints by hand first + +`coordinator/api/requests.http` walks the whole flow one request at a time +(register → claim → heartbeat → upload → result), and +`coordinator/scripts/smoke.sh` runs it end to end. Read those to see real +request/response bodies before writing code. + +## The rules you must not break + +1. Never touch the database — HTTP only. +2. Every mutating call carries `worker_id` **and** `attempt`. +3. Verify the input checksum before running. +4. Upload the result artifact **before** calling `/result`. +5. Never persist a `worker://` or local path as a result — the coordinator owns + artifacts. +6. Strip the bearer token on any cross-origin redirect. +7. Sanitize error output — no tracebacks, tokens, or absolute paths.