Documentation
¶
Overview ¶
Package cron provides distributed cron scheduling with leader election and per-tenant support.
Cron entries are stored in the database and fired only by the cluster leader. This guarantees at-most-once firing even when multiple Dispatch instances are running.
Entry ¶
An Entry represents a recurring job schedule:
- Schedule: standard cron expression (e.g., "0 9 * * 1-5")
- JobName: the registered job definition to enqueue when fired
- Queue: target queue (defaults to "default")
- Payload: static JSON payload passed to every triggered job
- ScopeAppID / ScopeOrgID: tenant scoping
- Enabled: whether the entry fires
- LockedBy / LockedUntil: distributed lock fields (managed internally)
Registering a Cron ¶
Use engine.RegisterCron to add a cron entry at startup:
engine.RegisterCron(ctx, eng, "daily-report", "0 9 * * *",
GenerateReport, ReportInput{Format: "pdf"})
Enable / Disable ¶
Cron entries can be enabled or disabled at runtime via the admin API (POST /v1/crons/:cronId/enable and POST /v1/crons/:cronId/disable). Operator code goes through the engine's EnableCron, DisableCron, DeleteCron and TriggerCron. Enabling computes the next fire time from now, so an entry that was off past its old one does not fire a catch-up. NextFires previews a schedule's next fire times.
Scheduler ¶
The Scheduler evaluates due entries on every tick, acquires a distributed lock on each entry, re-reads it, and fires only if the store still has it enabled and due. It enqueues the corresponding job and records LastRunAt and NextRunAt with targeted writes that never touch Enabled, so a fire cannot undo an operator's disable. The [ext.CronFired] extension hook fires after each enqueue.
Index ¶
- func NextFires(schedule string, from time.Time, n int) ([]time.Time, *time.Location, error)
- func ParseSchedule(expr string) (cronlib.Schedule, error)
- type Definition
- type Emitter
- type EnqueueFunc
- type Entry
- type Scheduler
- type SchedulerOption
- func WithCronRefreshInterval(d time.Duration) SchedulerOption
- func WithLeaderTTL(d time.Duration) SchedulerOption
- func WithLockTTL(d time.Duration) SchedulerOption
- func WithSchedulerStoreCallTimeout(d time.Duration) SchedulerOption
- func WithTelemetry(t *telemetry.Telemetry) SchedulerOption
- func WithTickInterval(d time.Duration) SchedulerOption
- type Settings
- type Store
- type TargetedUpdater
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NextFires ¶
NextFires returns the next n times schedule fires strictly after from, and the location the schedule is evaluated in.
It parses with ParseSchedule, so it accepts exactly what the scheduler and engine.RegisterCron accept, and it steps the schedule the way the scheduler does, from a UTC instant. A schedule with no CRON_TZ= or TZ= prefix therefore runs in UTC whatever the process's local zone is, and a prefixed one runs in the zone it names. The returned times are in that location, so their wall clock reads as the schedule does.
Fewer than n times come back when the schedule runs out. The parser searches five years ahead and then gives up, so "0 0 30 2 *" (30 February) yields none and no error. n must be positive.
Types ¶
type Definition ¶
type Definition[T any] struct { // Name is the unique identifier for this cron entry. Name string // Schedule is a cron expression (e.g., "*/5 * * * *" or "@every 30s"). Schedule string // JobName is the name of the job to enqueue on each tick. JobName string // Payload is the default payload to enqueue with the job. Payload T // Queue overrides the default job queue (optional). Queue string }
Definition is a typed cron definition. T is the payload type (must be JSON-serializable).
type Emitter ¶
Emitter emits cron lifecycle events. ext.Registry satisfies this interface via EmitCronFired.
type EnqueueFunc ¶
type EnqueueFunc func(ctx context.Context, name string, payload []byte, opts ...job.Option) (id.JobID, error)
EnqueueFunc is the callback the scheduler uses to enqueue jobs. This breaks the import cycle: the engine provides the implementation.
type Entry ¶
type Entry struct {
dispatch.Entity
ID id.CronID `json:"id"`
Name string `json:"name"`
Schedule string `json:"schedule"`
JobName string `json:"job_name"`
Queue string `json:"queue,omitempty"`
Payload []byte `json:"payload,omitempty"`
ScopeAppID string `json:"scope_app_id,omitempty"`
ScopeOrgID string `json:"scope_org_id,omitempty"`
LastRunAt *time.Time `json:"last_run_at,omitempty"`
NextRunAt *time.Time `json:"next_run_at,omitempty"`
LockedBy string `json:"locked_by,omitempty"`
LockedUntil *time.Time `json:"locked_until,omitempty"`
Enabled bool `json:"enabled"`
}
Entry represents a scheduled cron job.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler runs cron entries on a tick loop. Only the cluster leader executes cron ticks to prevent double-firing.
func NewScheduler ¶
func NewScheduler( cronStore Store, clusterStore cluster.Store, enqueue EnqueueFunc, emitter Emitter, workerID id.WorkerID, logger log.Logger, opts ...SchedulerOption, ) *Scheduler
NewScheduler creates a Scheduler.
func (*Scheduler) InvalidateCronCache ¶
func (s *Scheduler) InvalidateCronCache()
InvalidateCronCache forces the scheduler to re-list cron entries from the store on the next tick. Call it after registering, updating, or removing entries so changes take effect before the next scheduled refresh.
type SchedulerOption ¶
type SchedulerOption func(*Scheduler)
SchedulerOption configures a Scheduler.
func WithCronRefreshInterval ¶
func WithCronRefreshInterval(d time.Duration) SchedulerOption
WithCronRefreshInterval sets how often the leader re-lists cron entries from the store. Between refreshes ticks run against an in-memory cache; InvalidateCronCache forces an early refresh. Default 30s.
func WithLeaderTTL ¶
func WithLeaderTTL(d time.Duration) SchedulerOption
WithLeaderTTL sets the TTL for leader election.
func WithLockTTL ¶
func WithLockTTL(d time.Duration) SchedulerOption
WithLockTTL sets the TTL for per-entry distributed locks.
func WithSchedulerStoreCallTimeout ¶
func WithSchedulerStoreCallTimeout(d time.Duration) SchedulerOption
WithSchedulerStoreCallTimeout caps a single store roundtrip. Pass a positive duration to override defaultStoreCallTimeout, zero to keep the default, or negative to disable the timeout (test-only).
func WithTelemetry ¶
func WithTelemetry(t *telemetry.Telemetry) SchedulerOption
WithTelemetry sets where the scheduler records its spans and metrics: one root "fire cron entry" span per due entry, the fire counter and lag histogram, and the leadership gauge. engine.Build passes the engine's own.
func WithTickInterval ¶
func WithTickInterval(d time.Duration) SchedulerOption
WithTickInterval sets how often the scheduler checks for due entries.
type Settings ¶
type Settings struct {
TickInterval, LeaderTTL, RefreshInterval, LockTTL, StoreCallTimeout time.Duration
StoreCallsBounded bool
}
Settings is the configuration of this process's scheduler.
type Store ¶
type Store interface {
TargetedUpdater
// RegisterCron persists a new cron entry. Returns an error if the name
// already exists.
RegisterCron(ctx context.Context, entry *Entry) error
// GetCron retrieves a cron entry by ID.
GetCron(ctx context.Context, entryID id.CronID) (*Entry, error)
// ListCrons returns all cron entries.
ListCrons(ctx context.Context) ([]*Entry, error)
// AcquireCronLock attempts to acquire a distributed lock for a cron entry.
// Returns true if the lock was acquired. The lock expires after ttl.
AcquireCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID, ttl time.Duration) (bool, error)
// ReleaseCronLock releases the distributed lock for a cron entry.
ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID) error
// UpdateCronLastRun records when a cron entry last fired.
UpdateCronLastRun(ctx context.Context, entryID id.CronID, at time.Time) error
// UpdateCronEntry updates a cron entry (Enabled, NextRunAt, etc.).
UpdateCronEntry(ctx context.Context, entry *Entry) error
// DeleteCron removes a cron entry by ID.
DeleteCron(ctx context.Context, entryID id.CronID) error
}
Store defines the persistence contract for cron entries.
type TargetedUpdater ¶
type TargetedUpdater interface {
// SetCronEnabled sets enabled, sets next_run_at when nextRunAt is
// non-nil (nil leaves it unchanged), and stamps updated_at. It
// returns dispatch.ErrCronNotFound when the entry does not exist.
SetCronEnabled(ctx context.Context, entryID id.CronID, enabled bool, nextRunAt *time.Time) error
// UpdateCronNextRun sets next_run_at and stamps updated_at, and
// nothing else; in particular it never touches enabled. It returns
// dispatch.ErrCronNotFound when the entry does not exist.
UpdateCronNextRun(ctx context.Context, entryID id.CronID, nextRunAt time.Time) error
}
TargetedUpdater writes single fields so a scheduler fire can never overwrite an operator's enable or disable. The whole-row UpdateCronEntry write-back is what let a disabled cron come back: the scheduler read the entry while it was enabled, an operator disabled it, and the scheduler's post-fire write put the old row, enabled flag and all, back on top.
Each method changes only the columns it names. Every other field, including the lock and last_run_at, reads back exactly as it was.
Source Files