taskq

README

taskQ

A small, type-safe task queue for Go, built around a deliberately tiny broker interface. Producers enqueue strongly-typed payloads; a pluggable broker handles delivery, acknowledgement, and requeue. The core has no dependencies beyond google/uuid.

  producer                            consumer
Queue[T] ─Enqueue─▶ Broker ─Dequeue─▶ Pool[T] ─▶ Chain(mw…) ─▶ Handler[T]
 (T→JSON)          (transport)      (N workers,   (logging,      (your code)
                        ▲            backoff)      recovery,…)
                        └──────────── Ack / Nack ──────────┘

Status

taskQ is early and evolving. What's shipping today:

• ✅ Core types and the Broker contract
• ✅ membroker: an in-memory broker for local dev and tests
• ✅ redisbroker: a Redis Streams broker (consumer groups, ReceiptHandle-based Ack/Nack, XAUTOCLAIM crash recovery); live-Redis tests are gated behind -short
• ✅ pgbroker: a Postgres broker (SELECT … FOR UPDATE SKIP LOCKED, lease-based visibility timeout with crash recovery, LISTEN/NOTIFY); live-Postgres tests are gated behind -short
• ✅ Pool[T]: a concurrent worker pool (configurable concurrency + backoff) and Middleware[T] composition (Chain, LoggingMiddleware, RecoveryMiddleware, plus OpenTelemetry tracing in otelmw)
• ✅ taskqtest: one shared conformance suite that all three brokers pass, so backends stay interchangeable
• ✅ saga: durable multi-step workflows with automatic compensation; needs no Broker/Pool changes, works over every backend
• 🚧 Remaining polish: richer Pool Ack/Nack observability, still in progress.

How taskQ compares

A few other Go task queues exist; here's how taskQ differs, stated as plainly as I can manage about my own project.

	taskQ	vmihailenco/taskq	oshankkumar/taskqueue-go
Payload type safety	Generics: Task[T]/Handler[T]/Queue[T], compile-time checked	Reflection-based task registry, string task names	Concrete *Job with a JSON payload
Backends	In-memory, Redis Streams, Postgres (SKIP LOCKED)	Redis, SQS, IronMQ, in-memory	Redis (Postgres/MySQL job-store on the roadmap, not shipped)
Crash recovery	Built into every backend (XAUTOCLAIM, Postgres lease expiry)	Backend-dependent (SQS/IronMQ visibility timeout)	Redis Lua-script atomic dequeue plus visibility timeout
Cross-backend conformance	One shared suite (taskqtest) every backend must pass	Not present	Not present (single backend today)
Middleware	Chain() composition: logging, panic recovery, OpenTelemetry	Fallback handler for failed messages	Not documented
Delayed / scheduled delivery	Not yet (roadmap item)	—	Built in
Multi-step workflows / sagas	Yes, via Saga[S] with automatic reverse-order compensation	No	No
Web dashboard	No	No	Yes: queue/job monitoring, retry/delete from the UI
Maturity	New, pre-1.0	~1.3k GitHub stars; production use (built by the Uptrace team)	Early-stage, small
License	MIT	BSD-2-Clause	MIT

taskQ's actual bets: generics for payload safety instead of a reflection-based registry, a real Postgres backend that neither alternative has, one conformance suite that keeps backends genuinely interchangeable rather than each growing its own quirks, and sagas. None of the alternatives above support multi-step compensating workflows. In exchange, taskQ is newer and less battle-tested than vmihailenco/taskq, has no web dashboard the way oshankkumar/taskqueue-go does, and doesn't yet support delayed/scheduled delivery natively (it's on the roadmap).

Install

go get github.com/Nuvraxis/taskQ

Requires Go 1.26+.

Concepts

taskQ separates the typed producer side from the byte-level transport side, so a broker never needs to know about your payload types.

Type	Role
Broker	The backend contract: Enqueue, Dequeue, Ack, Nack. Deliberately minimal. Backend-specific behavior (delays, priority, and so on) lives in constructor options, not the interface.
Message	The broker-level envelope: an ID, queue name, JSON Payload, attempt counters, and timestamps. Brokers only ever see this.
Queue[T]	Binds a broker to a queue name and a payload type T. Enqueue marshals T to JSON and hands a Message to the broker.
Task[T]	A Message with its Payload decoded back into T, the thing a Handler[T] actually receives.
Handler[T]	func(ctx, Task[T]) error. Returning a non-nil error signals the task should be retried (subject to MaxRetry).
Pool[T]	The consumer counterpart to Queue[T]: runs a Handler[T] against dequeued tasks with N workers, Acking successes and scheduling backoff Nacks for failures. Configure with WithConcurrency / WithBackoffStrategy.
Middleware[T]	func(Handler[T]) Handler[T], wrapping a handler with cross-cutting behavior. Compose with Chain; built-ins: LoggingMiddleware, RecoveryMiddleware, and otelmw.Middleware for tracing.
Saga[S]	A durable, multi-step workflow: []Step[S] run in order, with automatic reverse-order compensation if a step fails permanently. Built on Queue[Envelope[S]] + Pool[Envelope[S]], with no Broker changes required.

Acknowledgement model

• Ack means the task is done and should be removed, whether it succeeded or the caller gave up after exhausting retries.
• Nack means requeue for redelivery. It does not decide retry policy; the consumer does, by comparing Message.Attempts against Message.MaxRetry and passing the count it wants persisted on redelivery.

Retry budget

MaxRetry is the number of redeliveries allowed after the first attempt, so a task is delivered at most MaxRetry + 1 times. Set a default per queue with WithDefaultMaxRetry, or override it per call with WithMaxRetry.

Quick start

package main

import (
	"context"
	"encoding/json"
	"errors"
	"log"

	taskq "github.com/Nuvraxis/taskQ"
	"github.com/Nuvraxis/taskQ/membroker"
)

type EmailJob struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
}

func main() {
	ctx := context.Background()

	broker := membroker.New()
	defer broker.Close()

	// Producer: a typed queue over the broker.
	emails := taskq.NewQueue[EmailJob](broker, "emails")
	if err := emails.Enqueue(ctx, EmailJob{To: "a@example.com", Subject: "Hi"}); err != nil {
		log.Fatal(err)
	}

	// Consumer: dequeue a Message, decode it, run your handler, then Ack.
	msg, err := broker.Dequeue(ctx, "emails")
	if err != nil {
		log.Fatal(err)
	}

	var job EmailJob
	if err := json.Unmarshal(msg.Payload, &job); err != nil {
		log.Fatal(err)
	}
	log.Printf("sending email to %s: %q", job.To, job.Subject)

	if err := broker.Ack(ctx, *msg); err != nil && !errors.Is(err, taskq.ErrQueueClosed) {
		log.Fatal(err)
	}
}

Consuming with Pool and middleware

The quick start dequeues by hand to show the mechanics; in practice you hand a Handler[T] to a Pool, which runs it across N workers and turns failures into backoff-spaced Nacks for you. Wrap the handler with Chain to add cross-cutting behavior:

handler := taskq.Chain[EmailJob](
	func(ctx context.Context, task taskq.Task[EmailJob]) error {
		// ... your work ...
		return nil
	},
	taskq.LoggingMiddleware[EmailJob](nil), // nil → slog.Default(); logs outcome + duration
	taskq.RecoveryMiddleware[EmailJob](),   // innermost: a panic becomes a retryable error
)

pool := taskq.NewPool(broker, "emails", handler,
	taskq.WithConcurrency(8),
	taskq.WithBackoffStrategy(taskq.ExponentialBackoff{
		Base: time.Second, Max: 30 * time.Second, Factor: 2, Jitter: true,
	}),
)

// Run blocks until ctx is cancelled or the broker is closed, then waits for
// in-flight handlers to finish.
if err := pool.Run(ctx); err != nil {
	log.Fatal(err)
}

Order matters. Chain(base, A, B, C) runs as A(B(C(base))): the first middleware listed is outermost. Put RecoveryMiddleware last (innermost, directly wrapping the handler) so the *PanicError it produces is visible to the logging/tracing middleware wrapped around it. For distributed tracing, add otelmw.Middleware between logging and recovery.

Toggling tracing. otelmw honors the standard OpenTelemetry OTEL_SDK_DISABLED environment variable (set it to true to turn tracing off). Use the env-gated constructors so you don't need an if at the call site: otelmw.MiddlewareFromEnv[T](tracer) and otelmw.NewTracingBrokerFromEnv(broker, tracer) become no-op passthroughs when tracing is disabled:

handler := taskq.Chain[EmailJob](
	base,
	taskq.LoggingMiddleware[EmailJob](nil),
	otelmw.MiddlewareFromEnv[EmailJob](tracer), // no-op if OTEL_SDK_DISABLED=true
	taskq.RecoveryMiddleware[EmailJob](),
)

See examples/ for complete, runnable programs covering the worker pool, middleware, and exponential-backoff retries. The in-memory set needs no setup; the Redis and Postgres sets need a running server (see each backend's README):

# in-memory (zero setup)
go run ./examples/membroker/basic
go run ./examples/membroker/retry
go run ./examples/membroker/pool
go run ./examples/membroker/middleware

# Redis Streams (needs Redis on localhost:6379, or set TASKQ_REDIS_ADDR)
go run ./examples/redisbroker/basic
go run ./examples/redisbroker/retry
go run ./examples/redisbroker/pool

# Postgres (needs Postgres on localhost:5432, or set TASKQ_POSTGRES_DSN)
go run ./examples/pgbroker/basic
go run ./examples/pgbroker/pool

# Sagas (zero setup, uses membroker)
go run ./examples/saga/basic        # happy path: reserve → charge → ship
go run ./examples/saga/compensate   # a step fails, watch it roll back

In-memory broker

membroker is a zero-config Broker for tests and local development:

broker := membroker.New()
defer broker.Close()

Characteristics:

• FIFO per queue; Nack requeues at the back.
• Blocking Dequeue: it waits until a message is available, the context is cancelled, or the broker is closed.
• No persistence and no visibility timeout: Dequeue removes a message immediately, so a crash between Dequeue and Ack loses it. Fine for dev and tests; not intended for production.
• Close unblocks pending Dequeue calls (they return taskq.ErrQueueClosed) but lets already-buffered messages drain first.

Redis broker

redisbroker maps each queue to a Redis Stream and reads through a consumer group, so Ack/Nack target the exact entry via ReceiptHandle. You pass a *redis.Client (which the broker never closes) and a running Redis:

client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
defer client.Close()

broker := redisbroker.New(client, redisbroker.WithConsumerGroup("workers"))

Option	Default	Purpose
WithKeyPrefix(p)	"taskq:"	Prefix for stream keys, for namespacing on a shared Redis.
WithConsumerGroup(name)	"taskq-workers"	Consumer group; brokers sharing it compete for a stream's entries.
WithConsumerName(name)	random UUID	This consumer's identity within the group; also which consumer an XAUTOCLAIM-reclaimed entry is reassigned to.
WithBlockTimeout(d)	5s	How long one XREADGROUP blocks before Dequeue re-checks the context.
WithClaimMinIdle(d)	30s	Crash recovery threshold: how long an entry must sit unacknowledged in another consumer's Pending Entries List before Dequeue reclaims it via XAUTOCLAIM.

There's no Close()/ErrQueueClosed: cancel the context to stop a consumer. Crash recovery is built in: Dequeue opportunistically sweeps for one stale pending entry (idle past WithClaimMinIdle) via XAUTOCLAIM before falling back to reading new entries, so a crashed consumer's stranded work is redelivered automatically. No separate reaper process or ticker needed.

Postgres broker

pgbroker stores messages in one taskq_messages table and dequeues with SELECT … FOR UPDATE SKIP LOCKED, so many workers pull concurrently without stepping on each other. A claimed row is hidden by a lease (visibility timeout) until it's Acked or the lease expires. That's how it recovers a crashed worker's message, with no separate reaper. Dequeue blocks on LISTEN/NOTIFY, falling back to polling.

Setup

pgbroker takes a caller-owned *pgxpool.Pool (it never closes it) and needs its table to exist. Run pgbroker/schema.sql once (all CREATE … IF NOT EXISTS); there's no migration tooling yet:

pool, err := pgxpool.New(ctx, os.Getenv("TASKQ_POSTGRES_DSN"))
if err != nil { /* ... */ }
defer pool.Close()

// once per database, e.g. `psql -f pgbroker/schema.sql`
broker := pgbroker.New(pool, pgbroker.WithLeaseDuration(60*time.Second))

Bring up a local Postgres (and Redis, for redisbroker) with:

make deps-up   # docker compose up -d, redis:7 + postgres:16, credentials taskq/taskq/taskq_test

Config

Option	Default	Purpose
WithLeaseDuration(d)	30s	Visibility timeout: how long a claimed message stays invisible before its lease expires and it's redelivered. Set it comfortably above your handler's worst-case runtime, or an in-flight message gets redelivered to a second worker.
WithPollInterval(d)	200ms	Ceiling on Dequeue wake-up latency: the fallback when a NOTIFY is missed or never sent (e.g. a lease expiring with no new Enqueue to trigger one).
WithNotifyChannel(name)	"taskq_new_message"	Postgres NOTIFY channel. It's cluster-wide, not table-scoped, so set a distinct name if multiple taskQ deployments share one database.
WithQueuePrefix(prefix)	""	Namespaces queue names at the storage layer (every row carries prefix+queue). Mainly for tests sharing one database; the Message.Queue returned to callers is always the unprefixed name.

Notes

• No Close()/ErrQueueClosed: cancel the context to stop a consumer or Pool.Run (same as redisbroker).
• Crash recovery is built in: an expired lease makes the row claimable again on the next Dequeue. No reaper process required.
• PgBouncer in transaction-pooling mode doesn't forward LISTEN/NOTIFY, so every wake-up degrades to the poll interval (this doesn't affect correctness, only latency).
• Admin methods beyond the Broker interface: QueueDepth, QueueDepthAvailable, PeekMessages, ReapExpiredLease, PurgeQueue, ListQueues, PurgeAll, for monitoring and cleanup. ListQueues and PurgeAll are table-wide and refuse to run on a Broker constructed with WithQueuePrefix (they'd otherwise leak or delete rows outside that Broker's own namespace). Use PurgeQueue for a single queue instead, or a Broker with no prefix.

Sagas

A saga is a durable multi-step workflow: steps run in order, and if one fails permanently after earlier steps already succeeded, those earlier steps are rolled back. Their compensations run in reverse order. It's built entirely on the primitives above. A saga run is just a Queue[Envelope[S]] enqueuing to itself, one hop per step, so it needs no Broker or Pool changes and runs unmodified over membroker, redisbroker, or pgbroker.

Saga[S].Handler() returns an ordinary Handler[Envelope[S]] that you run through an ordinary Pool. Each call processes exactly one hop: one step, in one direction, then returns. The next hop always lives in the queue, never in a goroutine's stack. That's what makes a saga survive a worker crash: a restarted worker simply dequeues wherever the run left off.

Define the steps, hand Handler() to a Pool, and Start a run:

type OrderState struct {
	OrderID       string
	ReservationID string // set by reserve, read by its Compensate
	ChargeID      string // set by charge, read by its Compensate
}

steps := []saga.Step[OrderState]{
	{
		Name:       "reserve-inventory",
		Do:         func(ctx context.Context, s *OrderState) error { s.ReservationID = "resv-" + s.OrderID; return nil },
		Compensate: func(ctx context.Context, s *OrderState) error { /* release s.ReservationID */ return nil },
	},
	{
		Name:       "charge-payment",
		Do:         func(ctx context.Context, s *OrderState) error { s.ChargeID = "charge-" + s.OrderID; return nil },
		Compensate: func(ctx context.Context, s *OrderState) error { /* refund s.ChargeID */ return nil },
	},
	{
		Name: "ship-order",
		Do:   func(ctx context.Context, s *OrderState) error { return ship(s) },
		// No Compensate: the last step has nothing to undo. A nil Compensate
		// is an automatic no-op if rollback ever reaches it.
	},
}

sg := saga.New(broker, "orders", steps,
	saga.WithOnComplete(func(ctx context.Context, id string, s OrderState) { /* finished */ }),
	saga.WithOnFailed(func(ctx context.Context, id string, s OrderState, failedStep string, cause error) {
		// failedStep/cause describe the step whose Do originally failed,
		// even after rollback has unwound all the way back to step 0.
	}),
)

// Drive it with an ordinary Pool; nothing saga-specific here.
pool := taskq.NewPool(broker, "orders", sg.Handler())
go pool.Run(ctx)

sagaID, err := sg.Start(ctx, OrderState{OrderID: "order-1001"})

Semantics worth knowing:

• Steps run in order; compensation runs in reverse. If step 2 of 3 fails permanently, only the steps that actually succeeded (0 and 1) are compensated, in reverse (1 then 0). The failed step has nothing of its own to undo. A step whose Compensate is nil is skipped as a no-op.
• The retry budget is per hop. WithMaxRetry(n) (default 3) applies to each individual Do or Compensate call, using the same MaxRetry + 1-attempts semantics as the rest of taskQ, not the run as a whole.
• Hooks report the real outcome. WithOnComplete fires once every step's Do succeeds. WithOnFailed fires when a run doesn't complete, always reporting the step whose Do originally failed (not whichever Compensate finished last). WithOnCompensationFailed fires when a Compensate itself exhausts its retries. That's the one outcome a saga can't resolve on its own, and it leaves the run partly rolled back. Treat it as page-a-human, not log-and-continue.

See examples/saga for runnable happy-path and rollback demos.

Development

Common tasks are wrapped in the Makefile (POSIX-shell first; on Windows use Git Bash or WSL2):

make build     # go build ./...
make test      # go test ./...
make test-race # go test -race ./...   (needs a C toolchain)
make vet       # go vet ./...
make fmt       # gofmt -l -w .
make lint      # golangci-lint run

Run the examples with go run, or make run EX=<path>. The membroker/* and saga/* examples need no setup; redisbroker/* needs a running Redis and pgbroker/* a running Postgres (make deps-up, or the docker run commands in each backend's README). See examples/:

go run ./examples/membroker/pool       # or: make run EX=membroker/pool
go run ./examples/redisbroker/pool     # or: make run EX=redisbroker/pool
go run ./examples/pgbroker/pool        # or: make run EX=pgbroker/pool
go run ./examples/saga/compensate      # or: make run EX=saga/compensate

CI runs build/vet/tidy and go test -short across Linux, macOS, and Windows; the race detector on Linux and macOS only (Windows lacks a reliable CGO toolchain in the runner image); and golangci-lint, govulncheck, and the taskqtest conformance suite (against live Redis and Postgres service containers) on Linux.

Need Support

Questions, bug reports, or feature requests? Open an issue on the repository, or reach out to the maintainers at hello@nuvraxis.com.

License

MIT. See LICENSE.

Documentation

Index

• Variables
• type BackoffStrategy
• type Broker
• type ConstantBackoff
  • func (b ConstantBackoff) Next(attempt int) time.Duration
• type EnqueueOption
  • func WithMaxRetry(n int) EnqueueOption
• type ExponentialBackoff
  • func (b ExponentialBackoff) Next(attempt int) time.Duration
• type Handler
  • func Chain[T any](h Handler[T], mws ...Middleware[T]) Handler[T]
• type Message
• type Middleware
  • func LoggingMiddleware[T any](logger *slog.Logger) Middleware[T]
  • func RecoveryMiddleware[T any]() Middleware[T]
• type PanicError
  • func (e *PanicError) Error() string
  • func (e *PanicError) Stack() []byte
• type Pool
  • func NewPool[T any](broker Broker, queue string, handler Handler[T], opts ...PoolOption) *Pool[T]
  • func (p *Pool[T]) Run(ctx context.Context) error
• type PoolOption
  • func WithBackoffStrategy(s BackoffStrategy) PoolOption
  • func WithConcurrency(n int) PoolOption
• type Queue
  • func NewQueue[T any](broker Broker, name string, opts ...QueueOption) *Queue[T]
  • func (q *Queue[T]) Enqueue(ctx context.Context, payload T, opts ...EnqueueOption) error
• type QueueOption
  • func WithDefaultMaxRetry(n int) QueueOption
• type Task

Variables

var ErrQueueClosed = errors.New("taskQ: queue closed")

ErrQueueClosed is returned by Broker methods once the broker has been closed — Dequeue returns it after any buffered messages drain, Enqueue and Nack return it immediately.

Types

type BackoffStrategy

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

BackoffStrategy computes the delay before the nth redelivery of a failed task. attempt is 1-indexed: Next(1) is the delay before the first retry, Next(2) before the second, and so on — it mirrors the Attempts value the worker pool is about to persist via Nack.

type Broker

type Broker interface {
	Enqueue(ctx context.Context, msg Message) error
	Dequeue(ctx context.Context, queue string) (*Message, error)
	Ack(ctx context.Context, msg Message) error
	Nack(ctx context.Context, msg Message, cause error) error
}

Broker is the entire backend contract. Deliberately tiny — resist adding to it. Backend-specific behavior (delayed delivery, priority, SKIP LOCKED, …) is modeled as broker-specific constructor options, not interface growth.

Ack always means "done, remove it" — whether the task succeeded or the caller gave up on it after exhausting retries. Nack always means "requeue for redelivery" — it does not decide retry policy; that's the worker pool's job (phase 2). The caller passes whatever Attempts value it wants persisted on redelivery; the broker just requeues msg as given, verbatim.

type ConstantBackoff

type ConstantBackoff struct {
	Delay time.Duration
}

ConstantBackoff waits a fixed Delay before every retry.

func (ConstantBackoff) Next

func (b ConstantBackoff) Next(attempt int) time.Duration

Next returns Delay, unaffected by attempt.

type EnqueueOption

type EnqueueOption func(*enqueueConfig)

EnqueueOption configures a single Enqueue call.

func WithMaxRetry

func WithMaxRetry(n int) EnqueueOption

WithMaxRetry overrides, for a single Enqueue call, the maximum number of redelivery attempts before the worker pool gives up on a task.

type ExponentialBackoff

type ExponentialBackoff struct {
	Base   time.Duration
	Max    time.Duration
	Factor float64
	Jitter bool
}

ExponentialBackoff waits Base * Factor^(attempt-1), capped at Max once Max is positive. Factor defaults to 2 if zero. If Jitter is true, the computed delay is randomized uniformly over [0, delay] to avoid thundering-herd redelivery when many tasks fail together.

func (ExponentialBackoff) Next

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

Next returns the exponential delay for attempt, capped and jittered per the struct's fields.

type Handler

type Handler[T any] func(ctx context.Context, task Task[T]) error

Handler processes a single Task[T]. Returning a non-nil error causes the worker pool to Nack the task for redelivery, subject to MaxRetry.

func Chain

func Chain[T any](h Handler[T], mws ...Middleware[T]) Handler[T]

Chain composes middleware around a base Handler[T]. Chain(h, A, B, C) runs as A(B(C(h))): the first middleware listed is outermost — it sees the task first and the returned error last.

Put RecoveryMiddleware LAST (innermost, directly wrapping h), not first: a panic it recovers becomes a normal *PanicError return, and only middleware listed *before* it in the chain — i.e. wrapping it from outside — ever sees that return value to log or trace. Recovery's job is to protect against a panicking Handler; it does not protect against a panic inside another middleware's own code, regardless of position.

h := taskq.Chain(baseHandler,
    taskq.LoggingMiddleware[T](logger),  // outermost: logs the outcome, including panics
    otelmw.Middleware[T](tracer),        // spans decode+handler, including panics
    taskq.RecoveryMiddleware[T](),       // innermost: converts a handler panic to an error
)
pool := taskq.NewPool(broker, queue, h, opts...)

type Message

type Message struct {
	ID         string    `json:"id"`
	Queue      string    `json:"queue"`
	Payload    []byte    `json:"payload"`
	Attempts   int       `json:"attempts"`
	MaxRetry   int       `json:"max_retry"`
	EnqueuedAt time.Time `json:"enqueued_at"`

	// Ack/Nack. membroker leaves it empty; redisbroker/pgbroker will use it
	ReceiptHandle string `json:"receipt_handle"`
}

Message is the broker-level envelope. Brokers only ever see this — they don't know about T. Task[T] is decoded from Message above the Broker boundary.

type Middleware

type Middleware[T any] func(Handler[T]) Handler[T]

Middleware wraps a Handler[T], adding cross-cutting behavior — logging, panic recovery, tracing — without changing what the handler does. Same shape as net/http middleware: a function from Handler to Handler.

func LoggingMiddleware

func LoggingMiddleware[T any](logger *slog.Logger) Middleware[T]

LoggingMiddleware logs each task's start, outcome, and duration via logger (log/slog; slog.Default() if nil).

A failed attempt is logged at Warn if the Pool will retry it, or Error if this was the last allowed attempt. That distinction is derived from task.Attempts and task.MaxRetry alone — Task.Attempts is the delivery count *before* this attempt, so per the MaxRetry+1 delivery budget (README), task.Attempts >= task.MaxRetry means Pool.handle will Ack (give up) rather than Nack (retry) after this call returns. No Pool changes are needed to know that.

If the error is a *PanicError, its stack trace is attached as a separate log attribute rather than folded into the error string.

func RecoveryMiddleware

func RecoveryMiddleware[T any]() Middleware[T]

RecoveryMiddleware recovers from a panic in the wrapped Handler and converts it into a *PanicError, so one bad task can't take down a Pool worker. Without this, Pool.runWorker has nothing catching a panic: it propagates up the goroutine's stack and, per normal Go semantics, crashes the process. With it, a panic becomes an ordinary handler error, subject to the same Attempts/MaxRetry retry decision as any other failure.

type PanicError

type PanicError struct {
	Value any
	// contains filtered or unexported fields
}

PanicError wraps a value recovered from a panicking Handler. Error() stays short so it doesn't bloat a broker's persisted Nack cause; the full stack trace captured at the moment of the panic is available via Stack() for middleware (LoggingMiddleware, or an OTel middleware) that wants to record it separately.

func (*PanicError) Error

func (e *PanicError) Error() string

func (*PanicError) Stack

func (e *PanicError) Stack() []byte

Stack returns the stack trace captured at the moment of the panic.

type Pool

type Pool[T any] struct {
	// contains filtered or unexported fields
}

Pool binds a Broker and queue name to a Handler[T] and runs it against dequeued tasks with a fixed number of concurrent workers. It's the consumer-side counterpart to Queue[T] — one Pool per broker/queue/T.

func NewPool

func NewPool[T any](broker Broker, queue string, handler Handler[T], opts ...PoolOption) *Pool[T]

NewPool creates a Pool bound to broker and queue, dispatching decoded tasks to handler. opts configure concurrency and backoff — see WithConcurrency and WithBackoffStrategy. The default is a single worker with an ExponentialBackoff (1s base, 30s max, factor 2, jittered).

func (*Pool[T]) Run

func (p *Pool[T]) Run(ctx context.Context) error

Run starts cfg.concurrency worker goroutines and blocks until ctx is canceled or the broker reports it's closed, then waits for every worker's in-flight Handler call to return before returning itself. A Handler that ignores ctx can therefore block Run's return indefinitely — the same caveat as net/http's graceful shutdown.

Retry delays use an async timer (see BackoffStrategy) rather than blocking a worker: a failed task's Nack is scheduled with time.AfterFunc and the worker resumes dequeuing immediately. Run does NOT wait for scheduled retries to fire — only for handlers already running when shutdown begins. A scheduled Nack uses context.Background(), so it still fires — and any error it returns is silently dropped — even after Run has returned. Observing that error is left to Phase 5 middleware.

Run returns nil on expected shutdown (ctx canceled or the broker closed). Any other Dequeue error is returned as-is.

type PoolOption

type PoolOption func(*poolConfig)

PoolOption configures a Pool[T] at construction time.

func WithBackoffStrategy

func WithBackoffStrategy(s BackoffStrategy) PoolOption

WithBackoffStrategy overrides the default backoff strategy used to space out redeliveries after a failed Handler call.

func WithConcurrency

func WithConcurrency(n int) PoolOption

WithConcurrency sets the number of worker goroutines a Pool runs concurrently. Values less than 1 are treated as 1.

type Queue

type Queue[T any] struct {
	// contains filtered or unexported fields
}

Queue binds a Broker to a queue name and a payload type T.

func NewQueue

func NewQueue[T any](broker Broker, name string, opts ...QueueOption) *Queue[T]

NewQueue creates a Queue bound to broker and the given queue name. opts set defaults — such as WithDefaultMaxRetry — applied to every task enqueued through it.

func (*Queue[T]) Enqueue

func (q *Queue[T]) Enqueue(ctx context.Context, payload T, opts ...EnqueueOption) error

Enqueue marshals payload to JSON and hands it to the broker.

type QueueOption

type QueueOption func(*queueConfig)

QueueOption configures a Queue[T] at construction time; these become defaults that EnqueueOption can override per call.

func WithDefaultMaxRetry

func WithDefaultMaxRetry(n int) QueueOption

WithDefaultMaxRetry sets the default MaxRetry applied to every task enqueued through this Queue, unless overridden per call with WithMaxRetry.

type Task

type Task[T any] struct {
	ID         string    `json:"id"`
	Queue      string    `json:"queue"`
	Payload    T         `json:"payload"`
	Attempts   int       `json:"attempts"`
	MaxRetry   int       `json:"max_retry"`
	EnqueuedAt time.Time `json:"enqueued_at"`
}

Task is what Handler[T] actually receives: a Message with Payload decoded into T.