runner

go-runner

A small Go library that runs a set of jobs every minute. Each minute the runner asks every registered job whether it should run (IsRun(now)), and concurrently executes the ones that say yes — up to a configurable concurrency limit.

• Push the current set of jobs to a channel; the runner queues batches internally and evaluates them on each minute tick.
• Jobs implement JobInterface (IsRun, GetID, Run).
• Concurrent execution within a tick, with a global concurrency cap.
• Backpressure-aware batch queue, optional dedupe, panic recovery, graceful shutdown, runtime stats.

Install

go get github.com/Sotaneum/go-runner

Quick Start

package main

import (
    "errors"
    "log"
    "time"

    runner "github.com/Sotaneum/go-runner"
)

type Job struct{ id string }

func (j *Job) GetID() string          { return j.id }
func (j *Job) IsRun(t time.Time) bool { return true } // every minute
func (j *Job) Run() (any, error)      { return "done", nil }

func main() {
    runnerCh := make(chan []runner.JobInterface)
    r := runner.NewRunnerWithLimit(runnerCh, 10)
    defer r.StopAndWait()

    runnerCh <- []runner.JobInterface{&Job{id: "a"}, &Job{id: "b"}}

    for res := range r.ResultCh {
        for _, jr := range res.Items {
            switch {
            case errors.Is(jr.Err, runner.ErrSkippedDuplicate):
                log.Printf("%s skipped: duplicate in flight", jr.ID)
            case jr.Err != nil:
                log.Printf("%s failed: %v", jr.ID, jr.Err) // *runner.PanicError
            default:
                log.Printf("%s -> %v (%v)", jr.ID, jr.Value, jr.EndedAt.Sub(jr.StartedAt))
            }
        }
    }
}

How It Works

caller ──[runnerCh]──▶ ingest ──▶ batch FIFO ──▶ createQueue ──▶ start ──▶ runQueue
                                       ▲              │           │
                              minute tick (timeChecker)            └─▶ goroutine per job (capped by limit) ──▶ ResultCh

• runnerCh accepts a batch (full job list) at a time. Each push is queued into the internal FIFO; the caller does not need to wait for the previous batch to finish.
• Every minute, createQueue pops one batch, evaluates IsRun(now) on each job, and forwards the surviving jobs to start.
• start hands the queue off to a background goroutine and immediately listens for the next queue, so a slow batch never blocks the next tick.
• Each job runs in its own goroutine, gated by a global concurrency semaphore (limit).
• Results for a queue are delivered as a single Result on ResultCh once all jobs in that queue finish.

Concurrency

• Jobs in a single queue execute concurrently.
• The concurrency limit is global across all in-flight queues (not per-queue).
• When the limit is reached, additional jobs block until a slot frees up.
• Default limit: 50. Set explicitly via NewRunnerWithLimit.

Because queues can overlap, the same GetID() appearing in two queues will, by default, run twice concurrently. Either make Run() safe for that, ensure uniqueness across queues, or enable WithDedupe().

Options

NewRunnerWithLimit accepts variadic Option values; existing callers without options remain source-compatible.

r := runner.NewRunnerWithLimit(runnerCh, 10,
    runner.WithDedupe(),
    runner.WithBatchBuffer(64),
)

WithDedupe()

Prevents the same GetID() from running concurrently. If a job with the same ID is already in flight (within the same queue or across overlapping queues), the duplicate is recorded as JobResult{ID, Err: ErrSkippedDuplicate} and Run() is not invoked. The skip is also reflected in Stats().DedupeSkipsTotal.

WithBatchBuffer(n)

Sets the size of the internal FIFO queue that holds batches received from runnerCh. Default 64, minimum 1.

• Each minute tick consumes one batch.
• If the caller pushes batches faster than they are consumed, the queue fills and subsequent runnerCh sends block, providing natural backpressure. The caller does not need to manually pace itself.
• If IsRun(now) returns false at the time the batch is finally dequeued (e.g. it lagged behind), that job is dropped from the queue.

Submitting Batches

Two equivalent ways:

// 1) caller-owned channel
runnerCh := make(chan []runner.JobInterface)
r := runner.NewRunnerWithLimit(runnerCh, 10)
runnerCh <- batch

// 2) library-managed channel
r := runner.New(10)
ok := r.Push(batch) // false if Stop has been called

For synchronous, immediate execution of a single batch (no tick wait) and per-batch correlation:

reply := r.PushAndAwait(batch)
res := <-reply // closed after delivery

Result and Errors

type Result struct {
    StartedAt time.Time
    EndedAt   time.Time
    Items     []JobResult
}

type JobResult struct {
    ID        string
    Value     any       // Run() return value, valid only when Err == nil
    Err       error     // *PanicError, ErrSkippedDuplicate, or nil
    StartedAt time.Time
    EndedAt   time.Time
}

type PanicError struct {
    ID        string
    Recovered any
    Stack     []byte
}

Result provides convenience helpers:

res.ByID()          // map[string]JobResult — O(1) lookup by ID
res.Errors()        // map[string]error — only entries with Err != nil
res.PanicErrors()   // map[string]*PanicError — panics only

Distinguish error sources with errors.As / errors.Is:

var pe *runner.PanicError
switch {
case errors.As(jr.Err, &pe):
    log.Printf("%s panicked: %v\n%s", pe.ID, pe.Recovered, pe.Stack)
case errors.Is(jr.Err, runner.ErrSkippedDuplicate):
    // skipped by WithDedupe
case jr.Err != nil:
    // (none currently produced, reserved for future)
}

Lifecycle

r := runner.NewRunner(runnerCh)
defer r.StopAndWait() // graceful: stop accepting new work and wait for in-flight queues

In-flight Run() invocations are not cancelled — JobInterface does not currently expose a context.Context.

Channel ownership

• The caller owns runnerCh. Closing it stops the ingest goroutine but does not stop the runner — call Stop() for that.
• The library owns ResultCh. It is buffered (size 8) and is closed by Wait() (or StopAndWait) once all in-flight queues have finished. After close, range r.ResultCh exits naturally.
• If the caller falls behind, the oldest result is dropped and Stats().DroppedResultsTotal is incremented.

Stats

s := r.Stats()
// s.InFlightJobs        — currently executing jobs (semaphore depth)
// s.BatchQueueDepth     — batches waiting to be consumed by createQueue
// s.BatchQueueCapacity  — configured via WithBatchBuffer
// s.DedupeSkipsTotal    — cumulative duplicate skips
// s.DroppedResultsTotal — cumulative ResultCh overflow drops

API

JobInterface

type JobInterface interface {
    IsRun(t time.Time) bool // called once per tick at evaluation time
    GetID() string          // stable identifier; result map key (and dedupe key if enabled)
    Run() (any, error)      // executed when IsRun returned true.
                            // value goes to JobResult.Value, error to JobResult.Err.
                            // a panic also lands in Err as *PanicError.
}

Examples

Runnable programs in examples/:

• examples/basic — minimal usage
• examples/dedupe — WithDedupe() + dedupe skip
• examples/with-stats — Stats() monitoring + PushAndAwait

Testing

go test -race ./...               # unit + leak (via go.uber.org/goleak)
go test -bench=. -benchmem ./...  # benchmarks
staticcheck ./...                 # style/lint

License

MIT — Copyright (c) 2021 Sotaneum