From 58da6ef139272fc39cf3824cef6dd81b7da66ff4 Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Thu, 23 Jul 2026 15:54:11 +0300 Subject: [PATCH] feat(coordinator): complete tasks with an artifact_id, not a URI (CTX-05, part 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task results are now coordinator-owned artifacts end to end. - domain.Task carries ResultArtifactID instead of ResultURI/ResultSHA256; CompleteWith and its idempotency key are keyed on the artifact id. - CompleteTask verifies the referenced artifact was stored for this exact task (rule 10): a worker cannot finish task B with task A's artifact, nor name an id that isn't a partial_result. Mismatch → 409. - POST /tasks/{id}/result takes {result:{artifact_id,...}}; ListResults and ResultManifest follow. - migration 0004 drops result_uri/result_sha256 and requires a completed task to reference its result_artifact_id. - smoke and requests.http exercise upload → complete-by-id → replay → conflict. --- coordinator/api/requests.http | 30 +++++---- coordinator/cmd/coordinator/main.go | 2 +- coordinator/internal/domain/task.go | 65 +++++++++---------- coordinator/internal/domain/task_test.go | 22 ++++--- .../internal/storage/postgres/task_repo.go | 27 ++++---- coordinator/internal/transport/http/dto.go | 18 +++-- .../internal/transport/http/handlers.go | 11 ++-- coordinator/internal/usecase/dto.go | 11 ++-- coordinator/internal/usecase/job.go | 13 ++-- coordinator/internal/usecase/task.go | 37 +++++++++-- .../migrations/0004_result_artifact.down.sql | 11 ++++ .../migrations/0004_result_artifact.up.sql | 14 ++++ coordinator/scripts/smoke.sh | 29 ++++++--- docs/api-contract.md | 10 +-- 14 files changed, 184 insertions(+), 116 deletions(-) create mode 100644 coordinator/migrations/0004_result_artifact.down.sql create mode 100644 coordinator/migrations/0004_result_artifact.up.sql diff --git a/coordinator/api/requests.http b/coordinator/api/requests.http index 181a3b5..303c3a3 100644 --- a/coordinator/api/requests.http +++ b/coordinator/api/requests.http @@ -98,7 +98,20 @@ CHEMBL25,CHEMBL139,0.87 GET {{host}}/artifacts/{{artifactId}}/download Authorization: Bearer {{token}} -### 4. Submit the result (200) +### 3c. Upload a second artifact — used by the conflict check below (200) +# @name uploadArtifact2 +PUT {{host}}/tasks/{{taskId}}/artifacts/secondary.csv +Authorization: Bearer {{token}} +Content-Type: text/csv +X-Worker-ID: {{worker}} +X-Task-Attempt: {{attempt}} + +query,match,score +CHEMBL25,CHEMBL521,0.42 + +@artifactId2 = {{uploadArtifact2.response.body.artifact_id}} + +### 4. Submit the result, referencing the uploaded artifact (200) POST {{host}}/tasks/{{taskId}}/result Authorization: Bearer {{token}} Content-Type: application/json @@ -106,8 +119,7 @@ Content-Type: application/json { "worker_id": "{{worker}}", "attempt": {{attempt}}, - "result_uri": "s3://results/shard-0.csv", - "result_sha256": "r0sha", + "result": { "artifact_id": "{{artifactId}}", "content_type": "text/csv" }, "metrics": { "elapsed_ms": 1234, "candidates": 50000 } } @@ -119,12 +131,10 @@ Content-Type: application/json { "worker_id": "{{worker}}", "attempt": {{attempt}}, - "result_uri": "s3://results/shard-0.csv", - "result_sha256": "r0sha", - "metrics": { "elapsed_ms": 1234, "candidates": 50000 } + "result": { "artifact_id": "{{artifactId}}" } } -### 4b. A different result for the same task — conflict (409) +### 4b. A different artifact for the same task — conflict (409) POST {{host}}/tasks/{{taskId}}/result Authorization: Bearer {{token}} Content-Type: application/json @@ -132,8 +142,7 @@ Content-Type: application/json { "worker_id": "{{worker}}", "attempt": {{attempt}}, - "result_uri": "s3://results/SOMETHING-ELSE.csv", - "result_sha256": "different" + "result": { "artifact_id": "{{artifactId2}}" } } ### 4c. Another worker submitting for this task — conflict (409) @@ -144,8 +153,7 @@ Content-Type: application/json { "worker_id": "impostor", "attempt": {{attempt}}, - "result_uri": "s3://results/x.csv", - "result_sha256": "x" + "result": { "artifact_id": "{{artifactId}}" } } ### 5. Report a failure instead (200) diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 32e0d45..95481fc 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -72,7 +72,7 @@ func run() error { 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), + CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, tx, clk), FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), UploadArtifact: usecase.NewUploadArtifact(taskRepo, artifactRepo, blobStore, clk), diff --git a/coordinator/internal/domain/task.go b/coordinator/internal/domain/task.go index 6704552..87fc555 100644 --- a/coordinator/internal/domain/task.go +++ b/coordinator/internal/domain/task.go @@ -28,27 +28,26 @@ const ErrCodeLeaseExpired = "lease_expired" // Nullable columns are pointers so "no lease" stays distinguishable from // "lease owned by the empty string" — a plain string cannot express both. type Task struct { - ID uuid.UUID - JobID uuid.UUID - ChunkIndex int - Workload string - InputURI string - InputSHA256 string - Parameters map[string]any - Status TaskStatus - Attempt int - MaxAttempts int - LeaseOwner *string - LeaseExpiresAt *time.Time - ResultURI *string - ResultSHA256 *string - Metrics map[string]any - ErrorCode *string - ErrorMessage *string - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time - Version int + ID uuid.UUID + JobID uuid.UUID + ChunkIndex int + Workload string + InputURI string + InputSHA256 string + Parameters map[string]any + Status TaskStatus + Attempt int + MaxAttempts int + LeaseOwner *string + LeaseExpiresAt *time.Time + ResultArtifactID *uuid.UUID + Metrics map[string]any + ErrorCode *string + ErrorMessage *string + CreatedAt time.Time + StartedAt *time.Time + CompletedAt *time.Time + Version int } // NewTask builds a pending task. maxAttempts <= 0 falls back to the default. @@ -146,17 +145,16 @@ func (t *Task) RenewLease(worker string, attempt int, until time.Time) error { // retry the same manifest, and that must succeed rather than trip the lease // check on a task the coordinator already finished. A *different* manifest for // an already-completed task is a genuine conflict. -func (t *Task) CompleteWith(resultURI, resultSHA256 string, metrics map[string]any, +func (t *Task) CompleteWith(resultArtifactID uuid.UUID, metrics map[string]any, worker string, attempt int, now time.Time) error { - if resultURI == "" || resultSHA256 == "" { + if resultArtifactID == uuid.Nil { return ErrInvalidInput } if t.Status == TaskCompleted { - if t.Attempt == attempt && t.ResultURI != nil && *t.ResultURI == resultURI && - t.ResultSHA256 != nil && *t.ResultSHA256 == resultSHA256 { - return nil // same attempt, same manifest — replay of a successful call + if t.Attempt == attempt && t.ResultArtifactID != nil && *t.ResultArtifactID == resultArtifactID { + return nil // same attempt, same artifact — replay of a successful call } return ErrResultConflict } @@ -166,8 +164,7 @@ func (t *Task) CompleteWith(resultURI, resultSHA256 string, metrics map[string]a } t.Status = TaskCompleted - t.ResultURI = &resultURI - t.ResultSHA256 = &resultSHA256 + t.ResultArtifactID = &resultArtifactID t.Metrics = metrics t.CompletedAt = &now t.LeaseOwner = nil @@ -234,11 +231,11 @@ type ClaimedTask struct { LeaseExpiresAt time.Time } -// ResultManifest is a completed task's output, ordered for the stitcher. +// ResultManifest is a completed task's output, ordered for the stitcher. It +// points at the coordinator-owned result artifact rather than a worker URI. type ResultManifest struct { - TaskID uuid.UUID - ChunkIndex int - ResultURI string - ResultSHA256 string - Metrics map[string]any + TaskID uuid.UUID + ChunkIndex int + ResultArtifactID uuid.UUID + Metrics map[string]any } diff --git a/coordinator/internal/domain/task_test.go b/coordinator/internal/domain/task_test.go index 3650478..7ccaccf 100644 --- a/coordinator/internal/domain/task_test.go +++ b/coordinator/internal/domain/task_test.go @@ -9,9 +9,11 @@ import ( ) var ( - testNow = time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC) - testLater = testNow.Add(time.Hour) - testWorker = "worker-1" + testNow = time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC) + testLater = testNow.Add(time.Hour) + testWorker = "worker-1" + testResult = uuid.New() + testResultAlt = uuid.New() ) // leasedTask builds a task already leased to testWorker at the given attempt. @@ -32,7 +34,7 @@ func leasedTask(attempt, maxAttempts int) *Task { func TestCompleteWithRecordsResult(t *testing.T) { task := leasedTask(1, 3) - if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + if err := task.CompleteWith(testResult, nil, testWorker, 1, testNow); err != nil { t.Fatalf("unexpected error: %v", err) } if task.Status != TaskCompleted { @@ -50,12 +52,12 @@ func TestCompleteWithRecordsResult(t *testing.T) { // rather than fail on the lease it has already given up. func TestCompleteWithIsIdempotentForSameManifest(t *testing.T) { task := leasedTask(1, 3) - if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + if err := task.CompleteWith(testResult, nil, testWorker, 1, testNow); err != nil { t.Fatalf("first call: %v", err) } versionAfterFirst := task.Version - if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testLater); err != nil { + if err := task.CompleteWith(testResult, nil, testWorker, 1, testLater); err != nil { t.Fatalf("replay must be idempotent, got %v", err) } if task.Version != versionAfterFirst { @@ -65,11 +67,11 @@ func TestCompleteWithIsIdempotentForSameManifest(t *testing.T) { func TestCompleteWithRejectsDifferentManifest(t *testing.T) { task := leasedTask(1, 3) - if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil { + if err := task.CompleteWith(testResult, nil, testWorker, 1, testNow); err != nil { t.Fatalf("first call: %v", err) } - err := task.CompleteWith("s3://other.csv", "def", nil, testWorker, 1, testLater) + err := task.CompleteWith(testResultAlt, nil, testWorker, 1, testLater) if !errors.Is(err, ErrResultConflict) { t.Errorf("err = %v, want ErrResultConflict", err) } @@ -78,7 +80,7 @@ func TestCompleteWithRejectsDifferentManifest(t *testing.T) { func TestCompleteWithRejectsForeignWorker(t *testing.T) { task := leasedTask(1, 3) - err := task.CompleteWith("s3://r.csv", "abc", nil, "worker-2", 1, testNow) + err := task.CompleteWith(testResult, nil, "worker-2", 1, testNow) if !errors.Is(err, ErrLeaseConflict) { t.Errorf("err = %v, want ErrLeaseConflict", err) } @@ -87,7 +89,7 @@ func TestCompleteWithRejectsForeignWorker(t *testing.T) { func TestCompleteWithRejectsStaleAttempt(t *testing.T) { task := leasedTask(2, 3) // task is on attempt 2 - err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow) // worker thinks it is 1 + err := task.CompleteWith(testResult, nil, testWorker, 1, testNow) // worker thinks it is 1 if !errors.Is(err, ErrStaleAttempt) { t.Errorf("err = %v, want ErrStaleAttempt", err) } diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go index d9e75df..d34f122 100644 --- a/coordinator/internal/storage/postgres/task_repo.go +++ b/coordinator/internal/storage/postgres/task_repo.go @@ -32,7 +32,7 @@ var _ usecase.TaskRepository = (*TaskRepo)(nil) var taskColumns = []string{ "id", "job_id", "chunk_index", "workload", "input_uri", "input_sha256", "parameters", "status", "attempt", "max_attempts", "lease_owner", "lease_expires_at", - "result_uri", "result_sha256", "metrics", "error_code", "error_message", + "result_artifact_id", "metrics", "error_code", "error_message", "created_at", "started_at", "completed_at", "version", } @@ -53,7 +53,7 @@ func scanTask(row pgx.Row) (*domain.Task, error) { err := row.Scan( &t.ID, &t.JobID, &t.ChunkIndex, &t.Workload, &t.InputURI, &t.InputSHA256, &t.Parameters, &status, &t.Attempt, &t.MaxAttempts, &t.LeaseOwner, &t.LeaseExpiresAt, - &t.ResultURI, &t.ResultSHA256, &t.Metrics, &t.ErrorCode, &t.ErrorMessage, + &t.ResultArtifactID, &t.Metrics, &t.ErrorCode, &t.ErrorMessage, &t.CreatedAt, &t.StartedAt, &t.CompletedAt, &t.Version, ) if err != nil { @@ -166,18 +166,17 @@ func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task func (r *TaskRepo) Update(ctx context.Context, t *domain.Task) error { sql, args, err := psql.Update("tasks"). SetMap(map[string]any{ - "status": string(t.Status), - "attempt": t.Attempt, - "lease_owner": t.LeaseOwner, - "lease_expires_at": t.LeaseExpiresAt, - "result_uri": t.ResultURI, - "result_sha256": t.ResultSHA256, - "metrics": t.Metrics, - "error_code": t.ErrorCode, - "error_message": t.ErrorMessage, - "started_at": t.StartedAt, - "completed_at": t.CompletedAt, - "version": t.Version, + "status": string(t.Status), + "attempt": t.Attempt, + "lease_owner": t.LeaseOwner, + "lease_expires_at": t.LeaseExpiresAt, + "result_artifact_id": t.ResultArtifactID, + "metrics": t.Metrics, + "error_code": t.ErrorCode, + "error_message": t.ErrorMessage, + "started_at": t.StartedAt, + "completed_at": t.CompletedAt, + "version": t.Version, }). Where(sq.Eq{"id": t.ID, "version": t.Version - 1}). ToSql() diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go index 6523c09..6e014a3 100644 --- a/coordinator/internal/transport/http/dto.go +++ b/coordinator/internal/transport/http/dto.go @@ -54,11 +54,19 @@ type heartbeatRequest struct { } type resultRequest struct { - WorkerID string `json:"worker_id"` - Attempt int `json:"attempt"` - ResultURI string `json:"result_uri"` - ResultSHA256 string `json:"result_sha256"` - Metrics map[string]any `json:"metrics"` + WorkerID string `json:"worker_id"` + Attempt int `json:"attempt"` + Result resultManifest `json:"result"` + Metrics map[string]any `json:"metrics"` +} + +// resultManifest references the artifact the worker already uploaded. sha256 and +// content_type are accepted for the worker's own cross-checking; the coordinator +// trusts its own stored metadata, not these. +type resultManifest struct { + ArtifactID uuid.UUID `json:"artifact_id"` + SHA256 string `json:"sha256"` + ContentType string `json:"content_type"` } type failureRequest struct { diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go index 9dbe659..d622e0f 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/coordinator/internal/transport/http/handlers.go @@ -133,12 +133,11 @@ func (s *Server) handleResult(w http.ResponseWriter, r *http.Request) { } task, err := s.uc.CompleteTask.Execute(ctx, usecase.CompleteTaskInput{ - TaskID: taskID, - WorkerID: req.WorkerID, - Attempt: req.Attempt, - ResultURI: req.ResultURI, - ResultSHA256: req.ResultSHA256, - Metrics: req.Metrics, + TaskID: taskID, + WorkerID: req.WorkerID, + Attempt: req.Attempt, + ResultArtifactID: req.Result.ArtifactID, + Metrics: req.Metrics, }) if err != nil { s.writeError(w, r, err) diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go index de4839c..d784b71 100644 --- a/coordinator/internal/usecase/dto.go +++ b/coordinator/internal/usecase/dto.go @@ -42,12 +42,11 @@ type RenewLeaseInput struct { } type CompleteTaskInput struct { - TaskID uuid.UUID - WorkerID string - Attempt int - ResultURI string - ResultSHA256 string - Metrics map[string]any + TaskID uuid.UUID + WorkerID string + Attempt int + ResultArtifactID uuid.UUID + Metrics map[string]any } type UploadArtifactInput struct { diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go index 1fc455f..6552565 100644 --- a/coordinator/internal/usecase/job.go +++ b/coordinator/internal/usecase/job.go @@ -99,15 +99,14 @@ func (uc *ListResults) Execute(ctx context.Context, jobID uuid.UUID) ([]domain.R manifests := make([]domain.ResultManifest, 0, len(tasks)) for _, t := range tasks { - if t.ResultURI == nil || t.ResultSHA256 == nil { - continue // a completed task always carries both; skip defensively + if t.ResultArtifactID == nil { + continue // a completed task always references its result; skip defensively } manifests = append(manifests, domain.ResultManifest{ - TaskID: t.ID, - ChunkIndex: t.ChunkIndex, - ResultURI: *t.ResultURI, - ResultSHA256: *t.ResultSHA256, - Metrics: t.Metrics, + TaskID: t.ID, + ChunkIndex: t.ChunkIndex, + ResultArtifactID: *t.ResultArtifactID, + Metrics: t.Metrics, }) } return manifests, nil diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index da177bf..0451187 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -4,6 +4,8 @@ import ( "context" "time" + "github.com/google/uuid" + "github.com/emil28092005/SciMesh/coordinator/internal/domain" ) @@ -105,14 +107,16 @@ func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain. // --- CompleteTask -------------------------------------------------------- type CompleteTask struct { - tasks TaskRepository - jobs JobRepository - tx TxManager - clock Clock + tasks TaskRepository + jobs JobRepository + artifacts ArtifactRepository + tx TxManager + clock Clock } -func NewCompleteTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *CompleteTask { - return &CompleteTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock} +func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts ArtifactRepository, + tx TxManager, clock Clock) *CompleteTask { + return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, tx: tx, clock: clock} } // Execute applies the result and, when that was the job's last outstanding @@ -129,9 +133,14 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom if err != nil { return err } + // Rule 10: never trust a worker-supplied artifact reference. The result + // must be an artifact the coordinator itself stored for *this* task. + if err := uc.verifyResultArtifact(ctx, in.TaskID, in.ResultArtifactID); err != nil { + return err + } now := uc.clock.Now() before := task.Version - if err := task.CompleteWith(in.ResultURI, in.ResultSHA256, in.Metrics, + if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, in.WorkerID, in.Attempt, now); err != nil { return err } @@ -155,6 +164,20 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom return out, nil } +// verifyResultArtifact enforces that the referenced artifact was stored by the +// coordinator for this exact task. It stops a worker from completing task B with +// an artifact it uploaded for task A, and from naming an id that isn't a result. +func (uc *CompleteTask) verifyResultArtifact(ctx context.Context, taskID, artifactID uuid.UUID) error { + art, err := uc.artifacts.Get(ctx, artifactID) + if err != nil { + return err + } + if art.TaskID == nil || *art.TaskID != taskID || art.Kind != domain.ArtifactPartialResult { + return domain.ErrResultConflict + } + return nil +} + // --- FailTask ------------------------------------------------------------ type FailTask struct { diff --git a/coordinator/migrations/0004_result_artifact.down.sql b/coordinator/migrations/0004_result_artifact.down.sql new file mode 100644 index 0000000..acbbdb3 --- /dev/null +++ b/coordinator/migrations/0004_result_artifact.down.sql @@ -0,0 +1,11 @@ +BEGIN; + +ALTER TABLE tasks DROP CONSTRAINT IF EXISTS ck_tasks_completed_result; +ALTER TABLE tasks ADD COLUMN result_uri text; +ALTER TABLE tasks ADD COLUMN result_sha256 text; + +ALTER TABLE tasks ADD CONSTRAINT ck_tasks_completed_result CHECK ( + status <> 'completed' OR (result_uri IS NOT NULL AND result_sha256 IS NOT NULL) +); + +COMMIT; diff --git a/coordinator/migrations/0004_result_artifact.up.sql b/coordinator/migrations/0004_result_artifact.up.sql new file mode 100644 index 0000000..e062e1a --- /dev/null +++ b/coordinator/migrations/0004_result_artifact.up.sql @@ -0,0 +1,14 @@ +BEGIN; + +-- Results are now coordinator-owned artifacts, not worker-supplied URIs. +-- Drop the URI-based completion guard and columns, and require a completed task +-- to reference its result artifact instead (PLAN.md §6.2). +ALTER TABLE tasks DROP CONSTRAINT IF EXISTS ck_tasks_completed_result; +ALTER TABLE tasks DROP COLUMN IF EXISTS result_uri; +ALTER TABLE tasks DROP COLUMN IF EXISTS result_sha256; + +ALTER TABLE tasks ADD CONSTRAINT ck_tasks_completed_result CHECK ( + status <> 'completed' OR result_artifact_id IS NOT NULL +); + +COMMIT; diff --git a/coordinator/scripts/smoke.sh b/coordinator/scripts/smoke.sh index 70c3550..762b784 100755 --- a/coordinator/scripts/smoke.sh +++ b/coordinator/scripts/smoke.sh @@ -102,8 +102,17 @@ 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) --------------------------- +# --- artifacts + result (uploads happen while the task is still leased) --- bearer=(-H "Authorization: Bearer ${TOKEN}") + +# upload -> prints the artifact_id +upload() { + curl -sS -X PUT "${HOST}/tasks/${task_id}/artifacts/$1" "${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' | + python3 -c 'import json,sys;print(json.load(sys.stdin)["artifact_id"])' 2>/dev/null +} + 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' @@ -111,26 +120,26 @@ check "foreign worker upload → 409" 409 -X PUT "${HOST}/tasks/${task_id -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) +# Two result artifacts, uploaded now while the lease is held: one to complete +# with, a second to prove a different manifest is rejected after completion. +art_id=$(upload primary.csv) +art_id2=$(upload secondary.csv) 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\"}" + -d "{\"worker_id\":\"impostor\",\"attempt\":${attempt},\"result\":{\"artifact_id\":\"${art_id}\"}}" check "submit result" 200 -X POST "${HOST}/tasks/${task_id}/result" "${auth[@]}" \ - -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result_uri\":\"s3://r0\",\"result_sha256\":\"rrr\"}" + -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result\":{\"artifact_id\":\"${art_id}\"}}" check "replay same result → idempotent" 200 -X POST "${HOST}/tasks/${task_id}/result" "${auth[@]}" \ - -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result_uri\":\"s3://r0\",\"result_sha256\":\"rrr\"}" + -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result\":{\"artifact_id\":\"${art_id}\"}}" check "different result → 409" 409 -X POST "${HOST}/tasks/${task_id}/result" "${auth[@]}" \ - -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result_uri\":\"s3://other\",\"result_sha256\":\"zzz\"}" + -d "{\"worker_id\":\"w1\",\"attempt\":${attempt},\"result\":{\"artifact_id\":\"${art_id2}\"}}" check "GET /jobs/{id}" 200 "${HOST}/jobs/${job_id}" "${auth[@]}" echo echo "input validation" check "malformed uuid → 400" 400 -X POST "${HOST}/tasks/not-a-uuid/result" "${auth[@]}" \ - -d '{"worker_id":"w1","attempt":1,"result_uri":"s3://x","result_sha256":"x"}' + -d '{"worker_id":"w1","attempt":1,"result":{"artifact_id":"00000000-0000-0000-0000-000000000000"}}' # Note: Go's encoding/json matches field names case-insensitively, so # "worker_ID" would be accepted as "worker_id". Only a genuinely unknown key # trips DisallowUnknownFields. diff --git a/docs/api-contract.md b/docs/api-contract.md index 603dc20..e67346c 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -19,7 +19,7 @@ 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` next | +| `POST /tasks/{id}/result` | complete | ✅ done, references `artifact_id` | | `POST /tasks/{id}/failure` | fail | ✅ done | | `GET /jobs/{id}` | progress | ✅ done | | `PUT /tasks/{id}/artifacts/{name}` | upload partial | ✅ done | @@ -146,10 +146,10 @@ Content-Type: application/json } ``` -> **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. +The worker uploads its partial result first (§5.5), then completes with that +`artifact_id`. The coordinator verifies the artifact was stored for this exact +task before accepting it — a worker cannot complete one task with another task's +artifact. No worker-supplied URI is ever persisted. ```http POST /tasks/{task_id}/failure