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
✅ Pool failure reporting: WithOnDecodeError and WithOnSettleError surface undecodable payloads and failed Ack/Nack calls, and Ack/Nack run on a context detached from shutdown (WithSettleTimeout, default 5s), so a task that finishes during shutdown isn't redelivered

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 `Nack`s for failures. Configure with `WithConcurrency` / `WithBackoffStrategy` / `WithSettleTimeout`. Failures that happen outside your handler (a payload that won't decode, a failed `Ack`/`Nack`) are reported through the `WithOnDecodeError` / `WithOnSettleError` hooks.
`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, and the Ack that follows each one, 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](),
)

Reporting failures outside the handler. Two kinds of failure happen where your handler can't see them, and Run doesn't return them either:

A payload that won't decode into T. The Pool acks and drops it, because it would never decode on redelivery.
An Ack or Nack the broker rejects. A failed Ack usually means the broker redelivers the task once its lease or idle window expires. A failed Nack means that scheduled retry is lost.

Both are silent unless you register a hook. Wiring them to slog:

logger := slog.Default()

pool := taskq.NewPool(broker, "emails", handler,
	taskq.WithOnDecodeError(func(ctx context.Context, msg taskq.Message, err error) {
		logger.ErrorContext(ctx, "dropping undecodable task",
			"task_id", msg.ID, "queue", msg.Queue, "error", err)
	}),
	taskq.WithOnSettleError(func(ctx context.Context, op string, msg taskq.Message, err error) {
		// op is taskq.SettleOpAck or taskq.SettleOpNack.
		logger.ErrorContext(ctx, "task settle failed",
			"op", op, "task_id", msg.ID, "queue", msg.Queue,
			"attempts", msg.Attempts, "error", err)
	}),
	taskq.WithSettleTimeout(10*time.Second), // default 5s
)

Shutdown doesn't cancel settlement. Ack and Nack run on a context that keeps Run's values (a trace span, say) but not its cancellation, bounded by WithSettleTimeout. A handler that finishes after you cancel Run's ctx still gets its Ack, instead of failing it on the dead context and having pgbroker or redisbroker redeliver the task later. Run waits for that Ack before returning. Hooks get the same never-cancelled context. WithOnSettleError can fire after Run returns, because a scheduled retry's Nack is fire-and-forget, so don't tear down anything a hook depends on when you shut down.

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.

Name		Name	Last commit message	Last commit date
.github		.github
examples		examples
membroker		membroker
otelmw		otelmw
pgbroker		pgbroker
redisbroker		redisbroker
saga		saga
taskqtest		taskqtest
.gitignore		.gitignore
.golangci.yaml		.golangci.yaml
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
backoff.go		backoff.go
broker.go		broker.go
errors.go		errors.go
go.mod		go.mod
go.sum		go.sum
handler.go		handler.go
message.go		message.go
middleware.go		middleware.go
middleware_test.go		middleware_test.go
options.go		options.go
pool.go		pool.go
pool_test.go		pool_test.go
queue.go		queue.go
queue_test.go		queue_test.go
task.go		task.go
taskQ_package_architecture.png		taskQ_package_architecture.png

Navigation Menu

Repository navigation

About

Repository files navigation

taskQ

Status

How taskQ compares

Install

Concepts

Acknowledgement model

Retry budget

Quick start

Consuming with `Pool` and middleware

In-memory broker

Redis broker

Postgres broker

Setup

Config

Notes

Sagas

Development

Need Support

License

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

taskQ

Status

How taskQ compares

Install

Concepts

Acknowledgement model

Retry budget

Quick start

Consuming with Pool and middleware

In-memory broker

Redis broker

Postgres broker

Setup

Config

Notes

Sagas

Development

Need Support

License

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Consuming with `Pool` and middleware