EricAndrechek · Sep 30, 2026 · Sep 30, 2026 · Sep 30, 2026 · Sep 30, 2026 · Sep 30, 2026
diff --git a/.testcoverage.yml b/.testcoverage.yml
@@ -112,7 +112,7 @@ exclude:
       - ^internal/mq/nats_topology\.go$
       - ^internal/mq/nats_manifests\.go$
       - ^internal/mq/subject_nats\.go$
-      - ^internal/mq/external\.go$
+      - ^internal/mq/external(_consumer)?\.go$
       - ^internal/mq/lease\.go$
       - ^cmd/wavehouse/mq\.go$
       # Their wiring, apart from wire.go for this: the NATS queue and lease
@@ -133,6 +133,6 @@ exclude:
       # The external NATS broker's tests start a server per case, which the
       # unit suite's 15s per package cannot hold: they are integration-tagged
       # (make test-integration), and the merged total counts them.
-      - ^internal/mq/external\.go$
+      - ^internal/mq/external(_consumer)?\.go$
       # The NATS KV leases, the same way.
       - ^internal/mq/lease\.go$
diff --git a/AGENTS.md b/AGENTS.md
diff --git a/CHANGELOG.md b/CHANGELOG.md
diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md
diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md
diff --git a/docs/src/content/docs/ingest-pipeline.md b/docs/src/content/docs/ingest-pipeline.md
@@ -17,7 +17,7 @@ It is deliberately detailed: this is a hot, concurrency-heavy path, and the goro
 | `backoff.go` | The retry backoff per ClickHouse pool: one outage backs off every table on it together, probing once per window; a failure of one table (read-only, too many parts or mutations, a missing grant) backs off that table alone — see [When ClickHouse cannot take an insert](#when-clickhouse-cannot-take-an-insert) |
 | `compact.go` | `EncodeCompactRow` — renders one record as a `JSONCompactEachRow` line over the table's **insertable** columns, in declaration order. Serialization only: it validates nothing and judges no value |
 | `sweeper.go` | The **Active Sweeper** — every minute, asks the MQ to purge the events that are both written to ClickHouse and past the SSE gap window (the purge arithmetic below lives in `internal/mq/purge.go`) |
-| `claims.go` | `ClaimShards` — over the external broker's shards, narrows the worker to this process's share: the membership lease, the per-tick assignment, handover by drain then release, the reset of a dead owner's shards at takeover, the `MaxHeld` cap on rows held, and the shard gauges. The behavior is in [Scaling ingest processes](/deployment#scaling-ingest-processes). |
+| `claims.go` | `ClaimShards` — over the external broker's shards, narrows the worker to this process's share: the membership lease, the per-tick assignment, handover by drain then release, the reset of a dead owner's shards at takeover, each shard's share of the `MaxHeld` budget of rows held, the halt that lets the worker write what it holds before a clean stop releases its shards, and the shard gauges. The behavior is in [Scaling ingest processes](/deployment#scaling-ingest-processes). |
 | `assign.go` | `assignUnits` — rendezvous hashing capped at ⌈units/slots⌉, the same owners in every process for the same view |
 | `types.go` | `EventMessage` wire format and the `BufferConsumerName` constant |
 
@@ -238,7 +238,7 @@ Several layers throttle the pipeline, inner to outer:
 2. **`msgChan`** (cap `maxBatch*2`) — when full, the consume callback blocks and delivery pauses.
 3. **`pullMaxMessages`** — nats.go's client-side prefetch buffer in front of `msgChan`, shared by the tenants' streams (at least one message each).
 4. **`maxAckPending`** — the server suspends a tenant's delivery once this many of its messages are delivered-but-unacked; no other tenant's delivery waits on it. The outermost in-memory bound, and a per-tenant one: while ClickHouse stalls, the worker can hold up to `maxAckPending` rows for every tenant served.
-5. **`MaxBytes` + `DiscardNew`** on each tenant's stream (its `mq.max_bytes_gb` in the [settings directory](/settings-directory#message-queue), resized in place on reload) — when it fills (e.g. ClickHouse is down so nothing acks/purges), that tenant's new publishes are rejected and the API returns 503. Steps 4 and 5 are the embedded broker's; under `mq.backend: nats` a shard durable's `max_ack_pending` pauses every table in that shard, a process holds at most 10,000 rows across its shards, and a full partition refuses every tenant in it (see [Limits that differ from the embedded queue](/deployment#limits-that-differ-from-the-embedded-queue)).
+5. **`MaxBytes` + `DiscardNew`** on each tenant's stream (its `mq.max_bytes_gb` in the [settings directory](/settings-directory#message-queue), resized in place on reload) — when it fills (e.g. ClickHouse is down so nothing acks/purges), that tenant's new publishes are rejected and the API returns 503. Steps 4 and 5 are the embedded broker's; under `mq.backend: nats` a shard slows to what keeps its pin at its share of the 10,000 rows a process may hold unsettled (an even share over its shards, at least 1,000), which pauses every table in that shard, and a full partition refuses every tenant in it (see [Limits that differ from the embedded queue](/deployment#limits-that-differ-from-the-embedded-queue)).
 
 | Knob | Default | Meaning / invariant |
 | --- | --- | --- |
@@ -288,9 +288,9 @@ flowchart TD
     W --> D
 ```
 
-- **Work distribution.** Each partition is split into V shards, and a table's rows all go to one shard. Each shard has its own durable with the `pinned_client` priority policy, so the server delivers a shard to one pulling process at a time: a table's rows are batched and written by one process at a time. The ingest processes share the shards out: each holds a membership lease (`ingest.m<j>`), and every process computes the same owner for every shard from the live members, by rendezvous hashing capped at an even share (`ClaimShards`). A shard given up is drained, then its pin released, so the next owner receives at once; a shard whose owner died is reset to its ack floor before its new owner pulls it, so the rows the dead owner held come back at once rather than after `ack_wait`. Rows delivered and unsettled across a process's shards are capped at 10,000.
+- **Work distribution.** Each partition is split into V shards, and a table's rows all go to one shard. Each shard has its own durable with the `pinned_client` priority policy, so the server delivers a shard to one pulling process at a time: a table's rows are batched and written by one process at a time. The ingest processes share the shards out: each holds a membership lease (`ingest.m<j>`), and every process computes the same owner for every shard from the live members, by rendezvous hashing capped at an even share (`ClaimShards`). A shard given up stops fetching but keeps its pin until the rows it delivered are written, then releases it, so the next owner receives at once and only newer rows; a stopping process first halts every shard the same way (`mq.Halter`), writes what it holds, and only then releases them. A shard whose owner died is reset to its ack floor before its new owner pulls it, so the rows the dead owner held come back at once rather than after `ack_wait`; the broker refuses the reset while the shard is pinned or was active within its pinned TTL. The worker never blocks a shard's pulls: at the shard's share of the 10,000 rows a process may hold unsettled (an even share over its shards, at least 1,000), the shard fetches only what keeps its pin (a row every 5 seconds, at most one `ack_wait` of them past its share, then pulls that take none), so a stuck table neither loses its shard nor stalls the process's other shards.
 - **Idempotent inserts matter more.** At-least-once delivery plus redelivery after a crash means another process can re-insert a batch the dead one had written but not acked. Use `ReplacingMergeTree` (or a dedup key). A single process already re-inserts after a crash, or after an insert whose outcome it could not see ([When ClickHouse cannot take an insert](#when-clickhouse-cannot-take-an-insert)); several make it routine.
-- **NATS resilience.** The external broker reconnects on its own, with backoff; while it is disconnected a publish answers `503` with `Retry-After: 5`, and consumption resumes after the reconnect. A consumer whose delivery ends for good (one of its configured shard durables deleted, or the connection closed) ends the worker and the process, as the embedded one does.
+- **NATS resilience.** The external broker reconnects on its own, with backoff; while it is disconnected a publish answers `503` with `Retry-After: 5`, and consumption resumes after the reconnect. A consumer whose delivery ends for good (one of its configured shard durables deleted, or its stream, even while no pull of it is waiting, which the next pull finds; a durable that no longer fits the worker's config; or the connection closed) ends the worker and the process, as the embedded one does.
 - **The sweeper.** Work-queue retention deletes each row once it is acked, one row at a time, so one tenant's unwritten rows never hold back another's reclaim, which a shared ack floor would, and it keeps a row no consumer covers rather than dropping it. SSE replay reads the history stream, which the partitions republish to and which expires by `max_age`; it is best effort, so an outage of the history costs SSE the rows republished meanwhile and never holds up ingest. So there is nothing for the sweeper to purge.
 
 ## Deferred / not yet implemented