Documentation
¶
Overview ¶
Package jobs provides a common abstraction for background jobs: handlers, dispatching, worker lifecycle, retries with backoff, observability hooks and graceful shutdown. Queue infrastructure is supplied by adapters implementing Queue (Redis, PostgreSQL, SQS, RabbitMQ, NATS, ...); MemoryQueue is included for development and tests.
m := jobs.NewManager(jobs.Options{Queue: queue})
m.Register("send-email", jobs.Typed(func(ctx context.Context, p Email) error {
return mailer.Send(ctx, p)
}))
_ = m.Dispatch(ctx, "send-email", Email{To: "a@example.com"})
Index ¶
- Constants
- Variables
- func ExponentialBackoff(base, max time.Duration) func(int) time.Duration
- func IsPermanent(err error) bool
- func Permanent(err error) error
- type Delivery
- type DispatchOption
- type Handler
- type HandlerFunc
- type Job
- type JobOption
- type Manager
- func (m *Manager) Dispatch(ctx context.Context, name string, payload any, opts ...DispatchOption) error
- func (m *Manager) Names() []string
- func (m *Manager) Queue() Queue
- func (m *Manager) Register(name string, h Handler, opts ...JobOption) error
- func (m *Manager) Start(context.Context) error
- func (m *Manager) Stop(ctx context.Context) error
- type MemoryQueue
- type Middleware
- type Options
- type Queue
Constants ¶
const ( MetaRequestID = "request_id" MetaTraceID = "trace_id" )
Metadata keys set by Dispatch.
Variables ¶
var ErrQueueClosed = errors.New("jobs: queue closed")
ErrQueueClosed is returned when enqueuing into a closed MemoryQueue.
var ErrUnknownJob = errors.New("jobs: unknown job")
ErrUnknownJob is returned when dispatching a job name with no handler.
Functions ¶
func ExponentialBackoff ¶
ExponentialBackoff returns base*2^(attempt-1) capped at max, with up to 20% random jitter to avoid synchronized retries.
func IsPermanent ¶
IsPermanent reports whether err was marked with Permanent.
Types ¶
type Delivery ¶
type Delivery interface {
Job() *Job
// Ack marks the job as processed.
Ack(ctx context.Context) error
// Retry schedules redelivery after delay with Attempt incremented.
Retry(ctx context.Context, delay time.Duration) error
// Reject marks the job as permanently failed (dead-lettered when the
// backend supports it).
Reject(ctx context.Context, reason error) error
}
Delivery is a received job awaiting acknowledgment.
type DispatchOption ¶
DispatchOption configures a dispatch.
func WithID ¶
func WithID(id string) DispatchOption
WithID sets the job ID, which backends may use for deduplication.
type HandlerFunc ¶
HandlerFunc adapts a function to Handler.
type Job ¶
type Job struct {
ID string `json:"id"`
Name string `json:"name"`
Payload json.RawMessage `json:"payload,omitempty"`
Attempt int `json:"attempt"`
MaxAttempts int `json:"max_attempts,omitempty"`
EnqueuedAt time.Time `json:"enqueued_at"`
Metadata map[string]string `json:"metadata,omitempty"`
}
Job is a unit of background work.
type JobOption ¶
type JobOption func(*registration)
JobOption configures a registered job.
func MaxAttempts ¶
MaxAttempts overrides the attempt limit for a job.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager registers handlers, dispatches jobs and runs workers. It implements Start and Stop so it can be managed by an application lifecycle.
func NewManager ¶
NewManager returns a Manager. It panics if opts.Queue is nil.
func (*Manager) Dispatch ¶
func (m *Manager) Dispatch(ctx context.Context, name string, payload any, opts ...DispatchOption) error
Dispatch enqueues a job. payload is JSON-encoded unless it is already []byte or json.RawMessage. The request and trace IDs in ctx are recorded so job logs correlate with the originating request. When handlers are registered, dispatching an unknown name fails with ErrUnknownJob; a manager without handlers (dispatch-only) accepts any name.
func (*Manager) Register ¶
Register adds a handler for name. It fails if name is already registered or the manager is running.
type MemoryQueue ¶
type MemoryQueue struct {
// contains filtered or unexported fields
}
MemoryQueue is an in-process queue for development and tests. Jobs are lost when the process exits and are not shared between instances; use a durable backend in production.
func NewMemoryQueue ¶
func NewMemoryQueue(capacity int) *MemoryQueue
NewMemoryQueue returns a queue buffering up to capacity ready jobs (default 1024). Enqueue blocks while the buffer is full.
func (*MemoryQueue) Close ¶
func (q *MemoryQueue) Close() error
Close stops pending delayed deliveries and rejects further enqueues.
func (*MemoryQueue) Failed ¶
func (q *MemoryQueue) Failed() []*Job
Failed returns permanently failed jobs (the dead-letter list).
func (*MemoryQueue) IsLocal ¶
func (q *MemoryQueue) IsLocal() bool
IsLocal reports that the queue's state is process-local.
func (*MemoryQueue) Len ¶
func (q *MemoryQueue) Len() int
Len returns the number of jobs ready for delivery.
type Middleware ¶
Middleware wraps job handlers, for example for tracing or metrics.
type Options ¶
type Options struct {
// Queue is the backend. Required.
Queue Queue
// Workers is the number of concurrent workers (default GOMAXPROCS).
// Set to -1 for a dispatch-only manager that never consumes.
Workers int
// MaxAttempts is the default attempt limit per job (default 3).
MaxAttempts int
// Timeout bounds a single attempt (default 5m).
Timeout time.Duration
// Backoff returns the delay before retry number attempt (1-based).
// Default: exponential from 1s, capped at 5m, with jitter.
Backoff func(attempt int) time.Duration
// Logger receives job failures (default slog.Default()).
Logger *slog.Logger
// Middleware wraps every handler; the first is outermost.
Middleware []Middleware
// OnFailure is called when a job fails permanently.
OnFailure func(ctx context.Context, job *Job, err error)
}
Options configures a Manager.
type Queue ¶
type Queue interface {
// Enqueue stores job for delivery after delay.
Enqueue(ctx context.Context, job *Job, delay time.Duration) error
// Receive blocks until a job is available or ctx is done, in which case
// it returns ctx.Err().
Receive(ctx context.Context) (Delivery, error)
}
Queue is implemented by queue backends.
Source Files