job

README

job

Install · Quick start · How it works · Configuration · Examples

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 automatic observability 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.

Installation

Requires Go 1.27 or later.

go get github.com/uchaloop/job

Quick start

package main

import (
    "context"
    "log/slog"
    "time"

    "github.com/uchaloop/job"
)

func main() {
    runner, err := job.MakeRunner(
        job.Config{Timeout: 3 * time.Minute},
        func(ctx context.Context) (int64, error) {
            // Replace with one batch of application work.
            return 0, ctx.Err()
        },
    )
    if err != nil {
        slog.Error("configure job", "error", err)
        return
    }
    result := runner.Run(context.Background())
    slog.Info("attempt finished", "outcome", result.Outcome, "processed", result.Processed)
}

How it works

flowchart LR
    A["Run: check caller context"] --> B{"Already cancelled?"}
    B -->|yes| C["Unmeasured canceled or timeout Result"]
    B -->|no| D["Apply timeout"]
    D --> E["Middleware → work"]
    E --> F["Measure and classify Result"]
    F --> H{"Returned error and ErrorHandler configured?"}
    H -->|yes| I["ErrorHandler: independent timeout"]
    I --> G["Caller observes and cleans up"]
    H -->|no| G
    C --> G

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	Start and total duration, including optional ErrorHandler
WorkDuration, ErrorHandlerDuration	Time spent in each stage
ErrorHandlerCalled	Whether error processing ran, even if its duration is zero
ErrorHandlerErr	Callback error and/or expired callback context, separate from the work error
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

Func returns an int64 count, stored unchanged in Result.Processed, including partial progress on error. Return zero or more; the runner does not validate or clamp the count. For slice lengths, use int64(len(items)).

[!IMPORTANT] A timeout cancels the context; it 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. Classification uses this precedence after the chain returns:

Outcome	Condition
panic	Error contains a recovered *job.PanicError
timeout	Attempt context deadline expired, including an earlier caller deadline
canceled	Attempt context was cancelled
error	Chain returned another non-nil error
ok	None of the above

A deadline can therefore produce timeout even when Err is nil.

Error handling

An optional callback can log an error or persist application-defined data in a DLQ. The library does not define the message format or retry delivery.

runner, err := job.MakeRunner(
    job.Config{Timeout: time.Minute, ErrorHandlerTimeout: 15 * time.Second},
    work,
    job.WithErrorHandler(func(ctx context.Context, err error) error {
        return failures.Store(ctx, err) // Application-owned storage and error model.
    }),
)

The callback runs synchronously once, only for a non-nil error returned by the work or middleware. It gets caller context values and a fresh deadline, without inheriting cancellation. A work timeout returning nil does not call it; work skipped because the caller was already cancelled does not call it either.

[!IMPORTANT] Error processing has its own budget, even during shutdown. Allow time for both stages before closing dependencies. Neither context can force code to return.

Err and Outcome retain the original work result. ErrorHandlerErr reports callback failure, joining its returned error with context expiration. Successful DLQ delivery does not turn failed work into success. Duration includes both stages, so schedulers account for error processing too. Persistence and source acknowledgement remain application responsibilities; this callback alone does not guarantee delivery. Recovery middleware does not cover callback panics.

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:  "el",
    Period:   4 * time.Hour,
})
if err != nil {
    return err
}

decision, err := rotation.Decide(assignment.Invocation{ScheduledFor: point})
if err != nil {
    return err
}
// Call Runner.Run only when decision.Execute is true.

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.

Decision.Slot is a non-negative uint64. Scheduled points must be on the epoch-anchored grid, from the Unix epoch through epoch + MaxInt64 nanoseconds. OwnedAfter(after, count) counts this cluster's slots among the count positions following after, excluding after itself. It returns uint64 and never adds the two arguments, so counting remains safe even when their sum exceeds MaxUint64.

Changing the cluster list or period can produce duplicates or gaps while old and new configurations coexist. Stop scheduling across all clusters and wait for old attempts to finish, update the configuration everywhere, then resume; accept the pause. A rolling restart is not sufficient. There is no configuration agreement protocol or automatic failover.

Configuration

Field	Default env name	Default
Timeout	JOB_TIMEOUT	1m
ErrorHandlerTimeout	JOB_ERROR_HANDLER_TIMEOUT	1m

job.MakeRunner accepts ordinary Go values and never reads the environment. The env name above applies only when using the optional loader below.

Uber Fx

The optional jobfx adapter is a separate module. The job module has no Fx dependency.

go get github.com/uchaloop/jobfx@v0.1.0

Provide job.Config and job.Func, then use jobfx.Module() and fx.Populate(&runner). The application starts dependencies, calls runner.Run once, observes the result and stops dependencies with a fresh cleanup budget. The adapter's executable example uses app.Wait() for Fx shutdown requests.

Documentation

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

Recommended configuration

[!TIP] We recommend confmaker for typed ENV configuration and confx for its Fx integration. Configuration loading stays in the application; it is optional for the work libraries.

Configure from ENV with confmaker / confx

The ordinary job.Config{...} in the quick start can be replaced with:

cfg, err := confmaker.Load[job.Config]("job")
if err != nil {
    return err
}
// Pass cfg to job.MakeRunner.

Import github.com/uchaloop/confmaker. No Fx dependency is needed.

Example environment: JOB_TIMEOUT=3m.

For an Fx application, replace fx.Supply(job.Config{...}) with:

confx.Module(),
confx.Provide[job.Config]("job"),

Import github.com/uchaloop/confx and include confx.Module() once per application.

Related libraries

Library	Purpose
beat	Run the same work on a schedule
jobfx	Provide a Runner through Fx

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) (int64, 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 work 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.

Error processing

WithErrorHandler registers an optional callback for non-nil errors returned by the Func and middleware. It runs once after the work returns, with caller context values but a fresh ErrorHandlerTimeout budget (default one minute). Cancellation of the work or caller does not cancel error processing. Already-cancelled input skips both stages; a nil work error never calls it.

Result.Err and Outcome describe the work. ErrorHandlerErr separately reports callback failure or deadline expiration, even if the callback returns nil. Duration includes both stages; WorkDuration and ErrorHandlerDuration separate them. Observability Handler delivery remains the caller's responsibility. Shutdown budgets must allow error processing before closing dependencies. A noncooperative callback still blocks Run; panics propagate to the caller.

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 (c *Config) SetDefaults()
  • func (c Config) Validate() error
• type ErrorHandler
• 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 WithErrorHandler(handler ErrorHandler) 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

Examples

• WithErrorHandler

Types

type Config

type Config struct {
	// Timeout bounds the work and middleware. Zero defaults to one minute;
	// negative values are invalid.
	//
	// 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" envDescription:"Cooperative timeout for work and middleware; zero uses one minute."`

	// ErrorHandlerTimeout bounds error processing after the work returns.
	// Zero defaults to one minute; negative values are invalid. The handler's
	// context retains caller values but not its deadline or cancellation.
	// This budget is additional to Timeout and matters during shutdown too.
	ErrorHandlerTimeout time.Duration `env:"ERROR_HANDLER_TIMEOUT" envDescription:"Additional timeout for error handling after work returns; zero uses one minute."`
}

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. Register with the explicit instance name "job" to read JOB_*.

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 ErrorHandler

type ErrorHandler func(context.Context, error) error

ErrorHandler handles a non-nil error returned by the work and middleware. It runs synchronously with an independent, bounded context. It must respect cancellation. Its own error is recorded separately and is never handled again. Panics propagate; work recovery middleware does not cover this callback.

type Func

type Func func(ctx context.Context) (int64, 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. The count is expected to be non-negative; the runner preserves it as returned rather than validating or clamping it.

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 WithErrorHandler

func WithErrorHandler(handler ErrorHandler) Option

WithErrorHandler sets the optional error callback. A nil callback disables it. It is not called for skipped work or a nil work error, even on timeout.

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/uchaloop/job"
)

func main() {
	failure := errors.New("batch rejected")
	handled := false
	runner, err := job.MakeRunner(
		job.Config{Timeout: time.Minute, ErrorHandlerTimeout: 10 * time.Second},
		func(context.Context) (int64, error) { return 3, failure },
		job.WithErrorHandler(func(ctx context.Context, err error) error {
			// Replace with application-owned DLQ persistence or additional logging.
			handled = errors.Is(err, failure)
			return ctx.Err()
		}),
	)
	if err != nil {
		panic(err)
	}
	result := runner.Run(context.Background())
	fmt.Println(result.Processed, result.Outcome, handled, result.ErrorHandlerErr)
}

Output:
3 error true <nil>

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 means the attempt context deadline expired, whether set
	// by Config.Timeout or by an earlier caller deadline. It is reported
	// even when the Func swallowed the cancellation and returned no error,
	// because the runner checks the context before classifying the result.
	// Cancellation is cooperative; the result is produced only after return.
	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 covers the whole attempt, including ErrorHandler when called.
	// Observability handlers are not part of it.
	Duration time.Duration

	// WorkDuration measures the Func and middleware only.
	WorkDuration time.Duration

	// ErrorHandlerCalled distinguishes an invoked callback from an absent one,
	// including callbacks that finish without advancing the clock.
	ErrorHandlerCalled bool

	// ErrorHandlerDuration measures the error callback, or zero when not called.
	ErrorHandlerDuration time.Duration

	// ErrorHandlerErr contains the callback error joined with any expiration
	// of its context. Err and Outcome always describe the original work.
	ErrorHandlerErr error

	// Processed is the number of items the Func reported, including partial
	// progress on failure. Values should be non-negative; the runner preserves
	// them unchanged. Callers are responsible for supplying valid counts.
	Processed int64

	// 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, its middleware and ErrorHandler are. Separate runners for 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 an observability Handler: the caller owns the Result. An optional ErrorHandler processes a returned error before Run returns.

The work is bounded by Timeout, and error processing by ErrorHandlerTimeout. Both deadlines cancel contexts without interrupting execution. A caller's context that is already done skips the work entirely and reports why.