scheduler

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 16, 2026 License: MIT Imports: 9 Imported by: 0

README

schedulerok

A minimal, Go-native scheduler that owns its timer loop end to end and lets schedules change their own mind.

Status: early development. The API is not stable yet.

About

schedulerok separates two concerns that most schedulers blur together: a Job only knows how to run, and a Schedule only knows how to calculate its next execution time. The scheduler owns everything in between — startup, timing, cancellation, and graceful shutdown — and reacts to either side changing, including a schedule that redefines itself after every run.

s := scheduler.New()

_, err := s.AddIntervalFunc(5*time.Minute, heartbeat)
if err != nil {
	return err
}

return s.Run(ctx)

Configuration loading stays outside the library. Applications translate their own YAML, environment variables, or database rows into Add calls with stable IDs; schedulerok never parses configuration itself.

Registering jobs

Add and AddFunc take any Schedule. AddIntervalJob/AddIntervalFunc and AddCronJob/AddCronFunc build the schedule for you. An AdaptiveJob can be passed through the same registration methods; the scheduler detects it and asks for its next schedule after each execution:

// Generic: bring your own Schedule.
s.Add(schedule, job)
s.AddFunc(schedule, func(ctx context.Context) error { return nil })

// Fixed interval and cron, job or plain function.
s.AddIntervalJob(time.Minute, job)
s.AddIntervalFunc(time.Minute, func(ctx context.Context) error { return nil })
s.AddCronJob("*/5 * * * *", job)
s.AddCronFunc("*/5 * * * *", func(ctx context.Context) error { return nil })

// Adaptive: use the regular registration methods. The scheduler detects
// AdaptiveJob and asks it for the next Schedule after Run returns.
s.Add(schedule, adaptiveJob)
s.AddIntervalJob(time.Minute, adaptiveJob)
s.AddCronJob("*/5 * * * *", adaptiveJob)

Adaptive schedules

Not every job runs on a fixed cadence. A poll loop driven by a server response — a chat API's PollingIntervalMillis, a rate-limited endpoint's retry hint — needs to pick its own next execution after it runs, not before. AdaptiveJob covers that case without changing the plain Job contract:

type pollJob struct {
	delay time.Duration
}

func (j *pollJob) Run(ctx context.Context) error {
	delay, err := poll(ctx)
	if err != nil {
		return err
	}

	j.delay = delay
	return nil
}

func (j *pollJob) NextSchedule(current scheduler.Schedule) (scheduler.Schedule, error) {
	if j.delay <= 0 {
		return nil, nil // keep the current schedule
	}

	return scheduler.NewIntervalSchedule(j.delay)
}

job := &pollJob{}
schedule, err := scheduler.NewIntervalSchedule(2 * time.Second)
if err != nil {
	return err
}

_, err = s.Add(schedule, job)

The scheduler calls Run(ctx) and then NextSchedule(current). Returning nil, nil keeps the current schedule; returning a schedule replaces it for the next execution. An error reports that the replacement could not be calculated. The current argument is available to stateful schedules that need to inspect the existing schedule before producing a replacement.

A plain Job passed to Add is wrapped internally, so both job kinds use the same registration API. For function-based jobs, AddAdaptiveFunc accepts the execution and schedule-selection functions separately:

_, err := s.AddAdaptiveFunc(
	schedule,
	func(ctx context.Context) error {
		return poll(ctx)
	},
	func(current scheduler.Schedule) (scheduler.Schedule, error) {
		return scheduler.NewIntervalSchedule(5 * time.Minute)
	},
)

If the schedule-selection function is nil, AddAdaptiveFunc delegates to AddFunc and keeps the schedule fixed.

Lifecycle, policies, and runtime control

Per-registration policies cover timeout, retry with isolated backoff state, and overlap behavior. Lifecycle hooks fire around each attempt without requiring a logging dependency in the core:

s.AddIntervalFunc(time.Minute, job,
	scheduler.WithTimeout(10*time.Second),
	scheduler.WithRetry(3, backoffFactory),
	scheduler.WithOverlap(scheduler.SkipOverlap),
	scheduler.WithHooks(scheduler.Hooks{
		OnFailure: func(_ context.Context, e scheduler.Event) {
			log.Printf("%s failed: %v", e.JobID, e.Error)
		},
	}),
)

Add and Remove both work while Run is already executing — a job can be registered or pulled out mid-flight, and the central loop wakes up to react immediately instead of waiting for its next tick.

Pause and Resume control dispatch without destroying the scheduler lifecycle. While paused, the loop remains alive but does not run jobs or call AdaptiveJob.NextSchedule; the latest schedule is preserved until resume.

A registration whose Schedule stops advancing does not take the rest of the scheduler down with it. It freezes in place, fires OnFailure, and stays out of consideration until Remove is called explicitly; FrozenIDs() reports which registrations are stuck.

Scheduler-level observability is separate from per-job hooks. Configure WithSchedulerHooks to receive lifecycle callbacks and optional tick events with the tick time, the JobIDs that were due, and the JobIDs actually dispatched. With tick observation disabled, the scheduler does not allocate tick events or job ID slices, keeping the normal path close to zero overhead. Callbacks must be fast and non-blocking.

Example

Run the basic interval scheduler and stop it with Ctrl+C:

go run ./examples/basic

The example demonstrates interval registration, a stable job ID, lifecycle hooks, and graceful shutdown. Output identifies its source:

[hook.OnStart] heartbeat started (attempt 1)
[job.Run] heartbeat
[hook.OnSuccess] heartbeat completed

Requirements

  • Go 1.25 or later

Support

schedulerok is one of Candango Open Source Group initiatives. It is available under the MIT License.

Documentation

Overview

Package scheduler provides primitives for defining and running scheduled jobs.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrSchedulerRunning indicates that Run was called while the scheduler is active.
	ErrSchedulerRunning = errors.New("scheduler is already running")
	// ErrNilSchedule indicates that a registration has no schedule.
	ErrNilSchedule = errors.New("scheduler schedule must not be nil")
	// ErrNilJob indicates that a registration has no job.
	ErrNilJob = errors.New("scheduler job must not be nil")
	// ErrInvalidSchedule indicates that a schedule did not advance in time.
	ErrInvalidSchedule = errors.New("scheduler schedule must return a future time")
	// ErrNilContext indicates that Run received a nil context.
	ErrNilContext = errors.New("scheduler context must not be nil")
	// ErrUnknownJob indicates that Remove did not find the requested registration.
	ErrUnknownJob = errors.New("scheduler job ID was not found")
)
View Source 
var ErrInvalidInterval = errors.New("interval must be greater than zero")

ErrInvalidInterval indicates that an interval is not positive.

Functions

This section is empty.

Types

type AdaptiveJob added in v0.1.0 

type AdaptiveJob interface {
	Job
	NextSchedule(current Schedule) (Schedule, error)
}

AdaptiveJob is a Job that may provide a replacement Schedule after each execution.

The scheduler calls Run and then NextSchedule with the current Schedule. A nil Schedule preserves the current schedule. A non-nil Schedule replaces it for the next execution. An error means that the replacement could not be calculated.

type BackoffFactory 

type BackoffFactory func() backoff.Backoff

BackoffFactory creates isolated backoff state for one job execution.

type Clock 

type Clock interface {
	Now() time.Time
	NewTimer(time.Duration) Timer
}

Clock provides the current time and creates timers for scheduled work.

Implementations let the scheduler control time deterministically in tests.

type Event 

type Event struct {
	JobID      JobID
	Attempt    int
	Error      error
	RetryDelay time.Duration
}

Event describes one scheduler lifecycle event.

type Hooks 

type Hooks struct {
	OnStart   func(context.Context, Event)
	OnSuccess func(context.Context, Event)
	OnFailure func(context.Context, Event)
	OnRetry   func(context.Context, Event)
	OnSkip    func(context.Context, Event)
}

Hooks receives lifecycle events for one job registration. Hook functions run synchronously and must return promptly.

type IntervalSchedule 

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

IntervalSchedule calculates execution times separated by a fixed interval.

func NewIntervalSchedule 

func NewIntervalSchedule(interval time.Duration) (IntervalSchedule, error)

NewIntervalSchedule returns a schedule with a positive fixed interval.

func (IntervalSchedule) Next 

func (s IntervalSchedule) Next(after time.Time) time.Time

Next returns the time one interval after after.

type Job 

type Job interface {
	Run(context.Context) error
}

Job performs one unit of scheduled work.

Run receives a context that is canceled when the scheduler stops. It returns a non-nil error when the scheduled work fails.

type JobFunc 

type JobFunc func(context.Context) error

JobFunc adapts a function to Job.

func (JobFunc) Run 

func (fn JobFunc) Run(ctx context.Context) error

Run executes fn.

type JobID 

type JobID string

JobID identifies one job registration in a Scheduler.

type NextScheduleFunc added in v0.2.0 

type NextScheduleFunc func(Schedule) (Schedule, error)

NextScheduleFunc adapts a function to the schedule-selection part of an AdaptiveJob.

type Option 

type Option func(*Scheduler)

Option configures a Scheduler during construction.

func WithClock 

func WithClock(clock Clock) Option

WithClock configures the clock used by the scheduler and its timers.

The clock must be set before the scheduler starts. Passing a nil clock is a programmer error.

func WithSchedulerHooks added in v0.2.0 

func WithSchedulerHooks(hooks SchedulerHooks) Option

WithSchedulerHooks attaches lifecycle hooks to the scheduler.

type OverlapPolicy 

type OverlapPolicy uint8

OverlapPolicy controls what happens when a job is due while it is running. 

const (
	// AllowOverlap starts every due run, even if an earlier run is still active.
	AllowOverlap OverlapPolicy = iota
	// SkipOverlap discards a due run while an earlier run is still active.
	SkipOverlap
)

type RegistrationOption 

type RegistrationOption func(*registrationOptions)

RegistrationOption configures one registered job.

func WithHooks 

func WithHooks(hooks Hooks) RegistrationOption

WithHooks attaches lifecycle hooks to one job registration.

func WithID 

func WithID(id JobID) RegistrationOption

WithID assigns a stable ID to one registered job.

func WithOverlap 

func WithOverlap(policy OverlapPolicy) RegistrationOption

WithOverlap configures overlap behavior for one job registration.

func WithRetry 

func WithRetry(attempts int, factory BackoffFactory) RegistrationOption

WithRetry retries a failed job up to attempts total attempts. factory creates a new backoff for each scheduled execution.

func WithTimeout 

func WithTimeout(timeout time.Duration) RegistrationOption

WithTimeout limits each job attempt to timeout.

type Schedule 

type Schedule interface {
	Next(after time.Time) time.Time
}

Schedule calculates the next time to run after a reference time.

Callers should pass the previous scheduled time so interval schedules do not drift when job execution takes longer than expected.

func NewCronSchedule 

func NewCronSchedule(expr string) (Schedule, error)

NewCronSchedule parses expr with intervalok and returns it as a Schedule.

type ScheduleFunc 

type ScheduleFunc func(time.Time) time.Time

ScheduleFunc adapts a function to Schedule.

func (ScheduleFunc) Next 

func (fn ScheduleFunc) Next(after time.Time) time.Time

Next returns the next time calculated by fn.

type Scheduler 

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

Scheduler coordinates registered jobs and schedules.

func New 

func New(options ...Option) *Scheduler

New creates a Scheduler with production defaults.

func (*Scheduler) Add 

func (s *Scheduler) Add(schedule Schedule, job Job, options ...RegistrationOption) (JobID, error)

Add registers a Job to run according to a Schedule. It may be called before or after Run starts; a registration added while running is scheduled from the current time and wakes the scheduler's timer loop.

func (*Scheduler) AddAdaptiveFunc added in v0.1.0 

func (s *Scheduler) AddAdaptiveFunc(
	schedule Schedule,
	run JobFunc,
	next NextScheduleFunc,
	options ...RegistrationOption,
) (JobID, error)

AddAdaptiveFunc adapts separate execution and schedule-selection functions to AdaptiveJob and registers them according to schedule. A nil next function registers run as a regular Job and preserves the current schedule.

func (*Scheduler) AddCronFunc 

func (s *Scheduler) AddCronFunc(
	spec string,
	fn JobFunc,
	options ...RegistrationOption,
) (JobID, error)

AddCronFunc parses spec as an intervalok cron schedule and registers fn.

func (*Scheduler) AddCronJob 

func (s *Scheduler) AddCronJob(
	spec string,
	job Job,
	options ...RegistrationOption,
) (JobID, error)

AddCronJob parses spec as an intervalok cron schedule and registers job.

func (*Scheduler) AddFunc 

func (s *Scheduler) AddFunc(
	schedule Schedule,
	fn JobFunc,
	options ...RegistrationOption,
) (JobID, error)

AddFunc adapts fn to Job and registers it according to schedule.

func (*Scheduler) AddIntervalFunc 

func (s *Scheduler) AddIntervalFunc(
	interval time.Duration,
	fn JobFunc,
	options ...RegistrationOption,
) (JobID, error)

AddIntervalFunc creates a fixed interval schedule and registers fn.

func (*Scheduler) AddIntervalJob 

func (s *Scheduler) AddIntervalJob(
	interval time.Duration,
	job Job,
	options ...RegistrationOption,
) (JobID, error)

AddIntervalJob creates a fixed interval schedule and registers job.

func (*Scheduler) FrozenIDs added in v0.1.0 

func (s *Scheduler) FrozenIDs() []JobID

FrozenIDs returns the IDs of registrations whose Schedule stopped advancing. A frozen registration stays registered and excluded from due consideration until Remove is called explicitly.

func (*Scheduler) Pause added in v0.2.0 

func (s *Scheduler) Pause()

Stop gracefully stops the scheduler. New jobs are not dispatched, active jobs are allowed to finish, and the scheduler can be started again with Run. The context controls how long the caller waits for shutdown. Pause keeps the scheduler loop alive while preventing job dispatches. Existing schedules are preserved and Resume re-evaluates them.

func (*Scheduler) Remove 

func (s *Scheduler) Remove(id JobID) error

Remove stops future runs for id. A job that is already running is not interrupted and remains subject to normal shutdown handling.

func (*Scheduler) Resume added in v0.2.0 

func (s *Scheduler) Resume()

Resume allows dispatches again. A schedule that became due while paused is dispatched once, then its normal scheduling policy continues.

func (*Scheduler) Run 

func (s *Scheduler) Run(ctx context.Context) error

Run starts the scheduler and blocks until ctx is cancelled or Stop is called. Context cancellation is propagated to active jobs; Stop waits for active jobs without cancelling them.

func (*Scheduler) Stop added in v0.2.0 

func (s *Scheduler) Stop(ctx context.Context) error

Stop gracefully stops the scheduler. New jobs are not dispatched, active jobs are allowed to finish, and the scheduler can be started again with Run. The context controls how long the caller waits for shutdown.

type SchedulerHooks added in v0.2.0 

type SchedulerHooks struct {
	OnStart    func(context.Context)
	OnStopping func(context.Context)
	OnStopped  func(context.Context)
	OnTick     func(context.Context, TickEvent)
}

SchedulerHooks receives lifecycle events for the scheduler itself. Hook functions run synchronously and must return promptly.

type TickEvent added in v0.2.0 

type TickEvent struct {
	At         time.Time
	DueJobs    []JobID
	Dispatched []JobID
}

TickEvent describes one scheduler timer tick.

type Timer 

type Timer interface {
	Chan() <-chan time.Time
	Stop()
}

Timer waits until a scheduled time.

Chan receives the time at which the timer fires. Stop releases the timer's resources and may be called after the timer has fired.

Source Files

View all Source files

Directories

Path Synopsis
Package clocktest provides deterministic clocks for scheduler tests.
 Package clocktest provides deterministic clocks for scheduler tests.
examples
adaptive command
basic command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.