One worker is not keeping up. Consumer concurrency in NATS

Part 2 of a series. Part 1 was about delivery guarantees: why Core NATS loses messages silently and what JetStream charges for not losing them.

Versions used in this text: nats-server 2.11.17, nats-py 2.15.0, nats CLI 0.4.0. Every number and every console output in this article comes from running the included code. Measured 2026-09-22 on one laptop, single-node server, no network between the client and the broker.

Table of contents

1. The second worker

Part 1 ended with a worker that doesn’t lose orders. In production the next question arrives about a week later: the queue grows faster than one process empties it. The obvious move is to start a second worker, and on the surface NATS makes that trivial — you run the same process twice.

What follows is the part nobody warns you about. The second worker doesn’t just add throughput; it changes what your code is allowed to assume. Two messages that used to be handled one after another can now be handled at the same time, in either order, by different processes. Everything you did in part 1 to survive duplicates does not help here, because this is a different problem wearing similar clothes.

This article is about that problem: what concurrency between consumers actually is, which parts of it NATS gives you, which parts stay your job, and what it costs when you need order back.

If you haven’t read part 1, three terms are enough to follow along. A stream is storage on the server that captures messages published to matching subjects. A consumer is a server-side bookmark into that storage — it remembers how far it has got and which messages it has handed out without getting an acknowledgement back. ack() is how your worker tells the server a message is done. The demos here use the same docker compose lab as part 1, plus one more stream:

docker compose exec nats-box nats stream add PART2 \
  --subjects="part2.>" --storage=file --max-age=1h --defaults

2. What concurrency actually is

The word gets used for several different arrangements, and the differences matter because they fail differently. Rather than start from NATS features, start from the property you want.

Consumer concurrency holds when all four of these are true:

1. One shared source of work. All workers draw from the same logical queue.
2. One task, one worker at a time. Something arbitrates, so two workers don’t pick up the same task.
3. More than one task in flight at once. Otherwise you have a standby setup, not concurrency.
4. Tasks are independent, or their dependencies are enforced. Nothing in the broker checks this one. It lives in your domain.

Each condition has a test that settles it, and the tests are more useful than the definitions.

Condition 1: is it one pool, or is everyone getting a copy?

The test: take a worker away. Do the others get more work, or the same amount? If the load per worker doesn’t change, you don’t have a shared pool — you have fan-out, where every worker receives every message.

In JetStream the difference is one string. Workers that pass the same durable name attach to the same server-side consumer and share one bookmark. Workers that each pass their own name get their own consumer, their own bookmark, and their own copy of everything.

if mode == "shared":
    durables = {"worker-a": "shared", "worker-b": "shared"}
else:
    durables = {"worker-a": "fanout-a", "worker-b": "fanout-b"}demo/06_fanout_vs_shared.py

Twenty messages published, two workers, the only difference being the consumer name:

mode=shared  published=20  processed in total=20
  worker-a: 10
  worker-b: 10

mode=fanout  published=20  processed in total=40
  worker-a: 20
  worker-b: 20

Forty units of work out of twenty messages. This is the failure mode behind a line like durable=f"worker-{os.getpid()}", which looks reasonable — every process gets a unique name, no collisions — and quietly turns your task queue into a broadcast. Every order confirmation gets sent twice, and adding workers makes it worse rather than better.

Condition 2: who stops two workers taking the same task?

In JetStream the server does. While a message is handed out and unacknowledged, it is not handed to anyone else.

But part 1 already showed the crack in this: the arbiter is a stopwatch, not a liveness check. After ack_wait elapses the server re-delivers, and it cannot tell a dead worker from a slow one. A worker that is merely slow keeps working while a second worker starts the same task. Condition 2 is therefore best-effort, not a guarantee, and the remedies are the ones from part 1 — set ack_wait above your realistic worst case, call msg.in_progress() during long work, and keep the operation idempotent as the net.

Core NATS queue groups arbitrate too, and differently. Both mechanisms get called “load balancing”, which is where the confusion starts:

A queue group satisfies conditions 1, 2 and 3 perfectly well. What it doesn’t do is survive failure. That is worth stating plainly, because it is the cleanest illustration of the point this series keeps making: concurrency and delivery guarantees are separate axes. You can have concurrent workers with no guarantee at all, and a single worker with a full guarantee.

Condition 3: in flight, or just fetched?

This is the one that fools people, because the server counters look identical either way.

msgs = await sub.fetch(batch=100, timeout=5)
for msg in msgs:          # one after another
    await handle(msg)

That fetches a hundred messages, and the server will report a hundred in flight. But the loop runs them sequentially — the hundred-and-first message waits for all hundred. Batching is about how many messages travel per round trip. Concurrency is about how many are being worked on at the same time. To get the second one you have to say so:

msgs = await sub.fetch(batch=100, timeout=5)
await asyncio.gather(*(handle(msg) for msg in msgs))

Which brings us to the question of what gather actually buys you in Python.

3. Two kinds of parallelism, and only one of them is real

nats-py is asyncio-only, so every NATS example you write is inside an event loop. That makes it easy to conflate two very different ways of doing more at once:

• N processes (containers, pods) all attached to the same consumer — this is horizontal scaling, and the work genuinely runs on several cores;
• N asyncio tasks inside one process — this is concurrency within one thread, and whether it buys you anything depends entirely on what your handler does.

The second one has a hard limit that catches Python developers out. An asyncio task only lets others run when it awaits something. A handler that waits on a database or an HTTP call yields, and the other tasks proceed while it waits. A handler that computes — parses, compresses, renders, hashes — never yields, and the GIL allows exactly one thread to execute Python bytecode at a time. A hundred such tasks in asyncio.gather run one after another.

The same worker, the same hundred messages, the same gather, the same ~20 ms of work each. Only the nature of the work differs:

async def handle(msg, kind: str) -> None:
    if kind == "io":
        await asyncio.sleep(0.02)   # a database write, an HTTP call
    else:
        cpu_work()                  # a busy loop of the same duration
    await msg.ack()demo/09_gil.py

work=io   one unit= 20.2 ms  100 messages concurrently in  0.03 s  (sequential would be  2.02 s)
work=cpu  one unit= 28.7 ms  100 messages concurrently in  2.89 s  (sequential would be  2.87 s)

I/O-bound work: a hundred messages in 30 milliseconds instead of two seconds. CPU-bound work: 2.89 seconds against a sequential 2.87 — the concurrency bought nothing at all, and the code looks identical in both cases.

The practical rule: gather scales waiting, processes scale computing. If your handler is CPU-bound, more asyncio tasks will not help and you need more processes — which in NATS costs nothing beyond starting them, because they simply pass the same durable name. There is no rebalance step, no partition reassignment, no pause while the group reorganises. (This is where NATS differs from Kafka, if you know it: Kafka distributes partitions among clients, so membership changes cost a rebalance. NATS keeps all of that on the server, so a worker joining is just another process pulling.)

4. The in-flight limit: max_ack_pending

Every broker has a knob for how much work may be outstanding at once. In NATS it is max_ack_pending, defaulting to 1000. It is the same idea as visibility-limited in-flight messages in Amazon SQS and as consumer prefetch in RabbitMQ — one knob, three names. If you know any of them, this is that; if you don’t, here is what it does.

A consumer will not hand out more than max_ack_pending messages that are waiting for an acknowledgement. When the limit is reached, the server stops delivering until acks come back. It is backpressure: it stops a fast producer from burying a slow worker under work it has already claimed but cannot finish.

It is one pool, not a limit per worker

This is the trap, and it is worth a measurement rather than a quotation. The documentation says the limit “applies across all of the consumer’s bound subscriptions”, which is easy to read past. Concretely: a consumer with max_ack_pending=5 and two subscriptions attached to it.

subscription 1: asked for 5, got 5 (not acked)
subscription 2: asked for 5, got 0 (not acked)
total handed out: 5   max_ack_pending=5   server num_ack_pending=5

The second subscription got nothing. Five processes sharing a durable do not get 1000 in-flight messages each — they share 1000. If you scale from two workers to twenty and throughput doesn’t move, this is the first thing to check: you may have been at the ceiling since worker three.

What it costs to turn it down

The measurement part 1 promised. One variable changes: max_ack_pending. Five hundred messages, a handler that sleeps 20 ms, the fetch batch matched to the limit (more on that in a moment), everything processed with asyncio.gather:

Two things to read out of this table.

Setting it to 1 costs a factor of sixty against the same code at 100 (47 against 2874 msg/s). That is what strict serialisation through a single consumer buys and what it charges. If you reached for max_ack_pending=1 to get ordering, this is the bill — and section 5 is about paying less of it.

Above 100 the curve flattens, and not because of the server. The fetch batch was capped at 100, so from that point the ceiling was the client: no more than a hundred messages were ever in one gather. Raising a server-side limit past what your worker actually asks for changes nothing. The knob that matters is whichever of the two is smaller.

Asking for more than the server may give

That interaction has a sharp edge. fetch(batch=100) against a consumer with max_ack_pending=1: the server hands out one message, then stops, because nothing may be in flight beside it. The client waits for the other ninety-nine until the fetch times out.

fetch: asked for 100, got 1 after 2.00 s
fetch: asked for 100, got 1 after 2.00 s
fetch: asked for 100, got 1 after 2.00 s
fetch: asked for 100, got 1 after 2.00 s
fetch: asked for 100, got 1 after 2.00 s
processed=5 in 10.01 s -> 0.5 msg/s

Half a message per second, with a two-second fetch timeout. Every message costs a full timeout, so the damage scales with how generous your timeout is — and a “safe” thirty-second timeout would make this roughly 0.03 msg/s. Nothing in the logs says what is wrong; the worker looks like it is patiently waiting for work that isn’t there, while the stream fills up. Keep the fetch batch at or below max_ack_pending.

5. Ordering against concurrency

Here is the trade-off the whole series builds towards, and the part the documentation covers worst.

A single consumer with one worker handling one message at a time processes a stream in order. Add concurrency and that stops being true — not because of a bug, but by definition: several messages are being worked on at once, and they finish when they finish. Part 1 already showed the weaker version of this, where a single redelivery is enough to reorder a stream. Concurrency makes it routine.

Two customers, four events each, published strictly interleaved and in order. The first event of each customer takes 50 ms, the rest take 10 ms. One consumer, the whole batch through asyncio.gather:

published in order: alice-1, bob-1, alice-2, bob-2, alice-3, bob-3, alice-4, bob-4
mode=concurrent
      12 ms  finished alice-2
      12 ms  finished bob-2
      12 ms  finished alice-3
      12 ms  finished bob-3
      12 ms  finished alice-4
      12 ms  finished bob-4
      52 ms  finished alice-1
      53 ms  finished bob-1
alice: alice-2 -> alice-3 -> alice-4 -> alice-1   OUT OF ORDER
bob: bob-2 -> bob-3 -> bob-4 -> bob-1   OUT OF ORDER

Every customer’s first event finished last. If those events were “customer created” followed by “address updated”, you just updated an address on a customer that doesn’t exist yet.

Idempotency is not commutativity

This is where the habits from part 1 stop protecting you, and it deserves naming precisely.

Idempotency means doing the same operation twice has the same effect as doing it once. That is what at-least-once delivery demands, and part 1 was about it.

Commutativity means the order of two different operations doesn’t matter. That is what concurrency demands, and no amount of idempotency provides it.

Someone who learned idempotency in part 1 feels safe. Adding a second worker is what proves otherwise.

Partial order: partitioning

The usual reflex is to serialise everything — max_ack_pending=1 — which the table in section 4 prices at a sixtyfold slowdown. But you almost never need global order. You need order within a customer, an account, a device. Events belonging to different customers can happily overlap.

The standard solution is to split the work into lanes such that everything with the same key lands in the same lane, and run the lanes in parallel with each lane sequential inside.

NATS has no partition key as a first-class concept — nothing you attach to a message that the broker uses to pin related messages together. What it has is the subject, and a subject is a dotted path you choose. So the key goes into the subject, and each consumer filters on its own slice:

await js.publish(f"part2.orders.{customer}.created", payload)

# one consumer per customer, each strictly sequential
await asyncio.gather(*(
    drain_consumer(js, f"order-{c}", f"part2.orders.{c}.>", max_ack_pending=1, ...)
    for c in CUSTOMERS
))demo/10_ordering.py

mode=partitioned
      52 ms  finished alice-1
      52 ms  finished bob-1
      63 ms  finished bob-2
      63 ms  finished alice-2
      74 ms  finished alice-3
      74 ms  finished bob-3
      86 ms  finished bob-4
      86 ms  finished alice-4
alice: alice-1 -> alice-2 -> alice-3 -> alice-4   in order
bob: bob-1 -> bob-2 -> bob-3 -> bob-4   in order

Both customers are being processed the whole time — look at the timestamps, alice and bob advance together — while each customer’s events stay in sequence. The unordered run finished in 53 ms and the partitioned one in 86 ms: correctness cost about 60% here, against the 6000% that global serialisation costs.

One consumer per customer obviously doesn’t scale to a million customers. The production shape is a fixed number of lanes with the key hashed into one of them — part2.orders.7.created for lane 7 — either computed by the publisher, or by the server through subject mapping, where {{partition(3, 2)}} derives the lane from a token of the incoming subject. The second option is a server configuration change rather than something an application sets up, which is worth knowing before you design around it.

The real work here isn’t the code. It is answering, for your domain: can two messages about the same thing be processed in either order? Nobody can answer that for you, and neither the broker nor the client library will notice when the answer is no.

6. What NATS gives you and what stays yours

The four conditions again, now with the split that matters when you are the one on call. Three of the four entries in the right-hand column are outside anything the NATS documentation talks about.

Worth noting what concurrency does not require, so you don’t build more than you need: not acknowledgements, not JetStream, not even multiple processes. A Core NATS queue group with gather in one process satisfies all four conditions. It just loses the work when a process dies — which is a delivery guarantee question, and that was part 1.

7. The operational layer

Three things that aren’t part of the definition but will happen to you in production.

Graceful shutdown. On SIGTERM: stop fetching, finish the messages in hand, acknowledge them, then exit. Without this, every deployment leaves a batch of messages in flight with no one to ack them; they all come back after ack_wait and get processed twice. A rolling deploy across ten pods turns into a wave of redeliveries — and it will look like a broker problem rather than a shutdown problem.

There is no dead letter queue. Covered in part 1 and it bites harder with concurrency, because a poison message that fails in every worker burns through all of them in turn. After max_deliver the server simply stops delivering, with no notification. msg.term() lets a worker say “never again” deliberately; anything automatic you have to build from the server’s advisory events.

Read the counters, don’t guess. ConsumerInfo tells you which of the four conditions is actually holding: num_pending (waiting for delivery), num_ack_pending (in flight), num_redelivered (currently being retried). A num_ack_pending that sits at 1 while your queue grows means your workers are taking turns — condition 3 is not holding, whatever your architecture diagram says.

8. What this adds up to

Scaling consumers in NATS is genuinely easy, and it is worth understanding why: everything that needs coordination was moved to the server. The bookmark, the in-flight set, the timers all live in the consumer, so the things you scale — your worker processes — are stateless. Adding one is starting a process with the same durable name. There is no rebalance, no leader election, no lock service.

That also draws the line around what NATS will not do for you. It will not tell you that your durable name has a process ID in it and you are fanning out. It will not notice that your handler is CPU-bound and your gather is decorative. It will not know that two events about the same customer must not overtake each other. Those are all yours, and they are the ones that produce incidents.

The compressed version, if you keep one paragraph from this article: give every worker the same durable name, keep the fetch batch at or below max_ack_pending, use gather only for work that waits and processes for work that computes, and decide — before you scale out — which messages are allowed to overtake which. The broker enforces the first three. The last one is on you.