Add pipeline run coordinator

2026-06-03 11:50:19 +00:00
parent 761a2f0bc2
commit 7b2caf4c01
2 changed files with 354 additions and 0 deletions
--- a/internal/app/run_coordinator.go
+++ b/internal/app/run_coordinator.go
@@ -0,0 +1,131 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+)
+
+type PipelineRunID string
+
+type PipelineRunStatus string
+
+const (
+	PipelineRunRunning   PipelineRunStatus = "running"
+	PipelineRunSucceeded PipelineRunStatus = "succeeded"
+	PipelineRunFailed    PipelineRunStatus = "failed"
+)
+
+type PipelineRunRecord struct {
+	ID         PipelineRunID     `json:"id"`
+	PipelineID string            `json:"pipeline_id"`
+	Status     PipelineRunStatus `json:"status"`
+	StartedAt  time.Time         `json:"started_at"`
+	FinishedAt *time.Time        `json:"finished_at,omitempty"`
+	Report     RunReport         `json:"report,omitempty"`
+	Error      string            `json:"error,omitempty"`
+}
+
+type DuplicatePipelineRunError struct {
+	PipelineID string
+	RunID      PipelineRunID
+}
+
+func (err DuplicatePipelineRunError) Error() string {
+	if err.RunID == "" {
+		return fmt.Sprintf("pipeline %q already has an active run", err.PipelineID)
+	}
+	return fmt.Sprintf("pipeline %q already has active run %s", err.PipelineID, err.RunID)
+}
+
+func IsDuplicatePipelineRun(err error) bool {
+	var duplicate DuplicatePipelineRunError
+	return errors.As(err, &duplicate)
+}
+
+type PipelineRunCoordinator struct {
+	ctx    context.Context
+	run    pipelineRunFunc
+	now    func() time.Time
+	mu     sync.Mutex
+	nextID uint64
+	active map[string]PipelineRunRecord
+}
+
+type pipelineRunFunc func(context.Context, RunPipelineOptions) (RunReport, error)
+
+func NewPipelineRunCoordinator(ctx context.Context) *PipelineRunCoordinator {
+	return newPipelineRunCoordinator(ctx, RunPipeline)
+}
+
+func newPipelineRunCoordinator(ctx context.Context, run pipelineRunFunc) *PipelineRunCoordinator {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return &PipelineRunCoordinator{
+		ctx:    ctx,
+		run:    run,
+		now:    time.Now,
+		active: map[string]PipelineRunRecord{},
+	}
+}
+
+func (coordinator *PipelineRunCoordinator) RunPipeline(ctx context.Context, options RunPipelineOptions) (PipelineRunRecord, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return PipelineRunRecord{}, err
+	}
+	record, err := coordinator.admit(options.PipelineID)
+	if err != nil {
+		return PipelineRunRecord{}, err
+	}
+	defer coordinator.clear(options.PipelineID)
+
+	report, runErr := coordinator.run(coordinator.ctx, options)
+	record.Report = report
+	finishedAt := coordinator.now().UTC()
+	record.FinishedAt = &finishedAt
+	if runErr != nil {
+		record.Status = PipelineRunFailed
+		record.Error = runErr.Error()
+		return record, runErr
+	}
+	record.Status = PipelineRunSucceeded
+	return record, nil
+}
+
+func (coordinator *PipelineRunCoordinator) admit(pipelineID string) (PipelineRunRecord, error) {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	if active, ok := coordinator.active[pipelineID]; ok {
+		return PipelineRunRecord{}, DuplicatePipelineRunError{
+			PipelineID: pipelineID,
+			RunID:      active.ID,
+		}
+	}
+	coordinator.nextID++
+	record := PipelineRunRecord{
+		ID:         PipelineRunID(fmt.Sprintf("run-%016d", coordinator.nextID)),
+		PipelineID: pipelineID,
+		Status:     PipelineRunRunning,
+		StartedAt:  coordinator.now().UTC(),
+	}
+	coordinator.active[pipelineID] = record
+	return record, nil
+}
+
+func (coordinator *PipelineRunCoordinator) clear(pipelineID string) {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	delete(coordinator.active, pipelineID)
+}
+
+func (coordinator *PipelineRunCoordinator) activeCount() int {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	return len(coordinator.active)
+}
--- a/internal/app/run_coordinator_test.go
+++ b/internal/app/run_coordinator_test.go
@@ -0,0 +1,223 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestPipelineRunCoordinatorRejectsDuplicateActiveRun(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var startedOnce sync.Once
+	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		startedOnce.Do(func() {
+			close(started)
+		})
+		<-release
+		return RunReport{}, nil
+	})
+	firstResult := make(chan runCoordinatorTestResult, 1)
+
+	go func() {
+		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+		firstResult <- runCoordinatorTestResult{record: record, err: err}
+	}()
+	waitForSignal(t, started, "first run to start")
+
+	_, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if err == nil || !IsDuplicatePipelineRun(err) {
+		t.Fatalf("RunPipeline() error = %v, want duplicate active run", err)
+	}
+	close(release)
+	result := waitForRunResult(t, firstResult)
+	if result.err != nil {
+		t.Fatalf("first RunPipeline() error = %v", result.err)
+	}
+	if result.record.Status != PipelineRunSucceeded || result.record.ID == "" || result.record.FinishedAt == nil {
+		t.Fatalf("first record = %#v, want succeeded completed record", result.record)
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+func TestPipelineRunCoordinatorAllowsDifferentActivePipelines(t *testing.T) {
+	started := make(chan string, 2)
+	release := make(chan struct{})
+	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		started <- options.PipelineID
+		<-release
+		return RunReport{}, nil
+	})
+	firstResult := make(chan runCoordinatorTestResult, 1)
+	secondResult := make(chan runCoordinatorTestResult, 1)
+
+	go func() {
+		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports-one"})
+		firstResult <- runCoordinatorTestResult{record: record, err: err}
+	}()
+	go func() {
+		record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports-two"})
+		secondResult <- runCoordinatorTestResult{record: record, err: err}
+	}()
+	startedPipelines := map[string]bool{
+		waitForPipelineID(t, started): true,
+		waitForPipelineID(t, started): true,
+	}
+	if !startedPipelines["reports-one"] || !startedPipelines["reports-two"] {
+		t.Fatalf("started pipelines = %#v, want both requested pipelines", startedPipelines)
+	}
+	if got := coordinator.activeCount(); got != 2 {
+		t.Fatalf("active count = %d, want 2", got)
+	}
+
+	close(release)
+	first := waitForRunResult(t, firstResult)
+	second := waitForRunResult(t, secondResult)
+	if first.err != nil || second.err != nil {
+		t.Fatalf("RunPipeline() errors = %v, %v; want nil", first.err, second.err)
+	}
+	if first.record.ID == second.record.ID {
+		t.Fatalf("run IDs matched: %q", first.record.ID)
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+func TestPipelineRunCoordinatorClearsActiveRunAfterSuccess(t *testing.T) {
+	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		return RunReport{DryRun: options.DryRun}, nil
+	})
+
+	first, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports", DryRun: true})
+	if err != nil {
+		t.Fatalf("first RunPipeline() error = %v", err)
+	}
+	second, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if err != nil {
+		t.Fatalf("second RunPipeline() error = %v", err)
+	}
+	if first.Status != PipelineRunSucceeded || second.Status != PipelineRunSucceeded {
+		t.Fatalf("statuses = %s, %s; want succeeded", first.Status, second.Status)
+	}
+	if !first.Report.DryRun {
+		t.Fatalf("first report dry_run = false, want true")
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+func TestPipelineRunCoordinatorClearsActiveRunAfterFailure(t *testing.T) {
+	runError := errors.New("run failed")
+	attempt := 0
+	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		attempt++
+		if attempt == 1 {
+			return RunReport{}, runError
+		}
+		return RunReport{}, nil
+	})
+
+	first, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if !errors.Is(err, runError) {
+		t.Fatalf("first RunPipeline() error = %v, want run failure", err)
+	}
+	if first.Status != PipelineRunFailed || first.Error != runError.Error() || first.FinishedAt == nil {
+		t.Fatalf("first record = %#v, want failed completed record", first)
+	}
+	second, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if err != nil {
+		t.Fatalf("second RunPipeline() error = %v", err)
+	}
+	if second.Status != PipelineRunSucceeded {
+		t.Fatalf("second status = %s, want succeeded", second.Status)
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+func TestPipelineRunCoordinatorClearsActiveRunAfterCancellation(t *testing.T) {
+	runContext, cancel := context.WithCancel(context.Background())
+	coordinator := newPipelineRunCoordinator(runContext, func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		return RunReport{}, ctx.Err()
+	})
+	cancel()
+
+	record, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("RunPipeline() error = %v, want context canceled", err)
+	}
+	if record.Status != PipelineRunFailed || record.Error != context.Canceled.Error() {
+		t.Fatalf("record = %#v, want failed cancellation record", record)
+	}
+	_, err = coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "reports"})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("second RunPipeline() error = %v, want context canceled", err)
+	}
+	if IsDuplicatePipelineRun(err) {
+		t.Fatalf("second RunPipeline() error = %v, want cancellation instead of duplicate", err)
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+func TestPipelineRunCoordinatorUnknownPipelineDoesNotRemainActive(t *testing.T) {
+	coordinator := newPipelineRunCoordinator(context.Background(), func(ctx context.Context, options RunPipelineOptions) (RunReport, error) {
+		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
+	})
+
+	_, err := coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "missing"})
+	if err == nil || !IsPipelineNotFound(err) {
+		t.Fatalf("RunPipeline() error = %v, want pipeline not found", err)
+	}
+	_, err = coordinator.RunPipeline(context.Background(), RunPipelineOptions{PipelineID: "missing"})
+	if err == nil || !IsPipelineNotFound(err) || IsDuplicatePipelineRun(err) {
+		t.Fatalf("second RunPipeline() error = %v, want pipeline not found without duplicate", err)
+	}
+	if got := coordinator.activeCount(); got != 0 {
+		t.Fatalf("active count = %d, want 0", got)
+	}
+}
+
+type runCoordinatorTestResult struct {
+	record PipelineRunRecord
+	err    error
+}
+
+func waitForSignal(t *testing.T, signal <-chan struct{}, name string) {
+	t.Helper()
+	select {
+	case <-signal:
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for %s", name)
+	}
+}
+
+func waitForPipelineID(t *testing.T, pipelineIDs <-chan string) string {
+	t.Helper()
+	select {
+	case pipelineID := <-pipelineIDs:
+		return pipelineID
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for pipeline start")
+		return ""
+	}
+}
+
+func waitForRunResult(t *testing.T, results <-chan runCoordinatorTestResult) runCoordinatorTestResult {
+	t.Helper()
+	select {
+	case result := <-results:
+		return result
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for run result")
+		return runCoordinatorTestResult{}
+	}
+}