NATS JetStream’s persistence mechanism isn’t just about picking "disk" or "memory"; it’s fundamentally about how you trade off data durability and latency for cost and operational simplicity.
Let’s see JetStream in action with a simple producer and consumer, pushing messages to a stream that’s configured for file-based persistence.
Producer (Go):
package main
import (
"context"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
// Connect to NATS
nc, err := nats.Connect("nats://localhost:4222")
if err != nil {
log.Fatalf("Failed to connect to NATS: %v", err)
}
defer nc.Close()
// JetStream context
js, err := nc.JetStream()
if err != nil {
log.Fatalf("Failed to get JetStream context: %v", err)
}
streamName := "ORDERS"
subject := "orders.new"
// Ensure the stream exists with file-based persistence
// This is crucial for the demonstration. If the stream doesn't exist,
// you'd create it first.
// Example stream creation (run once before producer/consumer):
// _, err = js.AddStream(&nats.StreamConfig{
// Name: streamName,
// Subjects: []string{subject},
// Storage: nats.FileStorage, // Explicitly file storage
// })
// if err != nil && err != nats.ErrStreamNameAlreadyInUse {
// log.Fatalf("Failed to add stream: %v", err)
// }
log.Printf("Publishing messages to subject '%s' on stream '%s'", subject, streamName)
for i := 0; i < 10; i++ {
msgData := []byte("order_id:" + time.Now().Format("20060102150405.000") + ":" + string(i))
_, err := js.Publish(subject, msgData)
if err != nil {
log.Printf("Error publishing message %d: %v", i, err)
} else {
log.Printf("Published message %d: %s", i, msgData)
}
time.Sleep(500 * time.Millisecond)
}
log.Println("Finished publishing messages.")
}
Consumer (Go):
package main
import (
"context"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
// Connect to NATS
nc, err := nats.Connect("nats://localhost:4222")
if err != nil {
log.Fatalf("Failed to connect to NATS: %v", err)
}
defer nc.Close()
// JetStream context
js, err := nc.JetStream()
if err != nil {
log.Fatalf("Failed to get JetStream context: %v", err)
}
streamName := "ORDERS"
subject := "orders.new"
consumerName := "order_processor"
// Create a durable consumer if it doesn't exist.
// This consumer will track its own progress.
// Example consumer creation (run once before consumer):
// _, err = js.AddConsumer(streamName, &nats.ConsumerConfig{
// Durable: consumerName,
// AckPolicy: nats.AckExplicit, // We'll manually ack messages
// FilterSubject: subject,
// })
// if err != nil && err != nats.ErrConsumerNameAlreadyInUse {
// log.Fatalf("Failed to add consumer: %v", err)
// }
log.Printf("Starting consumer '%s' for stream '%s' on subject '%s'", consumerName, streamName, subject)
// Subscribe to the stream with the durable consumer
sub, err := js.PullSubscribe(subject, consumerName, nats.PullMaxMessages(10))
if err != nil {
log.Fatalf("Failed to subscribe: %v", err)
}
defer sub.Unsubscribe()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
for {
// Fetch messages in batches
msgs, err := sub.Fetch(10, nats.Context(ctx))
if err != nil {
if err == context.DeadlineExceeded {
log.Println("Context deadline exceeded, stopping consumer.")
break
}
log.Printf("Error fetching messages: %v", err)
time.Sleep(1 * time.Second) // Wait before retrying
continue
}
if len(msgs) == 0 {
log.Println("No messages fetched, waiting...")
time.Sleep(1 * time.Second)
continue
}
for _, msg := range msgs {
log.Printf("Received message: Subject='%s', Data='%s', Sequence='%d'", msg.Subject, string(msg.Data), msg.Sequence)
// Acknowledge the message
err := msg.Ack()
if err != nil {
log.Printf("Failed to acknowledge message %d: %v", msg.Sequence, err)
}
}
}
log.Println("Consumer finished.")
}
To run this, you’d first start a NATS server with JetStream enabled (nats-server -js). Then, you’d typically create the stream and consumer once using the nats CLI or the Go client’s AddStream/AddConsumer methods (as commented out in the code). Finally, run the producer and consumer programs. You’ll see messages being published and then consumed, with the consumer’s progress being remembered across restarts due to the durable nature of the JetStream stream and consumer.
The core of JetStream’s persistence is its acknowledgment mechanism tied to a stream. When a message is published, it’s written to the configured storage (either memory or file). A consumer then pulls this message. Crucially, the message isn’t removed from the stream until the consumer explicitly acknowledges it (msg.Ack()). This acknowledgment signals to JetStream that the message has been successfully processed, and only then is it eligible for deletion based on stream retention policies. This ensures that even if a consumer crashes mid-processing, the message remains in the stream, ready to be redelivered to another (or the same) consumer upon restart.
The fundamental choice in JetStream persistence boils down to nats.MemoryStorage versus nats.FileStorage.
Memory Storage (nats.MemoryStorage)
- What it is: Messages are stored entirely in RAM.
- Pros:
- Extreme Speed: Reads and writes are incredibly fast because they bypass disk I/O entirely. This is ideal for scenarios where microsecond latency for message persistence is critical.
- Simplicity: No disk management, no worrying about disk fills, permissions, or I/O bottlenecks.
- Cons:
- No Durability: If the NATS server process crashes or restarts (for any reason – hardware failure, OOM kill, manual restart), all messages in memory storage are lost. This is the biggest drawback and makes it unsuitable for applications requiring any level of data safety.
- Limited by RAM: The total storage capacity is directly limited by the available RAM on the NATS server machine.
- Not for long-term storage: Not designed for retaining messages for extended periods if durability is a concern.
File Storage (nats.FileStorage)
- What it is: Messages are written to files on the server’s local disk. JetStream manages these files, often using a log-structured merge-tree (LSM tree) or similar optimized approach for efficient writes and reads.
- Pros:
- Durability: Messages persist even if the NATS server restarts. This is the primary reason to choose file storage. Data is safe across reboots.
- Larger Capacity: Limited by disk space rather than RAM, allowing for much larger streams.
- Cons:
- Higher Latency: Disk I/O is inherently slower than RAM access, leading to higher message persistence latency. This can be a bottleneck for extremely high-throughput, low-latency applications.
- Disk Management: Requires monitoring disk space, managing file permissions, and potentially dealing with disk I/O performance.
- I/O Bottlenecks: Heavy write loads can saturate the disk, impacting overall server performance.
The Mental Model: Durability vs. Performance
Think of it as a spectrum. On one end, you have raw speed (memory), and on the other, you have safety (file).
- Memory: Use this when message loss is acceptable, or when the message is immediately processed and acknowledged, and the only goal is to buffer briefly before immediate consumption. Think of ephemeral event notifications where a missed one isn’t catastrophic.
- File: Use this when every message must be accounted for, and data loss is unacceptable. This is the default and recommended choice for most production use cases where messages represent financial transactions, user actions, or critical state changes.
JetStream achieves durability with file storage by writing messages to a persistent log on disk. When a consumer acknowledges a message, JetStream marks that message as "deleted" within its internal index for that stream. The actual file segments containing the deleted messages are then garbage collected based on stream retention policies (e.g., retaining messages for 24 hours, or up to a certain size). This ensures that even if the NATS server crashes, the log on disk can be replayed upon restart, and JetStream will know which messages were already acknowledged and which still need to be delivered.
When using nats.FileStorage, JetStream internally uses a mechanism to manage the lifecycle of data segments on disk. It writes messages sequentially into append-only log files. As messages are acknowledged and retention policies are met, JetStream will eventually compact these logs, removing acknowledged messages and potentially merging smaller files into larger ones to optimize disk usage and read performance. This process is largely transparent but means that disk space usage isn’t a simple sum of unacknowledged messages; JetStream actively manages the storage footprint.
The next step after understanding persistence is exploring NATS JetStream’s various acknowledgment policies and their impact on consumer behavior and message redelivery guarantees.