beat

README

beat

beat runs a single background job on a schedule inside an Uber Fx application: once on an interval (@every 5s) or a cron expression (*/5 * * * * *), bounded by a per-run timeout, wrapped by a chain of middleware.

It 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 you supply. Configuration is a plain struct with koanf and env tags loaded by the application through confmaker/confx; beat never reads files or the environment itself.

The core package github.com/uchaloop/beat is the scheduler and has no Fx dependency - drive it directly with MakeBeat + (*Beat).Start/Stop. The Fx integration (Module, AsOption) lives in github.com/uchaloop/beat/beatfx.

Install

go get github.com/uchaloop/beat

Core contracts

type Job func(ctx context.Context) (int, error)     // your work, one run
type Middleware func(next Job) Job                   // classic composition

type Record struct {                                 // one completed run
    Iteration int64
    Start     time.Time
    Duration  time.Duration
    Processed int         // items processed this run; 0 if not applicable
    Err       error       // a recovered panic arrives as *beat.PanicError
    Mode      Mode        // "interval" or "cron"
}

type Handler interface { Handle(ctx context.Context, r Record) }
type HandlerFunc func(ctx context.Context, r Record) // adapter, like slog

Quick start (with confmaker/confx)

fx.New(
    // The application owns the config source.
    confx.LoadModule("config/local.toml"),
    confx.ProvideDefault[beat.Config]("beat"), // [beat] table + BEAT_* env
    fx.Supply(slog.Default()),                 // a *slog.Logger for the Handler

    // The business logic.
    fx.Provide(
        func() beat.Job {
            return func(ctx context.Context) (int, error) {
                var processed int
                // ... payload ...
                return processed, nil
            }
        },
    ),

    // Observability, injected from outside.
    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,
                        "err", r.Err,
                    )
                },
            )
        },
    ),

    beatfx.Module(),
).Run()

# config/local.toml
[beat]
spec        = "@every 5s"
job_timeout = "1m"
jitter      = "0s"

Without a config file

Supply the Config directly - for a service with no config file, or in tests:

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

Configuration

beat.Config declares both tag namespaces; confx fills koanf from the file and env from the environment (env overrides file).

Field	koanf	env	Default	Description
Spec	spec	SPEC (required)		@every 5s interval or cron */5 * * * * *
JobTimeout	job_timeout	JOB_TIMEOUT	1m	timeout on one Job execution; 0/unset uses the default
Jitter	jitter	JITTER	0	max random delay to stagger instances

JobTimeout: every execution is bounded, so a Job that ignores ctx cannot silently wedge the loop. Set a larger value for a legitimately long Job.

Jitter: 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.

Scheduling semantics

• @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. The first run fires right after the start delay.
• cron fires at fixed wall-clock points and skips a point a long run overruns - it never queues catch-up runs.

The core does not recover panics. A panic in the Job (or the Handler) propagates and crashes the process, with a full stack on stderr - the honest default for user code that misbehaves; let the orchestrator restart. To keep the scheduler alive instead, add middleware/recovery: it recovers the panic, optionally logs it with its stack (your logger), and reports it as a *beat.PanicError on the Record.

Stopping. By default the in-flight run's context is cancelled on shutdown. WithGracefulStop() instead lets the current run finish and only cancels it if it outlives fx.StopTimeout - use it for jobs that should not be interrupted mid-unit-of-work.

Lifecycle hooks and timeouts

WithOnStart / WithOnStop run during the Fx lifecycle and get the context Fx passes in. That context is bounded by the application's fx.StartTimeout / fx.StopTimeout (both default 15s) and is shared across all of the app's hooks - beat adds no per-hook timeout. A hook that needs longer than the default must be matched by raising the timeout on the top-level App:

fx.New( /* ... */, fx.StartTimeout(time.Minute), fx.StopTimeout(30*time.Second))

The run loop is not affected by this: it runs in its own goroutine, so a Job (and any middleware around it) can run as long as it likes, bounded only by JobTimeout per run and cancellation on shutdown. Keep the hooks for fast readiness work (open a pool, ping a dependency); put the real workload in the Job.

Options

Passed to beatfx.Module(...):

• WithMiddleware(mw ...Middleware) - the first middleware is the outermost layer and runs first.
• WithHandler(h Handler) - overrides a Handler from the container.
• WithOnStart(fn) / WithOnStop(fn) - lifecycle hooks.
• WithGracefulStop() - let the in-flight run finish on shutdown instead of cancelling it.

Options that need no dependencies go straight to Module. Options built from other container values are registered with AsOption, which accepts either a ready Option or a func(deps...) Option constructor Fx injects into; static and DI-built options are mixed (static apply first):

fx.New(
    // built from a dependency - Fx injects *sql.DB
    beatfx.AsOption(
        func(db *sql.DB) beat.Option {
            return beat.WithOnStart(db.PingContext)
        },
    ),
    beatfx.Module(beat.WithMiddleware(recovery.Middleware())), // no deps
)

Middleware

A Middleware is func(next Job) Job - standard Go composition, so each one is an ordinary function you can test in isolation. In WithMiddleware(a, b, c) the first is the outermost layer and runs first: a -> b -> c -> Job.

Built-in packages under github.com/uchaloop/beat/middleware:

recovery

The core lets a panic crash the process; add this middleware to keep the loop alive. It recovers the panic, optionally logs it with its stack (WithLogger), and always reports it as a *beat.PanicError on the Record - so downstream handlers (e.g. otelbeat) can classify it as a panic:

recovery.Middleware(recovery.WithLogger(logger))

Put it first in WithMiddleware so it also catches panics from the middleware inside it. For a custom panic-to-result mapping, write your own middleware (see Writing your own).

idle

Pauses after each run for a duration chosen from the run's error - a backoff on failure. The pause honours context cancellation:

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

batch

Like idle, but the delay is chosen from how much the run processed and its error - drain a queue fast while it is full, back off when it empties:

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

Writing your own

A Middleware is an ordinary function - no framework needed. A before/after wrapper that enriches the context and rewrites the result is just:

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

            processed, err := next(ctx)
            // inspect / wrap the result here
            return processed, err
        }
    }
}

Composing and injecting

Static middleware goes straight to Module:

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

Middleware that needs a container dependency is built with AsOption, where Fx injects the dependency:

beatfx.AsOption(
    func(log *slog.Logger) beat.Option {
        return beat.WithMiddleware(logging.Middleware(log))
    },
)

Metrics

The core has no metrics - it hands each run to a Handler. For OpenTelemetry, use the companion module github.com/uchaloop/otelbeat: a ready Handler recording run duration and processed counts, sliced by status and mode attributes. For any other backend, implement Handler yourself.

Run several sinks at once - metrics plus logging, say - with MultiHandler, which fans each Record out behind Module's single Handler slot:

beat.MultiHandler(metrics, logging)

Acknowledgements

beat builds on robfig/cron and uber-go/fx. Thanks to their authors and maintainers.

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