queue

README

go-queue

A lightweight, driver-based job queue for Go. go-queue defines a small Queue interface and ships three backends — SQLite, in-memory, and no-op — along with a Worker that polls one or more named queues and dispatches jobs to registered handlers with retry, per-job max attempts, priority ordering across queues, and graceful shutdown via context.Context. It is intended for services that want Laravel-style background jobs without a broker such as Redis or RabbitMQ.

Status

Pre-1.0 (v0.2.x). The public API is stable in shape — Queue, Worker, WorkerOpts, the driver packages — but minor breaks may still happen between v0.x releases. See CHANGELOG.md for per-release detail and pin a version in your go.mod.

Documentation: pkg.go.dev/github.com/hollis-labs/go-queue.

Install

go get github.com/hollis-labs/go-queue

Usage

Minimal example using the in-memory driver (adapted from worker_test.go):

package main

import (
    "context"
    "fmt"
    "time"

    queue "github.com/hollis-labs/go-queue"
    "github.com/hollis-labs/go-queue/driver/memory"
)

func main() {
    ctx := context.Background()
    q := memory.New()

    // Enqueue a job.
    if err := q.Push(ctx, "greet", []byte(`"hello"`)); err != nil {
        panic(err)
    }

    // Build a worker and register a handler for the "greet" job type.
    w := queue.NewWorker(q, queue.WorkerOpts{
        StopWhenEmpty: true,
        PollInterval:  10 * time.Millisecond,
        MaxTries:      3,
    })
    w.Register("greet", func(_ context.Context, job *queue.QueuedJob) error {
        fmt.Printf("got job %s: %s\n", job.ID, string(job.Payload))
        return nil
    })

    // Blocks until the queue is drained (because of StopWhenEmpty) or ctx cancels.
    if err := w.Start(ctx); err != nil {
        panic(err)
    }
}

For a persistent backend, construct the SQLite driver with an open *sql.DB:

import (
    "database/sql"

    qsqlite "github.com/hollis-labs/go-queue/driver/sqlite"
    _ "modernc.org/sqlite"
)

db, _ := sql.Open("sqlite", "jobs.db")
db.Exec("PRAGMA journal_mode=WAL")
q, err := qsqlite.New(db, qsqlite.Opts{})

Push options cover the common scheduling knobs:

q.Push(ctx, "email", payload,
    queue.OnQueue("high"),
    queue.WithDelay(30*time.Second),
    queue.WithMaxTries(5),
)

API Overview

Root package (github.com/hollis-labs/go-queue):

• Queue — interface every driver implements: Push, Pop, Delete, Release, Size, Failed.
• QueuedJob — struct returned by Pop, carrying ID, Type, Queue, Payload, Attempts, MaxTries, and timestamps.
• Handler — func(ctx context.Context, job *QueuedJob) error; registered on a Worker by job-type string.
• Worker / NewWorker(q, opts) — polls one or more queues and dispatches jobs to handlers.
• WorkerOpts — Queues, Concurrency, PollInterval, MaxTries, RetryAfter, MaxMemoryMB, StopWhenEmpty, CanReserve, optional Controller, plus OnProcessing, OnProcessed, OnFailed, OnError lifecycle callbacks.
• PushOption helpers: OnQueue(name), WithDelay(d), WithMaxTries(n).
• PushConfig / ResolvePushConfig(opts) — exported so custom drivers can consume the resolved push options.
• Sentinels: ErrNoJob, ErrHandlerNotFound.

Drivers:

• driver/memory — memory.New() returns a mutex-protected, slice-backed *Driver. Adds FailedJobs() for test inspection.
• driver/sqlite — sqlite.New(db, sqlite.Opts{...}) creates the jobs and failed_jobs tables (names configurable via Opts.Table / Opts.FailedTable) and returns a *Driver. Opts.RetryAfter (default 60s) controls how long a reserved job must sit before Pop reclaims it.
• driver/noop — noop.New() returns a *Driver that accepts pushes and drops them silently; Pop always returns (nil, nil). Used when queueing is disabled.

Architecture Notes

Reversible worker drain

Pass a *CycleController to opt into whole-cycle pause and settlement tracking. The zero value is ready to use:

controller := &queue.CycleController{}
worker := queue.NewWorker(q, queue.WorkerOpts{Controller: controller})
// Register handlers and run worker.Start(workerCtx) in its owning goroutine.

pause := controller.Pause()
report, err := controller.WaitQuiescent(waitCtx, pause)
if err != nil {
    // Inspect report internally. This resumes only the pause, not a fault.
    _ = controller.Resume(pause)
    return err
}
// This worker has no active cycles or unknown dispositions while pause is held.
// Keep the handle held for the operation that requires quiescence.
return controller.Resume(pause)

A cycle enrolls before CanReserve and Pop, and remains active through the handler, Delete/Release/Failed, and lifecycle callbacks. Thus pause cannot miss a poll that passed eligibility but has not reserved yet, or a handler whose result is not durably recorded. It does not cancel workerCtx. A canceled or expired waitCtx stops only the wait; the caller decides whether to resume. Callbacks must return before quiescence and must not wait on their own cycle.

Pause is nonnesting: concurrent/repeated calls while paused share one handle. The caller must serialize ownership of Resume; one caller resuming releases that pause for every holder. A zero, foreign or retired handle returns ErrStalePause. One controller belongs permanently to one Worker, and concurrent Starts or reuse by another worker return ErrControllerInUse. A paused worker does not treat the queue as empty for StopWhenEmpty. Normal idle polling still obeys PollInterval; resume does not promise to interrupt an existing poll sleep.

Managed cycles classify every driver error conservatively as unknown, even Pop(nil, err) or a read-only Size error: the existing driver interface has no richer outcome contract. An uncertain Pop may already have reserved a job, and retrying Release/Failed after a lost acknowledgement could repeat an effect. New cycles are quarantined as soon as the error is known; other enrolled cycles finish. There is no automatic settlement replay or fault reset, including across Starts of the same Worker. Resume may release a pause but cannot restore polling after a fault. Plan reconciliation separately instead of discarding the controller to claim a successful drain.

WaitQuiescent returns ErrUnknownDisposition after all active cycles finish when any outcome remains uncertain. CycleReport.Unknown retains at most one record per polling goroutine; Last is the most recent completed cycle. Results name the operation, diagnostic cycle number, optional opaque job ID and original error. Reports are detached snapshots; they may contain private driver details and are not safe wire/log payloads. A report without a held pause is only an observation. There is no claim that the shared queue is empty or other workers are idle.

In managed mode, settlement errors reach OnError; OnProcessed and OnFailed run only when their corresponding Delete/Failed succeeded. A missing handler still reaches OnError with ErrHandlerNotFound, even if Failed also errors. With Controller:nil, the prior behavior (including callbacks and handling of driver errors) is unchanged. Drivers, retry limits, FIFO/reclaim semantics and schema are unchanged. This API does not install a shutdown watcher or exit a consumer process.

Driver and worker ownership

The split between Queue (driver contract) and Worker (dispatch loop) means consumers depend on the interface, not on a concrete backend. The worker owns policy — poll interval, concurrency, retry counting, priority ordering — while drivers own storage and reservation semantics.

Priority is implemented by WorkerOpts.Queues order: popNextJob walks the slice and returns the first non-nil job, so listing ["high", "low"] drains high before looking at low (see TestWorkerPriorityQueues). Per-job WithMaxTries overrides WorkerOpts.MaxTries when greater than zero (TestWorkerPerJobMaxTries).

The SQLite driver uses a monotonically increasing id with explicit queue, reserved_at, and available_at columns, and reclaims stuck reservations whose reserved_at is older than Opts.RetryAfter. All mutations in Pop, Release, and Failed run inside a transaction. Release deletes and re-inserts the row so FIFO ordering is preserved across retries while the attempt count is carried over. The in-memory driver mirrors this semantics in-process with a sync.Mutex-guarded slice.

Dependencies

Direct:

• modernc.org/sqlite — pure-Go SQLite driver used by driver/sqlite and its tests. Pinned in go.mod.

All other entries in go.sum are transitive dependencies of modernc.org/sqlite. The root package and the memory/noop drivers have no external dependencies.

Testing

go test ./...

The SQLite driver tests use :memory: databases via the pure-Go modernc.org/sqlite driver, so no CGO toolchain or external SQLite install is required. No environment variables, fixtures, or external services are needed.

For runnable end-to-end examples, see the examples/ directory.

License

MIT © Hollis Labs.

Documentation

Overview

Package queue provides a driver-based job queue with pluggable backends, retries, per-job max attempts, priority queue ordering, and a polling worker.

Index

• Variables
• type CycleController
  • func (c *CycleController) Pause() PauseHandle
  • func (c *CycleController) Report() CycleReport
  • func (c *CycleController) Resume(h PauseHandle) error
  • func (c *CycleController) WaitQuiescent(ctx context.Context, h PauseHandle) (CycleReport, error)
• type CycleDisposition
• type CycleOperation
• type CycleReport
• type CycleResult
• type Handler
• type PauseHandle
• type PushConfig
  • func ResolvePushConfig(opts []PushOption) PushConfig
• type PushOption
  • func OnQueue(name string) PushOption
  • func WithDelay(d time.Duration) PushOption
  • func WithMaxTries(n int) PushOption
• type Queue
• type QueuedJob
• type Worker
  • func NewWorker(q Queue, opts WorkerOpts) *Worker
  • func (w *Worker) Register(jobType string, h Handler)
  • func (w *Worker) Start(ctx context.Context) error
• type WorkerOpts

Variables

var (
	ErrStalePause         = errors.New("queue: pause handle is not current")
	ErrUnknownDisposition = errors.New("queue: a cycle has an unknown disposition")
	ErrControllerInUse    = errors.New("queue: controller already belongs to another worker or active start")
)

Controller errors are comparable with errors.Is. Driver errors remain in CycleReport rather than being folded into these fixed messages.

var ErrHandlerNotFound = errors.New("handler not found for job type")

ErrHandlerNotFound is used when a job is popped but no handler is registered for its type.

var ErrNoJob = errors.New("no job available")

ErrNoJob is returned by Pop when no job is available. Drivers return nil, nil instead of this error — this sentinel exists for internal worker logic.

Types

type CycleController

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

CycleController gates entire polling cycles without canceling their contexts. Its zero value is ready to use. It belongs permanently to one Worker and must not be copied after use. Driver errors permanently quarantine new cycles for this controller, including across Starts. Resume never clears uncertainty. There is no reset API. Callbacks are part of a cycle and must not wait for their own quiescence.

func (*CycleController) Pause

func (c *CycleController) Pause() PauseHandle

Pause closes cycle admission. Repeated calls during the same pause return the same handle. It may be called before Worker.Start.

func (*CycleController) Report

func (c *CycleController) Report() CycleReport

Report returns detached result/slice values. It does not close admission.

func (*CycleController) Resume

func (c *CycleController) Resume(h PauseHandle) error

Resume reopens this pause only. Uncertainty still quarantines the worker.

func (*CycleController) WaitQuiescent

func (c *CycleController) WaitQuiescent(ctx context.Context, h PauseHandle) (CycleReport, error)

WaitQuiescent waits through eligibility, reservation, handlers, settlement and callbacks. A deadline stops only this wait. The caller must keep the pause held while relying on success. Unknown outcomes are reported only after active cycles finish; they never become a successful idle report.

type CycleDisposition

type CycleDisposition string

CycleDisposition describes what is known about a completed cycle.

const (
	DispositionNoReservation CycleDisposition = "no_reservation"
	DispositionSettled       CycleDisposition = "settled"
	DispositionUnknown       CycleDisposition = "unknown"
)

type CycleOperation

type CycleOperation string

CycleOperation identifies the last operation in a polling cycle.

const (
	OperationEligibility CycleOperation = "eligibility"
	OperationPop         CycleOperation = "pop"
	OperationSize        CycleOperation = "size"
	OperationDelete      CycleOperation = "delete"
	OperationRelease     CycleOperation = "release"
	OperationFailed      CycleOperation = "failed"
)

type CycleReport

type CycleReport struct {
	Active  int
	Last    *CycleResult
	Unknown []CycleResult
}

CycleReport is a snapshot. Only a successful WaitQuiescent with a still-held pause proves quiescence. Last is the most recently completed cycle; Unknown retains every uncertain cycle, bounded by the worker's polling concurrency.

type CycleResult

type CycleResult struct {
	Cycle       uint64
	Operation   CycleOperation
	Disposition CycleDisposition
	JobID       string
	Err         error
}

CycleResult carries internal evidence, not a safe-to-publish diagnostic. Err may contain driver details; JobID is opaque and Payload is never retained. Cycle is a diagnostic sequence, not an authorization token.

type Handler

type Handler func(ctx context.Context, job *QueuedJob) error

Handler processes a job payload. Registered with the worker by job type string.

type PauseHandle

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

PauseHandle belongs to one controller and one uninterrupted pause. Copies refer to the same pause; a handle cannot resume a later pause.

type PushConfig

type PushConfig struct {
	Queue    string
	Delay    time.Duration
	MaxTries int
}

PushConfig holds resolved push options. Exported so drivers can access the resolved values.

func ResolvePushConfig

func ResolvePushConfig(opts []PushOption) PushConfig

ResolvePushConfig applies options and returns the resolved config.

type PushOption

type PushOption func(*PushConfig)

PushOption configures optional parameters for Push.

func OnQueue

func OnQueue(name string) PushOption

OnQueue sets the queue name for the job. Default: "default".

func WithDelay

func WithDelay(d time.Duration) PushOption

WithDelay sets a delay before the job becomes available. Default: 0.

func WithMaxTries

func WithMaxTries(n int) PushOption

WithMaxTries sets the maximum number of attempts for this job. 0 means inherit from WorkerOpts.MaxTries.

type Queue

type Queue interface {
	// Push enqueues a new job. The jobType identifies which handler
	// processes it; payload is an opaque JSON blob.
	Push(ctx context.Context, jobType string, payload []byte, opts ...PushOption) error

	// Pop retrieves and reserves the next available job from the named
	// queue. Returns nil, nil when the queue is empty.
	Pop(ctx context.Context, queueName string) (*QueuedJob, error)

	// Delete removes a completed job from the queue.
	Delete(ctx context.Context, id string) error

	// Release puts a job back on the queue with a delay. The job gets
	// a new position (FIFO ordering preserved) but keeps its attempt
	// count.
	Release(ctx context.Context, id string, delay time.Duration) error

	// Size returns the total number of jobs on the named queue.
	Size(ctx context.Context, queueName string) (int, error)

	// Failed moves a job to the failed jobs store with the given error
	// message. The original job is removed from the active queue.
	Failed(ctx context.Context, job *QueuedJob, errMsg string) error
}

Queue is the driver interface. Each backend (SQLite, memory, noop) implements this contract. Consumers depend on this interface, not concrete drivers.

type QueuedJob

type QueuedJob struct {
	ID          string
	Type        string
	Queue       string
	Payload     []byte
	Attempts    int
	MaxTries    int
	CreatedAt   time.Time
	AvailableAt time.Time
	ReservedAt  *time.Time
}

QueuedJob is a job that has been popped from the queue. It carries the metadata the worker needs for dispatch, retry, and failure.

type Worker

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

Worker polls a Queue and dispatches jobs to registered handlers.

func NewWorker

func NewWorker(q Queue, opts WorkerOpts) *Worker

NewWorker creates a Worker bound to the given Queue.

func (*Worker) Register

func (w *Worker) Register(jobType string, h Handler)

Register adds a handler for the given job type.

func (*Worker) Start

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

Start begins polling. Blocks until ctx is cancelled or StopWhenEmpty triggers. Returns nil on clean shutdown.

type WorkerOpts

type WorkerOpts struct {
	// Queues to poll in priority order. Default: ["default"].
	Queues []string

	// Concurrency is the number of polling goroutines. Default: 1.
	Concurrency int

	// PollInterval is the sleep duration between empty polls. Default: 3s.
	PollInterval time.Duration

	// MaxTries is the default max attempts per job. Jobs with
	// WithMaxTries override this. 0 = unlimited. Default: 3.
	MaxTries int

	// RetryAfter is the delay before a failed job is retried.
	// 0 means no delay (immediate retry). Default: 0.
	RetryAfter time.Duration

	// MaxMemoryMB stops the worker if memory exceeds this. 0 = no limit.
	MaxMemoryMB int

	// StopWhenEmpty exits after the queue is drained. Default: false.
	StopWhenEmpty bool

	// CanReserve, when non-nil, is asked once per poll cycle whether this
	// worker may take work right now. Returning false skips the cycle: nothing
	// is reserved, the worker sleeps PollInterval, and it asks again.
	//
	// It gates ONLY new reservations. A job already held runs to completion
	// under a live context, which is the difference between "stop taking work"
	// and "abandon work in progress" — and the reason this is a gate rather
	// than something a caller can approximate by cancelling the worker's
	// context. Cancelling aborts the handler mid-flight and then fails the
	// bookkeeping that follows it, leaving the job reserved and its attempt
	// spent.
	//
	// Written for deference between processes that share a queue: a leader that
	// owns the work, a maintenance window, or a drain-down before shutdown. The
	// canonical case is a long-lived service and short-lived helpers on one
	// database, where only the service should run jobs while it is alive.
	//
	// nil means always eligible, which is exactly how every worker behaved
	// before this option existed.
	//
	// Called on the polling goroutine, so it should be cheap and must not
	// block for long — a slow check delays every cycle. It must not panic:
	// this package does not recover panics anywhere, here or in handlers, and
	// one raised here would take down the polling goroutine.
	//
	// Interaction with StopWhenEmpty, stated because it is a real corner: a
	// gated cycle never inspects the queue, so it cannot conclude the queue is
	// drained and will not trigger StopWhenEmpty. A worker told it may not
	// reserve is deferring, not finished, and it waits rather than exiting.
	CanReserve func(ctx context.Context) bool

	// Controller optionally supports reversible whole-cycle draining. It gates
	// before CanReserve and retains cycles through durable settlement and
	// callbacks. Driver errors quarantine new cycles; nil preserves legacy
	// behavior. One controller belongs to one worker, including across Starts.
	Controller *CycleController

	// Lifecycle callbacks.
	OnProcessing func(job *QueuedJob)
	OnProcessed  func(job *QueuedJob)
	OnFailed     func(job *QueuedJob, err error)
	OnError      func(err error)
}

WorkerOpts configures worker behavior.