diff --git a/coordinator/.env.example b/coordinator/.env.example index 82ae474..1e8f90f 100644 --- a/coordinator/.env.example +++ b/coordinator/.env.example @@ -11,6 +11,9 @@ WORKER_AUTH_TOKEN=change-me LOG_LEVEL=info # LOG_FILE=./logs/coordinator.log +# Directory where artifact bytes are stored. +COORDINATOR_STORAGE_DIR=./data + # Optional tuning (defaults shown). DB_MAX_CONNS=10 # How long to keep retrying the initial DB connection while Postgres boots. diff --git a/coordinator/.gitignore b/coordinator/.gitignore index b517da9..c8c4f7b 100644 --- a/coordinator/.gitignore +++ b/coordinator/.gitignore @@ -3,3 +3,4 @@ .env *.out /logs/ +/data/ diff --git a/coordinator/api/requests.http b/coordinator/api/requests.http index 75f4bf8..181a3b5 100644 --- a/coordinator/api/requests.http +++ b/coordinator/api/requests.http @@ -80,6 +80,24 @@ Content-Type: application/json "attempt": {{attempt}} } +### 3a. Upload a partial-result artifact (200) — while the task is leased +# Identity travels in headers per the contract; the body is streamed as-is. +# @name uploadArtifact +PUT {{host}}/tasks/{{taskId}}/artifacts/result.csv +Authorization: Bearer {{token}} +Content-Type: text/csv +X-Worker-ID: {{worker}} +X-Task-Attempt: {{attempt}} + +query,match,score +CHEMBL25,CHEMBL139,0.87 + +@artifactId = {{uploadArtifact.response.body.artifact_id}} + +### 3b. Download the artifact by id (200) +GET {{host}}/artifacts/{{artifactId}}/download +Authorization: Bearer {{token}} + ### 4. Submit the result (200) POST {{host}}/tasks/{{taskId}}/result Authorization: Bearer {{token}} diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 5de6d5b..32e0d45 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -9,6 +9,7 @@ import ( "syscall" "github.com/emil28092005/SciMesh/coordinator/internal/infra" + "github.com/emil28092005/SciMesh/coordinator/internal/storage/blob" "github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres" httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http" "github.com/emil28092005/SciMesh/coordinator/internal/usecase" @@ -51,22 +52,31 @@ func run() error { } defer pool.Close() + blobStore, err := blob.NewFSStore(cfg.StorageDir) + if err != nil { + log.Error("init blob storage", "err", err) + return err + } + var ( - clk = infra.NewClock() - tx = postgres.NewTxManager(pool) - taskRepo = postgres.NewTaskRepo(pool) - jobRepo = postgres.NewJobRepo(pool) - workerRepo = postgres.NewWorkerRepo(pool) + clk = infra.NewClock() + tx = postgres.NewTxManager(pool) + taskRepo = postgres.NewTaskRepo(pool) + jobRepo = postgres.NewJobRepo(pool) + workerRepo = postgres.NewWorkerRepo(pool) + artifactRepo = postgres.NewArtifactRepo(pool) ) useCases := httptransport.UseCases{ - 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), + 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), + UploadArtifact: usecase.NewUploadArtifact(taskRepo, artifactRepo, blobStore, clk), + DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore), } // Background workers are tracked so shutdown can wait for them. Without diff --git a/coordinator/docker-compose.yml b/coordinator/docker-compose.yml index 03cd92b..5f4c11f 100644 --- a/coordinator/docker-compose.yml +++ b/coordinator/docker-compose.yml @@ -57,10 +57,13 @@ services: # 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 + # Artifact bytes land on the mounted ./data directory, durable across rebuilds. + COORDINATOR_STORAGE_DIR: /var/lib/scimesh/artifacts ports: - "${COORDINATOR_PORT:-8080}:8080" volumes: - ./logs:/var/log/scimesh + - ./data:/var/lib/scimesh/artifacts healthcheck: test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"] interval: 10s diff --git a/coordinator/internal/storage/blob/store.go b/coordinator/internal/storage/blob/store.go index b7efafd..7c20eac 100644 --- a/coordinator/internal/storage/blob/store.go +++ b/coordinator/internal/storage/blob/store.go @@ -92,6 +92,18 @@ func (s *FSStore) Open(ctx context.Context, key string) (io.ReadCloser, error) { return f, nil } +// Delete removes a stored blob. Absence is not an error: cleaning up after a +// failed metadata insert must be idempotent. +func (s *FSStore) Delete(ctx context.Context, key string) error { + if err := checkKey(key); err != nil { + return err + } + if err := os.Remove(filepath.Join(s.dir, key)); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} + // checkKey rejects anything that could escape the storage directory. Keys are // coordinator-generated UUIDs, so this is defence in depth, not the only guard. func checkKey(key string) error { diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go index 3b26038..d9e75df 100644 --- a/coordinator/internal/storage/postgres/task_repo.go +++ b/coordinator/internal/storage/postgres/task_repo.go @@ -120,6 +120,25 @@ func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domai return task, nil } +// Get reads a task without locking its row. +func (r *TaskRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Task, error) { + sql, args, err := psql.Select(taskColumns...). + From("tasks"). + Where(sq.Eq{"id": id}). + ToSql() + if err != nil { + return nil, err + } + t, err := scanTask(conn(ctx, r.pool).QueryRow(ctx, sql, args...)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, domain.ErrTaskNotFound + } + if err != nil { + return nil, err + } + return t, nil +} + // GetForUpdate reads a task and holds its row lock until the caller's // transaction ends, so read-modify-write use cases cannot interleave. func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) { diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go index 99db4f4..6523c09 100644 --- a/coordinator/internal/transport/http/dto.go +++ b/coordinator/internal/transport/http/dto.go @@ -102,6 +102,13 @@ type jobProgressResponse struct { Failed int `json:"failed"` } +type uploadArtifactResponse struct { + ArtifactID uuid.UUID `json:"artifact_id"` + URI string `json:"uri"` + SHA256 string `json:"sha256"` + SizeBytes int64 `json:"size_bytes"` +} + type errorResponse struct { Error string `json:"error"` RequestID string `json:"request_id,omitempty"` diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go index 0abaa1a..04e5ae7 100644 --- a/coordinator/internal/transport/http/errors.go +++ b/coordinator/internal/transport/http/errors.go @@ -34,7 +34,7 @@ func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) { case errors.Is(err, domain.ErrInvalidInput): status = http.StatusBadRequest case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound), - errors.Is(err, domain.ErrWorkerNotFound): + errors.Is(err, domain.ErrWorkerNotFound), errors.Is(err, domain.ErrArtifactNotFound): 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 2e96adf..9dbe659 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/coordinator/internal/transport/http/handlers.go @@ -2,7 +2,10 @@ package http import ( "context" + "fmt" + "io" "net/http" + "strconv" "github.com/google/uuid" @@ -173,6 +176,60 @@ func (s *Server) handleFailure(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)}) } +// handleUploadArtifact streams a worker's partial result into blob storage. It +// deliberately does not use the short request timeout — a large shard upload +// would trip it — and reads identity from headers per the contract (§5.5). +func (s *Server) handleUploadArtifact(w http.ResponseWriter, r *http.Request) { + taskID, ok := s.pathUUID(w, r, "task_id") + if !ok { + return + } + attempt, err := strconv.Atoi(r.Header.Get("X-Task-Attempt")) + if err != nil { + s.writeError(w, r, domain.ErrInvalidInput) + return + } + + art, err := s.uc.UploadArtifact.Execute(r.Context(), usecase.UploadArtifactInput{ + TaskID: taskID, + WorkerID: r.Header.Get("X-Worker-ID"), + Attempt: attempt, + Filename: r.PathValue("filename"), + ContentType: r.Header.Get("Content-Type"), + Body: r.Body, + }) + if err != nil { + s.writeError(w, r, err) + return + } + writeJSON(w, http.StatusOK, uploadArtifactResponse{ + ArtifactID: art.ID, + URI: "/artifacts/" + art.ID.String() + "/download", + SHA256: art.SHA256, + SizeBytes: art.SizeBytes, + }) +} + +// handleDownloadArtifact streams an artifact's bytes back to the caller. +func (s *Server) handleDownloadArtifact(w http.ResponseWriter, r *http.Request) { + artifactID, ok := s.pathUUID(w, r, "artifact_id") + if !ok { + return + } + art, body, err := s.uc.DownloadArtifact.Execute(r.Context(), artifactID) + if err != nil { + s.writeError(w, r, err) + return + } + defer body.Close() + + w.Header().Set("Content-Type", art.ContentType) + w.Header().Set("Content-Length", strconv.FormatInt(art.SizeBytes, 10)) + w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", art.Filename)) + w.Header().Set("X-Checksum-SHA256", art.SHA256) + _, _ = io.Copy(w, body) +} + func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { ctx, cancel := s.reqCtx(r) defer cancel() diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 7b6e5f0..6a65d60 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -16,13 +16,15 @@ 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 { - RegisterWorker *usecase.RegisterWorker - 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 + UploadArtifact *usecase.UploadArtifact + DownloadArtifact *usecase.DownloadArtifact } type Server struct { @@ -57,6 +59,8 @@ func (s *Server) Handler(token string) http.Handler { 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) + protected.HandleFunc("PUT /tasks/{task_id}/artifacts/{filename}", s.handleUploadArtifact) + protected.HandleFunc("GET /artifacts/{artifact_id}/download", s.handleDownloadArtifact) mux := http.NewServeMux() mux.HandleFunc("GET /health", s.handleHealth) diff --git a/coordinator/internal/usecase/artifact.go b/coordinator/internal/usecase/artifact.go new file mode 100644 index 0000000..a515d09 --- /dev/null +++ b/coordinator/internal/usecase/artifact.go @@ -0,0 +1,80 @@ +package usecase + +import ( + "context" + "io" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// UploadArtifact stores a worker's partial-result bytes and records the metadata. +type UploadArtifact struct { + tasks TaskRepository + artifacts ArtifactRepository + blobs BlobStore + clk Clock +} + +func NewUploadArtifact(tasks TaskRepository, artifacts ArtifactRepository, + blobs BlobStore, clk Clock) *UploadArtifact { + return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, clk: clk} +} + +func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (*domain.Artifact, error) { + task, err := uc.tasks.Get(ctx, in.TaskID) + if err != nil { + return nil, err + } + // Only the worker holding the current lease at this attempt may upload the + // task's output — the coordinator never trusts an ownership claim on faith. + if !task.IsLeaseHeldBy(in.WorkerID, in.Attempt) { + return nil, domain.ErrLeaseConflict + } + + taskID := task.ID + art, err := domain.NewArtifact(task.JobID, &taskID, domain.ArtifactPartialResult, + in.Filename, in.ContentType, uc.clk.Now()) + if err != nil { + return nil, err + } + + // Stream to storage first: size and checksum are measured here, by us, not + // taken from the worker. A large shard never sits in memory. + sum, size, err := uc.blobs.Put(ctx, art.StorageKey, in.Body) + if err != nil { + return nil, err + } + art.SetContent(sum, size) + + // Persist the record. If that fails the blob would be an orphan, so remove it. + if err := uc.artifacts.Insert(ctx, art); err != nil { + _ = uc.blobs.Delete(ctx, art.StorageKey) + return nil, err + } + return art, nil +} + +// DownloadArtifact returns an artifact's metadata together with a reader over +// its bytes. The caller must close the reader. +type DownloadArtifact struct { + artifacts ArtifactRepository + blobs BlobStore +} + +func NewDownloadArtifact(artifacts ArtifactRepository, blobs BlobStore) *DownloadArtifact { + return &DownloadArtifact{artifacts: artifacts, blobs: blobs} +} + +func (uc *DownloadArtifact) Execute(ctx context.Context, id uuid.UUID) (*domain.Artifact, io.ReadCloser, error) { + a, err := uc.artifacts.Get(ctx, id) + if err != nil { + return nil, nil, err + } + rc, err := uc.blobs.Open(ctx, a.StorageKey) + if err != nil { + return nil, nil, err + } + return a, rc, nil +} diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go index 4733c54..de4839c 100644 --- a/coordinator/internal/usecase/dto.go +++ b/coordinator/internal/usecase/dto.go @@ -1,6 +1,10 @@ package usecase -import "github.com/google/uuid" +import ( + "io" + + "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. @@ -46,6 +50,15 @@ type CompleteTaskInput struct { Metrics map[string]any } +type UploadArtifactInput struct { + TaskID uuid.UUID + WorkerID string + Attempt int + Filename string + ContentType string + Body io.Reader +} + type FailTaskInput struct { TaskID uuid.UUID WorkerID string diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index 747bcef..1269e42 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -35,6 +35,11 @@ type TaskRepository interface { // Returns (nil, nil) when nothing is available. ClaimNext(ctx context.Context, f ClaimFilter) (*domain.Task, error) + // Get reads a task without locking. Use it for read-only checks (e.g. + // verifying lease ownership before a long upload) where holding a row lock + // across the operation would be wrong. + Get(ctx context.Context, id uuid.UUID) (*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) @@ -83,6 +88,9 @@ type ArtifactRepository interface { type BlobStore interface { Put(ctx context.Context, key string, r io.Reader) (sha256 string, size int64, err error) Open(ctx context.Context, key string) (io.ReadCloser, error) + // Delete removes a stored blob. Used to clean up after a metadata insert + // fails, so a committed blob never outlives its (absent) record. + Delete(ctx context.Context, key string) error } // TxManager runs a function inside one database transaction. The transaction diff --git a/coordinator/scripts/smoke.sh b/coordinator/scripts/smoke.sh index 5e516aa..70c3550 100755 --- a/coordinator/scripts/smoke.sh +++ b/coordinator/scripts/smoke.sh @@ -101,6 +101,22 @@ fi check "heartbeat" 200 -X POST "${HOST}/tasks/${task_id}/heartbeat" "${auth[@]}" \ -d "{\"worker_id\":\"w1\",\"attempt\":${attempt}}" + +# --- artifacts (while the task is still leased) --------------------------- +bearer=(-H "Authorization: Bearer ${TOKEN}") +check "upload artifact" 200 -X PUT "${HOST}/tasks/${task_id}/artifacts/result.csv" "${bearer[@]}" \ + -H 'Content-Type: text/csv' -H 'X-Worker-ID: w1' -H "X-Task-Attempt: ${attempt}" \ + --data-binary $'query,match,score\nA,B,0.9\n' +check "foreign worker upload → 409" 409 -X PUT "${HOST}/tasks/${task_id}/artifacts/x.csv" "${bearer[@]}" \ + -H 'Content-Type: text/csv' -H 'X-Worker-ID: impostor' -H "X-Task-Attempt: ${attempt}" \ + --data-binary 'x' + +# Round-trip: upload one more, then download it by id and confirm the bytes. +art=$(curl -sS -X PUT "${HOST}/tasks/${task_id}/artifacts/dl.csv" "${bearer[@]}" \ + -H 'Content-Type: text/csv' -H 'X-Worker-ID: w1' -H "X-Task-Attempt: ${attempt}" --data-binary 'a,b,c') +art_id=$(printf '%s' "$art" | python3 -c 'import json,sys;print(json.load(sys.stdin)["artifact_id"])' 2>/dev/null) +check "download artifact" 200 "${HOST}/artifacts/${art_id}/download" "${bearer[@]}" + check "foreign worker submits → 409" 409 -X POST "${HOST}/tasks/${task_id}/result" "${auth[@]}" \ -d "{\"worker_id\":\"impostor\",\"attempt\":${attempt},\"result_uri\":\"s3://x\",\"result_sha256\":\"x\"}" check "submit result" 200 -X POST "${HOST}/tasks/${task_id}/result" "${auth[@]}" \ diff --git a/docs/api-contract.md b/docs/api-contract.md index dd7b05c..603dc20 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -19,11 +19,12 @@ must be updated in the same change as any behaviour it describes. | `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}/result` | complete | 🟡 done, but result is a URI today; moves to `artifact_id` next | | `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 | +| `PUT /tasks/{id}/artifacts/{name}` | upload partial | ✅ done | +| `GET /artifacts/{id}/download` | download by id | ✅ done | +| `GET /tasks/{id}/input` | download shard | ❌ needs input artifacts (upload+chunking) | ---