hutch

README

hutch 🐰

A broker-agnostic worker-pool for queue consumers in Go.

hutch separates the orchestration that every queue consumer needs — a pool of workers, a bounded number of in-flight messages, a configurable ack/nack policy, per-message timeouts, and graceful draining on shutdown — from the broker itself. Brokers plug in behind two small interfaces, so the same consumer code runs on RabbitMQ, Redis, or (with a ~150-line driver) anything else.

sub, _ := rabbitmq.NewSubscriber("amqp://guest:guest@localhost:5672/")
pool := hutch.NewPool(sub, hutch.WithWorkers(20))

pool.Handle("orders", hutch.Handle(func(ctx context.Context, o Order) error {
    return process(ctx, o)
}))
defer pool.Close(context.Background())

---

Why another queue library?

hutch is small on purpose, but it bakes in lessons that are easy to get wrong and expensive to learn in production:

Lesson	What hutch does
Bounded prefetch is what makes consumers scale. Without a per-consumer in-flight cap, one replica greedily buffers the whole backlog and new replicas starve — adding pods does nothing.	Every queue gets a prefetch limit, defaulting to its worker count. Work spreads fairly across all replicas.
Be channel-light. Opening one broker channel per worker exhausts per-connection/per-user channel limits the instant you scale out.	A pool of N workers shares one channel/consumer. Channel usage stays flat as you add workers or replicas.
Isolate publishing from consuming. A consumer-side storm shouldn't be able to take down the producers your request path depends on.	The RabbitMQ driver publishes over a separate connection from consuming.
Fail predictably. Aggressive retry-to-DLQ floods; never-acking leaks unacked messages forever.	Explicit Drop / Requeue / Reject policy. The default (Drop) never piles up unacked or floods a DLQ.
Drain on shutdown.	Pool.Close(ctx) stops taking new work and waits for in-flight handlers, bounded by your context.
Reconnect transparently.	Drivers re-dial with exponential backoff + jitter and re-attach the delivery stream; the message channel survives reconnects.

These come from running a high-throughput notification consumer across many replicas — including an incident where a per-worker-channel design exhausted the broker's channel quota and took down publishing. hutch is the design that replaced it.

---

Install

go get github.com/akhiljns/hutch

Drivers live in subpackages so you only pull the client you use:

• github.com/akhiljns/hutch/connectors/rabbitmq — RabbitMQ (AMQP 0-9-1)
• github.com/akhiljns/hutch/connectors/redis — Redis Streams + consumer groups
• github.com/akhiljns/hutch/connectors/memory — in-process, zero-dependency (tests/local dev)

---

Architecture

        your handler
             │
        ┌────▼─────┐     Options: Workers, Prefetch, HandlerTimeout, ErrorPolicy
        │  Pool    │     • fan-out to N workers   • bounded in-flight (prefetch)
        │ (engine) │     • ack/nack policy        • graceful drain
        └────┬─────┘
   Subscriber │ Producer        ← tiny broker-agnostic interfaces
        ┌─────▼──────┐
        │   driver   │   rabbitmq · redis · memory · (kafka, nats, …)
        └─────┬──────┘
          the broker

The engine (hutch) is pure Go with no broker dependencies. Each driver implements Subscriber and/or Producer and owns the broker-specific bits (connections, prefetch/QoS, reconnection, ack semantics).

type Message interface {
    Body() []byte
    Ack() error
    Nack(requeue bool) error
}

type Subscriber interface {
    Subscribe(ctx context.Context, queue string, prefetch int) (<-chan Message, error)
    Close() error
}

type Producer interface {
    Publish(ctx context.Context, queue string, body []byte) error
    Close() error
}

---

Usage

Consume

sub, err := rabbitmq.NewSubscriber(url, rabbitmq.WithLogger(log.Default()))
if err != nil { log.Fatal(err) }

pool := hutch.NewPool(sub,
    hutch.WithWorkers(20),          // default workers per queue
    hutch.WithLogger(log.Default()),
)

// Typed handler: JSON body decoded into Order automatically.
pool.Handle("orders", hutch.Handle(func(ctx context.Context, o Order) error {
    return process(ctx, o)
}), hutch.OnError(hutch.Reject))

// Raw handler: full control over the message.
pool.Handle("audit", func(ctx context.Context, m hutch.Message) error {
    return store(ctx, m.Body())
}, hutch.Workers(4), hutch.Prefetch(8))

// On shutdown: stop consuming and drain in-flight, bounded by ctx.
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
pool.Close(ctx)

Publish

prod, _ := rabbitmq.NewProducer(url)   // separate connection from the consumer
pub := hutch.NewPublisher(prod)

pub.PublishJSON(ctx, "orders", Order{ID: 1, Item: "widget"})
pub.Publish(ctx, "audit", []byte("..."))
defer pub.Close()

No broker? Use the in-memory driver

broker := memory.New()              // implements both Subscriber and Producer
pub  := hutch.NewPublisher(broker)
pool := hutch.NewPool(broker)

Redis

The Redis driver is a drop-in swap — same pool, same handlers, just a different subscriber/producer:

sub, _ := redis.NewSubscriber("redis://localhost:6379/0")
prod, _ := redis.NewProducer("redis://localhost:6379/0")  // separate client
pool := hutch.NewPool(sub, hutch.WithWorkers(20))

It's built on Redis Streams + consumer groups, the only Redis primitive that gives per-message acks and redelivery (lists and pub/sub don't):

• XADD publishes; XREADGROUP load-balances entries across every replica in the group, so adding pods adds throughput.
• prefetch is honored by reading at most as many entries as there are free in-flight slots — no single replica hoards the backlog.
• Ack/Reject remove the entry (XACK+XDEL); Requeue re-adds it at the tail for redelivery.
• Entries left pending by a crashed consumer are reclaimed (XAUTOCLAIM) once idle and redelivered — the Redis analogue of RabbitMQ redelivering unacked messages when a channel drops.

The group starts at 0, so a consumer picks up the whole existing backlog, not just messages published after it connects.

Runnable examples: examples/memory (no broker needed), examples/rabbitmq, and examples/redis.

---

Configuration

Pool-wide defaults (hutch.With*) are inherited by every queue and can be overridden per Handle call (hutch.Workers, hutch.Prefetch, …):

Option	Default	Meaning
WithWorkers(n) / Workers(n)	10	worker goroutines per queue
WithPrefetch(n) / Prefetch(n)	= workers	max in-flight messages per queue (per replica)
WithHandlerTimeout(d) / HandlerTimeout(d)	30s	per-message timeout (0 disables)
WithErrorPolicy(p) / OnError(p)	Drop	what to do when a handler errors
OnErrorFunc(fn)	—	callback for every failed message (metrics, capture)
WithLogger(l)	no-op	anything with Printf, e.g. log.Default()

RabbitMQ driver options: WithBackoff, WithLogger, WithPublisherChannels (default 3), WithDeclareQueues (default true).

Redis driver options: WithGroup (consumer group, default "hutch"), WithConsumerName (default <hostname>-<pid>), WithClaimMinIdle (reclaim entries idle longer than this from crashed consumers, default 30s; set above your handler timeout), WithBlock (read block, default 5s), WithMaxLen (approximate stream cap, default unbounded), WithBackoff, WithLogger.

Error policies

Policy	Behavior	Use when
Drop (default)	ack the failed message	"process or move on"; pair with OnErrorFunc to capture
Requeue	nack + redeliver	failures are transient (⚠️ add your own attempt cap — a poison message loops forever)
Reject	nack, no redeliver	you have a dead-letter destination configured on the queue

---

Writing a driver (Kafka, Redis, NATS, …)

Implement Subscriber and/or Producer. The engine handles workers, prefetch fan-out, ack policy, timeouts, and draining — your driver only deals with the broker. Two responsibilities matter:

1. Honor prefetch. Keep at most prefetch messages outstanding per Subscribe. On RabbitMQ that's basic.qos; on Redis it's how many you pull before acking; on Kafka it maps to in-flight/max.poll.records.
2. Hide reconnection. Keep the returned channel open across transient broker disruptions; close it only when ctx is cancelled or Close is called.

Sketch:

type Driver struct { /* client, config */ }

func (d *Driver) Subscribe(ctx context.Context, queue string, prefetch int) (<-chan hutch.Message, error) {
    out := make(chan hutch.Message)
    go d.run(ctx, queue, prefetch, out) // pull, wrap as hutch.Message, forward; reconnect on loss
    return out, nil
}

func (d *Driver) Publish(ctx context.Context, queue string, body []byte) error { /* ... */ }
func (d *Driver) Close() error { /* ... */ }

See connectors/memory for a complete ~150-line reference driver, connectors/rabbitmq for a production-grade AMQP one (QoS, channel-light forwarding, reconnect, isolated publisher pool), and connectors/redis for a streams-based one (consumer groups, prefetch via batched reads, idle-entry reclaim).

Kafka and NATS drivers aren't bundled yet — they pull heavy clients and need a live broker to test honestly. The interface above is all it takes; PRs welcome.

---

Status & caveats

• Concurrent acks from multiple workers share one channel; the RabbitMQ client serializes channel writes internally, so this is safe (and is what keeps channel counts flat). Per-message ordering is not guaranteed across workers — use a single worker per queue if you need it.
• Drop is lossy by design. Use OnErrorFunc to record what you drop, or switch to Reject with a dead-letter queue.
• Tested with go test ./...: the engine against the in-memory driver, and the Redis driver against an in-process Redis (miniredis). The RabbitMQ driver is exercised via its example against a real broker.

License

MIT

Documentation

Overview

Package hutch is a broker-agnostic worker-pool for queue consumers in Go.

It separates the orchestration that every queue consumer needs — a pool of workers, a bounded number of in-flight messages, a configurable success/failure (ack/nack) policy, per-message timeouts, and graceful draining on shutdown — from the broker-specific transport. Brokers plug in behind two small interfaces (Subscriber and Producer); ship with RabbitMQ, Redis, and in-memory drivers, and add Kafka/NATS/etc. by implementing the same interfaces.

Why it exists

hutch encodes a set of hard-won production lessons:

• Bounded prefetch is what makes consumers scale horizontally. Without a per-consumer in-flight limit, one replica greedily buffers the whole backlog and newly-added replicas starve. The Pool always sets a prefetch (defaulting to the worker count) so work spreads fairly across every replica.
• Be channel-light. A pool of N workers shares one broker channel/consumer, not one channel per worker — opening a channel per worker exhausts broker channel limits the moment you scale out.
• Isolate publishing from consuming. The RabbitMQ driver publishes over a separate connection so a consumer-side disruption can never starve the producers that your request path depends on.
• Fail predictably. Choose Drop, Requeue, or Reject explicitly; the default (Drop) never lets failed messages pile up as unacked or flood a dead-letter queue.
• Drain on shutdown. Pool.Close stops accepting new work and waits for in-flight handlers to finish, bounded by the context you pass.

Quick start

sub, _ := rabbitmq.NewSubscriber("amqp://guest:guest@localhost:5672/")
pool := hutch.NewPool(sub, hutch.WithLogger(log.Default()))
pool.Handle("orders", hutch.Handle(func(ctx context.Context, o Order) error {
    return process(ctx, o)
}), hutch.Workers(20))
defer pool.Close(context.Background())

Index

• Variables
• type Backoff
  • func DefaultBackoff() Backoff
  • func (b Backoff) ForAttempt(attempt int) time.Duration
• type ErrorPolicy
  • func (p ErrorPolicy) String() string
• type HandlerFunc
  • func Handle[T any](fn func(ctx context.Context, v T) error) HandlerFunc
• type Logger
• type Message
• type Option
  • func WithErrorPolicy(p ErrorPolicy) Option
  • func WithHandlerTimeout(d time.Duration) Option
  • func WithLogger(l Logger) Option
  • func WithPrefetch(n int) Option
  • func WithWorkers(n int) Option
• type Pool
  • func NewPool(sub Subscriber, opts ...Option) *Pool
  • func (p *Pool) Close(ctx context.Context) error
  • func (p *Pool) Handle(queue string, h HandlerFunc, opts ...QueueOption) error
• type Producer
• type Publisher
  • func NewPublisher(prod Producer) *Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, queue string, body []byte) error
  • func (p *Publisher) PublishJSON(ctx context.Context, queue string, v any) error
• type QueueOption
  • func HandlerTimeout(d time.Duration) QueueOption
  • func OnError(p ErrorPolicy) QueueOption
  • func OnErrorFunc(fn func(Message, error)) QueueOption
  • func Prefetch(n int) QueueOption
  • func Workers(n int) QueueOption
• type Subscriber

Variables

var ErrClosed = errors.New("hutch: closed")

ErrClosed is returned by operations on a Pool, Publisher, or driver that has already been closed.

Types

type Backoff

type Backoff struct {
	Min    time.Duration // delay before the first retry
	Max    time.Duration // cap on the delay
	Factor float64       // multiplier per attempt (>= 1)
	Jitter bool          // randomize within [d/2, d] to avoid thundering herds
}

Backoff describes an exponential backoff schedule with optional jitter, used by drivers when reconnecting to a broker.

func DefaultBackoff

func DefaultBackoff() Backoff

DefaultBackoff returns a sensible schedule: 1s, 2s, 4s … capped at 30s, with jitter.

func (Backoff) ForAttempt

func (b Backoff) ForAttempt(attempt int) time.Duration

ForAttempt returns the delay for a 1-based attempt number.

type ErrorPolicy

type ErrorPolicy int

ErrorPolicy decides what happens to a message when its handler returns an error.

const (
	// Drop acknowledges the failed message, removing it (at-most-once). This is
	// the default: it never lets failures accumulate as unacked messages or
	// flood a dead-letter queue. Pair it with OnErrorFunc to capture failures.
	Drop ErrorPolicy = iota

	// Requeue negatively-acknowledges with redelivery. Use only when failures
	// are expected to be transient — a permanently-failing ("poison") message
	// will loop forever, so add your own attempt cap if you choose this.
	Requeue

	// Reject negatively-acknowledges without redelivery. On brokers with a
	// dead-letter destination configured for the queue the message is routed
	// there; otherwise it is dropped.
	Reject
)

func (ErrorPolicy) String

func (p ErrorPolicy) String() string

type HandlerFunc

type HandlerFunc func(ctx context.Context, msg Message) error

HandlerFunc processes a single message. Returning nil acks the message; returning an error applies the queue's ErrorPolicy. The context carries the per-message timeout (see HandlerTimeout) and is cancelled when the pool shuts down, so long-running handlers should respect it.

func Handle

func Handle[T any](fn func(ctx context.Context, v T) error) HandlerFunc

Handle adapts a typed handler into a HandlerFunc by JSON-decoding the message body into T. A decode failure is returned as an error (and so follows the queue's ErrorPolicy — typically Drop, since a malformed payload can never succeed on retry).

pool.Handle("orders", hutch.Handle(func(ctx context.Context, o Order) error {
    return process(ctx, o)
}))

type Logger

type Logger interface {
	Printf(format string, args ...any)
}

Logger is the minimal logging surface hutch needs. It is satisfied by the standard library's *log.Logger, so you can pass log.Default() directly. The default is a no-op.

type Message

type Message interface {
	// Body returns the raw payload bytes.
	Body() []byte

	// Ack acknowledges successful processing; the broker removes the message.
	Ack() error

	// Nack signals failure. If requeue is true the broker should redeliver the
	// message; if false it should drop it (or route it to a dead-letter
	// destination, where the broker supports one).
	Nack(requeue bool) error
}

Message is a single unit of work yielded by a Subscriber. The worker that receives it must eventually call exactly one of Ack or Nack — the Pool does this automatically based on the handler result and the configured ErrorPolicy.

type Option

type Option func(*poolConfig)

Option configures a Pool.

func WithErrorPolicy

func WithErrorPolicy(p ErrorPolicy) Option

WithErrorPolicy sets the default policy applied when a handler errors.

func WithHandlerTimeout

func WithHandlerTimeout(d time.Duration) Option

WithHandlerTimeout sets the default per-message handler timeout (0 disables).

func WithLogger

func WithLogger(l Logger) Option

WithLogger sets the logger (default: no-op). *log.Logger satisfies it.

func WithPrefetch

func WithPrefetch(n int) Option

WithPrefetch sets the default max in-flight messages per queue. When unset (or 0) it defaults to the queue's worker count, which keeps every worker busy without hoarding the backlog from other replicas.

func WithWorkers

func WithWorkers(n int) Option

WithWorkers sets the default number of worker goroutines per queue.

type Pool

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

Pool is a broker-agnostic consumer worker-pool. It draws messages from a Subscriber, fans them out to a configurable number of workers per queue, bounds in-flight work via prefetch, applies an ErrorPolicy on failure, and drains gracefully on shutdown.

A Pool is safe for concurrent use. Create one per Subscriber.

func NewPool

func NewPool(sub Subscriber, opts ...Option) *Pool

NewPool builds a Pool over the given Subscriber.

func (*Pool) Close

func (p *Pool) Close(ctx context.Context) error

Close stops accepting new messages and waits for in-flight handlers to finish, bounded by ctx. Messages still in flight when ctx expires are abandoned and will be redelivered by the broker after restart (at-least-once). Close is idempotent.

func (*Pool) Handle

func (p *Pool) Handle(queue string, h HandlerFunc, opts ...QueueOption) error

Handle starts a worker pool for queue. It may be called for multiple queues on the same Pool. It returns once the subscription is established and workers are running; it does not block.

type Producer

type Producer interface {
	Publish(ctx context.Context, queue string, body []byte) error
	Close() error
}

Producer is the publish side of a broker driver. Implementations must be safe for concurrent use.

type Publisher

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

Publisher is a thin, broker-agnostic convenience wrapper over a Producer. It adds JSON encoding; the resilience and batching/pooling live in the driver.

func NewPublisher

func NewPublisher(prod Producer) *Publisher

NewPublisher wraps a Producer driver.

func (*Publisher) Close

func (p *Publisher) Close() error

Close closes the underlying producer.

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, queue string, body []byte) error

Publish sends raw bytes to queue.

func (*Publisher) PublishJSON

func (p *Publisher) PublishJSON(ctx context.Context, queue string, v any) error

PublishJSON JSON-encodes v and sends it to queue.

type QueueOption

type QueueOption func(*queueConfig)

QueueOption overrides pool defaults for a single Pool.Handle call.

func HandlerTimeout

func HandlerTimeout(d time.Duration) QueueOption

HandlerTimeout overrides the per-message timeout for this queue (0 disables).

func OnError

func OnError(p ErrorPolicy) QueueOption

OnError overrides the ErrorPolicy for this queue.

func OnErrorFunc

func OnErrorFunc(fn func(Message, error)) QueueOption

OnErrorFunc registers a callback invoked with every failed message and its error (before the ErrorPolicy is applied) — handy for metrics or capturing dropped payloads.

func Prefetch

func Prefetch(n int) QueueOption

Prefetch overrides the max in-flight messages for this queue.

func Workers

func Workers(n int) QueueOption

Workers overrides the worker count for this queue.

type Subscriber

type Subscriber interface {
	Subscribe(ctx context.Context, queue string, prefetch int) (<-chan Message, error)
	Close() error
}

Subscriber is the consume side of a broker driver.

Subscribe delivers messages from queue on the returned channel, keeping at most prefetch messages in flight (delivered but not yet acked) at a time — the mechanism that lets multiple replicas share a queue fairly. A driver is expected to handle its own connection resilience: the returned channel should stay open across transient reconnects and close only when ctx is cancelled or the driver is closed.