Solving Message Loss with NATS JetStream Durable Consumers

The Problem: The \"Fire and Forget\" Gap

\n

In a standard NATS pub/sub model, messages are delivered to active subscribers and then immediately discarded by the server. If a subscriber crashes or restarts, any messages published during that downtime are lost forever. For critical workflows—like processing payments or updating inventory—this \"fire and forget\" behavior is a liability.

\n\n

The Solution: Durable Consumers

\n

JetStream extends NATS by adding a persistence layer. By using a Durable Consumer, the NATS server tracks which messages a specific client has acknowledged. If the client disconnects, the server remembers the last processed message (the ack floor). When the client reconnects using the same durable name, the server resumes delivery from exactly where it left off.

\n\n

Configuring the Persistence Layer

\n

To implement this, you must first define a Stream (the storage bucket) and then a Consumer (the pointer to the data). These commands should be run via the NATS CLI on a machine with network access to the server.

\n\n

1. Enable JetStream

\n

Start your server with the -js flag to enable the JetStream subsystem:

\n

nats-server -js

\n\n

2. Define the Stream

\n

Create a stream to capture messages on specific subjects. We use --storage=file to ensure data survives a full server reboot.

\n

# Create a stream named ORDERS for all subjects starting with ORDERS.\n# Example: nats stream add ORDERS --subjects=\"ORDERS.*\" --retention=limits --storage=file\n# nats stream add ORDERS --subjects=\"ORDERS.*\" --retention=limits --storage=file

\n\n

3. Create the Durable Consumer

\n

The --durable flag tells NATS to store the consumer's state. We use --ack explicit to ensure the server only marks a message as delivered once the client confirms successful processing.

\n

# Create a durable consumer named 'order-processor'\n# Example: nats consumer add ORDERS order-processor --durable --ack explicit --ack-wait=30s\n# nats consumer add ORDERS order-processor --durable --ack explicit --ack-wait=30s

\n\n

Worked Example: Go Implementation

\n

The following example demonstrates a publisher and a subscriber. Both require the github.com/nats-io/nats.go package. They should be run with network access to port 4222.

\n\n

The Publisher

\n

package main\n\nimport (\n    "github.com/nats-io/nats.go"\n    "log"\n)\n\nfunc main() {\n    nc, _ := nats.Connect(nats.DefaultURL)\n    js, _ := nc.JetStream()\n\n    for i := 1; i <= 5; i++ {\n        msg := []byte("Order Data " + string(rune('0'+i)))\n        js.Publish("ORDERS.new", msg)\n        log.Printf("Published message %d", i)\n    }\n}

\n\n

The Durable Subscriber

\n

Note how the subscriber specifies the durable name order-processor to retrieve its saved state.

\n

package main\n\nimport (\n    "github.com/nats-io/nats.go"\n    "log"\n)\n\nfunc main() {\n    nc, _ := nats.Connect(nats.DefaultURL)\n    js, _ := nc.JetStream()\n\n    // Bind to the existing durable consumer\n    cons, _ := js.Consume("ORDERS", "order-processor", func(msg *nats.Msg) {\n        log.Printf("Processing: %s", string(msg.Data))\n        msg.Ack() // Confirm processing is complete\n    })\n    _ = cons\n}

\n\n

Trade-offs and Operational Risks

\n

\n
• Disk Pressure: Because JetStream persists data to disk, unacknowledged messages or long retention periods can fill your storage. Always set a --max-bytes or --max-age limit on your streams.
\n
• The Redelivery Loop: If your code fails to call msg.Ack() within the --ack-wait window (30s in our example), NATS will redeliver the message. To prevent duplicate processing, your business logic must be idempotent (processing the same message twice has no additional effect).
\n

\n\n

Verifying the Result

\n

To confirm the durable consumer is working as expected, follow this sequence:

\n

\n
1. Run the publisher to send 5 messages.
\n
2. Run the subscriber; it should process all 5 messages.
\n
3. Stop the subscriber. Run the publisher again to send 5 more messages.
\n
4. Restart the subscriber. It should immediately process the 5 messages sent while it was offline.
\n

\n

You can check the current state of the consumer via the CLI:

\n

nats consumer info ORDERS order-processor

\n

Look for AckFloor to see how many messages have been permanently processed.

\n\n

Rollback

\n

To remove the stream and all associated consumer state from the server, run:

\n

nats stream rm ORDERS