beat

README

beat

A background-job scheduler for Go: one job, an interval or cron schedule, middleware, a run timeout, and Uber Fx integration.

• No built-in metrics - after every run a Record goes to a Handler, the way slog hands a record to its handler. Metrics, logging and tracing are adapters you supply.
• It reads no config source - Config is a plain struct with env tags the application loads and supplies.
• Every run is bounded, so a job that ignores its context cannot wedge the loop.
• Panics crash the process unless you say otherwise, which is a middleware away.

go get github.com/uchaloop/beat

Quick start

BEAT_SPEC=@every 5s
BEAT_JOB_TIMEOUT=1m

fx.New(
	confx.Module(),
	confx.Provide[beat.Config]("beat"),

	fx.Provide(func() beat.Job {
		return func(ctx context.Context) (processed int, err error) {
			// Do one unit of work.
			return processed, nil
		}
	}),

	fx.Provide(func(log *slog.Logger) beat.Handler {
		return beat.HandlerFunc(func(_ context.Context, r beat.Record) {
			log.Info("beat run",
				"iteration", r.Iteration,
				"processed", r.Processed,
				"duration", r.Duration,
				"error", r.Err,
			)
		})
	}),

	beatfx.Module(beat.WithMiddleware(recovery.Middleware())),
).Run()

Without Fx:

scheduler, err := beat.MakeBeat(cfg, job, handler, opts...)

Configuration

Field	Variable	Default
Spec	SPEC	none - the deployment has to supply it
JobTimeout	JOB_TIMEOUT	1m, from Config.SetDefaults
Jitter	JITTER	0

The prefix comes from the instance name, so confx.Provide[beat.Config]("beat") reads BEAT_SPEC and the rest, and confx.Manifest[beat.Config]("beat") lists the same set from the type.

Middleware

beat.WithMiddleware(
	recovery.Middleware(recovery.WithLogger(logger)),  // keep the loop alive
	idle.Middleware(backoff),                          // pause on failure
	batch.Middleware(drain),                           // pause when there is nothing to do
)

The first is outermost. A middleware is func(next beat.Job) beat.Job, so writing your own needs nothing from this package.

OpenTelemetry

otelbeat records run duration, processed counts, status and schedule mode:

fx.New(
	otelbeatfx.Module(),
	beatfx.Module(),
)

Documentation

The scheduling modes, the options, the record and the reasons behind them are in the package documentation: pkg.go.dev/github.com/uchaloop/beat. Each middleware and the Fx integration document themselves.

Acknowledgements

I am grateful to the authors of robfig/cron and Uber Fx. Their work made this library possible.

License

MIT

Documentation

Overview

Package beat runs a single background Job on a schedule: on an interval ("@every 5s") or a cron expression ("*/5 * * * * *"), optionally after a start hook, bounded by a per-run timeout, wrapped by a chain of Middleware. Build a Beat with MakeBeat and drive it with Start/Stop, or use the beat/beatfx subpackage to wire it into an Uber Fx application.

One Beat runs one Job. beat is single-instance per application: a process that needs two schedules is two processes, which is also how a deployment scales and stops them independently.

The job and its record

A Job reports how much it processed and whether it failed:

type Job func(context.Context) (int, error)

beat has no built-in metrics. After every run it hands a Record - iteration, start, duration, processed, error, scheduling mode - to a Handler, the same way slog hands a Record to its handler. Metrics, logging and tracing are adapters the caller supplies; MultiHandler sends one record to several. github.com/uchaloop/otelbeat is such an adapter for OpenTelemetry.

Scheduling

An "@every" interval measures the gap from the end of one run to the start of the next, so the effective period grows by the Job's duration. A cron spec fires at fixed wall-clock points and skips a point a long run overruns - it never queues catch-up runs.

Config.Jitter delays the first interval run, or every cron tick, by a random amount drawn once from [0, Jitter). Replicas of one deployment read identical configuration, so a fixed offset could not stagger them; the draw is what does.

Every run is bounded by Config.JobTimeout, one minute unless set, so a Job that ignores its context cannot silently wedge the loop.

Configuration

Config is a plain struct with env tags that beat itself never reads. An application loads it - typically through github.com/uchaloop/confmaker, under the prefix it gives the instance - and supplies the filled value:

SPEC          the schedule; declared notEmpty, so a deployment that forgets
              it is told which variable is missing before anything is built
JOB_TIMEOUT   bounds one run; one minute from Config.SetDefaults
JITTER        the upper bound of the start delay; zero disables it

A Config built in Go by hand never goes through SetDefaults, which is why MakeBeat treats a zero JobTimeout as the default too.

Options and middleware

WithMiddleware, WithHandler, WithOnStart, WithOnStop and WithGracefulStop configure what a Config cannot carry. Middleware wraps the Job:

type Middleware func(next Job) Job

In WithMiddleware(a, b, c) the first is outermost, so it is the order the wrapping reads in. Three are provided: beat/middleware/recovery, beat/middleware/idle and beat/middleware/batch.

Shutdown cancels the running job by default. WithGracefulStop lets it finish instead, within whatever stop timeout the application allows.

Panics

The core does not recover them. A panic in the Job, or in a Handler, propagates and crashes the process with a full stack on stderr - the honest default for user code that misbehaves, and one that a supervisor restarts. To keep the scheduler alive instead, add beat/middleware/recovery, which reports the panic as a PanicError on the Record.

Index

• func Wait(ctx context.Context, d time.Duration)
• type Beat
  • func MakeBeat(cfg Config, job Job, handler Handler, opts ...Option) (*Beat, error)
  • func (b *Beat) Start(ctx context.Context) error
  • func (b *Beat) Stop(ctx context.Context) error
• type Config
  • func (c *Config) SetDefaults()
  • func (c Config) Validate() error
• type Handler
  • func MultiHandler(handlers ...Handler) Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, r Record)
• type Job
• type Middleware
• type Mode
• type Option
  • func WithGracefulStop() Option
  • func WithHandler(h Handler) Option
  • func WithMiddleware(mw ...Middleware) Option
  • func WithOnStart(h func(context.Context) error) Option
  • func WithOnStop(h func(context.Context) error) Option
• type PanicError
  • func (e *PanicError) Error() string
• type Record

Functions

func Wait

func Wait(ctx context.Context, d time.Duration)

Wait blocks for d or until ctx is done, whichever comes first. A non-positive d returns immediately. It is exported for middleware authors that need a cancellable delay.

Types

type Beat

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

Beat is a scheduled runner for one Job. Build it with Module; the Fx lifecycle drives its hooks and loop.

It keeps two independent contexts: loopCtx gates scheduling (cancelled to stop launching new runs) and jobCtx is the parent of each run (cancelled to abort an in-flight run). Splitting them lets stop drain - end scheduling while letting the current run finish - or cancel, depending on the graceful flag.

func MakeBeat

func MakeBeat(cfg Config, job Job, handler Handler, opts ...Option) (*Beat, error)

MakeBeat resolves cfg against opts and builds a Beat. job is required; handler may be nil (a no-op handler is used). Drive the returned Beat with Start and Stop; the beatfx subpackage does this from the Fx lifecycle.

func (*Beat) Start

func (b *Beat) Start(ctx context.Context) error

Start runs the OnStart hook and launches the run loop. It is the OnStart half of the lifecycle; the loop runs in its own goroutine, so Start returns once it is launched. ctx bounds the hook. A returned error aborts startup.

func (*Beat) Stop

func (b *Beat) Stop(ctx context.Context) error

Stop ends scheduling, waits for the loop to finish (bounded by ctx), then runs the OnStop hook. By default the in-flight run's context is cancelled at once; with graceful stop the run is left to finish and is only cancelled if it outlives ctx.

type Config

type Config struct {
	// Spec is the schedule: an interval such as "@every 5s" or a cron
	// expression such as "*/5 * * * * *". The deployment has to supply it: there
	// is no schedule that makes sense for every job.
	Spec string `env:"SPEC,notEmpty"`

	// JobTimeout bounds the context of a single Job execution. Every execution is
	// bounded, so set a larger value for a legitimately long Job.
	JobTimeout time.Duration `env:"JOB_TIMEOUT"`

	// Jitter is the maximum random delay used to stagger instances so replicas
	// do not all fire at once. For an interval it is applied once before the
	// first run; for cron it is added to every tick, so keep it below the cron
	// interval. Zero disables it.
	Jitter time.Duration `env:"JITTER"`
}

Config is the configuration beat declares. The env tags are inert strings: an application loads them with confmaker/confx and provides the filled Config into the container. beat itself never reads the environment.

func (*Config) SetDefaults

func (c *Config) SetDefaults()

SetDefaults establishes the values a deployment does not have to think about. confmaker/confx 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 MakeBeat still treats a zero JobTimeout as the default. Both paths apply the same constant; neither states the value twice.

func (Config) Validate

func (c Config) Validate() error

Validate reports whether the Config is usable. confmaker/confx calls it after filling the struct, and it reports every problem at once rather than the first: a deployment is fixed in a config map and rolled out, so one report is one round trip.

The Spec check overlaps the notEmpty tag on purpose. The tag speaks to a deployment - it names the variable and fires before anything is built - while this speaks to any caller, including one that builds a Config in Go and never goes near a loader.

type Handler

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

Handler receives a Record after every run. It is the single observability seam of beat: metrics, logging and tracing are all built on top of it, so the core depends on none of them. Handle must not block for long - it runs inline on the run loop. It mirrors slog.Handler.

func MultiHandler

func MultiHandler(handlers ...Handler) Handler

MultiHandler returns a Handler that passes each Record to every handler in turn, skipping nil ones. Use it to run several sinks - metrics and logging, say - behind Module's single Handler slot. Handlers run in order and are not isolated: the core does not recover a panicking handler, so one panic propagates and crashes the process. Keep handlers panic-free.

type HandlerFunc

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

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

func (HandlerFunc) Handle

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

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

type Job

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

Job is the unit of work beat runs on each tick. It reports how many items it processed - return 0 when the job is not item-oriented - and an error if the run failed. A Job must respect ctx: beat bounds it with the job timeout and, unless graceful stop is enabled, cancels it on shutdown.

type Middleware

type Middleware func(next Job) Job

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

type Mode

type Mode string

Mode is the scheduling mode of the Beat that produced a run.

const (
	// ModeInterval is an "@every" interval schedule.
	ModeInterval Mode = "interval"
	// ModeCron is a cron-expression schedule.
	ModeCron Mode = "cron"
)

type Option

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

Option customizes a Beat built by Module.

func WithGracefulStop

func WithGracefulStop() Option

WithGracefulStop lets the in-flight run finish on shutdown instead of having its context cancelled. beat stops scheduling new runs immediately; the current run continues until it returns or the stop deadline (fx.StopTimeout) elapses, at which point its context is cancelled as a last resort.

func WithHandler

func WithHandler(h Handler) Option

WithHandler sets the observability Handler. It overrides a Handler provided through the Fx container.

func WithMiddleware

func WithMiddleware(mw ...Middleware) Option

WithMiddleware wraps the Job. The first middleware is the outermost layer and runs first.

func WithOnStart

func WithOnStart(h func(context.Context) error) Option

WithOnStart registers a hook run once before the loop starts. A returned error aborts startup.

func WithOnStop

func WithOnStop(h func(context.Context) error) Option

WithOnStop registers a hook run once after the loop has stopped.

type PanicError

type PanicError struct {
	Value any
	Stack []byte
}

PanicError wraps a value recovered from a panic in the Job. The recovery middleware returns it as the run error, so a Handler can tell a panic from an ordinary error with errors.As and inspect the recovered value and its stack. The beat core does not recover panics; without the recovery middleware a panic crashes the process (with a stack on stderr).

func (*PanicError) Error

func (e *PanicError) Error() string

type Record

type Record struct {
	// Iteration is the 1-based run counter for this Beat instance.
	Iteration int64
	// Start is the wall-clock time the run began.
	Start time.Time
	// Duration is how long the Job took.
	Duration time.Duration
	// Processed is the number of items the Job reported for this run.
	Processed int
	// Err is the error the Job returned, or nil on success. A panic recovered
	// by the recovery middleware is reported as a *PanicError.
	Err error
	// Mode is the scheduling mode that produced this run.
	Mode Mode
}

Record describes a single completed run of the Job. It is handed to a Handler after each run, mirroring the shape of slog.Record: the library produces the data, the caller decides what to do with it (metrics, logging, tracing).