diff --git a/backend/internal/worker/ci_worker.go b/backend/internal/worker/ci_worker.go index b21265c..ace889f 100644 --- a/backend/internal/worker/ci_worker.go +++ b/backend/internal/worker/ci_worker.go @@ -48,6 +48,10 @@ const ( ciStepTimeout = 10 * time.Minute + // maxParallelJobsPerRun bounds how many jobs within a single run execute + // steps concurrently. Independent jobs run in parallel up to this cap. + maxParallelJobsPerRun = 8 + planTierFree = "free" planTierPro = "pro" @@ -347,114 +351,186 @@ func (w *CIWorker) HandleCIRun(ctx context.Context, task *asynq.Task) error { return nil } + // Independent jobs run concurrently; a job starts only once every job in its + // `needs` has finished. Shared run state (the maps, the aggregate log + // buffer, the overall conclusion) is guarded by mu; the actual step + // execution happens outside the lock so jobs run in parallel. A bounded + // semaphore caps how many jobs execute at once. + var mu sync.Mutex + var fatalErr error + var fatalOnce sync.Once + runCtx, cancelRun := context.WithCancel(ctx) + defer cancelRun() + setFatal := func(err error) { + fatalOnce.Do(func() { + fatalErr = err + cancelRun() + }) + } + + done := make(map[string]chan struct{}, len(order)) + for _, name := range order { + done[name] = make(chan struct{}) + } + sem := make(chan struct{}, maxParallelJobsPerRun) + var wg sync.WaitGroup + for _, jobName := range order { - job := ir.Jobs[jobName] - - // A job whose dependencies did not all succeed is skipped, like GitHub - // Actions. The dependency's own failure already set the run conclusion, - // so a skip adds no new failure — but independent jobs still run. - if dep, unmet := firstUnsatisfiedNeed(job.Needs, jobResults); unmet { - if err := skipJob(jobName, fmt.Sprintf("dependency %q did not succeed", dep)); err != nil { - return err + wg.Add(1) + go func(jobName string) { + defer wg.Done() + defer close(done[jobName]) + job := ir.Jobs[jobName] + + // Wait for every dependency to finish before starting. + for _, need := range job.Needs { + if ch, ok := done[need]; ok { + select { + case <-ch: + case <-runCtx.Done(): + return + } + } + } + if runCtx.Err() != nil { + return } - continue - } - // A false job-level `if:` skips the whole job. Evaluate it against the - // github context and the workflow+job env. - if job.If != "" { - jobEnv := mergeStringMaps(ir.Env, job.Env) - ec := &workflow.EvalContext{Contexts: map[string]map[string]string{ - "github": githubCtx, "runner": runnerCtx, "secrets": secretsCtx, "env": jobEnv, "matrix": {}, - }} - run, ifErr := workflow.EvaluateCondition(job.If, ec) - if ifErr != nil { - if err := skipJob(jobName, "invalid job if: "+ifErr.Error()); err != nil { - return err + // Skip if a dependency did not succeed (GitHub semantics). + mu.Lock() + dep, unmet := firstUnsatisfiedNeed(job.Needs, jobResults) + mu.Unlock() + if unmet { + mu.Lock() + err := skipJob(jobName, fmt.Sprintf("dependency %q did not succeed", dep)) + mu.Unlock() + if err != nil { + setFatal(err) } - continue + return } - if !run { - if err := skipJob(jobName, fmt.Sprintf("job if condition %q evaluated false", job.If)); err != nil { - return err + + // Skip on a false job-level `if:` (evaluated against github + env). + if job.If != "" { + jobEnv := mergeStringMaps(ir.Env, job.Env) + ec := &workflow.EvalContext{Contexts: map[string]map[string]string{ + "github": githubCtx, "runner": runnerCtx, "secrets": secretsCtx, "env": jobEnv, "matrix": {}, + }} + run, ifErr := workflow.EvaluateCondition(job.If, ec) + reason := "" + if ifErr != nil { + reason = "invalid job if: " + ifErr.Error() + } else if !run { + reason = fmt.Sprintf("job if condition %q evaluated false", job.If) + } + if reason != "" { + mu.Lock() + err := skipJob(jobName, reason) + mu.Unlock() + if err != nil { + setFatal(err) + } + return } - continue } - } - // Expand the matrix into one instance per combination; a job without a - // matrix runs as a single instance. The logical job succeeds only if - // every instance succeeds (so dependents see one aggregated result). - combos := job.MatrixExpansion - if len(combos) == 0 { - combos = []map[string]any{nil} - } - logicalFailed := false - - for _, combo := range combos { - instanceName := jobName - matrixCtx := map[string]string{} - if combo != nil { - matrixCtx = stringifyMatrixCombo(combo) - instanceName = fmt.Sprintf("%s (%s)", jobName, matrixLabel(combo)) + // Bound the number of jobs executing steps at once. + select { + case sem <- struct{}{}: + case <-runCtx.Done(): + return } + defer func() { <-sem }() + + // Expand the matrix into one instance per combination (run + // sequentially within the job); a job without a matrix runs once. + // The logical job succeeds only if every instance succeeds. + combos := job.MatrixExpansion + if len(combos) == 0 { + combos = []map[string]any{nil} + } + logicalFailed := false - var jobUUID uuid.UUID - if w.jobRepo != nil { - jobUUID = int64CompatibleUUID() - jobIDs[instanceName] = jobUUID.String() - now := time.Now().UTC() - if createErr := w.jobRepo.Create(ctx, &entity.WorkflowJob{ - ID: jobUUID, WorkflowRunID: &runUUID, OrganizationID: orgUUID, RepositoryID: repoUUID, - Name: instanceName, Status: entity.WorkflowJobStatusInProgress, StartedAt: &now, CreatedAt: now, - }); createErr != nil { - return fmt.Errorf("create workflow job: %w", createErr) + for _, combo := range combos { + if runCtx.Err() != nil { + return + } + instanceName := jobName + matrixCtx := map[string]string{} + if combo != nil { + matrixCtx = stringifyMatrixCombo(combo) + instanceName = fmt.Sprintf("%s (%s)", jobName, matrixLabel(combo)) } - } - jobIDStr := instanceName - if w.jobRepo != nil { - jobIDStr = jobUUID.String() - } - jobLogStatus[instanceName] = jobLogStatusSuccess - instanceFailed := w.runJobSteps(ctx, runJobSpec{ - payload: payload, - wfEnv: ir.Env, - job: job, - instanceName: instanceName, - jobIDStr: jobIDStr, - matrixCtx: matrixCtx, - githubCtx: githubCtx, - runnerCtx: runnerCtx, - secretsCtx: secretsCtx, - secretEnv: secretEnv, - secretValues: secretValues, - workdir: workdir, - logBuf: logBuf, - useStreaming: useStreaming, - streamRunner: streamRunner, - jobLineCounts: jobLineCounts, - }) + var jobUUID uuid.UUID + jobIDStr := instanceName + if w.jobRepo != nil { + jobUUID = int64CompatibleUUID() + jobIDStr = jobUUID.String() + now := time.Now().UTC() + if createErr := w.jobRepo.Create(runCtx, &entity.WorkflowJob{ + ID: jobUUID, WorkflowRunID: &runUUID, OrganizationID: orgUUID, RepositoryID: repoUUID, + Name: instanceName, Status: entity.WorkflowJobStatusInProgress, StartedAt: &now, CreatedAt: now, + }); createErr != nil { + setFatal(fmt.Errorf("create workflow job: %w", createErr)) + return + } + } - if instanceFailed { - logicalFailed = true - jobLogStatus[instanceName] = jobLogStatusFailure - } - if w.jobRepo != nil && jobUUID != uuid.Nil { - jobConclusion := entity.WorkflowJobConclusionSuccess + acc := &jobAccumulator{} + instanceFailed := w.runJobSteps(runCtx, runJobSpec{ + payload: payload, + wfEnv: ir.Env, + job: job, + instanceName: instanceName, + jobIDStr: jobIDStr, + matrixCtx: matrixCtx, + githubCtx: githubCtx, + runnerCtx: runnerCtx, + secretsCtx: secretsCtx, + secretEnv: secretEnv, + secretValues: secretValues, + workdir: workdir, + useStreaming: useStreaming, + streamRunner: streamRunner, + acc: acc, + }) + + if w.jobRepo != nil && jobUUID != uuid.Nil { + jobConclusion := entity.WorkflowJobConclusionSuccess + if instanceFailed { + jobConclusion = entity.WorkflowJobConclusionFailure + } + _ = w.jobRepo.Complete(runCtx, jobUUID, jobConclusion, time.Now().UTC()) + } + + status := jobLogStatusSuccess if instanceFailed { - jobConclusion = entity.WorkflowJobConclusionFailure + status = jobLogStatusFailure + logicalFailed = true } - _ = w.jobRepo.Complete(ctx, jobUUID, jobConclusion, time.Now().UTC()) + mu.Lock() + jobIDs[instanceName] = jobIDStr + jobLogStatus[instanceName] = status + jobLineCounts[instanceName] = acc.lineCount + logBuf.WriteString(acc.buf.String()) + mu.Unlock() } - } - if logicalFailed { - jobResults[jobName] = jobResultFailure - conclusion = ciConclusionFailure - } else { - jobResults[jobName] = jobResultSuccess - } + mu.Lock() + if logicalFailed { + jobResults[jobName] = jobResultFailure + conclusion = ciConclusionFailure + } else { + jobResults[jobName] = jobResultSuccess + } + mu.Unlock() + }(jobName) + } + + wg.Wait() + if fatalErr != nil { + return fatalErr } finalStatus := ciStatusCompleted @@ -482,24 +558,31 @@ func (w *CIWorker) HandleCIRun(ctx context.Context, task *asynq.Task) error { return nil } +// jobAccumulator holds a single job instance's private log buffer and line +// counter. Each instance owns its own accumulator so instances can run +// concurrently without sharing mutable state. +type jobAccumulator struct { + buf strings.Builder + lineCount int64 +} + // runJobSpec bundles the per-run state a single job instance needs to execute. type runJobSpec struct { - payload CIRunPayload - wfEnv map[string]string - job workflow.IRJob - instanceName string - jobIDStr string - matrixCtx map[string]string - githubCtx map[string]string - runnerCtx map[string]string - secretsCtx map[string]string - secretEnv []string - secretValues []string - workdir string - logBuf *strings.Builder - useStreaming bool - streamRunner StreamingCommandRunner - jobLineCounts map[string]int64 + payload CIRunPayload + wfEnv map[string]string + job workflow.IRJob + instanceName string + jobIDStr string + matrixCtx map[string]string + githubCtx map[string]string + runnerCtx map[string]string + secretsCtx map[string]string + secretEnv []string + secretValues []string + workdir string + useStreaming bool + streamRunner StreamingCommandRunner + acc *jobAccumulator } // runJobSteps executes one job instance's steps, evaluating expressions against @@ -545,7 +628,7 @@ func (w *CIWorker) runJobSteps(ctx context.Context, s runJobSpec) bool { run, ifErr := workflow.EvaluateCondition(step.If, evalCtx) if ifErr != nil { jobFailed = true - fmt.Fprintf(s.logBuf, "[job=%s step=%d] if error: %s\n", s.instanceName, i, ifErr.Error()) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d] if error: %s\n", s.instanceName, i, ifErr.Error()) continue } if !run { @@ -571,10 +654,10 @@ func (w *CIWorker) runJobSteps(ctx context.Context, s runJobSpec) bool { if s.useStreaming { runErr := s.streamRunner(stepCtx, instWorkdir, stepEnv, script, i, func(stream, line string) { - s.jobLineCounts[s.instanceName]++ - lineNum := s.jobLineCounts[s.instanceName] + s.acc.lineCount++ + lineNum := s.acc.lineCount masked := maskSecrets(line, s.secretValues) - fmt.Fprintf(s.logBuf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, i, step.Name, masked) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, i, step.Name, masked) logLine := &entity.JobLogLine{ OrganizationID: s.payload.OrganizationID, @@ -598,18 +681,18 @@ func (w *CIWorker) runJobSteps(ctx context.Context, s runJobSpec) bool { if runErr != nil { jobFailed = true - fmt.Fprintf(s.logBuf, "[job=%s step=%d] error: %s\n", s.instanceName, i, maskSecrets(runErr.Error(), s.secretValues)) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d] error: %s\n", s.instanceName, i, maskSecrets(runErr.Error(), s.secretValues)) } } else { out, runErr := w.runStep(stepCtx, instWorkdir, stepEnv, script) cancel() masked := maskSecrets(string(out), s.secretValues) - fmt.Fprintf(s.logBuf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, i, step.Name, masked) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, i, step.Name, masked) if runErr != nil { jobFailed = true - fmt.Fprintf(s.logBuf, "[job=%s step=%d] error: %s\n", s.instanceName, i, maskSecrets(runErr.Error(), s.secretValues)) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d] error: %s\n", s.instanceName, i, maskSecrets(runErr.Error(), s.secretValues)) } } } @@ -642,18 +725,18 @@ func (w *CIWorker) runUsesStep(ctx context.Context, s runJobSpec, step workflow. // shows checkout/unsupported-action notes, mirroring how run-step output is // streamed. Falls back to the buffered log when no log repository is wired. func (w *CIWorker) emitUsesLine(ctx context.Context, s runJobSpec, stepIdx int, text string) { - fmt.Fprintf(s.logBuf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, stepIdx, "", text) + fmt.Fprintf(&s.acc.buf, "[job=%s step=%d name=%s]\n%s\n", s.instanceName, stepIdx, "", text) if w.logRepo == nil { return } - s.jobLineCounts[s.instanceName]++ + s.acc.lineCount++ line := &entity.JobLogLine{ OrganizationID: s.payload.OrganizationID, RepositoryID: s.payload.RepositoryID, RunID: s.payload.WorkflowRunID, JobID: s.jobIDStr, StepIndex: stepIdx, - LineNumber: s.jobLineCounts[s.instanceName], + LineNumber: s.acc.lineCount, Stream: entity.LogStreamStdout, Text: text, CreatedAt: time.Now().UTC(), diff --git a/backend/internal/worker/ci_worker_test.go b/backend/internal/worker/ci_worker_test.go index 2b13aac..9c67196 100644 --- a/backend/internal/worker/ci_worker_test.go +++ b/backend/internal/worker/ci_worker_test.go @@ -8,7 +8,9 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" + "time" "github.com/hibiken/asynq" _ "github.com/mattn/go-sqlite3" @@ -398,6 +400,58 @@ jobs: } } +// TestCIIndependentJobsRunConcurrently proves independent jobs execute in +// parallel: each job's step blocks until it observes that the other job has +// also started. If execution were sequential the first job would wait forever +// (until its 2s barrier times out) and fail the run. +func TestCIIndependentJobsRunConcurrently(t *testing.T) { + db := newCITestDB(t) + ctx := context.Background() + orgID, repoID, runID := "org-par", "repo-par", "run-par" + mustExec(t, db, `INSERT INTO organizations (id, login, name, plan_tier) VALUES (?, ?, ?, ?)`, orgID, "acme", "Acme", planTierPro) + mustExec(t, db, `INSERT INTO repositories (id, organization_id, name) VALUES (?, ?, ?)`, repoID, orgID, "widgets") + mustExec(t, db, `INSERT INTO workflow_runs (id, organization_id, repository_id, workflow, status) VALUES (?, ?, ?, ?, ?)`, runID, orgID, repoID, "ci.yml", ciStatusQueued) + + var started int32 + release := make(chan struct{}) + var releaseOnce sync.Once + worker := NewCIWorker(db).WithCommandRunner(func(_ context.Context, _ string, _ []string, _ string) ([]byte, error) { + if atomic.AddInt32(&started, 1) == 2 { + releaseOnce.Do(func() { close(release) }) + } + select { + case <-release: + return []byte("ok\n"), nil + case <-time.After(2 * time.Second): + return nil, errors.New("timeout: job did not run concurrently with its sibling") + } + }) + + payload, _ := json.Marshal(CIRunPayload{ + WorkflowRunID: runID, RepositoryID: repoID, OrganizationID: orgID, + WorkflowYAML: []byte(`name: CI +on: push +jobs: + a: + steps: + - run: echo A + b: + steps: + - run: echo B +`), + }) + if err := worker.HandleCIRun(ctx, asynq.NewTask(TypeCIRun, payload)); err != nil { + t.Fatalf("HandleCIRun: %v", err) + } + var conclusion sql.NullString + if err := db.QueryRowContext(ctx, `SELECT conclusion FROM workflow_runs WHERE id = ?`, runID).Scan(&conclusion); err != nil { + t.Fatalf("query conclusion: %v", err) + } + if got := strings.SplitN(conclusion.String, "\n", 2)[0]; got != ciConclusionSuccess { + t.Errorf("run conclusion = %q, want success — jobs did not run concurrently", got) + } +} + // runWorkflowCapture runs a workflow and returns the scripts the runner // actually received (post-interpolation), letting a caller adjust the payload. func runWorkflowCapture(t *testing.T, yamlSrc string, mutate func(*CIRunPayload)) []string {