gorun

README

gorun - A distributed job scheduling system

I am considering preparing this code for a more general release. At this point the code is completely subject to change and not yet ready for general use.

With that caveat, a variant of this code has worked for me in a production environment, and anyone is free to fork this repo to use as a starting place.

Documentation

Index

• Constants
• Variables
• func MarkIncompleteJobsHandler(ctx context.Context, args *MarkIncompleteJobs) (string, error)
• func ProcessTriggersHandler(ctx context.Context, args *ProcessTriggers) (string, error)
• func RegisterHandler[T JobData](h Handler[T])
• type GoRunService
  • func New(db *sql.DB, opts ...Option) (GoRunService, error)
  • func NewFromEnv(opts ...Option) (GoRunService, error)
• type Handler
• type JobData
• type MarkIncompleteJobs
  • func (a MarkIncompleteJobs) JobType() string
• type Option
  • func DisableLogging() Option
  • func OnJobComplete(...) Option
  • func OnJobInit(f func(ctx context.Context, jobType string, jobId string) context.Context) Option
  • func WithArgProcessor(f func(ctx context.Context, jobType string, jobId string, args any) error) Option
  • func WithBatchFreq(freq time.Duration) Option
  • func WithBatchSize(size int) Option
  • func WithJobTimeout(jobTimeout time.Duration) Option
• type ProcessTriggers
  • func (a ProcessTriggers) JobType() string
• type Trigger
• type Validateable

Constants

const (
	StatusScheduled = "scheduled"
	StatusRunning   = "running"
	StatusCompleted = "completed"
	StatusFailed    = "failed"
)

Variables

var ErrGorunInternalError = errors.Sentinel("internal gorun job service error")

var ErrUnregisteredJobType = errors.Sentinel("unregistered job type")

Functions

func MarkIncompleteJobsHandler

func MarkIncompleteJobsHandler(ctx context.Context, args *MarkIncompleteJobs) (string, error)

func ProcessTriggersHandler

func ProcessTriggersHandler(ctx context.Context, args *ProcessTriggers) (string, error)

func RegisterHandler

func RegisterHandler[T JobData](h Handler[T])

Types

type GoRunService

type GoRunService interface {
	ScheduleImmediately(ctx context.Context, job JobData) (jobId string, err error)
	ScheduleAfter(ctx context.Context, delay time.Duration, job JobData) (jobId string, err error)
	ScheduleCron(ctx context.Context, cronExpr string, loc *time.Location, job JobData) (triggerId string, err error)
	ScheduleRepeated(ctx context.Context, interval time.Duration, job JobData) (triggerId string, err error)

	ScheduleCronWithKey(ctx context.Context, triggerId string, cronExpr string, loc *time.Location, job JobData) error
	ScheduleRepeatedWithKey(ctx context.Context, triggerId string, interval time.Duration, job JobData) error

	GetJob(ctx context.Context, jobId string) (*gorundb.JobData, error)
	ListJobs(ctx context.Context, start, end time.Time) ([]*gorundb.JobData, error)
	ListTriggers(ctx context.Context) ([]*gorundb.JobTrigger, error)
	DeleteTrigger(ctx context.Context, triggerId string) error

	Start(ctx context.Context) error
	Close()
}

func New

func New(db *sql.DB, opts ...Option) (GoRunService, error)

func NewFromEnv

func NewFromEnv(opts ...Option) (GoRunService, error)

type Handler

type Handler[T JobData] func(ctx context.Context, args *T) (string, error)

type JobData

type JobData interface {
	// Return a constant name for the job that will process the request - this is used to uniquely identify the job handler
	JobType() string
}

type MarkIncompleteJobs

type MarkIncompleteJobs struct{}

func (MarkIncompleteJobs) JobType

func (a MarkIncompleteJobs) JobType() string

type Option

type Option func(*options)

func DisableLogging

func DisableLogging() Option

func OnJobComplete

func OnJobComplete(f func(ctx context.Context, jobType string, jobId string, result string, err error)) Option

func OnJobInit

func OnJobInit(f func(ctx context.Context, jobType string, jobId string) context.Context) Option

func WithArgProcessor

func WithArgProcessor(f func(ctx context.Context, jobType string, jobId string, args any) error) Option

func WithBatchFreq

func WithBatchFreq(freq time.Duration) Option

How often each job server will check for new jobs to run

func WithBatchSize

func WithBatchSize(size int) Option

How many new jobs to run in each batch. Currently there is no limit on the number of concurrent jobs.

func WithJobTimeout

func WithJobTimeout(jobTimeout time.Duration) Option

The maximum amount of time a job can run before it is considered to have timed out

type ProcessTriggers

type ProcessTriggers struct{}

func (ProcessTriggers) JobType

func (a ProcessTriggers) JobType() string

type Trigger

type Trigger = triggers.Trigger

type Validateable

type Validateable interface {
	Validate() error
}