core

Documentation

Overview

Package core provides error types for GopherQueue.

Package core provides the fundamental types for GopherQueue.

Package core provides configuration options for GopherQueue.

Index

• Variables
• func IsRetriable(err error) bool
• type BackoffStrategy
• type ErrorCategory
  • func ClassifyError(err error) ErrorCategory
• type Job
  • func NewJob(jobType string, payload []byte, opts ...JobOption) *Job
  • func (j *Job) CanRetry() bool
  • func (j *Job) IsTerminal() bool
• type JobContext
• type JobOption
  • func WithBackoff(strategy BackoffStrategy, initial, max time.Duration, multiplier float64) JobOption
  • func WithCorrelationID(id string) JobOption
  • func WithDelay(delay time.Duration) JobOption
  • func WithDependencies(deps ...uuid.UUID) JobOption
  • func WithID(id uuid.UUID) JobOption
  • func WithIdempotencyKey(key string) JobOption
  • func WithMaxAttempts(attempts int) JobOption
  • func WithMetadata(metadata map[string]interface{}) JobOption
  • func WithPriority(priority Priority) JobOption
  • func WithScheduledAt(t time.Time) JobOption
  • func WithTag(key, value string) JobOption
  • func WithTags(tags map[string]string) JobOption
  • func WithTimeout(timeout time.Duration) JobOption
• type JobResult
• type JobState
• type PermanentError
  • func NewPermanentError(message string, underlying error) *PermanentError
  • func (e *PermanentError) Error() string
  • func (e *PermanentError) Permanent() bool
  • func (e *PermanentError) Unwrap() error
• type Priority
• type RateLimitError
  • func (e *RateLimitError) Error() string
  • func (e *RateLimitError) RateLimited() bool
• type ServerConfig
  • func DefaultServerConfig() *ServerConfig
• type TemporaryError
  • func NewTemporaryError(message string, underlying error) *TemporaryError
  • func (e *TemporaryError) Error() string
  • func (e *TemporaryError) Temporary() bool
  • func (e *TemporaryError) Unwrap() error
• type TimeoutError
  • func NewTimeoutError(message string) *TimeoutError
  • func (e *TimeoutError) Error() string
  • func (e *TimeoutError) Timeout() bool
• type ValidationError
  • func (e *ValidationError) Error() string

Variables

var (
	ErrJobNotFound       = errors.New("job not found")
	ErrJobAlreadyExists  = errors.New("job already exists")
	ErrJobNotCancellable = errors.New("job cannot be cancelled in current state")
	ErrQueueFull         = errors.New("job queue is full")
	ErrShuttingDown      = errors.New("system is shutting down")
	ErrRateLimited       = errors.New("rate limit exceeded")
)

Common errors.

Functions

func IsRetriable

func IsRetriable(err error) bool

IsRetriable returns whether an error should be retried.

Types

type BackoffStrategy

type BackoffStrategy string

BackoffStrategy defines retry backoff algorithms.

const (
	BackoffConstant    BackoffStrategy = "constant"
	BackoffLinear      BackoffStrategy = "linear"
	BackoffExponential BackoffStrategy = "exponential"
)

type ErrorCategory

type ErrorCategory string

ErrorCategory classifies errors for retry decisions.

const (
	ErrorCategoryTemporary ErrorCategory = "temporary"
	ErrorCategoryPermanent ErrorCategory = "permanent"
	ErrorCategoryTimeout   ErrorCategory = "timeout"
	ErrorCategorySystem    ErrorCategory = "system"
	ErrorCategoryRateLimit ErrorCategory = "rate_limit"
)

func ClassifyError

func ClassifyError(err error) ErrorCategory

ClassifyError determines the category of an error.

type Job

type Job struct {
	// Identity
	ID             uuid.UUID         `json:"id"`
	IdempotencyKey string            `json:"idempotency_key,omitempty"`
	CorrelationID  string            `json:"correlation_id,omitempty"`
	Type           string            `json:"type"`
	Payload        []byte            `json:"payload"`
	Tags           map[string]string `json:"tags,omitempty"`

	// Priority and Scheduling
	Priority    Priority      `json:"priority"`
	Delay       time.Duration `json:"delay,omitempty"`
	ScheduledAt time.Time     `json:"scheduled_at,omitempty"`

	// Execution Control
	Timeout           time.Duration   `json:"timeout,omitempty"`
	MaxAttempts       int             `json:"max_attempts"`
	Attempt           int             `json:"attempt"`
	BackoffStrategy   BackoffStrategy `json:"backoff_strategy"`
	BackoffInitial    time.Duration   `json:"backoff_initial"`
	BackoffMax        time.Duration   `json:"backoff_max"`
	BackoffMultiplier float64         `json:"backoff_multiplier"`

	// Dependencies
	DependsOn []uuid.UUID `json:"depends_on,omitempty"`
	BlockedBy []uuid.UUID `json:"blocked_by,omitempty"`

	// State
	State           JobState `json:"state"`
	WorkerID        string   `json:"worker_id,omitempty"`
	Progress        float64  `json:"progress"`
	ProgressMessage string   `json:"progress_message,omitempty"`

	// Error information
	LastError     string `json:"last_error,omitempty"`
	ErrorCategory string `json:"error_category,omitempty"`

	// Timestamps
	CreatedAt   time.Time `json:"created_at"`
	UpdatedAt   time.Time `json:"updated_at"`
	StartedAt   time.Time `json:"started_at,omitempty"`
	CompletedAt time.Time `json:"completed_at,omitempty"`

	// Metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

Job represents a unit of work to be processed.

func NewJob

func NewJob(jobType string, payload []byte, opts ...JobOption) *Job

NewJob creates a new job with the given type and payload.

func (*Job) CanRetry

func (j *Job) CanRetry() bool

CanRetry returns whether the job can be retried.

func (*Job) IsTerminal

func (j *Job) IsTerminal() bool

IsTerminal returns whether the job is in a terminal state.

type JobContext

type JobContext interface {
	// Job returns the current job being executed.
	Job() *Job

	// Context returns the Go context for cancellation.
	Context() context.Context

	// Checkpoint saves progress for crash recovery.
	Checkpoint(data []byte) error

	// LastCheckpoint retrieves the last checkpoint data.
	LastCheckpoint() ([]byte, error)

	// SetOutput sets the job output.
	SetOutput(data []byte)

	// Progress reports execution progress.
	Progress(percent float64, message string)

	// Logger returns the job's logger.
	Logger() interface{}
}

JobContext provides the execution context for a job handler.

type JobOption

type JobOption func(*Job)

JobOption is a functional option for configuring a job.

func WithBackoff

func WithBackoff(strategy BackoffStrategy, initial, max time.Duration, multiplier float64) JobOption

WithBackoff configures the retry backoff strategy.

func WithCorrelationID

func WithCorrelationID(id string) JobOption

WithCorrelationID sets the correlation ID for tracing.

func WithDelay

func WithDelay(delay time.Duration) JobOption

WithDelay sets the initial delay before processing.

func WithDependencies

func WithDependencies(deps ...uuid.UUID) JobOption

WithDependencies sets job dependencies.

func WithID

func WithID(id uuid.UUID) JobOption

WithID sets a specific job ID.

func WithIdempotencyKey

func WithIdempotencyKey(key string) JobOption

WithIdempotencyKey sets the idempotency key.

func WithMaxAttempts

func WithMaxAttempts(attempts int) JobOption

WithMaxAttempts sets the maximum retry attempts.

func WithMetadata

func WithMetadata(metadata map[string]interface{}) JobOption

WithMetadata sets job metadata.

func WithPriority

func WithPriority(priority Priority) JobOption

WithPriority sets the job priority.

func WithScheduledAt

func WithScheduledAt(t time.Time) JobOption

WithScheduledAt sets the specific time to run.

func WithTag

func WithTag(key, value string) JobOption

WithTag adds a single tag.

func WithTags

func WithTags(tags map[string]string) JobOption

WithTags sets the job tags.

func WithTimeout

func WithTimeout(timeout time.Duration) JobOption

WithTimeout sets the maximum execution time.

type JobResult

type JobResult struct {
	JobID         uuid.UUID     `json:"job_id"`
	Success       bool          `json:"success"`
	Output        []byte        `json:"output,omitempty"`
	Error         string        `json:"error,omitempty"`
	ErrorCategory ErrorCategory `json:"error_category,omitempty"`
	Duration      time.Duration `json:"duration"`
	CompletedAt   time.Time     `json:"completed_at"`
	RetryAfter    time.Duration `json:"retry_after,omitempty"`
}

JobResult represents the outcome of a job execution.

type JobState

type JobState string

JobState represents the current state of a job.

const (
	JobStatePending    JobState = "pending"
	JobStateScheduled  JobState = "scheduled"
	JobStateRunning    JobState = "running"
	JobStateRetrying   JobState = "retrying"
	JobStateDelayed    JobState = "delayed"
	JobStateCompleted  JobState = "completed"
	JobStateFailed     JobState = "failed"
	JobStateDeadLetter JobState = "dead_letter"
	JobStateCancelled  JobState = "cancelled"
)

type PermanentError

type PermanentError struct {
	Message    string
	Underlying error
}

PermanentError indicates a non-retriable failure.

func NewPermanentError

func NewPermanentError(message string, underlying error) *PermanentError

NewPermanentError creates a new permanent error.

func (*PermanentError) Error

func (e *PermanentError) Error() string

func (*PermanentError) Permanent

func (e *PermanentError) Permanent() bool

func (*PermanentError) Unwrap

func (e *PermanentError) Unwrap() error

type Priority

type Priority int

Priority represents job priority levels.

const (
	PriorityCritical Priority = 0
	PriorityHigh     Priority = 1
	PriorityNormal   Priority = 2
	PriorityLow      Priority = 3
	PriorityBulk     Priority = 4
)

type RateLimitError

type RateLimitError struct {
	Message string
	RetryIn int64 // seconds
}

RateLimitError indicates a rate limit failure.

func (*RateLimitError) Error

func (e *RateLimitError) Error() string

func (*RateLimitError) RateLimited

func (e *RateLimitError) RateLimited() bool

type ServerConfig

type ServerConfig struct {
	// HTTP server configuration
	HTTPAddr string `json:"http_addr"`

	// Data directory for persistence
	DataDir string `json:"data_dir"`

	// Worker configuration
	Workers           int           `json:"workers"`
	ShutdownTimeout   time.Duration `json:"shutdown_timeout"`
	HeartbeatInterval time.Duration `json:"heartbeat_interval"`

	// Scheduler configuration
	SchedulerTickInterval time.Duration `json:"scheduler_tick_interval"`
	VisibilityTimeout     time.Duration `json:"visibility_timeout"`

	// Cleanup configuration
	RetentionPeriod time.Duration `json:"retention_period"`
	CleanupInterval time.Duration `json:"cleanup_interval"`

	// Observability
	MetricsEnabled bool   `json:"metrics_enabled"`
	MetricsAddr    string `json:"metrics_addr"`

	// Security
	APIKeyEnabled bool   `json:"api_key_enabled"`
	APIKey        string `json:"api_key"`
}

ServerConfig holds the main server configuration.

func DefaultServerConfig

func DefaultServerConfig() *ServerConfig

DefaultServerConfig returns sensible defaults.

type TemporaryError

type TemporaryError struct {
	Message    string
	Underlying error
}

TemporaryError indicates a retriable failure.

func NewTemporaryError

func NewTemporaryError(message string, underlying error) *TemporaryError

NewTemporaryError creates a new temporary error.

func (*TemporaryError) Error

func (e *TemporaryError) Error() string

func (*TemporaryError) Temporary

func (e *TemporaryError) Temporary() bool

func (*TemporaryError) Unwrap

func (e *TemporaryError) Unwrap() error

type TimeoutError

type TimeoutError struct {
	Message string
}

TimeoutError indicates a timeout failure.

func NewTimeoutError

func NewTimeoutError(message string) *TimeoutError

NewTimeoutError creates a new timeout error.

func (*TimeoutError) Error

func (e *TimeoutError) Error() string

func (*TimeoutError) Timeout

func (e *TimeoutError) Timeout() bool

type ValidationError

type ValidationError struct {
	Field   string
	Message string
}

ValidationError indicates invalid input.

func (*ValidationError) Error

func (e *ValidationError) Error() string