queue is a queue and workflow library with pluggable backends and runtime extensions.

Installation

go get github.com/goforj/queue

Existing deployments upgrading to application-type direct delivery must replace workers before switching producers. Follow the direct delivery migration guide, including its SQL schema step and backend-specific rollback constraints.

Applications upgrading from the retired bus or queuefake packages should follow the legacy API migration guide. The removed Temporal compatibility adapter has no root-package replacement.

Queue and its driver modules require Go 1.27 or newer. Projects on older Go releases can pin the v0.2 release line. The PostgreSQL driver continues to use pgx 5.9.2 or newer, the first release containing the GO-2026-5004 security fix.

Quick Start

import (
	"context"
	"fmt"

	"github.com/goforj/queue"
)

func main() {
	q, _ := queue.NewWorkerpool(
		queue.WithWorkers(2), // optional; default: runtime.NumCPU() (min 1)
	)
	type EmailPayload struct {
		To string `json:"to"`
	}

	q.Register("emails:send", func(ctx context.Context, m queue.Message) error {
		var payload EmailPayload
		_ = m.Bind(&payload)
		fmt.Println("send to", payload.To)
		return nil
	})

	_ = q.StartWorkers(context.Background())
	defer q.Shutdown(context.Background())

	_, _ = q.Dispatch(
		queue.NewJob("emails:send").
			Payload(EmailPayload{To: "[email protected]"}),
	)
}

Drivers

Each driver is thoroughly tested against the shared test suite using testcontainers or emulators where appropriate.

Driver / Backend	Mode	Notes	Durable	Async	Delay	Unique	Backoff	Timeout	Native Stats	Queue Admin
	Drop-only	Discards dispatched jobs; useful for disabled queue modes and smoke tests.	-	-	-	Instance	-	-	-	-
	Inline (caller)	Deterministic local execution with no external infra.	-	-	-	Instance	-	✓	-	-
	In-process pool	Local async behavior without external broker/database.	-	✓	✓	Instance	✓	✓	-	-
	SQL durable queue	MySQL driver module (driver/mysqlqueue) built on shared SQL queue core.	✓	✓	✓	Backend	✓	✓	✓	-
	SQL durable queue	Postgres driver module (driver/postgresqueue) built on shared SQL queue core.	✓	✓	✓	Backend	✓	✓	✓	-
	SQL durable queue	SQLite driver module (driver/sqlitequeue) built on shared SQL queue core.	✓	✓	✓	Backend	✓	✓	✓	-
	Redis/Asynq	Production Redis backend (Asynq semantics).	✓	✓	✓	Backend	-	✓	✓	✓
	Ephemeral broker	Core NATS subject routing; every plain subscription can receive a broadcast copy.	-	✓	✓	Instance	✓	✓	-	-
	Broker target	AWS SQS transport with endpoint overrides for localstack/testing.	-	✓	✓	Instance	✓	✓	-	-
	Broker target	RabbitMQ transport and worker consumption.	-	✓	✓	Instance	✓	✓	-	-

SQL-backed queues (sqlite, mysql, postgres) are durable and convenient, but they trade throughput for operational simplicity. They default to 1 worker, and increasing concurrency may require DB tuning (indexes, connection pool, lock contention). Prefer broker-backed drivers for higher-throughput workloads.

Unique scope: Instance suppresses duplicates only within one queue runtime instance; Backend shares claims through the configured database or Redis service. Identity is the effective queue, logical application job type, and canonical serialized payload. Absent, zero-byte, and exact JSON null payloads share one absence identity; generated workflow IDs and delivery options do not change it. See docs/backend-guarantees.md for acceptance, rollout, and failure-boundary details.

Queue Admin status: the cross-driver admin contract is defined in core (ListJobs, RetryJob, CancelJob, DeleteJob, ClearQueue, QueueHistory), but full queue admin operations are currently implemented only for Redis. Other drivers return ErrQueueAdminUnsupported for unsupported admin actions.

Driver constructor quick examples

Use root constructors for in-process backends, and driver-module constructors for external backends. See the Driver Constructors API section below for full constructor shapes (New(...) and NewWithConfig(...)). Driver backends live in separate packages so applications only import/link the optional backend dependencies they actually use (smaller builds, less dependency overhead, cleaner deploys).

package main

import (
	"github.com/goforj/queue"
	"github.com/goforj/queue/driver/mysqlqueue"
	"github.com/goforj/queue/driver/natsqueue"
	"github.com/goforj/queue/driver/postgresqueue"
	"github.com/goforj/queue/driver/rabbitmqqueue"
	"github.com/goforj/queue/driver/redisqueue"
	"github.com/goforj/queue/driver/sqlitequeue"
	"github.com/goforj/queue/driver/sqsqueue"
)

func main() {
	queue.NewSync()       // in-process sync
	queue.NewWorkerpool() // in-process worker pool
	queue.NewNull()       // drop-only / disabled mode

	sqlitequeue.New("file:queue.db?_busy_timeout=5000") // SQL durable queue (SQLite)
	mysqlqueue.New("user:pass@tcp(127.0.0.1:3306)/app") // SQL durable queue (MySQL)
	postgresqueue.New("postgres://user:[email protected]:5432/app?sslmode=disable") // SQL durable queue (Postgres)

	redisqueue.New("127.0.0.1:6379") // Redis/Asynq
	natsqueue.New("nats://127.0.0.1:4222") // NATS
	sqsqueue.New("us-east-1") // SQS
	rabbitmqqueue.New("amqp://guest:[email protected]:5672/") // RabbitMQ
}

Quick Start (Advanced: Workflows)

import (
	"context"

	"github.com/goforj/queue"
)

type EmailPayload struct {
	ID int `json:"id"`
}

func main() {
	q, _ := queue.NewWorkerpool()

	q.Register("reports:generate", func(ctx context.Context, m queue.Message) error {
		return nil
	})
	q.Register("reports:upload", func(ctx context.Context, m queue.Message) error {
		var payload EmailPayload
		if err := m.Bind(&payload); err != nil {
			return err
		}
		return nil
	})
	q.Register("users:notify_report_ready", func(ctx context.Context, m queue.Message) error {
		return nil
	})

	_ = q.StartWorkers(context.Background())
	defer q.Shutdown(context.Background())

	chainID, _ := q.Chain(
		// 1) generate report data
		queue.NewJob("reports:generate").Payload(map[string]any{"report_id": "rpt_123"}),
		// 2) upload report artifact after generate succeeds
		queue.NewJob("reports:upload").Payload(EmailPayload{ID: 123}),
		// 3) notify user only after upload succeeds
		queue.NewJob("users:notify_report_ready").Payload(map[string]any{"user_id": 123}),
	).OnQueue("critical").Dispatch(context.Background())
	_ = chainID
}

Run as a Worker Service

Use Run(ctx) for long-lived workers: it starts processing, waits for shutdown signals, and performs graceful termination.

import (
	"context"
	"log"
	"os/signal"
	"syscall"

	"github.com/goforj/queue"
)

func main() {
	q, _ := queue.NewWorkerpool()

	// Register handlers before starting workers.
	q.Register("emails:send", func(ctx context.Context, m queue.Message) error {
		return nil
	})

	// Create a context that is canceled on SIGINT/SIGTERM (Ctrl+C, container stop).
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Run starts workers, blocks until ctx is canceled, then gracefully shuts down.
	if err := q.Run(ctx); err != nil {
		log.Fatal(err)
	}
}

Core Concepts

Job: Typed work unit for app handlers.

_, _ = q.Dispatch(
	queue.NewJob("emails:send").Payload(EmailPayload{To: "[email protected]"}),
)

Chain: Ordered workflow (A then B then C).

_, _ = q.Chain(
	queue.NewJob("reports:generate"),
	queue.NewJob("reports:upload"),
	queue.NewJob("users:notify_report_ready"),
).Dispatch(context.Background())

Batch: Parallel workflow with callbacks.

_, _ = q.Batch(
	queue.NewJob("emails:send"),
	queue.NewJob("sms:send"),
).Then(func(context.Context, queue.BatchState) error {
	return nil
}).Dispatch(context.Background())

Middleware: Cross-cutting execution policy.

q, _ := queue.New(
	queue.Config{Driver: queue.DriverWorkerpool},
	queue.WithMiddleware(audit, skipMaintenance, fatalValidation),
)

Events: Lifecycle hooks and observability.

q, _ := queue.New(
	queue.Config{Driver: queue.DriverWorkerpool},
	queue.WithObserver(queue.NewStatsCollector()),
)

Backends: Driver/runtime transport selection.

q, _ := queue.NewWorkerpool()
rq, _ := redisqueue.New("127.0.0.1:6379")
_, _ = q, rq

Job builder options

// Define a struct for your job payload.
type EmailPayload struct {
	ID int `json:"id"`
	To string `json:"to"`
}

// Fluent builder pattern for job options.
job := queue.NewJob("emails:send").
	// Payload can be bytes, structs, maps, or JSON-marshalable values.
	// Default payload is empty.
	Payload(EmailPayload{ID: 123, To: "[email protected]"}).
	// OnQueue sets the queue name.
	// Default is empty; broker-style drivers expect an explicit queue.
	OnQueue("default").
	// Timeout sets per-job execution timeout.
	// Default is unset; some drivers may apply driver/runtime defaults.
	Timeout(20 * time.Second).
	// Retry sets max retries.
	// Default is 0, which means one total attempt.
	Retry(3).
	// Backoff sets retry delay.
	// Default is unset; Redis dispatch returns ErrBackoffUnsupported.
	Backoff(500 * time.Millisecond).
	// Delay schedules first execution in the future.
	// Default is 0 (run immediately).
	Delay(2 * time.Second).
	// UniqueFor deduplicates Type+Payload for a TTL window.
	// Default is 0 (no dedupe).
	UniqueFor(45 * time.Second)

// Dispatch the job to the queue.
_, _ = q.Dispatch(job)

// In handlers, use Bind to decode payload into a struct.
q.Register("emails:send", func(ctx context.Context, m queue.Message) error {
	var payload EmailPayload
	if err := m.Bind(&payload); err != nil {
		return err
	}
	return nil
})

Benchmarks

Run local + integration-backed benchmarks (requires Docker/testcontainers):

cd docs && GOWORK=off INTEGRATION_BACKEND=all GOCACHE=/tmp/queue-gocache go test -tags=benchrender ./bench -run '^TestRenderBenchmarks$'

Latency (ns/op)

Throughput (ops/s)

Allocated Bytes (B/op)

Allocations (allocs/op)

Tables

Class	Driver	ns/op	ops/s	B/op	allocs/op
External	nats	774	1291823	1258	13
External	redis	95295	10494	2113	33
External	rabbitmq	165780	6032	1882	57
External	sqlite	202380	4941	1931	47
External	postgres	1056731	946	3809	78
External	sqs	1873911	534	94784	1082
External	mysql	2286406	437	3303	62
Local	null	37	26673780	128	1
Local	sync	282	3539823	408	6
Local	workerpool	650	1538462	456	7

Middleware

Use queue.WithMiddleware(...) to apply cross-cutting workflow behavior to workflow job execution (logging, filtering, and error policy).

Common patterns:

• wrap handler execution (before/after logging, timing, tracing)
• skip jobs conditionally (maintenance mode, feature flags)
• convert matched errors into terminal failures (no retry)

var errValidation = errors.New("validation failed")
maintenanceMode := false

audit := queue.MiddlewareFunc(func(ctx context.Context, m queue.Message, next queue.Next) error {
	log.Printf("start job=%s", m.JobType)
	err := next(ctx, m)
	log.Printf("done job=%s err=%v", m.JobType, err)
	return err
})

skipMaintenance := queue.SkipWhen{
	Predicate: func(context.Context, queue.Message) bool {
		return maintenanceMode
	},
}

fatalValidation := queue.FailOnError{
	When: func(err error) bool {
		return errors.Is(err, errValidation)
	},
}

q, _ := queue.New(
	queue.Config{Driver: queue.DriverWorkerpool},
	queue.WithMiddleware(audit, skipMaintenance, fatalValidation),
)
_ = q

Observability

Use queue.Observer implementations to capture normalized runtime events across drivers.

collector := queue.NewStatsCollector()
observer := queue.MultiObserver(
    collector,
    queue.ObserverFunc(func(_ context.Context, event queue.Event) {
        _ = event.Kind
    }),
)

q, _ := queue.New(
    queue.Config{Driver: queue.DriverWorkerpool},
    queue.WithObserver(observer),
)
_ = q

Distributed counters and source of truth

• StatsCollector counters are process-local and event-driven.
• In multi-process deployments, aggregate metrics externally (OTel/Prometheus/etc.).
• Prefer backend-native stats when available.
• queue.SupportsNativeStats(q) indicates native driver snapshot support.
• queue.Snapshot(ctx, q, collector) merges native + collector where possible.

Compose observers

events := make(chan queue.Event, 100)
collector := queue.NewStatsCollector()
observer := queue.MultiObserver(
    collector,
    queue.ChannelObserver{
        Events:     events,
        DropIfFull: true,
    },
    queue.ObserverFunc(func(_ context.Context, e queue.Event) {
        _ = e
    }),
)

q, _ := queue.New(
    queue.Config{Driver: queue.DriverWorkerpool},
    queue.WithObserver(observer),
)
_ = q

Kitchen sink event logging

Runnable example: examples/observeall/main.go

logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
observer := queue.ObserverFunc(func(ctx context.Context, event queue.Event) {
	logger.InfoContext(ctx, "queue event",
		"layer", event.Layer,
		"kind", event.Kind,
		"driver", event.Driver,
		"queue", event.Queue,
		"dispatch_id", event.DispatchID,
		"job_id", event.JobID,
		"chain_id", event.ChainID,
		"batch_id", event.BatchID,
		"job_type", event.JobType,
		"attempt", event.Attempt,
		"max_retry", event.MaxRetry,
		"duration", event.Duration,
		"err", event.Err,
	)
})

q, _ := queue.New(
	queue.Config{Driver: queue.DriverSync},
	queue.WithObserver(observer),
)
_ = q

Events reference

Layer	EventKind	Meaning
queue	dispatch_started	After job validation, the root facade began a public dispatch.
queue	dispatch_succeeded	The backend accepted the public dispatch; synchronous execution may still return an application error.
queue	dispatch_failed	Public dispatch ended before backend acceptance.
queue	enqueue_accepted	The driver confirmed enqueue acceptance.
queue	enqueue_rejected	The driver rejected enqueue with an error.
queue	enqueue_duplicate	Uniqueness policy rejected the logical job as a duplicate.
queue	enqueue_canceled	Context cancellation or deadline expiry prevented enqueue.
worker	process_started	A physical handler attempt began.
worker	process_succeeded	Handler success crossed the driver's settlement boundary; runtimes without a settlement hook emit after handler return.
worker	process_failed	A handler attempt returned an error or panicked.
worker	process_retried	A numbered application retry attempt began; infrastructure redelivery may repeat the fact.
worker	process_archived	A SQL driver confirmed its fenced transition to terminal dead state; other built-in runtimes omit it.
worker	process_recovered	SQL bulk recovery requeued one stale in-flight claim; identity fields are unavailable at that boundary.
worker	republish_failed	An internal delay or retry replacement could not be published.
worker	settlement_failed	Delivery finalization, acknowledgement, deletion, or negative settlement failed or was ambiguous; redelivery remains possible.
queue	queue_paused	A supporting driver confirmed queue consumption was paused.
queue	queue_resumed	A supporting driver confirmed queue consumption was resumed.
workflow	job_started	A logical execution attempt began before handler lookup.
workflow	job_succeeded	Logical job success committed; settlement-aware drivers publish the fact only after settlement.
workflow	job_failed	A logical job reached permanent or exhausted failure.
workflow	chain_started	A chain record was created and initial dispatch began.
workflow	chain_advanced	A committed node outcome advanced the chain to its next node.
workflow	chain_completed	The final chain node committed terminal success.
workflow	chain_failed	A chain committed terminal failure.
workflow	batch_started	A batch record was created and initial member dispatch began.
workflow	batch_progressed	A batch member committed a terminal outcome.
workflow	batch_completed	Batch reached terminal success (or allowed-failure completion).
workflow	batch_failed	A non-allowed member failure or initial member dispatch rejection committed terminal batch failure.
workflow	batch_cancelled	Remaining batch work was cancelled after terminal failure.
workflow	callback_started	A claimed terminal Catch, Then, or Finally callback began execution.
workflow	callback_succeeded	Terminal callback success crossed the applicable settlement boundary.
workflow	callback_failed	A terminal callback was invalid or unavailable, returned an error, or panicked.

Handler panics now emit process_failed before the original panic value is rethrown. This adds truthful failure telemetry without changing backend panic recovery or retry behavior.

Callback lifecycle facts cover terminal Catch, Then, and Finally callbacks. Inline Batch.Progress closures do not emit callback_* facts.

Examples

Runnable examples live in the separate examples module (./examples). They are not included when applications import github.com/goforj/queue, which keeps dependency graphs and build/link overhead smaller.

Admin Support

Queue admin APIs are part of the core contract so additional drivers can implement them over time. At this time, full admin operations (ListJobs, RetryJob, CancelJob, DeleteJob, ClearQueue) are Redis-only. Use queue.SupportsQueueAdmin(q) (or handle queue.ErrQueueAdminUnsupported) to gate admin workflows per runtime.

API reference

The API section below is autogenerated; do not edit between the markers.

API Index

Group	Functions
Admin	CancelJob · Queue.CancelJob · ClearQueue · Queue.ClearQueue · DeleteJob · Queue.DeleteJob · History · ListJobs · Queue.ListJobs · Normalize · QueueHistory · RetryJob · Queue.RetryJob · SinglePointHistory · SupportsQueueAdmin · TimelineHistoryFromSnapshot
Constructors	New · NewMemoryStore · NewMessage · NewNull · NewSQLStore · NewSQLStoreWithManagedSchema · NewStatsCollector · NewSync · NewWorkerpool
Job	Backoff · Bind · Delay · NewJob · OnQueue · Payload · PayloadAs · PayloadBytes · PayloadJSON · Retry · Timeout · UniqueFor
Observability	Active · Archived · Failed · MultiObserver · ChannelObserver.Observe · Observer.Observe · ObserverFunc.Observe · StatsCollector.Observe · Pause · Paused · Pending · Processed · Queue · Queues · Ready · Resume · RetryCount · SafeObserve · Scheduled · Snapshot · StatsCollector.Snapshot · SupportsNativeStats · SupportsPause · Throughput
Other	Acquire · AdvanceChain · Allow · AllowFailures · CancelBatch · BatchBuilder.Catch · ChainBuilder.Catch · CreateBatch · CreateChain · BatchBuilder.Dispatch · ChainBuilder.Dispatch · FailChain · FailChainNode · BatchBuilder.Finally · ChainBuilder.Finally · GetBatch · GetChain · FailOnError.Handle · Middleware.Handle · MiddlewareFunc.Handle · RateLimit.Handle · RetryPolicy.Handle · SkipWhen.Handle · WithoutOverlapping.Handle · MarkBatchJobFailed · MarkBatchJobStarted · MarkBatchJobSucceeded · MarkCallbackInvoked · Name · BatchBuilder.OnQueue · ChainBuilder.OnQueue · PhysicalQueueName · PhysicalQueueWeights · Progress · Prune · Release · ResolveObservedJobType · SettleBatchJob · Then
Queue	Batch · Bind · Chain · Dispatch · Driver · FindBatch · FindChain · IsPermanent · Pause · PayloadAs · PayloadBytes · Permanent · Prune · Ready · Register · Resume · Run · Shutdown · StartWorkers · Stats · WithClock · WithContext · WithHandlerContextDecorator · WithLegacyDirectEnvelope · WithMiddleware · WithObserver · WithStore · WithWorkers · Queue.WithWorkers
Driver Constructors	mysqlqueue.New · mysqlqueue.NewWithConfig · natsqueue.New · natsqueue.NewWithConfig · postgresqueue.New · postgresqueue.NewWithConfig · rabbitmqqueue.New · rabbitmqqueue.NewWithConfig · redisqueue.New · redisqueue.NewWithConfig · sqlitequeue.New · sqlitequeue.NewWithConfig · sqsqueue.New · sqsqueue.NewWithConfig
Testing	AssertBatchCount · AssertBatched · AssertChained · AssertCount · AssertDispatched · AssertDispatchedOn · AssertDispatchedTimes · AssertNotDispatched · AssertNothingBatched · AssertNothingDispatched · Batch · BatchRecords · Chain · ChainRecords · Dispatch · Driver · FindBatch · FindChain · NewFake · Prune · Ready · Records · Register · Reset · Shutdown · StartWorkers · WithContext · Workers

API

Admin

CancelJob

CancelJob cancels a job when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
err = queue.CancelJob(context.Background(), q, "job-id")
_ = err

Queue.CancelJob

CancelJob cancels a job via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
if !queue.SupportsQueueAdmin(q) {
	return
}
err = q.CancelJob(context.Background(), "job-id")
_ = err

ClearQueue

ClearQueue clears queue jobs when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
err = queue.ClearQueue(context.Background(), q, "default")
_ = err

Queue.ClearQueue

ClearQueue clears queue jobs via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
if !queue.SupportsQueueAdmin(q) {
	return
}
err = q.ClearQueue(context.Background(), "default")
_ = err

DeleteJob

DeleteJob deletes a job when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
err = queue.DeleteJob(context.Background(), q, "default", "job-id")
_ = err

Queue.DeleteJob

DeleteJob deletes a job via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
if !queue.SupportsQueueAdmin(q) {
	return
}
err = q.DeleteJob(context.Background(), "default", "job-id")
_ = err

History

History returns queue history points via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
points, err := q.History(context.Background(), "default", queue.QueueHistoryHour)
_ = err

ListJobs

ListJobs lists jobs for a queue and state when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
_, err = queue.ListJobs(context.Background(), q, queue.ListJobsOptions{
	Queue: "default",
	State: queue.JobStatePending,
})
_ = err

Queue.ListJobs

ListJobs lists jobs via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
_, err = q.ListJobs(context.Background(), queue.ListJobsOptions{
	Queue: "default",
	State: queue.JobStatePending,
})
_ = err

Normalize

Normalize returns a safe options payload with defaults applied.

opts := queue.ListJobsOptions{Queue: "", State: "", Page: 0, PageSize: 1000}
normalized := opts.Normalize()
fmt.Println(normalized.Queue, normalized.State, normalized.Page, normalized.PageSize)
// Output: default pending 1 500

QueueHistory

QueueHistory returns queue history points when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
_, err = queue.QueueHistory(context.Background(), q, "default", queue.QueueHistoryHour)
_ = err

RetryJob

RetryJob retries (runs now) a job when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
err = queue.RetryJob(context.Background(), q, "default", "job-id")
_ = err

Queue.RetryJob

RetryJob retries (runs now) a job via queue admin capability when supported.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
if !queue.SupportsQueueAdmin(q) {
	return
}
err = q.RetryJob(context.Background(), "default", "job-id")
_ = err

SinglePointHistory

SinglePointHistory converts a snapshot into a single current-history point. This helper is intended for driver modules that do not expose historical buckets.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Processed: 12, Failed: 1},
	},
}
points := queue.SinglePointHistory(snapshot, "default")
fmt.Println(len(points), points[0].Processed, points[0].Failed)
// Output: 1 12 1

SupportsQueueAdmin

SupportsQueueAdmin reports whether queue admin operations are available.

q, err := redisqueue.New("127.0.0.1:6379")
if err != nil {
	return
}
fmt.Println(queue.SupportsQueueAdmin(q))
// Output: true

TimelineHistoryFromSnapshot

TimelineHistoryFromSnapshot records queue counters and returns windowed points. This is intended for drivers that don't expose native multi-point history.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Processed: 5, Failed: 1},
	},
}
points := queue.TimelineHistoryFromSnapshot(snapshot, "default", queue.QueueHistoryHour)
fmt.Println(len(points) >= 1)
// Output: true

Constructors

queue.New

New creates the high-level Queue API based on Config.Driver.

q, err := queue.New(queue.Config{Driver: queue.DriverWorkerpool})
if err != nil {
	return
}
type EmailPayload struct {
	ID int `json:"id"`
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error {
	var payload EmailPayload
	if err := m.Bind(&payload); err != nil {
		return err
	}
	_ = payload
	return nil
})
_ = q.WithWorkers(1).StartWorkers(context.Background()) // optional; default: runtime.NumCPU() (min 1)
defer q.Shutdown(context.Background())
_, _ = q.Dispatch(
	queue.NewJob("emails:send").
		Payload(EmailPayload{ID: 1}).
		OnQueue("default"),
)

NewMemoryStore

NewMemoryStore creates an in-memory workflow state store. It copies chain nodes and payload bytes on creation and return so callers retain independent ownership.

NewMessage

NewMessage creates a logical queue message from an application job type and exact payload bytes. The payload is copied so callers can safely reuse or mutate their input buffer.

NewNull

NewNull creates a Queue on the null backend.

q, err := queue.NewNull()
if err != nil {
	return
}

NewSQLStore

NewSQLStore creates a SQL-backed workflow state store.

NewSQLStoreWithManagedSchema

NewSQLStoreWithManagedSchema creates a SQL-backed workflow state store without executing schema DDL. The supplied database must already contain the dialect-correct workflow tables, including transition receipts.

NewStatsCollector

NewStatsCollector creates an event collector for queue counters.

collector := queue.NewStatsCollector()

NewSync

NewSync creates a Queue on the synchronous in-process backend.

q, err := queue.NewSync()
if err != nil {
	return
}

NewWorkerpool

NewWorkerpool creates a Queue on the in-process workerpool backend.

q, err := queue.NewWorkerpool()
if err != nil {
	return
}

Job

Backoff

Backoff sets delay between retries.

job := queue.NewJob("emails:send").Backoff(500 * time.Millisecond)

Job.Bind

Bind unmarshals job payload JSON into dst.

type EmailPayload struct {
	ID int    `json:"id"`
	To string `json:"to"`
}
job := queue.NewJob("emails:send").Payload(EmailPayload{
	ID: 1,
	To: "[email protected]",
})
var payload EmailPayload
if err := job.Bind(&payload); err != nil {
	return
}
_ = payload.To

Delay

Delay defers execution by duration.

job := queue.NewJob("emails:send").Delay(300 * time.Millisecond)

NewJob

NewJob creates a job value with a required job type.

job := queue.NewJob("emails:send")

Job.OnQueue

OnQueue sets the target queue name.

job := queue.NewJob("emails:send").OnQueue("critical")

Payload

Payload sets job payload from common value types.

Example: payload bytes

jobBytes := queue.NewJob("emails:send").Payload([]byte(`{"id":1}`))

Example: payload struct

type Meta struct {
	Nested bool `json:"nested"`
}
type EmailPayload struct {
	ID   int    `json:"id"`
	To   string `json:"to"`
	Meta Meta   `json:"meta"`
}
jobStruct := queue.NewJob("emails:send").Payload(EmailPayload{
	ID:   1,
	To:   "[email protected]",
	Meta: Meta{Nested: true},
})

Example: payload map

jobMap := queue.NewJob("emails:send").Payload(map[string]any{
	"id":  1,
	"to":  "[email protected]",
	"meta": map[string]any{"nested": true},
})

Job.PayloadAs

PayloadAs unmarshals the job payload JSON and returns it as T.

type EmailPayload struct {
	To string `json:"to"`
}
job := queue.NewJob("emails:send").Payload(EmailPayload{To: "[email protected]"})
payload, err := job.PayloadAs[EmailPayload]()
fmt.Println(err == nil, payload.To)
// true [email protected]

Job.PayloadBytes

PayloadBytes returns a copy of job payload bytes.

job := queue.NewJob("emails:send").Payload([]byte(`{"id":1}`))
payload := job.PayloadBytes()

PayloadJSON

PayloadJSON marshals payload as JSON.

job := queue.NewJob("emails:send").PayloadJSON(map[string]int{"id": 1})

Retry

Retry sets max retry attempts.

job := queue.NewJob("emails:send").Retry(4)

Timeout

Timeout sets per-job execution timeout.

job := queue.NewJob("emails:send").Timeout(10 * time.Second)

UniqueFor

UniqueFor enables uniqueness dedupe within the given TTL.

job := queue.NewJob("emails:send").UniqueFor(45 * time.Second)

Observability

Active

Active returns active count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Active: 2},
	},
}
fmt.Println(snapshot.Active("default"))
// Output: 2

Archived

Archived returns archived count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Archived: 7},
	},
}
fmt.Println(snapshot.Archived("default"))
// Output: 7

Failed

Failed returns failed count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Failed: 2},
	},
}
fmt.Println(snapshot.Failed("default"))
// Output: 2

MultiObserver

MultiObserver fans out events to multiple observers.

events := make(chan queue.Event, 2)
observer := queue.MultiObserver(
	queue.ChannelObserver{Events: events},
	queue.ObserverFunc(func(context.Context, queue.Event) {}),
)
observer.Observe(context.Background(), queue.Event{Kind: queue.EventEnqueueAccepted})
fmt.Println(len(events))
// Output: 1

ChannelObserver.Observe

Observe forwards an event to the configured channel.

ch := make(chan queue.Event, 1)
observer := queue.ChannelObserver{Events: ch}
observer.Observe(context.Background(), queue.Event{Kind: queue.EventProcessStarted, Queue: "default"})
event := <-ch

Observer.Observe

Observe handles a queue runtime event.

var observer queue.Observer
observer.Observe(context.Background(), queue.Event{
	Kind:   queue.EventEnqueueAccepted,
	Driver: queue.DriverSync,
	Queue:  "default",
})

ObserverFunc.Observe

Observe calls the wrapped function.

logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
observer := queue.ObserverFunc(func(ctx context.Context, event queue.Event) {
	logger.Info("queue event",
		"kind", event.Kind,
		"driver", event.Driver,
		"queue", event.Queue,
		"job_type", event.JobType,
		"attempt", event.Attempt,
		"max_retry", event.MaxRetry,
		"duration", event.Duration,
		"err", event.Err,
	)
})
observer.Observe(context.Background(), queue.Event{
	Kind:     queue.EventProcessSucceeded,
	Driver:   queue.DriverSync,
	Queue:    "default",
	JobType: "emails:send",
})

StatsCollector.Observe

Observe records an event and updates normalized counters.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventEnqueueAccepted,
	Driver: queue.DriverSync,
	Queue:  "default",
	Time:   time.Now(),
})

Pause

Pause pauses queue consumption for drivers that support it.

q, _ := queue.NewSync()
_ = queue.Pause(context.Background(), q, "default")
snapshot, _ := queue.Snapshot(context.Background(), q, nil)
fmt.Println(snapshot.Paused("default"))
// Output: 1

Paused

Paused returns the observed pause state for a queue as zero or one.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventQueuePaused,
	Driver: queue.DriverSync,
	Queue:  "default",
	Time:   time.Now(),
})
snapshot := collector.Snapshot()
fmt.Println(snapshot.Paused("default"))
// Output: 1

Pending

Pending returns pending count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Pending: 3},
	},
}
fmt.Println(snapshot.Pending("default"))
// Output: 3

Processed

Processed returns processed count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Processed: 11},
	},
}
fmt.Println(snapshot.Processed("default"))
// Output: 11

Queue

Queue returns queue counters for a queue name.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventEnqueueAccepted,
	Driver: queue.DriverSync,
	Queue:  "default",
	Time:   time.Now(),
})
snapshot := collector.Snapshot()
counters, ok := snapshot.Queue("default")
fmt.Println(ok, counters.Pending)
// Output: true 1

Queues

Queues returns sorted queue names present in the snapshot.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventEnqueueAccepted,
	Driver: queue.DriverSync,
	Queue:  "critical",
	Time:   time.Now(),
})
snapshot := collector.Snapshot()
names := snapshot.Queues()
fmt.Println(len(names), names[0])
// Output: 1 critical

Ready

Ready validates backend readiness for the provided queue runtime.

q, _ := queue.NewSync()
fmt.Println(queue.Ready(context.Background(), q) == nil)
// true

Resume

Resume resumes queue consumption for drivers that support it.

q, _ := queue.NewSync()
_ = queue.Pause(context.Background(), q, "default")
_ = queue.Resume(context.Background(), q, "default")
snapshot, _ := queue.Snapshot(context.Background(), q, nil)
fmt.Println(snapshot.Paused("default"))
// Output: 0

RetryCount

RetryCount returns retry count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Retry: 1},
	},
}
fmt.Println(snapshot.RetryCount("default"))
// Output: 1

SafeObserve

SafeObserve delivers an event to an observer and recovers observer panics.

This is an advanced helper intended for driver-module implementations.

Scheduled

Scheduled returns scheduled count for a queue.

snapshot := queue.StatsSnapshot{
	ByQueue: map[string]queue.QueueCounters{
		"default": {Scheduled: 4},
	},
}
fmt.Println(snapshot.Scheduled("default"))
// Output: 4

Snapshot

Snapshot returns driver-native stats, falling back to collector data.

q, _ := queue.NewSync()
snapshot, _ := q.Stats(context.Background())
_, ok := snapshot.Queue("default")
fmt.Println(ok)
// Output: true

StatsCollector.Snapshot

Snapshot returns a copy of collected counters.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventEnqueueAccepted,
	Driver: queue.DriverSync,
	Queue:  "default",
	Time:   time.Now(),
})
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventProcessStarted,
	Driver: queue.DriverSync,
	Queue:  "default",
	JobKey: "job-1",
	Time:   time.Now(),
})
collector.Observe(context.Background(), queue.Event{
	Kind:     queue.EventProcessSucceeded,
	Driver:   queue.DriverSync,
	Queue:    "default",
	JobKey:  "job-1",
	Duration: 12 * time.Millisecond,
	Time:     time.Now(),
})
snapshot := collector.Snapshot()
counters, _ := snapshot.Queue("default")
throughput, _ := snapshot.Throughput("default")
fmt.Printf("queues=%v\n", snapshot.Queues())
fmt.Printf("counters=%+v\n", counters)
fmt.Printf("hour=%+v\n", throughput.Hour)
// Output:
// queues=[default]
// counters={Pending:0 Active:0 Scheduled:0 Retry:0 Archived:0 Processed:1 Failed:0 Paused:0 AvgWait:0s AvgRun:12ms}
// hour={Processed:1 Failed:0}

SupportsNativeStats

SupportsNativeStats reports whether a queue runtime exposes native stats snapshots.

q, _ := queue.NewSync()
fmt.Println(queue.SupportsNativeStats(q))
// Output: true

SupportsPause

SupportsPause reports whether a queue runtime supports Pause/Resume.

q, _ := queue.NewSync()
fmt.Println(queue.SupportsPause(q))
// Output: true

Throughput

Throughput returns rolling throughput windows for a queue name.

collector := queue.NewStatsCollector()
collector.Observe(context.Background(), queue.Event{
	Kind:   queue.EventProcessSucceeded,
	Driver: queue.DriverSync,
	Queue:  "default",
	Time:   time.Now(),
})
snapshot := collector.Snapshot()
throughput, ok := snapshot.Throughput("default")
fmt.Printf("ok=%v hour=%+v day=%+v week=%+v\n", ok, throughput.Hour, throughput.Day, throughput.Week)
// Output: ok=true hour={Processed:1 Failed:0} day={Processed:1 Failed:0} week={Processed:1 Failed:0}

Other

Acquire

Acquire attempts to hold key for ttl.

AdvanceChain

AdvanceChain atomically claims completedNode and returns the current successor. Repeating the same (chainID, completedNode) claim must not advance again. When done is true, GetChain must immediately expose Completed or Failed state.

Allow

Allow returns whether key may execute and any suggested retry delay.

AllowFailures

AllowFailures keeps remaining members active after a terminal member failure.

CancelBatch

CancelBatch commits aggregate batch cancellation.

BatchBuilder.Catch

Catch registers the explicitly ephemeral batch failure callback.

ChainBuilder.Catch

Catch registers the explicitly ephemeral chain failure callback.

CreateBatch

CreateBatch persists a newly accepted batch. BatchID and every JobID must be non-empty, Jobs must contain at least one entry, and JobIDs must be unique.

CreateChain

CreateChain persists a newly accepted chain. ChainID and every NodeID must be non-empty, Nodes must contain at least one entry, and NodeIDs must be unique.

BatchBuilder.Dispatch

Dispatch persists and starts the batch workflow.

ChainBuilder.Dispatch

Dispatch persists and starts the chain workflow.

FailChain

FailChain commits terminal failure without replacing completed state.

FailChainNode

FailChainNode commits failure only while nodeID is the current unsettled node. owned remains true on replay while that node's failure owns the chain.

BatchBuilder.Finally

Finally registers the explicitly ephemeral batch terminal callback.

ChainBuilder.Finally

Finally registers the explicitly ephemeral chain terminal callback.

GetBatch

GetBatch returns current batch state.

GetChain

GetChain returns current chain state.

FailOnError.Handle

Handle wraps matched errors as fatal errors to stop retries.

Middleware.Handle

Handle wraps the remaining middleware and handler chain.

MiddlewareFunc.Handle

Handle calls the wrapped middleware function.

RateLimit.Handle

Handle applies limiter checks before executing the next handler.

RetryPolicy.Handle

Handle passes execution through without modification.

SkipWhen.Handle

Handle skips job execution when Predicate returns true.

WithoutOverlapping.Handle

Handle acquires a lock and prevents concurrent overlap for the same key.

MarkBatchJobFailed

MarkBatchJobFailed commits the first outcome for (batchID, jobID). Duplicate outcomes must return current state without changing counters.

MarkBatchJobStarted

MarkBatchJobStarted records that one batch member began execution.

MarkBatchJobSucceeded

MarkBatchJobSucceeded commits the first outcome for (batchID, jobID). Duplicate outcomes must return current state without changing counters.

MarkCallbackInvoked

MarkCallbackInvoked atomically claims one callback idempotency key.

Name

Name assigns an application-facing label to the batch.

BatchBuilder.OnQueue

OnQueue applies a default queue to batch jobs without an explicit target.

ChainBuilder.OnQueue

OnQueue applies a default queue to chain jobs without an explicit target.

PhysicalQueueName

PhysicalQueueName maps a logical queue name into the physical name used by the backing queue driver.

PhysicalQueueWeights

PhysicalQueueWeights maps logical weighted queue names into their physical backend names.

Progress

Progress registers the explicitly ephemeral batch progress callback.

WorkflowStore.Prune

Prune removes terminal workflow state older than before.

Release

Release relinquishes the acquired lease.

ResolveObservedJobType

ResolveObservedJobType returns the effective application job type that should be emitted to observers. External workers may process private workflow delivery envelopes (for example, "bus:job") whose payload embeds the real application job type. When possible, this helper unwraps that payload so dashboards and metrics reflect the user-facing job type instead of the transport wrapper.

SettleBatchJob

SettleBatchJob returns the first committed outcome for one batch member. owned remains true on same-outcome replay and false when the opposite outcome won. Ownership covers the outcome category; BatchState does not retain a per-member cause.

Then

Then registers the explicitly ephemeral batch success callback.

Queue

Queue.Batch

Batch creates a batch builder for fan-out workflow execution.

q, err := queue.NewSync()
if err != nil {
	return
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error { return nil })
if err := q.StartWorkers(context.Background()); err != nil {
	return
}
defer q.Shutdown(context.Background())
_, _ = q.Batch(
	queue.NewJob("emails:send").Payload(map[string]any{"id": 1}),
	queue.NewJob("emails:send").Payload(map[string]any{"id": 2}),
).Name("send-emails").OnQueue("default").Dispatch(context.Background())

Message.Bind

Bind unmarshals the raw job payload into dst.

Queue.Chain

Chain creates a chain builder for sequential workflow execution.

q, err := queue.NewSync()
if err != nil {
	return
}
q.Register("first", func(ctx context.Context, m queue.Message) error { return nil })
q.Register("second", func(ctx context.Context, m queue.Message) error { return nil })
if err := q.StartWorkers(context.Background()); err != nil {
	return
}
defer q.Shutdown(context.Background())
_, _ = q.Chain(
	queue.NewJob("first"),
	queue.NewJob("second"),
).OnQueue("default").Dispatch(context.Background())

Queue.Dispatch

Dispatch enqueues a high-level job using its application type and exact payload bytes together with the queue's bound context.

q, err := queue.NewSync()
if err != nil {
	return
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error { return nil })
if err := q.StartWorkers(context.Background()); err != nil {
	return
}
defer q.Shutdown(context.Background())
job := queue.NewJob("emails:send").Payload(map[string]any{"id": 1}).OnQueue("default")
_, _ = q.Dispatch(job)

Queue.Driver

Driver reports the configured backend driver for the underlying queue runtime.

q, err := queue.NewSync()
if err != nil {
	return
}
fmt.Println(q.Driver())
// Output: sync

Queue.FindBatch

FindBatch returns current batch state by ID.

q, err := queue.NewSync()
if err != nil {
	return
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error { return nil })
batchID, err := q.Batch(queue.NewJob("emails:send")).Dispatch(context.Background())
if err != nil {
	return
}
_, _ = q.FindBatch(context.Background(), batchID)

Queue.FindChain

FindChain returns current chain state by ID.

q, err := queue.NewSync()
if err != nil {
	return
}
q.Register("first", func(ctx context.Context, m queue.Message) error { return nil })
chainID, err := q.Chain(queue.NewJob("first")).Dispatch(context.Background())
if err != nil {
	return
}
_, _ = q.FindChain(context.Background(), chainID)

IsPermanent

IsPermanent reports whether an error requests terminal application settlement.

Queue.Pause

Pause pauses consumption for a queue when supported by the underlying driver. See the README "Queue Backends" table for Pause/Resume support and docs/backend-guarantees.md (Capability Matrix) for broader backend differences.

q, err := queue.NewSync()
if err != nil {
	return
}
if queue.SupportsPause(q) {
	_ = q.Pause(context.Background(), "default")
}

Message.PayloadAs

PayloadAs unmarshals the delivered payload and returns it as T.

type EmailPayload struct {
	To string `json:"to"`
}
message := queue.NewMessage("emails:send", []byte(`{"to":"[email protected]"}`))
payload, err := message.PayloadAs[EmailPayload]()
fmt.Println(err == nil, payload.To)
// true [email protected]

Message.PayloadBytes

PayloadBytes returns an isolated copy of the raw job payload.

Permanent

Permanent marks an error as terminal so workers do not spend the remaining application retry budget on it.

Queue.Prune

Prune deletes old workflow state records.

q, err := queue.NewSync()
if err != nil {
	return
}
_ = q.Prune(context.Background(), time.Now().Add(-24*time.Hour))

Queue.Ready

Ready validates queue backend readiness for dispatch/worker operation.

q, err := queue.NewSync()
if err != nil {
	return
}
fmt.Println(q.Ready(context.Background()) == nil)
// true

Queue.Register

Register binds a handler for a high-level job type.

q, err := queue.NewSync()
if err != nil {
	return
}
type EmailPayload struct {
	ID int `json:"id"`
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error {
	var payload EmailPayload
	if err := m.Bind(&payload); err != nil {
		return err
	}
	_ = payload
	return nil
})

Queue.Resume

Resume resumes consumption for a queue when supported by the underlying driver.

q, err := queue.NewSync()
if err != nil {
	return
}
if queue.SupportsPause(q) {
	_ = q.Resume(context.Background(), "default")
}

Run

Run starts worker processing, blocks until ctx is canceled, then gracefully shuts down.

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
q, err := queue.NewWorkerpool()
if err != nil {
	return
}
q.Register("emails:send", func(ctx context.Context, m queue.Message) error { return nil })
go func() {
	time.Sleep(100 * time.Millisecond)
	cancel()
}()
_ = q.Run(ctx)

Queue.Shutdown

Shutdown drains workers and closes underlying resources.

q, err := queue.NewWorkerpool()
if err != nil {
	return
}
_ = q.StartWorkers(context.Background())
_ = q.Shutdown(context.Background())

Queue.StartWorkers

StartWorkers starts worker processing.

q, err := queue.NewWorkerpool()
if err != nil {
	return
}
_ = q.StartWorkers(context.Background())

Stats

Stats returns a normalized snapshot when supported by the underlying driver.

q, err := queue.NewSync()
if err != nil {
	return
}
if queue.SupportsNativeStats(q) {
	_, _ = q.Stats(context.Background())
}

WithClock

WithClock overrides the workflow runtime clock.

q, err := queue.New(
	queue.Config{Driver: queue.DriverSync},
	queue.WithClock(func() time.Time { return time.Unix(0, 0) }),
)
if err != nil {
	return
}

Queue.WithContext

WithContext returns a derived queue handle bound to ctx.

WithHandlerContextDecorator

WithHandlerContextDecorator decorates queue handler execution context before process lifecycle events and handler execution run.

q, err := queue.New(
	queue.Config{Driver: queue.DriverSync},
	queue.WithHandlerContextDecorator(func(ctx context.Context) context.Context {
		return context.WithValue(ctx, "source", "jobs")
	}),
)
if err != nil {
	return
}

WithLegacyDirectEnvelope

WithLegacyDirectEnvelope keeps ordinary dispatches on the version-one bus:job wire route during a workers-first migration. Remove this option only after every consumer can process canonical direct deliveries. See the direct delivery migration guide for backend-specific rollout and rollback.

WithMiddleware

WithMiddleware appends queue workflow middleware.

mw := queue.MiddlewareFunc(func(ctx context.Context, m queue.Message, next queue.Next) error {
	return next(ctx, m)
})
q, err := queue.New(queue.Config{Driver: queue.DriverSync}, queue.WithMiddleware(mw))
if err != nil {
	return
}

WithObserver

WithObserver installs one observer for queue, worker, and workflow lifecycle events.

observer := queue.ObserverFunc(func(_ context.Context, event queue.Event) {
	_ = event.Kind
})
q, err := queue.New(queue.Config{Driver: queue.DriverSync}, queue.WithObserver(observer))
if err != nil {
	return
}

WithStore

WithStore overrides the workflow orchestration store.

var store queue.WorkflowStore
q, err := queue.New(queue.Config{Driver: queue.DriverSync}, queue.WithStore(store))
if err != nil {
	return
}

WithWorkers

WithWorkers sets desired worker concurrency before StartWorkers. It applies to high-level queue constructors (for example NewWorkerpool/New/NewSync).

q, err := queue.NewWorkerpool(
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

Queue.WithWorkers

WithWorkers sets desired worker concurrency before StartWorkers.

q, err := queue.NewWorkerpool()
if err != nil {
	return
}
q.WithWorkers(4) // optional; default: runtime.NumCPU() (min 1)

Driver Constructors

mysqlqueue

mysqlqueue.New

New creates a high-level Queue using the MySQL SQL backend.

q, err := mysqlqueue.New(
	"user:pass@tcp(127.0.0.1:3306)/queue?parseTime=true",
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

mysqlqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit MySQL SQL driver config.

q, err := mysqlqueue.NewWithConfig(
	mysqlqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		DB: nil, // optional; provide *sql.DB instead of DSN
		DSN: "user:pass@tcp(127.0.0.1:3306)/queue?parseTime=true", // optional if DB is set
		DisableAutoMigrate: false, // set true when schema migrations are managed externally
		ProcessingRecoveryGrace:  2 * time.Second, // default if <=0: 2s
		ProcessingLeaseNoTimeout: 5 * time.Minute, // default if <=0: 5m
	},
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

natsqueue

natsqueue.New

New creates a high-level Queue using the NATS backend.

q, err := natsqueue.New(
	"nats://127.0.0.1:4222",
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

natsqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit NATS driver config.

q, err := natsqueue.NewWithConfig(
	natsqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		URL: "nats://127.0.0.1:4222", // required
	},
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

postgresqueue

postgresqueue.New

New creates a high-level Queue using the Postgres SQL backend.

q, err := postgresqueue.New(
	"postgres://user:[email protected]:5432/queue?sslmode=disable",
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

postgresqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit Postgres SQL driver config.

q, err := postgresqueue.NewWithConfig(
	postgresqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		DB: nil, // optional; provide *sql.DB instead of DSN
		DSN: "postgres://user:[email protected]:5432/queue?sslmode=disable", // optional if DB is set
		DisableAutoMigrate: false, // set true when schema migrations are managed externally
		ProcessingRecoveryGrace:  2 * time.Second, // default if <=0: 2s
		ProcessingLeaseNoTimeout: 5 * time.Minute, // default if <=0: 5m
	},
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

rabbitmqqueue

rabbitmqqueue.New

New creates a high-level Queue using the RabbitMQ backend.

q, err := rabbitmqqueue.New(
	"amqp://guest:[email protected]:5672/",
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

rabbitmqqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit RabbitMQ driver config.

q, err := rabbitmqqueue.NewWithConfig(
	rabbitmqqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		URL: "amqp://guest:[email protected]:5672/", // required
	},
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

redisqueue

redisqueue.New

New creates a high-level Queue using the Redis backend.

q, err := redisqueue.New(
	"127.0.0.1:6379",
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

redisqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit Redis driver config.

q, err := redisqueue.NewWithConfig(
	redisqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		Addr: "127.0.0.1:6379", // required
		Password: "",           // optional; default empty
		DB: 0,                  // optional; default 0
		Logger: nil,            // optional; default backend logger
		ServerLogLevel: redisqueue.ServerLogLevelDefault, // optional
	},
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

sqlitequeue

sqlitequeue.New

New creates a high-level Queue using the SQLite SQL backend.

q, err := sqlitequeue.New(
	"file:queue.db?_busy_timeout=5000",
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

sqlitequeue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit SQLite SQL driver config.

q, err := sqlitequeue.NewWithConfig(
	sqlitequeue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		DB: nil, // optional; provide *sql.DB instead of DSN
		DSN: "file:queue.db?_busy_timeout=5000", // optional if DB is set
		DisableAutoMigrate: false, // set true when schema migrations are managed externally
		ProcessingRecoveryGrace:  2 * time.Second, // default if <=0: 2s
		ProcessingLeaseNoTimeout: 5 * time.Minute, // default if <=0: 5m
	},
	queue.WithWorkers(4), // optional; default: 1 worker
)
if err != nil {
	return
}

sqsqueue

sqsqueue.New

New creates a high-level Queue using the SQS backend.

q, err := sqsqueue.New(
	"us-east-1",
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

sqsqueue.NewWithConfig

NewWithConfig creates a high-level Queue using an explicit SQS driver config.

q, err := sqsqueue.NewWithConfig(
	sqsqueue.Config{
		DriverBaseConfig: queueconfig.DriverBaseConfig{
			DefaultQueue: "critical", // default if empty: "default"
			Observer:     nil,        // default: nil
		},
		Region: "us-east-1", // default if empty: "us-east-1"
		Endpoint: "",        // optional; set for LocalStack/custom endpoint
		AccessKey: "",       // optional; static credentials
		SecretKey: "",       // optional; static credentials
	},
	queue.WithWorkers(4), // optional; default: runtime.NumCPU() (min 1)
)
if err != nil {
	return
}

Testing API

queue.NewFake is a recording fake with its established Dispatch(any) error surface. Inject it where *queue.FakeQueue or that recording contract is accepted; it is not a drop-in *queue.Queue.

Examples in this section assume they are used inside tests and t is a *testing.T (or testing.TB).

FakeQueue.AssertBatchCount

AssertBatchCount fails unless the accepted batch count equals expected.

fake := queue.NewFake()
_, _ = fake.Batch(queue.NewJob("emails:send")).Dispatch(context.Background())
fake.AssertBatchCount(t, 1)

FakeQueue.AssertBatched

AssertBatched fails unless an accepted canonical batch matches predicate. The predicate runs outside the recorder lock so it may safely inspect the fake.

fake := queue.NewFake()
_, _ = fake.Batch(queue.NewJob("emails:send")).Name("nightly").Dispatch(context.Background())
fake.AssertBatched(t, func(record queue.BatchRecord) bool { return record.Name == "nightly" })

FakeQueue.AssertChained

AssertChained fails unless an accepted chain has the expected ordered job types.

fake := queue.NewFake()
_, _ = fake.Chain(
	queue.NewJob("reports:build"),
	queue.NewJob("reports:publish"),
).Dispatch(context.Background())
fake.AssertChained(t, []string{"reports:build", "reports:publish"})

FakeQueue.AssertCount

AssertCount fails when the direct dispatch count is not expected.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send"))
fake.AssertCount(t, 1)

FakeQueue.AssertDispatched

AssertDispatched fails when jobType was not dispatched.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send"))
fake.AssertDispatched(t, "emails:send")

FakeQueue.AssertDispatchedOn

AssertDispatchedOn fails when jobType was not dispatched on queueName.

fake := queue.NewFake()
_ = fake.Dispatch(
	queue.NewJob("emails:send").
		OnQueue("critical"),
)
fake.AssertDispatchedOn(t, "critical", "emails:send")

FakeQueue.AssertDispatchedTimes

AssertDispatchedTimes fails when jobType dispatch count does not match expected.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send"))
_ = fake.Dispatch(queue.NewJob("emails:send"))
fake.AssertDispatchedTimes(t, "emails:send", 2)

FakeQueue.AssertNotDispatched

AssertNotDispatched fails when jobType was dispatched.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send"))
fake.AssertNotDispatched(t, "emails:cancel")

FakeQueue.AssertNothingBatched

AssertNothingBatched fails when any accepted batch was recorded.

FakeQueue.AssertNothingDispatched

AssertNothingDispatched fails when any direct dispatch was recorded.

fake := queue.NewFake()
fake.AssertNothingDispatched(t)

FakeQueue.Batch

Batch creates a fake batch backed by the production workflow builder and records it only when Dispatch accepts all initial member deliveries. Fluent function callbacks are accepted for compatibility but are not retained in fake runtime state or executed.

FakeQueue.BatchRecords

BatchRecords returns isolated creation records for accepted fake batches.

fake := queue.NewFake()
_, _ = fake.Batch(
	queue.NewJob("emails:first"),
	queue.NewJob("emails:second"),
).Name("nightly").AllowFailures().Dispatch(context.Background())
record := fake.BatchRecords()[0]
fmt.Println(record.Name, len(record.Jobs), record.AllowFailed)
// Output: nightly 2 true

FakeQueue.Chain

Chain creates a fake chain backed by the production workflow builder and records it only when Dispatch accepts its initial delivery. Fluent function callbacks are accepted for compatibility but are not retained in fake runtime state or executed.

FakeQueue.ChainRecords

ChainRecords returns isolated creation records for accepted fake chains.

fake := queue.NewFake()
_, _ = fake.Chain(
	queue.NewJob("reports:build"),
	queue.NewJob("reports:publish"),
).OnQueue("workflow").Dispatch(context.Background())
record := fake.ChainRecords()[0]
fmt.Println(len(record.Nodes), record.Queue)
// Output: 2 workflow

FakeQueue.Dispatch

Dispatch records a typed job payload in-memory using the fake default queue.

fake := queue.NewFake()
err := fake.Dispatch(queue.NewJob("emails:send").OnQueue("default"))

FakeQueue.Driver

Driver returns the active queue driver.

fake := queue.NewFake()
driver := fake.Driver()

FakeQueue.FindBatch

FindBatch returns workflow state created by the fake's production engine.

FakeQueue.FindChain

FindChain returns workflow state created by the fake's production engine.

NewFake

NewFake creates the canonical fake for direct and workflow tests.

fake := queue.NewFake()
_ = fake.Dispatch(
	queue.NewJob("emails:send").
		Payload(map[string]any{"id": 1}).
		OnQueue("critical"),
)
records := fake.Records()
fmt.Println(len(records), records[0].Queue, records[0].Job.Type)
// Output: 1 critical emails:send

FakeQueue.Prune

Prune removes terminal workflow state while retaining fake dispatch records.

FakeQueue.Ready

Ready validates fake queue readiness.

fake := queue.NewFake()
fmt.Println(fake.Ready(context.Background()) == nil)
// Output: true

FakeQueue.Records

Records returns isolated records for accepted direct dispatches. Chain and batch creation is available through ChainRecords and BatchRecords.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send").OnQueue("default"))
records := fake.Records()
fmt.Println(len(records), records[0].Job.Type)
// Output: 1 emails:send

FakeQueue.Register

Register is a compatibility no-op because the recording fake never executes handlers.

fake := queue.NewFake()
fake.Register("emails:send", func(context.Context, queue.Job) error { return nil })

FakeQueue.Reset

Reset clears direct dispatches and all workflow records through every fake view.

fake := queue.NewFake()
_ = fake.Dispatch(queue.NewJob("emails:send").OnQueue("default"))
fmt.Println(len(fake.Records()))
fake.Reset()
fmt.Println(len(fake.Records()))
// Output:
// 1
// 0

FakeQueue.Shutdown

Shutdown is a compatibility no-op because the recording fake owns no worker resources.

fake := queue.NewFake()
err := fake.Shutdown(context.Background())

FakeQueue.StartWorkers

StartWorkers is a compatibility no-op because the recording fake owns no workers.

fake := queue.NewFake()
err := fake.StartWorkers(context.Background())

FakeQueue.WithContext

WithContext returns a derived fake queue handle bound to ctx.

FakeQueue.Workers

Workers preserves fluent lifecycle compatibility without creating workers.

fake := queue.NewFake()
q := fake.Workers(4)
fmt.Println(q != nil)
// Output: true

Development

Use make test for root-module tests, make vet for static checks, and make generate to refresh generated documentation. make test-integration runs the separate integration module; pass a backend such as make test-integration sqlite to narrow the matrix. Integration tests may need local services.

Driver, docs, examples, and integration directories are independent Go modules; test each changed module from its directory. Matrix status and backend notes are tracked in docs/integration-scenarios.md.

Using with GoForj

GoForj Apps expose named queues through generated accessors. Dispatch jobs through those accessors and keep backend selection in queue configuration.

For the App workflow, see Queues.