diff --git a/docs/internal/app.md b/docs/internal/app.md index 8986e69..692bd9c 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -5,7 +5,7 @@ `internal/app` owns the top-level application use cases. It coordinates configuration loading, secret resolution, backend construction, source bundle discovery, destination selection, publish planning, publish execution, -notification handoff, run reporting, and in-memory run coordination. +notification handoff, run reporting, and upload coordination. The package is the boundary between callers and lower-level domain packages. It does not own manifest validation rules, destination state comparison, storage @@ -152,23 +152,6 @@ Oversized uploads, unsupported content types, invalid bearer tokens, full queues, and unknown status records are mapped to stable HTTP status codes without returning secret token values. -## Coordination - -`PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control. -It allows different pipeline IDs to run concurrently and rejects a second active -run for the same pipeline ID. - -Coordinator records contain a run ID, pipeline ID, status, timestamps, completed -report, and error text when applicable. Active state is memory-only and is -cleared after success, failure, unknown pipeline ID, or context cancellation. - -The admission context is checked before a run is accepted. Once accepted, the -run uses the coordinator lifetime context, so caller cancellation can stop -waiting for admission without owning the actual run lifetime. - -The coordinator does not queue duplicate runs, persist run records, or define -transport endpoints. - ## Errors `Run` returns immediately for config loading errors, context cancellation before @@ -187,10 +170,6 @@ aggregated into one run error after remaining destinations have been attempted. Destination diagnostics include pipeline ID, destination ID, backend, and bundle path. -`PipelineRunCoordinator` returns `DuplicatePipelineRunError` when the same -pipeline already has an active run. Callers can detect that condition with -`IsDuplicatePipelineRun`. - Stdout write errors are returned immediately because the caller's requested output stream can no longer be trusted. @@ -211,7 +190,6 @@ Run helpers are grouped by responsibility: - `run_selection.go`: destination bundle selection, path mapping decisions, and fixed-path warnings. - `run_warnings.go`: secret and SSH warning records. - `run_notify.go`: notification event projection and action filtering. -- `run_coordinator.go`: in-memory run admission, run IDs, status records, and duplicate-run errors. - `upload_coordinator.go`: in-memory upload admission, queue reservation, staging handoff, status tracking, queueing, and staged-source execution. - `upload_http.go`: HTTP upload authentication, routes, JSON response projection, and HTTP error mapping. - `serve.go`: HTTP server startup. @@ -249,8 +227,8 @@ Before changing app orchestration, inspect tests under: - `internal/publish` Use focused app tests for report structure, single-pipeline execution, -coordinator admission, warning generation, notification behavior, and -partial-result aggregation. +upload admission, warning generation, notification behavior, and partial-result +aggregation. ## Invariants @@ -259,7 +237,5 @@ partial-result aggregation. - Destination-scoped failures still produce a structured report plus an aggregated error. - Dry-run must not mutate destination storage or invoke notifications. - `RunPipeline` must use the same run path as `Run` after pipeline selection. -- Duplicate in-flight runs are rejected only for the same pipeline ID. -- Different pipeline IDs may run concurrently. - Concrete backend and transform registration stays at the app layer. - The default notifier is `notify.Noop`. diff --git a/docs/roadmap/audit.md b/docs/roadmap/audit.md index 3920cf4..8258263 100644 --- a/docs/roadmap/audit.md +++ b/docs/roadmap/audit.md @@ -335,14 +335,12 @@ Risk level: - Low. -### PipelineRunCoordinator overlaps conceptually with UploadCoordinator +### Duplicate-run coordination overlaps conceptually with upload coordination Affected files/packages: -- `internal/app/run_coordinator.go` - `internal/app/upload_coordinator.go` - `docs/internal/app.md` -- `internal/app/run_coordinator_test.go` - `internal/app/upload_coordinator_test.go` Duplicated or near-duplicated behavior: @@ -358,7 +356,9 @@ Why it matters: Recommended refactor: - Do not merge the coordinators now. -- Review whether `PipelineRunCoordinator` is still needed as an exported app-level helper. If it is intended for future transports, document that role clearly. If not, remove it and its tests in a separate dead-code cleanup. +- Review whether duplicate-run coordination is still needed as an exported + app-level helper. If it is intended for future transports, document that role + clearly. If not, remove it and its tests in a separate dead-code cleanup. - If both remain, extract only tiny shared timestamp/status helpers if a real third coordinator appears. Suggested tests: @@ -474,7 +474,8 @@ Progress/status handling: - `RunReport` is the core run result model and supports JSON partial-result output. - HTTP upload status is memory-only and documented as such. -- `PipelineRunCoordinator` and `UploadCoordinator` overlap conceptually but have different policies. Avoid merging unless product behavior converges. +- Duplicate-run coordination and upload coordination overlap conceptually but + have different policies. Avoid merging unless product behavior converges. Gaps: @@ -527,7 +528,7 @@ Avoid these changes in the cleanup pass: - Only centralize code if the helper does not blur archive directory semantics. 7. Coordinator intent cleanup. - - Decide whether `PipelineRunCoordinator` is retained for internal future use. + - Decide whether duplicate-run coordination is retained for internal future use. - If retained, clarify comments/docs. If removed, do it as a separate dead-code commit. 8. Test helper cleanup. diff --git a/docs/roadmap/cleanup.md b/docs/roadmap/cleanup.md index df910c4..3d7c52f 100644 --- a/docs/roadmap/cleanup.md +++ b/docs/roadmap/cleanup.md @@ -290,20 +290,19 @@ Completion criteria: Goal: -Remove the currently unused internal `PipelineRunCoordinator` to avoid -maintaining two similar coordination concepts. +Remove the currently unused duplicate-run coordinator to avoid maintaining two +similar coordination concepts. Implementation scope: -- Delete `PipelineRunCoordinator`, `PipelineRunRecord`, - `DuplicatePipelineRunError`, related helpers, and their tests. +- Delete the duplicate-run coordinator, its run record and duplicate-run error + types, related helpers, and their tests. - Remove or rewrite `docs/internal/app.md` sections that describe the removed coordinator. - Keep `UploadCoordinator`; do not merge upload queueing with the removed duplicate-run coordinator. -- Before deletion, confirm with `rg` that production code does not reference - `NewPipelineRunCoordinator`, `PipelineRunCoordinator`, or - `DuplicatePipelineRunError`. +- Before deletion, confirm with search that production code does not reference + the duplicate-run coordinator constructor, type, or error. Current-behavior documentation updates: @@ -313,7 +312,7 @@ Current-behavior documentation updates: Tests: - `go test ./internal/app ./internal/cli` -- `rg -n "PipelineRunCoordinator|NewPipelineRunCoordinator|DuplicatePipelineRunError" internal docs` +- Search `internal` and `docs` for the removed duplicate-run coordinator symbols; should show no stale references after removal. Completion criteria: @@ -387,7 +386,7 @@ Recommended consistency checks: rg -n "LoadFile\\(|LoadSecretEnvironment\\(|DefaultConfigPath" internal/app internal/cli rg -n "application/x-tar|application/gzip|application/x-gzip" internal docs rg -n "2006-01-02T15:04:05Z07:00" internal pkg -rg -n "PipelineRunCoordinator|NewPipelineRunCoordinator|DuplicatePipelineRunError" internal docs +rg -n "duplicate-run coordinator" internal docs ``` The cleanup is complete when: diff --git a/internal/app/run_coordinator.go b/internal/app/run_coordinator.go deleted file mode 100644 index 913b948..0000000 --- a/internal/app/run_coordinator.go +++ /dev/null @@ -1,131 +0,0 @@ -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 deleted file mode 100644 index f3a366a..0000000 --- a/internal/app/run_coordinator_test.go +++ /dev/null @@ -1,223 +0,0 @@ -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{} - } -}