job

README

job

One attempt at a piece of work, and nothing about when the next one happens.

A short-lived process started by a Kubernetes CronJob runs one attempt and exits. beat runs the same work repeatedly inside a long-lived process. Both drive the same Runner, so the work, its middleware, its timeout and its classification are written once.

• One attempt per call - no retry, no goroutine of its own, no handler call.
• A cooperative timeout - it cancels the attempt's context and says so even when the work stayed quiet about it.
• No dependencies beyond the standard library and uchaloop/validate.

go get github.com/uchaloop/job

Quick start

runner, err := job.MakeRunner(
    job.Config{Timeout: 3 * time.Minute},
    func(ctx context.Context) (int, error) {
        return queue.ProcessBatch(ctx, 1000)
    },
    job.WithMiddleware(recovery.Middleware()),
)
if err != nil {
    return err
}

result := runner.Run(workCtx)
handler.Handle(observeCtx, result)
// The exit code follows result.Outcome, after cleanup.

The contract

Run calls the chain at most once, synchronously, and waits for it to return. A caller's context that is already done skips the work entirely and reports why, measuring nothing.

Field	Meaning
Start, Duration	When the chain ran and for how long. Handlers are not part of it
Processed	What the work reported, kept even when it then failed
Err	The chain's error, which stays nil when a timeout cut a quiet Func short
Outcome	ok, error, panic, timeout or canceled - authoritative

The timeout is cooperative, as every deadline in Go is: it cancels the context and cannot interrupt the function. Work must return when its context is done and must wait for its own goroutines first. Work that ignores cancellation holds its caller for as long as it likes, and no result is reported until it returns.

Outcome, not Err, decides whether an attempt succeeded: a Func cut short by the deadline may still return a nil error, and that is not a success.

Panics

They propagate by default. To turn one into a *job.PanicError on the result instead, add job/middleware/recovery:

job.WithMiddleware(recovery.Middleware(recovery.WithLogger(logger)))

It protects only the wrapped work - not a handler, not goroutines the work started.

Choosing which process runs the work

job/assignment is an optional, pure policy for a deployment that runs in several clusters: one cluster owns each scheduled point, and all of its replicas run it.

rotation, err := assignment.MakeRotation(assignment.Config{
    Clusters: []string{"el", "xc", "dm"},
    Current:  os.Getenv("CLUSTER"),
    Period:   4 * time.Hour,
})

decision, err := rotation.Decide(assignment.Invocation{ScheduledFor: point})

It reads no clock and keeps no state, so every process computes the same answer from the same configuration - nothing is coordinated at run time. It decides who should try, not that anyone did: a cluster that is down leaves its points unserved, and no other cluster takes over. Work must survive a skipped attempt.

The caller passes the scheduled point, never the current time, so a pod that starts late and a retry of the same scheduled Job decide as the original point did.

Changing the cluster list or the period changes who owns which point, and nothing checks that every process agrees: a rolling update produces duplicates and gaps at once. Change the topology by restarting every pod in every cluster and accept the pause - there is no hot reconfiguration, by choice.

Configuration

Field	Default env name	Default
Timeout	JOB_TIMEOUT	1m

Config names its default instance job, so confx.Provide[job.Config]() reads JOB_TIMEOUT.

Uber Fx

job/jobfx is the only package here that imports Fx, so a process that builds its Runner by hand never compiles it in. It ships in this module for now - the Fx requirement is therefore in go.mod and in your build list; a split into its own module is left for a later release.

It provides the Runner from the container and runs nothing: a one-shot process owns its own sequence.

app.Start(startupCtx)          // dependencies first
result := runner.Run(workCtx)  // one attempt
handler.Handle(observeCtx, result)
app.Stop(cleanupCtx)           // a fresh budget, not the cancelled work context
os.Exit(code(result))

Running the work inside OnStart would fold its duration into the startup timeout, and fx.Shutdowner would report an ordinary completion as a termination signal. See the package's ExampleModule for the whole sequence, including the cluster decision and the exit code.

Documentation

The contract, the outcomes and the reasons behind them are in the package documentation: pkg.go.dev/github.com/uchaloop/job.

License

MIT

Documentation

Overview

Package job runs one attempt at a piece of work and reports how it went.

It is the execution half of a scheduled job, with no opinion about when the next attempt happens. A short-lived process started by a Kubernetes CronJob runs one attempt and exits; github.com/uchaloop/beat runs the same work repeatedly inside a long-lived process. Both use the same Runner, so the work, its middleware, its timeout and its classification are written once.

runner, err := job.MakeRunner(
	job.Config{Timeout: 3 * time.Minute},
	func(ctx context.Context) (int, error) {
		return queue.ProcessBatch(ctx, 1000)
	},
	job.WithMiddleware(recovery.Middleware()),
)

result := runner.Run(ctx)

What one attempt means

Runner.Run calls the chain at most once, synchronously, and waits for it to return. It never retries, never starts a goroutine of its own and never calls a Handler: the caller decides what to do with the Result and when to try again. Two attempts never overlap because the caller does not overlap them.

The timeout is cooperative

Config.Timeout bounds the attempt's context. Like every deadline in Go it cancels; it cannot interrupt a running function. A Func must return when its context is done, and must wait for its own goroutines first. One that ignores cancellation holds its caller for as long as it likes, and no Result is reported until it returns.

Result.Outcome, not Result.Err, says whether an attempt succeeded: a Func cut short by the deadline may still return a nil error, and that is not a success. The runner classifies the attempt before releasing its own context, so a deadline or a cancellation is visible even when the work stayed quiet about it.

Panics

The runner does not recover them. A panic propagates to whoever called Run - the honest default for work that misbehaved. To turn one into a PanicError on the Result instead, add job/middleware/recovery.

Choosing which process runs the work

job/assignment is an optional, pure policy that picks one cluster to own a scheduled point when the same deployment runs in several of them. It is off unless an application asks for it, and it decides nothing about execution.

Index

• type Config
  • func (Config) ConfigName() string
  • func (c *Config) SetDefaults()
  • func (c Config) Validate() error
• type Func
• type Handler
  • func MultiHandler(handlers ...Handler) Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, r Result)
• type Middleware
• type Option
  • func WithMiddleware(mw ...Middleware) Option
• type Outcome
• type PanicError
  • func (e *PanicError) Error() string
• type Result
• type Runner
  • func MakeRunner(cfg Config, fn Func, opts ...Option) (*Runner, error)
  • func (r *Runner) Run(ctx context.Context) Result

Types

type Config

type Config struct {
	// Timeout bounds the context of a single attempt. Zero means the default of
	// one minute; a negative value is a configuration error.
	//
	// The bound is cooperative: it cancels the attempt's context and cannot
	// interrupt the function. A Func that ignores cancellation holds its caller
	// for as long as it likes, and no outcome is reported until it returns.
	Timeout time.Duration `env:"TIMEOUT"`
}

Config is the configuration job declares. The env tags are inert strings: an application loads them with github.com/uchaloop/confmaker and provides the filled Config. job itself never reads the environment.

func (Config) ConfigName

func (Config) ConfigName() string

ConfigName is the default instance name, "job": a loader such as confmaker reads JOB_TIMEOUT unless the application names the instance itself.

func (*Config) SetDefaults

func (c *Config) SetDefaults()

SetDefaults establishes the values a deployment does not have to think about. confmaker calls it before the environment is applied, so a variable left unset keeps what is set here, and a generated .env.example carries the real default rather than a blank.

A Config built in Go by hand does not go through it, which is why MakeRunner still treats a zero Timeout as the default. Both paths apply one constant.

func (Config) Validate

func (c Config) Validate() error

Validate reports whether the Config is usable.

type Func

type Func func(ctx context.Context) (int, error)

Func is one attempt at the work. It reports how many items it processed - return 0 when the work is not item-oriented - and an error if the attempt failed.

A Func must respect ctx: the runner bounds it with the configured timeout, and whoever owns the process cancels it on shutdown. The bound is cooperative, as every deadline in Go is: it cancels the context, it does not interrupt the running function. A Func must return when its context is done, and must wait for its own goroutines before it returns.

type Handler

type Handler interface {
	Handle(ctx context.Context, r Result)
}

Handler receives a Result after an attempt. It is the observability seam of job: metrics, logging and tracing are built on top of it, so the package depends on none of them.

The Runner never calls a Handler. Whoever owns the attempt delivers its Result exactly once - a one-shot process straight from Run, a scheduler through its own richer record - so there is no hidden second delivery.

func MultiHandler

func MultiHandler(handlers ...Handler) Handler

MultiHandler returns a Handler that passes each Result to every handler in turn, skipping nil ones. Handlers run in order and are not isolated: a panic in one propagates to the caller, so keep them panic-free.

type HandlerFunc

type HandlerFunc func(ctx context.Context, r Result)

HandlerFunc adapts an ordinary function to a Handler, like http.HandlerFunc.

func (HandlerFunc) Handle

func (f HandlerFunc) Handle(ctx context.Context, r Result)

Handle calls f. A nil HandlerFunc is a no-op.

type Middleware

type Middleware func(next Func) Func

Middleware wraps a Func to add behaviour around it - recovery, context enrichment, extra logging. It is classic Go composition: a Middleware receives the next Func and returns a Func that calls it.

A Middleware must keep the one-attempt contract: it may suppress the call to next, but must not call it twice or concurrently. Deciding when the next attempt happens is the caller's business, not a Middleware's.

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option customizes a Runner built by MakeRunner.

func WithMiddleware

func WithMiddleware(mw ...Middleware) Option

WithMiddleware wraps the work. The first middleware is the outermost layer and therefore runs first, which is the order the call reads in. Nil entries are skipped, and the chain is composed once, when the Runner is built.

type Outcome

type Outcome string

Outcome classifies how an attempt ended. It is the one label an observability adapter needs; Result.Err carries the detail behind it.

const (
	// OutcomeOK is an attempt that finished and returned no error.
	OutcomeOK Outcome = "ok"

	// OutcomeError is an attempt that returned an error of its own.
	OutcomeError Outcome = "error"

	// OutcomePanic is an attempt whose panic the recovery middleware turned
	// into a *PanicError. Without that middleware a panic never reaches a
	// Result - it propagates to the caller.
	OutcomePanic Outcome = "panic"

	// OutcomeTimeout is an attempt cut off by Config.Timeout. It is reported
	// even when the Func swallowed the cancellation and returned no error,
	// because the deadline belongs to the runner and the runner can see it
	// fire.
	OutcomeTimeout Outcome = "timeout"

	// OutcomeCanceled is an attempt cut off by the caller's context - a
	// shutdown, most often. It is not a failure of the work, so it carries its
	// own label rather than adding to the error count every time a process
	// stops.
	OutcomeCanceled Outcome = "canceled"
)

type PanicError

type PanicError struct {
	Value any
	Stack []byte
}

PanicError wraps a value recovered from a panic in the work. The recovery middleware returns it as the attempt's error, so a handler can tell a panic from an ordinary error - Result.Outcome reads OutcomePanic - and inspect the recovered value and its stack.

The runner does not recover panics; without the recovery middleware a panic propagates to whoever called Run.

func (*PanicError) Error

func (e *PanicError) Error() string

type Result

type Result struct {
	// Start is the wall-clock time the attempt began. It is the zero Time when
	// the caller's context was already done and the work was never called.
	Start time.Time

	// Duration is how long the chain took: the Func and its middleware, and
	// nothing else. Result handlers are not part of it.
	Duration time.Duration

	// Processed is the number of items the Func reported.
	Processed int

	// Err is the error the chain returned, or nil. It stays nil when a timeout
	// or a cancellation cut a Func short that chose to report no error, which
	// is why Outcome and not Err decides whether an attempt succeeded.
	Err error

	// Outcome classifies the attempt and is authoritative.
	Outcome Outcome
}

Result describes one completed attempt.

type Runner

type Runner struct {
	// contains filtered or unexported fields
}

Runner executes one attempt per call to Run. It holds the resolved configuration and the middleware chain, built once, so a caller that runs the same work repeatedly - beat, for instance - pays for the composition once.

A Runner keeps no state between attempts. Concurrent calls to Run are the caller's business: the runner serializes nothing, so they are safe only when the Func and its middleware are. Separate runners for separate logical jobs are simpler than one runner guarding shared state.

func MakeRunner

func MakeRunner(cfg Config, fn Func, opts ...Option) (*Runner, error)

MakeRunner resolves cfg against opts and builds a Runner. fn is required.

func (*Runner) Run

func (r *Runner) Run(ctx context.Context) Result

Run performs one attempt and reports how it went.

It calls the chain at most once, synchronously, on the calling goroutine, and waits for it to return. There is no retry, no goroutine of its own and no delivery to a handler: what to do with the Result, and when to attempt again, belong to the caller.

The attempt is bounded by the configured timeout, as far as a context can bound anything - the deadline cancels the attempt, it cannot interrupt it. A caller's context that is already done skips the work entirely and reports why.