Documentation
¶
Overview ¶
Package scheduler runs recurring jobs on wall-clock minute boundaries and one-shot jobs on demand, keyed by id, in one process.
Every instance of an app runs every recurring job at every boundary unless the Scheduler has a Claimer (WithClaimer), which lets one instance run each boundary of a recurring job. A keyed id excludes other runs of the id only in its own process unless the Scheduler has a Leaser (WithLeaser), which lets at most one run of an id go across the instances that share it. The claims subpackage has both on PostgreSQL.
Index ¶
- Constants
- Variables
- func ShouldRunLocalHour(tz string, hour int) bool
- func ShouldRunLocalTime(tz string, hour, minute int) bool
- type Claimer
- type Job
- type Lease
- type Leaser
- type Option
- type Schedule
- type Scheduler
- func (s *Scheduler) Active() int64
- func (s *Scheduler) Add(id string, when Schedule, job Job) error
- func (s *Scheduler) Idled() <-chan struct{}
- func (s *Scheduler) RunNext(ctx context.Context, id string, job Job) bool
- func (s *Scheduler) RunOnce(ctx context.Context, id string, job Job) bool
- func (s *Scheduler) Start(ctx context.Context)
- func (s *Scheduler) Supersede(ctx context.Context, id string, job Job) bool
Constants ¶
const ( // DefaultShutdownTimeout is how long Start waits for running jobs after its // ctx is done, unless WithShutdownTimeout sets another value. DefaultShutdownTimeout = 60 * time.Second // DefaultJobTimeout is the per-job timeout applied when the job's ctx has no // deadline, unless WithJobTimeout sets another value. DefaultJobTimeout = time.Hour )
Variables ¶
var ErrLeaseLost error = ez.New(ez.ECANCELED, "scheduler: lease lost", context.Canceled)
ErrLeaseLost is the context cause seen by a job that was cancelled because the Scheduler could no longer be sure it holds the lease of its id: a renewal answered that the lease is no longer this run's, or no renewal succeeded for the lease's TTL minus a safety margin. errors.Is(context.Cause(ctx), ErrLeaseLost) and errors.Is(context.Cause(ctx), context.Canceled) both hold.
var ErrSuperseded error = ez.New(ez.ECANCELED, "scheduler: superseded", context.Canceled)
ErrSuperseded is the context cause seen by a job that was cancelled because Supersede queued a replacement for the same id. errors.Is(context.Cause(ctx), ErrSuperseded) and errors.Is(context.Cause(ctx), context.Canceled) both hold.
Functions ¶
func ShouldRunLocalHour ¶ added in v1.6.0
ShouldRunLocalHour returns true if the local time in tz is exactly hour:00.
func ShouldRunLocalTime ¶ added in v1.6.4
ShouldRunLocalTime returns true if the local time in tzName matches hour:minute exactly.
Types ¶
type Claimer ¶ added in v1.33.0
type Claimer interface {
// Claim reports whether this instance may run job id for the minute
// boundary tick. It returns false when tick or a later boundary of id was
// already claimed by any instance, this one included (after its clock
// stepped back, or a restart on a host whose clock is behind), or when a
// claimed run of id is still going: not released and its lease not
// expired. lease is how long a claim blocks later boundaries when its
// instance never releases it; the Scheduler passes its job timeout
// (WithJobTimeout) plus one minute, capped at the largest Duration. An
// error means the store could not answer or refused tick.
Claim(ctx context.Context, id string, tick time.Time, lease time.Duration) (bool, error)
// Release ends the claim on tick for id, so it no longer blocks later
// boundaries. It must not end a claim on any other tick, so a late
// release cannot free a newer claim.
Release(ctx context.Context, id string, tick time.Time) error
}
Claimer decides which instance of an app runs a recurring job at a minute boundary, so that a job registered with Add on every instance runs once per boundary instead of once per instance. Set it with WithClaimer. The claims subpackage implements it on PostgreSQL.
The Scheduler claims only recurring runs. At a due boundary it first applies its own rule, skipping the boundary while the id is busy in this process without claiming. Otherwise it admits the run, which makes the id busy, and the run calls Claim before the job body. While the claim is in flight the id is busy as for a running job: RunOnce returns false, RunNext and Supersede queue behind it, and Active counts the id. Supersede and shutdown cancel the context passed to Claim. A store can still answer that the claim is won, and the claims subpackage's always answers a claim it has started (bounded by the 10 second claim timeout); the job is then not run and the claim is released. A won claim runs the job, and Release is called when the job returns, after a panic too, and before a pending job of the id runs. A refused claim skips the job, and so does a claim error, which is logged at error level unless the run was cancelled: missing a boundary is safer than every instance running it. After a late wake the Scheduler claims, once, the latest boundary of the caught-up range that the job's schedule matches. RunOnce, Supersede and RunNext run on the instance they are called on and are never claimed; with a Leaser they, and claimed runs too, also hold the lease of their id (see Leaser).
During shutdown the claim is released when the job returns and its release finishes within the shutdown timeout, and the process waits for Start to return.
type Lease ¶ added in v1.36.0
type Lease struct {
ID string
Owner string
// Superseded is set in Renew's answer when an Acquire with supersede
// marked the lease.
Superseded bool
}
Lease names one acquisition of an id. Owner is unique per acquisition.
type Leaser ¶ added in v1.36.0
type Leaser interface {
// Acquire takes id for owner when no lease on id is live, or when the live
// lease is already owner's (a retry of an acquire whose answer was lost;
// its superseded mark is kept). A fresh take clears the mark. When the
// acquire is refused and supersede is true, it marks the live lease
// superseded in the same step. ttl is measured on the store's clock.
Acquire(ctx context.Context, id, owner string, ttl time.Duration, supersede bool) (bool, error)
// Renew extends each lease that is still its owner's by ttl from the
// store's clock and returns those, with Superseded set on the marked
// ones. A lease missing from the result is lost.
Renew(ctx context.Context, leases []Lease, ttl time.Duration) ([]Lease, error)
// Release ends owner's lease on id. It never ends another owner's.
Release(ctx context.Context, id, owner string) error
}
Leaser keeps, across the instances of an app, which run holds each keyed job id, so that at most one run of an id is going across all of them and Supersede on one instance cancels the run on another. Set it with WithLeaser. The claims subpackage implements it on PostgreSQL.
With a Leaser every invocation of the Scheduler, recurring ones included, holds the lease of its id while it runs. It acquires the lease after the recurring claim of a Claimer, if any, and before the permit of a WithLimit option. A RunOnce or recurring run whose acquire is refused or fails is skipped; a RunNext or Supersede run retries until it wins, its context ends or its job timeout plus the lease's TTL passes, and a Supersede run asks to supersede on every attempt. While a run holds its lease the Scheduler renews it every 2 seconds for 30 seconds, and cancels the run with ErrSuperseded when a renewal answers that the lease was marked superseded, and with ErrLeaseLost when a renewal answers that the lease is no longer the run's or when no renewal succeeded for 20 seconds. The lease is released when the run returns, unless a pending job of the id queued on the same instance runs next on it.
type Option ¶ added in v1.6.0
type Option func(*Scheduler)
Option configures the Scheduler. Pass options to New.
func WithClaimer ¶ added in v1.33.0
WithClaimer makes recurring jobs run on one instance per minute boundary: before a due recurring job runs, the Scheduler claims the boundary from c and skips the run unless the claim is won (see Claimer). Run every instance in the same time zone. Every and Hourly compare only the minute of the hour in local time, so zones whose UTC offsets differ by whole hours agree, but zones whose offsets differ by a fraction of an hour (+5:30, +5:45, -3:30) select different instants for a schedule that fraction does not fit, such as Hourly or Every(time.Hour), and each instance then wins its own boundaries. Daily compares in its own location, the same on every instance unless it is time.Local.
The claim's lease is the job timeout plus one minute (capped at the largest Duration), counted from the claim, while the job timeout starts when the job body is about to run. A run outlives its lease, and can overlap a run of a later boundary on another instance, when the claim's answer, the lease's answer when a Leaser is set, the wait for a WithLimit permit and the time the job takes to return once its job timeout cancels it add up to more than that minute, when the job ignores its context, or when it runs under a Start ctx with a deadline, which replaces the job timeout. Nil => no claims, every instance runs every due boundary.
func WithJobTimeout ¶ added in v1.6.0
WithJobTimeout sets the per-job timeout applied when the job's ctx has no deadline. The timer starts when the job body is about to run, after any permit from WithLimit was acquired, so time spent waiting for a permit does not count. d <= 0 is ignored.
func WithLeaser ¶ added in v1.36.0
WithLeaser makes every run of a keyed id hold the id's lease from l while it runs (see Leaser), so that across the instances that share l's store at most one run of an id is going, and Supersede on one instance cancels the run on another. Set it on every instance, together with WithClaimer: the lease stops two runs of a recurring job from overlapping, but without a Claimer a boundary still runs on every instance that asks for the lease after the run before it returned. Nil leaves the Scheduler as without this option: ids exclude each other in this process only.
func WithLimit ¶ added in v1.12.0
WithLimit caps at n the number of jobs whose id starts with prefix that run at the same time, across RunOnce, Supersede, RunNext and recurring jobs. A job over the cap waits for a permit, still counts in Active, and gives up without running if its context is cancelled while waiting. New rejects an empty prefix, n < 1, a prefix given twice, and a prefix that is a prefix of another limit's prefix.
func WithLogger ¶ added in v1.6.0
WithLogger sets the logger. Nil => Noop logger.
func WithShutdownTimeout ¶ added in v1.6.0
WithShutdownTimeout sets how long Start waits for jobs to drain after its ctx is done. d <= 0 is ignored.
type Schedule ¶ added in v1.12.0
type Schedule struct {
// contains filtered or unexported fields
}
Schedule is an immutable description of the wall-clock minutes at which a recurring job is due. Build it with Every, Hourly or Daily. The zero value is invalid. A value can be reused for several ids.
func Daily ¶ added in v1.12.0
Daily returns a Schedule that is due once a day at hour:minute wall-clock time in loc. hour must be in 0..23, minute in 0..59 and loc non-nil. On a day where that local time does not exist (DST spring-forward) the schedule is not due; on a day where it occurs twice (DST fall-back) it is due at both instants. Invalid input yields a Schedule that Add rejects with an ez.EINVALID error.
func Every ¶ added in v1.12.0
Every returns a Schedule that is due at minute 0 of every hour and every d after it, in server-local time. d must be a whole number of minutes that divides one hour evenly (1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30 or 60 minutes). Every(time.Hour) is equivalent to Hourly(0). An invalid d yields a Schedule that Add rejects with an ez.EINVALID error.
func Hourly ¶ added in v1.12.0
Hourly returns a Schedule that is due at each of the given minutes of every hour, in server-local time. At least one minute is required, each must be in 0..59 and none may repeat. Order does not matter. Invalid input yields a Schedule that Add rejects with an ez.EINVALID error.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler runs recurring jobs on wall-clock minute boundaries and one-shot jobs on demand. Work is keyed by id: at most one invocation per id runs at a time, and each busy id holds at most one pending job that runs after the current one returns. Create it with New.
func New ¶
New creates a Scheduler with the given options. It returns an ez.EINVALID error when a WithLimit option has an empty prefix, a size below 1, a prefix given twice, or a prefix that is a prefix of another limit's prefix.
func (*Scheduler) Active ¶ added in v1.6.2
Active returns the number of ids with admitted work: running, waiting for a permit from a WithLimit option, claiming their boundary from a Claimer, or acquiring, waiting for or releasing their lease from a Leaser. A pending job adds nothing to the count.
func (*Scheduler) Add ¶
Add registers job to run at the minute boundaries selected by when. It returns an ez.EINVALID error for an empty id, a nil job or an invalid Schedule, and an ez.ECONFLICT error when a recurring job with the same id is already registered; on error nothing is registered. Add may be called before or after Start. The id is shared with RunOnce, Supersede and RunNext: a due boundary is skipped while the id is busy.
Daily schedules follow wall time in their location: on a DST spring-forward day a nonexistent local time does not fire, and on a fall-back day a repeated local time fires at both instants.
func (*Scheduler) Idled ¶ added in v1.6.2
func (s *Scheduler) Idled() <-chan struct{}
Idled returns a channel that receives a value when the last busy id finishes and Active drops to zero. It has a buffer of one and sends never block, so a receiver may see one value for several idle transitions. It does not fire between a job and the pending job that follows it. Check Active after receiving, since new work may already have been admitted.
func (*Scheduler) RunNext ¶ added in v1.12.0
RunNext starts job under id now when id is idle. When id is busy it replaces the id's pending job with job and returns true without cancelling the current invocation; job starts after the current invocation returns. It returns false when id is empty, job is nil, ctx is already done, or the Scheduler is stopping. A pending job whose ctx is done by then is dropped.
With a Leaser, true means job was accepted on this instance. While a run on another instance holds the lease of id, job waits for it, asking every second, and is skipped when its ctx ends first, the Scheduler stops, or its job timeout plus 30 seconds passes.
func (*Scheduler) RunOnce ¶
RunOnce starts job under id now. It returns false without running anything when id is empty, job is nil, ctx is already done, the Scheduler is stopping, or id is busy; a busy id's pending job is left untouched.
With a Leaser, true means job was accepted on this instance: the run asks for the lease of id once, and job is skipped when a run on another instance holds it or the acquire fails.
func (*Scheduler) Start ¶
Start runs the minute engine and blocks until ctx is done. It sleeps until the next wall-clock minute boundary, runs the recurring jobs due at it, and repeats. It never fires immediately: the first boundary evaluated is the first one after Start is called.
A late wake, because the process was frozen or the clock jumped forward, catches up: every minute boundary between the last one handled and the clock is evaluated, and a registration due at any of them is admitted once. A due job whose id is still busy is skipped for that wake. Nothing is remembered across restarts, so minutes that pass while the process is not running are never replayed. If the clock is stepped back by a minute or more the engine follows it and evaluates those minutes again as they pass.
When ctx is done Start stops the Scheduler: nothing new is admitted, pending jobs are discarded, running jobs are cancelled with context.Canceled, and Start waits for them up to the shutdown timeout before returning. Jobs that ignore their context may outlive that wait. Call Start once; once it has returned, a later call returns immediately.
func (*Scheduler) Supersede ¶ added in v1.12.0
Supersede starts job under id now when id is idle. When id is busy it replaces the id's pending job with job, cancels the current invocation with cause ErrSuperseded and returns true without waiting; job starts after the current invocation returns. It returns false when id is empty, job is nil, ctx is already done, or the Scheduler is stopping.
With a Leaser, true means job was accepted on this instance. Unless job runs on the lease this instance already holds for id, it asks for the lease of id every second until it gets it, and each refused attempt marks the lease superseded: the run holding a marked lease is cancelled with ErrSuperseded at its instance's next successful renewal of the lease, which that instance attempts every 2 seconds. So until job gets the lease or gives up, every run that holds the id is cancelled, also one that started after this call, such as the holder's own pending job or the job of another Supersede. Of two Supersede calls on different instances at the same time, one job runs to completion and the other is cancelled, runs after it or gives up, in no guaranteed order. job gives up, and is skipped, when its ctx ends first, the Scheduler stops, or its job timeout plus 30 seconds passes. A mark it left stays on the lease while the holder keeps it, so at the holder's next successful renewal it can cancel the run then holding the lease, even one queued after job gave up, with no job replacing it. In one process, likewise, a job queued as pending whose ctx is done by the time it would run is dropped after this call already cancelled the run.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package claims keeps the claims and the leases of a scheduler.Scheduler in PostgreSQL tables, so that an app running several instances runs each minute boundary of a recurring job on one of them (NewBunStore), and runs at most one run of a keyed job id at a time across them (NewBunLeaser).
|
Package claims keeps the claims and the leases of a scheduler.Scheduler in PostgreSQL tables, so that an app running several instances runs each minute boundary of a recurring job on one of them (NewBunStore), and runs at most one run of a keyed job id at a time across them (NewBunLeaser). |