beat

README

beat

A background-job scheduler for Go with interval and cron schedules, middleware, run timeouts, result handlers, and Uber Fx integration.

Installation

go get github.com/uchaloop/beat

Fx

[beat]
spec = "@every 5s"
job_timeout = "1m"
jitter = "0s"

fx.New(
	confx.LoadDir("config"),
	confx.ProvideDefault[beat.Config]("beat"),

	fx.Provide(func() beat.Job {
		return func(ctx context.Context) (int, error) {
			processed := 0
			// Do one unit of work.

			return processed, nil
		}
	}),

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

	beatfx.Module(),
).Run()

Without a configuration file:

fx.New(
	fx.Supply(beat.Config{
		Spec:       "@every 5s",
		JobTimeout: time.Minute,
	}),
	fx.Provide(func() beat.Job { return work }),
	beatfx.Module(),
)

Configuration

Field	TOML	Environment	Default
Spec	spec	SPEC	required
JobTimeout	job_timeout	JOB_TIMEOUT	1m
Jitter	jitter	JITTER	0

Examples:

spec = "@every 5s"

spec = "*/5 * * * * *"

For interval schedules, the delay is measured after the previous run finishes. Cron schedules follow wall-clock times and do not queue missed runs.

JobTimeout limits each execution. Jitter delays the first interval run or each cron tick to reduce synchronized work across replicas.

Results

A job returns the number of processed items and an error:

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

After every run, a handler receives:

type Record struct {
	Iteration int64
	Start     time.Time
	Duration  time.Duration
	Processed int
	Err       error
	Mode      beat.Mode
}

Send a record to several destinations:

beat.MultiHandler(metricsHandler, loggingHandler)

Options

Pass static options to beatfx.Module:

beatfx.Module(
	beat.WithMiddleware(
		recovery.Middleware(),
		idle.Middleware(backoff),
	),
	beat.WithGracefulStop(),
)

Available options:

• WithMiddleware
• WithHandler
• WithOnStart
• WithOnStop
• WithGracefulStop

Build an option from Fx dependencies:

beatfx.AsOption(func(db *sql.DB) beat.Option {
	return beat.WithOnStart(db.PingContext)
})

By default, shutdown cancels the running job. WithGracefulStop lets it finish within the application's Fx stop timeout.

Middleware

Middleware wraps a job:

type Middleware func(next beat.Job) beat.Job

In WithMiddleware(a, b, c), a is the outermost middleware.

Panic recovery

By default, a panic terminates the process. Add recovery middleware to keep the scheduler running:

recovery.Middleware(
	recovery.WithLogger(logger),
)

The recovered panic is reported as *beat.PanicError.

Idle backoff

idle.Middleware(func(err error) time.Duration {
	if err != nil {
		return 5 * time.Second
	}

	return 0
})

Batch delay

batch.Middleware(func(processed int, err error) time.Duration {
	if processed == 0 {
		return 3 * time.Second
	}

	return 0
})

Custom middleware

func WithOperation(name string) beat.Middleware {
	return func(next beat.Job) beat.Job {
		return func(ctx context.Context) (int, error) {
			ctx = context.WithValue(ctx, operationKey{}, name)

			return next(ctx)
		}
	}
}

Without Fx

scheduler, err := beat.MakeBeat(cfg, job, handler, opts...)
if err != nil {
	return err
}

if err := scheduler.Start(ctx); err != nil {
	return err
}
defer scheduler.Stop(context.Background())

OpenTelemetry

Use otelbeat to record run duration, processed item counts, status, and schedule mode:

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

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.

beat has no built-in metrics: after every run it hands a Record to a Handler, the same way slog hands a Record to its handler. Metrics, logging and tracing are adapters the caller supplies. Configuration is declared as a plain struct with koanf and env tags and loaded by the application through confmaker/confx, so beat never reads files or the environment itself.

Scheduling has two modes. 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.

One Beat runs one Job; beat is single-instance per application. The core does not recover panics: a panic in the Job (or Handler) propagates and crashes the process, with a full stack on stderr - the honest default for user code that misbehaves. To keep the scheduler alive instead, add middleware/recovery, which recovers the panic, optionally logs it with its stack, and reports it 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) 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 * * * * *". Required.
	Spec string `koanf:"spec" env:"SPEC,required"`

	// JobTimeout bounds the context of a single Job execution. Zero (unset)
	// uses the default, 1m; every execution is bounded, so set a larger value
	// for a legitimately long Job.
	JobTimeout time.Duration `koanf:"job_timeout" 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 `koanf:"jitter" env:"JITTER"`
}

Config is the configuration beat declares. Both tag namespaces are inert strings: an application loads them with confmaker/confx (koanf from the file, env from the environment) and provides the filled Config into the container. beat itself never reads files or the environment.

func (Config) Validate

func (c Config) Validate() error

Validate reports whether the Config is usable. confmaker/confx calls it after filling the struct.

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).