From 18d58cce847b4bb5724704d7b60d7154e2123aed Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Sun, 26 Jul 2026 22:55:51 +0300 Subject: [PATCH] feat(coordinator): quorum verification for untrusted (volunteer) results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the C1 quarantine with real verification. Trusted results (lab token, verified, or admin worker) are accepted directly as before. An untrusted worker's result is recorded as one vote per (task, owner) in a new task_results table; the task only completes once QUORUM_SIZE distinct owners submit the same result hash, otherwise it returns to the queue for another independent compute. - migration 0013 task_results (one vote per owner, quorum by result_sha256) - CompleteTask branches on worker trust; unknown worker defaults trusted (safe: completing needs the lease, whose owner is always a known registered worker) - claim drops the quarantine and excludes chunks the owner already voted on - domain Task.ReleaseAfterVote; QUORUM_SIZE config (default 2) - unit tests: trusted direct-complete, untrusted needs-quorum, can-claim Reducer and job done/total logic untouched — still one completed task per chunk. --- coordinator/cmd/coordinator/main.go | 17 +-- coordinator/internal/domain/task.go | 31 +++++ coordinator/internal/infra/config.go | 10 ++ coordinator/internal/memstore/memstore.go | 33 +++++ .../internal/storage/postgres/task_repo.go | 5 +- .../storage/postgres/task_result_repo.go | 45 ++++++ .../internal/transport/http/server_test.go | 2 +- coordinator/internal/usecase/ports.go | 9 ++ coordinator/internal/usecase/task.go | 129 ++++++++++++++---- coordinator/internal/usecase/usecase_test.go | 107 +++++++++++---- .../migrations/0013_task_results.down.sql | 5 + .../migrations/0013_task_results.up.sql | 23 ++++ 12 files changed, 356 insertions(+), 60 deletions(-) create mode 100644 coordinator/internal/storage/postgres/task_result_repo.go create mode 100644 coordinator/migrations/0013_task_results.down.sql create mode 100644 coordinator/migrations/0013_task_results.up.sql 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;