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, 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

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 Handler
• 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 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 Handler

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

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

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

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

WorkerOpts configures worker behavior.