jobs

Documentation

Overview

Package jobs provides a durable, adapter-pluggable job queue for maniflex.

The core package has no third-party imports. Choose an adapter sub-package for the backing store:

• jobs/inproc — goroutine pool; for tests and single-binary apps
• jobs/sql — Postgres/SQLite with transactional outbox semantics
• jobs/redis — Redis Streams; for high-throughput worker fleets

A scheduled trigger is available in jobs/cron. Status persistence for the REST layer lives in jobs/maniflex (3C.4).

Index

• func MaxRetryFor(j Job) int
• func TraceIDFromContext(ctx context.Context) string
• type BackoffPolicy
  • func BackoffFor(j Job) BackoffPolicy
• type BlockingSource
• type Cancellable
• type ExponentialBackoff
  • func (e ExponentialBackoff) Next(attempt int) time.Duration
• type FixedBackoff
  • func (f FixedBackoff) Next(_ int) time.Duration
• type Handler
• type Inspector
• type Job
• type JobState
• type LeaseRenewer
• type ListQuery
• type Queue
• type Result
• type Source
• type Status
• type StatusInfo
• type StatusSink
• type Worker
  • func NewWorker(cfg WorkerConfig) (*Worker, error)
  • func (w *Worker) Run(ctx context.Context) error
  • func (w *Worker) Shutdown(ctx context.Context) error
• type WorkerConfig

Functions

func MaxRetryFor

func MaxRetryFor(j Job) int

MaxRetryFor returns the effective max retry count for j.

func TraceIDFromContext

func TraceIDFromContext(ctx context.Context) string

TraceIDFromContext returns the trace ID stored by the Worker, or "".

Types

type BackoffPolicy

type BackoffPolicy interface {
	Next(attempt int) time.Duration
}

BackoffPolicy computes the delay before a job's next retry attempt.

func BackoffFor

func BackoffFor(j Job) BackoffPolicy

BackoffFor returns the effective backoff policy for j.

type BlockingSource

type BlockingSource interface {
	Source
	// DequeueBlocking is like Dequeue but waits up to max for at least one
	// job to become available. Returns the empty slice (not an error) on
	// timeout or context cancellation. Implementations must respect ctx.
	DequeueBlocking(ctx context.Context, n int, max time.Duration) ([]Job, error)
}

BlockingSource is an optional capability adapters may implement to avoid poll-and-sleep behaviour during idle periods. When the Worker has free slots and no other jobs to dispatch it calls DequeueBlocking, which is expected to wait — up to `max` — until at least one job becomes available (Redis BRPOP, Postgres LISTEN/NOTIFY, ...) or the context is cancelled.

Adapters that do not implement BlockingSource fall back to the standard poll cycle controlled by WorkerConfig.EmptyQueueBackoff. There is no downside to implementing both; the worker prefers the blocking variant whenever the slot pool is fully idle so it does not hammer the source.

type Cancellable

type Cancellable interface {
	Cancel(ctx context.Context, id string) error
}

Cancellable is an optional capability adapters may implement.

type ExponentialBackoff

type ExponentialBackoff struct {
	Base time.Duration
	Max  time.Duration
}

ExponentialBackoff doubles the delay on each attempt, capped at Max. This is the default policy when Job.Backoff is nil.

func (ExponentialBackoff) Next

func (e ExponentialBackoff) Next(attempt int) time.Duration

type FixedBackoff

type FixedBackoff struct {
	Delay time.Duration
}

FixedBackoff returns the same delay on every attempt.

func (FixedBackoff) Next

func (f FixedBackoff) Next(_ int) time.Duration

type Handler

type Handler func(ctx context.Context, j Job) (Result, error)

Handler is the function signature for job processing logic.

type Inspector

type Inspector interface {
	Get(ctx context.Context, id string) (JobState, error)
	List(ctx context.Context, q ListQuery) ([]JobState, error)
}

Inspector is an optional capability adapters may implement for job introspection.

type Job

type Job struct {
	// ID is a ULID-style sortable identifier. Auto-assigned on Enqueue when empty.
	ID string

	// Type identifies the handler to dispatch to (e.g. "export_invoices").
	Type string

	// Payload is the opaque, handler-specific data. Handlers decode it themselves.
	Payload json.RawMessage

	// TraceID is the W3C traceparent value propagated to the Handler context.
	TraceID string

	ActorID  string
	TenantID string

	// MaxRetry is the maximum number of attempts before the job is marked dead.
	// Zero means the adapter default (3).
	MaxRetry int

	// Backoff controls the delay between retries. nil uses ExponentialBackoff{1s, 5m}.
	Backoff BackoffPolicy

	// Priority is a hint to the adapter; higher values run first.
	Priority int

	// NotBefore is the earliest time the job may be executed. Zero means now.
	NotBefore time.Time

	// GroupKey serialises execution: at most one job per key runs at a time.
	// e.g. tenant_id for per-tenant payroll runs.
	GroupKey string

	// Headers are arbitrary routing hints or idempotency keys.
	Headers map[string]string

	// Attempts is populated by the Source on Dequeue. Handlers and the Worker
	// use it to compute backoff delay; it is not set by the caller.
	Attempts int
}

Job is the unit of work passed to a Handler. All fields except Type are optional; adapters supply sensible defaults on Enqueue.

type JobState

type JobState struct {
	Job
	Status      Status
	Error       string
	StartedAt   *time.Time
	CompletedAt *time.Time
	CreatedAt   time.Time
	UpdatedAt   time.Time
}

JobState is the full runtime representation of a job, including persistence metadata.

type LeaseRenewer

type LeaseRenewer interface {
	RenewLease(ctx context.Context, id string, d time.Duration) error
}

LeaseRenewer is an optional interface Source adapters may implement to allow the worker to extend a job's lease while it is running.

type ListQuery

type ListQuery struct {
	Type     string
	Status   Status
	TenantID string
	ActorID  string
	Limit    int
	Offset   int
}

ListQuery filters results for Inspector.List.

type Queue

type Queue interface {
	// Enqueue submits j for immediate execution. The returned id is the assigned
	// Job.ID (auto-generated if j.ID is empty).
	Enqueue(ctx context.Context, j Job) (id string, err error)

	// EnqueueAt submits j for execution no earlier than at.
	EnqueueAt(ctx context.Context, j Job, at time.Time) (id string, err error)

	// EnqueueBatch submits multiple jobs atomically where the adapter supports it.
	// Returns one id per job in input order.
	EnqueueBatch(ctx context.Context, js []Job) ([]string, error)

	Close() error
}

Queue is the producer surface. All three methods are safe for concurrent use.

type Result

type Result struct {
	// Output is a small, structured result (stored inline).
	Output json.RawMessage

	// URL is a pre-signed or server-relative URL to a large output artefact.
	URL string

	// Mime is the content-type of the artefact at URL.
	Mime string

	SizeBytes int64
}

Result is the structured output of a successful Handler invocation.

type Source

type Source interface {
	// Dequeue claims up to n ready jobs and returns them. It sets Job.Attempts
	// to the current attempt count for each returned job.
	// Blocks briefly (adapter-defined) when no jobs are ready; returns an empty
	// slice (not an error) on timeout.
	Dequeue(ctx context.Context, n int) ([]Job, error)

	// Ack marks id as succeeded.
	Ack(ctx context.Context, id string) error

	// Nack marks id as failed. If the job's attempt count has reached MaxRetry
	// the adapter marks it dead; otherwise it reschedules with the given delay.
	Nack(ctx context.Context, id string, jobErr error, delay time.Duration) error

	// Dead unconditionally marks id as dead (permanent failure).
	Dead(ctx context.Context, id string, jobErr error) error
}

Source is the consumer surface, implemented by adapters. The Worker calls these methods; application code does not call them directly.

type Status

type Status string

Status describes the lifecycle state of a job.

const (
	StatusEnqueued  Status = "enqueued"
	StatusRunning   Status = "running"
	StatusSucceeded Status = "succeeded"
	StatusFailed    Status = "failed" // retryable terminal
	StatusDead      Status = "dead"   // permanent — MaxRetry exhausted
	StatusCancelled Status = "cancelled"
)

type StatusInfo

type StatusInfo struct {
	Attempt  int
	Error    string
	Result   *Result
	JobType  string
	ActorID  string
	TenantID string
}

StatusInfo carries supplemental data for a StatusSink.Transition call.

type StatusSink

type StatusSink interface {
	Transition(ctx context.Context, id string, from, to Status, info StatusInfo) error
}

StatusSink receives lifecycle transitions from the Worker. Implement this to persist job status alongside your business data. jobs/maniflex provides one backed by the Server model layer.

type Worker

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

Worker dispatches jobs from a Source to the registered Handlers.

func NewWorker

func NewWorker(cfg WorkerConfig) (*Worker, error)

NewWorker validates cfg and returns a ready Worker.

func (*Worker) Run

func (w *Worker) Run(ctx context.Context) error

Run starts the dequeue-dispatch loop. It blocks until ctx is cancelled, then returns nil. Use Shutdown to drain in-flight handlers after cancellation.

Loop shape:

• All slots busy: wait on the semaphore (no polling).
• Slots available: call Source.Dequeue, or DequeueBlocking when the adapter implements BlockingSource AND the pool is fully idle (no in-flight jobs). This trades a brief wakeup latency on warm-up for zero idle traffic on cold paths.
• Empty result: exponential backoff from EmptyQueueBackoff to MaxEmptyQueueBackoff, reset on the next successful dispatch.

func (*Worker) Shutdown

func (w *Worker) Shutdown(ctx context.Context) error

Shutdown waits for all in-flight handlers to complete, bounded by the deadline on ctx. Returns an error if the deadline is exceeded.

type WorkerConfig

type WorkerConfig struct {
	// Source is the adapter the worker dequeues from. Required.
	Source Source

	// Handlers maps Job.Type to the processing function.
	Handlers map[string]Handler

	// Concurrency is the maximum number of jobs processed simultaneously.
	// Defaults to runtime.GOMAXPROCS(0).
	Concurrency int

	// Status receives lifecycle transitions (enqueued→running→succeeded|failed|dead).
	// nil disables status tracking.
	Status StatusSink

	Logger *slog.Logger

	// LeaseRenew is how often the worker renews the job lease for long-running
	// handlers. Only used by adapters that implement LeaseRenewer.
	// Defaults to 30s.
	LeaseRenew time.Duration

	// ShutdownWait is the maximum duration Shutdown() waits for in-flight
	// handlers to complete. Defaults to 30s.
	ShutdownWait time.Duration

	// EmptyQueueBackoff is how long the worker waits after Dequeue returns
	// no jobs before trying again. Pre-fix this was a hard-coded 100ms,
	// which hammered Redis BRPOP / Postgres SKIP LOCKED at scale. Default
	// 1s; the worker grows the delay exponentially (×2 per consecutive
	// empty) up to MaxEmptyQueueBackoff. Adapters that implement
	// BlockingSource bypass this entirely when the slot pool is fully idle.
	EmptyQueueBackoff time.Duration

	// MaxEmptyQueueBackoff caps the exponential backoff above. Default 30s.
	MaxEmptyQueueBackoff time.Duration

	// DLQType is the job type used for dead-letter routing. When non-empty and
	// a matching handler is registered, dead jobs are re-enqueued under this
	// type with the original job as payload.
	DLQType string

	// OnPanic is called when a handler panics. When nil the panic is logged and
	// the job is nacked.
	OnPanic func(j Job, recovered any)

	// EventBus is an optional bus for publishing job.{type}.completed and
	// job.{type}.failed events. Must implement EventPublisher if non-nil.
	// The worker checks for the interface at runtime; supplying a concrete type
	// avoids a compile-time import of the events package.
	EventBus any
}

WorkerConfig configures a Worker.