diff --git a/users/.dockerignore b/users/.dockerignore index a4d0d5c..545cfcf 100644 --- a/users/.dockerignore +++ b/users/.dockerignore @@ -9,6 +9,8 @@ Dockerfile .dockerignore # Local build artifacts -/coordinator +/userservice /bin/ *.out +/data/ +/logs/ diff --git a/users/.env.example b/users/.env.example index 4c1e596..53b8ecf 100644 --- a/users/.env.example +++ b/users/.env.example @@ -1,32 +1,23 @@ # Copy to .env and adjust. All settings are read from the environment. -COORDINATOR_ADDR=:8080 -DATABASE_URL=postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable +USERSERVICE_ADDR=:8081 +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). -WORKER_AUTH_TOKEN=change-me - -# Optional local operator UI. Use a separate value; never reuse the worker token. -# When empty, /ui is disabled. -UI_AUTH_TOKEN= +# Shared HS256 secret used to sign JWTs. The coordinator verifies tokens with +# 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 +# How long an issued token stays valid. +JWT_TTL=24h +# 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; # set a path to also write a size-rotated file (kept across restarts). LOG_LEVEL=info -# LOG_FILE=./logs/coordinator.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 +# LOG_FILE=./logs/userservice.log # Optional tuning (defaults shown). DB_MAX_CONNS=10 # How long to keep retrying the initial DB connection while Postgres boots. DB_CONNECT_TIMEOUT=30s 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 diff --git a/users/.gitignore b/users/.gitignore index c8c4f7b..8ef0865 100644 --- a/users/.gitignore +++ b/users/.gitignore @@ -1,4 +1,4 @@ -/coordinator +/userservice /bin/ .env *.out diff --git a/users/.golangci.yml b/users/.golangci.yml index 428bfed..3492657 100644 --- a/users/.golangci.yml +++ b/users/.golangci.yml @@ -51,4 +51,4 @@ formatters: settings: goimports: local-prefixes: - - github.com/emil28092005/SciMesh/coordinator + - github.com/emil28092005/SciMesh/users diff --git a/users/ARCHITECTURE.md b/users/ARCHITECTURE.md deleted file mode 100644 index 77b6976..0000000 --- a/users/ARCHITECTURE.md +++ /dev/null @@ -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`, осталось их подключить. diff --git a/users/README.md b/users/README.md index 7240b99..63b6587 100644 --- a/users/README.md +++ b/users/README.md @@ -1,230 +1,65 @@ -# SciMesh Coordinator +# SciMesh userservice -Durable task-queue server for SciMesh, in Go on PostgreSQL. It owns all database -access; workers talk to it only over HTTP and never receive DB credentials. +Authentication service for SciMesh, in Go on PostgreSQL. It owns user accounts +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 -layers, dependencies pointing strictly inward. See -`docs/database-integration-task.md` and `docs/worker-daemon-task.md` in the repo -root for the full contract. - -## Layers +Built as a modular monolith following Clean Architecture — one binary, four +layers, dependencies pointing strictly inward: ``` - infra config, pgxpool, http.Server, clock ← frameworks & drivers - transport http handlers ← inbound: who calls us - storage sql repositories ← outbound: who we call - usecase business operations + PORTS ← application rules - domain Task, Job + their invariants ← enterprise rules - - ┌── transport ──┐ - domain ◄── usecase ◄┤ ├◄── infra - └── storage ────┘ + infra config, DB pool, clock, HTTP server ← drivers + transport HTTP handlers + JWT middleware ← incoming + storage SQL repository ← outgoing + usecase Register / Login + PORTS (interfaces) ← application rules + domain User, Role, invariants ← business rules + auth bcrypt hasher, HS256 JWT issuer ← crypto adapters ``` -`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 -| 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) | +| Method | Path | Auth | Purpose | +|--------|-------------|-------------|------------------------------------------| +| GET | `/health` | none | Liveness probe (checks the database) | +| POST | `/register` | none | Create an account (always role `user`) | +| POST | `/login` | none | Verify credentials, return a signed JWT | +| GET | `/me` | Bearer JWT | Return the caller's own account | -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). +Roles are `user` and `admin`. Registration always creates a `user`; promotion to +`admin` is a manual database operation, never a request. The role→permission +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 -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 -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`): +## Verify ```sh -make test # go test ./... -make vet -make lint -go test -race ./... +make test # unit tests +make check # vet, lint, race, integration, smoke — needs Docker +make smoke # end-to-end against a running service ``` -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. - -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. +Password hashing uses bcrypt (`golang.org/x/crypto/bcrypt`); the salt and cost +are embedded in the stored hash, so there is no separate salt column. Tokens are +HS256 (`github.com/golang-jwt/jwt/v5`). diff --git a/users/api/requests.http b/users/api/requests.http deleted file mode 100644 index 6c50a59..0000000 --- a/users/api/requests.http +++ /dev/null @@ -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. diff --git a/users/internal/infra/clock.go b/users/internal/infra/clock.go index eedbcde..e326bce 100644 --- a/users/internal/infra/clock.go +++ b/users/internal/infra/clock.go @@ -8,6 +8,6 @@ type System struct{} 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. func (System) Now() time.Time { return time.Now().UTC() } diff --git a/users/internal/infra/db.go b/users/internal/infra/db.go index d1135cd..4a09674 100644 --- a/users/internal/infra/db.go +++ b/users/internal/infra/db.go @@ -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. // 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. if err := pingWithRetry(ctx, pool, cfg.DBConnectTimeout, log); err != nil { pool.Close() diff --git a/users/internal/infra/server.go b/users/internal/infra/server.go index fc49a4c..80999e2 100644 --- a/users/internal/infra/server.go +++ b/users/internal/infra/server.go @@ -1,5 +1,4 @@ -// Server: the HTTP listener and the background lease reaper, both shut down -// cleanly on a signal. +// Server: the HTTP listener, shut down cleanly on a signal. package infra import ( @@ -12,7 +11,7 @@ import ( 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 { srv := &http.Server{ Addr: addr, @@ -43,32 +42,3 @@ func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http. defer cancel() 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) - } - } - } -} diff --git a/users/internal/storage/postgres/retry.go b/users/internal/storage/postgres/retry.go index 8f0ae86..3bd35ce 100644 --- a/users/internal/storage/postgres/retry.go +++ b/users/internal/storage/postgres/retry.go @@ -9,8 +9,8 @@ import ( "github.com/jackc/pgx/v5/pgconn" ) -// Transient PostgreSQL failures. Under concurrent claiming these are expected -// rather than exceptional: two coordinators touching neighbouring rows can +// Transient PostgreSQL failures. Under concurrent writes these are expected +// rather than exceptional: two service instances touching neighbouring rows can // deadlock or fail to serialize, and the correct response is to try again. const ( codeSerializationFailure = "40001" @@ -60,7 +60,7 @@ func isTransient(err error) bool { // withRetry runs op, retrying only transient database failures with // 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. func withRetry(ctx context.Context, op func(context.Context) error) error { b := backoff.NewExponentialBackOff()