From bda22666d72393d5029b4840f8916d352d2b5104 Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Wed, 22 Jul 2026 13:43:58 +0300 Subject: [PATCH 01/17] feat(coordinator): scaffold task-queue service in Go Adds the SciMesh coordinator: a durable task-queue server on PostgreSQL that owns all database access, with workers reaching it over HTTP only. Structured as a modular monolith following Clean Architecture: domain entities and their invariants, no I/O usecase business operations + repository/clock ports transport HTTP handlers, DTOs, auth, error mapping storage PostgreSQL repositories, transactions carried in context infra config, pool, clock, server, lease reaper Dependencies point strictly inward; domain imports nothing from the module. Working: layer wiring, routing, shared-token auth, access logging, request IDs, domain-error to status-code mapping, transactional boundaries, graceful shutdown (HTTP drain -> reaper stop -> pool close), migrations, and a Compose stack starting Postgres -> migrations -> coordinator. The domain is complete and covered by unit tests that need no database: lease ownership, stale attempts, idempotent result replay, retry budgets, and lease expiry. Repository methods are stubs returning ErrNotImplemented (HTTP 501). The SQL for atomic claiming (FOR UPDATE SKIP LOCKED) and for lease expiry is written and ready to wire up. See coordinator/ARCHITECTURE.md for the layer map and a request traced through every layer. --- coordinator/.dockerignore | 14 + coordinator/.env.example | 14 + coordinator/.gitignore | 4 + coordinator/.golangci.yml | 54 ++++ coordinator/ARCHITECTURE.md | 144 +++++++++++ coordinator/Dockerfile | 47 ++++ coordinator/Makefile | 60 +++++ coordinator/README.md | 143 ++++++++++ coordinator/cmd/coordinator/main.go | 98 +++++++ coordinator/docker-compose.yml | 67 +++++ coordinator/go.mod | 19 ++ coordinator/go.sum | 34 +++ coordinator/internal/domain/errors.go | 18 ++ coordinator/internal/domain/job.go | 107 ++++++++ coordinator/internal/domain/task.go | 244 ++++++++++++++++++ coordinator/internal/domain/task_test.go | 183 +++++++++++++ coordinator/internal/infra/clock.go | 13 + coordinator/internal/infra/config.go | 135 ++++++++++ coordinator/internal/infra/db.go | 30 +++ coordinator/internal/infra/doc.go | 7 + coordinator/internal/infra/server.go | 72 ++++++ .../internal/storage/postgres/job_repo.go | 40 +++ .../internal/storage/postgres/retry.go | 82 ++++++ .../internal/storage/postgres/retry_test.go | 98 +++++++ .../internal/storage/postgres/task_repo.go | 114 ++++++++ coordinator/internal/storage/postgres/tx.go | 69 +++++ coordinator/internal/transport/http/dto.go | 119 +++++++++ coordinator/internal/transport/http/errors.go | 55 ++++ .../internal/transport/http/handlers.go | 181 +++++++++++++ .../internal/transport/http/middleware.go | 103 ++++++++ coordinator/internal/transport/http/server.go | 60 +++++ coordinator/internal/usecase/dto.go | 51 ++++ coordinator/internal/usecase/job.go | 174 +++++++++++++ coordinator/internal/usecase/ports.go | 80 ++++++ coordinator/internal/usecase/task.go | 206 +++++++++++++++ coordinator/migrations/0001_init.down.sql | 8 + coordinator/migrations/0001_init.up.sql | 58 +++++ 37 files changed, 3005 insertions(+) create mode 100644 coordinator/.dockerignore create mode 100644 coordinator/.env.example create mode 100644 coordinator/.gitignore create mode 100644 coordinator/.golangci.yml create mode 100644 coordinator/ARCHITECTURE.md create mode 100644 coordinator/Dockerfile create mode 100644 coordinator/Makefile create mode 100644 coordinator/README.md create mode 100644 coordinator/cmd/coordinator/main.go create mode 100644 coordinator/docker-compose.yml create mode 100644 coordinator/go.mod create mode 100644 coordinator/go.sum create mode 100644 coordinator/internal/domain/errors.go create mode 100644 coordinator/internal/domain/job.go create mode 100644 coordinator/internal/domain/task.go create mode 100644 coordinator/internal/domain/task_test.go create mode 100644 coordinator/internal/infra/clock.go create mode 100644 coordinator/internal/infra/config.go create mode 100644 coordinator/internal/infra/db.go create mode 100644 coordinator/internal/infra/doc.go create mode 100644 coordinator/internal/infra/server.go create mode 100644 coordinator/internal/storage/postgres/job_repo.go create mode 100644 coordinator/internal/storage/postgres/retry.go create mode 100644 coordinator/internal/storage/postgres/retry_test.go create mode 100644 coordinator/internal/storage/postgres/task_repo.go create mode 100644 coordinator/internal/storage/postgres/tx.go create mode 100644 coordinator/internal/transport/http/dto.go create mode 100644 coordinator/internal/transport/http/errors.go create mode 100644 coordinator/internal/transport/http/handlers.go create mode 100644 coordinator/internal/transport/http/middleware.go create mode 100644 coordinator/internal/transport/http/server.go create mode 100644 coordinator/internal/usecase/dto.go create mode 100644 coordinator/internal/usecase/job.go create mode 100644 coordinator/internal/usecase/ports.go create mode 100644 coordinator/internal/usecase/task.go create mode 100644 coordinator/migrations/0001_init.down.sql create mode 100644 coordinator/migrations/0001_init.up.sql diff --git a/coordinator/.dockerignore b/coordinator/.dockerignore new file mode 100644 index 0000000..a4d0d5c --- /dev/null +++ b/coordinator/.dockerignore @@ -0,0 +1,14 @@ +# Keep the build context small and never bake secrets or local state into an image. +.env +.git +.gitignore +*.md +Makefile +docker-compose.yml +Dockerfile +.dockerignore + +# Local build artifacts +/coordinator +/bin/ +*.out diff --git a/coordinator/.env.example b/coordinator/.env.example new file mode 100644 index 0000000..2887295 --- /dev/null +++ b/coordinator/.env.example @@ -0,0 +1,14 @@ +# 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 + +# Shared bearer token every worker must present. Leave empty to disable auth (dev only). +WORKER_AUTH_TOKEN=change-me + +# Optional tuning (defaults shown). +DB_MAX_CONNS=10 +REQUEST_TIMEOUT=15s +LEASE_DURATION=2m +DEFAULT_MAX_ATTEMPTS=3 +REAPER_INTERVAL=30s diff --git a/coordinator/.gitignore b/coordinator/.gitignore new file mode 100644 index 0000000..c44a5e4 --- /dev/null +++ b/coordinator/.gitignore @@ -0,0 +1,4 @@ +/coordinator +/bin/ +.env +*.out diff --git a/coordinator/.golangci.yml b/coordinator/.golangci.yml new file mode 100644 index 0000000..428bfed --- /dev/null +++ b/coordinator/.golangci.yml @@ -0,0 +1,54 @@ +version: "2" + +run: + timeout: 3m + +linters: + # "standard" = errcheck, govet, ineffassign, staticcheck, unused. + default: standard + enable: + # Catches `err == ErrFoo` where errors.Is is required. Directly relevant + # here: domain exposes sentinel errors that use cases may wrap with %w. + - errorlint + # Returning nil after checking a non-nil error — a silent bug factory. + - nilerr + # http.Get/Do without a context: every outbound call must be cancellable. + - noctx + # Unclosed response bodies leak connections. + - bodyclose + # Common security mistakes (weak crypto, unhandled file perms). + - gosec + # Style and naming consistency. + - revive + - misspell + - unconvert + + settings: + errcheck: + # Deferred Close/Rollback are intentionally ignored in a few places + # (rollback after commit is a documented no-op). + check-type-assertions: true + revive: + rules: + - name: exported + disabled: true # internal packages need no exported-symbol comments + gosec: + excludes: + - G404 # math/rand is fine for jitter; nothing here is security-sensitive + + exclusions: + rules: + # Tests may skip error checks and use long literals freely. + - path: _test\.go + linters: + - errcheck + - gosec + +formatters: + enable: + - gofmt + - goimports + settings: + goimports: + local-prefixes: + - github.com/emil28092005/SciMesh/coordinator diff --git a/coordinator/ARCHITECTURE.md b/coordinator/ARCHITECTURE.md new file mode 100644 index 0000000..77b6976 --- /dev/null +++ b/coordinator/ARCHITECTURE.md @@ -0,0 +1,144 @@ +# Архитектура координатора + +Карта кода. Читать сверху вниз: сначала «где что лежит», потом «как проходит +запрос», в конце — «куда добавлять новое». + +--- + +## 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/coordinator/Dockerfile b/coordinator/Dockerfile new file mode 100644 index 0000000..bf0b1dc --- /dev/null +++ b/coordinator/Dockerfile @@ -0,0 +1,47 @@ +# syntax=docker/dockerfile:1 +# +# Requires BuildKit (the RUN --mount cache lines below). Docker 23+ enables it +# by default when the buildx plugin is present; install `docker-buildx` if a +# build fails with "the --mount option requires BuildKit". + +# --- build stage ---------------------------------------------------------- +FROM golang:1.24-alpine AS build + +WORKDIR /src + +# Copy manifests first: this layer stays cached until dependencies actually +# change, so editing Go sources does not re-download the module graph. +COPY go.mod go.sum ./ +RUN --mount=type=cache,target=/go/pkg/mod go mod download + +COPY . . + +# The cache mounts persist the module cache and the compiler's build cache +# *across* builds, so a rebuild after a code edit recompiles only what changed +# instead of the whole dependency tree. +# +# CGO_ENABLED=0 produces a fully static binary, so the runtime image needs no +# libc. -trimpath strips local paths; -s -w drop the symbol table and DWARF. +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + CGO_ENABLED=0 GOOS=linux go build \ + -trimpath -ldflags="-s -w" \ + -o /out/coordinator ./cmd/coordinator + +# --- runtime stage -------------------------------------------------------- +FROM alpine:3.20 + +# ca-certificates for outbound TLS; wget backs the container healthcheck. +RUN apk add --no-cache ca-certificates wget \ + && adduser -D -H -u 10001 coordinator + +COPY --from=build /out/coordinator /usr/local/bin/coordinator + +# Never run as root: a compromised process should not own the container. +USER coordinator + +EXPOSE 8080 + +# Exec form, not shell: the binary becomes PID 1 and receives SIGTERM directly, +# which is what its graceful shutdown depends on. +ENTRYPOINT ["/usr/local/bin/coordinator"] diff --git a/coordinator/Makefile b/coordinator/Makefile new file mode 100644 index 0000000..d85c19e --- /dev/null +++ b/coordinator/Makefile @@ -0,0 +1,60 @@ +.PHONY: build run test vet tidy migrate-up migrate-down up down logs ps rebuild psql + +# --- build / run --------------------------------------------------------- +build: + go build ./... + +run: + go run ./cmd/coordinator + +test: + go test ./... + +vet: + go vet ./... + +# Runs golangci-lint without installing it system-wide. Install it for speed: +# pacman -S golangci-lint (Arch) +LINT_VERSION := v2.12.2 +lint: + @command -v golangci-lint >/dev/null 2>&1 \ + && golangci-lint run ./... \ + || go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(LINT_VERSION) run ./... + +tidy: + go mod tidy + +# --- migrations ---------------------------------------------------------- +# Requires the golang-migrate CLI: +# go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest +# DATABASE_URL must be set, e.g.: +# export DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable' +migrate-up: + migrate -path migrations -database "$(DATABASE_URL)" up + +migrate-down: + migrate -path migrations -database "$(DATABASE_URL)" down 1 + +# --- docker -------------------------------------------------------------- +# `up` starts Postgres, applies migrations, then launches the coordinator. +up: + docker compose up -d --build + +down: + docker compose down + +# Also drops the database volume — use when the schema is beyond repair. +down-clean: + docker compose down -v + +logs: + docker compose logs -f coordinator + +ps: + docker compose ps + +rebuild: + docker compose up -d --build --force-recreate coordinator + +psql: + docker compose exec postgres psql -U scimesh -d scimesh diff --git a/coordinator/README.md b/coordinator/README.md new file mode 100644 index 0000000..f83a9be --- /dev/null +++ b/coordinator/README.md @@ -0,0 +1,143 @@ +# SciMesh Coordinator + +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. + +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 + +``` + 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 ────┘ +``` + +`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) +``` + +`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 | `/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) | + +## Status + +Scaffold with a **complete, tested domain**. Layers, wiring, routing, auth, +access logging, error mapping, transactions, migrations, and graceful shutdown +are in place. Repository methods are stubs returning `ErrNotImplemented` +(→ HTTP 501); the SQL for claiming and lease expiry is written and ready to wire. + +Roadmap: + +1. schema + migrations ✅ +2. `ClaimNext`, `InsertBatch` — atomic claim via `FOR UPDATE SKIP LOCKED` +3. `GetForUpdate`, `Update`, `CountByStatus` — completes the result/failure paths +4. file upload / chunk download +5. `ExpireLeases` — SQL is written, needs wiring +6. stitcher: merge per-chunk top-k into the final CSV +7. integration tests against real Postgres via `TEST_DATABASE_URL` + +## 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: + +```sh +go test ./... +go vet ./... +``` + +Integration tests (concurrent claiming, migrations) come in phase 7 and require +a real PostgreSQL instance supplied through `TEST_DATABASE_URL`. diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go new file mode 100644 index 0000000..d2fb7ec --- /dev/null +++ b/coordinator/cmd/coordinator/main.go @@ -0,0 +1,98 @@ +// Command coordinator is the SciMesh task-queue server. It owns all database +// access; workers reach it only over HTTP and never receive DB credentials. +// Migrations are a separate explicit command (see Makefile) — this binary never +// mutates schema at startup. +package main + +import ( + "context" + "log/slog" + "os" + "os/signal" + "sync" + "syscall" + + "github.com/emil28092005/SciMesh/coordinator/internal/infra" + "github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres" + httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +func main() { + log := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + + // All work happens in run() so its defers (pool.Close, signal stop) still + // execute: os.Exit skips deferred calls entirely. + if err := run(log); err != nil { + log.Error("fatal", "err", err) + os.Exit(1) + } +} + +func run(log *slog.Logger) error { + cfg, err := infra.Load() + if err != nil { + return err + } + + // One cancellation source for the whole process: HTTP server and reaper + // both observe it and wind down together. + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + pool, err := infra.NewPool(ctx, cfg) + if err != nil { + return err + } + defer pool.Close() + + // --- composition root: the only place that knows concrete types --- + // + // Wiring reads outward-in: adapters are constructed, then injected into + // use cases through their ports. Nothing below this function can see a + // pgxpool, and nothing above the repositories can see SQL. + var ( + clk = infra.NewClock() + tx = postgres.NewTxManager(pool) + taskRepo = postgres.NewTaskRepo(pool) + jobRepo = postgres.NewJobRepo(pool) + ) + + useCases := httptransport.UseCases{ + CreateJob: usecase.NewCreateJob(jobRepo, taskRepo, tx, clk), + ClaimTask: usecase.NewClaimTask(taskRepo, clk, cfg.LeaseDuration), + RenewLease: usecase.NewRenewLease(taskRepo, tx, clk, cfg.LeaseDuration), + CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, tx, clk), + FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk), + GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), + } + + // Background workers are tracked so shutdown can wait for them. Without + // this the process would exit while the reaper sat mid-UPDATE, and the + // deferred pool.Close() would pull connections out from under it. + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + infra.RunReaper(ctx, log, usecase.NewExpireLeases(taskRepo, clk), cfg.ReaperInterval) + }() + + api := httptransport.NewServer(useCases, log, cfg.RequestTimeout) + err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.WorkerAuthToken)) + + // Shutdown order matters, and defers alone cannot express it (they run + // LIFO, so the deferred stop() would fire *after* the wait below). + // + // 1. stop() cancel the context, telling the reaper to finish + // 2. wg.Wait() let it return from its current tick + // 3. deferred pool.Close() closes an idle pool, not a busy one + // + // Calling stop() here also covers the path where RunServer failed on its + // own: the context would never be cancelled otherwise and wg.Wait() + // would block forever. + stop() + wg.Wait() + log.Info("shutdown complete") + + return err +} diff --git a/coordinator/docker-compose.yml b/coordinator/docker-compose.yml new file mode 100644 index 0000000..1cbe5b9 --- /dev/null +++ b/coordinator/docker-compose.yml @@ -0,0 +1,67 @@ +name: scimesh + +services: + postgres: + image: postgres:16-alpine + environment: + POSTGRES_USER: ${POSTGRES_USER:-scimesh} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-scimesh} + POSTGRES_DB: ${POSTGRES_DB:-scimesh} + ports: + - "${POSTGRES_PORT:-5432}:5432" + volumes: + - pgdata:/var/lib/postgresql/data + healthcheck: + # Everything else waits on this, so the check must prove the server + # accepts queries — not merely that the port is open. + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-scimesh} -d ${POSTGRES_DB:-scimesh}"] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + + # One-shot: applies migrations, then exits. Schema changes stay an explicit + # deployment step — the coordinator binary never migrates on startup. + migrate: + image: migrate/migrate:v4.17.1 + depends_on: + postgres: + condition: service_healthy + volumes: + - ./migrations:/migrations:ro + command: + - -path=/migrations + - -database=postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh}?sslmode=disable + - up + restart: on-failure + + coordinator: + build: + context: . + depends_on: + postgres: + condition: service_healthy + # Start only once the schema exists, otherwise the first query fails. + migrate: + condition: service_completed_successfully + environment: + COORDINATOR_ADDR: ":8080" + # Host is the service name: compose resolves it on the project network. + DATABASE_URL: postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh}?sslmode=disable + WORKER_AUTH_TOKEN: ${WORKER_AUTH_TOKEN:-dev-token} + DB_MAX_CONNS: "10" + REQUEST_TIMEOUT: "15s" + LEASE_DURATION: "2m" + REAPER_INTERVAL: "30s" + ports: + - "${COORDINATOR_PORT:-8080}:8080" + healthcheck: + test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"] + interval: 10s + timeout: 3s + retries: 3 + start_period: 5s + restart: unless-stopped + +volumes: + pgdata: diff --git a/coordinator/go.mod b/coordinator/go.mod new file mode 100644 index 0000000..dd02541 --- /dev/null +++ b/coordinator/go.mod @@ -0,0 +1,19 @@ +module github.com/emil28092005/SciMesh/coordinator + +go 1.22 + +require ( + github.com/cenkalti/backoff/v4 v4.3.0 + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.6.0 + github.com/joho/godotenv v1.5.1 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect + github.com/jackc/puddle/v2 v2.2.1 // indirect + golang.org/x/crypto v0.17.0 // indirect + golang.org/x/sync v0.1.0 // indirect + golang.org/x/text v0.14.0 // indirect +) diff --git a/coordinator/go.sum b/coordinator/go.sum new file mode 100644 index 0000000..c078c1d --- /dev/null +++ b/coordinator/go.sum @@ -0,0 +1,34 @@ +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY= +github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw= +github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= +github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k= +golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/coordinator/internal/domain/errors.go b/coordinator/internal/domain/errors.go new file mode 100644 index 0000000..f1e69e8 --- /dev/null +++ b/coordinator/internal/domain/errors.go @@ -0,0 +1,18 @@ +package domain + +import "errors" + +// Business-rule violations. They live in the innermost layer because they +// describe what the rules are, not how a transport reports them: the HTTP +// adapter maps these to status codes, and nothing here knows 409 exists. +// +// Always compare with errors.Is — outer layers may wrap these with %w. +var ( + ErrJobNotFound = errors.New("job not found") + ErrTaskNotFound = errors.New("task not found") + ErrLeaseConflict = errors.New("task leased to another worker") + ErrStaleAttempt = errors.New("attempt does not match lease") + ErrResultConflict = errors.New("different result already recorded") + ErrInvalidInput = errors.New("invalid input") + ErrTaskNotLeased = errors.New("task is not currently leased") +) diff --git a/coordinator/internal/domain/job.go b/coordinator/internal/domain/job.go new file mode 100644 index 0000000..b6e9f4a --- /dev/null +++ b/coordinator/internal/domain/job.go @@ -0,0 +1,107 @@ +package domain + +import ( + "time" + + "github.com/google/uuid" +) + +type JobStatus string + +const ( + JobPending JobStatus = "pending" + JobRunning JobStatus = "running" + JobCompleted JobStatus = "completed" + JobFailed JobStatus = "failed" + JobCancelled JobStatus = "cancelled" +) + +// Job is one user submission that fans out into one or more tasks. +type Job struct { + ID uuid.UUID + Workload string + InputURI string + Parameters map[string]any + Status JobStatus + CreatedAt time.Time + CompletedAt *time.Time +} + +// ChunkSpec describes one piece a job is split into. Callers build these from +// whatever chunking strategy the workload uses; the domain only validates them. +type ChunkSpec struct { + ChunkIndex int + Workload string // empty inherits the job's workload + InputURI string + InputSHA256 string + Parameters map[string]any + MaxAttempts int +} + +// NewJobWithTasks builds a job together with all of its tasks, validating the +// set as a whole. Returning both from one constructor keeps the invariant +// visible: a job without tasks, or with duplicate chunk indexes, cannot exist. +func NewJobWithTasks(workload, inputURI string, params map[string]any, + chunks []ChunkSpec, now time.Time) (*Job, []*Task, error) { + + if workload == "" || inputURI == "" || len(chunks) == 0 { + return nil, nil, ErrInvalidInput + } + + job := &Job{ + ID: uuid.New(), + Workload: workload, + InputURI: inputURI, + Parameters: params, + Status: JobPending, + CreatedAt: now, + } + + seen := make(map[int]struct{}, len(chunks)) + tasks := make([]*Task, 0, len(chunks)) + for _, c := range chunks { + if _, dup := seen[c.ChunkIndex]; dup { + return nil, nil, ErrInvalidInput // unique (job_id, chunk_index) + } + seen[c.ChunkIndex] = struct{}{} + + w := c.Workload + if w == "" { + w = workload + } + task, err := NewTask(job.ID, c.ChunkIndex, w, c.InputURI, c.InputSHA256, + c.Parameters, c.MaxAttempts, now) + if err != nil { + return nil, nil, err + } + tasks = append(tasks, task) + } + return job, tasks, nil +} + +// JobProgress is the aggregate view of a job and the state of its tasks. +type JobProgress struct { + Job Job + Total int + Pending int + Leased int + Done int + Failed int +} + +// DeriveStatus computes what the job's status should be from its task counts, +// so the rule lives here rather than in a SQL trigger or a handler. +func (p JobProgress) DeriveStatus() JobStatus { + switch { + case p.Total == 0: + return JobPending + case p.Done == p.Total: + return JobCompleted + case p.Failed > 0 && p.Done+p.Failed == p.Total: + return JobFailed + case p.Leased > 0 || p.Done > 0 || p.Failed > 0: + return JobRunning + default: + return JobPending + } +} diff --git a/coordinator/internal/domain/task.go b/coordinator/internal/domain/task.go new file mode 100644 index 0000000..6704552 --- /dev/null +++ b/coordinator/internal/domain/task.go @@ -0,0 +1,244 @@ +// Package domain holds SciMesh's entities and the rules that govern them. It +// is the innermost layer: it imports nothing from this module and knows nothing +// about HTTP, SQL, or configuration. Every state transition a task can undergo +// is a method here, so the rules are unit-testable without a database. +package domain + +import ( + "time" + + "github.com/google/uuid" +) + +type TaskStatus string + +const ( + TaskPending TaskStatus = "pending" + TaskLeased TaskStatus = "leased" + TaskCompleted TaskStatus = "completed" + TaskFailed TaskStatus = "failed" + TaskCancelled TaskStatus = "cancelled" +) + +// ErrCodeLeaseExpired marks tasks failed by the reaper rather than by a worker. +const ErrCodeLeaseExpired = "lease_expired" + +// Task is one independently executable chunk of a job. +// +// Nullable columns are pointers so "no lease" stays distinguishable from +// "lease owned by the empty string" — a plain string cannot express both. +type Task struct { + ID uuid.UUID + JobID uuid.UUID + ChunkIndex int + Workload string + InputURI string + InputSHA256 string + Parameters map[string]any + Status TaskStatus + Attempt int + MaxAttempts int + LeaseOwner *string + LeaseExpiresAt *time.Time + ResultURI *string + ResultSHA256 *string + Metrics map[string]any + ErrorCode *string + ErrorMessage *string + CreatedAt time.Time + StartedAt *time.Time + CompletedAt *time.Time + Version int +} + +// NewTask builds a pending task. maxAttempts <= 0 falls back to the default. +func NewTask(jobID uuid.UUID, chunkIndex int, workload, inputURI, inputSHA256 string, + params map[string]any, maxAttempts int, now time.Time) (*Task, error) { + + if inputURI == "" { + return nil, ErrInvalidInput + } + if inputSHA256 == "" { + return nil, ErrInvalidInput // checksum is mandatory: workers verify inputs + } + if chunkIndex < 0 { + return nil, ErrInvalidInput + } + if maxAttempts <= 0 { + maxAttempts = DefaultMaxAttempts + } + return &Task{ + ID: uuid.New(), + JobID: jobID, + ChunkIndex: chunkIndex, + Workload: workload, + InputURI: inputURI, + InputSHA256: inputSHA256, + Parameters: params, + Status: TaskPending, + Attempt: 0, + MaxAttempts: maxAttempts, + CreatedAt: now, + }, nil +} + +// DefaultMaxAttempts applies when a task does not specify its own ceiling. +const DefaultMaxAttempts = 3 + +// CanRetry reports whether any attempts remain. +func (t *Task) CanRetry() bool { return t.Attempt < t.MaxAttempts } + +// IsLeaseHeldBy reports whether worker currently holds this task at attempt. +func (t *Task) IsLeaseHeldBy(worker string, attempt int) bool { + return t.LeaseOwner != nil && *t.LeaseOwner == worker && t.Attempt == attempt +} + +// AsClaimed projects the task into the trimmed view handed to a worker: +// everything needed to execute, nothing it has no business seeing. +func (t *Task) AsClaimed() ClaimedTask { + ct := ClaimedTask{ + TaskID: t.ID, + JobID: t.JobID, + ChunkIndex: t.ChunkIndex, + Workload: t.Workload, + InputURI: t.InputURI, + InputSHA256: t.InputSHA256, + Parameters: t.Parameters, + Attempt: t.Attempt, + } + if t.LeaseOwner != nil { + ct.LeaseOwner = *t.LeaseOwner + } + if t.LeaseExpiresAt != nil { + ct.LeaseExpiresAt = *t.LeaseExpiresAt + } + return ct +} + +// verifyLease is the guard every worker-driven transition shares: the caller +// must own the lease and reference the attempt it was granted. +func (t *Task) verifyLease(worker string, attempt int) error { + if t.Status != TaskLeased { + return ErrTaskNotLeased + } + if t.LeaseOwner == nil || *t.LeaseOwner != worker { + return ErrLeaseConflict + } + if t.Attempt != attempt { + return ErrStaleAttempt + } + return nil +} + +// RenewLease extends the lease of the worker that holds it. +func (t *Task) RenewLease(worker string, attempt int, until time.Time) error { + if err := t.verifyLease(worker, attempt); err != nil { + return err + } + t.LeaseExpiresAt = &until + t.Version++ + return nil +} + +// CompleteWith records a successful result. +// +// Idempotency comes first deliberately: a worker whose network dropped will +// retry the same manifest, and that must succeed rather than trip the lease +// check on a task the coordinator already finished. A *different* manifest for +// an already-completed task is a genuine conflict. +func (t *Task) CompleteWith(resultURI, resultSHA256 string, metrics map[string]any, + worker string, attempt int, now time.Time) error { + + if resultURI == "" || resultSHA256 == "" { + return ErrInvalidInput + } + + if t.Status == TaskCompleted { + if t.Attempt == attempt && t.ResultURI != nil && *t.ResultURI == resultURI && + t.ResultSHA256 != nil && *t.ResultSHA256 == resultSHA256 { + return nil // same attempt, same manifest — replay of a successful call + } + return ErrResultConflict + } + + if err := t.verifyLease(worker, attempt); err != nil { + return err + } + + t.Status = TaskCompleted + t.ResultURI = &resultURI + t.ResultSHA256 = &resultSHA256 + t.Metrics = metrics + t.CompletedAt = &now + t.LeaseOwner = nil + t.LeaseExpiresAt = nil + t.ErrorCode = nil + t.ErrorMessage = nil + t.Version++ + return nil +} + +// Fail records a worker-reported failure. A retryable failure with attempts +// left returns the task to the queue; otherwise it terminates as failed. +func (t *Task) Fail(worker string, attempt int, code, message string, retryable bool, now time.Time) error { + if err := t.verifyLease(worker, attempt); err != nil { + return err + } + t.ErrorCode = &code + t.ErrorMessage = &message + t.LeaseOwner = nil + t.LeaseExpiresAt = nil + t.Version++ + + if retryable && t.CanRetry() { + t.Status = TaskPending + return nil + } + t.Status = TaskFailed + t.CompletedAt = &now + return nil +} + +// ExpireLease is applied by the reaper when a lease elapses without a +// heartbeat: requeue while attempts remain, otherwise fail terminally. +func (t *Task) ExpireLease(now time.Time) { + if t.Status != TaskLeased { + return + } + t.LeaseOwner = nil + t.LeaseExpiresAt = nil + t.Version++ + + if t.CanRetry() { + t.Status = TaskPending + return + } + code, msg := ErrCodeLeaseExpired, "lease expired after the final attempt" + t.ErrorCode = &code + t.ErrorMessage = &msg + t.Status = TaskFailed + t.CompletedAt = &now +} + +// ClaimedTask is the worker-facing projection of a leased task. +type ClaimedTask struct { + TaskID uuid.UUID + JobID uuid.UUID + ChunkIndex int + Workload string + InputURI string + InputSHA256 string + Parameters map[string]any + Attempt int + LeaseOwner string + LeaseExpiresAt time.Time +} + +// ResultManifest is a completed task's output, ordered for the stitcher. +type ResultManifest struct { + TaskID uuid.UUID + ChunkIndex int + ResultURI string + ResultSHA256 string + Metrics map[string]any +} diff --git a/coordinator/internal/domain/task_test.go b/coordinator/internal/domain/task_test.go new file mode 100644 index 0000000..3650478 --- /dev/null +++ b/coordinator/internal/domain/task_test.go @@ -0,0 +1,183 @@ +package domain + +import ( + "errors" + "testing" + "time" + + "github.com/google/uuid" +) + +var ( + testNow = time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC) + testLater = testNow.Add(time.Hour) + testWorker = "worker-1" +) + +// leasedTask builds a task already leased to testWorker at the given attempt. +func leasedTask(attempt, maxAttempts int) *Task { + owner := testWorker + expires := testLater + return &Task{ + ID: uuid.New(), + JobID: uuid.New(), + Status: TaskLeased, + Attempt: attempt, + MaxAttempts: maxAttempts, + LeaseOwner: &owner, + LeaseExpiresAt: &expires, + } +} + +func TestCompleteWithRecordsResult(t *testing.T) { + task := leasedTask(1, 3) + + if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != TaskCompleted { + t.Errorf("status = %q, want completed", task.Status) + } + if task.LeaseOwner != nil || task.LeaseExpiresAt != nil { + t.Error("lease must be released on completion") + } + if task.CompletedAt == nil || !task.CompletedAt.Equal(testNow) { + t.Error("completed_at must be stamped") + } +} + +// A worker whose network dropped retries the same manifest; that must succeed +// rather than fail on the lease it has already given up. +func TestCompleteWithIsIdempotentForSameManifest(t *testing.T) { + task := leasedTask(1, 3) + if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + t.Fatalf("first call: %v", err) + } + versionAfterFirst := task.Version + + if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testLater); err != nil { + t.Fatalf("replay must be idempotent, got %v", err) + } + if task.Version != versionAfterFirst { + t.Error("replay must not mutate the task") + } +} + +func TestCompleteWithRejectsDifferentManifest(t *testing.T) { + task := leasedTask(1, 3) + if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + t.Fatalf("first call: %v", err) + } + + err := task.CompleteWith("s3://other.csv", "def", nil, testWorker, 1, testLater) + if !errors.Is(err, ErrResultConflict) { + t.Errorf("err = %v, want ErrResultConflict", err) + } +} + +func TestCompleteWithRejectsForeignWorker(t *testing.T) { + task := leasedTask(1, 3) + + err := task.CompleteWith("s3://r.csv", "abc", nil, "worker-2", 1, testNow) + if !errors.Is(err, ErrLeaseConflict) { + t.Errorf("err = %v, want ErrLeaseConflict", err) + } +} + +func TestCompleteWithRejectsStaleAttempt(t *testing.T) { + task := leasedTask(2, 3) // task is on attempt 2 + + err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow) // worker thinks it is 1 + if !errors.Is(err, ErrStaleAttempt) { + t.Errorf("err = %v, want ErrStaleAttempt", err) + } +} + +func TestFailRequeuesWhileAttemptsRemain(t *testing.T) { + task := leasedTask(1, 3) + + if err := task.Fail(testWorker, 1, "boom", "exploded", true, testNow); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != TaskPending { + t.Errorf("status = %q, want pending", task.Status) + } + if task.LeaseOwner != nil { + t.Error("lease must be released so another worker can claim it") + } +} + +func TestFailTerminatesOnFinalAttempt(t *testing.T) { + task := leasedTask(3, 3) // no attempts left + + if err := task.Fail(testWorker, 3, "boom", "exploded", true, testNow); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != TaskFailed { + t.Errorf("status = %q, want failed", task.Status) + } +} + +func TestFailIsTerminalWhenNotRetryable(t *testing.T) { + task := leasedTask(1, 3) // attempts remain, but the error is fatal + + if err := task.Fail(testWorker, 1, "bad_input", "checksum mismatch", false, testNow); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != TaskFailed { + t.Errorf("status = %q, want failed", task.Status) + } +} + +// This is the MVP acceptance criterion: a dead worker must not strand its task. +func TestExpireLeaseRequeuesWhileAttemptsRemain(t *testing.T) { + task := leasedTask(1, 3) + + task.ExpireLease(testNow) + + if task.Status != TaskPending { + t.Errorf("status = %q, want pending", task.Status) + } + if task.LeaseOwner != nil || task.LeaseExpiresAt != nil { + t.Error("expired lease must be cleared") + } +} + +func TestExpireLeaseFailsAfterFinalAttempt(t *testing.T) { + task := leasedTask(3, 3) + + task.ExpireLease(testNow) + + if task.Status != TaskFailed { + t.Errorf("status = %q, want failed", task.Status) + } + if task.ErrorCode == nil || *task.ErrorCode != ErrCodeLeaseExpired { + t.Error("expected a lease_expired error code") + } +} + +func TestExpireLeaseIgnoresUnleasedTasks(t *testing.T) { + task := &Task{Status: TaskCompleted, Attempt: 1, MaxAttempts: 3} + + task.ExpireLease(testNow) + + if task.Status != TaskCompleted { + t.Errorf("status = %q, completed tasks must be untouched", task.Status) + } +} + +func TestRenewLeaseExtendsOnlyForHolder(t *testing.T) { + task := leasedTask(1, 3) + until := testLater.Add(time.Hour) + + if err := task.RenewLease(testWorker, 1, until); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !task.LeaseExpiresAt.Equal(until) { + t.Error("lease must be extended") + } + + if err := task.RenewLease("worker-2", 1, until); !errors.Is(err, ErrLeaseConflict) { + t.Errorf("err = %v, want ErrLeaseConflict", err) + } +} diff --git a/coordinator/internal/infra/clock.go b/coordinator/internal/infra/clock.go new file mode 100644 index 0000000..eedbcde --- /dev/null +++ b/coordinator/internal/infra/clock.go @@ -0,0 +1,13 @@ +// Clock: the real implementation of the usecase.Clock port. It lives out here +// because reading the system clock is infrastructure; tests substitute a fixed one. +package infra + +import "time" + +type System struct{} + +func NewClock() System { return System{} } + +// Now returns UTC so every timestamp the coordinator writes is comparable +// regardless of the host's timezone. +func (System) Now() time.Time { return time.Now().UTC() } diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go new file mode 100644 index 0000000..db07876 --- /dev/null +++ b/coordinator/internal/infra/config.go @@ -0,0 +1,135 @@ +// Config: coordinator settings, read only from the environment, so the same +// binary behaves identically in CI, local, and prod. +package infra + +import ( + "errors" + "fmt" + "io/fs" + "math" + "os" + "strconv" + "time" + + "github.com/joho/godotenv" +) + +// defaultEnvFile is loaded by Load unless ENV_FILE points elsewhere. +const defaultEnvFile = ".env" + +type Config struct { + // HTTP listen address, e.g. ":8080". + Addr string + // PostgreSQL connection string (pgx format / libpq URL). + DatabaseURL string + // Shared bearer token workers must present. Empty disables auth (dev only). + WorkerAuthToken string + + // Connection pool upper bound. + DBMaxConns int32 + // Per-request context timeout applied to handlers and DB calls. + RequestTimeout time.Duration + + // Default lease length handed out on claim. + LeaseDuration time.Duration + // Default attempt ceiling for newly created tasks. + DefaultMaxAttempts int + // How often the background lease-reaper runs. + ReaperInterval time.Duration +} + +// Load reads the environment and fails fast on anything required-but-missing +// or malformed, so a misconfigured process never limps along half-wired. +// +// A .env file (path overridable via ENV_FILE) is loaded first as a local-dev +// convenience. It only fills variables the environment does not already define. +func Load() (Config, error) { + envFile := os.Getenv("ENV_FILE") + if envFile == "" { + envFile = defaultEnvFile + } + // godotenv.Load never overwrites variables already present in the + // environment, so an orchestrator's values always beat the file. A missing + // file is expected in production, where env vars are injected directly. + if err := godotenv.Load(envFile); err != nil && !errors.Is(err, fs.ErrNotExist) { + return Config{}, fmt.Errorf("load env file %q: %w", envFile, err) + } + + cfg := Config{ + Addr: getEnv("COORDINATOR_ADDR", ":8080"), + DatabaseURL: os.Getenv("DATABASE_URL"), + WorkerAuthToken: os.Getenv("WORKER_AUTH_TOKEN"), + DBMaxConns: 10, + RequestTimeout: 15 * time.Second, + LeaseDuration: 2 * time.Minute, + DefaultMaxAttempts: 3, + ReaperInterval: 30 * time.Second, + } + + if cfg.DatabaseURL == "" { + return Config{}, fmt.Errorf("DATABASE_URL is required") + } + + var err error + if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil { + return Config{}, err + } + if cfg.RequestTimeout, err = getEnvDuration("REQUEST_TIMEOUT", cfg.RequestTimeout); err != nil { + return Config{}, err + } + if cfg.LeaseDuration, err = getEnvDuration("LEASE_DURATION", cfg.LeaseDuration); err != nil { + return Config{}, err + } + if cfg.ReaperInterval, err = getEnvDuration("REAPER_INTERVAL", cfg.ReaperInterval); err != nil { + return Config{}, err + } + if cfg.DefaultMaxAttempts, err = getEnvInt("DEFAULT_MAX_ATTEMPTS", cfg.DefaultMaxAttempts); err != nil { + return Config{}, err + } + + return cfg, nil +} + +func getEnv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func getEnvInt(key string, def int) (int, error) { + v := os.Getenv(key) + if v == "" { + return def, nil + } + n, err := strconv.Atoi(v) + if err != nil { + return 0, fmt.Errorf("%s: %w", key, err) + } + return n, nil +} + +func getEnvInt32(key string, def int32) (int32, error) { + n, err := getEnvInt(key, int(def)) + if err != nil { + return 0, err + } + // On 64-bit builds int is wider than int32, so an oversized value would + // wrap silently — DB_MAX_CONNS=2147483648 becoming a negative pool size. + if n < math.MinInt32 || n > math.MaxInt32 { + return 0, fmt.Errorf("%s: %d is out of range for int32", key, n) + } + return int32(n), nil +} + +func getEnvDuration(key string, def time.Duration) (time.Duration, error) { + v := os.Getenv(key) + if v == "" { + return def, nil + } + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf("%s: %w", key, err) + } + return d, nil +} diff --git a/coordinator/internal/infra/db.go b/coordinator/internal/infra/db.go new file mode 100644 index 0000000..29899b3 --- /dev/null +++ b/coordinator/internal/infra/db.go @@ -0,0 +1,30 @@ +// DB: the PostgreSQL connection pool. +package infra + +import ( + "context" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// NewPool builds the single shared pool. The caller owns its lifetime and must +// Close() it on shutdown. +func NewPool(ctx context.Context, cfg Config) (*pgxpool.Pool, error) { + poolCfg, err := pgxpool.ParseConfig(cfg.DatabaseURL) + if err != nil { + return nil, err + } + poolCfg.MaxConns = cfg.DBMaxConns + + pool, err := pgxpool.NewWithConfig(ctx, poolCfg) + if err != nil { + return nil, err + } + // pgxpool.New is lazy, so without this ping a bad DATABASE_URL would only + // surface on the first request instead of at startup. + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, err + } + return pool, nil +} diff --git a/coordinator/internal/infra/doc.go b/coordinator/internal/infra/doc.go new file mode 100644 index 0000000..4e64f47 --- /dev/null +++ b/coordinator/internal/infra/doc.go @@ -0,0 +1,7 @@ +// Package infra holds the outermost layer: frameworks, drivers, and process +// wiring. It reads configuration, opens the database pool, supplies the real +// clock, and runs the HTTP server and background reaper. +// +// Nothing inward depends on this package — it is the last thing constructed and +// the first thing that would be swapped when the runtime environment changes. +package infra diff --git a/coordinator/internal/infra/server.go b/coordinator/internal/infra/server.go new file mode 100644 index 0000000..c017726 --- /dev/null +++ b/coordinator/internal/infra/server.go @@ -0,0 +1,72 @@ +// Server: the HTTP listener and the background lease reaper, both shut down +// cleanly on a signal. +package infra + +import ( + "context" + "errors" + "log/slog" + "net/http" + "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +const shutdownGrace = 15 * time.Second + +// Run 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, + Handler: handler, + ReadHeaderTimeout: 5 * time.Second, + } + + // Buffered so this goroutine can exit even when nobody reads the channel + // (the ctx.Done branch below) — an unbuffered send would leak it forever. + errCh := make(chan error, 1) + go func() { + log.Info("coordinator listening", "addr", addr) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + select { + case err := <-errCh: + return err + case <-ctx.Done(): + log.Info("shutdown signal received") + } + + // A fresh context: ctx is already cancelled, and reusing it would abort the + // very requests we are trying to let finish. + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace) + 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. +func RunReaper(ctx context.Context, log *slog.Logger, uc *usecase.ExpireLeases, interval time.Duration) { + t := time.NewTicker(interval) + defer t.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-t.C: + n, err := uc.Execute(ctx) + if err != nil { + // Demoted to debug while the repository is still a stub; + // raise to Warn once phase 5 lands. + log.Debug("reaper skipped", "err", err) + continue + } + if n > 0 { + log.Info("reaper requeued expired leases", "count", n) + } + } + } +} diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go new file mode 100644 index 0000000..3751801 --- /dev/null +++ b/coordinator/internal/storage/postgres/job_repo.go @@ -0,0 +1,40 @@ +package postgres + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// JobRepo implements usecase.JobRepository. +type JobRepo struct { + pool *pgxpool.Pool +} + +func NewJobRepo(pool *pgxpool.Pool) *JobRepo { + return &JobRepo{pool: pool} +} + +var _ usecase.JobRepository = (*JobRepo)(nil) + +// TODO(phase 2-3): replace stubs with real pgx queries. + +func (r *JobRepo) Insert(ctx context.Context, j *domain.Job) error { + // Phase 2: INSERT INTO jobs ...; runs inside the caller's transaction. + return usecase.ErrNotImplemented +} + +func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) { + // Phase 3: SELECT ... WHERE id = $1; no rows -> domain.ErrJobNotFound. + return nil, usecase.ErrNotImplemented +} + +func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error { + // Phase 3: UPDATE jobs SET status = $2, completed_at = $3 WHERE id = $1. + return usecase.ErrNotImplemented +} diff --git a/coordinator/internal/storage/postgres/retry.go b/coordinator/internal/storage/postgres/retry.go new file mode 100644 index 0000000..8f0ae86 --- /dev/null +++ b/coordinator/internal/storage/postgres/retry.go @@ -0,0 +1,82 @@ +package postgres + +import ( + "context" + "errors" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/jackc/pgx/v5/pgconn" +) + +// Transient PostgreSQL failures. Under concurrent claiming these are expected +// rather than exceptional: two coordinators touching neighbouring rows can +// deadlock or fail to serialize, and the correct response is to try again. +const ( + codeSerializationFailure = "40001" + codeDeadlockDetected = "40P01" + codeTooManyConnections = "53300" + codeCannotConnectNow = "57P03" +) + +// Retry budget: short and bounded. A worker polling for tasks would rather get +// a fast error and poll again than have its request hang for half a minute. +const ( + retryInitialInterval = 50 * time.Millisecond + retryMaxInterval = 1 * time.Second + retryMaxElapsedTime = 5 * time.Second +) + +// isTransient reports whether err is worth retrying. +// +// The default is *not* to retry: a constraint violation or a syntax error will +// fail identically every time, and retrying it only multiplies the damage. +func isTransient(err error) bool { + if err == nil { + return false + } + // A cancelled caller does not want another attempt. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + switch pgErr.Code { + case codeSerializationFailure, codeDeadlockDetected, + codeTooManyConnections, codeCannotConnectNow: + return true + default: + return false + } + } + + // Connection-level trouble (dropped socket, closed pool). pgconn knows + // whether the query could have been executed before the failure — retrying + // a maybe-executed write would risk duplicating it. + return pgconn.SafeToRetry(err) +} + +// 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 +// 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() + b.InitialInterval = retryInitialInterval + b.MaxInterval = retryMaxInterval + b.MaxElapsedTime = retryMaxElapsedTime + // RandomizationFactor defaults to 0.5, which is the jitter. + + return backoff.Retry(func() error { + err := op(ctx) + if err == nil { + return nil + } + if !isTransient(err) { + return backoff.Permanent(err) // stop now, do not burn the budget + } + return err + }, backoff.WithContext(b, ctx)) +} diff --git a/coordinator/internal/storage/postgres/retry_test.go b/coordinator/internal/storage/postgres/retry_test.go new file mode 100644 index 0000000..bdf77ec --- /dev/null +++ b/coordinator/internal/storage/postgres/retry_test.go @@ -0,0 +1,98 @@ +package postgres + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +func TestIsTransient(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"serialization failure", &pgconn.PgError{Code: codeSerializationFailure}, true}, + {"deadlock", &pgconn.PgError{Code: codeDeadlockDetected}, true}, + {"too many connections", &pgconn.PgError{Code: codeTooManyConnections}, true}, + // A unique-violation repeats identically forever — retrying is pointless. + {"unique violation", &pgconn.PgError{Code: "23505"}, false}, + {"syntax error", &pgconn.PgError{Code: "42601"}, false}, + {"context cancelled", context.Canceled, false}, + {"deadline exceeded", context.DeadlineExceeded, false}, + {"unknown error", errors.New("boom"), false}, + // Wrapping must not hide the cause: errors.As walks the chain. + {"wrapped deadlock", errors2Wrap(&pgconn.PgError{Code: codeDeadlockDetected}), true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isTransient(tt.err); got != tt.want { + t.Errorf("isTransient(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +func errors2Wrap(err error) error { + return errors.Join(errors.New("query failed"), err) +} + +func TestWithRetrySucceedsAfterTransientFailures(t *testing.T) { + calls := 0 + err := withRetry(context.Background(), func(context.Context) error { + calls++ + if calls < 3 { + return &pgconn.PgError{Code: codeSerializationFailure} + } + return nil + }) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if calls != 3 { + t.Errorf("calls = %d, want 3", calls) + } +} + +func TestWithRetryStopsOnPermanentError(t *testing.T) { + permanent := &pgconn.PgError{Code: "23505"} // unique violation + calls := 0 + + err := withRetry(context.Background(), func(context.Context) error { + calls++ + return permanent + }) + + if !errors.Is(err, permanent) { + t.Errorf("err = %v, want the original error", err) + } + if calls != 1 { + t.Errorf("calls = %d, want 1 — a permanent error must not be retried", calls) + } +} + +func TestWithRetryHonoursContextCancellation(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + calls := 0 + start := time.Now() + err := withRetry(ctx, func(context.Context) error { + calls++ + return &pgconn.PgError{Code: codeDeadlockDetected} + }) + + if err == nil { + t.Fatal("expected an error once the context expired") + } + // Must abort at the deadline, not run the full 5s retry budget. + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("took %v, expected to stop at the context deadline", elapsed) + } +} diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go new file mode 100644 index 0000000..631a88b --- /dev/null +++ b/coordinator/internal/storage/postgres/task_repo.go @@ -0,0 +1,114 @@ +package postgres + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// TaskRepo implements usecase.TaskRepository. +type TaskRepo struct { + pool *pgxpool.Pool +} + +func NewTaskRepo(pool *pgxpool.Pool) *TaskRepo { + return &TaskRepo{pool: pool} +} + +var _ usecase.TaskRepository = (*TaskRepo)(nil) + +// claimNextSQL leases one task in a single statement. +// +// FOR UPDATE SKIP LOCKED is what makes concurrent coordinators safe: each +// process locks a different candidate row instead of queueing on the same one, +// so no task is ever handed to two workers and no claim blocks behind another. +// Splitting this into SELECT + UPDATE would reintroduce exactly that race. +// +//nolint:unused // wired up in phase 2; kept beside the repository it belongs to +const claimNextSQL = ` +WITH candidate AS ( + SELECT id + FROM tasks + WHERE status = 'pending' + AND attempt < max_attempts + AND (cardinality($1::text[]) = 0 OR workload = ANY($1)) + ORDER BY created_at, chunk_index + FOR UPDATE SKIP LOCKED + LIMIT 1 +) +UPDATE tasks +SET status = 'leased', + attempt = attempt + 1, + lease_owner = $2, + lease_expires_at = $3, + started_at = COALESCE(started_at, $4), + version = version + 1 +FROM candidate +WHERE tasks.id = candidate.id +RETURNING tasks.id, tasks.job_id, tasks.chunk_index, tasks.workload, + tasks.input_uri, tasks.input_sha256, tasks.parameters, + tasks.status, tasks.attempt, tasks.max_attempts, + tasks.lease_owner, tasks.lease_expires_at, tasks.version; +` + +// expireLeasesSQL applies the lease-expiry rule set-based, mirroring +// domain.Task.ExpireLease: requeue while attempts remain, otherwise fail. +// +//nolint:unused // wired up in phase 5; mirrors domain.Task.ExpireLease +const expireLeasesSQL = ` +UPDATE tasks +SET status = CASE WHEN attempt < max_attempts THEN 'pending'::task_status + ELSE 'failed'::task_status END, + lease_owner = NULL, + lease_expires_at = NULL, + error_code = CASE WHEN attempt >= max_attempts THEN $2 ELSE error_code END, + error_message = CASE WHEN attempt >= max_attempts + THEN 'lease expired after the final attempt' + ELSE error_message END, + completed_at = CASE WHEN attempt >= max_attempts THEN $1 ELSE completed_at END, + version = version + 1 +WHERE status = 'leased' AND lease_expires_at < $1; +` + +// TODO(phase 2-6): replace stubs with real pgx queries; the SQL above is ready +// to wire up. Each method maps 1:1 to a roadmap phase. + +func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domain.Task, error) { + // Phase 2: run claimNextSQL; pgx.ErrNoRows -> (nil, nil). + return nil, usecase.ErrNotImplemented +} + +func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) { + // Phase 3: SELECT ... WHERE id = $1 FOR UPDATE; no rows -> domain.ErrTaskNotFound. + return nil, usecase.ErrNotImplemented +} + +func (r *TaskRepo) Update(ctx context.Context, t *domain.Task) error { + // Phase 3: UPDATE ... WHERE id = $1 AND version = $2 (optimistic concurrency). + return usecase.ErrNotImplemented +} + +func (r *TaskRepo) InsertBatch(ctx context.Context, tasks []*domain.Task) error { + // Phase 2: pgx.Batch or COPY; runs inside the caller's transaction. + return usecase.ErrNotImplemented +} + +func (r *TaskRepo) ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error) { + // Phase 6: WHERE job_id = $1 AND status = 'completed' ORDER BY chunk_index. + return nil, usecase.ErrNotImplemented +} + +func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error) { + // Phase 3: SELECT status, count(*) ... GROUP BY status. + return nil, usecase.ErrNotImplemented +} + +func (r *TaskRepo) ExpireLeases(ctx context.Context, now time.Time) (int64, error) { + // Phase 5: run expireLeasesSQL, return the affected row count. + return 0, usecase.ErrNotImplemented +} diff --git a/coordinator/internal/storage/postgres/tx.go b/coordinator/internal/storage/postgres/tx.go new file mode 100644 index 0000000..72dfe9b --- /dev/null +++ b/coordinator/internal/storage/postgres/tx.go @@ -0,0 +1,69 @@ +// Package postgres implements the usecase repository ports on PostgreSQL. +// SQL and pgx types never escape this package. +package postgres + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +// querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting every +// repository method run identically inside or outside a transaction. +// +//nolint:unused // used by repository methods once phase 2 replaces the stubs +type querier interface { + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) +} + +// txKey is an unexported struct type, so no other package can collide with it +// or reach the transaction we stash in the context. +type txKey struct{} + +// TxManager implements usecase.TxManager. +type TxManager struct { + pool *pgxpool.Pool +} + +func NewTxManager(pool *pgxpool.Pool) *TxManager { + return &TxManager{pool: pool} +} + +// WithinTx runs fn inside one transaction, committing on success and rolling +// back on any error or panic. +// +// The transaction travels in the context rather than in fn's signature, which +// is what lets the usecase layer express "do these repository calls atomically" +// without its port ever mentioning pgx. +func (m *TxManager) WithinTx(ctx context.Context, fn func(ctx context.Context) error) error { + if _, ok := ctx.Value(txKey{}).(pgx.Tx); ok { + return fn(ctx) // already inside a transaction — join it, don't nest + } + + tx, err := m.pool.Begin(ctx) + if err != nil { + return err + } + // Rollback after a successful Commit is a no-op, so this defer is safe and + // also covers the panic path. + defer func() { _ = tx.Rollback(ctx) }() + + if err := fn(context.WithValue(ctx, txKey{}, tx)); err != nil { + return err + } + return tx.Commit(ctx) +} + +// conn returns the transaction bound to ctx, or the pool when there is none. +// +//nolint:unused // every repository method will route through this in phase 2 +func conn(ctx context.Context, pool *pgxpool.Pool) querier { + if tx, ok := ctx.Value(txKey{}).(pgx.Tx); ok { + return tx + } + return pool +} diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go new file mode 100644 index 0000000..b308843 --- /dev/null +++ b/coordinator/internal/transport/http/dto.go @@ -0,0 +1,119 @@ +package http + +import ( + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// Wire formats. Keeping them separate from domain entities means the API +// contract can evolve without reshaping the database, and nothing internal +// (version counters, other workers' errors) leaks by accident. + +type createJobRequest struct { + Workload string `json:"workload"` + InputURI string `json:"input_uri"` + Parameters map[string]any `json:"parameters"` + Chunks []chunkDTO `json:"chunks"` +} + +type chunkDTO struct { + ChunkIndex int `json:"chunk_index"` + Workload string `json:"workload"` + InputURI string `json:"input_uri"` + InputSHA256 string `json:"input_sha256"` + Parameters map[string]any `json:"parameters"` + MaxAttempts int `json:"max_attempts"` +} + +type claimRequest struct { + WorkerID string `json:"worker_id"` + Workloads []string `json:"workloads"` +} + +type heartbeatRequest struct { + WorkerID string `json:"worker_id"` + Attempt int `json:"attempt"` +} + +type resultRequest struct { + WorkerID string `json:"worker_id"` + Attempt int `json:"attempt"` + ResultURI string `json:"result_uri"` + ResultSHA256 string `json:"result_sha256"` + Metrics map[string]any `json:"metrics"` +} + +type failureRequest struct { + WorkerID string `json:"worker_id"` + Attempt int `json:"attempt"` + ErrorCode string `json:"error_code"` + ErrorMessage string `json:"error_message"` + Retryable bool `json:"retryable"` +} + +type jobResponse struct { + ID uuid.UUID `json:"id"` + Status string `json:"status"` +} + +type taskResponse struct { + ID uuid.UUID `json:"id"` + JobID uuid.UUID `json:"job_id"` + Status string `json:"status"` +} + +type claimedTaskResponse struct { + TaskID uuid.UUID `json:"task_id"` + JobID uuid.UUID `json:"job_id"` + ChunkIndex int `json:"chunk_index"` + Workload string `json:"workload"` + InputURI string `json:"input_uri"` + InputSHA256 string `json:"input_sha256"` + Parameters map[string]any `json:"parameters"` + Attempt int `json:"attempt"` + LeaseExpiresAt time.Time `json:"lease_expires_at"` +} + +type jobProgressResponse struct { + ID uuid.UUID `json:"id"` + Status string `json:"status"` + Total int `json:"total"` + Pending int `json:"pending"` + Leased int `json:"leased"` + Done int `json:"completed"` + Failed int `json:"failed"` +} + +type errorResponse struct { + Error string `json:"error"` + RequestID string `json:"request_id,omitempty"` +} + +func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse { + return claimedTaskResponse{ + TaskID: c.TaskID, + JobID: c.JobID, + ChunkIndex: c.ChunkIndex, + Workload: c.Workload, + InputURI: c.InputURI, + InputSHA256: c.InputSHA256, + Parameters: c.Parameters, + Attempt: c.Attempt, + LeaseExpiresAt: c.LeaseExpiresAt, + } +} + +func toJobProgressResponse(p domain.JobProgress) jobProgressResponse { + return jobProgressResponse{ + ID: p.Job.ID, + Status: string(p.DeriveStatus()), + Total: p.Total, + Pending: p.Pending, + Leased: p.Leased, + Done: p.Done, + Failed: p.Failed, + } +} diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go new file mode 100644 index 0000000..d052a90 --- /dev/null +++ b/coordinator/internal/transport/http/errors.go @@ -0,0 +1,55 @@ +package http + +import ( + "encoding/json" + "errors" + "net/http" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +func decodeJSON(r *http.Request, dst any) error { + dec := json.NewDecoder(r.Body) + // Reject unknown fields: silently ignoring a misspelled "worker_ID" would + // surface later as a baffling validation failure. + dec.DisallowUnknownFields() + return dec.Decode(dst) +} + +// writeError translates domain errors into status codes. This mapping is the +// only place in the codebase that knows HTTP status codes exist — the inner +// layers speak only in business terms. +func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) { + reqID := requestIDFrom(r.Context()) + + status := http.StatusInternalServerError + switch { + case errors.Is(err, domain.ErrInvalidInput): + status = http.StatusBadRequest + case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound): + status = http.StatusNotFound + case errors.Is(err, domain.ErrLeaseConflict), + errors.Is(err, domain.ErrStaleAttempt), + errors.Is(err, domain.ErrResultConflict), + errors.Is(err, domain.ErrTaskNotLeased): + status = http.StatusConflict + case errors.Is(err, usecase.ErrNotImplemented): + status = http.StatusNotImplemented + } + + if status >= 500 { + // Never echo an internal error: it can carry table names, query + // fragments, and values. The request ID is the bridge to the logs. + s.log.Error("request failed", "request_id", reqID, "path", r.URL.Path, "err", err) + writeJSON(w, status, errorResponse{Error: "internal error", RequestID: reqID}) + return + } + writeJSON(w, status, errorResponse{Error: err.Error(), RequestID: reqID}) +} diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go new file mode 100644 index 0000000..00e4f0e --- /dev/null +++ b/coordinator/internal/transport/http/handlers.go @@ -0,0 +1,181 @@ +package http + +import ( + "context" + "net/http" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// Every handler follows the same shape: decode, map to a use-case input, +// execute, translate. Anything resembling a rule belongs one layer inward. + +func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + var req createJobRequest + if err := decodeJSON(r, &req); err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + in := usecase.CreateJobInput{ + Workload: req.Workload, + InputURI: req.InputURI, + Parameters: req.Parameters, + } + for _, c := range req.Chunks { + in.Chunks = append(in.Chunks, usecase.ChunkInput(c)) + } + + job, err := s.uc.CreateJob.Execute(ctx, in) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusCreated, jobResponse{ID: job.ID, Status: string(job.Status)}) +} + +func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + var req claimRequest + if err := decodeJSON(r, &req); err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + claimed, err := s.uc.ClaimTask.Execute(ctx, usecase.ClaimTaskInput{ + WorkerID: req.WorkerID, + Workloads: req.Workloads, + }) + if err != nil { + s.writeError(w, r, err) + return + } + if claimed == nil { + w.WriteHeader(http.StatusNoContent) // empty queue, not an error + return + } + writeJSON(w, http.StatusOK, toClaimedTaskResponse(*claimed)) +} + +func (s *Server) handleHeartbeat(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + taskID, ok := s.pathUUID(w, r, "task_id") + if !ok { + return + } + var req heartbeatRequest + if err := decodeJSON(r, &req); err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + claimed, err := s.uc.RenewLease.Execute(ctx, usecase.RenewLeaseInput{ + TaskID: taskID, + WorkerID: req.WorkerID, + Attempt: req.Attempt, + }) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, toClaimedTaskResponse(*claimed)) +} + +func (s *Server) handleResult(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + taskID, ok := s.pathUUID(w, r, "task_id") + if !ok { + return + } + var req resultRequest + if err := decodeJSON(r, &req); err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + task, err := s.uc.CompleteTask.Execute(ctx, usecase.CompleteTaskInput{ + TaskID: taskID, + WorkerID: req.WorkerID, + Attempt: req.Attempt, + ResultURI: req.ResultURI, + ResultSHA256: req.ResultSHA256, + Metrics: req.Metrics, + }) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)}) +} + +func (s *Server) handleFailure(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + taskID, ok := s.pathUUID(w, r, "task_id") + if !ok { + return + } + var req failureRequest + if err := decodeJSON(r, &req); err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + task, err := s.uc.FailTask.Execute(ctx, usecase.FailTaskInput{ + TaskID: taskID, + WorkerID: req.WorkerID, + Attempt: req.Attempt, + ErrorCode: req.ErrorCode, + ErrorMessage: req.ErrorMessage, + Retryable: req.Retryable, + }) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)}) +} + +func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + + jobID, ok := s.pathUUID(w, r, "job_id") + if !ok { + return + } + progress, err := s.uc.GetJobStatus.Execute(ctx, jobID) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, toJobProgressResponse(progress)) +} + +// --- helpers --- + +func (s *Server) reqCtx(r *http.Request) (context.Context, context.CancelFunc) { + return context.WithTimeout(r.Context(), s.requestTimeout) +} + +func (s *Server) pathUUID(w http.ResponseWriter, r *http.Request, name string) (uuid.UUID, bool) { + id, err := uuid.Parse(r.PathValue(name)) + if err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return uuid.Nil, false + } + return id, true +} diff --git a/coordinator/internal/transport/http/middleware.go b/coordinator/internal/transport/http/middleware.go new file mode 100644 index 0000000..feb3f4f --- /dev/null +++ b/coordinator/internal/transport/http/middleware.go @@ -0,0 +1,103 @@ +package http + +import ( + "context" + "crypto/rand" + "crypto/subtle" + "encoding/hex" + "log/slog" + "net/http" + "strings" + "time" +) + +type ctxKey string + +const requestIDKey ctxKey = "request_id" + +// withRequestID stamps every request with an ID for correlated logs and error +// bodies. It wraps the auth middleware rather than the other way round, so even +// a rejected request carries an ID the caller can quote in a bug report. +func withRequestID(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id := newRequestID() + w.Header().Set("X-Request-ID", id) + next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), requestIDKey, id))) + }) +} + +func requestIDFrom(ctx context.Context) string { + if v, ok := ctx.Value(requestIDKey).(string); ok { + return v + } + return "" +} + +func newRequestID() string { + var b [8]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} + +// withAuth enforces the shared bearer token every worker presents. +// An empty token disables the check (local development only). +func withAuth(token string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if token == "" { + next.ServeHTTP(w, r) + return + } + presented := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + // Constant-time compare: a byte-by-byte early exit would let an + // attacker recover the token by timing responses. + if subtle.ConstantTimeCompare([]byte(presented), []byte(token)) != 1 { + w.Header().Set("WWW-Authenticate", "Bearer") + writeJSON(w, http.StatusUnauthorized, errorResponse{ + Error: "unauthorized", + RequestID: requestIDFrom(r.Context()), + }) + return + } + next.ServeHTTP(w, r) + }) + } +} + +// statusRecorder captures the status code for the access log. +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (s *statusRecorder) WriteHeader(code int) { + s.status = code + s.ResponseWriter.WriteHeader(code) +} + +// withAccessLog records one structured line per request — the minimum needed to +// debug a distributed system after the fact. +func withAccessLog(log *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} + next.ServeHTTP(rec, r) + log.Info("request", + "request_id", requestIDFrom(r.Context()), + "method", r.Method, + "path", r.URL.Path, + "status", rec.status, + "duration_ms", time.Since(start).Milliseconds(), + ) + }) + } +} + +// chain applies middleware so that the first argument is the outermost layer. +func chain(h http.Handler, mw ...func(http.Handler) http.Handler) http.Handler { + for i := len(mw) - 1; i >= 0; i-- { + h = mw[i](h) + } + return h +} diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go new file mode 100644 index 0000000..d2731a3 --- /dev/null +++ b/coordinator/internal/transport/http/server.go @@ -0,0 +1,60 @@ +// Package http adapts the use-case layer to HTTP. Handlers decode requests, +// map them onto use-case inputs, and translate results and errors back — no +// business rules live here. +package http + +import ( + "log/slog" + "net/http" + "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +// UseCases collects everything the transport needs. Depending on concrete +// use-case types (not one fat interface) keeps each handler's dependency +// explicit and the wiring visible in the composition root. +type UseCases struct { + CreateJob *usecase.CreateJob + ClaimTask *usecase.ClaimTask + RenewLease *usecase.RenewLease + CompleteTask *usecase.CompleteTask + FailTask *usecase.FailTask + GetJobStatus *usecase.GetJobStatus +} + +type Server struct { + uc UseCases + log *slog.Logger + requestTimeout time.Duration +} + +func NewServer(uc UseCases, log *slog.Logger, requestTimeout time.Duration) *Server { + return &Server{uc: uc, log: log, requestTimeout: requestTimeout} +} + +// Handler builds the router. Go 1.22's ServeMux matches on method and path +// wildcards, so no third-party router is needed. +func (s *Server) Handler(token string) http.Handler { + protected := http.NewServeMux() + protected.HandleFunc("POST /jobs", s.handleCreateJob) + protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob) + protected.HandleFunc("POST /tasks/claim", s.handleClaim) + protected.HandleFunc("POST /tasks/{task_id}/heartbeat", s.handleHeartbeat) + protected.HandleFunc("POST /tasks/{task_id}/result", s.handleResult) + protected.HandleFunc("POST /tasks/{task_id}/failure", s.handleFailure) + + mux := http.NewServeMux() + // A more specific pattern wins, so /health stays outside the auth wall. + mux.HandleFunc("GET /health", s.handleHealth) + mux.Handle("/", chain(protected, + withRequestID, // outermost: every response gets an ID, + withAccessLog(s.log), // including the 401s below + withAuth(token), + )) + return mux +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go new file mode 100644 index 0000000..d7cb76c --- /dev/null +++ b/coordinator/internal/usecase/dto.go @@ -0,0 +1,51 @@ +package usecase + +import "github.com/google/uuid" + +// Use-case boundary types. Adapters map their wire formats onto these, so the +// HTTP shape can change without touching business code. + +type CreateJobInput struct { + Workload string + InputURI string + Parameters map[string]any + Chunks []ChunkInput +} + +type ChunkInput struct { + ChunkIndex int + Workload string + InputURI string + InputSHA256 string + Parameters map[string]any + MaxAttempts int +} + +type ClaimTaskInput struct { + WorkerID string + Workloads []string +} + +type RenewLeaseInput struct { + TaskID uuid.UUID + WorkerID string + Attempt int +} + +type CompleteTaskInput struct { + TaskID uuid.UUID + WorkerID string + Attempt int + ResultURI string + ResultSHA256 string + Metrics map[string]any +} + +type FailTaskInput struct { + TaskID uuid.UUID + WorkerID string + Attempt int + ErrorCode string + ErrorMessage string + Retryable bool +} diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go new file mode 100644 index 0000000..1fc455f --- /dev/null +++ b/coordinator/internal/usecase/job.go @@ -0,0 +1,174 @@ +package usecase + +import ( + "context" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// Job operations: the submitter-facing lifecycle of a whole submission. +// +// CreateJob register a job and fan it out into tasks +// GetJobStatus aggregate progress +// ListResults completed manifests, ordered for the stitcher +// StitchJob merge partial results into the final artifact + +// --- CreateJob ----------------------------------------------------------- + +type CreateJob struct { + jobs JobRepository + tasks TaskRepository + tx TxManager + clock Clock +} + +func NewCreateJob(jobs JobRepository, tasks TaskRepository, tx TxManager, clock Clock) *CreateJob { + return &CreateJob{jobs: jobs, tasks: tasks, tx: tx, clock: clock} +} + +// Execute builds the job and its tasks, then writes them in one transaction. +// The all-or-none guarantee comes from TxManager: a half-created job would +// leave chunks no worker could ever complete. +func (uc *CreateJob) Execute(ctx context.Context, in CreateJobInput) (*domain.Job, error) { + chunks := make([]domain.ChunkSpec, 0, len(in.Chunks)) + for _, c := range in.Chunks { + chunks = append(chunks, domain.ChunkSpec(c)) + } + + job, tasks, err := domain.NewJobWithTasks(in.Workload, in.InputURI, in.Parameters, chunks, uc.clock.Now()) + if err != nil { + return nil, err + } + + err = uc.tx.WithinTx(ctx, func(ctx context.Context) error { + if err := uc.jobs.Insert(ctx, job); err != nil { + return err + } + return uc.tasks.InsertBatch(ctx, tasks) + }) + if err != nil { + return nil, err + } + return job, nil +} + +// --- GetJobStatus -------------------------------------------------------- + +type GetJobStatus struct { + jobs JobRepository + tasks TaskRepository +} + +func NewGetJobStatus(jobs JobRepository, tasks TaskRepository) *GetJobStatus { + return &GetJobStatus{jobs: jobs, tasks: tasks} +} + +func (uc *GetJobStatus) Execute(ctx context.Context, jobID uuid.UUID) (domain.JobProgress, error) { + job, err := uc.jobs.Get(ctx, jobID) + if err != nil { + return domain.JobProgress{}, err + } + counts, err := uc.tasks.CountByStatus(ctx, jobID) + if err != nil { + return domain.JobProgress{}, err + } + return progressFrom(*job, counts), nil +} + +// --- ListResults --------------------------------------------------------- + +type ListResults struct { + tasks TaskRepository +} + +func NewListResults(tasks TaskRepository) *ListResults { + return &ListResults{tasks: tasks} +} + +// Execute preserves chunk_index order: the stitcher merges these into one +// artifact, and a non-deterministic order would make the final result depend on +// which worker happened to finish first. +func (uc *ListResults) Execute(ctx context.Context, jobID uuid.UUID) ([]domain.ResultManifest, error) { + tasks, err := uc.tasks.ListCompleted(ctx, jobID) + if err != nil { + return nil, err + } + + manifests := make([]domain.ResultManifest, 0, len(tasks)) + for _, t := range tasks { + if t.ResultURI == nil || t.ResultSHA256 == nil { + continue // a completed task always carries both; skip defensively + } + manifests = append(manifests, domain.ResultManifest{ + TaskID: t.ID, + ChunkIndex: t.ChunkIndex, + ResultURI: *t.ResultURI, + ResultSHA256: *t.ResultSHA256, + Metrics: t.Metrics, + }) + } + return manifests, nil +} + +// --- StitchJob ----------------------------------------------------------- + +// StitchJob merges every chunk's partial result into the job's final artifact. +// For similarity search that means concatenating each worker's local top-k, +// sorting by similarity, and keeping the global top-k — the distributed result +// must match what a single local run would produce. +type StitchJob struct { + results *ListResults +} + +func NewStitchJob(results *ListResults) *StitchJob { + return &StitchJob{results: results} +} + +// Execute returns the URI of the assembled artifact. +// +// TODO(phase 6): fetch each manifest's CSV, merge, and persist the result. +func (uc *StitchJob) Execute(ctx context.Context, jobID uuid.UUID) (string, error) { + if _, err := uc.results.Execute(ctx, jobID); err != nil { + return "", err + } + return "", ErrNotImplemented +} + +// --- shared helpers ------------------------------------------------------ + +// progressFrom turns a status histogram into the domain's progress view. +func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobProgress { + p := domain.JobProgress{ + Job: job, + Pending: counts[domain.TaskPending], + Leased: counts[domain.TaskLeased], + Done: counts[domain.TaskCompleted], + Failed: counts[domain.TaskFailed], + } + for _, n := range counts { + p.Total += n + } + return p +} + +// syncJobStatus recomputes a job's status from its task counts and persists it. +// Shared by CompleteTask and FailTask so both close a job by the same rule — +// the rule itself lives in domain.JobProgress.DeriveStatus. +func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository, + jobID uuid.UUID, now time.Time) error { + + counts, err := tasks.CountByStatus(ctx, jobID) + if err != nil { + return err + } + status := progressFrom(domain.Job{}, counts).DeriveStatus() + + var completedAt *time.Time + if status == domain.JobCompleted || status == domain.JobFailed { + completedAt = &now + } + return jobs.UpdateStatus(ctx, jobID, status, completedAt) +} diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go new file mode 100644 index 0000000..af0e06d --- /dev/null +++ b/coordinator/internal/usecase/ports.go @@ -0,0 +1,80 @@ +// Package usecase holds the application's business operations. Each use case is +// a small type with its dependencies injected and a single Execute method. +// +// The interfaces below are *ports*: they are declared here, by the consumer, +// and implemented further out in storage/postgres. That is what keeps the +// dependency rule intact — usecase never imports storage or transport. +package usecase + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// ClaimFilter narrows which task a worker may be handed. +type ClaimFilter struct { + Workloads []string // workloads this worker can execute + Owner string // worker ID taking the lease + Now time.Time + LeaseUntil time.Time +} + +// TaskRepository persists tasks. +// +// ClaimNext is deliberately coarse: leasing must be a single atomic statement +// (SELECT ... FOR UPDATE SKIP LOCKED + UPDATE), so it cannot be decomposed into +// Get+Update without losing the guarantee that one task goes to one worker. +type TaskRepository interface { + // ClaimNext atomically leases one matching pending task. + // Returns (nil, nil) when nothing is available. + ClaimNext(ctx context.Context, f ClaimFilter) (*domain.Task, error) + + // GetForUpdate reads a task and locks its row for the enclosing + // transaction, so read-modify-write use cases stay serialized. + GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) + + // Update persists a mutated task, honouring its Version for optimistic + // concurrency. + Update(ctx context.Context, t *domain.Task) error + + InsertBatch(ctx context.Context, tasks []*domain.Task) error + + // ListCompleted returns completed tasks ordered by chunk_index. + ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error) + + // CountByStatus aggregates a job's tasks for progress reporting. + CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error) + + // ExpireLeases applies the lease-expiry rule to every elapsed task and + // reports how many were affected. + ExpireLeases(ctx context.Context, now time.Time) (int64, error) +} + +// JobRepository persists jobs. +type JobRepository interface { + Insert(ctx context.Context, j *domain.Job) error + Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) + UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error +} + +// TxManager runs a function inside one database transaction. The transaction +// travels in the context, so repositories pick it up without this port ever +// mentioning pgx. +type TxManager interface { + WithinTx(ctx context.Context, fn func(ctx context.Context) error) error +} + +// Clock supplies the current time. Injecting it keeps lease and expiry rules +// testable without sleeping or freezing the system clock. +type Clock interface { + Now() time.Time +} + +// ErrNotImplemented marks scaffold code with no body yet. Unlike the errors in +// domain, it describes the state of this codebase, not a business rule. +var ErrNotImplemented = errors.New("not implemented") diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go new file mode 100644 index 0000000..c9dc801 --- /dev/null +++ b/coordinator/internal/usecase/task.go @@ -0,0 +1,206 @@ +package usecase + +import ( + "context" + "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// Task operations: the worker-facing lifecycle of a single chunk. +// +// ClaimTask lease the next available task +// RenewLease extend a held lease (heartbeat) +// CompleteTask record a successful result +// FailTask record a failure +// ExpireLeases reclaim leases that elapsed without a heartbeat + +// --- ClaimTask ----------------------------------------------------------- + +type ClaimTask struct { + tasks TaskRepository + clock Clock + leaseDuration time.Duration +} + +func NewClaimTask(tasks TaskRepository, clock Clock, leaseDuration time.Duration) *ClaimTask { + return &ClaimTask{tasks: tasks, clock: clock, leaseDuration: leaseDuration} +} + +// Execute reclaims elapsed leases first, then hands out one task. +// +// Sweeping before claiming matters: otherwise a task abandoned by a dead worker +// stays invisible until the reaper's next tick, and a waiting worker is told the +// queue is empty while work sits idle. +// +// This use case is thin by design — the atomicity that makes claiming correct +// lives in one SQL statement behind ClaimNext, and splitting it across the layer +// boundary would break it. +func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.ClaimedTask, error) { + if in.WorkerID == "" { + return nil, domain.ErrInvalidInput + } + now := uc.clock.Now() + + if _, err := uc.tasks.ExpireLeases(ctx, now); err != nil { + return nil, err + } + + task, err := uc.tasks.ClaimNext(ctx, ClaimFilter{ + Workloads: in.Workloads, + Owner: in.WorkerID, + Now: now, + LeaseUntil: now.Add(uc.leaseDuration), + }) + if err != nil { + return nil, err + } + if task == nil { + return nil, nil // empty queue is a normal state, not an error + } + + claimed := task.AsClaimed() + return &claimed, nil +} + +// --- RenewLease ---------------------------------------------------------- + +type RenewLease struct { + tasks TaskRepository + tx TxManager + clock Clock + leaseDuration time.Duration +} + +func NewRenewLease(tasks TaskRepository, tx TxManager, clock Clock, leaseDuration time.Duration) *RenewLease { + return &RenewLease{tasks: tasks, tx: tx, clock: clock, leaseDuration: leaseDuration} +} + +// Execute is a read-modify-write, so it runs inside a transaction with the row +// locked: two concurrent heartbeats must not interleave into a lost update. +// Whether the caller may renew at all is decided by the entity, not here. +func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain.ClaimedTask, error) { + var claimed domain.ClaimedTask + + err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { + task, err := uc.tasks.GetForUpdate(ctx, in.TaskID) + if err != nil { + return err + } + if err := task.RenewLease(in.WorkerID, in.Attempt, uc.clock.Now().Add(uc.leaseDuration)); err != nil { + return err + } + if err := uc.tasks.Update(ctx, task); err != nil { + return err + } + claimed = task.AsClaimed() + return nil + }) + if err != nil { + return nil, err + } + return &claimed, nil +} + +// --- CompleteTask -------------------------------------------------------- + +type CompleteTask struct { + tasks TaskRepository + jobs JobRepository + tx TxManager + clock Clock +} + +func NewCompleteTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *CompleteTask { + return &CompleteTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock} +} + +// Execute applies the result and, when that was the job's last outstanding +// task, closes the job in the same transaction — so a caller who sees a +// completed task never observes its job still marked running. +// +// Lease ownership, staleness, and idempotent replays are all decided by +// Task.CompleteWith; this use case only orchestrates. +func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*domain.Task, error) { + var out *domain.Task + + err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { + task, err := uc.tasks.GetForUpdate(ctx, in.TaskID) + if err != nil { + return err + } + now := uc.clock.Now() + if err := task.CompleteWith(in.ResultURI, in.ResultSHA256, in.Metrics, + in.WorkerID, in.Attempt, now); err != nil { + return err + } + if err := uc.tasks.Update(ctx, task); err != nil { + return err + } + out = task + return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) + }) + if err != nil { + return nil, err + } + return out, nil +} + +// --- FailTask ------------------------------------------------------------ + +type FailTask struct { + tasks TaskRepository + jobs JobRepository + tx TxManager + clock Clock +} + +func NewFailTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *FailTask { + return &FailTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock} +} + +// Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps +// the parent job's status consistent in the same transaction. +func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task, error) { + var out *domain.Task + + err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { + task, err := uc.tasks.GetForUpdate(ctx, in.TaskID) + if err != nil { + return err + } + now := uc.clock.Now() + if err := task.Fail(in.WorkerID, in.Attempt, in.ErrorCode, in.ErrorMessage, in.Retryable, now); err != nil { + return err + } + if err := uc.tasks.Update(ctx, task); err != nil { + return err + } + out = task + return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) + }) + if err != nil { + return nil, err + } + return out, nil +} + +// --- ExpireLeases -------------------------------------------------------- + +type ExpireLeases struct { + tasks TaskRepository + clock Clock +} + +func NewExpireLeases(tasks TaskRepository, clock Clock) *ExpireLeases { + return &ExpireLeases{tasks: tasks, clock: clock} +} + +// Execute reports how many tasks were reclaimed. +// +// The sweep is one set-based statement rather than a load-decide-save loop: +// several coordinators run it concurrently, and a single atomic UPDATE makes +// the duplicate work harmless — the loser simply updates 0 rows. +func (uc *ExpireLeases) Execute(ctx context.Context) (int64, error) { + return uc.tasks.ExpireLeases(ctx, uc.clock.Now()) +} diff --git a/coordinator/migrations/0001_init.down.sql b/coordinator/migrations/0001_init.down.sql new file mode 100644 index 0000000..78df969 --- /dev/null +++ b/coordinator/migrations/0001_init.down.sql @@ -0,0 +1,8 @@ +BEGIN; + +DROP TABLE IF EXISTS tasks; +DROP TABLE IF EXISTS jobs; +DROP TYPE IF EXISTS task_status; +DROP TYPE IF EXISTS job_status; + +COMMIT; diff --git a/coordinator/migrations/0001_init.up.sql b/coordinator/migrations/0001_init.up.sql new file mode 100644 index 0000000..c84c469 --- /dev/null +++ b/coordinator/migrations/0001_init.up.sql @@ -0,0 +1,58 @@ +BEGIN; + +CREATE TYPE job_status AS ENUM ('pending','running','completed','failed','cancelled'); +CREATE TYPE task_status AS ENUM ('pending','leased','completed','failed','cancelled'); + +-- One user submission, possibly split into several tasks. +CREATE TABLE jobs ( + id uuid PRIMARY KEY, + workload text NOT NULL, + input_uri text NOT NULL, + parameters jsonb NOT NULL DEFAULT '{}'::jsonb, + status job_status NOT NULL DEFAULT 'pending', + created_at timestamptz NOT NULL DEFAULT now(), + completed_at timestamptz +); + +-- One independently executable chunk. +CREATE TABLE tasks ( + id uuid PRIMARY KEY, + job_id uuid NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + chunk_index integer NOT NULL, + workload text NOT NULL, + input_uri text NOT NULL, + input_sha256 text NOT NULL, + parameters jsonb NOT NULL DEFAULT '{}'::jsonb, + status task_status NOT NULL DEFAULT 'pending', + attempt integer NOT NULL DEFAULT 0, + max_attempts integer NOT NULL DEFAULT 3, + lease_owner text, + lease_expires_at timestamptz, + result_uri text, + result_sha256 text, + metrics jsonb, + error_code text, + error_message text, + created_at timestamptz NOT NULL DEFAULT now(), + started_at timestamptz, + completed_at timestamptz, + version integer NOT NULL DEFAULT 0, + + CONSTRAINT uq_tasks_job_chunk UNIQUE (job_id, chunk_index), + CONSTRAINT ck_tasks_attempt CHECK (attempt >= 0), + CONSTRAINT ck_tasks_max_attempts CHECK (max_attempts > 0), + -- A completed task must carry its result manifest. + CONSTRAINT ck_tasks_completed_result CHECK ( + status <> 'completed' OR (result_uri IS NOT NULL AND result_sha256 IS NOT NULL) + ), + -- A leased task must carry its lease. + CONSTRAINT ck_tasks_leased_owner CHECK ( + status <> 'leased' OR (lease_owner IS NOT NULL AND lease_expires_at IS NOT NULL) + ) +); + +-- Claim path: find the oldest pending task fast. +CREATE INDEX ix_tasks_claim ON tasks (status, lease_expires_at, created_at); +CREATE INDEX ix_tasks_job ON tasks (job_id); + +COMMIT; From f1c3163be4315ae0261c01ac927280e79498358f Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Wed, 22 Jul 2026 14:48:58 +0300 Subject: [PATCH 02/17] feat(coordinator): implement PostgreSQL repositories MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the repository stubs with real pgx queries, so the queue now works end to end: a job is split into tasks, leased to workers one at a time, heartbeated, completed, and reflected in job progress. Task claiming is a single statement — SELECT ... FOR UPDATE SKIP LOCKED feeding an UPDATE — so concurrent coordinators lease different rows instead of blocking on the same one. Writes use optimistic concurrency: the entity increments its version in memory, and the UPDATE guards on the previous value. Retries moved to the transaction level. Once Postgres aborts a transaction with a serialization failure, replaying one statement inside it cannot help; the unit of retry is Begin -> fn -> Commit, which is safe because each attempt re-reads its rows through GetForUpdate. Adds integration tests behind the `integration` build tag, run against a real PostgreSQL through TEST_DATABASE_URL: concurrent claiming hands each task to exactly one worker, job creation rolls back whole, stale writes are refused, completed results keep chunk order, and expired leases return to the queue. Two bugs the tests caught: - a nil parameters map reached a NOT NULL jsonb column as SQL NULL, since pgx sends NULL rather than omitting the column and letting DEFAULT '{}' apply; - replaying an already-recorded result returned 409. The idempotent path leaves the entity untouched, so the version guard matched nothing and a successful no-op looked like a conflict. CompleteTask now skips the write when the entity did not change. --- coordinator/Makefile | 9 +- coordinator/README.md | 19 +- .../storage/postgres/integration_test.go | 360 ++++++++++++++++++ .../internal/storage/postgres/job_repo.go | 49 ++- .../internal/storage/postgres/task_repo.go | 257 ++++++++++--- coordinator/internal/storage/postgres/tx.go | 34 +- coordinator/internal/usecase/task.go | 11 +- 7 files changed, 668 insertions(+), 71 deletions(-) create mode 100644 coordinator/internal/storage/postgres/integration_test.go diff --git a/coordinator/Makefile b/coordinator/Makefile index d85c19e..38e4b54 100644 --- a/coordinator/Makefile +++ b/coordinator/Makefile @@ -10,6 +10,11 @@ run: test: go test ./... +# Needs a running PostgreSQL; the spec forbids mocks for these guarantees. +# make test-integration TEST_DATABASE_URL='postgres://...' +test-integration: + TEST_DATABASE_URL="$(TEST_DATABASE_URL)" go test -tags=integration ./... -v + vet: go vet ./... @@ -18,8 +23,8 @@ vet: LINT_VERSION := v2.12.2 lint: @command -v golangci-lint >/dev/null 2>&1 \ - && golangci-lint run ./... \ - || go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(LINT_VERSION) run ./... + && golangci-lint run --build-tags=integration ./... \ + || go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(LINT_VERSION) run --build-tags=integration ./... tidy: go mod tidy diff --git a/coordinator/README.md b/coordinator/README.md index f83a9be..ef54ec1 100644 --- a/coordinator/README.md +++ b/coordinator/README.md @@ -113,20 +113,21 @@ See `.env.example`; only `DATABASE_URL` is required. ## Status -Scaffold with a **complete, tested domain**. Layers, wiring, routing, auth, -access logging, error mapping, transactions, migrations, and graceful shutdown -are in place. Repository methods are stubs returning `ErrNotImplemented` -(→ HTTP 501); the SQL for claiming and lease expiry is written and ready to wire. +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. Roadmap: 1. schema + migrations ✅ -2. `ClaimNext`, `InsertBatch` — atomic claim via `FOR UPDATE SKIP LOCKED` -3. `GetForUpdate`, `Update`, `CountByStatus` — completes the result/failure paths -4. file upload / chunk download -5. `ExpireLeases` — SQL is written, needs wiring +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. integration tests against real Postgres via `TEST_DATABASE_URL` +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. ## Tests diff --git a/coordinator/internal/storage/postgres/integration_test.go b/coordinator/internal/storage/postgres/integration_test.go new file mode 100644 index 0000000..d6ec8a7 --- /dev/null +++ b/coordinator/internal/storage/postgres/integration_test.go @@ -0,0 +1,360 @@ +//go:build integration + +// Integration tests run against a real PostgreSQL instance supplied through +// TEST_DATABASE_URL. The spec forbids mocks or SQLite here: the guarantees +// being verified — FOR UPDATE SKIP LOCKED, optimistic concurrency, transaction +// rollback — are properties of Postgres, not of our Go code. +// +// docker compose up -d +// TEST_DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable' \ +// go test -tags=integration ./internal/storage/postgres/ -v +package postgres + +import ( + "context" + "errors" + "fmt" + "os" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/usecase" +) + +func testPool(t *testing.T) *pgxpool.Pool { + t.Helper() + url := os.Getenv("TEST_DATABASE_URL") + if url == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + pool, err := pgxpool.New(context.Background(), url) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(pool.Close) + return pool +} + +// seedJob creates a job with n pending tasks and removes them afterwards, so +// tests stay independent of each other and of leftovers from earlier runs. +func seedJob(t *testing.T, pool *pgxpool.Pool, n int) (*domain.Job, []*domain.Task) { + t.Helper() + ctx := context.Background() + + chunks := make([]domain.ChunkSpec, 0, n) + for i := 0; i < n; i++ { + chunks = append(chunks, domain.ChunkSpec{ + ChunkIndex: i, + InputURI: fmt.Sprintf("s3://chunk-%d", i), + InputSHA256: fmt.Sprintf("sha-%d", i), + }) + } + job, tasks, err := domain.NewJobWithTasks("similarity_search", "s3://ds", nil, chunks, time.Now().UTC()) + if err != nil { + t.Fatalf("build job: %v", err) + } + + jobs, taskRepo, tx := NewJobRepo(pool), NewTaskRepo(pool), NewTxManager(pool) + err = tx.WithinTx(ctx, func(ctx context.Context) error { + if err := jobs.Insert(ctx, job); err != nil { + return err + } + return taskRepo.InsertBatch(ctx, tasks) + }) + if err != nil { + t.Fatalf("seed: %v", err) + } + + t.Cleanup(func() { + // ON DELETE CASCADE removes the tasks with it. + _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, job.ID) + }) + return job, tasks +} + +func TestCreateJobPersistsEveryTask(t *testing.T) { + pool := testPool(t) + job, _ := seedJob(t, pool, 3) + + counts, err := NewTaskRepo(pool).CountByStatus(context.Background(), job.ID) + if err != nil { + t.Fatalf("count: %v", err) + } + if counts[domain.TaskPending] != 3 { + t.Errorf("pending = %d, want 3", counts[domain.TaskPending]) + } +} + +// A job must land whole or not at all: a half-created job leaves chunks no +// worker could ever complete. +func TestCreateJobRollsBackOnFailure(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + + chunks := []domain.ChunkSpec{{ChunkIndex: 0, InputURI: "s3://c0", InputSHA256: "sha0"}} + job, tasks, err := domain.NewJobWithTasks("similarity_search", "s3://ds", nil, chunks, time.Now().UTC()) + if err != nil { + t.Fatal(err) + } + + jobs, taskRepo, tx := NewJobRepo(pool), NewTaskRepo(pool), NewTxManager(pool) + boom := errors.New("boom") + err = tx.WithinTx(ctx, func(ctx context.Context) error { + if err := jobs.Insert(ctx, job); err != nil { + return err + } + if err := taskRepo.InsertBatch(ctx, tasks); err != nil { + return err + } + return boom // fail after both writes + }) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } + + if _, err := jobs.Get(ctx, job.ID); !errors.Is(err, domain.ErrJobNotFound) { + t.Errorf("job survived the rollback: %v", err) + } +} + +// The acceptance criterion: N workers claiming at once must each get a +// different task, and no task may be handed out twice. +func TestConcurrentClaimGivesEachTaskToExactlyOneWorker(t *testing.T) { + pool := testPool(t) + const tasks = 8 + job, _ := seedJob(t, pool, tasks) + + repo := NewTaskRepo(pool) + now := time.Now().UTC() + + var ( + mu sync.Mutex + claimed = make(map[uuid.UUID]string) + wg sync.WaitGroup + ) + // More workers than tasks, so the surplus must come back empty rather than + // steal an already-leased row. + for i := 0; i < tasks*2; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{ + Owner: fmt.Sprintf("worker-%d", n), + Now: now, + LeaseUntil: now.Add(time.Minute), + }) + if err != nil { + t.Errorf("claim: %v", err) + return + } + if task == nil || task.JobID != job.ID { + return // empty queue, or a task from another test's job + } + mu.Lock() + defer mu.Unlock() + if prev, dup := claimed[task.ID]; dup { + t.Errorf("task %s handed to both %s and worker-%d", task.ID, prev, n) + } + claimed[task.ID] = fmt.Sprintf("worker-%d", n) + }(i) + } + wg.Wait() + + if len(claimed) != tasks { + t.Errorf("claimed %d tasks, want %d", len(claimed), tasks) + } +} + +func TestClaimNextReturnsNilOnEmptyQueue(t *testing.T) { + pool := testPool(t) + now := time.Now().UTC() + + // Drain everything first, then ask once more. + repo := NewTaskRepo(pool) + for { + task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{ + Owner: "drainer", Now: now, LeaseUntil: now.Add(time.Minute), + }) + if err != nil { + t.Fatalf("drain: %v", err) + } + if task == nil { + break + } + } + + task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{ + Owner: "worker-1", Now: now, LeaseUntil: now.Add(time.Minute), + }) + if err != nil { + t.Fatalf("claim: %v", err) + } + if task != nil { + t.Errorf("expected nil on an empty queue, got %s", task.ID) + } +} + +func TestUpdateRejectsStaleVersion(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + job, _ := seedJob(t, pool, 1) + + repo, tx := NewTaskRepo(pool), NewTxManager(pool) + now := time.Now().UTC() + + task, err := repo.ClaimNext(ctx, usecase.ClaimFilter{ + Owner: "worker-1", Now: now, LeaseUntil: now.Add(time.Minute), + }) + if err != nil || task == nil || task.JobID != job.ID { + t.Skipf("could not claim this job's task (got %v, %v)", task, err) + } + + // A stale copy: same row, but the version it remembers is behind. + stale := *task + stale.Version = task.Version // pretend the caller mutated it once + + err = tx.WithinTx(ctx, func(ctx context.Context) error { + fresh, err := repo.GetForUpdate(ctx, task.ID) + if err != nil { + return err + } + if err := fresh.RenewLease("worker-1", fresh.Attempt, now.Add(2*time.Minute)); err != nil { + return err + } + return repo.Update(ctx, fresh) + }) + if err != nil { + t.Fatalf("legitimate update failed: %v", err) + } + + // Now the stale copy's version is behind by one; its write must be refused. + stale.Version++ // as a domain method would have done + if err := repo.Update(ctx, &stale); !errors.Is(err, domain.ErrLeaseConflict) { + t.Errorf("stale update err = %v, want ErrLeaseConflict", err) + } +} + +func TestListCompletedIsOrderedByChunkIndex(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + job, tasks := seedJob(t, pool, 4) + + repo, tx := NewTaskRepo(pool), NewTxManager(pool) + now := time.Now().UTC() + + // Complete them out of order to prove the ordering comes from SQL. + for _, i := range []int{2, 0, 3, 1} { + task := tasks[i] + err := tx.WithinTx(ctx, func(ctx context.Context) error { + fresh, err := repo.GetForUpdate(ctx, task.ID) + if err != nil { + return err + } + owner := "worker-1" + fresh.Status = domain.TaskLeased + fresh.LeaseOwner = &owner + expires := now.Add(time.Minute) + fresh.LeaseExpiresAt = &expires + if err := fresh.CompleteWith( + fmt.Sprintf("s3://result-%d", fresh.ChunkIndex), + fmt.Sprintf("rsha-%d", fresh.ChunkIndex), + nil, owner, fresh.Attempt, now, + ); err != nil { + return err + } + return repo.Update(ctx, fresh) + }) + if err != nil { + t.Fatalf("complete chunk %d: %v", i, err) + } + } + + done, err := repo.ListCompleted(ctx, job.ID) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(done) != 4 { + t.Fatalf("got %d completed, want 4", len(done)) + } + for i, task := range done { + if task.ChunkIndex != i { + t.Errorf("position %d holds chunk_index %d — order is not deterministic", i, task.ChunkIndex) + } + } +} + +// A worker whose network dropped resends the same manifest. That must succeed: +// the entity is unchanged, so nothing is written, and the optimistic-concurrency +// guard must not turn the replay into a conflict. +func TestCompleteTaskReplayIsIdempotent(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + job, _ := seedJob(t, pool, 1) + + tasks, jobs, tx := NewTaskRepo(pool), NewJobRepo(pool), NewTxManager(pool) + clk := fixedClock{now: time.Now().UTC()} + uc := usecase.NewCompleteTask(tasks, jobs, tx, clk) + + claimed, err := tasks.ClaimNext(ctx, usecase.ClaimFilter{ + Owner: "worker-1", Now: clk.now, LeaseUntil: clk.now.Add(time.Minute), + }) + if err != nil || claimed == nil || claimed.JobID != job.ID { + t.Skipf("could not claim this job's task (got %v, %v)", claimed, err) + } + + in := usecase.CompleteTaskInput{ + TaskID: claimed.ID, WorkerID: "worker-1", Attempt: claimed.Attempt, + ResultURI: "s3://r0", ResultSHA256: "rsha", + } + if _, err := uc.Execute(ctx, in); err != nil { + t.Fatalf("first submission: %v", err) + } + if _, err := uc.Execute(ctx, in); err != nil { + t.Errorf("replay must be idempotent, got %v", err) + } + + // A different manifest for the same task is a genuine conflict. + other := in + other.ResultURI = "s3://different" + if _, err := uc.Execute(ctx, other); !errors.Is(err, domain.ErrResultConflict) { + t.Errorf("err = %v, want ErrResultConflict", err) + } +} + +type fixedClock struct{ now time.Time } + +func (c fixedClock) Now() time.Time { return c.now } + +func TestExpireLeasesRequeuesElapsedTasks(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + job, _ := seedJob(t, pool, 1) + + repo := NewTaskRepo(pool) + past := time.Now().UTC().Add(-time.Hour) + + // Lease it with an expiry already in the past. + task, err := repo.ClaimNext(ctx, usecase.ClaimFilter{ + Owner: "dead-worker", Now: past, LeaseUntil: past.Add(time.Minute), + }) + if err != nil || task == nil || task.JobID != job.ID { + t.Skipf("could not claim this job's task (got %v, %v)", task, err) + } + + if _, err := repo.ExpireLeases(ctx, time.Now().UTC()); err != nil { + t.Fatalf("expire: %v", err) + } + + counts, err := repo.CountByStatus(ctx, job.ID) + if err != nil { + t.Fatalf("count: %v", err) + } + if counts[domain.TaskPending] != 1 { + t.Errorf("pending = %d, want 1 — a dead worker must not strand its task", counts[domain.TaskPending]) + } +} diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go index 3751801..1945a07 100644 --- a/coordinator/internal/storage/postgres/job_repo.go +++ b/coordinator/internal/storage/postgres/job_repo.go @@ -2,9 +2,11 @@ package postgres import ( "context" + "errors" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/emil28092005/SciMesh/coordinator/internal/domain" @@ -22,19 +24,50 @@ func NewJobRepo(pool *pgxpool.Pool) *JobRepo { var _ usecase.JobRepository = (*JobRepo)(nil) -// TODO(phase 2-3): replace stubs with real pgx queries. +const jobColumns = `id, workload, input_uri, parameters, status, created_at, completed_at` +const insertJobSQL = ` +INSERT INTO jobs (id, workload, input_uri, parameters, status, created_at) +VALUES ($1, $2, $3, $4, $5, $6)` + +// Insert runs inside the caller's transaction, alongside the job's tasks — that +// is what makes "all tasks or none" hold. func (r *JobRepo) Insert(ctx context.Context, j *domain.Job) error { - // Phase 2: INSERT INTO jobs ...; runs inside the caller's transaction. - return usecase.ErrNotImplemented + _, err := conn(ctx, r.pool).Exec(ctx, insertJobSQL, + j.ID, j.Workload, j.InputURI, jsonbOrEmpty(j.Parameters), string(j.Status), j.CreatedAt) + return err } +const getJobSQL = `SELECT ` + jobColumns + ` FROM jobs WHERE id = $1` + func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) { - // Phase 3: SELECT ... WHERE id = $1; no rows -> domain.ErrJobNotFound. - return nil, usecase.ErrNotImplemented + var ( + j domain.Job + status string + ) + err := conn(ctx, r.pool).QueryRow(ctx, getJobSQL, id).Scan( + &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, domain.ErrJobNotFound + } + if err != nil { + return nil, err + } + j.Status = domain.JobStatus(status) + return &j, nil } -func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error { - // Phase 3: UPDATE jobs SET status = $2, completed_at = $3 WHERE id = $1. - return usecase.ErrNotImplemented +const updateJobStatusSQL = `UPDATE jobs SET status = $2, completed_at = $3 WHERE id = $1` + +func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, + status domain.JobStatus, completedAt *time.Time) error { + + tag, err := conn(ctx, r.pool).Exec(ctx, updateJobStatusSQL, id, string(status), completedAt) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return domain.ErrJobNotFound + } + return nil } diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go index 631a88b..4c0abe1 100644 --- a/coordinator/internal/storage/postgres/task_repo.go +++ b/coordinator/internal/storage/postgres/task_repo.go @@ -2,9 +2,11 @@ package postgres import ( "context" + "errors" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/emil28092005/SciMesh/coordinator/internal/domain" @@ -22,6 +24,37 @@ func NewTaskRepo(pool *pgxpool.Pool) *TaskRepo { var _ usecase.TaskRepository = (*TaskRepo)(nil) +// taskColumns is the single source of truth for the shape scanTask expects. +// Every query that returns a task selects exactly this list, in this order — +// three hand-written column lists would drift apart within a week. +const taskColumns = `id, job_id, chunk_index, workload, input_uri, input_sha256, + parameters, status, attempt, max_attempts, lease_owner, lease_expires_at, + result_uri, result_sha256, metrics, error_code, error_message, + created_at, started_at, completed_at, version` + +// scanTask maps one row onto an entity. +// +// status is read into a plain string rather than domain.TaskStatus: pgx does +// not know the task_status enum, and going through string keeps the driver out +// of the domain's type system. +func scanTask(row pgx.Row) (*domain.Task, error) { + var ( + t domain.Task + status string + ) + err := row.Scan( + &t.ID, &t.JobID, &t.ChunkIndex, &t.Workload, &t.InputURI, &t.InputSHA256, + &t.Parameters, &status, &t.Attempt, &t.MaxAttempts, &t.LeaseOwner, &t.LeaseExpiresAt, + &t.ResultURI, &t.ResultSHA256, &t.Metrics, &t.ErrorCode, &t.ErrorMessage, + &t.CreatedAt, &t.StartedAt, &t.CompletedAt, &t.Version, + ) + if err != nil { + return nil, err + } + t.Status = domain.TaskStatus(status) + return &t, nil +} + // claimNextSQL leases one task in a single statement. // // FOR UPDATE SKIP LOCKED is what makes concurrent coordinators safe: each @@ -29,10 +62,11 @@ var _ usecase.TaskRepository = (*TaskRepo)(nil) // so no task is ever handed to two workers and no claim blocks behind another. // Splitting this into SELECT + UPDATE would reintroduce exactly that race. // -//nolint:unused // wired up in phase 2; kept beside the repository it belongs to +// The CTE column is aliased to cid so the RETURNING list below can use bare +// column names without colliding with the candidate relation. const claimNextSQL = ` WITH candidate AS ( - SELECT id + SELECT id AS cid FROM tasks WHERE status = 'pending' AND attempt < max_attempts @@ -49,17 +83,174 @@ SET status = 'leased', started_at = COALESCE(started_at, $4), version = version + 1 FROM candidate -WHERE tasks.id = candidate.id -RETURNING tasks.id, tasks.job_id, tasks.chunk_index, tasks.workload, - tasks.input_uri, tasks.input_sha256, tasks.parameters, - tasks.status, tasks.attempt, tasks.max_attempts, - tasks.lease_owner, tasks.lease_expires_at, tasks.version; -` +WHERE tasks.id = candidate.cid +RETURNING ` + taskColumns + +// ClaimNext atomically leases the next eligible task. +func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domain.Task, error) { + workloads := f.Workloads + if workloads == nil { + workloads = []string{} // NULL would make the cardinality() guard fail + } + + var task *domain.Task + err := withRetry(ctx, func(ctx context.Context) error { + row := conn(ctx, r.pool).QueryRow(ctx, claimNextSQL, workloads, f.Owner, f.LeaseUntil, f.Now) + t, err := scanTask(row) + if errors.Is(err, pgx.ErrNoRows) { + task = nil + return nil // an empty queue is a normal state, not a failure + } + if err != nil { + return err + } + task = t + return nil + }) + if err != nil { + return nil, err + } + return task, nil +} + +const getForUpdateSQL = `SELECT ` + taskColumns + ` FROM tasks WHERE id = $1 FOR UPDATE` + +// GetForUpdate reads a task and holds its row lock until the caller's +// transaction ends, so read-modify-write use cases cannot interleave. +func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) { + t, err := scanTask(conn(ctx, r.pool).QueryRow(ctx, getForUpdateSQL, id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, domain.ErrTaskNotFound + } + if err != nil { + return nil, err + } + return t, nil +} + +// updateTaskSQL writes the mutated entity back under optimistic concurrency. +// +// The entity has already incremented its Version in memory, so the new value +// goes into SET while the guard in WHERE compares against the previous one. +const updateTaskSQL = ` +UPDATE tasks +SET status = $2, + attempt = $3, + lease_owner = $4, + lease_expires_at = $5, + result_uri = $6, + result_sha256 = $7, + metrics = $8, + error_code = $9, + error_message = $10, + started_at = $11, + completed_at = $12, + version = $13 +WHERE id = $1 AND version = $13 - 1` + +func (r *TaskRepo) Update(ctx context.Context, t *domain.Task) error { + tag, err := conn(ctx, r.pool).Exec(ctx, updateTaskSQL, + t.ID, string(t.Status), t.Attempt, t.LeaseOwner, t.LeaseExpiresAt, + t.ResultURI, t.ResultSHA256, t.Metrics, t.ErrorCode, t.ErrorMessage, + t.StartedAt, t.CompletedAt, t.Version, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + // Either the row vanished or someone else advanced its version while we + // held a stale copy. Both mean this write must not land. + return domain.ErrLeaseConflict + } + return nil +} + +const insertTaskSQL = ` +INSERT INTO tasks (id, job_id, chunk_index, workload, input_uri, input_sha256, + parameters, status, attempt, max_attempts, created_at, version) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)` + +// InsertBatch writes every task in one round trip. It runs inside the caller's +// transaction, which is what makes "all tasks or none" hold. +func (r *TaskRepo) InsertBatch(ctx context.Context, tasks []*domain.Task) error { + if len(tasks) == 0 { + return nil + } + + batch := &pgx.Batch{} + for _, t := range tasks { + batch.Queue(insertTaskSQL, + t.ID, t.JobID, t.ChunkIndex, t.Workload, t.InputURI, t.InputSHA256, + jsonbOrEmpty(t.Parameters), string(t.Status), t.Attempt, t.MaxAttempts, t.CreatedAt, t.Version, + ) + } + + results := conn(ctx, r.pool).SendBatch(ctx, batch) + for range tasks { + if _, err := results.Exec(); err != nil { + _ = results.Close() + return err + } + } + return results.Close() +} + +const listCompletedSQL = ` +SELECT ` + taskColumns + ` +FROM tasks +WHERE job_id = $1 AND status = 'completed' +ORDER BY chunk_index` + +// ListCompleted returns results in chunk order, which the stitcher relies on: +// a non-deterministic order would make the merged output depend on which worker +// happened to finish first. +func (r *TaskRepo) ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error) { + rows, err := conn(ctx, r.pool).Query(ctx, listCompletedSQL, jobID) + if err != nil { + return nil, err + } + defer rows.Close() + + var tasks []*domain.Task + for rows.Next() { + t, err := scanTask(rows) + if err != nil { + return nil, err + } + tasks = append(tasks, t) + } + return tasks, rows.Err() +} + +const countByStatusSQL = `SELECT status, count(*) FROM tasks WHERE job_id = $1 GROUP BY status` + +func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error) { + rows, err := conn(ctx, r.pool).Query(ctx, countByStatusSQL, jobID) + if err != nil { + return nil, err + } + defer rows.Close() + + counts := make(map[domain.TaskStatus]int) + for rows.Next() { + var ( + status string + n int + ) + if err := rows.Scan(&status, &n); err != nil { + return nil, err + } + counts[domain.TaskStatus(status)] = n + } + return counts, rows.Err() +} // expireLeasesSQL applies the lease-expiry rule set-based, mirroring // domain.Task.ExpireLease: requeue while attempts remain, otherwise fail. // -//nolint:unused // wired up in phase 5; mirrors domain.Task.ExpireLease +// It is one statement rather than a load-decide-save loop because several +// coordinators run it concurrently; an atomic UPDATE makes the duplicate work +// harmless — the loser simply updates zero rows. const expireLeasesSQL = ` UPDATE tasks SET status = CASE WHEN attempt < max_attempts THEN 'pending'::task_status @@ -72,43 +263,17 @@ SET status = CASE WHEN attempt < max_attempts THEN 'pending'::task_sta ELSE error_message END, completed_at = CASE WHEN attempt >= max_attempts THEN $1 ELSE completed_at END, version = version + 1 -WHERE status = 'leased' AND lease_expires_at < $1; -` - -// TODO(phase 2-6): replace stubs with real pgx queries; the SQL above is ready -// to wire up. Each method maps 1:1 to a roadmap phase. - -func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domain.Task, error) { - // Phase 2: run claimNextSQL; pgx.ErrNoRows -> (nil, nil). - return nil, usecase.ErrNotImplemented -} - -func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) { - // Phase 3: SELECT ... WHERE id = $1 FOR UPDATE; no rows -> domain.ErrTaskNotFound. - return nil, usecase.ErrNotImplemented -} - -func (r *TaskRepo) Update(ctx context.Context, t *domain.Task) error { - // Phase 3: UPDATE ... WHERE id = $1 AND version = $2 (optimistic concurrency). - return usecase.ErrNotImplemented -} - -func (r *TaskRepo) InsertBatch(ctx context.Context, tasks []*domain.Task) error { - // Phase 2: pgx.Batch or COPY; runs inside the caller's transaction. - return usecase.ErrNotImplemented -} - -func (r *TaskRepo) ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error) { - // Phase 6: WHERE job_id = $1 AND status = 'completed' ORDER BY chunk_index. - return nil, usecase.ErrNotImplemented -} - -func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error) { - // Phase 3: SELECT status, count(*) ... GROUP BY status. - return nil, usecase.ErrNotImplemented -} +WHERE status = 'leased' AND lease_expires_at < $1` func (r *TaskRepo) ExpireLeases(ctx context.Context, now time.Time) (int64, error) { - // Phase 5: run expireLeasesSQL, return the affected row count. - return 0, usecase.ErrNotImplemented + var affected int64 + err := withRetry(ctx, func(ctx context.Context) error { + tag, err := conn(ctx, r.pool).Exec(ctx, expireLeasesSQL, now, domain.ErrCodeLeaseExpired) + if err != nil { + return err + } + affected = tag.RowsAffected() + return nil + }) + return affected, err } diff --git a/coordinator/internal/storage/postgres/tx.go b/coordinator/internal/storage/postgres/tx.go index 72dfe9b..4b70e17 100644 --- a/coordinator/internal/storage/postgres/tx.go +++ b/coordinator/internal/storage/postgres/tx.go @@ -12,12 +12,11 @@ import ( // querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting every // repository method run identically inside or outside a transaction. -// -//nolint:unused // used by repository methods once phase 2 replaces the stubs type querier interface { Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults } // txKey is an unexported struct type, so no other package can collide with it @@ -39,11 +38,28 @@ func NewTxManager(pool *pgxpool.Pool) *TxManager { // The transaction travels in the context rather than in fn's signature, which // is what lets the usecase layer express "do these repository calls atomically" // without its port ever mentioning pgx. +// Retrying happens here, around the whole transaction, and deliberately not +// inside the repositories. Once Postgres aborts a transaction with a +// serialization failure or deadlock, every further statement in it fails too — +// replaying a single query would accomplish nothing. The unit of retry is +// Begin → fn → Commit. +// +// This is safe because fn re-reads its rows (via GetForUpdate) on each attempt, +// so a retry starts from the current state rather than stale entities. func (m *TxManager) WithinTx(ctx context.Context, fn func(ctx context.Context) error) error { if _, ok := ctx.Value(txKey{}).(pgx.Tx); ok { - return fn(ctx) // already inside a transaction — join it, don't nest + // Already inside a transaction — join it. Retrying here would be wrong + // twice over: the outer transaction owns the retry, and re-running fn + // alone cannot undo what the outer one already wrote. + return fn(ctx) } + return withRetry(ctx, func(ctx context.Context) error { + return m.runTx(ctx, fn) + }) +} + +func (m *TxManager) runTx(ctx context.Context, fn func(ctx context.Context) error) error { tx, err := m.pool.Begin(ctx) if err != nil { return err @@ -58,9 +74,17 @@ func (m *TxManager) WithinTx(ctx context.Context, fn func(ctx context.Context) e return tx.Commit(ctx) } +// jsonbOrEmpty keeps a nil map from reaching a NOT NULL jsonb column. pgx +// encodes a nil map as SQL NULL rather than omitting the column, so the +// DEFAULT '{}' never gets a chance to apply. +func jsonbOrEmpty(m map[string]any) map[string]any { + if m == nil { + return map[string]any{} + } + return m +} + // conn returns the transaction bound to ctx, or the pool when there is none. -// -//nolint:unused // every repository method will route through this in phase 2 func conn(ctx context.Context, pool *pgxpool.Pool) querier { if tx, ok := ctx.Value(txKey{}).(pgx.Tx); ok { return tx diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index c9dc801..da177bf 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -130,14 +130,23 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom return err } now := uc.clock.Now() + before := task.Version if err := task.CompleteWith(in.ResultURI, in.ResultSHA256, in.Metrics, in.WorkerID, in.Attempt, now); err != nil { return err } + out = task + + // A replay of an already-recorded result leaves the entity untouched. + // Writing anyway would fail the optimistic-concurrency guard (the stored + // version already equals ours) and turn an idempotent call into a 409. + if task.Version == before { + return nil + } + if err := uc.tasks.Update(ctx, task); err != nil { return err } - out = task return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) }) if err != nil { From 6517145622b38fc665631da6e186d69dd3065da1 Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Wed, 22 Jul 2026 14:53:27 +0300 Subject: [PATCH 03/17] chore(coordinator): add API smoke script and request collection Two ways to exercise every endpoint, both living in the repo rather than in a personal Postman workspace: - scripts/smoke.sh walks the full lifecycle and asserts each status, exiting non-zero on the first surprise, so it works in CI as well as by hand; - api/requests.http drives the same calls from an editor's REST client, with later requests reusing ids captured from earlier responses. It doubles as API documentation for the worker author. The script claims until it sees its own job's chunks instead of assuming an empty queue: a shared development database usually holds pending tasks from earlier runs, and it takes the attempt number from the claim response, since a task requeued after an expired lease comes back with attempt 2 or 3. Note for whoever extends the validation cases: Go matches JSON field names case-insensitively, so "worker_ID" is accepted as "worker_id". Only a genuinely unknown key trips DisallowUnknownFields. --- coordinator/Makefile | 7 ++ coordinator/README.md | 12 +++ coordinator/api/requests.http | 157 ++++++++++++++++++++++++++++++++++ coordinator/scripts/smoke.sh | 122 ++++++++++++++++++++++++++ 4 files changed, 298 insertions(+) create mode 100644 coordinator/api/requests.http create mode 100755 coordinator/scripts/smoke.sh diff --git a/coordinator/Makefile b/coordinator/Makefile index 38e4b54..7dceb9b 100644 --- a/coordinator/Makefile +++ b/coordinator/Makefile @@ -63,3 +63,10 @@ rebuild: psql: docker compose exec postgres psql -U scimesh -d scimesh + +# --- api ------------------------------------------------------------------ +# Exercises every endpoint against a running coordinator; exits non-zero on the +# first unexpected status. See also api/requests.http for clicking through them +# one at a time in an editor. +smoke: + ./scripts/smoke.sh diff --git a/coordinator/README.md b/coordinator/README.md index ef54ec1..8a10bac 100644 --- a/coordinator/README.md +++ b/coordinator/README.md @@ -111,6 +111,18 @@ See `.env.example`; only `DATABASE_URL` is required. | GET | `/jobs/{job_id}` | Aggregate job progress | | GET | `/health` | Liveness (unauthenticated) | +## Poking the API + +Two ways, both checked in: + +```sh +make smoke # every endpoint, asserted; non-zero exit on failure +``` + +`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 The queue works end to end: a job can be submitted, split into tasks, leased to diff --git a/coordinator/api/requests.http b/coordinator/api/requests.http new file mode 100644 index 0000000..5a4b356 --- /dev/null +++ b/coordinator/api/requests.http @@ -0,0 +1,157 @@ +# 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 + +### Health — the only unauthenticated endpoint +GET {{host}}/health + +### Auth check — no token must be rejected with 401 +POST {{host}}/tasks/claim +Content-Type: application/json + +{ "worker_id": "{{worker}}" } + +### 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}}", + "workloads": ["similarity_search"] +} + +@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}} +} + +### 4. Submit the result (200) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "result_uri": "s3://results/shard-0.csv", + "result_sha256": "r0sha", + "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_uri": "s3://results/shard-0.csv", + "result_sha256": "r0sha", + "metrics": { "elapsed_ms": 1234, "candidates": 50000 } +} + +### 4b. A different result for the same task — conflict (409) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "result_uri": "s3://results/SOMETHING-ELSE.csv", + "result_sha256": "different" +} + +### 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_uri": "s3://results/x.csv", + "result_sha256": "x" +} + +### 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/coordinator/scripts/smoke.sh b/coordinator/scripts/smoke.sh new file mode 100755 index 0000000..16702f9 --- /dev/null +++ b/coordinator/scripts/smoke.sh @@ -0,0 +1,122 @@ +#!/usr/bin/env bash +# +# End-to-end smoke test against a running coordinator. +# +# ./scripts/smoke.sh # localhost:8080, token from .env +# HOST=http://1.2.3.4:8080 TOKEN=x ./scripts/smoke.sh +# +# Exits non-zero on the first unexpected status, so it is usable in CI. + +set -uo pipefail + +HOST="${HOST:-http://localhost:8080}" +TOKEN="${TOKEN:-$(grep -s '^WORKER_AUTH_TOKEN=' .env | cut -d= -f2- || echo change-me)}" + +pass=0 +fail=0 + +# check