diff --git a/README.md b/README.md index b51582f..947e27e 100644 --- a/README.md +++ b/README.md @@ -3,10 +3,10 @@ SciMesh is a scientific-workload framework for molecular datasets. Its public CLI runs exact similarity search and sparse similarity-graph construction locally in one Python process; it creates no dense similarity matrix. The Go/PostgreSQL -coordinator and Python worker can run a diagnostic, shard-based -`similarity-search` pipeline locally. Its CSV artifacts are not a global result -until CTX-07--09 add planning and reduction; use the local CLI for scientific -results today. See [`STATUS.md`](STATUS.md). +coordinator and Python worker can run a shard-based `similarity-search` +pipeline locally. After every shard succeeds, the coordinator deterministically +merges its candidates into one final global top-k CSV. See +[`STATUS.md`](STATUS.md). The ChEMBL TSV database is intentionally not included in this repository. Download it separately and pass its path to the commands below. The expected columns are `chembl_id` and `canonical_smiles`. diff --git a/STATUS.md b/STATUS.md index cebfd27..1555d37 100644 --- a/STATUS.md +++ b/STATUS.md @@ -1,7 +1,7 @@ # SciMesh Status **Updated:** 2026-07-24 -**Branch baseline:** `main` at `f953112` (distributed pipeline hardening) +**Branch baseline:** `main` at `6e67daa` (distributed similarity-search) ## Current state @@ -18,8 +18,10 @@ the reference behaviour for future distributed execution: The Go coordinator and its PostgreSQL-backed task lifecycle are implemented: registration, atomic claiming, lease renewal, artifact storage, dataset chunking, result/failure reporting, and job progress. The Python worker now -uses the live coordinator contract; its HTTP path was exercised against a real -Docker PostgreSQL stack on 2026-07-23. +uses the live coordinator contract. Completed similarity-search shard results +are reduced once into a checksum-protected final CSV, which is downloadable +through the coordinator. The full Go checks (including a fresh migration and +real PostgreSQL smoke test) passed on 2026-07-24. ## Milestone tracker @@ -33,30 +35,24 @@ Docker PostgreSQL stack on 2026-07-23. | CTX-05 Artifact storage | Implemented | Coordinator-owned inputs/results, checksum verification, and upload flow. | | CTX-06 Python Worker live-contract alignment | Implemented | Worker completed a real uploaded shard via HTTP on 2026-07-23. | | CTX-07 Distributed workload protocol | Implemented | Versioned Python contract models, registry, strict plan validation, and deterministic reduction ordering are in `scimesh/distributed/`. | -| CTX-08 Distributed similarity-search | Implemented (scientific layer) | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. Coordinator persistence/orchestration remains CTX-09. | -| CTX-09 Reducer and final-result API | Not started | Depends on CTX-07 and CTX-08. | +| CTX-08 Distributed similarity-search | Implemented | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. | +| CTX-09 Reducer and final-result API | Implemented | Atomic `reducing` claim, deterministic coordinator-side top-k reducer, sanitized reducer failure, final artifact persistence, `result_uri`, and final CSV download. | | CTX-10 Distributed similarity-graph | Not started | Local reference exists. | -| CTX-11 Dashboard/operator view | Implemented (diagnostic scope) | Protected local view: job/task/worker status, validated similarity-search upload, diagnostic partial-artifact download, and bounded polling. Final-result reduction remains CTX-09. | +| CTX-11 Dashboard/operator view | Implemented | Protected local view: job/task/worker status, validated similarity-search upload, partial-artifact diagnostics, final-result download, and bounded polling. | | CTX-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. | ## Next recommended assignment -Assign **CTX-09** to the coordinator role: materialize planned shards, -persist them transactionally, invoke the registered reducer once, and expose a -durable final artifact. +Assign **CTX-10** to the distributed-science role: implement deterministic +block-pair planning and reduction for `similarity-graph`. ## Known constraints -- The Python `similarity-search` planner/reducer is implemented, but the Go - coordinator does not yet invoke it or persist its final artifact. The - operator UI labels `partial_result` files as diagnostic and cannot present - them as final output. - Use the local `scimesh` CLI for complete workload results. - The worker/coordinator flow currently accepts both underscore API workload names and hyphenated CLI names while the contract is consolidated. - A real-stack worker test uses a small `query_smiles` shard. The Python - planner resolves `query_id` once and shares `query_smiles`; connecting that - planner to uploaded coordinator jobs belongs to CTX-09. + planner resolves `query_id` once and shares `query_smiles`; the upload UI + currently accepts `query_smiles` only. - The coordinator accepts uploaded distributed jobs only for `similarity-search` with `query_smiles`. It rejects `similarity-graph` until CTX-10 supplies cross-shard pair planning. diff --git a/coordinator/README.md b/coordinator/README.md index 43239cc..eeebca2 100644 --- a/coordinator/README.md +++ b/coordinator/README.md @@ -76,9 +76,10 @@ UI_AUTH_TOKEN='local-ui-secret' make up ``` The UI is disabled by default and never accepts the worker bearer token. -It shows recent jobs, task/worker state, and the per-job partial artifacts. -Those files are explicitly diagnostic until the CTX-09 reducer creates a final -result; the UI does not present them as final scientific output. +It shows recent jobs, task/worker state, diagnostic shard artifacts, and the +final CSV for completed similarity-search jobs. The coordinator enters +`reducing` after the last shard completes, then exposes the final deterministic +global top-k result when merging succeeds. `up` starts three services in order: Postgres waits until `pg_isready` passes, a one-shot `migrate` container applies the schema and exits, and only then does the diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 8447414..a200584 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -75,11 +75,13 @@ func run() error { ClaimTask: usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration), RenewLease: usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration), CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, tx, clk), + ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk), FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), CancelJob: usecase.NewCancelJob(jobRepo, taskRepo, tx, clk), UploadArtifact: usecase.NewUploadArtifact(taskRepo, artifactRepo, blobStore, tx, clk), DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore), + GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)), GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore), Dashboard: usecase.NewDashboard(uiReadRepo), } diff --git a/coordinator/internal/domain/job.go b/coordinator/internal/domain/job.go index 7d6be3d..94ff189 100644 --- a/coordinator/internal/domain/job.go +++ b/coordinator/internal/domain/job.go @@ -11,6 +11,7 @@ type JobStatus string const ( JobPending JobStatus = "pending" JobRunning JobStatus = "running" + JobReducing JobStatus = "reducing" JobCompleted JobStatus = "completed" JobFailed JobStatus = "failed" JobCancelled JobStatus = "cancelled" @@ -18,14 +19,18 @@ const ( // Job is one user submission that fans out into one or more tasks. type Job struct { - ID uuid.UUID - Workload string - InputURI string // external input URI; empty for uploaded datasets - InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions - Parameters map[string]any - Status JobStatus - CreatedAt time.Time - CompletedAt *time.Time + ID uuid.UUID + Workload string + InputURI string // external input URI; empty for uploaded datasets + InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions + ResultArtifactID *uuid.UUID + Parameters map[string]any + Status JobStatus + CreatedAt time.Time + CompletedAt *time.Time + ReducerStartedAt *time.Time + ErrorCode *string + ErrorMessage *string } // NewUploadedJob builds a job whose input was uploaded to the coordinator. The @@ -114,6 +119,8 @@ func (p JobProgress) DeriveStatus() JobStatus { switch { case p.Job.Status == JobCancelled: return JobCancelled + case p.Job.Status == JobReducing: + return JobReducing case p.Total == 0: return JobPending case p.Done == p.Total: diff --git a/coordinator/internal/memstore/memstore.go b/coordinator/internal/memstore/memstore.go index c7f1076..28f4318 100644 --- a/coordinator/internal/memstore/memstore.go +++ b/coordinator/internal/memstore/memstore.go @@ -216,6 +216,51 @@ func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain. return nil } +func (r *JobRepo) ClaimReduction(_ context.Context, id uuid.UUID, startedAt time.Time) (bool, error) { + r.mu.Lock() + defer r.mu.Unlock() + j, ok := r.jobs[id] + if !ok { + return false, domain.ErrJobNotFound + } + if j.Status != domain.JobReducing || j.ReducerStartedAt != nil { + return false, nil + } + j.ReducerStartedAt = &startedAt + return true, nil +} + +func (r *JobRepo) CompleteWithResult(_ context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error { + r.mu.Lock() + defer r.mu.Unlock() + j, ok := r.jobs[id] + if !ok { + return domain.ErrJobNotFound + } + j.ResultArtifactID = &resultArtifactID + j.Status = domain.JobCompleted + j.CompletedAt = &completedAt + j.ReducerStartedAt = nil + j.ErrorCode = nil + j.ErrorMessage = nil + return nil +} + +func (r *JobRepo) FailReduction(_ context.Context, id uuid.UUID, code, message string, completedAt time.Time) error { + r.mu.Lock() + defer r.mu.Unlock() + j, ok := r.jobs[id] + if !ok { + return domain.ErrJobNotFound + } + j.Status = domain.JobFailed + j.CompletedAt = &completedAt + j.ReducerStartedAt = nil + j.ErrorCode = &code + j.ErrorMessage = &message + return nil +} + // --- WorkerRepo ---------------------------------------------------------- type WorkerRepo struct { diff --git a/coordinator/internal/reducer/similarity_search.go b/coordinator/internal/reducer/similarity_search.go new file mode 100644 index 0000000..0d43a53 --- /dev/null +++ b/coordinator/internal/reducer/similarity_search.go @@ -0,0 +1,195 @@ +// Package reducer contains deterministic, coordinator-side result reductions. +package reducer + +import ( + "bytes" + "encoding/csv" + "errors" + "fmt" + "io" + "math" + "sort" + "strconv" +) + +var searchHeader = []string{"rank", "chembl_id", "canonical_smiles", "similarity"} + +type similarityMatch struct { + similarity float64 + id string + smiles string +} + +// ReduceSimilaritySearch streams worker-local top-k CSVs into the exact global +// top-k. Each partial is validated before it can affect the final artifact. +func ReduceSimilaritySearch(partials []io.Reader, parameters map[string]any) ([]byte, error) { + topK, err := positiveInt(parameters["top_k"], 20) + if err != nil { + return nil, err + } + direction, err := thresholdDirection(parameters["threshold_direction"]) + if err != nil { + return nil, err + } + h := &matchHeap{direction: direction} + for _, partial := range partials { + if err := readPartial(partial, direction, func(match similarityMatch) { + if len(h.items) < topK { + heapPush(h, match) + return + } + if better(match, h.items[0], direction) { + h.items[0] = match + heapDown(h, 0) + } + }); err != nil { + return nil, err + } + } + + matches := append([]similarityMatch(nil), h.items...) + sort.Slice(matches, func(i, j int) bool { return better(matches[i], matches[j], direction) }) + var out bytes.Buffer + writer := csv.NewWriter(&out) + if err := writer.Write(searchHeader); err != nil { + return nil, err + } + for index, match := range matches { + if err := writer.Write([]string{ + strconv.Itoa(index + 1), match.id, match.smiles, fmt.Sprintf("%.6f", match.similarity), + }); err != nil { + return nil, err + } + } + writer.Flush() + if err := writer.Error(); err != nil { + return nil, err + } + return out.Bytes(), nil +} + +func readPartial(input io.Reader, direction string, consume func(similarityMatch)) error { + reader := csv.NewReader(input) + header, err := reader.Read() + if err != nil { + return fmt.Errorf("read partial header: %w", err) + } + if !equalStrings(header, searchHeader) { + return fmt.Errorf("partial result has an invalid CSV header") + } + var previous *similarityMatch + for rank := 1; ; rank++ { + row, err := reader.Read() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return fmt.Errorf("read partial row: %w", err) + } + if len(row) != len(searchHeader) || row[0] != strconv.Itoa(rank) { + return fmt.Errorf("partial result has an invalid rank") + } + score, err := strconv.ParseFloat(row[3], 64) + if err != nil || math.IsNaN(score) || math.IsInf(score, 0) || score < 0 || score > 1 { + return fmt.Errorf("partial result has an invalid similarity") + } + match := similarityMatch{similarity: score, id: row[1], smiles: row[2]} + if previous != nil && better(match, *previous, direction) { + return fmt.Errorf("partial result is not sorted deterministically") + } + previous = &match + consume(match) + } +} + +func positiveInt(value any, fallback int) (int, error) { + if value == nil { + return fallback, nil + } + switch n := value.(type) { + case int: + if n > 0 { + return n, nil + } + case int64: + if n > 0 && n <= math.MaxInt { + return int(n), nil + } + case float64: + if n > 0 && n == math.Trunc(n) && n <= math.MaxInt { + return int(n), nil + } + } + return 0, fmt.Errorf("top_k must be a positive integer") +} + +func thresholdDirection(value any) (string, error) { + if value == nil { + return "greater", nil + } + direction, ok := value.(string) + if !ok || (direction != "greater" && direction != "less") { + return "", fmt.Errorf("threshold_direction must be greater or less") + } + return direction, nil +} + +func better(left, right similarityMatch, direction string) bool { + if left.similarity != right.similarity { + if direction == "less" { + return left.similarity < right.similarity + } + return left.similarity > right.similarity + } + if left.id != right.id { + return left.id < right.id + } + return left.smiles < right.smiles +} + +func equalStrings(left, right []string) bool { + if len(left) != len(right) { + return false + } + for index := range left { + if left[index] != right[index] { + return false + } + } + return true +} + +// matchHeap keeps the worst retained match at index zero. +type matchHeap struct { + items []similarityMatch + direction string +} + +func heapPush(h *matchHeap, value similarityMatch) { + h.items = append(h.items, value) + for child := len(h.items) - 1; child > 0; { + parent := (child - 1) / 2 + if !better(h.items[parent], h.items[child], h.direction) { + break + } + h.items[parent], h.items[child] = h.items[child], h.items[parent] + child = parent + } +} + +func heapDown(h *matchHeap, parent int) { + for { + child := parent*2 + 1 + if child >= len(h.items) { + return + } + if right := child + 1; right < len(h.items) && better(h.items[child], h.items[right], h.direction) { + child = right + } + if !better(h.items[parent], h.items[child], h.direction) { + return + } + h.items[parent], h.items[child] = h.items[child], h.items[parent] + parent = child + } +} diff --git a/coordinator/internal/reducer/similarity_search_test.go b/coordinator/internal/reducer/similarity_search_test.go new file mode 100644 index 0000000..ff4edc8 --- /dev/null +++ b/coordinator/internal/reducer/similarity_search_test.go @@ -0,0 +1,41 @@ +package reducer + +import ( + "io" + "strings" + "testing" +) + +func TestReduceSimilaritySearchKeepsExactCrossShardRanking(t *testing.T) { + first := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,B,CCC,0.50000048\n2,C,CCCC,0.1\n") + second := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.50000049\n") + + output, err := ReduceSimilaritySearch([]io.Reader{first, second}, map[string]any{"top_k": 2}) + if err != nil { + t.Fatal(err) + } + want := "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.500000\n2,B,CCC,0.500000\n" + if string(output) != want { + t.Fatalf("output = %q, want %q", output, want) + } +} + +func TestReduceSimilaritySearchSupportsLeastSimilarDirection(t *testing.T) { + partial := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.1\n2,B,CCC,0.8\n") + output, err := ReduceSimilaritySearch([]io.Reader{partial}, map[string]any{ + "top_k": 1, "threshold_direction": "less", + }) + if err != nil { + t.Fatal(err) + } + if got, want := string(output), "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.100000\n"; got != want { + t.Fatalf("output = %q, want %q", got, want) + } +} + +func TestReduceSimilaritySearchRejectsMalformedPartial(t *testing.T) { + partial := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n2,A,CC,0.1\n") + if _, err := ReduceSimilaritySearch([]io.Reader{partial}, nil); err == nil { + t.Fatal("expected malformed rank error") + } +} diff --git a/coordinator/internal/storage/postgres/integration_test.go b/coordinator/internal/storage/postgres/integration_test.go index b4b18ac..7628a50 100644 --- a/coordinator/internal/storage/postgres/integration_test.go +++ b/coordinator/internal/storage/postgres/integration_test.go @@ -90,6 +90,49 @@ func TestCreateJobPersistsEveryTask(t *testing.T) { } } +func TestClaimReductionIsAtomic(t *testing.T) { + pool := testPool(t) + job, _ := seedJob(t, pool, 1) + repo := NewJobRepo(pool) + ctx := context.Background() + if err := repo.UpdateStatus(ctx, job.ID, domain.JobReducing, nil); err != nil { + t.Fatal(err) + } + + var ( + wg sync.WaitGroup + mu sync.Mutex + claimed int + ) + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + ok, err := repo.ClaimReduction(context.Background(), job.ID, time.Now().UTC()) + if err != nil { + t.Errorf("claim reduction: %v", err) + return + } + if ok { + mu.Lock() + claimed++ + mu.Unlock() + } + }() + } + wg.Wait() + if claimed != 1 { + t.Fatalf("reducer claims = %d, want 1", claimed) + } + stored, err := repo.Get(ctx, job.ID) + if err != nil { + t.Fatal(err) + } + if stored.Status != domain.JobReducing || stored.ReducerStartedAt == nil { + t.Fatalf("stored reduction state = %+v", stored) + } +} + // A job must land whole or not at all: a half-created job leaves chunks no // worker could ever complete. func TestCreateJobRollsBackOnFailure(t *testing.T) { diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go index f2a7e2e..d889a23 100644 --- a/coordinator/internal/storage/postgres/job_repo.go +++ b/coordinator/internal/storage/postgres/job_repo.go @@ -25,7 +25,10 @@ func NewJobRepo(pool *pgxpool.Pool) *JobRepo { var _ usecase.JobRepository = (*JobRepo)(nil) -var jobColumns = []string{"id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at"} +var jobColumns = []string{ + "id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at", + "input_artifact_id", "result_artifact_id", "error_code", "error_message", "reducer_started_at", +} // Insert runs inside the caller's transaction, alongside the job's tasks — that // is what makes "all tasks or none" hold. @@ -55,7 +58,8 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) { status string ) err = conn(ctx, r.pool).QueryRow(ctx, sql, args...).Scan( - &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt) + &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt, + &j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, domain.ErrJobNotFound } @@ -66,6 +70,64 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) { return &j, nil } +func (r *JobRepo) ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error) { + sql, args, err := psql.Update("jobs"). + Set("reducer_started_at", startedAt). + Where(sq.Eq{"id": id, "status": string(domain.JobReducing), "reducer_started_at": nil}). + ToSql() + if err != nil { + return false, err + } + tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...) + if err != nil { + return false, err + } + return tag.RowsAffected() == 1, nil +} + +func (r *JobRepo) CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error { + sql, args, err := psql.Update("jobs"). + SetMap(map[string]any{ + "status": string(domain.JobCompleted), + "result_artifact_id": resultArtifactID, + "completed_at": completedAt, + "reducer_started_at": nil, + "error_code": nil, + "error_message": nil, + }). + Where(sq.Eq{"id": id, "status": string(domain.JobReducing)}). + ToSql() + if err != nil { + return err + } + tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return domain.ErrJobNotFound + } + return nil +} + +func (r *JobRepo) FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error { + sql, args, err := psql.Update("jobs"). + SetMap(map[string]any{ + "status": string(domain.JobFailed), + "completed_at": completedAt, + "error_code": code, + "error_message": message, + "reducer_started_at": nil, + }). + Where(sq.Eq{"id": id, "status": string(domain.JobReducing)}). + ToSql() + if err != nil { + return err + } + _, err = conn(ctx, r.pool).Exec(ctx, sql, args...) + return err +} + func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error { diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go index f944482..42e706d 100644 --- a/coordinator/internal/transport/http/dto.go +++ b/coordinator/internal/transport/http/dto.go @@ -119,6 +119,8 @@ type jobProgressResponse struct { Done int `json:"completed"` Failed int `json:"failed"` Cancelled int `json:"cancelled"` + ResultURI string `json:"result_uri,omitempty"` + ErrorCode string `json:"error_code,omitempty"` } type uploadArtifactResponse struct { @@ -153,7 +155,7 @@ func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse { } func toJobProgressResponse(p domain.JobProgress) jobProgressResponse { - return jobProgressResponse{ + out := jobProgressResponse{ ID: p.Job.ID, Status: string(p.DeriveStatus()), Total: p.Total, @@ -163,4 +165,11 @@ func toJobProgressResponse(p domain.JobProgress) jobProgressResponse { Failed: p.Failed, Cancelled: p.Cancelled, } + if p.Job.ResultArtifactID != nil && out.Status == string(domain.JobCompleted) { + out.ResultURI = "/jobs/" + p.Job.ID.String() + "/result" + } + if p.Job.ErrorCode != nil { + out.ErrorCode = *p.Job.ErrorCode + } + return out } diff --git a/coordinator/internal/transport/http/dto_test.go b/coordinator/internal/transport/http/dto_test.go new file mode 100644 index 0000000..3fb7608 --- /dev/null +++ b/coordinator/internal/transport/http/dto_test.go @@ -0,0 +1,25 @@ +package http + +import ( + "testing" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +func TestJobProgressResponseExposesFinalResultOnlyWhenCompleted(t *testing.T) { + id := uuid.New() + result := uuid.New() + progress := domain.JobProgress{Job: domain.Job{ + ID: id, Status: domain.JobCompleted, ResultArtifactID: &result, + }, Total: 1, Done: 1} + if got, want := toJobProgressResponse(progress).ResultURI, "/jobs/"+id.String()+"/result"; got != want { + t.Fatalf("result URI = %q, want %q", got, want) + } + + progress.Job.Status = domain.JobReducing + if got := toJobProgressResponse(progress).ResultURI; got != "" { + t.Fatalf("reducing job exposes result URI %q", got) + } +} diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go index 33ad06b..e8c827b 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/coordinator/internal/transport/http/handlers.go @@ -150,6 +150,12 @@ func (s *Server) handleResult(w http.ResponseWriter, r *http.Request) { s.writeError(w, r, err) return } + if s.uc.ReduceJob != nil { + if err := s.uc.ReduceJob.Execute(ctx, task.JobID); err != nil { + s.writeError(w, r, err) + return + } + } writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)}) } @@ -399,6 +405,25 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, toJobProgressResponse(progress)) } +func (s *Server) handleGetJobResult(w http.ResponseWriter, r *http.Request) { + ctx, cancel := s.reqCtx(r) + defer cancel() + jobID, ok := s.pathUUID(w, r, "job_id") + if !ok { + return + } + art, body, err := s.uc.GetJobResult.Execute(ctx, jobID) + if err != nil { + s.writeError(w, r, err) + return + } + defer func() { _ = 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)) + _, _ = io.Copy(w, body) +} + // handleCancelJob stops all non-terminal shards for an operator-requested job. // It is available to both the bearer API and the separately authenticated UI. func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) { diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 00e80d8..560c85b 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -22,8 +22,10 @@ type UseCases struct { ClaimTask *usecase.ClaimTask RenewLease *usecase.RenewLease CompleteTask *usecase.CompleteTask + ReduceJob *usecase.ReduceJob FailTask *usecase.FailTask GetJobStatus *usecase.GetJobStatus + GetJobResult *usecase.GetJobResult CancelJob *usecase.CancelJob UploadArtifact *usecase.UploadArtifact DownloadArtifact *usecase.DownloadArtifact @@ -62,6 +64,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler { protected.HandleFunc("POST /jobs", s.handleCreateJob) protected.HandleFunc("POST /jobs/upload", s.handleUploadDataset) protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob) + protected.HandleFunc("GET /jobs/{job_id}/result", s.handleGetJobResult) protected.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob) protected.HandleFunc("POST /tasks/claim", s.handleClaim) protected.HandleFunc("GET /tasks/{task_id}/input", s.handleGetTaskInput) diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 52c453f..97c7681 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -42,6 +42,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur clk := memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)) tx := memstore.Tx{} lease := 2 * time.Minute + downloadArtifact := usecase.NewDownloadArtifact(arts, blobs) uc := coordhttp.UseCases{ RegisterWorker: usecase.NewRegisterWorker(work, clk), @@ -50,11 +51,13 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur ClaimTask: usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease), RenewLease: usecase.NewRenewLease(tasks, work, tx, clk, lease), CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, tx, clk), + ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk), FailTask: usecase.NewFailTask(tasks, jobs, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobs, tasks), + GetJobResult: usecase.NewGetJobResult(jobs, downloadArtifact), CancelJob: usecase.NewCancelJob(jobs, tasks, tx, clk), UploadArtifact: usecase.NewUploadArtifact(tasks, arts, blobs, tx, clk), - DownloadArtifact: usecase.NewDownloadArtifact(arts, blobs), + DownloadArtifact: downloadArtifact, GetTaskInput: usecase.NewGetTaskInput(tasks, arts, blobs), Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)), } @@ -400,6 +403,42 @@ func TestFullLifecycle(t *testing.T) { } } +func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) { + e := newEnv(t, healthy) + code, job := e.uploadDataset(t, "similarity-search", 10, "chembl_id\tcanonical_smiles\nA\tCC\n") + if code != http.StatusCreated { + t.Fatalf("upload job: %d", code) + } + jobID := job["job_id"].(string) + code, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["similarity-search"]}`) + if code != http.StatusOK { + t.Fatalf("claim: %d", code) + } + taskID := claim["task_id"].(string) + attempt := int(claim["attempt"].(float64)) + artifactID := e.putArtifact(t, taskID, "w1", attempt, "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n") + if code, _ := e.do(t, "POST", "/tasks/"+taskID+"/result", + `{"worker_id":"w1","attempt":`+itoa(attempt)+`,"result":{"artifact_id":"`+artifactID+`"}}`); code != http.StatusOK { + t.Fatalf("complete: %d", code) + } + + code, progress := e.do(t, "GET", "/jobs/"+jobID, "") + if code != http.StatusOK || progress["status"] != "completed" || progress["result_uri"] != "/jobs/"+jobID+"/result" { + t.Fatalf("progress = (%d, %v)", code, progress) + } + req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+progress["result_uri"].(string), nil) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n" { + t.Fatalf("final result = (%d, %q)", resp.StatusCode, body) + } +} + func TestForeignArtifactResultConflict(t *testing.T) { e := newEnv(t, healthy) e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in", diff --git a/coordinator/internal/transport/http/templates/dashboard.html b/coordinator/internal/transport/http/templates/dashboard.html index 5e24f90..e643cc7 100644 --- a/coordinator/internal/transport/http/templates/dashboard.html +++ b/coordinator/internal/transport/http/templates/dashboard.html @@ -12,7 +12,7 @@

Local coordinator

SciMesh operator dashboard

See where a computation is and what should happen next.

Start a check
-
This screen currently diagnoses shard jobs.Workers upload partial CSVs to the coordinator. Until a reducer is implemented, those files are not one final scientific result.
1. Upload TSVThe coordinator splits the file into shard tasks.
2. Wait for a workerA worker claims a shard, calculates similarity, and returns a CSV.
3. Inspect artifactsDownload a partial result from the job page.
+
Similarity-search jobs produce a final CSV.Workers return shard-level candidates; after every shard succeeds, the coordinator deterministically merges them into one global top-k result.
1. Upload TSVThe coordinator splits the file into shard tasks.
2. Wait for a workerA worker claims a shard, calculates similarity, and returns a CSV.
3. Download resultWhen merging finishes, download the final CSV from the job page.

Recent jobs

{{range .Jobs}}{{else}}{{end}}
ComputationStateProgressCreated
{{workloadLabel .Workload}}
Open job details		{{statusLabel .Status}}
{{statusHint .Status}}		{{.Completed}} / {{.Total}} complete{{if gt .Failed 0}} · failed: {{.Failed}}{{end}}{{if gt .Cancelled 0}} · stopped: {{.Cancelled}}{{end}}
{{time .CreatedAt}}
No jobs yet.
Click “Start a check”, upload a small TSV, and leave a worker running.

Workers

diff --git a/coordinator/internal/transport/http/templates/job.html b/coordinator/internal/transport/http/templates/job.html index b985a2e..596a12d 100644 --- a/coordinator/internal/transport/http/templates/job.html +++ b/coordinator/internal/transport/http/templates/job.html @@ -13,14 +13,14 @@
← Back to jobs

{{workloadLabel .Workload}}

Execution progress

{{statusLabel .Status}}

{{statusHint .Status}}

{{if cancellable .Status}}This cancels every shard that is not finished yet.{{else}}Summary refreshes automatically every two seconds.{{end}}

{{.Completed}} of {{.Total}} tasks complete

{{.Total}}total shards
{{.Completed}}complete
{{.Pending}}waiting
{{add .Leased .Running}}with workers
{{.Failed}}failed
{{.Cancelled}}stopped
Technical details

Job ID: {{.ID}}
Workload: {{.Workload}}
Created: {{time .CreatedAt}}

-
What can be downloaded now?
partial_result files come from individual shards. They are useful for checking the pipeline, but are not a merged final CSV because the reducer is not implemented yet.
+
{{if .FinalResultAvailable}}Final result ready
Download the final_result CSV below. It is the deterministic global top-k across all completed shards. The partial_result files remain available for diagnostics.{{else if eq .Status "reducing"}}Merging completed shards
Every shard has finished. The coordinator is building one deterministic global CSV; refresh in a moment to download it.{{else}}What can be downloaded now?
partial_result files come from individual shards. They are useful for checking the pipeline, but are not a merged final CSV yet.{{end}}

Shard tasks

If a task fails, its code and message appear here. Refresh the page to update the detailed rows.

{{range .Tasks}}{{else}}{{end}}
ShardStateAttemptWorker / leaseError
#{{.ChunkIndex}}{{statusLabel .Status}}{{.Attempt}} / {{.MaxAttempts}}{{if .LeaseOwner}}{{.LeaseOwner}}{{if .LeaseExpiresAt}}
until {{time .LeaseExpiresAt}}{{end}}{{else}}—{{end}}		{{if .ErrorCode}}{{taskErrorLabel .ErrorCode}}
{{taskErrorHint .ErrorCode}}{{else}}—{{end}}
No tasks have appeared yet.

Coordinator artifacts

{{range .Artifacts}}{{else}}{{end}}
TypeFileSizeIntegrity check
{{if .Diagnostic}}Partial result
diagnostic{{else}}{{.Kind}}{{end}}		{{.Filename}}{{bytes .SizeBytes}}{{.SHA256}}{{if .Downloadable}}Download CSV{{else}}Unavailable{{end}}
No artifacts yet. The worker uploads a CSV after it completes a shard.
diff --git a/coordinator/internal/transport/http/ui.go b/coordinator/internal/transport/http/ui.go index ddf1168..abcba26 100644 --- a/coordinator/internal/transport/http/ui.go +++ b/coordinator/internal/transport/http/ui.go @@ -48,8 +48,10 @@ func uiStatusLabel(status string) string { return "Assigned to a worker" case "running": return "Running" + case "reducing": + return "Merging results" case "completed": - return "Tasks complete" + return "Completed" case "failed": return "Needs attention" case "cancelled": @@ -67,8 +69,10 @@ func uiStatusHint(status string) string { return "A worker has claimed the task and should begin processing shortly." case "running": return "A worker is reading a shard, calculating fingerprints, and uploading its result through the coordinator." + case "reducing": + return "All shards are complete. The coordinator is merging their candidates into one final CSV." case "completed": - return "Every shard task is complete. Files below are still partial results." + return "The final result is ready to download." case "failed": return "One or more shard tasks failed. Open the task list below for details." case "cancelled": @@ -86,7 +90,7 @@ func uiStatusClass(status string) string { return "danger" case "cancelled": return "waiting" - case "running", "leased": + case "running", "leased", "reducing": return "active" default: return "waiting" diff --git a/coordinator/internal/transport/http/ui_test.go b/coordinator/internal/transport/http/ui_test.go index a6f82cb..034af7c 100644 --- a/coordinator/internal/transport/http/ui_test.go +++ b/coordinator/internal/transport/http/ui_test.go @@ -10,7 +10,8 @@ func TestUIStatusPresentation(t *testing.T) { }{ {"pending", "Waiting for a worker", "waiting"}, {"running", "Running", "active"}, - {"completed", "Tasks complete", "success"}, + {"reducing", "Merging results", "active"}, + {"completed", "Completed", "success"}, {"failed", "Needs attention", "danger"}, } for _, test := range tests { diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go index f6e0497..eda4b49 100644 --- a/coordinator/internal/usecase/job.go +++ b/coordinator/internal/usecase/job.go @@ -100,7 +100,7 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error if job.Status == domain.JobCancelled { return nil } - if job.Status == domain.JobCompleted || job.Status == domain.JobFailed { + if job.Status == domain.JobReducing || job.Status == domain.JobCompleted || job.Status == domain.JobFailed { return domain.ErrJobNotCancellable } // The lease reaper can be the transition that exhausted the final task. @@ -111,7 +111,7 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error return err } derived := progressFrom(*job, counts).DeriveStatus() - if derived == domain.JobCompleted || derived == domain.JobFailed { + if derived == domain.JobReducing || derived == domain.JobCompleted || derived == domain.JobFailed { return domain.ErrJobNotCancellable } cancelled, err = uc.tasks.CancelByJob(ctx, jobID, now) @@ -226,10 +226,20 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository if err != nil { return err } - status := progressFrom(domain.Job{}, counts).DeriveStatus() + job, err := jobs.Get(ctx, jobID) + if err != nil { + return err + } + status := progressFrom(*job, counts).DeriveStatus() + // All worker shards being complete means scientific reduction is ready, not + // that the job's final artifact already exists. CTX-09 owns the transition + // from reducing to completed after it persists that artifact. + if status == domain.JobCompleted && job.Workload == "similarity-search" { + status = domain.JobReducing + } var completedAt *time.Time - if status == domain.JobCompleted || status == domain.JobFailed { + if status == domain.JobFailed { completedAt = &now } return jobs.UpdateStatus(ctx, jobID, status, completedAt) diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index 119a0c9..933453c 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -70,6 +70,9 @@ type JobRepository interface { Insert(ctx context.Context, j *domain.Job) error Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error + ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error) + CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error + FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error } // WorkerRepository persists the worker registry. diff --git a/coordinator/internal/usecase/reduce.go b/coordinator/internal/usecase/reduce.go new file mode 100644 index 0000000..e73f575 --- /dev/null +++ b/coordinator/internal/usecase/reduce.go @@ -0,0 +1,133 @@ +package usecase + +import ( + "bytes" + "context" + "io" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/reducer" +) + +// ReduceJob turns completed coordinator-owned partial artifacts into one final +// artifact. It performs no worker I/O and never trusts a worker URI or path. +type ReduceJob struct { + jobs JobRepository + tasks TaskRepository + artifacts ArtifactRepository + blobs BlobStore + tx TxManager + clock Clock +} + +func NewReduceJob(jobs JobRepository, tasks TaskRepository, artifacts ArtifactRepository, + blobs BlobStore, tx TxManager, clock Clock) *ReduceJob { + return &ReduceJob{jobs: jobs, tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clock: clock} +} + +// Execute is idempotent for jobs that are not currently reducing. The worker +// completion path may call it after every result; only the last task changes a +// similarity-search job into reducing state. +func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error { + claimed, err := uc.jobs.ClaimReduction(ctx, jobID, uc.clock.Now()) + if err != nil || !claimed { + return err + } + job, err := uc.jobs.Get(ctx, jobID) + if err != nil { + return err + } + if job.Status != domain.JobReducing { + return nil + } + if job.Workload != "similarity-search" { + return uc.fail(ctx, jobID) + } + + completed, err := uc.tasks.ListCompleted(ctx, jobID) + if err != nil { + return uc.fail(ctx, jobID) + } + if len(completed) == 0 { + return uc.fail(ctx, jobID) + } + readers := make([]io.Reader, 0, len(completed)) + closers := make([]io.Closer, 0, len(completed)) + for _, task := range completed { + if task.ResultArtifactID == nil { + closeAll(closers) + return uc.fail(ctx, jobID) + } + artifact, err := uc.artifacts.Get(ctx, *task.ResultArtifactID) + if err != nil || artifact.JobID != jobID || artifact.TaskID == nil || *artifact.TaskID != task.ID || artifact.Kind != domain.ArtifactPartialResult { + closeAll(closers) + return uc.fail(ctx, jobID) + } + body, err := uc.blobs.Open(ctx, artifact.StorageKey) + if err != nil { + closeAll(closers) + return uc.fail(ctx, jobID) + } + readers = append(readers, body) + closers = append(closers, body) + } + output, reduceErr := reducer.ReduceSimilaritySearch(readers, job.Parameters) + closeAll(closers) + if reduceErr != nil { + return uc.fail(ctx, jobID) + } + + final, err := domain.NewArtifact(jobID, nil, domain.ArtifactFinalResult, "similarity-search.csv", "text/csv", uc.clock.Now()) + if err != nil { + return uc.fail(ctx, jobID) + } + sum, size, err := uc.blobs.Put(ctx, final.StorageKey, bytes.NewReader(output)) + if err != nil { + return uc.fail(ctx, jobID) + } + final.SetContent(sum, size) + if err := uc.tx.WithinTx(ctx, func(ctx context.Context) error { + if err := uc.artifacts.Insert(ctx, final); err != nil { + return err + } + return uc.jobs.CompleteWithResult(ctx, jobID, final.ID, uc.clock.Now()) + }); err != nil { + _ = uc.blobs.Delete(ctx, final.StorageKey) + return err + } + return nil +} + +func (uc *ReduceJob) fail(ctx context.Context, jobID uuid.UUID) error { + // The public state carries a stable sanitized failure, never parser/storage + // internals that may include local paths or implementation details. + return uc.jobs.FailReduction(ctx, jobID, "reducer_failed", "final result reduction failed", uc.clock.Now()) +} + +func closeAll(closers []io.Closer) { + for _, closer := range closers { + _ = closer.Close() + } +} + +type GetJobResult struct { + jobs JobRepository + download *DownloadArtifact +} + +func NewGetJobResult(jobs JobRepository, download *DownloadArtifact) *GetJobResult { + return &GetJobResult{jobs: jobs, download: download} +} + +func (uc *GetJobResult) Execute(ctx context.Context, jobID uuid.UUID) (*domain.Artifact, io.ReadCloser, error) { + job, err := uc.jobs.Get(ctx, jobID) + if err != nil { + return nil, nil, err + } + if job.Status != domain.JobCompleted || job.ResultArtifactID == nil { + return nil, nil, domain.ErrArtifactNotFound + } + return uc.download.Execute(ctx, *job.ResultArtifactID) +} diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index 2083c5c..2272c51 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -55,6 +55,8 @@ type harness struct { getInput *usecase.GetTaskInput expire *usecase.ExpireLeases cancel *usecase.CancelJob + reduce *usecase.ReduceJob + jobResult *usecase.GetJobResult } func newHarness() *harness { @@ -81,9 +83,88 @@ func newHarness() *harness { h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs) h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk) h.cancel = usecase.NewCancelJob(h.jobs, h.tasks, tx, h.clk) + h.reduce = usecase.NewReduceJob(h.jobs, h.tasks, h.arts, h.blobs, tx, h.clk) + h.jobResult = usecase.NewGetJobResult(h.jobs, h.downloadArt) return h } +func TestSimilaritySearchReductionCreatesFinalArtifact(t *testing.T) { + h := newHarness() + jobID := h.seedJob(t, "similarity-search", 2) + _, err := h.jobs.Get(ctx, jobID) + if err != nil { + t.Fatal(err) + } + if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + partials := []string{ + "rank,chembl_id,canonical_smiles,similarity\n1,B,CCC,0.50000048\n", + "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.50000049\n", + } + for _, partial := range partials { + taskID, attempt := h.leaseOne(t, "w1", "similarity-search") + art, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, Filename: "partial.csv", ContentType: "text/csv", Body: strings.NewReader(partial)}) + if err != nil { + t.Fatal(err) + } + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, ResultArtifactID: art.ID}); err != nil { + t.Fatal(err) + } + } + if err := h.reduce.Execute(ctx, jobID); err != nil { + t.Fatal(err) + } + progress, err := h.status.Execute(ctx, jobID) + if err != nil || progress.Job.Status != domain.JobCompleted { + t.Fatalf("status=%s err=%v", progress.Job.Status, err) + } + art, body, err := h.jobResult.Execute(ctx, jobID) + if err != nil { + t.Fatal(err) + } + defer body.Close() + bytes, _ := io.ReadAll(body) + if art.Kind != domain.ArtifactFinalResult || string(bytes) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.500000\n2,B,CCC,0.500000\n" { + t.Fatalf("unexpected final %q", bytes) + } +} + +func TestSimilaritySearchReductionFailureIsSanitized(t *testing.T) { + h := newHarness() + jobID := h.seedJob(t, "similarity-search", 1) + if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + taskID, attempt := h.leaseOne(t, "w1", "similarity-search") + art, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{ + TaskID: taskID, WorkerID: "w1", Attempt: attempt, Filename: "partial.csv", + ContentType: "text/csv", Body: strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n2,A,CC,0.9\n"), + }) + if err != nil { + t.Fatal(err) + } + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{ + TaskID: taskID, WorkerID: "w1", Attempt: attempt, ResultArtifactID: art.ID, + }); err != nil { + t.Fatal(err) + } + if err := h.reduce.Execute(ctx, jobID); err != nil { + t.Fatal(err) + } + job, err := h.jobs.Get(ctx, jobID) + if err != nil { + t.Fatal(err) + } + if job.Status != domain.JobFailed || job.ErrorCode == nil || *job.ErrorCode != "reducer_failed" || + job.ErrorMessage == nil || *job.ErrorMessage != "final result reduction failed" { + t.Fatalf("unexpected failed job: %+v", job) + } + if job.ResultArtifactID != nil { + t.Fatal("failed reduction must not expose a final result") + } +} + // seedJob creates a URI-chunked job with n chunks and returns its id. func (h *harness) seedJob(t *testing.T, workload string, n int) uuid.UUID { t.Helper() diff --git a/coordinator/migrations/0010_job_reduction.down.sql b/coordinator/migrations/0010_job_reduction.down.sql new file mode 100644 index 0000000..da2d01b --- /dev/null +++ b/coordinator/migrations/0010_job_reduction.down.sql @@ -0,0 +1,7 @@ +BEGIN; + +ALTER TABLE jobs DROP COLUMN IF EXISTS error_message; +ALTER TABLE jobs DROP COLUMN IF EXISTS error_code; +ALTER TABLE jobs DROP COLUMN IF EXISTS reducer_started_at; + +COMMIT; diff --git a/coordinator/migrations/0010_job_reduction.up.sql b/coordinator/migrations/0010_job_reduction.up.sql new file mode 100644 index 0000000..b2a28b8 --- /dev/null +++ b/coordinator/migrations/0010_job_reduction.up.sql @@ -0,0 +1,7 @@ +-- PostgreSQL enum values must be committed before they are used by a later +-- transaction, so this migration intentionally has no BEGIN/COMMIT wrapper. +ALTER TYPE job_status ADD VALUE IF NOT EXISTS 'reducing'; + +ALTER TABLE jobs ADD COLUMN IF NOT EXISTS error_code text; +ALTER TABLE jobs ADD COLUMN IF NOT EXISTS error_message text; +ALTER TABLE jobs ADD COLUMN IF NOT EXISTS reducer_started_at timestamptz; diff --git a/docs/api-contract.md b/docs/api-contract.md index 103b2d1..62589e1 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -26,7 +26,8 @@ must be updated in the same change as any behaviour it describes. | `POST /tasks/{id}/heartbeat` | renew lease | ✅ done | | `POST /tasks/{id}/result` | complete | ✅ done, references `artifact_id` | | `POST /tasks/{id}/failure` | fail | ✅ done | -| `GET /jobs/{id}` | progress | ✅ done | +| `GET /jobs/{id}` | progress and final-result URI | ✅ done | +| `GET /jobs/{id}/result` | download final CSV | ✅ done | | `PUT /tasks/{id}/artifacts/{name}` | upload partial | ✅ done | | `GET /artifacts/{id}/download` | download by id | ✅ done | | `POST /jobs/upload` | upload dataset, coordinator chunks it | ✅ done | @@ -67,6 +68,35 @@ artifact. The coordinator splits the selected TSV rows into shard artifacts Each resulting task's claim response carries `input.uri = /tasks/{id}/input`, served by §5.4. +## Job progress and final result + +```http +GET /jobs/{job_id} +Authorization: Bearer +``` + +The response contains task counters and a derived status. A successful +`similarity-search` enters `reducing` after the last shard completes, then +becomes `completed` only after the coordinator stores its deterministic final +CSV. At that point `result_uri` is present: + +```json +{ + "id": "uuid", + "status": "completed", + "total": 3, + "pending": 0, + "leased": 0, + "completed": 3, + "failed": 0, + "cancelled": 0, + "result_uri": "/jobs/uuid/result" +} +``` + +`GET /jobs/{job_id}/result` downloads that final coordinator-owned CSV. Before +the job is completed (or for a failed/cancelled job), it returns `404`. + ## Stop a job ```http diff --git a/docs/building-workers.md b/docs/building-workers.md index 08f8d6a..956d302 100644 --- a/docs/building-workers.md +++ b/docs/building-workers.md @@ -56,9 +56,10 @@ Response: `{ "worker_id": "", "heartbeat_interval_seconds": 15 }`. - **Keep `worker_id`**. Use it as your identity in every later call. Using the registered UUID is what lets the coordinator track your liveness (it marks workers offline after they go silent). -- Current diagnostic uploads use `similarity-search` with `query_smiles`. The - reference worker accepts the legacy `similarity_search` spelling too. Do not - advertise `similarity-graph` until CTX-10 implements cross-shard pair planning. +- Current distributed uploads use `similarity-search` with `query_smiles`; the + coordinator merges completed shard candidates into a final CSV. The reference + worker accepts the legacy `similarity_search` spelling too. Do not advertise + `similarity-graph` until CTX-10 implements cross-shard pair planning. ## 2. Claim a task diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 4e574df..901d76d 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -97,8 +97,9 @@ paths: multipart/form-data. The text fields (`workload`, `parameters`, `chunk_rows`, `max_rows`) MUST precede the `file` part: the file is streamed, not buffered, so the fields have to be parsed before it arrives. Currently - only diagnostic `similarity-search` with `parameters.query_smiles` is - accepted; distributed graph planning is not implemented. + only `similarity-search` with `parameters.query_smiles` is accepted. + When every shard succeeds, the coordinator merges their candidates into + one final CSV; distributed graph planning is not implemented. requestBody: required: true content: @@ -124,7 +125,9 @@ paths: - $ref: "#/components/parameters/JobID" responses: "200": - description: Progress counts and derived status. + description: > + Progress counts and derived status. A completed similarity-search + response includes `result_uri` for its final CSV. content: application/json: schema: { $ref: "#/components/schemas/JobProgress" } @@ -132,6 +135,21 @@ paths: "401": { $ref: "#/components/responses/Unauthorized" } "404": { $ref: "#/components/responses/NotFound" } + /jobs/{job_id}/result: + get: + tags: [jobs] + summary: Download a completed job's final result + parameters: + - $ref: "#/components/parameters/JobID" + responses: + "200": + description: Final coordinator-owned CSV. + content: + text/csv: + schema: { type: string, format: binary } + "401": { $ref: "#/components/responses/Unauthorized" } + "404": { $ref: "#/components/responses/NotFound" } + /jobs/{job_id}/cancel: post: tags: [jobs] @@ -483,6 +501,13 @@ components: completed: { type: integer } failed: { type: integer } cancelled: { type: integer } + result_uri: + type: string + description: Present only when the final result is available. + example: /jobs/5a4c3a7f-ccfc-47d6-b78d-2d1fa565bafd/result + error_code: + type: string + description: Sanitized terminal reducer failure code, when applicable. ClaimRequest: type: object @@ -577,7 +602,7 @@ components: JobStatus: type: string - enum: [pending, running, completed, failed, cancelled] + enum: [pending, running, reducing, completed, failed, cancelled] TaskStatus: type: string