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.
This commit is contained in:
Efremenko Arhip
@@ -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
|
||||
@@ -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
|
||||
@@ -0,0 +1,4 @@
|
||||
/coordinator
|
||||
/bin/
|
||||
.env
|
||||
*.out
|
||||
@@ -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
|
||||
@@ -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`, осталось их подключить.
|
||||
@@ -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"]
|
||||
@@ -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
|
||||
@@ -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`.
|
||||
@@ -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
|
||||
}
|
||||
@@ -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:
|
||||
@@ -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
|
||||
)
|
||||
@@ -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=
|
||||
@@ -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")
|
||||
)
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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() }
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
@@ -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())
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
Reference in New Issue
Block a user