chore(users): remove coordinator leftovers, tidy for userservice
- delete api/requests.http and ARCHITECTURE.md (coordinator content) - rewrite README.md and .env.example for the userservice - drop dead RunPeriodic (reaper machinery userservice has no use for) - fix .gitignore/.dockerignore/.golangci.yml module + artifact names - degeneralize stale copied comments that said "coordinator"
This commit is contained in:
Efremenko Arhip
+3
-1
@@ -9,6 +9,8 @@ Dockerfile
|
|||||||
.dockerignore
|
.dockerignore
|
||||||
|
|
||||||
# Local build artifacts
|
# Local build artifacts
|
||||||
/coordinator
|
/userservice
|
||||||
/bin/
|
/bin/
|
||||||
*.out
|
*.out
|
||||||
|
/data/
|
||||||
|
/logs/
|
||||||
|
|||||||
+10
-19
@@ -1,32 +1,23 @@
|
|||||||
# Copy to .env and adjust. All settings are read from the environment.
|
# Copy to .env and adjust. All settings are read from the environment.
|
||||||
|
|
||||||
COORDINATOR_ADDR=:8080
|
USERSERVICE_ADDR=:8081
|
||||||
DATABASE_URL=postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable
|
DATABASE_URL=postgres://scimesh:scimesh@localhost:5433/scimesh_users?sslmode=disable
|
||||||
|
|
||||||
# Shared bearer token every worker must present. Leave empty to disable auth (dev only).
|
# Shared HS256 secret used to sign JWTs. The coordinator verifies tokens with
|
||||||
WORKER_AUTH_TOKEN=change-me
|
# this SAME secret, so the two values must match exactly. Minimum 32 bytes.
|
||||||
|
JWT_SECRET=change-me-to-a-long-random-secret-min-32-bytes
|
||||||
# Optional local operator UI. Use a separate value; never reuse the worker token.
|
# How long an issued token stays valid.
|
||||||
# When empty, /ui is disabled.
|
JWT_TTL=24h
|
||||||
UI_AUTH_TOKEN=
|
# bcrypt work factor. Empty/0 uses the library default (10).
|
||||||
|
# BCRYPT_COST=10
|
||||||
|
|
||||||
# Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only;
|
# Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only;
|
||||||
# set a path to also write a size-rotated file (kept across restarts).
|
# set a path to also write a size-rotated file (kept across restarts).
|
||||||
LOG_LEVEL=info
|
LOG_LEVEL=info
|
||||||
# LOG_FILE=./logs/coordinator.log
|
# LOG_FILE=./logs/userservice.log
|
||||||
|
|
||||||
# Directory where artifact bytes are stored.
|
|
||||||
COORDINATOR_STORAGE_DIR=./data
|
|
||||||
# Upper bound on an uploaded dataset or artifact body (bytes). Default 1 GiB.
|
|
||||||
MAX_UPLOAD_BYTES=1073741824
|
|
||||||
|
|
||||||
# Optional tuning (defaults shown).
|
# Optional tuning (defaults shown).
|
||||||
DB_MAX_CONNS=10
|
DB_MAX_CONNS=10
|
||||||
# How long to keep retrying the initial DB connection while Postgres boots.
|
# How long to keep retrying the initial DB connection while Postgres boots.
|
||||||
DB_CONNECT_TIMEOUT=30s
|
DB_CONNECT_TIMEOUT=30s
|
||||||
REQUEST_TIMEOUT=15s
|
REQUEST_TIMEOUT=15s
|
||||||
LEASE_DURATION=2m
|
|
||||||
DEFAULT_MAX_ATTEMPTS=3
|
|
||||||
REAPER_INTERVAL=30s
|
|
||||||
# A worker silent longer than this is marked offline by the reaper.
|
|
||||||
WORKER_OFFLINE_AFTER=1m
|
|
||||||
|
|||||||
+1
-1
@@ -1,4 +1,4 @@
|
|||||||
/coordinator
|
/userservice
|
||||||
/bin/
|
/bin/
|
||||||
.env
|
.env
|
||||||
*.out
|
*.out
|
||||||
|
|||||||
+1
-1
@@ -51,4 +51,4 @@ formatters:
|
|||||||
settings:
|
settings:
|
||||||
goimports:
|
goimports:
|
||||||
local-prefixes:
|
local-prefixes:
|
||||||
- github.com/emil28092005/SciMesh/coordinator
|
- github.com/emil28092005/SciMesh/users
|
||||||
|
|||||||
@@ -1,144 +0,0 @@
|
|||||||
# Архитектура координатора
|
|
||||||
|
|
||||||
Карта кода. Читать сверху вниз: сначала «где что лежит», потом «как проходит
|
|
||||||
запрос», в конце — «куда добавлять новое».
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 1. Четыре слоя
|
|
||||||
|
|
||||||
```
|
|
||||||
infra конфиг, пул БД, часы, HTTP-сервер, reaper ← драйверы
|
|
||||||
transport HTTP-хендлеры ← входящее: кто зовёт нас
|
|
||||||
storage репозитории на SQL ← исходящее: кого зовём мы
|
|
||||||
usecase операции + ПОРТЫ (интерфейсы) ← прикладные правила
|
|
||||||
domain Task, Job и их инварианты ← бизнес-правила
|
|
||||||
|
|
||||||
┌── transport ──┐
|
|
||||||
domain ◄── usecase ◄┤ ├◄── infra
|
|
||||||
└── storage ────┘
|
|
||||||
```
|
|
||||||
|
|
||||||
`transport` и `storage` — один и тот же слой (в книгах он зовётся «адаптеры»),
|
|
||||||
просто разделённый по направлению: транспорт принимает запросы снаружи, storage
|
|
||||||
обращается наружу сам. Так путь к файлу говорит о его роли, а не о категории.
|
|
||||||
|
|
||||||
**Единственное правило:** зависимости идут только внутрь. `domain` не импортирует
|
|
||||||
ничего из проекта. `usecase` видит только `domain`. `transport` и `storage` не
|
|
||||||
знают друг о друге.
|
|
||||||
|
|
||||||
Проверить в любой момент:
|
|
||||||
|
|
||||||
```sh
|
|
||||||
go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal
|
|
||||||
# пусто = правило соблюдено
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 2. Где что лежит
|
|
||||||
|
|
||||||
| Файл | Что внутри | Строк |
|
|
||||||
| --- | --- | --- |
|
|
||||||
| `domain/task.go` | `Task` и **все** переходы состояний: аренда, завершение, провал, истечение | ~245 |
|
|
||||||
| `domain/job.go` | `Job`, разбиение на чанки, вывод статуса из счётчиков задач | ~107 |
|
|
||||||
| `domain/errors.go` | Нарушения бизнес-правил (`ErrLeaseConflict`, `ErrStaleAttempt`, …) | ~18 |
|
|
||||||
| `usecase/ports.go` | **Порты**: `TaskRepository`, `JobRepository`, `TxManager`, `Clock` | ~79 |
|
|
||||||
| `usecase/task.go` | Операции над задачей: claim, renew, complete, fail, expire | ~200 |
|
|
||||||
| `usecase/job.go` | Операции над job: create, status, results, stitch | ~180 |
|
|
||||||
| `usecase/dto.go` | Входные структуры юзкейсов | ~51 |
|
|
||||||
| `transport/http/server.go` | Роутер и сборка middleware | ~60 |
|
|
||||||
| `transport/http/handlers.go` | По хендлеру на эндпоинт | ~180 |
|
|
||||||
| `transport/http/dto.go` | JSON-форматы запросов и ответов | ~118 |
|
|
||||||
| `transport/http/middleware.go` | request-ID, access-лог, bearer-авторизация | ~103 |
|
|
||||||
| `transport/http/errors.go` | Маппинг доменных ошибок в HTTP-коды | ~55 |
|
|
||||||
| `storage/postgres/task_repo.go` | SQL по задачам, включая атомарный claim | ~109 |
|
|
||||||
| `storage/postgres/job_repo.go` | SQL по job'ам | ~39 |
|
|
||||||
| `storage/postgres/tx.go` | `TxManager`: транзакция через контекст | ~65 |
|
|
||||||
| `infra/*.go` | Конфиг, пул, часы, сервер, reaper | ~240 |
|
|
||||||
| `cmd/coordinator/main.go` | **Composition root** — единственное место со всеми конкретными типами | ~73 |
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 3. Трасса запроса: `POST /tasks/claim`
|
|
||||||
|
|
||||||
Как воркер получает задачу. Четыре остановки, по одной на слой:
|
|
||||||
|
|
||||||
```
|
|
||||||
① transport/http/handlers.go → handleClaim
|
|
||||||
разбирает JSON, отдаёт usecase.ClaimTaskInput
|
|
||||||
│
|
|
||||||
▼
|
|
||||||
② usecase/task.go → ClaimTask.Execute
|
|
||||||
сначала подчищает протухшие аренды, потом просит одну задачу
|
|
||||||
через ПОРТ TaskRepository (реализацию не знает)
|
|
||||||
│
|
|
||||||
▼
|
|
||||||
③ usecase/ports.go → TaskRepository.ClaimNext
|
|
||||||
контракт: «атомарно выдай одну задачу»
|
|
||||||
│
|
|
||||||
▼
|
|
||||||
④ storage/postgres/task_repo.go → claimNextSQL
|
|
||||||
SELECT ... FOR UPDATE SKIP LOCKED + UPDATE одним запросом
|
|
||||||
```
|
|
||||||
|
|
||||||
Обратно поднимается `*domain.Task`, юзкейс сужает его до `domain.ClaimedTask`
|
|
||||||
(воркеру не отдаём `version`, `max_attempts` и чужие ошибки), хендлер
|
|
||||||
превращает в JSON. Пустая очередь — это `nil, nil` на шаге ② и `204` на ①.
|
|
||||||
|
|
||||||
**Трасса `POST /tasks/{id}/result`** такая же, но с одним отличием: решение
|
|
||||||
принимает **сущность**, а не юзкейс.
|
|
||||||
|
|
||||||
```
|
|
||||||
handlers.go → CompleteTask.Execute → tx.WithinTx(
|
|
||||||
GetForUpdate → task.CompleteWith(...) ←── ЗДЕСЬ правила
|
|
||||||
│ (чужая аренда? устаревший
|
|
||||||
Update ←─────────────┘ attempt? повтор того же
|
|
||||||
syncJobStatus манифеста?)
|
|
||||||
)
|
|
||||||
```
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 4. Куда добавлять новое
|
|
||||||
|
|
||||||
| Хочу… | Правлю |
|
|
||||||
| --- | --- |
|
|
||||||
| новое бизнес-правило (когда задачу можно повторить) | `domain/task.go` + тест рядом |
|
|
||||||
| новую операцию (отменить job) | `usecase/job.go` + порт в `ports.go`, если нужен новый запрос к БД |
|
|
||||||
| новый HTTP-эндпоинт | `transport/http/handlers.go` + маршрут в `server.go` + DTO в `dto.go` |
|
|
||||||
| новый SQL-запрос | `storage/postgres/*_repo.go` |
|
|
||||||
| новую настройку | `infra/config.go` + `.env.example` |
|
|
||||||
| поменять код ответа на ошибку | `transport/http/errors.go` |
|
|
||||||
|
|
||||||
**Правило при сомнении:** если код можно описать фразой «когда X, то Y» без
|
|
||||||
упоминания HTTP, SQL и конфигов — это `domain`. Если он оркеструет несколько
|
|
||||||
шагов и транзакцию — `usecase`. Если знает про JSON — `transport`, про SQL — `storage`.
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 5. Три вещи, которые надо понять один раз
|
|
||||||
|
|
||||||
**Порты объявляет потребитель.** `TaskRepository` описан в `usecase/ports.go`, а
|
|
||||||
реализован в `storage/postgres`. Поэтому `usecase` не импортирует `storage` —
|
|
||||||
стрелка зависимости смотрит внутрь, хотя вызов на рантайме идёт наружу.
|
|
||||||
|
|
||||||
**Транзакция едет в контексте.** `TxManager.WithinTx` кладёт `pgx.Tx` в контекст
|
|
||||||
по неэкспортируемому ключу; репозитории достают её через `conn(ctx, pool)`.
|
|
||||||
Благодаря этому юзкейс говорит «сделай это атомарно», ни разу не упомянув pgx.
|
|
||||||
|
|
||||||
**Атомарный claim нельзя разложить на шаги.** `ClaimNext` — один SQL-запрос,
|
|
||||||
потому что `SELECT` + отдельный `UPDATE` вернул бы гонку, при которой одну
|
|
||||||
задачу выдают двум воркерам. Поэтому `ClaimTask.Execute` выглядит тонким: там
|
|
||||||
нечего оркестровать, вся гарантия — внутри запроса.
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 6. Что уже работает, а что заглушка
|
|
||||||
|
|
||||||
Работает: слои и проводка, роутинг, авторизация, access-лог, маппинг ошибок,
|
|
||||||
транзакции, graceful shutdown, миграции, **весь domain с 12 юнит-тестами без БД**.
|
|
||||||
|
|
||||||
Заглушки (`ErrNotImplemented` → HTTP 501): методы репозиториев. SQL для двух
|
|
||||||
главных операций уже написан в `task_repo.go` — `claimNextSQL` и
|
|
||||||
`expireLeasesSQL`, осталось их подключить.
|
|
||||||
+46
-211
@@ -1,230 +1,65 @@
|
|||||||
# SciMesh Coordinator
|
# SciMesh userservice
|
||||||
|
|
||||||
Durable task-queue server for SciMesh, in Go on PostgreSQL. It owns all database
|
Authentication service for SciMesh, in Go on PostgreSQL. It owns user accounts
|
||||||
access; workers talk to it only over HTTP and never receive DB credentials.
|
and issues the JWTs the coordinator trusts. It is a **separate bounded context**
|
||||||
|
from the coordinator: its own database, its own binary. The only thing shared
|
||||||
|
between the two services is the JWT signing secret.
|
||||||
|
|
||||||
Built as a **modular monolith following Clean Architecture** — one binary, four
|
Built as a modular monolith following Clean Architecture — one binary, four
|
||||||
layers, dependencies pointing strictly inward. See
|
layers, dependencies pointing strictly inward:
|
||||||
`docs/database-integration-task.md` and `docs/worker-daemon-task.md` in the repo
|
|
||||||
root for the full contract.
|
|
||||||
|
|
||||||
## Layers
|
|
||||||
|
|
||||||
```
|
```
|
||||||
infra config, pgxpool, http.Server, clock ← frameworks & drivers
|
infra config, DB pool, clock, HTTP server ← drivers
|
||||||
transport http handlers ← inbound: who calls us
|
transport HTTP handlers + JWT middleware ← incoming
|
||||||
storage sql repositories ← outbound: who we call
|
storage SQL repository ← outgoing
|
||||||
usecase business operations + PORTS ← application rules
|
usecase Register / Login + PORTS (interfaces) ← application rules
|
||||||
domain Task, Job + their invariants ← enterprise rules
|
domain User, Role, invariants ← business rules
|
||||||
|
auth bcrypt hasher, HS256 JWT issuer ← crypto adapters
|
||||||
┌── transport ──┐
|
|
||||||
domain ◄── usecase ◄┤ ├◄── infra
|
|
||||||
└── storage ────┘
|
|
||||||
```
|
```
|
||||||
|
|
||||||
`transport` and `storage` are one layer — the "interface adapters" ring — split
|
|
||||||
by direction rather than by category, so a file's path tells you its role.
|
|
||||||
|
|
||||||
The rule that matters: **source dependencies point only inward**. `domain`
|
|
||||||
imports nothing from this module; `usecase` sees only `domain`; `transport` and
|
|
||||||
`storage` know nothing of each other. Verify it at any time with:
|
|
||||||
|
|
||||||
```sh
|
|
||||||
go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal # must be empty
|
|
||||||
```
|
|
||||||
|
|
||||||
## Layout
|
|
||||||
|
|
||||||
```
|
|
||||||
coordinator/
|
|
||||||
cmd/coordinator/main.go # composition root: the only place with concrete types
|
|
||||||
internal/
|
|
||||||
domain/ # entities + rules, no I/O
|
|
||||||
task.go Task, lease/complete/fail/expire transitions
|
|
||||||
job.go Job, chunk fan-out, status derivation
|
|
||||||
errors.go business-rule violations
|
|
||||||
usecase/ # one type per operation, dependencies injected
|
|
||||||
ports.go TaskRepository, JobRepository, TxManager, Clock
|
|
||||||
dto.go use-case boundary inputs
|
|
||||||
task.go claim, renew, complete, fail, expire
|
|
||||||
job.go create, status, results, stitch
|
|
||||||
transport/http/ # routing, DTOs, middleware, error mapping
|
|
||||||
storage/postgres/ # SQL behind the ports; TxManager via context
|
|
||||||
infra/ # config.go db.go clock.go server.go
|
|
||||||
migrations/ # golang-migrate SQL, run as an explicit command
|
|
||||||
```
|
|
||||||
|
|
||||||
A full map — file-by-file table, a request traced through every layer, and a
|
|
||||||
"where do I add X" guide — lives in [ARCHITECTURE.md](ARCHITECTURE.md).
|
|
||||||
|
|
||||||
## Quickstart
|
|
||||||
|
|
||||||
### With Docker (nothing to install but Docker)
|
|
||||||
|
|
||||||
```sh
|
|
||||||
make up # Postgres → migrations → coordinator
|
|
||||||
curl localhost:8080/health
|
|
||||||
make logs # follow the coordinator
|
|
||||||
make down # stop (add down-clean to drop the DB volume)
|
|
||||||
```
|
|
||||||
|
|
||||||
To enable the local operator UI, set a separate credential before starting:
|
|
||||||
|
|
||||||
```sh
|
|
||||||
UI_AUTH_TOKEN='local-ui-secret' make up
|
|
||||||
# Open http://localhost:8080/ui and use any username with this value as password.
|
|
||||||
```
|
|
||||||
|
|
||||||
The UI is disabled by default and never accepts the worker bearer token.
|
|
||||||
The **control room** shows live workers, recent runs, shard state/attempts,
|
|
||||||
safe failures, coordinator artifacts, and the final CSV for completed
|
|
||||||
similarity-search jobs. The job page follows the real stages: TSV accepted →
|
|
||||||
shards execute → workers return CSVs → `reducing` → final deterministic global
|
|
||||||
top-k result. It polls only its own coordinator read-model and never controls
|
|
||||||
or exposes worker processes.
|
|
||||||
|
|
||||||
For a hands-on run, open `/ui`, choose **New similarity search**, select a
|
|
||||||
small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes
|
|
||||||
running in separate terminals. The detail page updates every two seconds and
|
|
||||||
stops polling after a completed, failed, or cancelled job. Use **Preview CSV**
|
|
||||||
to inspect a bounded first page of a partial or completed final result before
|
|
||||||
downloading it. The UI never exposes source datasets or shard inputs; partial
|
|
||||||
CSVs remain available only as diagnostics.
|
|
||||||
|
|
||||||
### One-command manual demo
|
|
||||||
|
|
||||||
From the repository root, create the Python environment once, then start a
|
|
||||||
self-contained UI demo with two local reference workers:
|
|
||||||
|
|
||||||
```sh
|
|
||||||
python3 -m venv .venv
|
|
||||||
.venv/bin/pip install -e '.[dev]'
|
|
||||||
make demo-ui
|
|
||||||
```
|
|
||||||
|
|
||||||
This uses a separate Docker project and ports `18080` (coordinator) and
|
|
||||||
`55432` (PostgreSQL), so it does not conflict with the normal stack. Open
|
|
||||||
`http://localhost:18080/ui`, use username `operator` and password
|
|
||||||
`demo-ui-secret`, upload a small ChEMBL TSV, and observe the workers process
|
|
||||||
it. Change the worker count with `make demo-ui WORKERS=3`; stop all demo
|
|
||||||
services and workers with `make demo-down`.
|
|
||||||
|
|
||||||
The job page shows a live **Processing speed** graph in completed shards per
|
|
||||||
minute. It uses the coordinator snapshots observed by the open browser tab, so
|
|
||||||
it is a transparent local-session measurement rather than a persisted metric.
|
|
||||||
Use **Preview CSV** before downloading a partial diagnostic or completed final
|
|
||||||
result. Run `make help` from either the repository root or this directory for
|
|
||||||
the full list of demo commands.
|
|
||||||
|
|
||||||
`up` starts three services in order: Postgres waits until `pg_isready` passes, a
|
|
||||||
one-shot `migrate` container applies the schema and exits, and only then does the
|
|
||||||
coordinator start — so it never queries a database that has no tables.
|
|
||||||
|
|
||||||
> **Needs BuildKit.** The Dockerfile uses `RUN --mount=type=cache` to reuse the
|
|
||||||
> Go module and compiler caches between builds. If the build fails with
|
|
||||||
> *"the --mount option requires BuildKit"*, install the buildx plugin —
|
|
||||||
> `pacman -S docker-buildx` on Arch, `apt install docker-buildx-plugin` on Debian.
|
|
||||||
|
|
||||||
### Locally, against your own Postgres
|
|
||||||
|
|
||||||
```sh
|
|
||||||
cp .env.example .env # then edit DATABASE_URL / WORKER_AUTH_TOKEN
|
|
||||||
# it is loaded automatically — no export needed
|
|
||||||
|
|
||||||
make tidy # fetch deps (needs network once)
|
|
||||||
make migrate-up # apply schema (needs the migrate CLI)
|
|
||||||
make run # start the server
|
|
||||||
```
|
|
||||||
|
|
||||||
## Configuration
|
|
||||||
|
|
||||||
Settings come from the environment. A `.env` file is loaded at startup via
|
|
||||||
`godotenv` as a local-dev convenience (override its path with `ENV_FILE`):
|
|
||||||
|
|
||||||
- a missing `.env` is not an error — production injects real env vars;
|
|
||||||
- **real environment variables always win** over the file, so an orchestrator's
|
|
||||||
values are never shadowed by a stale `.env` baked into an image.
|
|
||||||
|
|
||||||
See `.env.example`; only `DATABASE_URL` is required.
|
|
||||||
|
|
||||||
## Endpoints
|
## Endpoints
|
||||||
|
|
||||||
| Method | Path | Purpose |
|
| Method | Path | Auth | Purpose |
|
||||||
| ------ | ---------------------------------- | --------------------------------------------- |
|
|--------|-------------|-------------|------------------------------------------|
|
||||||
| POST | `/workers/register` | Register a worker, get its id |
|
| GET | `/health` | none | Liveness probe (checks the database) |
|
||||||
| POST | `/jobs` | Create job + tasks from chunk URIs |
|
| POST | `/register` | none | Create an account (always role `user`) |
|
||||||
| POST | `/jobs/upload` | Upload a dataset; coordinator chunks it |
|
| POST | `/login` | none | Verify credentials, return a signed JWT |
|
||||||
| GET | `/jobs/{job_id}` | Aggregate job progress |
|
| GET | `/me` | Bearer JWT | Return the caller's own account |
|
||||||
| 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
|
Roles are `user` and `admin`. Registration always creates a `user`; promotion to
|
||||||
[`docs/openapi.yaml`](../docs/openapi.yaml); a worker-author guide is in
|
`admin` is a manual database operation, never a request. The role→permission
|
||||||
[`docs/building-workers.md`](../docs/building-workers.md).
|
mapping lives in the coordinator's authorization checks, not in a table.
|
||||||
|
|
||||||
## Poking the API
|
## How it connects to the coordinator
|
||||||
|
|
||||||
Two ways, both checked in:
|
The coordinator never calls this service at runtime. A client logs in here, gets
|
||||||
|
a JWT, and presents it to the coordinator, which verifies the signature locally
|
||||||
|
with the same `JWT_SECRET` and reads `sub` (the user id) into `jobs.owner_id`.
|
||||||
|
|
||||||
|
That link is **off by default**: until the coordinator is given a matching
|
||||||
|
`JWT_SECRET`, it accepts only the shared worker token and stores `owner_id` as
|
||||||
|
NULL. Set the same secret (≥ 32 bytes, byte-for-byte identical) on both services
|
||||||
|
to turn it on.
|
||||||
|
|
||||||
|
## Run
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
make smoke # every endpoint, asserted; non-zero exit on failure
|
# whole stack: Postgres + migrations + the service on :8081
|
||||||
|
make up
|
||||||
|
|
||||||
|
# or locally against your own Postgres
|
||||||
|
cp .env.example .env # then edit JWT_SECRET and DATABASE_URL
|
||||||
|
make run
|
||||||
```
|
```
|
||||||
|
|
||||||
`api/requests.http` runs the same calls one at a time from an editor with a REST
|
## Verify
|
||||||
client (VSCodium/VS Code "REST Client", JetBrains HTTP Client). Later requests
|
|
||||||
reuse ids captured from earlier responses, so it doubles as API documentation.
|
|
||||||
|
|
||||||
## Status
|
|
||||||
|
|
||||||
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.
|
|
||||||
|
|
||||||
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.
|
|
||||||
|
|
||||||
Still stubbed: `StitchJob.Execute` — merging per-chunk top-k into the final CSV
|
|
||||||
is workload semantics that belongs to the Python side (reducer).
|
|
||||||
|
|
||||||
## Tests
|
|
||||||
|
|
||||||
Unit tests need **no database** — domain rules, use-case orchestration (over
|
|
||||||
in-memory `internal/memstore`), and HTTP handlers (via `httptest`):
|
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
make test # go test ./...
|
make test # unit tests
|
||||||
make vet
|
make check # vet, lint, race, integration, smoke — needs Docker
|
||||||
make lint
|
make smoke # end-to-end against a running service
|
||||||
go test -race ./...
|
|
||||||
```
|
```
|
||||||
|
|
||||||
Integration tests run against a **real PostgreSQL** (the spec forbids mocks
|
Password hashing uses bcrypt (`golang.org/x/crypto/bcrypt`); the salt and cost
|
||||||
here — they verify `FOR UPDATE SKIP LOCKED`, optimistic concurrency, rollback):
|
are embedded in the stored hash, so there is no separate salt column. Tokens are
|
||||||
|
HS256 (`github.com/golang-jwt/jwt/v5`).
|
||||||
```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.
|
|
||||||
|
|
||||||
For the complete local verification, including an isolated Docker PostgreSQL
|
|
||||||
and the HTTP smoke flow, run:
|
|
||||||
|
|
||||||
```sh
|
|
||||||
make check
|
|
||||||
```
|
|
||||||
|
|
||||||
It uses Compose project `scimesh-check` and ports `55432`/`18080` by default,
|
|
||||||
so it does not connect to a PostgreSQL already running on `5432`. Override
|
|
||||||
`CHECK_POSTGRES_PORT`, `CHECK_COORDINATOR_PORT`, or `CHECK_PROJECT` if needed.
|
|
||||||
|
|||||||
@@ -1,234 +0,0 @@
|
|||||||
# SciMesh Coordinator — API requests
|
|
||||||
#
|
|
||||||
# Runnable from any editor with a REST client (VSCodium/VS Code "REST Client",
|
|
||||||
# JetBrains HTTP Client). Click "Send Request" above each block, top to bottom:
|
|
||||||
# later requests reuse ids captured from earlier responses.
|
|
||||||
#
|
|
||||||
# Start the stack first: docker compose up -d
|
|
||||||
|
|
||||||
@host = http://localhost:8080
|
|
||||||
@token = change-me
|
|
||||||
@worker = worker-1
|
|
||||||
|
|
||||||
### Readiness — the only unauthenticated endpoint (probes the database)
|
|
||||||
GET {{host}}/health
|
|
||||||
|
|
||||||
### Auth check — no token must be rejected with 401
|
|
||||||
POST {{host}}/tasks/claim
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{ "worker_id": "{{worker}}" }
|
|
||||||
|
|
||||||
### 0. Register a worker (201)
|
|
||||||
# @name register
|
|
||||||
POST {{host}}/workers/register
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"name": "lab-worker-01",
|
|
||||||
"capabilities": ["similarity_search"],
|
|
||||||
"cpu_count": 8,
|
|
||||||
"memory_mb": 16384
|
|
||||||
}
|
|
||||||
|
|
||||||
@workerId = {{register.response.body.worker_id}}
|
|
||||||
|
|
||||||
### 0b. Upload a dataset — the coordinator splits it into shard tasks (201)
|
|
||||||
# Text fields first, the file part last (it is streamed, not buffered).
|
|
||||||
# @name uploadJob
|
|
||||||
POST {{host}}/jobs/upload
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: multipart/form-data; boundary=----scimesh
|
|
||||||
|
|
||||||
------scimesh
|
|
||||||
Content-Disposition: form-data; name="workload"
|
|
||||||
|
|
||||||
similarity_search
|
|
||||||
------scimesh
|
|
||||||
Content-Disposition: form-data; name="parameters"
|
|
||||||
|
|
||||||
{"top_k":10}
|
|
||||||
------scimesh
|
|
||||||
Content-Disposition: form-data; name="chunk_rows"
|
|
||||||
|
|
||||||
2
|
|
||||||
------scimesh
|
|
||||||
Content-Disposition: form-data; name="file"; filename="chembl.tsv"
|
|
||||||
Content-Type: text/tab-separated-values
|
|
||||||
|
|
||||||
id smiles
|
|
||||||
A CC
|
|
||||||
B CCC
|
|
||||||
C CCCC
|
|
||||||
D CCCCC
|
|
||||||
------scimesh--
|
|
||||||
|
|
||||||
### Download a task's input shard (200) — taskId must be a shard task from an
|
|
||||||
### uploaded job (claim one first; its input.uri is /tasks/{id}/input).
|
|
||||||
GET {{host}}/tasks/{{taskId}}/input
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
|
|
||||||
### 1. Create a job and its chunks (201)
|
|
||||||
# The coordinator splits the submission into one task per chunk, transactionally.
|
|
||||||
# @name createJob
|
|
||||||
POST {{host}}/jobs
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"workload": "similarity_search",
|
|
||||||
"input_uri": "s3://chembl/full.sdf",
|
|
||||||
"parameters": { "top_k": 10 },
|
|
||||||
"chunks": [
|
|
||||||
{ "chunk_index": 0, "input_uri": "s3://chembl/shard-0.sdf", "input_sha256": "aaa", "max_attempts": 3 },
|
|
||||||
{ "chunk_index": 1, "input_uri": "s3://chembl/shard-1.sdf", "input_sha256": "bbb", "max_attempts": 3 },
|
|
||||||
{ "chunk_index": 2, "input_uri": "s3://chembl/shard-2.sdf", "input_sha256": "ccc", "max_attempts": 3 }
|
|
||||||
]
|
|
||||||
}
|
|
||||||
|
|
||||||
@jobId = {{createJob.response.body.id}}
|
|
||||||
|
|
||||||
### 2. Claim a task (200, or 204 when the queue is empty)
|
|
||||||
# Each call leases a different task; run it repeatedly to see chunk_index advance.
|
|
||||||
# @name claim
|
|
||||||
POST {{host}}/tasks/claim
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"capabilities": ["similarity_search"],
|
|
||||||
"max_concurrency": 1
|
|
||||||
}
|
|
||||||
|
|
||||||
@taskId = {{claim.response.body.task_id}}
|
|
||||||
@attempt = {{claim.response.body.attempt}}
|
|
||||||
|
|
||||||
### 3. Heartbeat — renew the lease while the task is still running (200)
|
|
||||||
POST {{host}}/tasks/{{taskId}}/heartbeat
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"attempt": {{attempt}}
|
|
||||||
}
|
|
||||||
|
|
||||||
### 3a. Upload a partial-result artifact (200) — while the task is leased
|
|
||||||
# Identity travels in headers per the contract; the body is streamed as-is.
|
|
||||||
# @name uploadArtifact
|
|
||||||
PUT {{host}}/tasks/{{taskId}}/artifacts/result.csv
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: text/csv
|
|
||||||
X-Worker-ID: {{worker}}
|
|
||||||
X-Task-Attempt: {{attempt}}
|
|
||||||
|
|
||||||
query,match,score
|
|
||||||
CHEMBL25,CHEMBL139,0.87
|
|
||||||
|
|
||||||
@artifactId = {{uploadArtifact.response.body.artifact_id}}
|
|
||||||
|
|
||||||
### 3b. Download the artifact by id (200)
|
|
||||||
GET {{host}}/artifacts/{{artifactId}}/download
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
|
|
||||||
### 3c. Upload a second artifact — used by the conflict check below (200)
|
|
||||||
# @name uploadArtifact2
|
|
||||||
PUT {{host}}/tasks/{{taskId}}/artifacts/secondary.csv
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: text/csv
|
|
||||||
X-Worker-ID: {{worker}}
|
|
||||||
X-Task-Attempt: {{attempt}}
|
|
||||||
|
|
||||||
query,match,score
|
|
||||||
CHEMBL25,CHEMBL521,0.42
|
|
||||||
|
|
||||||
@artifactId2 = {{uploadArtifact2.response.body.artifact_id}}
|
|
||||||
|
|
||||||
### 4. Submit the result, referencing the uploaded artifact (200)
|
|
||||||
POST {{host}}/tasks/{{taskId}}/result
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"attempt": {{attempt}},
|
|
||||||
"result": { "artifact_id": "{{artifactId}}", "content_type": "text/csv" },
|
|
||||||
"metrics": { "elapsed_ms": 1234, "candidates": 50000 }
|
|
||||||
}
|
|
||||||
|
|
||||||
### 4a. Replay the same result — must be idempotent (200, not 409)
|
|
||||||
POST {{host}}/tasks/{{taskId}}/result
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"attempt": {{attempt}},
|
|
||||||
"result": { "artifact_id": "{{artifactId}}" }
|
|
||||||
}
|
|
||||||
|
|
||||||
### 4b. A different artifact for the same task — conflict (409)
|
|
||||||
POST {{host}}/tasks/{{taskId}}/result
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"attempt": {{attempt}},
|
|
||||||
"result": { "artifact_id": "{{artifactId2}}" }
|
|
||||||
}
|
|
||||||
|
|
||||||
### 4c. Another worker submitting for this task — conflict (409)
|
|
||||||
POST {{host}}/tasks/{{taskId}}/result
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "impostor",
|
|
||||||
"attempt": {{attempt}},
|
|
||||||
"result": { "artifact_id": "{{artifactId}}" }
|
|
||||||
}
|
|
||||||
|
|
||||||
### 5. Report a failure instead (200)
|
|
||||||
# retryable=true returns the task to the queue while attempts remain;
|
|
||||||
# retryable=false fails it terminally.
|
|
||||||
POST {{host}}/tasks/{{taskId}}/failure
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{
|
|
||||||
"worker_id": "{{worker}}",
|
|
||||||
"attempt": {{attempt}},
|
|
||||||
"error_code": "download_failed",
|
|
||||||
"error_message": "checksum mismatch on shard",
|
|
||||||
"retryable": true
|
|
||||||
}
|
|
||||||
|
|
||||||
### 6. Job progress (200)
|
|
||||||
GET {{host}}/jobs/{{jobId}}
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
|
|
||||||
### --- error cases -------------------------------------------------------
|
|
||||||
|
|
||||||
### Malformed UUID in the path (400)
|
|
||||||
POST {{host}}/tasks/not-a-uuid/result
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{ "worker_id": "{{worker}}", "attempt": 1, "result_uri": "s3://x", "result_sha256": "x" }
|
|
||||||
|
|
||||||
### Unknown field in the body (400) — a misspelled key must not pass silently
|
|
||||||
POST {{host}}/tasks/claim
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
Content-Type: application/json
|
|
||||||
|
|
||||||
{ "worker_ID": "{{worker}}" }
|
|
||||||
|
|
||||||
### Unknown job (404)
|
|
||||||
GET {{host}}/jobs/00000000-0000-0000-0000-000000000000
|
|
||||||
Authorization: Bearer {{token}}
|
|
||||||
|
|
||||||
### Stitching is not implemented yet (501)
|
|
||||||
# Any endpoint whose use case is still a stub answers 501.
|
|
||||||
@@ -8,6 +8,6 @@ type System struct{}
|
|||||||
|
|
||||||
func NewClock() System { return System{} }
|
func NewClock() System { return System{} }
|
||||||
|
|
||||||
// Now returns UTC so every timestamp the coordinator writes is comparable
|
// Now returns UTC so every timestamp this service writes is comparable
|
||||||
// regardless of the host's timezone.
|
// regardless of the host's timezone.
|
||||||
func (System) Now() time.Time { return time.Now().UTC() }
|
func (System) Now() time.Time { return time.Now().UTC() }
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ func NewPool(ctx context.Context, cfg Config, log *slog.Logger) (*pgxpool.Pool,
|
|||||||
}
|
}
|
||||||
// pgxpool.New is lazy, so a ping is needed to actually reach the server.
|
// pgxpool.New is lazy, so a ping is needed to actually reach the server.
|
||||||
// It is retried because at startup — especially under docker-compose, where
|
// It is retried because at startup — especially under docker-compose, where
|
||||||
// the coordinator can boot before Postgres is accepting connections — a
|
// this service can boot before Postgres is accepting connections — a
|
||||||
// service should wait for its database rather than crash-loop.
|
// service should wait for its database rather than crash-loop.
|
||||||
if err := pingWithRetry(ctx, pool, cfg.DBConnectTimeout, log); err != nil {
|
if err := pingWithRetry(ctx, pool, cfg.DBConnectTimeout, log); err != nil {
|
||||||
pool.Close()
|
pool.Close()
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
// Server: the HTTP listener and the background lease reaper, both shut down
|
// Server: the HTTP listener, shut down cleanly on a signal.
|
||||||
// cleanly on a signal.
|
|
||||||
package infra
|
package infra
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -12,7 +11,7 @@ import (
|
|||||||
|
|
||||||
const shutdownGrace = 15 * time.Second
|
const shutdownGrace = 15 * time.Second
|
||||||
|
|
||||||
// Run serves handler until ctx is cancelled, then drains in-flight requests.
|
// RunServer serves handler until ctx is cancelled, then drains in-flight requests.
|
||||||
func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler) error {
|
func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler) error {
|
||||||
srv := &http.Server{
|
srv := &http.Server{
|
||||||
Addr: addr,
|
Addr: addr,
|
||||||
@@ -43,32 +42,3 @@ func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.
|
|||||||
defer cancel()
|
defer cancel()
|
||||||
return srv.Shutdown(shutdownCtx)
|
return srv.Shutdown(shutdownCtx)
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunReaper periodically reclaims tasks whose lease elapsed, so a worker that
|
|
||||||
// died without a heartbeat cannot strand its task in 'leased' forever.
|
|
||||||
// RunPeriodic invokes fn on an interval until ctx is done, logging how many rows
|
|
||||||
// each tick affected. It backs the background reapers (expired leases, offline
|
|
||||||
// workers) — each is a set-based UPDATE that is safe to run repeatedly and
|
|
||||||
// concurrently across coordinators.
|
|
||||||
func RunPeriodic(ctx context.Context, log *slog.Logger, name string, interval time.Duration,
|
|
||||||
fn func(context.Context) (int64, error)) {
|
|
||||||
|
|
||||||
t := time.NewTicker(interval)
|
|
||||||
defer t.Stop()
|
|
||||||
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
|
||||||
return
|
|
||||||
case <-t.C:
|
|
||||||
n, err := fn(ctx)
|
|
||||||
if err != nil {
|
|
||||||
log.Debug(name+" skipped", "err", err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if n > 0 {
|
|
||||||
log.Info(name, "count", n)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -9,8 +9,8 @@ import (
|
|||||||
"github.com/jackc/pgx/v5/pgconn"
|
"github.com/jackc/pgx/v5/pgconn"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Transient PostgreSQL failures. Under concurrent claiming these are expected
|
// Transient PostgreSQL failures. Under concurrent writes these are expected
|
||||||
// rather than exceptional: two coordinators touching neighbouring rows can
|
// rather than exceptional: two service instances touching neighbouring rows can
|
||||||
// deadlock or fail to serialize, and the correct response is to try again.
|
// deadlock or fail to serialize, and the correct response is to try again.
|
||||||
const (
|
const (
|
||||||
codeSerializationFailure = "40001"
|
codeSerializationFailure = "40001"
|
||||||
@@ -60,7 +60,7 @@ func isTransient(err error) bool {
|
|||||||
// withRetry runs op, retrying only transient database failures with
|
// withRetry runs op, retrying only transient database failures with
|
||||||
// exponential backoff and jitter, and giving up as soon as ctx is done.
|
// exponential backoff and jitter, and giving up as soon as ctx is done.
|
||||||
//
|
//
|
||||||
// Jitter matters here: without it, several coordinators that collide once will
|
// Jitter matters here: without it, several instances that collide once will
|
||||||
// retry in lockstep and collide again at exactly the same moment.
|
// retry in lockstep and collide again at exactly the same moment.
|
||||||
func withRetry(ctx context.Context, op func(context.Context) error) error {
|
func withRetry(ctx context.Context, op func(context.Context) error) error {
|
||||||
b := backoff.NewExponentialBackOff()
|
b := backoff.NewExponentialBackOff()
|
||||||
|
|||||||
Reference in New Issue
Block a user