go-job-pipeline

name	go-job-pipeline
description	The Highway pattern - a resumable multi-job pipeline for Go CLI tools, with worker lanes, a UI-agnostic progress channel, and Ctrl+C resume. Use when a CLI processes many items through one pipeline (downloads, scans, migrations), needs configurable worker counts, needs progress that can drive a terminal or a web UI, or needs to pick up where an interrupt left off. Triggers on internal/highway/, internal/display/, internal/jobs/, a Job interface, a Progress channel, RegisterType, LoadState, and a resume command.
user-invocable	false

Go Job Pipeline

Many items, N worker lanes, one progress channel, and a state file that survives Ctrl+C.

This applies to CLI Only projects and the command surface of a CLI + Web hybrid, since it assumes the utils package exists.

Reach for it when a tool runs many items through a shared pipeline that needs progress reporting and resumability. One-shot work with no resume and no progress uses plain goroutine primitives, because the Job interface and the state file are pure overhead when nothing is going to be resumed.

CLI entry (cmd/download.go, cmd/scan.go)
   ↓  parse flags into []Job
Highway: N worker lanes pull from the queue, run job.Run(ctx, progress), track completion
   ↓                                    ↓
Progress channel (any UI)      State file (.toolname-resume-state.json)

Layout

cmd/
├── download.go     # builds download jobs, submits them
├── scan.go         # builds scan jobs, submits them
└── resume.go       # loads the state file, submits what is pending

internal/
├── highway/        # the engine: highway.go, state.go, progress.go
├── display/        # terminal UI consuming the progress channel
└── jobs/           # concrete job types

Design Decisions

Decision	Reason
A job writes its own output	the engine never learns what a job produces, so a new job type needs no engine change
A job marshals its own state	only the job knows which fields matter for resume
Progress is data with no behavior	any consumer can render it, including a websocket feeding a browser
The engine knows only the Job interface	adding a job type is one file, not a switch statement in the engine
The state file sits in the working directory	it is discoverable next to the output it accompanies
A failed job is marked done	resume skips it rather than retrying automatically, since a retry loop on a permanent failure never terminates

The Job Interface

type Job interface {
    ID() string    // unique, used for tracking and for skipping on resume
    Type() string  // resolves the unmarshaler on resume
    Run(ctx context.Context, progress chan<- Progress) error
    Marshal() ([]byte, error)
}

type JobUnmarshaler func(data []byte) (Job, error)

A Run does its work, emits Progress as it goes, and finishes by sending Progress{JobID: j.ID(), Done: true} or returning an error.

A job that carries partial progress records it in its own fields and skips what is already finished, which is what makes resume mean something more than restart:

type HTTPDownloadJob struct {
    URL            string `json:"url"`
    OutputDir      string `json:"outputDir"`
    TotalSize      int64  `json:"totalSize"`      // discovered on the first run
    CompletedParts []int  `json:"completedParts"` // the resume state
}

func (j *HTTPDownloadJob) Run(ctx context.Context, progress chan<- highway.Progress) error {
    totalParts := int((j.TotalSize + partSize - 1) / partSize)
    for part := range totalParts {
        if slices.Contains(j.CompletedParts, part) {
            continue
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        if err := j.downloadPart(ctx, part); err != nil {
            return fmt.Errorf("download part %d: %w", part, err)
        }
        j.CompletedParts = append(j.CompletedParts, part) // only after the part lands
        progress <- highway.Progress{
            JobID:   j.ID(),
            Message: fmt.Sprintf("Part %d/%d", len(j.CompletedParts), totalParts),
            Current: int64(len(j.CompletedParts)),
            Total:   int64(totalParts),
        }
    }
    progress <- highway.Progress{JobID: j.ID(), Done: true}
    return nil
}

  Marshal() ([], ) {  json.Marshal(j) }

 (highway.Job, ) {
     job HTTPDownloadJob
     err := json.Unmarshal(data, &job); err !=  {
         , err
    }
     &job, 
}

A part is appended to CompletedParts only after it succeeds, because recording it first means a crash mid-part loses those bytes forever.

Progress

Two kinds of update, distinguished so a renderer knows whether it can draw a bar.

Field	Purpose
JobID	which job sent it
Type	ProgressTypeProgress when the total is known, ProgressTypeSubStatus when it is not
Message	short status shown next to the job ID
SubStatus	detailed line, for the SubStatus kind
Current / Total	numerator and denominator
Extra	trailing detail after the percentage, such as 125MB/1GB
Done	the job finished
Error	the job failed

The Engine

package highway

type Highway struct {
    workers      int
    statePath    string
    unmarshalers map[string]JobUnmarshaler

    mu        sync.Mutex
    pending   []Job
    completed map[string]bool
    progress  chan Progress
}

func New(workers int, statePath string) *Highway {
    return &Highway{
        workers:      max(workers, 1),
        statePath:    statePath,
        unmarshalers: make(map[string]JobUnmarshaler),
        completed:    make(map[string]bool),
        progress:     make(chan Progress, 100),
    }
}

func (h *Highway) RegisterType(jobType string, unmarshal JobUnmarshaler) {
    h.unmarshalers[jobType] = unmarshal
}

func (h *Highway) Submit(jobs ...Job) {
    h.mu.Lock()
    defer h.mu.Unlock()
    h.pending = append(h.pending, jobs...)
}

func (h *Highway) Progress() <-chan Progress { return h.progress }

func (h *Highway) Run(ctx context.Context) error {
    jobCh := make(chan Job)
     wg sync.WaitGroup

      h.workers {
        wg.Go( {
             job :=  jobCh {
                h.executeJob(ctx, job)
            }
        })
    }

      {
         (jobCh)
        h.mu.Lock()
        jobs := slices.Clone(h.pending)
        h.mu.Unlock()

         _, job :=  jobs {
             h.isCompleted(job.ID()) {
                
            }
             {
             <-ctx.Done():
                
             jobCh <- job:
            }
        }
    }()

    done := ( {})
      {
        wg.Wait()
        (done)
    }()

     {
     <-done:
        (h.progress)
        os.Remove(h.statePath)
         
     <-ctx.Done():
        wg.Wait()
        (h.progress)
         h.saveState()
    }
}

 executeJob(ctx context.Context, job Job) {
     err := job.Run(ctx, h.progress); err !=  {
        h.progress <- Progress{JobID: job.ID(), Done: , Error: err, ErrMsg: err.Error()}
    }
    h.markCompleted(job.ID()) 
}

The progress channel is buffered so a job emitting updates faster than the display consumes them keeps working instead of blocking on a render.

The feeder goroutine clones the pending slice under the lock and then releases it, since holding a mutex while sending on a channel deadlocks against anything that submits during the run.

The state file is deleted on clean completion, because a leftover file makes the next run look like a resume.

State

{
  "completed": ["job-1", "job-2"],
  "pending": [
    {
      "id": "http-bigfile.zip",
      "type": "http-download",
      "data": { "url": "https://example.com/bigfile.zip", "completedParts": [0, 1, 2] }
    }
  ]
}

func (h *Highway) saveState() error {
    h.mu.Lock()
    defer h.mu.Unlock()

    var pendingJobs []persistedJob
    for _, job := range h.pending {
        if h.completed[job.ID()] {
            continue
        }
        data, err := job.Marshal()
        if err != nil {
            continue
        }
        pendingJobs = append(pendingJobs, persistedJob{ID: job.ID(), Type: job.Type(), Data: data})
    }

    state := persistedState{
        Completed: slices.Collect(maps.Keys(h.completed)),
        Pending:   pendingJobs,
    }
    data, err := json.Marshal(state, jsontext.WithIndent("  "))
    if err != nil {
        return err
    }
    if err := os.WriteFile(h.statePath, data, 0600); err != nil {
        return err
    }
    utils.PrintInfo("State saved to " + h.statePath)
    return nil
}

func (h *Highway) LoadState() error {
    data, err := os.ReadFile(h.statePath)
    if err != nil {
        return err
    }
    var state persistedState
    if err := json.Unmarshal(data, &state); err != nil {
        return err
    }

    h.mu.Lock()
    for _, id := range state.Completed {
        h.completed[id] = true
    }
    h.mu.Unlock()

    for _, pj := range state.Pending {
        unmarshal, ok := h.unmarshalers[pj.Type]
         !ok {
             fmt.Errorf(, pj.Type)
        }
        job, err := unmarshal(pj.Data)
         err !=  {
             fmt.Errorf(, pj.ID, err)
        }
        h.Submit(job)
    }
     
}

 persistedState  {
    Completed []       
    Pending   []persistedJob 
}

 persistedJob  {
    ID             
    Type           
    Data jsontext.Value 
}

The state file is written at 0600, matching every other file these projects write into a user's directory, because a job payload commonly carries paths, URLs, and identifiers from their environment.

The save reports through utils.PrintInfo rather than fmt.Printf, so the line obeys the same three tiers as everything else the tool prints.

Wiring a Command

func runDownload(cmd *cobra.Command, args []string) error {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
    defer cancel()

    hw := highway.New(workers, ".downloader-state.json")
    hw.RegisterType("http-download", jobs.UnmarshalHTTPDownload)

    disp := display.New(display.DefaultConfig())
    for _, url := range urls {
        job := jobs.NewHTTPDownload(url, outputDir)
        disp.RegisterJob(job.ID())
        hw.Submit(job)
    }

    disp.Start(hw.Progress())
    err := hw.Run(ctx)
    disp.Stop()
    return err
}

signal.NotifyContext turns Ctrl+C into a context cancellation, which is what lets the engine save state instead of the process dying mid-write.

A resume command is the same function with hw.LoadState() in place of the submit loop. Every job type is registered before LoadState runs, since the unmarshaler is what turns stored bytes back into a job.

Display Contract

The display consumes the progress channel and owns the terminal. It lives in its own internal/display/ package so nothing about the engine depends on how progress is rendered, and so the same channel can feed a websocket for a web UI.

• The refresh loop redraws on a ticker, 200ms by default. Redrawing on every update makes the terminal flicker under a fast job.
• A running job renders as two lines: the job line, then either a progress bar or a substatus line, never both. ProgressTypeProgress with a non-zero Total gets the bar; ProgressTypeSubStatus gets the text.
• Only the first few jobs render in detail, with the rest summarized as a queued count, because a hundred concurrent jobs cannot all be on screen.
• A final summary replaces the live view on Stop, reporting completed and failed counts.
• Colors follow the same rule as the rest of the CLI: ANSI indices 0 through 15, never hex, so the display adopts the user's terminal theme.

Without a terminal, or under --debug, the display skips the live view entirely and prints one line per update, since redrawing in place produces cursor escapes that whoever reads the output later cannot use.

func (d *Display) Start(updates <-chan Progress) {
    if utils.GlobalDebugFlag || !utils.StdoutIsTerminal {
        d.startLinear(updates)
        return
    }
    // consume updates into state, then redraw on the ticker
}

[INFO] http-bigfile.zip: Downloading 62% (485MB/782MB)
[OK] http-bigfile.zip: Done
[ERROR] s3-scan: timeout connecting to server