Message processing becomes more complicated when an application must survive restarts, temporary failures, and network interruptions. A service may receive a message, begin processing it, and then terminate before completing the operation. When the service starts again, it must determine which messages still need attention without losing work or creating unintended duplicate effects.
NATS JetStream provides persistence, acknowledgments, and durable consumers to help applications recover from these situations. Unlike a basic in-memory messaging workflow, JetStream can retain messages and track consumer progress so that processing can continue after a consumer disconnects.
However, persistence alone does not guarantee that every message is processed exactly once at the application level. Developers must configure acknowledgment behavior, understand redelivery, and make message handlers safe to execute more than once.
This article explains how JetStream consumers work, how durable consumers resume processing after a restart, and how to implement a reliable message-processing workflow using Go.
What Is NATS JetStream?
NATS is a messaging system designed for communication between distributed applications. It supports publish-subscribe messaging, request-reply communication, and other messaging patterns.
JetStream adds persistence and message-streaming capabilities to NATS. It allows messages to be stored in streams and consumed by applications that may connect and disconnect over time.
A JetStream deployment typically involves three important components:
Stream: Stores messages that match configured subjects.
Consumer: Tracks how messages from a stream are delivered and acknowledged.
Subscriber application: Receives messages and performs the required business operation.
For example, an order-processing application might publish events to a subject named orders.created. A JetStream stream can retain those events, while a durable consumer delivers them to a worker responsible for processing orders.
If the worker restarts, the consumer can continue delivering messages according to its stored state and acknowledgment configuration.
Understanding Durable and Ephemeral Consumers
Choosing the correct consumer type is essential when message processing must survive application restarts.
Durable consumers
A durable consumer has a persistent identity and retains its delivery and acknowledgment state according to the server's configuration and retention rules.
When a worker disconnects, its durable consumer can remain available. When the worker reconnects and binds to the same consumer, it can continue consuming messages based on the consumer's stored state.
This makes durable consumers appropriate for background workers, order processing, event-driven integrations, and other workflows that need continuity across restarts.
A durable consumer does not mean the application process itself remains running. It means the consumer's state is maintained independently of the application's connection.
Ephemeral consumers
An ephemeral consumer is generally intended for temporary consumption. Its lifecycle is managed by the server, and it can be removed when it is no longer active, depending on its configuration and server behavior.
Ephemeral consumers are useful for short-lived subscriptions and workloads that do not require a stable consumer identity across application restarts.
For a worker that must resume processing after a restart, a durable consumer is usually the more appropriate choice.
Feature | Durable consumer | Ephemeral consumer |
|---|
Consumer identity | Persistent identity | Temporary identity |
Recovery after application restart | Can reconnect to the same consumer | May require creating a new consumer |
Stored delivery state | Retained according to configuration | Depends on its lifecycle and configuration |
Typical use case | Background processing and reliable workflows | Temporary subscriptions and short-lived tasks |
How JetStream Resumes Message Processing
JetStream uses message acknowledgments to track whether delivered messages have been successfully processed.
A typical workflow looks like this:
The application subscribes to a durable consumer.
JetStream delivers a message to the worker.
The worker performs the business operation.
The worker acknowledges the message after successful processing.
JetStream records the acknowledgment and advances the consumer's delivery state.
If the worker crashes before acknowledging the message, JetStream can redeliver it according to the consumer's acknowledgment policy and acknowledgment timeout.
This behavior allows the application to recover work that was interrupted.
However, it also creates an important distinction between message delivery and business processing. JetStream can track acknowledgments, but it cannot automatically determine whether an external database transaction or payment operation completed successfully before the worker crashed.
That is why reliable consumers must be designed with failure recovery and duplicate handling in mind.
Setting Up NATS JetStream
Before implementing the consumer, start a NATS server with JetStream enabled.
For local development, the NATS server can be started with the JetStream flag:
nats-server -js
The command assumes that the NATS server is installed and available in the current environment.
For a local development environment, you can also use Docker:
docker run --name nats-jetstream \
-p 4222:4222 \
nats -js
This starts a NATS server with JetStream enabled and exposes the client port.
The example is suitable for local experimentation, not production deployment. Production systems need appropriate persistence volumes, resource limits, monitoring, access controls, and a deployment strategy for server recovery.
JetStream stores messages according to the stream's storage and retention configuration. A container that is removed without persistent storage may lose its locally stored data.
Implementing a Durable Consumer in Go
The following example uses the official Go NATS client with its JetStream API.
Install the client package:
go get github.com/nats-io/nats.go
The application will create a stream, configure a durable consumer, publish messages, and process them with explicit acknowledgments.
Step 1: Connect to NATS
Create a Go file named main.go and establish a connection.
package main
import (
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
nc, err := nats.Connect(
nats.DefaultURL,
nats.Name("Order Processor"),
nats.Timeout(5*time.Second),
)
if err != nil {
log.Fatalf("failed to connect to NATS: %v", err)
}
defer nc.Close()
log.Println("Connected to NATS")
}
The connection establishes communication with the NATS server.
The timeout controls how long the client waits during connection establishment. Production applications should also configure reconnection behavior and handle connection lifecycle events where appropriate.
The defer nc.Close() call ensures that the connection is closed when the program exits normally. It does not replace graceful shutdown handling for a long-running worker.
Step 2: Create a stream
A stream defines which subjects are stored and how messages are retained.
Add the following code after connecting:
js, err := nc.JetStream()
if err != nil {
log.Fatalf("failed to initialize JetStream: %v", err)
}
_, err = js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"orders.created"},
Storage: nats.FileStorage,
Retention: nats.LimitsPolicy,
MaxAge: 24 * time.Hour,
})
if err != nil {
log.Fatalf("failed to create stream: %v", err)
}
The ORDERS stream stores messages published to orders.created.
The example uses file storage and a maximum message age of 24 hours. These settings are illustrative. A production retention policy must reflect message volume, recovery objectives, storage capacity, and business requirements.
The code assumes the stream does not already exist. If the application is restarted, AddStream may return an error because the stream already exists.
For production deployments, provision streams through infrastructure configuration or use an explicit create-or-update process that validates existing settings. Avoid blindly recreating streams every time a worker starts.
Step 3: Create a durable consumer
Configure a durable consumer with explicit acknowledgments.
_, err = js.AddConsumer("ORDERS", &nats.ConsumerConfig{
Durable: "order-worker",
AckPolicy: nats.AckExplicitPolicy,
AckWait: 30 * time.Second,
MaxDeliver: 5,
FilterSubject: "orders.created",
})
if err != nil {
log.Fatalf("failed to create consumer: %v", err)
}
The consumer belongs to the ORDERS stream and uses the durable name order-worker.
The acknowledgment policy is important:
AckExplicitPolicy requires the application to acknowledge each message.
AckWait defines how long JetStream waits for an acknowledgment before making the message eligible for redelivery.
MaxDeliver limits delivery attempts according to the consumer's configuration.
FilterSubject restricts delivery to messages matching the specified subject.
The AckWait value must be appropriate for the expected processing duration. If processing regularly exceeds the timeout, the same message may be delivered again while the original processing attempt is still running.
In production, consider configuring a backoff schedule for repeated delivery attempts and implementing an explicit strategy for messages that repeatedly fail.
As with stream creation, handle the case where the consumer already exists. A worker should reconnect to its intended durable consumer instead of failing because the consumer was created during a previous run.
Step 4: Publish a message
The application can publish an event to the stream.
payload := []byte(`{
"orderId": "ORD-1001",
"customerId": "CUST-42",
"amount": 2499.00
}`)
ack, err := js.Publish("orders.created", payload)
if err != nil {
log.Fatalf("failed to publish order: %v", err)
}
log.Printf(
"Published message to stream %s at sequence %d",
ack.Stream,
ack.Sequence,
)
The JetStream publish operation waits for a server acknowledgment that confirms the message was accepted according to the server's publishing behavior.
This is different from publishing through ordinary Core NATS, where a successful client-side publish does not provide the same JetStream persistence acknowledgment.
For important events, the publisher should handle failed publishes and retries. Retrying without a deduplication strategy can create duplicate events, so consider message identifiers and JetStream's deduplication capabilities where appropriate.
Step 5: Subscribe and process messages
The worker subscribes using the same durable consumer name.
sub, err := js.PullSubscribe(
"orders.created",
"order-worker",
)
if err != nil {
log.Fatalf("failed to subscribe: %v", err)
}
defer sub.Unsubscribe()
for {
messages, err := sub.Fetch(
10,
nats.MaxWait(5*time.Second),
)
if err != nil {
if err == nats.ErrTimeout {
continue
}
log.Printf("failed to fetch messages: %v", err)
time.Sleep(time.Second)
continue
}
for _, msg := range messages {
if err := processOrder(msg.Data); err != nil {
log.Printf("order processing failed: %v", err)
if nakErr := msg.Nak(); nakErr != nil {
log.Printf(
"failed to negatively acknowledge message: %v",
nakErr,
)
}
continue
}
if err := msg.Ack(); err != nil {
log.Printf(
"failed to acknowledge message: %v",
err,
)
}
}
}
This example uses a pull consumer. The worker explicitly requests batches of messages rather than receiving them continuously without requesting delivery.
Pull consumers are useful when the application needs control over how many messages it fetches at a time.
The code acknowledges a message only after processOrder returns successfully. If processing fails, it sends a negative acknowledgment so JetStream can redeliver the message according to the consumer's policy.
The Fetch call can return a timeout when no messages arrive within the configured wait period. The worker handles that condition by continuing to poll.
The example focuses on the message-processing flow. Production code should also handle cancellation, shutdown, subscription errors, and delivery limits carefully.
The helper function can be defined as follows:
func processOrder(data []byte) error {
// Decode the message.
// Validate the order.
// Apply the business operation.
// Return an error if processing fails.
log.Printf("Processing order event: %s", string(data))
return nil
}
Replace the placeholder with actual business logic. Logging a message is not equivalent to completing an order-processing operation.
What Happens When the Consumer Restarts?
Suppose the worker receives a message and begins processing it. Before it sends an acknowledgment, the process crashes.
When the worker starts again and reconnects to the same durable consumer, JetStream retains the consumer's delivery state. Because the previous message was not acknowledged, it can be delivered again.
The outcome depends on where the failure occurred.
Failure point | Expected behavior |
|---|
Worker crashes before receiving the message | The message remains available for delivery |
Worker crashes after delivery but before processing | The unacknowledged message can be redelivered |
Worker crashes during processing | The message can be redelivered after the applicable timeout or another redelivery trigger |
Worker completes processing but crashes before acknowledgment | The operation may run again when the message is redelivered |
Worker acknowledges after successful processing | JetStream records the acknowledgment and normally does not redeliver that message as unacknowledged work |
The fourth case is especially important.
Imagine that the worker writes an order to a database successfully, but crashes before acknowledging the message. JetStream cannot know that the database operation succeeded. It may deliver the message again, causing the worker to attempt the same operation twice.
This is a fundamental challenge in distributed message processing.
Handling Duplicate Messages Safely
Durable consumers support recovery, but reliable application behavior also requires idempotent processing.
An idempotent operation can be repeated without producing an unintended additional effect.
For example, instead of blindly inserting an order every time an event arrives, the application can use a unique order identifier and enforce uniqueness in the database.
func processOrder(data []byte) error {
// Decode and validate the order event.
// Start a database transaction.
// Check whether the event or order has already been processed.
// Apply the business change only if it has not been processed.
// Record the event identifier in the same transaction.
// Commit the transaction.
return nil
}
This code illustrates the approach rather than a complete database implementation.
The duplicate check and business change should be coordinated transactionally when possible. Otherwise, a crash between the check and the write can still introduce duplicate effects.
A practical design may use an inbox table with a unique event identifier. The application records that identifier and applies the business update in the same database transaction. If the event arrives again, the unique constraint prevents the business operation from being applied a second time.
JetStream's message deduplication capabilities can also help publishers avoid duplicate stream entries when publishing with stable message identifiers within the configured deduplication window. This does not replace idempotency in consumers.
For external side effects, such as calling a payment provider, use the provider's idempotency mechanism when available.
Choosing an Acknowledgment Strategy
JetStream supports several acknowledgment modes. The correct choice depends on the workload and the consequences of processing a message more than once.
Explicit acknowledgment
With explicit acknowledgment, the application acknowledges each message individually after processing it.
This is generally a good default for independent background jobs because it provides clear control over when each message is considered complete.
Negative acknowledgment
A negative acknowledgment tells JetStream that the application did not successfully process the message and that it may be delivered again.
Use it for failures that are potentially recoverable. However, repeatedly retrying a permanently invalid message can create an endless failure loop.
Consider distinguishing transient failures, such as a temporary database outage, from permanent failures, such as malformed input.
Delayed negative acknowledgment
A delayed negative acknowledgment can postpone redelivery. This helps prevent a failing dependency from being overwhelmed by immediate retries.
Use a retry delay that reflects the failure type and the application's recovery expectations.
Acknowledgment timeout
If the worker does not acknowledge a message within the configured acknowledgment timeout, JetStream can redeliver it.
For long-running tasks, consider sending an acknowledgment progress signal using the appropriate client API, or adjust the timeout based on measured processing duration. Ensure the handler remains safe if the message is eventually redelivered.
Common Mistakes and Troubleshooting
The consumer starts from the beginning unexpectedly
Verify that the application is using the intended durable consumer name and that it has not created a different consumer accidentally.
Also inspect the consumer's delivery policy and stored state. A newly created consumer may start from a configured point in the stream rather than resume another consumer's progress.
Messages are delivered more than once
Check whether the worker acknowledges messages only after completing the business operation. Investigate acknowledgment timeouts, negative acknowledgments, process crashes, and overlapping processing attempts.
Duplicate delivery is a normal possibility in a reliable messaging workflow. The application should not assume that every message will be delivered only once.
The same failing message keeps returning
Inspect the delivery count and the reason processing fails. Configure retry delays and a maximum delivery policy appropriate to the workload.
Messages that repeatedly fail may require a dead-letter or quarantine workflow. JetStream does not automatically move every message that reaches its delivery limit into a separate dead-letter stream; the application must implement the desired handling.
The worker stops receiving messages
Confirm that the stream captures the expected subject, that the durable consumer's filter matches it, and that the worker is connected to the correct NATS server.
Check consumer state, pending messages, acknowledgment-pending counts, delivery limits, and any pull-fetch errors. Also confirm that stream retention policies have not removed the messages.
Stream creation fails after a restart
The stream may already exist. Separate infrastructure provisioning from worker startup or implement an idempotent configuration process that checks existing settings before attempting changes.
Do not delete and recreate a stream merely to make startup succeed. Doing so can remove retained messages or disrupt consumer state.
Best Practices for Production
Provision streams and consumers deliberately
Treat stream and consumer configuration as part of infrastructure management. Validate subjects, retention policies, storage requirements, and acknowledgment settings during deployment.
Keep configuration changes separate from ordinary message-processing code where practical.
Use bounded pull batches
Choose batch sizes based on message size, processing time, concurrency, and available memory.
Fetching too many messages can increase memory consumption and the number of messages waiting for acknowledgment. Fetching too few may limit throughput.
Set retry and acknowledgment policies intentionally
Measure actual processing durations and use realistic acknowledgment timeouts. Configure redelivery limits and retry delays based on the expected failure patterns.
Avoid treating every error as permanently fatal or retrying every failure immediately.
Monitor consumer state
Track pending messages, acknowledgment-pending counts, redelivery counts, processing latency, and failure rates.
A growing pending-message count can indicate that the producer is publishing faster than consumers can process messages or that a worker has stopped making progress.
Design for graceful shutdown
When a worker receives a termination signal, stop fetching new messages and allow in-flight processing to complete within a controlled shutdown period.
Do not acknowledge a message merely because the process is shutting down. Acknowledge it only when the business operation has completed successfully.
If a task cannot finish before shutdown, leave the message unacknowledged so it can be recovered according to the consumer's configuration.
Conclusion
NATS JetStream provides the persistence and consumer state needed to resume message processing after a worker restarts. Durable consumers preserve a stable consumer identity, while explicit acknowledgments allow JetStream to distinguish completed work from messages that may need to be delivered again.
The key is to design the entire processing workflow around possible failures. A message may be delivered more than once, especially when a worker crashes after completing a database operation but before acknowledging the message.
Use durable consumers, acknowledge only after successful processing, configure retries and timeouts carefully, and make business operations idempotent. Combine these practices with monitoring and graceful shutdown handling to build message-processing services that recover predictably from interruptions.
Join the conversation! Your thoughts help the community grow.