feat(coordinator): worker registry, API contract, logging & DB retry
Align the coordinator with the master PLAN.md (CTX-00, CTX-04) and harden process startup. - CTX-00: freeze docs/api-contract.md as the v1 source of truth for the Go coordinator and Python worker. - CTX-04: worker registry — workers table (migration 0002), domain.Worker, RegisterWorker use case, WorkerRepository, and POST /workers/register. - Contract alignment: claim uses `capabilities` (was `workloads`), COORDINATOR_TOKEN env (WORKER_AUTH_TOKEN kept as fallback), and GET /health now reports database readiness (503 when the DB is down). - Logging: logs are teed to stdout and an optional rotated file (LOG_FILE) via lumberjack, so they survive a container rebuild. - Startup resilience: the initial DB connection is retried with backoff, so the coordinator waits for Postgres to boot instead of crash-looping.
This commit is contained in:
Efremenko Arhip
@@ -6,8 +6,15 @@ 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
|
||||
|
||||
# Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only;
|
||||
# set a path to also write a size-rotated file (kept across restarts).
|
||||
LOG_LEVEL=info
|
||||
# LOG_FILE=./logs/coordinator.log
|
||||
|
||||
# Optional tuning (defaults shown).
|
||||
DB_MAX_CONNS=10
|
||||
# How long to keep retrying the initial DB connection while Postgres boots.
|
||||
DB_CONNECT_TIMEOUT=30s
|
||||
REQUEST_TIMEOUT=15s
|
||||
LEASE_DURATION=2m
|
||||
DEFAULT_MAX_ATTEMPTS=3
|
||||
|
||||
@@ -2,3 +2,4 @@
|
||||
/bin/
|
||||
.env
|
||||
*.out
|
||||
/logs/
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
@token = change-me
|
||||
@worker = worker-1
|
||||
|
||||
### Health — the only unauthenticated endpoint
|
||||
### Readiness — the only unauthenticated endpoint (probes the database)
|
||||
GET {{host}}/health
|
||||
|
||||
### Auth check — no token must be rejected with 401
|
||||
@@ -19,6 +19,21 @@ Content-Type: application/json
|
||||
|
||||
{ "worker_id": "{{worker}}" }
|
||||
|
||||
### 0. Register a worker (201)
|
||||
# @name register
|
||||
POST {{host}}/workers/register
|
||||
Authorization: Bearer {{token}}
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"name": "lab-worker-01",
|
||||
"capabilities": ["similarity_search"],
|
||||
"cpu_count": 8,
|
||||
"memory_mb": 16384
|
||||
}
|
||||
|
||||
@workerId = {{register.response.body.worker_id}}
|
||||
|
||||
### 1. Create a job and its chunks (201)
|
||||
# The coordinator splits the submission into one task per chunk, transactionally.
|
||||
# @name createJob
|
||||
@@ -48,7 +63,8 @@ Content-Type: application/json
|
||||
|
||||
{
|
||||
"worker_id": "{{worker}}",
|
||||
"workloads": ["similarity_search"]
|
||||
"capabilities": ["similarity_search"],
|
||||
"max_concurrency": 1
|
||||
}
|
||||
|
||||
@taskId = {{claim.response.body.task_id}}
|
||||
|
||||
@@ -1,7 +1,3 @@
|
||||
// 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 (
|
||||
@@ -19,52 +15,58 @@ import (
|
||||
)
|
||||
|
||||
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)
|
||||
// All work happens in run() so its defers (pool.Close, log flush, signal
|
||||
// stop) still execute: os.Exit skips deferred calls entirely.
|
||||
if err := run(); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func run(log *slog.Logger) error {
|
||||
cfg, err := infra.Load()
|
||||
func run() error {
|
||||
// Bootstrap logger, used only until config says where logs should go. It
|
||||
// writes to stderr so it never contaminates the configured stdout stream.
|
||||
boot := slog.New(slog.NewJSONHandler(os.Stderr, nil))
|
||||
|
||||
cfg, err := infra.LoadConfig()
|
||||
if err != nil {
|
||||
boot.Error("load config", "err", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// One cancellation source for the whole process: HTTP server and reaper
|
||||
// both observe it and wind down together.
|
||||
// The real logger: stdout plus an optional rotated file (LOG_FILE).
|
||||
log, logCloser, err := infra.NewLogger(cfg)
|
||||
if err != nil {
|
||||
boot.Error("init logger", "err", err)
|
||||
return err
|
||||
}
|
||||
defer func() { _ = logCloser.Close() }()
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
pool, err := infra.NewPool(ctx, cfg)
|
||||
pool, err := infra.NewPool(ctx, cfg, log)
|
||||
if err != nil {
|
||||
log.Error("connect database", "err", err)
|
||||
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)
|
||||
clk = infra.NewClock()
|
||||
tx = postgres.NewTxManager(pool)
|
||||
taskRepo = postgres.NewTaskRepo(pool)
|
||||
jobRepo = postgres.NewJobRepo(pool)
|
||||
workerRepo = postgres.NewWorkerRepo(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),
|
||||
RegisterWorker: usecase.NewRegisterWorker(workerRepo, clk),
|
||||
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
|
||||
@@ -77,8 +79,10 @@ func run(log *slog.Logger) error {
|
||||
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))
|
||||
// pool.Ping backs /health: readiness means the database answers, not just
|
||||
// that the process is alive.
|
||||
api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, pool.Ping)
|
||||
err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token))
|
||||
|
||||
// Shutdown order matters, and defers alone cannot express it (they run
|
||||
// LIFO, so the deferred stop() would fire *after* the wait below).
|
||||
|
||||
@@ -53,8 +53,14 @@ services:
|
||||
REQUEST_TIMEOUT: "15s"
|
||||
LEASE_DURATION: "2m"
|
||||
REAPER_INTERVAL: "30s"
|
||||
LOG_LEVEL: ${LOG_LEVEL:-info}
|
||||
# Logs are teed to stdout (docker logs) and this rotated file, which lives
|
||||
# on the mounted ./logs directory so it survives a rebuild.
|
||||
LOG_FILE: /var/log/scimesh/coordinator.log
|
||||
ports:
|
||||
- "${COORDINATOR_PORT:-8080}:8080"
|
||||
volumes:
|
||||
- ./logs:/var/log/scimesh
|
||||
healthcheck:
|
||||
test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"]
|
||||
interval: 10s
|
||||
|
||||
@@ -7,6 +7,7 @@ require (
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/jackc/pgx/v5 v5.6.0
|
||||
github.com/joho/godotenv v1.5.1
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1
|
||||
)
|
||||
|
||||
require (
|
||||
|
||||
@@ -29,6 +29,8 @@ 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/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
|
||||
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=
|
||||
|
||||
@@ -10,6 +10,7 @@ import "errors"
|
||||
var (
|
||||
ErrJobNotFound = errors.New("job not found")
|
||||
ErrTaskNotFound = errors.New("task not found")
|
||||
ErrWorkerNotFound = errors.New("worker 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")
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type WorkerStatus string
|
||||
|
||||
const (
|
||||
WorkerOnline WorkerStatus = "online"
|
||||
WorkerBusy WorkerStatus = "busy"
|
||||
WorkerOffline WorkerStatus = "offline"
|
||||
)
|
||||
|
||||
// Worker is a registered process/machine allowed to claim tasks. Its
|
||||
// capabilities are the allowlisted workload names it can run; the coordinator
|
||||
// never hands it a task outside that set.
|
||||
type Worker struct {
|
||||
ID uuid.UUID
|
||||
Name string
|
||||
Capabilities []string
|
||||
Status WorkerStatus
|
||||
LastHeartbeatAt time.Time
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
// NewWorker registers a worker. A worker with no capabilities could never be
|
||||
// handed a task, so an empty set is rejected rather than silently stored.
|
||||
func NewWorker(name string, capabilities []string, now time.Time) (*Worker, error) {
|
||||
if len(capabilities) == 0 {
|
||||
return nil, ErrInvalidInput
|
||||
}
|
||||
return &Worker{
|
||||
ID: uuid.New(),
|
||||
Name: name,
|
||||
Capabilities: capabilities,
|
||||
Status: WorkerOnline,
|
||||
LastHeartbeatAt: now,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}, nil
|
||||
}
|
||||
@@ -23,13 +23,23 @@ type Config struct {
|
||||
// PostgreSQL connection string (pgx format / libpq URL).
|
||||
DatabaseURL string
|
||||
// Shared bearer token workers must present. Empty disables auth (dev only).
|
||||
WorkerAuthToken string
|
||||
Token string
|
||||
|
||||
// Minimum log level: debug, info, warn, error.
|
||||
LogLevel string
|
||||
// Path to a rotated log file. Empty logs to stdout only.
|
||||
LogFile string
|
||||
|
||||
// Connection pool upper bound.
|
||||
DBMaxConns int32
|
||||
// How long to keep retrying the initial database connection at startup
|
||||
// before giving up. Covers a Postgres container that is still booting.
|
||||
DBConnectTimeout time.Duration
|
||||
// Per-request context timeout applied to handlers and DB calls.
|
||||
RequestTimeout time.Duration
|
||||
|
||||
// Suggested heartbeat cadence returned to workers on registration.
|
||||
HeartbeatInterval time.Duration
|
||||
// Default lease length handed out on claim.
|
||||
LeaseDuration time.Duration
|
||||
// Default attempt ceiling for newly created tasks.
|
||||
@@ -43,7 +53,7 @@ type Config struct {
|
||||
//
|
||||
// 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) {
|
||||
func LoadConfig() (Config, error) {
|
||||
envFile := os.Getenv("ENV_FILE")
|
||||
if envFile == "" {
|
||||
envFile = defaultEnvFile
|
||||
@@ -56,11 +66,17 @@ func Load() (Config, error) {
|
||||
}
|
||||
|
||||
cfg := Config{
|
||||
Addr: getEnv("COORDINATOR_ADDR", ":8080"),
|
||||
DatabaseURL: os.Getenv("DATABASE_URL"),
|
||||
WorkerAuthToken: os.Getenv("WORKER_AUTH_TOKEN"),
|
||||
Addr: getEnv("COORDINATOR_ADDR", ":8080"),
|
||||
DatabaseURL: os.Getenv("DATABASE_URL"),
|
||||
// COORDINATOR_TOKEN is the contract name; WORKER_AUTH_TOKEN is the
|
||||
// former name, still honoured so existing .env files keep working.
|
||||
Token: getEnv("COORDINATOR_TOKEN", os.Getenv("WORKER_AUTH_TOKEN")),
|
||||
LogLevel: getEnv("LOG_LEVEL", "info"),
|
||||
LogFile: os.Getenv("LOG_FILE"),
|
||||
DBMaxConns: 10,
|
||||
DBConnectTimeout: 30 * time.Second,
|
||||
RequestTimeout: 15 * time.Second,
|
||||
HeartbeatInterval: 15 * time.Second,
|
||||
LeaseDuration: 2 * time.Minute,
|
||||
DefaultMaxAttempts: 3,
|
||||
ReaperInterval: 30 * time.Second,
|
||||
@@ -74,9 +90,15 @@ func Load() (Config, error) {
|
||||
if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.DBConnectTimeout, err = getEnvDuration("DB_CONNECT_TIMEOUT", cfg.DBConnectTimeout); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.RequestTimeout, err = getEnvDuration("REQUEST_TIMEOUT", cfg.RequestTimeout); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.HeartbeatInterval, err = getEnvDuration("HEARTBEAT_INTERVAL", cfg.HeartbeatInterval); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.LeaseDuration, err = getEnvDuration("LEASE_DURATION", cfg.LeaseDuration); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
|
||||
@@ -3,13 +3,16 @@ package infra
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
"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) {
|
||||
func NewPool(ctx context.Context, cfg Config, log *slog.Logger) (*pgxpool.Pool, error) {
|
||||
poolCfg, err := pgxpool.ParseConfig(cfg.DatabaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -20,11 +23,43 @@ func NewPool(ctx context.Context, cfg Config) (*pgxpool.Pool, error) {
|
||||
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 {
|
||||
// pgxpool.New is lazy, so a ping is needed to actually reach the server.
|
||||
// It is retried because at startup — especially under docker-compose, where
|
||||
// the coordinator can boot before Postgres is accepting connections — a
|
||||
// service should wait for its database rather than crash-loop.
|
||||
if err := pingWithRetry(ctx, pool, cfg.DBConnectTimeout, log); err != nil {
|
||||
pool.Close()
|
||||
return nil, err
|
||||
}
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
// pingWithRetry waits for the database to accept connections, backing off
|
||||
// between attempts until the budget elapses or ctx is cancelled.
|
||||
//
|
||||
// Unlike the transaction retry in storage/postgres, this retries *any* ping
|
||||
// error: at startup a "connection refused" is the expected, retryable state,
|
||||
// not an anomaly.
|
||||
func pingWithRetry(ctx context.Context, pool *pgxpool.Pool, budget time.Duration, log *slog.Logger) error {
|
||||
b := backoff.NewExponentialBackOff()
|
||||
b.InitialInterval = 200 * time.Millisecond
|
||||
b.MaxInterval = 3 * time.Second
|
||||
b.MaxElapsedTime = budget
|
||||
|
||||
attempt := 0
|
||||
return backoff.RetryNotify(
|
||||
func() error {
|
||||
// A bounded per-attempt timeout so one hung dial cannot eat the
|
||||
// whole budget in a single try.
|
||||
pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
defer cancel()
|
||||
return pool.Ping(pingCtx)
|
||||
},
|
||||
backoff.WithContext(b, ctx),
|
||||
func(err error, next time.Duration) {
|
||||
attempt++
|
||||
log.Warn("database not ready, retrying",
|
||||
"attempt", attempt, "retry_in", next.String(), "err", err)
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1,7 +0,0 @@
|
||||
// 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,65 @@
|
||||
package infra
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"gopkg.in/natefinch/lumberjack.v2"
|
||||
)
|
||||
|
||||
// NewLogger builds the process logger.
|
||||
//
|
||||
// It always writes JSON to stdout, so `docker logs` and any 12-factor log
|
||||
// collector keep working. When LogFile is set it *also* writes to a
|
||||
// size-rotated file, so logs survive a container rebuild instead of vanishing
|
||||
// with the previous stdout stream. Rotation is delegated to lumberjack rather
|
||||
// than hand-rolled.
|
||||
//
|
||||
// The returned Closer flushes and closes the file; call it on shutdown.
|
||||
func NewLogger(cfg Config) (*slog.Logger, io.Closer, error) {
|
||||
opts := &slog.HandlerOptions{Level: parseLevel(cfg.LogLevel)}
|
||||
|
||||
var (
|
||||
out io.Writer = os.Stdout
|
||||
closer io.Closer = noopCloser{}
|
||||
)
|
||||
|
||||
if cfg.LogFile != "" {
|
||||
if err := os.MkdirAll(filepath.Dir(cfg.LogFile), 0o755); err != nil {
|
||||
return nil, nil, fmt.Errorf("create log directory: %w", err)
|
||||
}
|
||||
rotator := &lumberjack.Logger{
|
||||
Filename: cfg.LogFile,
|
||||
MaxSize: 50, // megabytes before a rotation
|
||||
MaxBackups: 5, // keep this many rotated files
|
||||
MaxAge: 30, // days
|
||||
Compress: true,
|
||||
}
|
||||
// Tee to both: the console stays live while the file is the durable copy.
|
||||
out = io.MultiWriter(os.Stdout, rotator)
|
||||
closer = rotator
|
||||
}
|
||||
|
||||
return slog.New(slog.NewJSONHandler(out, opts)), closer, nil
|
||||
}
|
||||
|
||||
func parseLevel(s string) slog.Level {
|
||||
switch strings.ToLower(strings.TrimSpace(s)) {
|
||||
case "debug":
|
||||
return slog.LevelDebug
|
||||
case "warn", "warning":
|
||||
return slog.LevelWarn
|
||||
case "error":
|
||||
return slog.LevelError
|
||||
default:
|
||||
return slog.LevelInfo
|
||||
}
|
||||
}
|
||||
|
||||
type noopCloser struct{}
|
||||
|
||||
func (noopCloser) Close() error { return nil }
|
||||
@@ -0,0 +1,65 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
|
||||
)
|
||||
|
||||
// WorkerRepo implements usecase.WorkerRepository.
|
||||
type WorkerRepo struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewWorkerRepo(pool *pgxpool.Pool) *WorkerRepo {
|
||||
return &WorkerRepo{pool: pool}
|
||||
}
|
||||
|
||||
const workerColumns = `id, name, capabilities, status, last_heartbeat_at, created_at, updated_at`
|
||||
|
||||
const insertWorkerSQL = `
|
||||
INSERT INTO workers (` + workerColumns + `)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)`
|
||||
|
||||
func (r *WorkerRepo) Insert(ctx context.Context, w *domain.Worker) error {
|
||||
// capabilities is a jsonb column; pgx marshals the []string to a JSON array.
|
||||
_, err := conn(ctx, r.pool).Exec(ctx, insertWorkerSQL,
|
||||
w.ID, w.Name, w.Capabilities, string(w.Status),
|
||||
w.LastHeartbeatAt, w.CreatedAt, w.UpdatedAt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("insert worker: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const getWorkerSQL = `SELECT ` + workerColumns + ` FROM workers WHERE id = $1`
|
||||
|
||||
func (r *WorkerRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Worker, error) {
|
||||
w, err := scanWorker(conn(ctx, r.pool).QueryRow(ctx, getWorkerSQL, id))
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, domain.ErrWorkerNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get worker: %w", err)
|
||||
}
|
||||
return w, nil
|
||||
}
|
||||
|
||||
func scanWorker(row pgx.Row) (*domain.Worker, error) {
|
||||
var (
|
||||
w domain.Worker
|
||||
status string
|
||||
)
|
||||
if err := row.Scan(&w.ID, &w.Name, &w.Capabilities, &status,
|
||||
&w.LastHeartbeatAt, &w.CreatedAt, &w.UpdatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
w.Status = domain.WorkerStatus(status)
|
||||
return &w, nil
|
||||
}
|
||||
@@ -28,9 +28,24 @@ type chunkDTO struct {
|
||||
MaxAttempts int `json:"max_attempts"`
|
||||
}
|
||||
|
||||
type registerRequest struct {
|
||||
Name string `json:"name"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
// Accepted per the contract for forward compatibility; not yet persisted.
|
||||
CPUCount int `json:"cpu_count"`
|
||||
MemoryMB int `json:"memory_mb"`
|
||||
}
|
||||
|
||||
type registerResponse struct {
|
||||
WorkerID uuid.UUID `json:"worker_id"`
|
||||
HeartbeatIntervalSeconds int `json:"heartbeat_interval_seconds"`
|
||||
}
|
||||
|
||||
type claimRequest struct {
|
||||
WorkerID string `json:"worker_id"`
|
||||
Workloads []string `json:"workloads"`
|
||||
WorkerID string `json:"worker_id"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
// Accepted per the contract; the coordinator leases one task per call.
|
||||
MaxConcurrency int `json:"max_concurrency"`
|
||||
}
|
||||
|
||||
type heartbeatRequest struct {
|
||||
|
||||
@@ -33,7 +33,8 @@ func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) {
|
||||
switch {
|
||||
case errors.Is(err, domain.ErrInvalidInput):
|
||||
status = http.StatusBadRequest
|
||||
case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound):
|
||||
case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound),
|
||||
errors.Is(err, domain.ErrWorkerNotFound):
|
||||
status = http.StatusNotFound
|
||||
case errors.Is(err, domain.ErrLeaseConflict),
|
||||
errors.Is(err, domain.ErrStaleAttempt),
|
||||
|
||||
@@ -40,6 +40,30 @@ func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusCreated, jobResponse{ID: job.ID, Status: string(job.Status)})
|
||||
}
|
||||
|
||||
func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := s.reqCtx(r)
|
||||
defer cancel()
|
||||
|
||||
var req registerRequest
|
||||
if err := decodeJSON(r, &req); err != nil {
|
||||
s.writeError(w, r, domain.ErrInvalidInput)
|
||||
return
|
||||
}
|
||||
|
||||
worker, err := s.uc.RegisterWorker.Execute(ctx, usecase.RegisterWorkerInput{
|
||||
Name: req.Name,
|
||||
Capabilities: req.Capabilities,
|
||||
})
|
||||
if err != nil {
|
||||
s.writeError(w, r, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, registerResponse{
|
||||
WorkerID: worker.ID,
|
||||
HeartbeatIntervalSeconds: int(s.heartbeatInterval.Seconds()),
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := s.reqCtx(r)
|
||||
defer cancel()
|
||||
@@ -52,7 +76,7 @@ func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
claimed, err := s.uc.ClaimTask.Execute(ctx, usecase.ClaimTaskInput{
|
||||
WorkerID: req.WorkerID,
|
||||
Workloads: req.Workloads,
|
||||
Workloads: req.Capabilities,
|
||||
})
|
||||
if err != nil {
|
||||
s.writeError(w, r, err)
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
@@ -15,28 +16,41 @@ import (
|
||||
// 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
|
||||
RegisterWorker *usecase.RegisterWorker
|
||||
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
|
||||
uc UseCases
|
||||
log *slog.Logger
|
||||
requestTimeout time.Duration
|
||||
heartbeatInterval time.Duration
|
||||
// ready probes downstream dependencies (the database) for /health. Kept as
|
||||
// a func so the transport layer never imports pgx.
|
||||
ready func(context.Context) error
|
||||
}
|
||||
|
||||
func NewServer(uc UseCases, log *slog.Logger, requestTimeout time.Duration) *Server {
|
||||
return &Server{uc: uc, log: log, requestTimeout: requestTimeout}
|
||||
func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration,
|
||||
ready func(context.Context) error) *Server {
|
||||
return &Server{
|
||||
uc: uc,
|
||||
log: log,
|
||||
requestTimeout: requestTimeout,
|
||||
heartbeatInterval: heartbeatInterval,
|
||||
ready: ready,
|
||||
}
|
||||
}
|
||||
|
||||
// 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 /workers/register", s.handleRegister)
|
||||
protected.HandleFunc("POST /jobs", s.handleCreateJob)
|
||||
protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
|
||||
protected.HandleFunc("POST /tasks/claim", s.handleClaim)
|
||||
@@ -45,7 +59,6 @@ func (s *Server) Handler(token string) http.Handler {
|
||||
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,
|
||||
@@ -55,6 +68,16 @@ func (s *Server) Handler(token string) http.Handler {
|
||||
return mux
|
||||
}
|
||||
|
||||
// handleHealth reports readiness. It probes the database so an orchestrator
|
||||
// learns the difference between "process is up" and "process can serve".
|
||||
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
if s.ready != nil {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
|
||||
defer cancel()
|
||||
if err := s.ready(ctx); err != nil {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
|
||||
return
|
||||
}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
@@ -21,6 +21,11 @@ type ChunkInput struct {
|
||||
MaxAttempts int
|
||||
}
|
||||
|
||||
type RegisterWorkerInput struct {
|
||||
Name string
|
||||
Capabilities []string
|
||||
}
|
||||
|
||||
type ClaimTaskInput struct {
|
||||
WorkerID string
|
||||
Workloads []string
|
||||
|
||||
@@ -62,6 +62,12 @@ type JobRepository interface {
|
||||
UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error
|
||||
}
|
||||
|
||||
// WorkerRepository persists the worker registry.
|
||||
type WorkerRepository interface {
|
||||
Insert(ctx context.Context, w *domain.Worker) error
|
||||
Get(ctx context.Context, id uuid.UUID) (*domain.Worker, 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.
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
|
||||
)
|
||||
|
||||
// RegisterWorker records a worker in the registry and hands back its identity.
|
||||
type RegisterWorker struct {
|
||||
workers WorkerRepository
|
||||
clk Clock
|
||||
}
|
||||
|
||||
func NewRegisterWorker(workers WorkerRepository, clk Clock) *RegisterWorker {
|
||||
return &RegisterWorker{workers: workers, clk: clk}
|
||||
}
|
||||
|
||||
func (uc *RegisterWorker) Execute(ctx context.Context, in RegisterWorkerInput) (*domain.Worker, error) {
|
||||
w, err := domain.NewWorker(in.Name, in.Capabilities, uc.clk.Now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := uc.workers.Insert(ctx, w); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return w, nil
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
BEGIN;
|
||||
|
||||
DROP TABLE IF EXISTS workers;
|
||||
DROP TYPE IF EXISTS worker_status;
|
||||
|
||||
COMMIT;
|
||||
@@ -0,0 +1,20 @@
|
||||
BEGIN;
|
||||
|
||||
CREATE TYPE worker_status AS ENUM ('online','busy','offline');
|
||||
|
||||
-- A registered process/machine that can claim tasks. Registration returns the
|
||||
-- id; liveness is tracked by last_heartbeat_at.
|
||||
CREATE TABLE workers (
|
||||
id uuid PRIMARY KEY,
|
||||
name text NOT NULL DEFAULT '',
|
||||
capabilities jsonb NOT NULL DEFAULT '[]'::jsonb,
|
||||
status worker_status NOT NULL DEFAULT 'online',
|
||||
last_heartbeat_at timestamptz NOT NULL DEFAULT now(),
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
updated_at timestamptz NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Liveness sweep: find workers that have gone quiet.
|
||||
CREATE INDEX ix_workers_liveness ON workers (status, last_heartbeat_at);
|
||||
|
||||
COMMIT;
|
||||
@@ -45,6 +45,13 @@ check "GET /health" 200 "${HOST}/health"
|
||||
check "claim without a token → 401" 401 -X POST "${HOST}/tasks/claim" \
|
||||
-H 'Content-Type: application/json' -d '{"worker_id":"w1"}'
|
||||
|
||||
echo
|
||||
echo "worker registry"
|
||||
check "register worker" 201 -X POST "${HOST}/workers/register" "${auth[@]}" \
|
||||
-d '{"name":"smoke-worker","capabilities":["similarity_search"],"cpu_count":4,"memory_mb":8192}'
|
||||
check "register without capabilities → 400" 400 -X POST "${HOST}/workers/register" "${auth[@]}" \
|
||||
-d '{"name":"bad"}'
|
||||
|
||||
echo
|
||||
echo "job lifecycle"
|
||||
job=$(curl -sS "${auth[@]}" -X POST "${HOST}/jobs" -d '{
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
# SciMesh coordinator ↔ worker API contract (v1)
|
||||
|
||||
**Status marker:** `v1`. This document is the single source of truth for the Go
|
||||
coordinator and the Python Worker Daemon. It is derived from `PLAN.md` §5 and
|
||||
must be updated in the same change as any behaviour it describes.
|
||||
|
||||
- **Auth:** every endpoint except readiness requires `Authorization: Bearer <token>`.
|
||||
- **Identity:** every mutating worker request carries `worker_id` and `attempt`;
|
||||
they are checked against the current task lease in PostgreSQL. A stale attempt
|
||||
gets `409`.
|
||||
- **Timestamps:** UTC, RFC 3339 (e.g. `2026-07-22T12:05:00Z`).
|
||||
- **Unknown JSON fields are rejected** with `400`.
|
||||
|
||||
## Implementation status
|
||||
|
||||
| Endpoint | Contract | Coordinator |
|
||||
| --- | --- | --- |
|
||||
| `GET /health` | readiness incl. DB | ✅ done |
|
||||
| `POST /workers/register` | register + capabilities | ✅ done |
|
||||
| `POST /tasks/claim` | atomic lease | ✅ done |
|
||||
| `POST /tasks/{id}/heartbeat` | renew lease | ✅ done |
|
||||
| `POST /tasks/{id}/result` | complete | 🟡 done, but result is a URI today; moves to `artifact_id` in CTX-05 |
|
||||
| `POST /tasks/{id}/failure` | fail | ✅ done |
|
||||
| `GET /jobs/{id}` | progress | ✅ done |
|
||||
| `GET /tasks/{id}/input` | download shard | ❌ CTX-05 |
|
||||
| `PUT /tasks/{id}/artifacts/{name}` | upload partial | ❌ CTX-05 |
|
||||
|
||||
---
|
||||
|
||||
## Readiness
|
||||
|
||||
```http
|
||||
GET /health
|
||||
```
|
||||
|
||||
`200 {"status":"ok"}` when the database is reachable; `503 {"status":"unavailable"}`
|
||||
otherwise. Unauthenticated.
|
||||
|
||||
## Register worker
|
||||
|
||||
```http
|
||||
POST /workers/register
|
||||
Authorization: Bearer <token>
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"name": "lab-worker-01",
|
||||
"capabilities": ["similarity-search", "similarity-graph"],
|
||||
"cpu_count": 8,
|
||||
"memory_mb": 16384
|
||||
}
|
||||
```
|
||||
|
||||
`201`:
|
||||
|
||||
```json
|
||||
{ "worker_id": "uuid", "heartbeat_interval_seconds": 15 }
|
||||
```
|
||||
|
||||
`cpu_count`/`memory_mb` are accepted for forward compatibility and not yet
|
||||
persisted. `capabilities` must be non-empty (an allowlisted workload set).
|
||||
|
||||
## Claim task
|
||||
|
||||
```http
|
||||
POST /tasks/claim
|
||||
Authorization: Bearer <token>
|
||||
Content-Type: application/json
|
||||
|
||||
{ "worker_id": "uuid", "capabilities": ["similarity-search"], "max_concurrency": 1 }
|
||||
```
|
||||
|
||||
- `204 No Content`: no compatible task.
|
||||
- `200 OK`: a task is leased atomically.
|
||||
|
||||
```json
|
||||
{
|
||||
"task_id": "uuid",
|
||||
"attempt": 1,
|
||||
"lease_expires_at": "2026-07-22T12:05:00Z",
|
||||
"workload": "similarity-search",
|
||||
"input": { "uri": "https://coordinator/tasks/uuid/input", "sha256": "hex" },
|
||||
"parameters": { "query_id": "CHEMBL939", "top_k": 20 }
|
||||
}
|
||||
```
|
||||
|
||||
`max_concurrency` is accepted; the coordinator leases one task per call for now.
|
||||
|
||||
## Renew lease (heartbeat)
|
||||
|
||||
```http
|
||||
POST /tasks/{task_id}/heartbeat
|
||||
Authorization: Bearer <token>
|
||||
Content-Type: application/json
|
||||
|
||||
{ "worker_id": "uuid", "attempt": 1 }
|
||||
```
|
||||
|
||||
Response **must** contain a renewed deadline:
|
||||
|
||||
```json
|
||||
{ "lease_expires_at": "2026-07-22T12:10:00Z" }
|
||||
```
|
||||
|
||||
The worker schedules the next heartbeat before half of the returned TTL, never
|
||||
on a fixed interval alone.
|
||||
|
||||
## Download input or shard (CTX-05)
|
||||
|
||||
`GET /tasks/{task_id}/input` returns the artifact owned by the current task. The
|
||||
worker verifies its SHA-256 before execution. If the URI redirects to another
|
||||
origin, the worker removes the coordinator bearer token.
|
||||
|
||||
## Upload a partial artifact (CTX-05)
|
||||
|
||||
```http
|
||||
PUT /tasks/{task_id}/artifacts/{filename}
|
||||
Authorization: Bearer <token>
|
||||
Content-Type: text/csv
|
||||
X-Worker-ID: uuid
|
||||
X-Task-Attempt: 1
|
||||
|
||||
<streamed bytes>
|
||||
```
|
||||
|
||||
`200`:
|
||||
|
||||
```json
|
||||
{ "artifact_id": "uuid", "uri": "https://coordinator/artifacts/uuid/download",
|
||||
"sha256": "hex", "size_bytes": 1234 }
|
||||
```
|
||||
|
||||
## Complete or fail task
|
||||
|
||||
```http
|
||||
POST /tasks/{task_id}/result
|
||||
Authorization: Bearer <token>
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"worker_id": "uuid",
|
||||
"attempt": 1,
|
||||
"result": { "artifact_id": "uuid", "sha256": "hex", "content_type": "text/csv" },
|
||||
"metrics": { "elapsed_seconds": 12.4, "processed_rows": 10000 }
|
||||
}
|
||||
```
|
||||
|
||||
> **Transitional:** the coordinator currently accepts `result_uri` + `result_sha256`
|
||||
> instead of `result.artifact_id`. This switches to the artifact form in CTX-05,
|
||||
> once upload exists. Until then no worker-supplied `file://`/`worker://` URI is
|
||||
> valid in persisted metadata.
|
||||
|
||||
```http
|
||||
POST /tasks/{task_id}/failure
|
||||
```
|
||||
|
||||
Same identity fields, plus sanitized `error_code`, `error_message`, `retryable`.
|
||||
Never a traceback, token, or absolute worker path.
|
||||
|
||||
## Idempotency and errors
|
||||
|
||||
| Situation | Response |
|
||||
| --- | --- |
|
||||
| No compatible task | `204` |
|
||||
| Worker/attempt does not own lease | `409` |
|
||||
| Artifact does not belong to task/attempt | `409` |
|
||||
| Same completion, same manifest | `200` idempotent |
|
||||
| Same attempt, different manifest | `409` |
|
||||
| Invalid parameters/input | `400` |
|
||||
| Auth failure | `401` |
|
||||
| Unknown job/task | `404` |
|
||||
Reference in New Issue
Block a user