NATS JetStream Event-Driven Architecture

Overview

NATS JetStream is the persistence and streaming layer built into the NATS messaging system. It adds durable message storage, at-least-once delivery, consumer groups, and replay capabilities on top of NATS’ core publish-subscribe model, making it suitable for event-driven architectures that require reliability guarantees without deploying a separate message broker like Kafka or RabbitMQ.

Event-driven architecture (EDA) decouples services by having them communicate through events rather than direct RPC calls. Producers emit events to streams, and consumers subscribe to those streams to react asynchronously. This pattern improves resilience, scalability, and deployability—services can be added, removed, or upgraded independently as long as they understand the event schema.

In a typical document search application, NATS JetStream handles analytics events, document ingestion pipelines, and inter-service coordination. Search indexing workers consume document-change events from a JetStream stream, process them asynchronously, and acknowledge completion—enabling the system to handle traffic spikes by buffering events in the stream rather than dropping or backpressuring producers.

Core Concepts

• Stream: A named, durable log of messages on one or more subjects. Streams define retention policies (limits on age, count, or bytes), storage type (file or memory), and replication factor for high availability.
• Consumer: A stateful subscription to a stream that tracks which messages have been delivered and acknowledged. Consumers can be durable (surviving restarts) or ephemeral, and can operate in push or pull mode.
• Subject Mapping: Streams capture messages published to matching NATS subjects (with wildcards). A single stream can aggregate events from multiple subjects, e.g., events.analytics.* and events.ingestion.*.
• Acknowledgements: JetStream supports explicit ack, negative ack (nak), and in-progress signals. Unacknowledged messages are redelivered after a configurable timeout, providing at-least-once semantics.
• Exactly-Once Deduplication: Publishers can attach a unique Nats-Msg-Id header; JetStream deduplicates within a configurable window, enabling exactly-once publishing semantics.

How It Works

Stream Configuration and Storage

When you create a JetStream stream, you specify which subjects it captures, the storage backend (file for durability, memory for speed), retention policy, and replication factor. JetStream stores messages in a Raft-replicated log across the configured number of NATS server nodes, ensuring durability even if servers fail.

Each message in the stream receives a monotonically increasing sequence number. The stream maintains an index from sequence number to message, enabling efficient replay from any point. Retention policies automatically discard old messages—by age (e.g., keep 7 days), count (e.g., keep last 1M messages), or total bytes.

Consumer Lifecycle

A durable consumer is created with a name and a delivery policy (deliver all, deliver last, deliver new, or deliver from a specific sequence or time). The server tracks which messages have been delivered and acknowledged. When the consumer reconnects, it resumes from its last acknowledged position.

For pull consumers, the client calls Fetch(batch) to request a batch of messages. The server delivers up to batch messages and the client processes and acknowledges them. If processing fails, the client sends a Nak or simply lets the ack timeout expire, causing redelivery. A MaxDeliver setting prevents infinite redelivery loops by sending repeatedly failing messages to a dead-letter subject.

Event Flow in a Microservices Architecture

In a typical EDA setup, a web API publishes events (e.g., events.documents.created) when users upload content. Multiple JetStream consumers process these events independently: an indexing service updates the search index, an analytics service records metrics, and a notification service alerts subscribers. Each consumer operates at its own pace, and if one falls behind, the stream buffers events until it catches up.

Practical Applications

In a typical search application, the document ingestion pipeline uses JetStream to decouple the upload API from the indexing workers. When a document is uploaded, the API publishes an ingest.document.new event with metadata (document ID, source, content hash). A pool of indexing workers, configured as a consumer group, pulls events in batches, downloads the document content, extracts text, builds index segments, and acknowledges completion.

Analytics events flow through a separate stream (analytics.*) that captures user search queries, click-through data, and latency measurements. A real-time dashboard consumer processes these events with a sliding window to power live metrics, while a batch consumer replays the same stream nightly to generate aggregate reports.

JetStream’s replay capability is valuable for debugging and development. When a bug is discovered in event processing logic, engineers can reset a consumer’s position to replay historical events through the fixed code, without needing to re-emit events from upstream services.

Implementation Patterns

Idempotent Event Handlers

Since JetStream provides at-least-once delivery, consumers must be prepared to handle duplicate messages. The standard pattern is to derive a deterministic idempotency key from each event (e.g., the document ID and event type) and check whether the event has already been processed before applying side effects.

Ordered Consumers for Sequential Processing

When events for a specific entity must be processed in order (e.g., document version 1 before version 2), use NATS subject partitioning: publish events to events.documents.{docID} and configure the consumer with a subject filter. Because JetStream preserves per-subject ordering, all events for a given document are processed sequentially by the same consumer instance.

Dead Letter Queues

Configure MaxDeliver on consumers to limit redelivery attempts. When a message exceeds the limit, JetStream publishes it to an advisory subject. A dead-letter consumer monitors this subject, logs the failure, and stores the poisoned message for manual inspection or automated retry with different processing logic.

Code Examples

Setting Up a JetStream Stream and Consumer in Go

package main

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

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func setupJetStream() (jetstream.Stream, jetstream.Consumer, error) {
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        return nil, nil, fmt.Errorf("connect: %w", err)
    }

    js, err := jetstream.New(nc)
    if err != nil {
        return nil, nil, fmt.Errorf("jetstream: %w", err)
    }

    ctx := context.Background()

    // Create or update the stream
    stream, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
        Name:      "EVENTS",
        Subjects:  []string{"events.>"},
        Storage:   jetstream.FileStorage,
        Retention: jetstream.LimitsPolicy,
        MaxAge:    7 * 24 * time.Hour, // Keep 7 days
        MaxBytes:  10 << 30,           // 10 GB cap
        Replicas:  3,                   // Raft replication
        Discard:   jetstream.DiscardOld,
    })
    if err != nil {
        return nil, nil, fmt.Errorf("create stream: %w", err)
    }

    // Create a durable pull consumer for indexing
    consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
        Durable:       "indexing-workers",
        FilterSubject: "events.documents.>",
        AckPolicy:     jetstream.AckExplicitPolicy,
        AckWait:       30 * time.Second,
        MaxDeliver:    5,
        DeliverPolicy: jetstream.DeliverAllPolicy,
    })
    if err != nil {
        return nil, nil, fmt.Errorf("create consumer: %w", err)
    }

    return stream, consumer, nil
}

Publishing Events with Deduplication

package publisher

import (
    "context"
    "encoding/json"
    "fmt"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

type DocumentEvent struct {
    DocumentID  string `json:"document_id"`
    Action      string `json:"action"`
    ContentHash string `json:"content_hash"`
    Timestamp   int64  `json:"timestamp"`
}

type EventPublisher struct {
    js jetstream.JetStream
}

func NewEventPublisher(nc *nats.Conn) (*EventPublisher, error) {
    js, err := jetstream.New(nc)
    if err != nil {
        return nil, err
    }
    return &EventPublisher{js: js}, nil
}

func (p *EventPublisher) PublishDocumentEvent(ctx context.Context, evt DocumentEvent) error {
    data, err := json.Marshal(evt)
    if err != nil {
        return fmt.Errorf("marshal: %w", err)
    }

    subject := fmt.Sprintf("events.documents.%s", evt.Action)

    // Nats-Msg-Id enables deduplication within the stream's
    // dedup window (default 2 minutes)
    msgID := fmt.Sprintf("%s-%s-%d", evt.DocumentID, evt.Action, evt.Timestamp)

    ack, err := p.js.Publish(ctx, subject, data,
        jetstream.WithMsgID(msgID),
    )
    if err != nil {
        return fmt.Errorf("publish: %w", err)
    }

    if ack.Duplicate {
        // Message was already published (dedup hit)
        return nil
    }
    return nil
}

// ProcessEvents pulls and processes events in batches
func ProcessEvents(consumer jetstream.Consumer, handler func([]byte) error) error {
    ctx := context.Background()

    for {
        // Pull a batch of up to 10 messages, wait up to 5s
        msgs, err := consumer.Fetch(10,
            jetstream.FetchMaxWait(5*time.Second),
        )
        if err != nil {
            return fmt.Errorf("fetch: %w", err)
        }

        for msg := range msgs.Messages() {
            if err := handler(msg.Data()); err != nil {
                // Processing failed; nak for redelivery
                msg.Nak()
                continue
            }
            msg.Ack()
        }

        if msgs.Error() != nil {
            // Timeout or other fetch error; retry
            continue
        }
    }
}

Common Pitfalls

• Not handling redeliveries: At-least-once delivery means consumers will see duplicates during failures and rebalances. Always implement idempotent processing or deduplication at the consumer level.
• Unbounded ack wait: If the ack timeout is too long, failed messages sit in limbo instead of being redelivered. If too short, slow but successful processing is interrupted. Tune AckWait based on your actual processing time distribution with headroom for p99 latency.
• Ignoring consumer lag: A consumer that falls behind can exhaust stream storage or miss the retention window. Monitor consumer lag (the gap between the stream’s last sequence and the consumer’s ack’d sequence) and alert when it exceeds a threshold.
• Publishing without backpressure: When the NATS server or JetStream stream is overloaded, publish calls will fail or time out. Implement retry with exponential backoff on the publisher side, and consider buffering events locally during outages.
• Overloading a single stream: Putting all events into one stream with a wildcard subject makes it hard to tune retention and performance per event type. Use separate streams for event categories with distinct retention and throughput requirements.