Emil_Shanaty/SciMesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(coordinator): complete tasks with an artifact_id, not a URI (CTX-05, part 3)

Browse Source 
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.
This commit is contained in:
Efremenko Arhip
2026-07-23 15:54:11 +03:00
parent 6d45406ee0
commit 58da6ef139
14 changed files with 184 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coordinator/internal/domain/task.go
+31 -34
View File
@@ -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
}
coordinator/internal/domain/task_test.go
+12 -10
View File
@@ -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)
}
coordinator/internal/storage/postgres/task_repo.go
+13 -14
View File
@@ -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()
coordinator/internal/transport/http/dto.go
+13 -5
View File
@@ -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 {
coordinator/internal/transport/http/handlers.go
+5 -6
View File
@@ -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)
coordinator/internal/usecase/dto.go
+5 -6
View File
@@ -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 {
coordinator/internal/usecase/job.go
+6 -7
View File
@@ -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
coordinator/internal/usecase/task.go
+30 -7
View File
@@ -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 {