jobs

Documentation

Index

• Variables
• type BackoffFunc
  • func ExponentialBackoff(base, max time.Duration) BackoffFunc
  • func FixedBackoff(delay time.Duration) BackoffFunc
• type Config
• type CronFunc
• type EnqueueConfig
• type Event
• type EventHandler
• type EventType
• type HandlerFunc
• type Job
  • func (j Job) Decode(v any) error
• type Queue
  • func New() *Queue
  • func NewWithConfig(config Config) *Queue
  • func (q *Queue) Cron(name, spec string, handler CronFunc, config ...ScheduleConfig) (Schedule, error)
  • func (q *Queue) Enqueue(ctx context.Context, name string, payload any, config ...EnqueueConfig) (Job, error)
  • func (q *Queue) EnqueueIn(ctx context.Context, name string, payload any, delay time.Duration, ...) (Job, error)
  • func (q *Queue) Failed() []Job
  • func (q *Queue) Handle(name string, handler HandlerFunc) error
  • func (q *Queue) Pending() int
  • func (q *Queue) Schedule(name, spec string, payload any, config ...ScheduleConfig) (Schedule, error)
  • func (q *Queue) Start(ctx context.Context, workers int) (*Runner, error)
  • func (q *Queue) Unschedule(id string) error
• type Runner
  • func (r *Runner) Stop(ctx context.Context) error
  • func (r *Runner) Wait()
• type Schedule
• type ScheduleConfig
• type ScheduleSpec
  • func ParseSchedule(spec string) (ScheduleSpec, error)

Variables

var (
	ErrJobNameRequired      = errors.New("zincjobs: job name is required")
	ErrHandlerRequired      = errors.New("zincjobs: handler is required")
	ErrHandlerExists        = errors.New("zincjobs: handler already registered")
	ErrHandlerNotFound      = errors.New("zincjobs: handler not found")
	ErrInvalidWorkerCount   = errors.New("zincjobs: worker count must be greater than zero")
	ErrInvalidMaxAttempts   = errors.New("zincjobs: max attempts must be greater than zero")
	ErrInvalidSchedule      = errors.New("zincjobs: invalid schedule")
	ErrQueueAlreadyRunning  = errors.New("zincjobs: queue is already running")
	ErrScheduleExists       = errors.New("zincjobs: schedule already registered")
	ErrScheduleNotFound     = errors.New("zincjobs: schedule not found")
	ErrScheduleIDRequired   = errors.New("zincjobs: schedule id is required")
	ErrScheduleNameRequired = errors.New("zincjobs: schedule name is required")
)

var DefaultConfig = Config{
	DefaultQueue:       "default",
	DefaultMaxAttempts: 1,
	Backoff:            FixedBackoff(time.Second),
	Now:                time.Now,
}

DefaultConfig is used by New.

Types

type BackoffFunc

type BackoffFunc func(attempt int, err error) time.Duration

BackoffFunc returns the delay before the next attempt after a failed job.

func ExponentialBackoff

func ExponentialBackoff(base, max time.Duration) BackoffFunc

ExponentialBackoff returns a backoff function that doubles from base up to max.

func FixedBackoff

func FixedBackoff(delay time.Duration) BackoffFunc

FixedBackoff returns a backoff function that always uses delay.

type Config

type Config struct {
	DefaultQueue       string
	DefaultMaxAttempts int
	Backoff            BackoffFunc
	EventHandler       EventHandler
	Now                func() time.Time
}

Config controls queue defaults.

type CronFunc

type CronFunc func(ctx context.Context) error

CronFunc processes a scheduled function job.

type EnqueueConfig

type EnqueueConfig struct {
	ID          string
	Queue       string
	RunAt       time.Time
	MaxAttempts int
}

EnqueueConfig customizes a single enqueued job.

type Event

type Event struct {
	Type      EventType
	Job       Job
	Error     error
	NextRunAt time.Time
}

Event describes a job lifecycle transition.

type EventHandler

type EventHandler func(Event)

EventHandler observes job lifecycle events emitted by workers and schedules.

type EventType

type EventType string

EventType identifies a job lifecycle event.

const (
	EventEnqueued  EventType = "enqueued"
	EventStarted   EventType = "started"
	EventCompleted EventType = "completed"
	EventRetrying  EventType = "retrying"
	EventFailed    EventType = "failed"
	EventScheduled EventType = "scheduled"
)

type HandlerFunc

type HandlerFunc func(ctx context.Context, job Job) error

HandlerFunc processes a job attempt.

type Job

type Job struct {
	ID          string
	Name        string
	Queue       string
	Payload     []byte
	CreatedAt   time.Time
	RunAt       time.Time
	Attempts    int
	MaxAttempts int
	LastError   string
}

Job is the immutable job snapshot passed to handlers.

func (Job) Decode

func (j Job) Decode(v any) error

Decode unmarshals the JSON job payload into v.

type Queue

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

Queue stores jobs, handlers, and schedules for a worker pool.

func New

func New() *Queue

New creates a queue with DefaultConfig.

func NewWithConfig

func NewWithConfig(config Config) *Queue

NewWithConfig creates a queue with explicit defaults.

func (*Queue) Cron

func (q *Queue) Cron(name, spec string, handler CronFunc, config ...ScheduleConfig) (Schedule, error)

Cron registers a named scheduled function.

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, name string, payload any, config ...EnqueueConfig) (Job, error)

Enqueue adds a job to the queue.

func (*Queue) EnqueueIn

func (q *Queue) EnqueueIn(ctx context.Context, name string, payload any, delay time.Duration, config ...EnqueueConfig) (Job, error)

EnqueueIn adds a job that becomes runnable after delay.

func (*Queue) Failed

func (q *Queue) Failed() []Job

Failed returns failed jobs that exhausted their attempts.

func (*Queue) Handle

func (q *Queue) Handle(name string, handler HandlerFunc) error

Handle registers a handler for a named job.

func (*Queue) Pending

func (q *Queue) Pending() int

Pending returns the number of jobs waiting to run.

func (*Queue) Schedule

func (q *Queue) Schedule(name, spec string, payload any, config ...ScheduleConfig) (Schedule, error)

Schedule registers a recurring job using a cron expression or @every duration.

func (*Queue) Start

func (q *Queue) Start(ctx context.Context, workers int) (*Runner, error)

Start launches workers and the scheduler until the context is canceled.

func (*Queue) Unschedule

func (q *Queue) Unschedule(id string) error

Unschedule removes a recurring job by id.

type Runner

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

Runner controls a running queue.

func (*Runner) Stop

func (r *Runner) Stop(ctx context.Context) error

Stop cancels the runner and waits for workers to exit.

func (*Runner) Wait

func (r *Runner) Wait()

Wait blocks until the runner exits.

type Schedule

type Schedule struct {
	ID          string
	Name        string
	Spec        string
	Queue       string
	NextRunAt   time.Time
	MaxAttempts int
}

Schedule describes a registered recurring job.

type ScheduleConfig

type ScheduleConfig struct {
	ID          string
	Queue       string
	MaxAttempts int
}

ScheduleConfig customizes jobs created by a schedule.

type ScheduleSpec

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

ScheduleSpec calculates the next time a recurring job should run.

func ParseSchedule

func ParseSchedule(spec string) (ScheduleSpec, error)

ParseSchedule parses a five-field cron expression or aliases such as @every.