Emil_Shanaty/SciMesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ci+docs: GitHub Actions pipeline and a worker-author guide

Browse Source 
- .github/workflows/coordinator.yml: vet, gofmt, race unit tests, golangci-lint,
  and the integration suite against a Postgres service — on every push/PR that
  touches coordinator/.
- docs/building-workers.md: a from-scratch guide for implementing a worker (the
  claim/heartbeat/upload/complete loop, auth, lease semantics, status codes,
  and the do-not-break rules), pointing at openapi.yaml for client generation.
- README: refresh the endpoint table, status, and test sections to match reality.
This commit is contained in:
Efremenko Arhip
2026-07-23 17:36:26 +03:00
parent 3b41455b20
commit 6829632651
3 changed files with 331 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/coordinator.yml
+66
View File
@@ -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
coordinator/README.md
+44 -29
View File
@@ -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.
docs/building-workers.md
+221
View File
@@ -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 <COORDINATOR_TOKEN>
```
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": "<uuid>", "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": "<uuid>", "capabilities": ["similarity_search"] }
```
- `200` → a leased task (below).
- `204` → nothing to do; back off a little and poll again.
```json
{
"task_id": "<uuid>",
"attempt": 1,
"lease_expires_at": "2026-07-22T12:05:00Z",
"workload": "similarity_search",
"input": { "uri": "/tasks/<uuid>/input", "sha256": "<hex>" },
"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/<uuid>/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": "<uuid>", "attempt": 1 }
```
Response: `{ "lease_expires_at": "<new deadline>" }`.
- 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: <uuid>
X-Task-Attempt: 1
<streamed result bytes>
```
Response: `{ "artifact_id": "<uuid>", "uri": "...", "sha256": "<hex>", "size_bytes": 1234 }`.
Keep the returned `artifact_id`.
## 6. Complete the task
```http
POST /tasks/{task_id}/result
{ "worker_id": "<uuid>", "attempt": 1,
"result": { "artifact_id": "<uuid>" },
"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": "<uuid>", "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.