eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Remove unused pipeline run coordinator

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-04 00:43:19 +00:00
parent 2ac2bbdf79
commit 982e7e9863
5 changed files with 18 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
docs/internal/app.md
View File

@@ -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`.

13
docs/roadmap/audit.md
View File

@@ -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.

17
docs/roadmap/cleanup.md
View File

@@ -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:

131
internal/app/run_coordinator.go
View File

@@ -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)
}

223
internal/app/run_coordinator_test.go
View File

@@ -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{}
}
}