beat

README

beat

Install · Quick start · Scheduling · Configuration · Shutdown

Run one recurring job in a long-lived Go process. beat provides sequential execution, fixed-rate or fixed-delay scheduling, replica offsets, cooperative timeouts, and optional Fx integration. Requires Go 1.27 or later.

Use it for frequent polling and background work that benefits from reusable connections and in-memory state. beat has no calendar expressions, persistent schedule, replay after downtime, or coordination between replicas.

Installation

go get github.com/uchaloop/beat

Quick start

package main

import (
    "context"
    "log/slog"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/uchaloop/beat"
    "github.com/uchaloop/job"
)

func main() {
    shutdown, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()

    work := func(ctx context.Context) (int64, error) {
        // Replace this timer with one bounded batch of application work.
        timer := time.NewTimer(20 * time.Millisecond)
        defer timer.Stop()
        select {
        case <-timer.C:
            return 1, nil
        case <-ctx.Done():
            return 0, ctx.Err()
        }
    }
    handler := beat.HandlerFunc(func(_ context.Context, r beat.Record) {
        slog.Info("attempt completed", "outcome", r.Result.Outcome,
            "processed", r.Result.Processed, "duration", r.Result.Duration,
            "missed", r.Missed)
    })

    // The work, its middleware and its timeout belong to the runner.
    runner, err := job.MakeRunner(job.Config{Timeout: 2 * time.Second}, work)
    if err != nil {
        slog.Error("configure runner", "error", err)
        return
    }

    scheduler, err := beat.MakeBeat(
        beat.Config{Period: 5 * time.Second, Jitter: time.Second},
        runner,
        beat.WithHandler(handler),
        beat.WithGracefulStop(),
    )
    if err != nil {
        slog.Error("configure beat", "error", err)
        return
    }
    if err := scheduler.Start(shutdown); err != nil {
        slog.Error("start beat", "error", err)
        return
    }
    select {
    case <-shutdown.Done():
    case <-scheduler.Done(): // A terminal loop error is returned by Stop.
    }
    stopCtx, stopCancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer stopCancel()
    if err := scheduler.Stop(stopCtx); err != nil {
        slog.Error("stop beat", "error", err)
    }
}

Start's context applies to startup only. To stop a running Beat, call Stop with a fresh context. A nil Handler is allowed.

Scheduling

Mode	First run	Following runs
ModeFixedRate (default)	Next grid point strictly after startup	Grid points k × Period + offset, anchored to the Unix epoch
ModeFixedDelay	Startup time + offset	At least Period after the previous Job returns

[!NOTE] An attempt error does not stop the schedule or trigger an immediate retry.

A Beat never overlaps its Job calls. Fixed-rate runs that overrun grid points move to the first available point; no backlog of runs is queued. A pending target that is already past when the loop wakes runs once immediately, so the following run may be close behind. A restart joins the grid afresh.

For example, with a 5-minute period, a 25-second offset and a 4m50s Job:

Fixed rate:   10:00:25 ── Job ── 10:05:15   → next start 10:05:25
Fixed delay:  10:00:25 ── Job ── 10:05:15   → next start 10:10:15

Run loop

flowchart TD
    A[Start: run OnStart] --> B[Choose first target]
    B --> C[Wait for target or stop request]
    C --> D{Stopping?}
    D -->|yes| Z[Exit loop and close Done]
    D -->|no| E{Point owned here?}
    E -->|no| H
    E -->|yes| Q{"Attempt admitted before Stop?"}
    Q -->|no| Z
    Q -->|yes| R["job.Runner.Run: middleware and work"]
    R --> F[Build Record and call Handler inline]
    F --> G[Evaluate backoff from Record]
    G --> H[Compute next target and Missed for the next Record]
    H --> C

Result.Duration covers work, middleware and optional job.ErrorHandler. It excludes the observability Handler and backoff. The observability Handler and backoff callback still occupy the loop: keep both short. Configure error processing with job.WithErrorHandler on the Runner. Its separate timeout also consumes scheduling and shutdown time; foreign cluster points never call it because they do not run the work. Fixed-rate waits periodically recheck wall time; fixed-delay waits use monotonic time. Neither mode is a real-time execution guarantee.

Configuration and options

beat accepts an ordinary Config and never reads environment variables. The env names below apply to the optional loader described separately.

Field	Default env name	Default	Constraint
Period	BEAT_PERIOD	Required	Greater than zero
Jitter	BEAT_JITTER	0	Between zero and Period, inclusive

The attempt's timeout is not here: it belongs to the job.Runner beat drives, and is configured with job.Config.Timeout. It cancels the work's context and cannot interrupt the function - work must respect cancellation and join its own goroutines before returning. Work that never returns blocks subsequent runs and their records.

Options select the mode, offset, backoff, Handler, cluster rotation and lifecycle hooks. Repeated setter options use the last value. WithGracefulStop enables draining on shutdown. Middleware belongs to the Runner, not here.

Polling with backoff

With a short fixed-delay period, continue polling when a batch is full and rest longer after an empty or partial batch:

opts := []beat.Option{
    beat.WithMode(beat.ModeFixedDelay), // for example, Config.Period = 1s
    beat.WithBackoff(func(r beat.Record) time.Duration {
        if r.Result.Outcome != job.OutcomeOK {
            return 30 * time.Second
        }
        if r.Result.Processed < 1000 {
            return 5 * time.Minute
        }
        return 0
    }),
}

Backoff is a minimum pause measured from completion of the entire attempt (including job.ErrorHandler), not an addition to Period. In fixed-delay mode the next target is the latest of Job end + Period, Job end + positive backoff, and the time scheduling resumes. In fixed-rate mode it is the next eligible grid point no earlier than scheduling resumes or the backoff permits. Returning zero or a negative value adds no restriction; it does not remove Period. Deliberately bypassed backoff points are not Missed.

Replica offsets

Jitter draws one offset in [0, Jitter) per Beat; zero gives no offset. In fixed-rate mode it shifts every grid point. In fixed-delay mode it delays only the first run. Jitter == Period permits spreading across the full period. A wider range spreads planned starts but can increase initial waiting time.

To derive an offset from an application-provided identity:

offset := beat.OffsetFor(service+"/"+cluster+"/"+instance, cfg.Jitter)
scheduler, err := beat.MakeBeat(cfg, runner,
    beat.WithHandler(handler),
    beat.WithOffset(offset),
)

WithOffset requires 0 <= offset < Period. OffsetFor is stable only while its input is stable; replacing a Deployment Pod usually changes its name. Hashing and random draws do not guarantee gaps between replicas. Services may share the same Jitter range.

Offsets spread starts; they do not divide work or prevent duplicate effects. Queue claiming, retries and idempotency belong to the application, including when consumers run in different clusters.

Records and observability

A Handler receives one Record after each completed Job:

Fields	Meaning
Iteration, Mode, Period, LocalPeriod	Attempt number, mode, base step and local nominal step
GridPoint, ScheduledFor	Shared point before offset and this replica's target
Result.Start, Result.Duration	Actual start and total attempt time, including ErrorHandler
Result.WorkDuration, Result.ErrorHandlerDuration	Duration of each execution stage
Result.ErrorHandlerCalled	Whether error processing ran, even if its duration is zero
Result.ErrorHandlerErr	Failure of error processing, separate from Result.Err
Result.Processed, Result.Err	Values returned by the work
Result.Outcome	ok, error, panic, timeout or canceled
Missed	Unintentional grid losses computed after the preceding run

Missed is a uint64; counts exceeding MaxUint64 saturate instead of wrapping.

Result.Processed is an int64, preserved as returned by job.Func. The work should report a non-negative count, including partial progress on error.

Outcome is authoritative even if Err is nil. A recovered panic takes precedence, then cancellation of the Job context (timeout or shutdown), then the Job's error. Missed is zero in fixed-delay mode. It excludes intentional backoff and is reported with the next completed Job, so a shutdown or a stuck Job may prevent its delivery. The current Record's duration does not identify the cause of its Missed count.

Handlers run inline with a context without cancellation or a deadline. Give any I/O its own budget. MultiHandler calls sinks sequentially. The application owns its observability handlers and exporters.

Lifecycle and shutdown

• Start may run once. Repeated Start returns ErrAlreadyStarted; a stopped Beat returns ErrStopped and cannot be restarted.
• Stop prevents further scheduling. Normally it cancels the active attempt at once; WithGracefulStop lets it finish within Stop's budget. The runner's own timeout still applies.
• Concurrent Stop callers share the first Stop's result, but each waiting caller may return early if its own context expires.
• Stop waits for an in-progress OnStart. If OnStart fails, OnStop is not called; the startup hook must clean up partially acquired resources on failure.
• If OnStart succeeds but startup's context is cancelled, Start arranges OnStop with a fresh 15-second context unless a concurrent Stop already owns cleanup.
• If startup or the loop has not finished when Stop stops waiting, Stop returns ErrStillRunning joined with its context error. OnStop is skipped permanently.
• Otherwise OnStop runs synchronously after the loop exits. Hooks must respect their contexts; a blocked hook can outlive the supplied deadline.

[!IMPORTANT] Done means loop activity has ended. It does not mean cleanup has completed.

Do not call Stop synchronously from the work, handlers, backoff callback or lifecycle hooks: it waits for that callback to return. Signal the application's lifecycle owner through a channel, return, and let the owner call Stop.

Done() closes after startup/run-loop activity finishes, before OnStop. It closes for two reasons: a Stop you called, and a loop that ended by itself after an error it cannot carry on past - Stop then reports that error. Use it for fallback cleanup after Stop reports ErrStillRunning, and to notice the second case early. Bound any additional wait according to the application's shutdown policy:

if errors.Is(stopErr, beat.ErrStillRunning) {
    select {
    case <-scheduler.Done():
        // Startup and the loop have ended; perform application-owned cleanup.
    case <-cleanupCtx.Done():
        // Leave final termination to the process supervisor.
    }
}

Under beatfx this is connected for you: a loop that ends by itself asks the Fx application to stop, with exit code 1, and its error surfaces through the OnStop hook. It is a request - fx.Shutdowner broadcasts a signal. An application using app.Run() receives it and stops; one driving the lifecycle by hand must wait on app.Wait() and call Stop itself, or the request goes unanswered.

Fx and panic recovery

The optional beatfx adapter is a separate module. Install it with go get github.com/uchaloop/beatfx@v0.1.0.

The adapter supplies lifecycle wiring; the application supplies beat.Config, a *job.Runner and optional handlers. See the beatfx quick start for a complete Fx daemon. The standalone example above needs no DI framework.

Panics propagate by default. Recovery middleware catches panics only inside the wrapped work and returns *job.PanicError; it does not protect the Handler, the backoff callback, hooks, or goroutines the work created.

Running in several clusters

WithAssignment adds an optional rotation from job/assignment: one cluster owns each grid point and all of its replicas run it, while the others skip. It is off by default, requires ModeFixedRate and must share Config.Period.

rotation, err := assignment.MakeRotation(assignment.Config{
    Clusters: []string{"el", "xc", "dm"},
    Current:  "el",
    Period:   5 * time.Minute,
})

A point another cluster owns produces no run, no Record and no backoff, and is not counted as Missed - it was never this process's to serve. Use WithDecisionHandler to observe those decisions, since nothing else reports them. With N clusters Record.LocalPeriod is N * Period: three clusters on a five-minute period means each attempts every fifteen minutes.

The rotation names an owner; it does not guarantee the owner runs. A cluster that is down leaves its points unserved, and no other cluster takes over - that would need shared state the policy deliberately does not have. Work must therefore survive a skipped attempt.

When changing the cluster set or period, stop scheduling in all clusters and wait for old attempts to finish before starting the new configuration. A rolling restart can leave old and new ownership rules active together, producing both duplicate executions and unserved points.

Documentation

API reference · Compilable examples

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 beat.Config{...} in the quick start can be replaced with:

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

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

Example environment: BEAT_PERIOD=5m BEAT_JITTER=30s. Configure the Runner separately with job.Config (JOB_TIMEOUT).

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

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

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

Related libraries

Library	Purpose
job	One attempt, middleware and timeout
beatfx	Connect the scheduler to Fx

License

MIT

Documentation

Overview

Package beat schedules one job.Runner in a long-lived process. Attempts are sequential and never overlap. beat suits frequent polling with reusable process state; rare or calendar work belongs in a Kubernetes CronJob.

The work, its middleware and its timeout live in the Runner, so the same attempt can also be driven by a one-shot process. beat adds only the schedule.

Scheduling

ModeFixedRate, the default, places runs on an absolute grid of Period since the Unix epoch, shifted by an offset. Cadence does not depend on run duration, and a run that outlives its period reports the points it covered as Record.Missed rather than queueing them. ModeFixedDelay measures Period from the end of each run; only fixed rate preserves a shared grid.

The grid is anchored in UTC, so there is no time zone and no daylight saving. A point already past is served at once - one late run, no catch-up queue.

Config.Jitter draws an offset in [0, Jitter) once per process; WithOffset sets one outright. It may reach Period. Because the grid is absolute, the offset is the only thing separating replicas, and it holds for the life of the process. It spreads load and guards nothing: queue claiming and idempotency belong to the application.

WithBackoff sets a minimum pause after the complete attempt, including any job.ErrorHandler. The pause is outside timeouts and Result.Duration. It cannot shorten the configured schedule, and the points it holds back are not counted as losses.

Running in several clusters

WithAssignment adds an optional job/assignment rotation: one cluster owns each grid point and all of its replicas run it, while the others skip. It is off by default, needs ModeFixedRate, and must share Config.Period. A point another cluster owns produces no run, no Record and no backoff, and is not a Missed point; WithDecisionHandler observes those decisions. With N clusters Record.LocalPeriod is N*Period - the spacing a handler should measure a run against. An owner that is down simply leaves its points unserved.

Records

Handler receives a Record after each completed attempt. Record.Result is what the work reported - see job.Result - and everything beside it describes the schedule. Result.Outcome is authoritative even when Result.Err is nil.

Missed carries previously computed losses of this process's own points, excluding intentional backoff and points owned elsewhere. It travels with the next completed attempt, so the final losses can go unreported. A single Record cannot say why a point was lost; successive records can.

Lifecycle

MakeBeat builds a scheduler; Start calls OnStart and launches its loop. Start's context bounds startup only. A second Start fails; a stopped Beat cannot be restarted. Call Stop explicitly to stop scheduling and cancel the active attempt, or use WithGracefulStop to drain within Stop's context.

Whether a new attempt may begin is decided under the same mutex Stop moves the lifecycle state with, so a graceful stop lets an attempt already accepted finish and refuses every later one.

Stop calls OnStop only after startup succeeded and the loop ended. If waiting expires first, it returns ErrStillRunning with the context error and skips OnStop permanently. Done closes when loop activity has ended, not when OnStop has completed; it also closes when a loop ends by itself, and Stop then reports why. Concurrent Stop callers share the first result unless their own wait expires.

A failing OnStart is responsible for its partial cleanup; OnStop is not run. If OnStart succeeds but startup's context is cancelled, Start initiates cleanup with a fresh 15-second context unless a concurrent Stop already owns it. All hooks run synchronously and must respect their contexts.

Integration

The beatfx package connects the scheduler to Fx. job/middleware/recovery catches panics in the wrapped work as job.PanicError; panics elsewhere - Handler, hooks, backoff, child goroutines - propagate.

beat provides no calendar scheduling, persistence, replay after downtime, or delivery guarantee for business work. See the README for usage examples.

Index

• Variables
• func OffsetFor(id string, max time.Duration) time.Duration
• type Beat
  • func MakeBeat(cfg Config, runner *job.Runner, opts ...Option) (*Beat, error)
  • func (b *Beat) Done() <-chan struct{}
  • 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 Mode
• type Option
  • func WithAssignment(r *assignment.Rotation) Option
  • func WithBackoff(fn func(Record) time.Duration) Option
  • func WithDecisionHandler(fn func(assignment.Decision)) Option
  • func WithGracefulStop() Option
  • func WithHandler(h Handler) Option
  • func WithMode(m Mode) Option
  • func WithOffset(d time.Duration) Option
  • func WithOnStart(h func(context.Context) error) Option
  • func WithOnStop(h func(context.Context) error) Option
• type Record

Examples

• MakeBeat
• OffsetFor
• WithAssignment
• WithBackoff

Variables

var (
	// ErrAlreadyStarted reports a second Start on a Beat that is already
	// running. One Beat runs one loop: a second one would break the guarantee
	// that runs never overlap.
	ErrAlreadyStarted = errors.New("beat is already started")

	// ErrStopped reports a Start on a Beat that has been stopped. A stopped
	// Beat is not restartable - build a new one with MakeBeat.
	ErrStopped = errors.New("beat is stopped")

	// ErrStillRunning reports that Stop returned before the run did: the run
	// was cancelled, but had not returned by the time the stop deadline
	// elapsed. The OnStop hook is not run in that case, and [Beat.Done] is how
	// an application learns when the run finally ends.
	ErrStillRunning = errors.New("stop deadline elapsed before the run finished")
)

Functions

func OffsetFor

func OffsetFor(id string, max time.Duration) time.Duration

OffsetFor derives an offset in [0, max) from id, so a replica keeps the same slot for as long as that id is the same, instead of drawing a new one every time it boots:

beat.WithOffset(beat.OffsetFor(os.Getenv("POD_NAME"), 30*time.Second))

The result is stable but still arbitrary, so two replicas can land close together. When the replicas are numbered - a StatefulSet, or any deployment that can tell a pod its index - spacing them by hand is strictly better, because it is the only way to guarantee a gap:

beat.WithOffset(time.Duration(ordinal) * max / time.Duration(replicas))

An empty id or a non-positive max yields 0.

Example

package main

import (
	"context"
	"time"

	"github.com/uchaloop/beat"
	"github.com/uchaloop/job"
)

func work(ctx context.Context) (int64, error) {

	timer := time.NewTimer(20 * time.Millisecond)
	defer timer.Stop()

	select {
	case <-timer.C:
		return 1, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

func main() {
	cfg := beat.Config{Period: 5 * time.Minute, Jitter: 5 * time.Minute}

	offset := beat.OffsetFor("orders/cluster-a/worker-0", cfg.Jitter)

	runner, err := job.MakeRunner(job.Config{}, work)
	if err != nil {
		panic(err)
	}

	scheduler, err := beat.MakeBeat(cfg, runner, beat.WithOffset(offset))
	if err != nil {
		panic(err)
	}

	_ = scheduler
}

Types

type Beat

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

Beat runs one job.Runner on a schedule. Build it with MakeBeat and drive it with Start and Stop; the separate beatfx module does that from the Fx lifecycle.

It keeps two independent contexts: loopCtx gates scheduling (cancelled to stop launching new runs) and workCtx 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, runner *job.Runner, opts ...Option) (*Beat, error)

MakeBeat resolves cfg against opts and builds a Beat around runner. runner is required and carries the work, its middleware and its timeout; handler may be supplied with WithHandler, and a no-op is used without it.

Drive the returned Beat with Start and Stop; the separate beatfx module does this from the Fx lifecycle.

Example

package main

import (
	"context"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/uchaloop/beat"
	"github.com/uchaloop/job"
	"github.com/uchaloop/job/middleware/recovery"
)

func work(ctx context.Context) (int64, error) {

	timer := time.NewTimer(20 * time.Millisecond)
	defer timer.Stop()

	select {
	case <-timer.C:
		return 1, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

func main() {
	shutdown, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer cancel()

	// The work, its middleware and its timeout belong to the runner, so the
	// same attempt can also be driven by a one-shot process.
	runner, err := job.MakeRunner(
		job.Config{Timeout: 2 * time.Second},
		work,
		job.WithMiddleware(recovery.Middleware()),
	)
	if err != nil {
		slog.Error("configure runner", "error", err)

		return
	}

	handler := beat.HandlerFunc(func(_ context.Context, r beat.Record) {
		slog.Info("attempt completed",
			"outcome", r.Result.Outcome,
			"processed", r.Result.Processed,
			"duration", r.Result.Duration,
			"missed", r.Missed,
		)
	})

	scheduler, err := beat.MakeBeat(
		beat.Config{Period: 5 * time.Second, Jitter: time.Second},
		runner,
		beat.WithHandler(handler),
		beat.WithGracefulStop(),
	)
	if err != nil {
		slog.Error("configure beat", "error", err)

		return
	}

	if err := scheduler.Start(shutdown); err != nil {
		slog.Error("start beat", "error", err)

		return
	}

	<-shutdown.Done()

	stopCtx, stopCancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer stopCancel()

	if err := scheduler.Stop(stopCtx); err != nil {
		slog.Error("stop beat", "error", err)
	}
}

func (*Beat) Done

func (b *Beat) Done() <-chan struct{}

Done returns a channel that closes when the run loop has left and no work is in flight - or, for a Beat that never started, as soon as it is stopped.

It closes for two reasons: a Stop you called, and a loop that ended by itself after an error it cannot recover from. Stop returns on its own deadline and may return first, because a job timeout cancels an attempt's context but cannot interrupt it. Watch this to learn that the run finally ended - so a resource the work was still using can be released - or that a loop stopped without being asked to, in which case Stop reports why.

func (*Beat) Start

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

Start runs the OnStart hook and launches the run loop. The loop runs in its own goroutine, so Start returns once it is launched. ctx bounds the hook, and a returned error aborts startup.

It may be called once. A second call reports ErrAlreadyStarted, and a call on a stopped Beat reports ErrStopped; neither launches a second loop, which is what keeps runs from overlapping. If the hook fails the Beat is left stopped, and the OnStop hook is not run - the hooks are a pair, and the first one did not complete. If OnStart succeeds but ctx has been cancelled, Start invokes Stop with a fresh 15-second cleanup context and returns both errors. A concurrent Stop owns cleanup instead. Hooks must respect their contexts.

func (*Beat) Stop

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

Stop ends scheduling, waits for the run loop to leave (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.

If ctx elapses while a run is still going, Stop cancels the run, returns ErrStillRunning joined with ctx's error, and does not run the OnStop hook: the hook is documented to run after the loop has stopped, and running it against work that is still going would race it for the very resources the hook exists to release. Beat.Done reports when the run finally ends.

A loop that stopped by itself - a cluster rotation that could not decide a point - leaves its error here, joined with whatever the hook reports.

Stop is safe to call more than once and from several goroutines: the first call does the work, and the others report its result without running the hook again. Stopping a Beat that was never started does nothing and reports no error. Waiting callers may return their own context error before the first Stop finishes. The deadline bounds waiting; OnStop runs synchronously and must respect ctx. Stop during OnStart waits for that hook before cleanup.

Do not call Stop synchronously from the work, handlers, backoff callback or lifecycle hooks: it would wait for the calling callback itself to return. Signal the application's lifecycle owner instead, then return from the callback.

type Config

type Config struct {
	// Period is how often the work runs. Under ModeFixedRate it is the spacing
	// of the grid the runs sit on; under ModeFixedDelay it is the pause between
	// the end of one run and the start of the next. The deployment has to
	// supply it: there is no period that makes sense for every job.
	Period time.Duration `env:"PERIOD,notEmpty" envDescription:"Interval between scheduled runs; must be greater than zero."`

	// Jitter is the upper bound of the random offset that staggers replicas so
	// they do not all fire at once. The draw is half-open, [0, Jitter), and
	// happens once when the Beat is built, so the offset holds for the life of
	// the process. Zero disables it; WithOffset replaces the draw with a value
	// the application chooses.
	//
	// It may be as large as Period - spreading replicas over the whole period is
	// what an evenly polling deployment wants. Work tied to a boundary is the
	// exception: a wide jitter delays it by that much.
	//
	// It spreads load. It is not a guard against two replicas doing the same
	// work: runs longer than the offset overlap regardless.
	Jitter time.Duration `` /* 146-byte string literal not displayed */
}

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

The timeout of one attempt is not here: it belongs to the job.Runner beat drives, and lives in job.Config, so the same work carries the same bound whether a scheduler or a one-shot process runs it. Register with the explicit instance name "beat" to read BEAT_*.

func (Config) Validate

func (c Config) Validate() error

Validate reports whether the Config is usable. confmaker 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.

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. It mirrors slog.Handler.

Handle runs inline on the run loop, so its time comes out of the schedule: a slow Handler delays the next run and can cost a scheduled point. It is handed a context that is deliberately not cancellable, so that the last record of a shutdown still arrives - which means a Handler that blocks holds the loop, and Stop will give up on it with ErrStillRunning. Return promptly, and give any network call a budget of its own.

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 the 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 Mode

type Mode string

Mode is the timing policy of the Beat that produced a run.

const (
	// ModeFixedRate starts runs on an absolute grid: every Period since the
	// Unix epoch, shifted by the offset. The cadence does not depend on how
	// long a run takes, so two processes that share a Period agree on the
	// points without sharing anything else - which is what lets a per-replica
	// offset hold them apart for as long as they live, and what lets an
	// optional cluster rotation name an owner for each point.
	//
	// Runs still never overlap. A run that outlives its period covers the
	// points it ran through, and they are reported as Record.Missed rather
	// than queued for later.
	ModeFixedRate Mode = "fixed_rate"

	// ModeFixedDelay starts the next run one Period after the previous one
	// ended, so a slow run pushes the whole schedule back. Use it for a job
	// that should rest between runs rather than keep a cadence.
	//
	// There is no grid, so nothing is ever missed, an offset only delays the
	// first run, and a cluster rotation cannot be used: replicas have no shared
	// numbering to rotate over.
	ModeFixedDelay Mode = "fixed_delay"
)

type Option

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

Option customizes a Beat built by MakeBeat.

func WithAssignment

func WithAssignment(r *assignment.Rotation) Option

WithAssignment makes the loop ask a cluster rotation who owns each scheduled point, and skip the ones it does not own. It is off by default: without it every instance runs every point, which is what a deployment scaling for throughput wants.

It requires ModeFixedRate - replicas have no shared numbering under ModeFixedDelay - and the rotation's period must match Config.Period, since they describe the same grid. Both are checked by MakeBeat.

A point owned by another cluster produces no run, no Record and no backoff, and is not counted as Record.Missed: it was never this process's to serve. Use WithDecisionHandler to observe those points.

The rotation names an owner; it does not guarantee the owner runs. A cluster that is down simply leaves its points unserved - no other cluster takes over, because that would need shared state the policy deliberately does not have.

Example

package main

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

	"github.com/uchaloop/beat"
	"github.com/uchaloop/job"
	"github.com/uchaloop/job/assignment"
)

func work(ctx context.Context) (int64, error) {

	timer := time.NewTimer(20 * time.Millisecond)
	defer timer.Stop()

	select {
	case <-timer.C:
		return 1, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

func main() {
	// One cluster owns each point; every replica of that cluster runs it. With
	// three clusters on a five-minute period each one attempts every fifteen
	// minutes, and the deployment as a whole every five.
	rotation, err := assignment.MakeRotation(assignment.Config{
		Clusters: []string{"el", "xc", "dm"},
		Current:  os.Getenv("CLUSTER"),
		Period:   5 * time.Minute,
	})
	if err != nil {
		panic(err)
	}

	runner, err := job.MakeRunner(job.Config{Timeout: 4 * time.Minute}, work)
	if err != nil {
		panic(err)
	}

	scheduler, err := beat.MakeBeat(
		beat.Config{Period: 5 * time.Minute, Jitter: 30 * time.Second},
		runner,
		beat.WithAssignment(rotation),
		beat.WithDecisionHandler(func(d assignment.Decision) {
			slog.Debug("scheduled point", "slot", d.Slot, "owner", d.Owner, "ours", d.Execute)
		}),
	)
	if err != nil {
		panic(err)
	}

	_ = scheduler
}

func WithBackoff

func WithBackoff(fn func(Record) time.Duration) Option

WithBackoff pauses the loop after a run, for the duration fn reports from the Record of that run. Returning 0 - or not setting it at all - leaves the schedule as configured.

beat.WithBackoff(func(r beat.Record) time.Duration {
	if r.Result.Processed == 0 {
		return 30 * time.Second // the queue was empty, come back later
	}

	return 0
})

It sets a floor rather than a delay: the next run starts no earlier than the pause allows and no earlier than the schedule says, whichever is later. Under ModeFixedRate a pause that outlives the next grid points simply moves the run past them; they are not counted as Record.Missed, because a pause the application asked for is its schedule and not a loss.

The pause belongs here and not in a job.Middleware because the loop owns the timing. A pause taken inside the work would spend the job timeout, land in job.Result.Duration and be invisible to the schedule; this one does none of that. A point another cluster owns produces no Record and no backoff.

Example

package main

import (
	"context"
	"time"

	"github.com/uchaloop/beat"
	"github.com/uchaloop/job"
)

func work(ctx context.Context) (int64, error) {

	timer := time.NewTimer(20 * time.Millisecond)
	defer timer.Stop()

	select {
	case <-timer.C:
		return 1, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

func main() {
	runner, err := job.MakeRunner(job.Config{Timeout: 30 * time.Second}, work)
	if err != nil {
		panic(err)
	}

	// Drain while batches come back full, rest once the queue empties.
	scheduler, err := beat.MakeBeat(
		beat.Config{Period: time.Second},
		runner,
		beat.WithMode(beat.ModeFixedDelay),
		beat.WithBackoff(func(r beat.Record) time.Duration {
			switch {
			case r.Result.Outcome != job.OutcomeOK:
				return 30 * time.Second
			case r.Result.Processed < 1000:
				return 5 * time.Minute
			}

			return 0
		}),
	)
	if err != nil {
		panic(err)
	}

	// Call Start and Stop from the application's lifecycle.
	_ = scheduler
}

func WithDecisionHandler

func WithDecisionHandler(fn func(assignment.Decision)) Option

WithDecisionHandler observes every decision the cluster rotation makes, including the ones that allow the run. It is the only way to tell a point this process skipped from one it never had, since neither produces a Record.

A decision that allows a run is not a promise that one followed. The loop decides, reports here, and only then takes the last word on whether a new attempt may begin, so a Stop landing in between leaves a reported decision with no Record behind it. The order cannot be reversed without reopening the window in which a graceful stop would start fresh work.

It runs inline on the loop and must return promptly; a panic propagates.

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, 15s unless the application raises it) elapses, at which point its context is cancelled as a last resort and Stop reports ErrStillRunning.

A job whose duration varies should be matched by a stop timeout above its worst case, or the drain only sometimes happens - and a stop that runs out of time skips the OnStop hook, which is exactly when a graceful stop was supposed to help.

func WithHandler

func WithHandler(h Handler) Option

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

func WithMode

func WithMode(m Mode) Option

WithMode selects the timing policy. The default is ModeFixedRate, which keeps a cadence on an absolute grid; ModeFixedDelay measures the period from the end of the previous run instead.

This is an option rather than a Config field because the choice follows from what the job is - a poller that should rest between runs against a scraper that should keep a rate - and not from how a particular deployment is tuned.

func WithOffset

func WithOffset(d time.Duration) Option

WithOffset sets the offset within the period explicitly, instead of drawing one below Config.Jitter. Use it to give each replica a slot that survives a restart - see job/assignment for the cluster-level equivalent - or to spread replicas evenly by their index. The offset must be in [0, Period).

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 and leaves the Beat stopped, without running the OnStop hook - the two are a pair, and this one did not complete. If the hook succeeds after the startup context is cancelled, Start arranges cleanup; see Beat.Start.

func WithOnStop

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

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

"After the loop has stopped" is exact: if Stop's context elapses while a run is still going, the hook does not run at all, because it would otherwise race the work for whatever it is meant to release. Watch Beat.Done for the moment the run finally ends, after Stop returned ErrStillRunning. Done does not wait for OnStop. The hook runs synchronously and must respect its context.

type Record

type Record struct {
	// Result includes work and optional error processing: durations, count, errors and
	// outcome. See job.Result.
	Result job.Result

	// Iteration is the 1-based run counter for this Beat, useful as a
	// correlation id. It counts runs, not scheduled points - see Missed - and
	// does not advance for a point another cluster owns.
	Iteration int64

	// GridPoint is the point of the shared grid this run served, before the
	// per-replica offset. It is what every replica and every cluster agrees on,
	// and therefore what a cluster rotation decides over. Under ModeFixedDelay
	// it equals ScheduledFor, which is not a shared tick at all.
	GridPoint time.Time

	// ScheduledFor is the point this replica aimed at: GridPoint plus the
	// offset. Result.Start minus this is how late the run actually began.
	ScheduledFor time.Time

	// Period is the configured spacing of the shared grid.
	Period time.Duration

	// LocalPeriod is the spacing of the points assigned to this process: Period
	// times the number of clusters when a rotation is in use, and Period itself
	// otherwise. A handler reporting how much of its schedule the work uses
	// divides by this, not by Period.
	LocalPeriod time.Duration

	// Mode is the timing policy that produced this run.
	Mode Mode

	// Missed is how many scheduled points belonging to this process passed
	// without a run between the previous run and this one - because that run
	// and its Handler outlasted them, or because the clock moved.
	//
	// Points the loop chose not to serve are not counted: a backoff asking it to
	// wait is the schedule the application requested, and a point another
	// cluster owns was never this process's to serve. It is always 0 under
	// ModeFixedDelay, which has no grid to miss.
	// Counts larger than MaxUint64 are capped at that maximum.
	//
	// This Record's timestamps and Result describe the current attempt, not the
	// earlier work that lost the points. A single Record therefore cannot
	// identify the cause - the previous attempt may have been long while this
	// one is short and on time, and under a rotation other clusters' points sit
	// between them. Comparing successive records can.
	Missed uint64
}

Record describes a single completed run. It is handed to a Handler after each run: the scheduling around the work, plus the job.Result the work produced.

The nesting is the boundary made visible. Everything under Result belongs to the attempt and means the same in a one-shot process; everything beside it belongs to the schedule and exists only here.