A multi-stage data pipeline has a basic reliability problem: each stage can fail independently. The ingester fetches a document from PubMed and publishes it. The processor picks it up, runs it through parsing, embedding, and COI matching, then publishes the enriched result. The indexer fans that out to Meilisearch, Milvus, and Neo4j. Any of these steps can fail — a transient network error, an OOM during embedding, a timeout on a slow Neo4j batch write. The question isn’t whether failures happen, but whether the pipeline loses work when they do.
NATS JetStream is a persistent messaging layer built on top of NATS that changes the answer. With core NATS, messages are fire-and-forget — if the subscriber isn’t listening when a message arrives, it’s gone. JetStream persists messages to disk and tracks consumer state, so a stage that crashes and restarts picks up where it left off. This tutorial walks through how EvidenceLens uses JetStream’s consumer mechanics to build a pipeline that handles failures, prevents redelivery storms, and backpressures ingest when downstream stages fall behind.
What You Need Coming In
- Familiarity with async Python (asyncio, coroutines, tasks)
- Basic pub/sub concepts — publishers emit messages on subjects, subscribers consume them
- No prior NATS experience required; the relevant concepts are explained as they come up
The Pipeline Topology
graph LR
A[Ingesters<br/>Go] -->|raw-docs.source| B[NATS JetStream<br/>EVIDENCELENS stream]
B -->|raw-docs.>| C[Processor<br/>Python]
C -->|indexable-docs.source| B
B -->|indexable-docs.>| D[Indexer<br/>Go]
D -->|batch writes| E[Meilisearch]
D -->|batch writes| F[Milvus]
D -->|batch writes| G[Neo4j]
D -->|failed docs| H[dlq.indexer]
One NATS stream named EVIDENCELENS carries three subject namespaces: raw-docs.> for ingester output, indexable-docs.> for processor output, and dlq.> for dead-letter routing. The > wildcard means each ingester and processor publishes to a source-specific subject (raw-docs.pubmed, raw-docs.fda, etc.) while consumers subscribe to the whole namespace at once.
Both the processor and indexer create the stream idempotently at startup, so whichever service starts first wins and the other just confirms the stream exists:
// index/cmd/indexer/main.go
_, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
Name: streamName,
Subjects: []string{"raw-docs.>", "indexable-docs.>", "dlq.>"},
Storage: jetstream.FileStorage,
Replicas: 1,
})FileStorage persists messages to disk. If the entire server restarts, unprocessed messages survive.
Durable Consumers and What They Track
A durable consumer is a named subscription that persists its state in JetStream — specifically, the sequence number of the last acknowledged message. When the consumer reconnects after a crash, it resumes from where it left off rather than starting at the beginning or missing messages that arrived while it was down.
The processor subscribes with a durable named “processor”:
# process/main.py
self._sub = await js.subscribe(
"raw-docs.>",
cb=self._on_message,
durable="processor",
manual_ack=True,
config=ConsumerConfig(
max_deliver=5,
ack_wait=ack_wait,
max_ack_pending=max_ack_pending,
),
)manual_ack=True means JetStream doesn’t consider a message delivered until the consumer explicitly calls await msg.ack(). If the subscriber dies mid-processing, the message hasn’t been acknowledged, so JetStream redelivers it to the next subscriber instance that comes up.
The Redelivery Storm
max_deliver, ack_wait, and max_ack_pending look like minor configuration details until you understand what happens when they’re misconfigured. The comments in the source are honest about the actual bug:
# ack_wait: how long NATS waits for an ack before redelivering.
# The old 60s was far too short under high concurrency — with N pipelines
# sharing one embedder, a doc routinely takes >60s to finish, so it got
# redelivered WHILE still processing, exploding the redelivered counter
# and pinning the ack floor (a redelivery storm, not real progress).The processor runs 50 concurrent pipelines. Each pipeline hits the BGE-M3 embedder, which batches requests and processes them at its own pace. Under load, all 50 pipelines queue behind the embedder. A document that enters processing might not complete embedding for 90-120 seconds.
The old ack_wait=60 meant JetStream assumed a document was lost if not acknowledged within 60 seconds. It redelivered the document. Now the processor has two copies of the same document in flight — the original still waiting for the embedder, and the redelivery also waiting for the embedder. This doubled the queue depth. JetStream redelivered again. The consumer’s redelivered counter grew exponentially while actual throughput went nowhere. Classic redelivery storm.
The fix:
ack_wait = int(os.getenv("PROCESSOR_ACK_WAIT_SEC", "300")) # 5 minutes300 seconds gives even heavily loaded pipelines time to complete before JetStream assumes the message is lost.
The max_ack_pending parameter has a subtler effect:
# max_ack_pending: cap on in-flight (unacked) messages — must be >=
# MAX_CONCURRENT_PIPELINES or the extra pipelines starve (default 1000
# was throttling a 200-wide processor to 1000 in flight regardless).
max_ack_pending = int(os.getenv("PROCESSOR_MAX_ACK_PENDING",
str(max(1000, self.cfg.max_concurrent_pipelines * 2))))JetStream stops delivering new messages when the count of unacknowledged messages hits max_ack_pending. If you have 200 concurrent pipelines and max_ack_pending=1000, the 200 pipelines can accept 1000 messages before JetStream pauses delivery. That’s fine. But the default of 1000 was below max_concurrent_pipelines * 2 for larger deployments, so some pipelines were starving while others completed and freed slots.
How Ack and Nak Route Messages
The message callback wraps the entire pipeline in a try/except:
async def _on_message(self, msg) -> None:
async with self._sem: # limits concurrency to max_concurrent_pipelines
try:
payload = json.loads(msg.data)
await self._process(payload)
await msg.ack()
except Exception as e:
log.error("processor.error", err=str(e))
await msg.nak()msg.ack() tells JetStream the message was processed successfully — advance the consumer’s position. msg.nak() signals failure — redeliver this message according to the consumer’s delivery policy.
max_deliver=5 caps redeliveries. After 5 failed attempts, JetStream stops redelivering the message. In EvidenceLens, permanently failed messages end up in dlq.indexer (the dead-letter subject), where they’re logged for audit. This prevents one malformed document from getting the processor stuck in an infinite redelivery loop.
Inside _process, some failures are treated as permanent and acked immediately rather than nacked:
# A missing object (e.g. the raw was purged) is unrecoverable —
# skip it (ack) rather than nak-ing into an infinite redelivery loop.
try:
raw = await asyncio.to_thread(self.store.get, object_key)
except Exception as e:
if "NoSuchKey" in str(e) or "404" in str(e):
log.warning("process.skip_missing_object", ...)
return # causes _on_message to ack
raiseIf the S3 object is gone — because the raw archive was purged or the key was wrong — no amount of redelivery will fix that. Acknowledging it removes it from the consumer’s backlog without consuming all 5 delivery attempts.
Backpressure: Pausing When the Indexer Falls Behind
The consumer semaphore limits concurrency (asyncio.Semaphore(cfg.max_concurrent_pipelines)), but it only prevents the processor from overwhelming itself. It doesn’t prevent the processor from overwhelming the indexer.
The _backpressure_loop task polls the indexer’s JetStream consumer info and pauses the processor’s subscription when the indexer’s pending message count exceeds a threshold:
async def _backpressure_loop(self) -> None:
js = self.nc.jetstream()
max_pending = int(self.cfg.max_concurrent_pipelines * 4)
paused = False
while True:
await asyncio.sleep(5)
try:
info = await js.consumer_info("EVIDENCELENS", "indexer")
pending = info.num_pending
if pending > max_pending and not paused:
log.warning("backpressure: pausing", indexer_pending=pending)
await self._sub.drain()
paused = True
elif pending <= max_pending // 2 and paused:
log.info("backpressure: resuming", indexer_pending=pending)
self._sub = await js.subscribe(
"raw-docs.>", cb=self._on_message,
durable="processor", manual_ack=True,
)
paused = False
except Exception as e:
log.debug("backpressure check skipped", err=str(e))Every 5 seconds, the loop reads how many messages are waiting in the indexer’s queue. If the indexer is more than max_concurrent_pipelines * 4 messages behind, the processor drains its subscription — it finishes whatever is currently in flight, then stops pulling new messages. When the indexer catches up to half the threshold, the processor subscribes again.
The drain/resubscribe approach is blunt but reliable. An alternative is flow control at the JetStream consumer level (the max_ack_pending cap acts as implicit backpressure), but explicit drain gives the processor control over when it resumes and lets the resume threshold differ from the pause threshold, preventing rapid oscillation between paused and active states.
Why JetStream Over Core NATS
Core NATS is a publish-subscribe message bus with no persistence. It’s fast and simple, but any subscriber that isn’t actively listening when a message is published misses it permanently. For an ETL pipeline where stages restart independently, that’s unacceptable.
JetStream adds a log-based persistence layer: messages are written to disk before being acknowledged by the server. Consumer state (which messages have been delivered and acknowledged) is also persisted. A stage that restarts picks up from its last acknowledged position — no messages lost, no need to replay from the beginning of the source.
The tradeoff is operational complexity: you’re now reasoning about streams, consumers, delivery policies, and ack semantics instead of just subjects and subscriptions. But for a pipeline where each stage might take seconds to process a single message and can fail for transient reasons, that complexity is worth it. The alternative — wrapping everything in try/retry loops with in-memory queues — loses work on process death and is harder to reason about than JetStream’s explicit delivery guarantees.
