# Queues with Ack/Nack & DLQ (/integrations/watermill/how-to/queues)



## Overview [#overview]

The **Queues** pattern (`PatternQueues`) maps Watermill's `message.Publisher` and `message.Subscriber` onto KubeMQ's reliable, point-to-point queues. Unlike Events (fire-and-forget) and EventsStore (persistent fan-out), Queues give you **at-least-once** delivery with **explicit acknowledgment**: every message is persisted on the broker until a consumer acknowledges it, and a failed delivery is redelivered rather than lost. This is the pattern to reach for when you are building task queues and job processing, where each unit of work must be handled exactly once and reliably.

For what the Queues pattern guarantees at the broker level, see [Queues](/learn/queues). This page documents how the plugin exposes it through the Watermill interfaces — the queue-specific config fields, ack/nack bridging, and native DLQ routing.

| Property    | Queues                                         |
| ----------- | ---------------------------------------------- |
| Delivery    | At-least-once                                  |
| Ack/Nack    | Explicit (`msg.Ack()` / `msg.Nack()`)          |
| Persistence | Yes (until acked)                              |
| Best for    | Task queues, job processing, reliable delivery |

A publisher writes messages to a queue channel; a subscriber polls the channel, processes each message, and settles it with an ack or a nack. Because the broker holds the message until it is acked, a consumer can crash mid-processing and the message will be redelivered.

<Mermaid
  chart="sequenceDiagram
    participant P as Publisher
    participant K as KubeMQ Queue
    participant C as Consumer
    P->>K: Publish (persisted)
    C->>K: Poll (MaxItems)
    K-->>C: Deliver message
    C->>C: Process
    C->>K: Ack (remove) / Nack (redeliver)"
/>

<Callout type="info">
  The Watermill plugin is a native gRPC client — it dials the broker's gRPC port (`50000`) directly, with no HTTP connector or enable flag to configure. If you have not set up a broker yet, see [Getting Started](/integrations/watermill/tutorials/getting-started).
</Callout>

## Publisher and Subscriber Configuration [#publisher-and-subscriber-configuration]

Both the publisher and subscriber are created with `Pattern: kubemq.PatternQueues`. The subscriber has two queue-specific fields that control polling:

<TypeTable
  type="{
  MaxItems: {
    description: 'Maximum messages fetched per poll cycle (Queues only).',
    type: 'int32',
    default: '1',
  },
  WaitTimeoutSeconds: {
    description: 'How long the server waits for messages before returning an empty poll (Queues only).',
    type: 'int32',
    default: '1',
  },
}"
/>

`MaxItems` accepts a value in the range 1–1000, and `WaitTimeoutSeconds` is the server-side poll timeout. For the Queues pattern, both fields are &#x2A;*default-corrected to `1`** whenever they are set to a value less than or equal to zero, so an unset subscriber still polls one message at a time with a one-second wait:

```go title="config.go (Validate)"
if c.Pattern == PatternQueues {
    if c.MaxItems <= 0 {
        c.MaxItems = 1
    }
    if c.WaitTimeoutSeconds <= 0 {
        c.WaitTimeoutSeconds = 1
    }
}
```

A minimal publish-and-receive setup looks like this — note that queues are persistent, so the publisher can write before any subscriber exists:

```go title="queues/send-receive/main.go"
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

func main() {
	logger := watermill.NewStdLogger(false, false)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Create queue publisher
	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		Logger:  logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	// Create queue subscriber
	// MaxItems: how many messages to fetch per poll cycle
	// WaitTimeoutSeconds: how long to wait for messages before returning empty
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	// Publish 3 queue messages (queues are persistent -- no need to subscribe first)
	for i := 1; i <= 3; i++ {
		msg := message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("Queue message %d", i)))
		if err := pub.Publish("watermill-queues.send-receive", msg); err != nil {
			log.Fatal(err)
		}
		fmt.Printf("Published: %s\n", string(msg.Payload))
	}

	// Subscribe and receive messages
	msgs, err := sub.Subscribe(ctx, "watermill-queues.send-receive")
	if err != nil {
		log.Fatal(err)
	}

	received := 0
	for received < 3 {
		select {
		case msg := <-msgs:
			fmt.Printf("Received: UUID=%s, Payload=%s\n", msg.UUID, string(msg.Payload))
			// Queue messages MUST be acknowledged to remove them from the queue.
			msg.Ack()
			received++
		case <-ctx.Done():
			log.Fatal("Timeout waiting for messages")
		}
	}

	fmt.Println("Done! All 3 queue messages received and acknowledged.")
}
```

A larger `MaxItems` reduces the number of round trips when the queue is busy: the subscriber fetches a batch per poll and delivers the messages one at a time on its output channel. See [Batch send](#batch-send) below for the matching publisher side.

## Ack and Nack Semantics [#ack-and-nack-semantics]

Queue messages are not removed from the broker on delivery — they are removed only when acknowledged. Each delivered Watermill message carries the two standard settlement calls:

* `msg.Ack()` — acknowledges the message, **removing it from the queue**.
* `msg.Nack()` — negative-acknowledges, **returning the message to the queue for redelivery**.

Internally, the subscriber bridges Watermill's settlement signals to the KubeMQ queue. For each delivered message it starts a small goroutine that waits on the message's `Acked()` or `Nacked()` channel and forwards the result to the underlying KubeMQ queue message. Critically, if the subscription context is cancelled while a message is still in flight, the bridge **nacks** it so the broker can redeliver it to another consumer rather than dropping it:

```go title="subscriber.go (ack/nack bridge)"
// Bridge goroutine: Watermill ack/nack -> KubeMQ ack/nack
s.wg.Add(1)
go func(qm *kubemqSDK.QueueDownstreamMessage, wm *message.Message) {
	defer s.wg.Done()
	select {
	case <-wm.Acked():
		if err := qm.Ack(); err != nil {
			s.config.Logger.Error("Queue ack error", err, watermill.LogFields{
				"topic": topic, "uuid": wm.UUID,
			})
		}
	case <-wm.Nacked():
		if err := qm.Nack(); err != nil {
			s.config.Logger.Error("Queue nack error", err, watermill.LogFields{
				"topic": topic, "uuid": wm.UUID,
			})
		}
	case <-subCtx.Done():
		_ = qm.Nack()
	}
}(qMsg, wmMsg)
```

The following example publishes one message, nacks it on first receive to simulate a processing failure (triggering redelivery), then acks it on the second receive:

```go title="queues/ack-nack/main.go"
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

func main() {
	logger := watermill.NewStdLogger(false, false)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		Logger:  logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	// Publish 1 message
	msg := message.NewMessage(watermill.NewUUID(), []byte("Important task"))
	if err := pub.Publish("watermill-queues.ack-nack", msg); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Published: %s\n", string(msg.Payload))

	msgs, err := sub.Subscribe(ctx, "watermill-queues.ack-nack")
	if err != nil {
		log.Fatal(err)
	}

	// First receive: Nack the message (simulates processing failure)
	select {
	case received := <-msgs:
		fmt.Printf("First receive: %s -- Nacking (simulating failure)\n", string(received.Payload))
		received.Nack() // Message goes back to the queue for redelivery
	case <-ctx.Done():
		log.Fatal("Timeout waiting for first message")
	}

	// Brief pause for redelivery
	time.Sleep(2 * time.Second)

	// Second receive: Ack the message (successful processing)
	select {
	case received := <-msgs:
		fmt.Printf("Second receive: %s -- Acking (processing success)\n", string(received.Payload))
		received.Ack() // Message removed from queue
	case <-ctx.Done():
		log.Fatal("Timeout waiting for redelivered message")
	}

	fmt.Println("Done! Message was nacked, redelivered, and then acked.")
}
```

## Queue Message Policy [#queue-message-policy]

A publisher can attach a `QueueMessagePolicy` that controls delivery behavior for every message it sends. The policy is set once on the `PublisherConfig` and applies to all messages from that publisher.

<TypeTable
  type="{
  ExpirationSeconds: {
    description: 'Message TTL in seconds. After this, an unconsumed message expires. 0 = no expiration.',
    type: 'int',
  },
  DelaySeconds: {
    description: 'Delay before the message becomes visible to consumers.',
    type: 'int',
  },
  MaxReceiveCount: {
    description: 'Maximum delivery attempts before the message is routed to the DLQ.',
    type: 'int',
  },
  MaxReceiveQueue: {
    description: 'DLQ channel name that over-delivered messages are moved to.',
    type: 'string',
  },
}"
/>

These four fields cover the three reliability features most task queues need: time-to-live (`ExpirationSeconds`), delayed/scheduled visibility (`DelaySeconds`), and automatic dead-lettering (`MaxReceiveCount` + `MaxReceiveQueue`). The sections below show each in isolation.

## Dead Letter Queue [#dead-letter-queue]

When a message is repeatedly redelivered without ever being acked, you usually do not want it to cycle forever. Set `MaxReceiveCount` and `MaxReceiveQueue` on the publisher's policy and KubeMQ will **automatically route** a message that exceeds the receive count to the named DLQ channel — no extra code on the consumer side.

In the example below the publisher routes anything received more than three times to `watermill-queues.dlq`. The consumer nacks the message three times, after which the broker moves it to the DLQ, where a second subscriber picks it up:

```go title="queues/dead-letter-queue/main.go"
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

func main() {
	logger := watermill.NewStdLogger(false, false)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Create publisher with DLQ policy
	// MaxReceiveCount: message moves to DLQ after 3 failed receive attempts
	// MaxReceiveQueue: the DLQ channel name
	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		QueueMessagePolicy: &kubemq.QueueMessagePolicy{
			MaxReceiveCount: 3,
			MaxReceiveQueue: "watermill-queues.dlq",
		},
		Logger: logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	// Publish 1 message
	msg := message.NewMessage(watermill.NewUUID(), []byte("Problematic task"))
	if err := pub.Publish("watermill-queues.dlq-source", msg); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Published: %s\n", string(msg.Payload))

	// Create subscriber for the source queue
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	msgs, err := sub.Subscribe(ctx, "watermill-queues.dlq-source")
	if err != nil {
		log.Fatal(err)
	}

	// Nack the message 3 times to trigger DLQ
	for attempt := 1; attempt <= 3; attempt++ {
		select {
		case received := <-msgs:
			fmt.Printf("Attempt %d: Nacking message: %s\n", attempt, string(received.Payload))
			received.Nack()
		case <-ctx.Done():
			log.Fatal("Timeout waiting for message")
		}
		time.Sleep(time.Second) // Brief pause between attempts
	}

	time.Sleep(2 * time.Second) // Allow DLQ transfer

	// Subscribe to the DLQ to confirm the message was moved there
	dlqSub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer dlqSub.Close()

	dlqMsgs, err := dlqSub.Subscribe(ctx, "watermill-queues.dlq")
	if err != nil {
		log.Fatal(err)
	}

	select {
	case dlqMsg := <-dlqMsgs:
		fmt.Printf("DLQ received: %s\n", string(dlqMsg.Payload))
		dlqMsg.Ack()
	case <-ctx.Done():
		log.Fatal("Timeout waiting for DLQ message")
	}

	fmt.Println("Done! Message moved to dead letter queue after 3 failed attempts.")
}
```

### Native DLQ vs. Poison Queue Middleware [#native-dlq-vs-poison-queue-middleware]

KubeMQ's native DLQ is not the only way to dead-letter a message. Watermill ships a `PoisonQueue` middleware that does something similar at a different layer. The two solve different problems and can be combined.

| Mechanism                             | Layer       | Catches                                       | Configured by                                      |
| ------------------------------------- | ----------- | --------------------------------------------- | -------------------------------------------------- |
| **Native DLQ** (`QueueMessagePolicy`) | Transport   | Consumer crashes, repeatedly unacked messages | Publisher `MaxReceiveCount` + `MaxReceiveQueue`    |
| **Poison Queue middleware**           | Application | Handler panics, business-logic errors         | `middleware.PoisonQueue(pub, topic)` on the Router |

The native DLQ is driven by the broker counting receive attempts, so it captures transport-level failures even when the consumer never gets a chance to return an error (for example, it crashes mid-processing). The Poison Queue middleware wraps your handler and catches the **error it returns**, publishing the failed message to a dedicated topic — so it captures application-level failures that the broker would otherwise see as a normal nack.

```go title="middleware/poison-queue/main.go (excerpt)"
// PoisonQueue middleware: messages that fail processing are sent to the poison topic.
// It wraps the handler and catches errors, publishing the failed message to the
// specified topic via the provided publisher.
poisonQueue, err := middleware.PoisonQueue(poisonPub, "watermill-mw.poison")
if err != nil {
	log.Fatal(err)
}
router.AddMiddleware(poisonQueue)

// Handler that always fails -- the message will end up in the poison queue
router.AddNoPublisherHandler(
	"always-failing",
	"watermill-mw.poison-input",
	sub,
	func(msg *message.Message) error {
		fmt.Printf("Handler received (will fail): %s\n", string(msg.Payload))
		return fmt.Errorf("processing failed: invalid data format")
	},
)
```

<Callout type="info">
  Use the native DLQ to guard against infrastructure failures and the Poison Queue middleware to isolate messages your handler cannot process. They operate independently — running both gives you coverage at the transport and application layers at once.
</Callout>

## Delayed and Expiring Messages [#delayed-and-expiring-messages]

The same `QueueMessagePolicy` controls message timing.

**Delayed visibility** (`DelaySeconds`): the message is published immediately but the broker withholds it from consumers until the delay elapses — useful for scheduled or deferred work.

```go title="queues/delayed-messages/main.go (excerpt)"
// Create publisher with a 5-second delay policy
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
	Address: "localhost:50000",
	Pattern: kubemq.PatternQueues,
	QueueMessagePolicy: &kubemq.QueueMessagePolicy{
		DelaySeconds: 5,
	},
	Logger: logger,
})
```

**Expiration / TTL** (`ExpirationSeconds`): if no consumer reads the message within the window, it expires and is never delivered — useful for time-sensitive work that is worthless if stale.

```go title="queues/expiration/main.go (excerpt)"
// Create publisher with a 3-second expiration policy
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
	Address: "localhost:50000",
	Pattern: kubemq.PatternQueues,
	QueueMessagePolicy: &kubemq.QueueMessagePolicy{
		ExpirationSeconds: 3,
	},
	Logger: logger,
})
```

Run the full programs to see the timing in action:

```bash
go run ./examples/queues/delayed-messages/main.go
go run ./examples/queues/expiration/main.go
```

## Competing Consumers [#competing-consumers]

To process a queue in parallel, run multiple subscribers that share the same `ConsumerGroup`. KubeMQ distributes the queue's messages across the group members, so each message is handled by exactly one consumer — the standard competing-consumers pattern for scaling out workers.

```go title="queues/competing-consumers/main.go (excerpt)"
// Create 3 subscribers with the same consumer group
channels := make([]<-chan *message.Message, 3)
for i := 0; i < 3; i++ {
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		ConsumerGroup:      "queue-workers",
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	msgs, err := sub.Subscribe(ctx, "watermill-queues.competing-consumers")
	if err != nil {
		log.Fatal(err)
	}
	channels[i] = msgs
}
```

The `queues/competing-consumers` example publishes nine tasks and watches them spread across three subscribers in the `queue-workers` group. For a Router-based take on the same idea — where each worker is a Router handler rather than a raw channel read — see `advanced/competing-consumers`.

```bash
go run ./examples/queues/competing-consumers/main.go
go run ./examples/advanced/competing-consumers/main.go
```

## Batch Send [#batch-send]

`Publish` is variadic, so you can hand it several messages in a single call. On the Queues pattern these are sent upstream to the broker as one batch rather than one request per message, which cuts round trips when you are enqueuing many tasks at once:

```go title="queues/batch-send/main.go (excerpt)"
// Build 10 messages
msgs := make([]*message.Message, 10)
for i := 0; i < 10; i++ {
	msgs[i] = message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("Batch message %d", i+1)))
}

// Publish all 10 messages in one call.
// Internally this sends them as a single batch via queueUpstream.Send(requestID, queueMsgs).
if err := pub.Publish("watermill-queues.batch-send", msgs...); err != nil {
	log.Fatal(err)
}
fmt.Printf("Published batch of %d messages\n", len(msgs))
```

Pair a batch publisher with a subscriber that sets a higher `MaxItems` (the `batch-send` example uses `MaxItems: 5`) to fetch several messages per poll on the consumer side as well.

## Using the Router with Queues [#using-the-router-with-queues]

The Watermill Router works the same with Queues as it does with Events, with one important behavioral difference: because Queues support ack/nack, the Router **acks a message only after the handler returns successfully** and nacks it on error. That turns the Router into a reliable processing loop — a handler that fails will have its message redelivered.

The `queues/router-handler` example wires a handler that reads from a pending queue, transforms the payload, and forwards the result to a completed queue:

```go title="queues/router-handler/main.go (excerpt)"
router.AddMiddleware(middleware.Recoverer)

// Add handler: process pending tasks and publish to the completed queue
router.AddHandler(
	"task-processor",
	"watermill-queues.pending",
	sub,
	"watermill-queues.completed",
	pub,
	func(msg *message.Message) ([]*message.Message, error) {
		result := fmt.Sprintf("DONE: %s", strings.ToUpper(string(msg.Payload)))
		fmt.Printf("Handler processed: %s -> %s\n", string(msg.Payload), result)
		outMsg := message.NewMessage(watermill.NewUUID(), []byte(result))
		return []*message.Message{outMsg}, nil
	},
)
```

```bash
go run ./examples/queues/router-handler/main.go
```

## A Note on the Retry Middleware [#a-note-on-the-retry-middleware]

Watermill's `Retry` middleware re-invokes a handler when it returns an error, but it only does anything useful when a failed message can actually be redelivered. Because **Queues are the only pattern where a nack triggers redelivery**, `Retry` is meaningful for Queues and is not applicable to Events or EventsStore:

| Middleware   | Events | EventsStore | Queues | Notes                                                    |
| ------------ | ------ | ----------- | ------ | -------------------------------------------------------- |
| Retry        | N/A    | N/A         | Yes    | Nack triggers redelivery; only meaningful for Queues     |
| Poison Queue | N/A    | N/A         | Yes    | Application-level DLQ (alternative to KubeMQ native DLQ) |

<Callout type="info">
  Pair `Retry` with the Poison Queue middleware: retry handles transient failures in place, and anything that exhausts its retries is routed to the poison topic instead of blocking the consumer.
</Callout>

## Next Steps [#next-steps]

<Cards>
  <Card title="Concepts" href="/integrations/watermill/concepts/concepts" description="How the plugin maps Watermill Publishers, Subscribers, and the Router onto KubeMQ's three messaging patterns." />

  <Card title="Getting Started" href="/integrations/watermill/tutorials/getting-started" description="Install the plugin, start a broker, and run a first end-to-end example." />

  <Card title="Reference" href="/integrations/watermill/reference/configuration" description="Full PublisherConfig, SubscriberConfig, and QueueMessagePolicy field reference." />
</Cards>