scheduler

Documentation

Overview

Package scheduler provides a background task scheduler that allows for the registration, execution, and management of periodic tasks. Tasks can be grouped, named, and configured with various options such as custom intervals, log fields, and skip conditions.

The scheduler supports dynamic interval updates and can gracefully stop tasks either individually or by group.

Index

• Constants
• func BlockDelay(flags observertypes.OperationalFlags, block cometbft.EventDataNewBlock) time.Duration
• func BlockFromContext(ctx context.Context) (cometbft.EventDataNewBlock, bool)
• func BlockFromContextWithDelay(ctx context.Context) (cometbft.EventDataNewBlock, time.Duration, error)
• type Executable
• type Group
• type Opt
  • func BlockTicker(blocks <-chan cometbft.EventDataNewBlock) Opt
  • func GroupName(group Group) Opt
  • func Interval(interval time.Duration) Opt
  • func IntervalUpdater(intervalUpdater func() time.Duration) Opt
  • func LogFields(fields map[string]any) Opt
  • func Name(name string) Opt
  • func Skipper(skipper func() bool) Opt
• type Scheduler
  • func New(logger zerolog.Logger) *Scheduler
  • func (s *Scheduler) Register(ctx context.Context, exec Executable, opts ...Opt) *Task
  • func (s *Scheduler) Stop()
  • func (s *Scheduler) StopGroup(group Group)
  • func (s *Scheduler) Tasks() map[uuid.UUID]*Task
• type Task
  • func (t *Task) Group() Group
  • func (t *Task) Name() string
  • func (t *Task) Stop()

Constants

const DefaultGroup = Group("default")

DefaultGroup is the default task group.

Functions

func BlockDelay

func BlockDelay(flags observertypes.OperationalFlags, block cometbft.EventDataNewBlock) time.Duration

BlockDelay calculates block sleep delay based on a given operational flags and a block. Sleep duration represents artificial "lag" before processing outbound transactions.

Use-case: coordinate outbound signatures between different observer-signers that might be located in different regions (e.g. Alice is in EU, Bob is in US)

func BlockFromContext

func BlockFromContext(ctx context.Context) (cometbft.EventDataNewBlock, bool)

BlockFromContext returns cometbft.EventDataNewBlock from the context or false.

func BlockFromContextWithDelay

func BlockFromContextWithDelay(ctx context.Context) (cometbft.EventDataNewBlock, time.Duration, error)

BlockFromContextWithDelay a combination of BlockFromContext and BlockDelay

Types

type Executable

type Executable func(ctx context.Context) error

Executable arbitrary function that can be executed.

type Group

type Group string

Group represents Task group. Tasks can be grouped for easier management.

type Opt

type Opt func(task *Task, taskOpts *taskOpts)

Opt Task option

func BlockTicker

func BlockTicker(blocks <-chan cometbft.EventDataNewBlock) Opt

BlockTicker makes Task to listen for new zeta blocks instead of using interval ticker. IntervalUpdater is ignored.

func GroupName

func GroupName(group Group) Opt

GroupName sets task group. Otherwise, defaults to DefaultGroup.

func Interval

func Interval(interval time.Duration) Opt

Interval sets initial task interval.

func IntervalUpdater

func IntervalUpdater(intervalUpdater func() time.Duration) Opt

IntervalUpdater sets interval updater function. Overrides Interval.

func LogFields

func LogFields(fields map[string]any) Opt

LogFields augments Task's logger with some fields.

func Name

func Name(name string) Opt

Name sets task name.

func Skipper

func Skipper(skipper func() bool) Opt

Skipper sets task skipper function. If it returns true, the task is skipped.

type Scheduler

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

Scheduler represents background task scheduler.

func New

func New(logger zerolog.Logger) *Scheduler

New Scheduler instance.

func (*Scheduler) Register

func (s *Scheduler) Register(ctx context.Context, exec Executable, opts ...Opt) *Task

Register registers and starts new Task in the background

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop stops all tasks.

func (*Scheduler) StopGroup

func (s *Scheduler) StopGroup(group Group)

StopGroup stops all tasks in the group.

func (*Scheduler) Tasks

func (s *Scheduler) Tasks() map[uuid.UUID]*Task

type Task

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

Task represents scheduler's task.

func (*Task) Group

func (t *Task) Group() Group

func (*Task) Name

func (t *Task) Name() string

func (*Task) Stop

func (t *Task) Stop()

Stop stops the task and offloads it from the scheduler.