scheduler

Documentation

Overview

Package scheduler defines the scheduling interface and provides implementations for GopherQueue.

Index

• func CalculateBackoff(attempt int, strategy core.BackoffStrategy, initial, max time.Duration, ...) time.Duration
• type PriorityScheduler
  • func NewPriorityScheduler(store persistence.JobStore, config *SchedulerConfig) *PriorityScheduler
  • func (s *PriorityScheduler) Cancel(ctx context.Context, id uuid.UUID, reason string, force bool) error
  • func (s *PriorityScheduler) Complete(ctx context.Context, id uuid.UUID, result *core.JobResult) error
  • func (s *PriorityScheduler) Enqueue(ctx context.Context, job *core.Job) error
  • func (s *PriorityScheduler) Fail(ctx context.Context, id uuid.UUID, result *core.JobResult) error
  • func (s *PriorityScheduler) GetNextJob(ctx context.Context) (*core.Job, error)
  • func (s *PriorityScheduler) IsHealthy() bool
  • func (s *PriorityScheduler) Retry(ctx context.Context, id uuid.UUID, resetAttempts bool) error
  • func (s *PriorityScheduler) Schedule(ctx context.Context, id uuid.UUID) error
  • func (s *PriorityScheduler) Start(ctx context.Context) error
  • func (s *PriorityScheduler) Stats() *SchedulerStats
  • func (s *PriorityScheduler) Stop(ctx context.Context) error
• type Scheduler
• type SchedulerConfig
  • func DefaultSchedulerConfig() *SchedulerConfig
• type SchedulerStats

Functions

func CalculateBackoff

func CalculateBackoff(attempt int, strategy core.BackoffStrategy, initial, max time.Duration, multiplier float64) time.Duration

CalculateBackoff calculates the retry delay based on the backoff strategy.

Types

type PriorityScheduler

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

PriorityScheduler is a full-featured scheduler implementation with priority queue.

func NewPriorityScheduler

func NewPriorityScheduler(store persistence.JobStore, config *SchedulerConfig) *PriorityScheduler

NewPriorityScheduler creates a new priority-based scheduler.

func (*PriorityScheduler) Cancel

func (s *PriorityScheduler) Cancel(ctx context.Context, id uuid.UUID, reason string, force bool) error

Cancel requests cancellation of a job.

func (*PriorityScheduler) Complete

func (s *PriorityScheduler) Complete(ctx context.Context, id uuid.UUID, result *core.JobResult) error

Complete marks a job as successfully completed.

func (*PriorityScheduler) Enqueue

func (s *PriorityScheduler) Enqueue(ctx context.Context, job *core.Job) error

Enqueue adds a new job to the scheduler.

func (*PriorityScheduler) Fail

func (s *PriorityScheduler) Fail(ctx context.Context, id uuid.UUID, result *core.JobResult) error

Fail marks a job as failed and determines retry behavior.

func (*PriorityScheduler) GetNextJob

func (s *PriorityScheduler) GetNextJob(ctx context.Context) (*core.Job, error)

GetNextJob returns the next job to execute.

func (*PriorityScheduler) IsHealthy

func (s *PriorityScheduler) IsHealthy() bool

IsHealthy returns whether the scheduler is operating normally.

func (*PriorityScheduler) Retry

func (s *PriorityScheduler) Retry(ctx context.Context, id uuid.UUID, resetAttempts bool) error

Retry reschedules a failed or dead-lettered job.

func (*PriorityScheduler) Schedule

func (s *PriorityScheduler) Schedule(ctx context.Context, id uuid.UUID) error

Schedule moves a delayed job to scheduled state.

func (*PriorityScheduler) Start

func (s *PriorityScheduler) Start(ctx context.Context) error

Start begins the scheduler's main loop.

func (*PriorityScheduler) Stats

func (s *PriorityScheduler) Stats() *SchedulerStats

Stats returns current scheduler statistics.

func (*PriorityScheduler) Stop

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

Stop gracefully shuts down the scheduler.

type Scheduler

type Scheduler interface {
	// Start begins the scheduler's main loop.
	Start(ctx context.Context) error

	// Stop gracefully shuts down the scheduler.
	Stop(ctx context.Context) error

	// Enqueue adds a new job to the scheduler.
	Enqueue(ctx context.Context, job *core.Job) error

	// Schedule moves a delayed job to scheduled state.
	Schedule(ctx context.Context, id uuid.UUID) error

	// Complete marks a job as successfully completed.
	Complete(ctx context.Context, id uuid.UUID, result *core.JobResult) error

	// Fail marks a job as failed and determines retry behavior.
	Fail(ctx context.Context, id uuid.UUID, result *core.JobResult) error

	// Cancel requests cancellation of a job.
	Cancel(ctx context.Context, id uuid.UUID, reason string, force bool) error

	// Retry reschedules a failed or dead-lettered job.
	Retry(ctx context.Context, id uuid.UUID, resetAttempts bool) error

	// GetNextJob returns the next job to execute, blocking until available.
	GetNextJob(ctx context.Context) (*core.Job, error)

	// Stats returns current scheduler statistics.
	Stats() *SchedulerStats

	// IsHealthy returns whether the scheduler is operating normally.
	IsHealthy() bool
}

Scheduler coordinates job scheduling and dispatching. It maintains the priority queue and manages job lifecycle transitions.

type SchedulerConfig

type SchedulerConfig struct {
	TickInterval      time.Duration
	BatchSize         int
	VisibilityTimeout time.Duration
	MaxPendingJobs    int64
}

SchedulerConfig configures scheduler behavior.

func DefaultSchedulerConfig

func DefaultSchedulerConfig() *SchedulerConfig

DefaultSchedulerConfig returns sensible defaults.

type SchedulerStats

type SchedulerStats struct {
	PendingCount   int64 `json:"pending_count"`
	ScheduledCount int64 `json:"scheduled_count"`
	RunningCount   int64 `json:"running_count"`
	DelayedCount   int64 `json:"delayed_count"`

	EnqueueRate  float64 `json:"enqueue_rate"`
	CompleteRate float64 `json:"complete_rate"`
	FailRate     float64 `json:"fail_rate"`

	ScheduleLatencyP50 time.Duration `json:"schedule_latency_p50"`
	ScheduleLatencyP99 time.Duration `json:"schedule_latency_p99"`

	LastCycleTime time.Time     `json:"last_cycle_time"`
	CycleDuration time.Duration `json:"cycle_duration"`
	Healthy       bool          `json:"healthy"`
}

SchedulerStats contains scheduler metrics.