diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index f9f26f6..dd6e33d 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -60,13 +60,14 @@ func run() error { } var ( - clk = infra.NewClock() - tx = postgres.NewTxManager(pool) - taskRepo = postgres.NewTaskRepo(pool) - jobRepo = postgres.NewJobRepo(pool) - workerRepo = postgres.NewWorkerRepo(pool) - artifactRepo = postgres.NewArtifactRepo(pool) - uiReadRepo = postgres.NewUIReadRepo(pool) + clk = infra.NewClock() + tx = postgres.NewTxManager(pool) + taskRepo = postgres.NewTaskRepo(pool) + jobRepo = postgres.NewJobRepo(pool) + workerRepo = postgres.NewWorkerRepo(pool) + artifactRepo = postgres.NewArtifactRepo(pool) + uiReadRepo = postgres.NewUIReadRepo(pool) + taskResultRepo = postgres.NewTaskResultRepo(pool) ) useCases := httptransport.UseCases{ @@ -75,7 +76,7 @@ func run() error { SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts), 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), + CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize), ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk), FailTask: usecase.NewFailTask(taskRepo, jobRepo, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), diff --git a/coordinator/internal/domain/task.go b/coordinator/internal/domain/task.go index 7864660..1148f18 100644 --- a/coordinator/internal/domain/task.go +++ b/coordinator/internal/domain/task.go @@ -24,6 +24,10 @@ const ( // ErrCodeLeaseExpired marks tasks failed by the reaper rather than by a worker. const ErrCodeLeaseExpired = "lease_expired" +// ErrCodeQuorumFailed marks a task whose untrusted results never reached a +// verifying quorum before its attempts ran out. +const ErrCodeQuorumFailed = "quorum_failed" + // Task is one independently executable chunk of a job. // // Nullable columns are pointers so "no lease" stays distinguishable from @@ -216,6 +220,33 @@ func (t *Task) CompleteWith(resultArtifactID uuid.UUID, metrics map[string]any, return nil } +// ReleaseAfterVote returns an untrusted worker's task to the queue after its +// result was recorded as a quorum vote but quorum was not yet reached, so a +// different owner can compute it independently. When no attempts remain the task +// fails: its untrusted results could not be verified. +func (t *Task) ReleaseAfterVote(worker string, attempt int, now time.Time) error { + if t.Status == TaskCompleted { + return nil // settled by a concurrent quorum + } + if err := t.verifyLease(worker, attempt, now); err != nil { + return err + } + t.LeaseOwner = nil + t.LeaseExpiresAt = nil + t.Version++ + + if t.CanRetry() { + t.Status = TaskPending + return nil + } + code, msg := ErrCodeQuorumFailed, "untrusted results did not reach quorum" + t.ErrorCode = &code + t.ErrorMessage = &msg + t.Status = TaskFailed + t.CompletedAt = &now + return nil +} + // Fail records a worker-reported failure. A retryable failure with attempts // left returns the task to the queue; otherwise it terminates as failed. func (t *Task) Fail(worker string, attempt int, code, message string, retryable bool, now time.Time) error { diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go index 3d62e02..5a414bd 100644 --- a/coordinator/internal/infra/config.go +++ b/coordinator/internal/infra/config.go @@ -61,6 +61,9 @@ type Config struct { LeaseDuration time.Duration // Default attempt ceiling for newly created tasks. DefaultMaxAttempts int + // How many distinct owners must agree on an untrusted result before it is + // accepted (trusted workers are accepted directly). + QuorumSize int // How often the background lease-reaper runs. ReaperInterval time.Duration // A worker silent for longer than this is marked offline by the reaper. @@ -103,6 +106,7 @@ func LoadConfig() (Config, error) { HeartbeatInterval: 15 * time.Second, LeaseDuration: 2 * time.Minute, DefaultMaxAttempts: 3, + QuorumSize: 2, ReaperInterval: 30 * time.Second, WorkerOfflineAfter: 1 * time.Minute, } @@ -147,6 +151,12 @@ func LoadConfig() (Config, error) { if cfg.DefaultMaxAttempts, err = getEnvInt("DEFAULT_MAX_ATTEMPTS", cfg.DefaultMaxAttempts); err != nil { return Config{}, err } + if cfg.QuorumSize, err = getEnvInt("QUORUM_SIZE", cfg.QuorumSize); err != nil { + return Config{}, err + } + if cfg.QuorumSize < 1 { + return Config{}, fmt.Errorf("QUORUM_SIZE must be positive") + } if cfg.DefaultMaxAttempts < 1 { return Config{}, fmt.Errorf("DEFAULT_MAX_ATTEMPTS must be positive") } diff --git a/coordinator/internal/memstore/memstore.go b/coordinator/internal/memstore/memstore.go index 28f4318..9c911e4 100644 --- a/coordinator/internal/memstore/memstore.go +++ b/coordinator/internal/memstore/memstore.go @@ -417,3 +417,36 @@ func contains(ss []string, s string) bool { } return false } + +// TaskResultRepo is an in-memory usecase.TaskResultRepository: one vote per +// (task, owner). +type TaskResultRepo struct { + mu sync.Mutex + votes map[uuid.UUID]map[uuid.UUID]string // taskID -> ownerID -> sha256 +} + +func NewTaskResultRepo() *TaskResultRepo { + return &TaskResultRepo{votes: make(map[uuid.UUID]map[uuid.UUID]string)} +} + +func (r *TaskResultRepo) RecordVote(_ context.Context, taskID, ownerID uuid.UUID, sha256 string, _ uuid.UUID) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.votes[taskID] == nil { + r.votes[taskID] = make(map[uuid.UUID]string) + } + r.votes[taskID][ownerID] = sha256 + return nil +} + +func (r *TaskResultRepo) CountAgreeing(_ context.Context, taskID uuid.UUID, sha256 string) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + n := 0 + for _, s := range r.votes[taskID] { + if s == sha256 { + n++ + } + } + return n, nil +} diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go index 11b0616..a373087 100644 --- a/coordinator/internal/storage/postgres/task_repo.go +++ b/coordinator/internal/storage/postgres/task_repo.go @@ -84,6 +84,9 @@ WITH candidate AS ( WHERE status = 'pending' AND attempt < max_attempts AND (cardinality($1::text[]) = 0 OR workload = ANY($1)) + AND ($5::uuid IS NULL OR NOT EXISTS ( + SELECT 1 FROM task_results tr + WHERE tr.task_id = tasks.id AND tr.owner_id = $5)) ORDER BY created_at, chunk_index FOR UPDATE SKIP LOCKED LIMIT 1 @@ -108,7 +111,7 @@ func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domai var task *domain.Task err := withRetry(ctx, func(ctx context.Context) error { - row := conn(ctx, r.pool).QueryRow(ctx, claimNextSQL, workloads, f.Owner, f.LeaseUntil, f.Now) + row := conn(ctx, r.pool).QueryRow(ctx, claimNextSQL, workloads, f.Owner, f.LeaseUntil, f.Now, f.VoterOwner) t, err := scanTask(row) if errors.Is(err, pgx.ErrNoRows) { task = nil diff --git a/coordinator/internal/storage/postgres/task_result_repo.go b/coordinator/internal/storage/postgres/task_result_repo.go new file mode 100644 index 0000000..aa5907a --- /dev/null +++ b/coordinator/internal/storage/postgres/task_result_repo.go @@ -0,0 +1,45 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// TaskResultRepo records and tallies quorum votes for untrusted task results. +type TaskResultRepo struct { + pool *pgxpool.Pool +} + +func NewTaskResultRepo(pool *pgxpool.Pool) *TaskResultRepo { + return &TaskResultRepo{pool: pool} +} + +// RecordVote stores (or replaces) one owner's vote for a task's result. +func (r *TaskResultRepo) RecordVote(ctx context.Context, taskID, ownerID uuid.UUID, sha256 string, artifactID uuid.UUID) error { + const sql = ` +INSERT INTO task_results (task_id, owner_id, result_sha256, result_artifact_id) +VALUES ($1, $2, $3, $4) +ON CONFLICT (task_id, owner_id) DO UPDATE +SET result_sha256 = EXCLUDED.result_sha256, + result_artifact_id = EXCLUDED.result_artifact_id, + created_at = now()` + if _, err := conn(ctx, r.pool).Exec(ctx, sql, taskID, ownerID, sha256, artifactID); err != nil { + return fmt.Errorf("record vote: %w", err) + } + return nil +} + +// CountAgreeing returns how many distinct owners have voted for the given result +// hash on this task — the size of the agreeing set the quorum is measured +// against. +func (r *TaskResultRepo) CountAgreeing(ctx context.Context, taskID uuid.UUID, sha256 string) (int, error) { + const sql = `SELECT count(DISTINCT owner_id) FROM task_results WHERE task_id = $1 AND result_sha256 = $2` + var n int + if err := conn(ctx, r.pool).QueryRow(ctx, sql, taskID, sha256).Scan(&n); err != nil { + return 0, fmt.Errorf("count agreeing: %w", err) + } + return n, nil +} diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 96dc777..56cb049 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -50,7 +50,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3), 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), + CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2), ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk), FailTask: usecase.NewFailTask(tasks, jobs, tx, clk), GetJobStatus: usecase.NewGetJobStatus(jobs, tasks), diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index 933453c..d903816 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -23,6 +23,15 @@ type ClaimFilter struct { Owner string // worker ID taking the lease Now time.Time LeaseUntil time.Time + // VoterOwner, when set, excludes tasks this owner has already voted on, so + // an untrusted worker never verifies its own chunk twice. + VoterOwner *uuid.UUID +} + +// TaskResultRepository records and tallies quorum votes for untrusted results. +type TaskResultRepository interface { + RecordVote(ctx context.Context, taskID, ownerID uuid.UUID, sha256 string, artifactID uuid.UUID) error + CountAgreeing(ctx context.Context, taskID uuid.UUID, sha256 string) (int, error) } // TaskRepository persists tasks. diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index 1d96591..0fec738 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -2,6 +2,7 @@ package usecase import ( "context" + "errors" "time" "github.com/google/uuid" @@ -47,6 +48,7 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl return nil, domain.ErrInvalidInput } workloads := in.Workloads + var voterOwner *uuid.UUID if workerID, err := uuid.Parse(in.WorkerID); err == nil { worker, err := uc.workers.Get(ctx, workerID) if err != nil { @@ -55,21 +57,18 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl // Bind the caller to the worker it claims as. A JWT-authenticated // volunteer may operate only its own workers; without this the trust // tier would be read off a caller-supplied worker_id, letting anyone who - // knows a trusted worker's id claim as it and bypass the quarantine - // below. A shared-token caller (no requester) is a lab operator and may - // act as any worker, preserving the original behaviour. + // knows a trusted worker's id claim as it. A shared-token caller (no + // requester) is a lab operator and may act as any worker. if r, ok := authctx.From(ctx); ok { if worker.OwnerID == nil || *worker.OwnerID != r.UserID { // Don't disclose that another user's worker exists. return nil, domain.ErrWorkerNotFound } } - // C1 quarantine: an untrusted volunteer worker may register but receives - // no tasks, because there is not yet (until quorum, C2) any way to verify - // its results. Report an empty queue rather than an error, so its poller - // simply idles. + // An untrusted volunteer may claim, but never a chunk its owner has + // already voted on — so quorum needs genuinely independent computations. if worker.TrustLevel == domain.WorkerUntrusted { - return nil, nil + voterOwner = worker.OwnerID } // Never trust caller-supplied capabilities: registration is the durable // worker identity and its allowlist. @@ -91,6 +90,7 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl Owner: in.WorkerID, Now: now, LeaseUntil: now.Add(uc.leaseDuration), + VoterOwner: voterOwner, }) if err != nil { return err @@ -161,13 +161,22 @@ type CompleteTask struct { tasks TaskRepository jobs JobRepository artifacts ArtifactRepository + workers WorkerRepository + results TaskResultRepository tx TxManager clock Clock + // quorum is how many distinct owners must agree on an untrusted result + // before it is accepted; a trusted worker's result is accepted directly. + quorum int } 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} + workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int) *CompleteTask { + if quorum < 1 { + quorum = 2 + } + return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, workers: workers, + results: results, tx: tx, clock: clock, quorum: quorum} } // Execute applies the result and, when that was the job's last outstanding @@ -186,24 +195,35 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom } // 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.Attempt, in.ResultArtifactID); err != nil { + art, err := uc.verifyResultArtifact(ctx, in.TaskID, in.Attempt, in.ResultArtifactID) + if err != nil { + return err + } + + trusted, ownerID, err := uc.workerTrust(ctx, in.WorkerID) + if err != nil { return err } now := uc.clock.Now() + + // Untrusted (volunteer) worker: record a vote and only complete once a + // quorum of distinct owners agree; otherwise return the task to the queue. + if !trusted { + return uc.recordVote(ctx, task, in, art, ownerID, now, &out) + } + + // Trusted worker (lab token, verified, or admin): accept directly. before := task.Version - if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, - in.WorkerID, in.Attempt, now); err != nil { + if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, in.WorkerID, in.Attempt, now); err != nil { return err } out = task - // A replay of an already-recorded result leaves the entity untouched. // Writing anyway would fail the optimistic-concurrency guard (the stored // version already equals ours) and turn an idempotent call into a 409. if task.Version == before { return nil } - if err := uc.tasks.Update(ctx, task); err != nil { return err } @@ -215,18 +235,81 @@ 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 uuid.UUID, attempt int, artifactID uuid.UUID) error { - art, err := uc.artifacts.Get(ctx, artifactID) +// recordVote handles an untrusted result: it stores the vote, then completes the +// task when the submitter's result hash has reached quorum, or returns the task +// to the queue so another owner can compute it independently. +func (uc *CompleteTask) recordVote(ctx context.Context, task *domain.Task, in CompleteTaskInput, + art *domain.Artifact, ownerID uuid.UUID, now time.Time, out **domain.Task) error { + + *out = task + if task.Status == domain.TaskCompleted { + return nil // already settled by an earlier quorum; nothing to record + } + if err := uc.results.RecordVote(ctx, task.ID, ownerID, art.SHA256, in.ResultArtifactID); err != nil { + return err + } + agree, err := uc.results.CountAgreeing(ctx, task.ID, art.SHA256) if err != nil { return err } - if art.TaskID == nil || *art.TaskID != taskID || art.Attempt == nil || *art.Attempt != attempt || art.Kind != domain.ArtifactPartialResult { - return domain.ErrResultConflict + + if agree >= uc.quorum { + // The submitter's own (already verified) artifact carries the winning + // hash, so complete with it. + if err := task.CompleteWith(in.ResultArtifactID, in.Metrics, in.WorkerID, in.Attempt, now); err != nil { + return err + } + } else if err := task.ReleaseAfterVote(in.WorkerID, in.Attempt, now); err != nil { + return err } - return nil + + if err := uc.tasks.Update(ctx, task); err != nil { + return err + } + return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) +} + +// workerTrust reports whether the worker's results are accepted directly, and +// the owner to attribute a vote to when they are not. +func (uc *CompleteTask) workerTrust(ctx context.Context, workerID string) (trusted bool, ownerID uuid.UUID, err error) { + // When the worker can't be resolved, default to trusted — the pre-quorum + // behaviour. This is safe because completing a task requires holding its + // lease, and the lease owner is always a real registered worker whose trust + // is therefore known; only an untrusted worker ever takes the quorum path. + id, err := uuid.Parse(workerID) + if err != nil { + return true, uuid.Nil, nil + } + w, err := uc.workers.Get(ctx, id) + if err != nil { + if errors.Is(err, domain.ErrWorkerNotFound) { + return true, uuid.Nil, nil + } + return false, uuid.Nil, err + } + if w.TrustLevel != domain.WorkerUntrusted { + return true, uuid.Nil, nil + } + if w.OwnerID == nil { + // An untrusted worker always has an owner (it registered via a user JWT); + // a missing one is a data error, not a silent trust upgrade. + return false, uuid.Nil, domain.ErrInvalidInput + } + return false, *w.OwnerID, 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 uuid.UUID, attempt int, artifactID uuid.UUID) (*domain.Artifact, error) { + art, err := uc.artifacts.Get(ctx, artifactID) + if err != nil { + return nil, err + } + if art.TaskID == nil || *art.TaskID != taskID || art.Attempt == nil || *art.Attempt != attempt || art.Kind != domain.ArtifactPartialResult { + return nil, domain.ErrResultConflict + } + return art, nil } // --- FailTask ------------------------------------------------------------ diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index c0a4566..246c452 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -35,12 +35,13 @@ func (s expiringBlobStore) Put(ctx context.Context, key string, body io.Reader) // harness wires every use case to in-memory stores so orchestration can be // tested without a database. type harness struct { - tasks *memstore.TaskRepo - jobs *memstore.JobRepo - work *memstore.WorkerRepo - arts *memstore.ArtifactRepo - blobs *memstore.BlobStore - clk *memstore.Clock + tasks *memstore.TaskRepo + jobs *memstore.JobRepo + work *memstore.WorkerRepo + arts *memstore.ArtifactRepo + blobs *memstore.BlobStore + clk *memstore.Clock + taskResults *memstore.TaskResultRepo createJob *usecase.CreateJob submit *usecase.SubmitDataset @@ -62,19 +63,20 @@ type harness struct { func newHarness() *harness { h := &harness{ - tasks: memstore.NewTaskRepo(), - jobs: memstore.NewJobRepo(), - work: memstore.NewWorkerRepo(), - arts: memstore.NewArtifactRepo(), - blobs: memstore.NewBlobStore(), - clk: memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)), + tasks: memstore.NewTaskRepo(), + jobs: memstore.NewJobRepo(), + work: memstore.NewWorkerRepo(), + arts: memstore.NewArtifactRepo(), + blobs: memstore.NewBlobStore(), + clk: memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)), + taskResults: memstore.NewTaskResultRepo(), } tx := memstore.Tx{} h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk) h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3) h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease) h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease) - h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, tx, h.clk) + h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2) h.fail = usecase.NewFailTask(h.tasks, h.jobs, tx, h.clk) h.status = usecase.NewGetJobStatus(h.jobs, h.tasks) h.results = usecase.NewListResults(h.tasks) @@ -333,9 +335,9 @@ func TestJWTCallerClaimsAsOwnTrustedWorker(t *testing.T) { } } -func TestUntrustedWorkerIsQuarantinedFromClaims(t *testing.T) { +func TestUntrustedWorkerCanClaim(t *testing.T) { h := newHarness() - h.seedJob(t, "w", 1) // a task is waiting + h.seedJob(t, "w", 1) owner := uuid.New() worker, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ Name: "volunteer", Capabilities: []string{"w"}, @@ -344,22 +346,73 @@ func TestUntrustedWorkerIsQuarantinedFromClaims(t *testing.T) { if err != nil { t.Fatal(err) } - - // Even with a matching task available, an untrusted worker gets nothing: - // its results cannot be verified until quorum (C2) exists. + // Volunteers are no longer quarantined — they may claim; their results are + // gated by quorum at completion, not by withholding work. claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: worker.ID.String()}) - if err != nil { - t.Fatalf("claim: %v", err) + if err != nil || claimed == nil { + t.Fatalf("untrusted claim = (%v, %v), want a task", claimed, err) } - if claimed != nil { - t.Error("untrusted worker must receive no task (quarantine)") +} + +// registerUntrusted registers a volunteer worker under a fresh owner. +func (h *harness) registerUntrusted(t *testing.T, name, workload string) (*domain.Worker, uuid.UUID) { + t.Helper() + owner := uuid.New() + w, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{ + Name: name, Capabilities: []string{workload}, + OwnerID: &owner, TrustLevel: domain.WorkerUntrusted, + }) + if err != nil { + t.Fatal(err) + } + return w, owner +} + +func TestUntrustedResultNeedsQuorum(t *testing.T) { + h := newHarness() + jobID := h.seedJob(t, "w", 1) + if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + w1, _ := h.registerUntrusted(t, "v1", "w") + w2, _ := h.registerUntrusted(t, "v2", "w") + + // First volunteer computes and submits — one vote, not yet quorum (2). + taskID, attempt := h.leaseOne(t, w1.ID.String(), "w") + art1 := h.uploadResult(t, taskID, w1.ID.String(), attempt) + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: w1.ID.String(), Attempt: attempt, ResultArtifactID: art1}); err != nil { + t.Fatalf("first vote: %v", err) + } + if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskPending { + t.Fatalf("after one vote status = %s, want pending", tk.Status) } - // A trusted worker still drains the same queue. - trusted, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{Name: "lab", Capabilities: []string{"w"}}) - got, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: trusted.ID.String()}) - if err != nil || got == nil { - t.Fatalf("trusted claim = (%v, %v), want a task", got, err) + // Second volunteer (distinct owner) computes the same bytes -> quorum -> done. + taskID2, attempt2 := h.leaseOne(t, w2.ID.String(), "w") + art2 := h.uploadResult(t, taskID2, w2.ID.String(), attempt2) + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID2, WorkerID: w2.ID.String(), Attempt: attempt2, ResultArtifactID: art2}); err != nil { + t.Fatalf("second vote: %v", err) + } + if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskCompleted { + t.Fatalf("after quorum status = %s, want completed", tk.Status) + } +} + +func TestTrustedResultCompletesDirectly(t *testing.T) { + h := newHarness() + jobID := h.seedJob(t, "w", 1) + if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + // A trusted (default) worker's single result completes the task immediately. + worker, _ := h.register.Execute(ctx, usecase.RegisterWorkerInput{Name: "lab", Capabilities: []string{"w"}}) + taskID, attempt := h.leaseOne(t, worker.ID.String(), "w") + art := h.uploadResult(t, taskID, worker.ID.String(), attempt) + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: worker.ID.String(), Attempt: attempt, ResultArtifactID: art}); err != nil { + t.Fatal(err) + } + if tk, _ := h.tasks.Get(ctx, taskID); tk.Status != domain.TaskCompleted { + t.Fatalf("trusted result status = %s, want completed", tk.Status) } } diff --git a/coordinator/migrations/0013_task_results.down.sql b/coordinator/migrations/0013_task_results.down.sql new file mode 100644 index 0000000..dc4fc2c --- /dev/null +++ b/coordinator/migrations/0013_task_results.down.sql @@ -0,0 +1,5 @@ +BEGIN; + +DROP TABLE IF EXISTS task_results; + +COMMIT; diff --git a/coordinator/migrations/0013_task_results.up.sql b/coordinator/migrations/0013_task_results.up.sql new file mode 100644 index 0000000..c71f33f --- /dev/null +++ b/coordinator/migrations/0013_task_results.up.sql @@ -0,0 +1,23 @@ +BEGIN; + +-- Quorum votes for a task computed by untrusted (volunteer) workers. A trusted +-- worker's result completes the task directly and never lands here; an untrusted +-- result is recorded as one vote, and the task is only completed once enough +-- distinct owners submit the same result_sha256. +-- +-- One vote per (task, owner): a single volunteer cannot stuff the ballot by +-- running many workers under one account. A resubmission updates their vote. +CREATE TABLE task_results ( + task_id uuid NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + owner_id uuid NOT NULL, + result_sha256 text NOT NULL, + result_artifact_id uuid NOT NULL REFERENCES artifacts(id) ON DELETE CASCADE, + created_at timestamptz NOT NULL DEFAULT now(), + + PRIMARY KEY (task_id, owner_id) +); + +-- Quorum check groups a task's votes by result_sha256. +CREATE INDEX ix_task_results_quorum ON task_results (task_id, result_sha256); + +COMMIT;