diff --git a/internal/app/run_coordinator.go b/internal/app/run_coordinator.go new file mode 100644 index 0000000..913b948 --- /dev/null +++ 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) +} diff --git a/internal/app/run_coordinator_test.go b/internal/app/run_coordinator_test.go new file mode 100644 index 0000000..f3a366a --- /dev/null +++ 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{} + } +}