[EricAndrechek]

Part of #613. Stacked on #624 (feat/mq-nats-topology); merge into it, not into main.

This fixes four problems with shard ownership under mq.backend: nats. Each one broke the promise that a table has one writer and keeps its order under load, across handovers and across rolling deploys.

What changes

1. A busy owner keeps its pin

The pull consumer ran the worker's handler on its delivery goroutine and only pulled again once the handler returned. The server renews a pin only when it gets a pull. So a process that sat at its held-rows cap for longer than the pinned TTL (10s) lost its pins. wavehouse_ingest_shards_unowned then counted its shards, and a peer that took the process for dead reset them.

• Each unit now has two goroutines (internal/mq/external_consumer.go, moved out of external.go):
  • A puller that never runs the handler. It fetches in 1s pulls, sized to what the unit's cap leaves room for.
  • A deliverer that hands rows to the handler in order.
• At the cap, the puller renews the pin every 5s with a one-row fetch, up to one ack_wait of such rows past the cap (12 at the generated 1-minute ack_wait).
• Past that, and once halted, it renews with a pull of max_bytes: 1. The server holds the row back, so that pull delivers nothing.
• The one-row fetch at the cap is there for order. A held-back redelivery goes to the back of the server's redelivery queue: with max_bytes-1 renewals at the cap, rows NAKed as 0, 1, 2, 3 came back as 2, 3, 0, 1 (measured).
• ResetOrphaned and Unowned now also require that the durable delivered nothing and had no ack for its pinned TTL. A missing pin alone no longer means the owner is dead.

2. Each shard gets its own budget

The process-wide cap of 10,000 held rows let one stuck table take every slot, which stalled every shard the process owned.

• Each shard now holds an even share of the budget over the shards its process holds, and never less than 1,000 (two batches).
• The share goes through ConsumerConfig.MaxHeld and is read before every fetch.
• wavehouse_ingest_rows_held_waits_total is removed, because nothing waits at the cap any more.

3. A clean stop and a handover keep order

Before:

• A stop cancelled the worker while its shards kept pulling. Rows delivered during the final flush were dropped, and they only came back after ack_wait, behind newer rows.
• A handover released its shard after at most 15 seconds, even if its rows had not been written yet.

Now:

• A clean stop halts every shard first (mq.Halter) and keeps the pins. The worker writes what it holds plus whatever the shards still deliver. Only then does the stop release the shards.
• A handover keeps the pin until its rows are written, bounded by the worker's 60s ack wait.
• A row the worker drops while stopping is NAKed before the release.

4. A deleted durable ends the worker

• A durable deleted while no pull of it was waiting stalled silently until the five-minute topology check. Two pulls in a row with no responders, or one renewal with none, now look the durable up. If the durable or its stream is gone, delivery ends through failed.
• A bind whose durable or stream is gone, or whose durable no longer fits (mq.ErrConsumerMismatch), now ends the worker.
• A bind failure that lasts a minute now logs an error.

The release check-then-unpin race is documented, not changed:

• The unpin names no pin id, so a check that the pin is still ours comes first.
• A halted unit renews its pin until the unpin returns, so the pin cannot lapse between the check and the unpin.
• Only a consumer leader change in that window could move the pin.

Measurements

These were run against an in-process single-node nats-server v2.14.6.

• Blocked owner. Its insert hung for 13s, longer than the 10s pinned TTL. It kept one pin id throughout, took 2 rows past its share, and a peer's ResetOrphaned answered ErrUnitHeld.
• Throughput of one unit. 160k to 192k rows/s, against 158k to 172k before, with 20,000 rows of 256 bytes, run three times.
• Handover and clean stop under continuous publishing. 16 tables, with the first process's ClickHouse taking 6s per insert. Every table's rows arrived exactly once and in order. The next owner wrote its first rows about 2s after the stop. With the halt removed that was 9s, while the dropped rows were waited out.
• Deleted durable. failed fired 5s after the delete, with the handler busy.

Tests

Each new test was checked by removing its fix and confirming the test then fails:

• TestShardClaims_BlockedOwnerKeepsItsShard
• TestShardClaims_StuckShardDoesNotStallTheOthers
• TestShardClaims_TableOrderAcrossHandoverAndStop
• TestExternalNATS_DurableDeletedWhileBusyEndsDelivery
• TestExternalNATS_RedeliveryOrderKeptAtTheCap
• TestExternalNATS_BlockedHandlerTakesABoundedOvershoot
• TestExternalNATS_BlockedHandlerOvershootBoundedByTheShare
• TestExternalNATS_HaltKeepsThePinPastASlowDrain
• TestIngestWorker_StopHaltsBeforeTheFlushAndStopsAfter
• TestIngestWorker_StopNaksWhatArrivesAfterItStopsReading
• TestClaims_UnitShareOfMaxHeld
• the fatal-bind cases in TestClaims_MissingDurableEndsDelivery

make ci passes locally.

Left for later

• Rows redelivered across a handover or stop. They come back ahead of newer rows, but not always in order among themselves. While the next owner's pull waits for the pin, the server requeues a due redelivery behind the others. The normal handover and stop hand nothing back, so this only affects rows already NAKed or past ack_wait. It is documented.
• Cost of the change. An idle shard now costs one pull per second (1s fetches bound how long a halt takes). A stuck shard can hold up to 12 rows past its share.
• Deploy and tooling fixes to the manifests, the permissions check and the mq CLI are in a separate PR.
• The other follow-ups found in validation: mixed-version rollouts, the history's edge cases, and the storage-limit status codes.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FyrXjhR7iDg33paioLQHFq

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Advanced

Run ID: b45c4644-48bb-4f43-8421-4973ee04f1db

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

• Autopilot · Keep fixing CodeRabbit findings and required CI, and resolving merge conflicts

Autopilot is currently an internal CodeRabbit preview.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}