diff --git a/coordinator/.env.example b/coordinator/.env.example index 2887295..82ae474 100644 --- a/coordinator/.env.example +++ b/coordinator/.env.example @@ -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 diff --git a/coordinator/.gitignore b/coordinator/.gitignore index c44a5e4..b517da9 100644 --- a/coordinator/.gitignore +++ b/coordinator/.gitignore @@ -2,3 +2,4 @@ /bin/ .env *.out +/logs/ diff --git a/coordinator/api/requests.http b/coordinator/api/requests.http index 5a4b356..75f4bf8 100644 --- a/coordinator/api/requests.http +++ b/coordinator/api/requests.http @@ -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}} diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index d2fb7ec..5de6d5b 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -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). diff --git a/coordinator/docker-compose.yml b/coordinator/docker-compose.yml index 1cbe5b9..03cd92b 100644 --- a/coordinator/docker-compose.yml +++ b/coordinator/docker-compose.yml @@ -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 diff --git a/coordinator/go.mod b/coordinator/go.mod index dd02541..e424a5e 100644 --- a/coordinator/go.mod +++ b/coordinator/go.mod @@ -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 ( diff --git a/coordinator/go.sum b/coordinator/go.sum index c078c1d..7e95bc7 100644 --- a/coordinator/go.sum +++ b/coordinator/go.sum @@ -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= diff --git a/coordinator/internal/domain/errors.go b/coordinator/internal/domain/errors.go index f1e69e8..6603844 100644 --- a/coordinator/internal/domain/errors.go +++ b/coordinator/internal/domain/errors.go @@ -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") diff --git a/coordinator/internal/domain/worker.go b/coordinator/internal/domain/worker.go new file mode 100644 index 0000000..e7e0bf9 --- /dev/null +++ b/coordinator/internal/domain/worker.go @@ -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 +} diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go index db07876..ce8bc12 100644 --- a/coordinator/internal/infra/config.go +++ b/coordinator/internal/infra/config.go @@ -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 } diff --git a/coordinator/internal/infra/db.go b/coordinator/internal/infra/db.go index 29899b3..d1135cd 100644 --- a/coordinator/internal/infra/db.go +++ b/coordinator/internal/infra/db.go @@ -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) + }, + ) +} diff --git a/coordinator/internal/infra/doc.go b/coordinator/internal/infra/doc.go deleted file mode 100644 index 4e64f47..0000000 --- a/coordinator/internal/infra/doc.go +++ /dev/null @@ -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 diff --git a/coordinator/internal/infra/logging.go b/coordinator/internal/infra/logging.go new file mode 100644 index 0000000..4d12e61 --- /dev/null +++ b/coordinator/internal/infra/logging.go @@ -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 } diff --git a/coordinator/internal/storage/postgres/worker_repo.go b/coordinator/internal/storage/postgres/worker_repo.go new file mode 100644 index 0000000..68f6aaf --- /dev/null +++ b/coordinator/internal/storage/postgres/worker_repo.go @@ -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 +} diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go index b308843..99db4f4 100644 --- a/coordinator/internal/transport/http/dto.go +++ b/coordinator/internal/transport/http/dto.go @@ -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 { diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go index 933b574..0abaa1a 100644 --- a/coordinator/internal/transport/http/errors.go +++ b/coordinator/internal/transport/http/errors.go @@ -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), diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go index 00e4f0e..2e96adf 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/coordinator/internal/transport/http/handlers.go @@ -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) diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index d2731a3..7b6e5f0 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -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"}) } diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go index d7cb76c..4733c54 100644 --- a/coordinator/internal/usecase/dto.go +++ b/coordinator/internal/usecase/dto.go @@ -21,6 +21,11 @@ type ChunkInput struct { MaxAttempts int } +type RegisterWorkerInput struct { + Name string + Capabilities []string +} + type ClaimTaskInput struct { WorkerID string Workloads []string diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index af0e06d..183141b 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -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. diff --git a/coordinator/internal/usecase/worker.go b/coordinator/internal/usecase/worker.go new file mode 100644 index 0000000..915879b --- /dev/null +++ b/coordinator/internal/usecase/worker.go @@ -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 +} diff --git a/coordinator/migrations/0002_workers.down.sql b/coordinator/migrations/0002_workers.down.sql new file mode 100644 index 0000000..c01563d --- /dev/null +++ b/coordinator/migrations/0002_workers.down.sql @@ -0,0 +1,6 @@ +BEGIN; + +DROP TABLE IF EXISTS workers; +DROP TYPE IF EXISTS worker_status; + +COMMIT; diff --git a/coordinator/migrations/0002_workers.up.sql b/coordinator/migrations/0002_workers.up.sql new file mode 100644 index 0000000..25547fd --- /dev/null +++ b/coordinator/migrations/0002_workers.up.sql @@ -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; diff --git a/coordinator/scripts/smoke.sh b/coordinator/scripts/smoke.sh index 16702f9..5e516aa 100755 --- a/coordinator/scripts/smoke.sh +++ b/coordinator/scripts/smoke.sh @@ -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 '{ diff --git a/docs/api-contract.md b/docs/api-contract.md new file mode 100644 index 0000000..dd7b05c --- /dev/null +++ b/docs/api-contract.md @@ -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 `. +- **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 +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 +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 +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 +Content-Type: text/csv +X-Worker-ID: uuid +X-Task-Attempt: 1 + + +``` + +`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 +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` |