[EricAndrechek]

Summary

WaveHouse can now run on a NATS JetStream cluster the operator owns, instead of only on its embedded single-process broker. With mq.backend: nats and coord.backend: nats, any number of WaveHouse processes share one queue: several api,ingest replicas, or API and ingest processes deployed separately (with cache.backend: redis).

• The operator creates and owns every stream, durable and KV bucket. WaveHouse never creates, changes, purges or deletes one. It checks the topology at boot, refuses to start with every finding listed while it is wrong, and checks again every five minutes.
• wavehouse mq manifests prints the whole topology as nack resources, sized to fit the servers' file store, and wavehouse mq permissions prints the restricted wavehouse user's NATS permissions. deployments/nats/ ships both, for four partitions of eight shards, with a Helm values snippet for a three-node cluster whose volume the shipped streams fit.
• Ingest processes share the work out between them with no configuration beyond the process count. Handover is clean on a scale event, and a crashed process's rows come back in about a pin TTL rather than after ack_wait.
• A shard's owner keeps it while its ClickHouse is slow, one stuck shard does not stall the process's other shards, and a handover or clean stop keeps each table's rows in order.
• The deployment guide gains an External NATS section covering creating the topology, sizing the file store, durability, permissions, the limits that differ from the embedded queue, changing N and V, scaling, manual pre-production checks and monitoring.

Topology

The operator creates all of this, and wavehouse mq manifests generates it:

• N ingest partitions, streams holding <prefix>.ingest.<p>.>. A row's subject is <prefix>.ingest.<p>.<s>.<tenant>.<table>. The partition p and shard s come from independent jump consistent hashes of tenant and table, so changing N or V moves only about 1/N or 1/V of the tables, and a tenant's tables spread over the partitions. Partitions use work-queue retention: a row stays until the ingest worker acks it, even while no consumer covers it, and is deleted then. One tenant's unwritten rows never hold back another's reclaim. A partition has no age limit, and uses discard: new with a byte limit, plus a per-subject cap that refuses one table's flood before it fills the partition.
• V shard durables per partition, wh-ingest-<s>, each filtering its shard's subjects. They use the pinned_client priority policy in the group wavehouse, with a pinned TTL of at least 10 seconds, twice the 5 seconds between the pulls that keep the pin of a shard at its share of held rows, or stopping. The server delivers a shard to one pinned puller at a time. Names are not zero-padded, so a durable keeps its name when V grows.
• The history stream, <prefix>.hist.>, with no sources. Every partition republishes each row it stores to <prefix>.hist.<tenant>.<table>, which drops the partition and shard. SSE replay and every API process's live events read from it through short-lived consumers of their own, the only objects WaveHouse creates. The generated max_age is 15m, the seed's gap window. The history is best effort: a row republished while the history is unavailable reaches ClickHouse but not SSE, and a full or missing history never holds up ingest.
• One dead-letter stream, <prefix>.dlq.>, shared by every tenant.
• The lease bucket, <prefix>_coord, a KV bucket that holds the ingest processes' leases.

The wavehouse user can publish to the ingest and dead-letter subjects. It can read stream and consumer info, and pull from, unpin and reset each shard durable by name (a NATS wildcard matches a whole token only, so wh-ingest-* names nothing). It can ack only for the shard durables, under both ack subject layouts ($JS.ACK.*.<durable>.>, and $JS.ACK.*.*.*.<durable>.> for the js_ack_fc_v2 flag). It can read and write lease keys. It cannot create, change, purge or delete a stream or bucket, create a durable on a partition, ack for another consumer, or publish to the history.

• js_domain. Under a JetStream domain nats.go sends every API request and KV write under $JS.<domain>.API.. A server in the domain maps that to $JS.API. before checking permissions, while a leafnode outside it checks it as sent. wavehouse mq permissions --js-domain <d> allows and denies both forms, which covers the raw pulls that renew a busy or stopping shard's pin.
• Boot permission probe. Boot probes every shard durable as the connecting user with a pull and an unpin that the server rejects on their merits, so neither delivers a row nor moves a pin. A durable the user may not pull is a required finding, with the full wavehouse mq permissions command that regenerates the set. Permissions generated for fewer shards than the topology used to boot and leave the extra shards unpulled. A consumer API request refused after boot sets wavehouse_mq_topology_ok to 0 at once.

Sizing

Each server reserves every stream replica's max_bytes against its max_file_store, which the NATS Helm chart sets to the JetStream PVC size (100Gi shipped). wavehouse mq manifests --file-store (default 100Gi) sizes each partition at 15% of the store, the history at 10% and the dead-letter stream at 5%, so the shipped four partitions reserve 75 GiB. A partition's size does not follow N, so lowering N never needs more store while the removed partitions drain. The generator refuses streams that together exceed --file-store, and --partition-max-bytes resizes the partitions (from six partitions at the defaults). The test fixture now takes max_file_store from the shipped values the way the chart does.

wavehouse mq manifests also takes --ingest-consumer, --history-stream and --publish-timeout (the partitions' duplicate window follows it), matching the config keys.

Worker model

• Shards, pinned. Each shard durable is pulled by one process at a time, the one the server pinned, so a table's rows are batched and written by one process at a time.
• Membership-based balancing. Each ingest process holds a membership lease (lease.ingest.m<j>, up to 64 slots) in the lease bucket. Every process computes the same owner for every shard from the live members, by rendezvous hashing capped at an even share. When a process joins or leaves, roughly one to two times its share of the shards move while the rounded-up share holds, and never more than half of them. Durables left over after N or V is lowered are assigned apart from the configured ones and drained.
• Leases on NATS KV. The KV revision a term was taken at is its fencing token. A candidate takes another holder's lease only after seeing the same revision unchanged for 15 seconds on its own clock, so no two servers' clocks are compared and the bucket needs no TTL. The holder renews every 2 seconds and steps down after 10 seconds without a renewal. A clean stop deletes the key.
• A per-shard budget of held rows. Each shard the process holds gets an even share of 10,000 rows delivered and not yet acked, and never less than 1,000 (two batches), so a process holds about max(10,000, 1,000 × shards) at most. One stuck table fills only its own shard's share; the process's other shards keep flowing. A row stops counting at its first ack or nak, or after ack_wait. A per-shard max_ack_pending of 10,000/V was not used: at V=32 it is below the worker's batch of 500, so a busy table's batch could never fill.
• A busy owner keeps its pin. Each shard has a puller that never runs the handler and a deliverer that hands rows to it in order, so the delivery goroutine never blocks. At its share a shard renews its pin every 5 seconds with a one-row fetch (which keeps redeliveries in order), up to one ack_wait of such rows past its share (12 at the generated 1-minute ack_wait), then with pulls that deliver nothing. Unowned and the takeover reset also require that the durable delivered nothing and had no ack for the pinned TTL, so a missing pin alone no longer reads as a dead owner.
• Handover. A shard that changes owner stops fetching at the old owner. The old owner keeps the pin until the rows it delivered are acked or handed back, bounded by the worker's 60-second ack wait (the shortest ack_wait a durable may have), then releases it, and the new owner receives at once.
• Clean stop. The stop halts every shard first and keeps the pins. The worker flushes what it holds and whatever the shards still deliver, NAKs any row it could not take, and only then releases the shards and resigns membership. Each table's rows stay in order across a stop and a handover.
• A missing durable ends delivery. A durable deleted while no pull was waiting used to stall silently until the five-minute check. Now repeated no-responders, or one on a renewal, look the durable up, and a durable or stream that is gone ends delivery through failed, which ends the process so the next boot names what is missing. A bind whose durable is gone or no longer fits does the same.
• Crash. A dead process keeps its membership lease until the others have seen it unrenewed for 15 seconds. It keeps its pins until the pinned TTL after its last pull. Each new owner then resets the shard to its ack floor before pulling it, so the dead process's unacked rows come back at once instead of after ack_wait. Delivery is at least once. Measured:

Scenario	Result
Crash with 20 rows in flight (lease 3s, pinned TTL 10s, ack_wait 60s)	The survivor had all 20 rows 10.1s after the kill, bounded by the pin TTL, not ack_wait. No acked row was replayed.
kill -9 of one of the real binary's ingest processes (15s lease)	The survivors owned its shards 18s after the kill, and the next batch was written exactly once.
Scale from 2 to 3 ingest processes	The shards were split again 3s after the third process went live.
Clean stop	Every table was written again within 90ms.
Graceful unpin (TTL 30s)	The next puller's first row arrived 0s after the unpin.
Reset before pull	5 held rows were redelivered within 1ms, and no acked row was replayed.
Owner whose insert hangs 13s (pinned TTL 10s)	It kept one pin id throughout and took 2 rows past its share, and a peer's reset was refused with ErrUnitHeld.
Handover and clean stop under continuous publishing, 16 tables, old owner's inserts 6s each	Every table's rows arrived exactly once and in order. The next owner wrote about 2s after the stop.
Durable deleted with the handler busy	failed fired 5s after the delete.
One unit's throughput, 20,000 rows of 256 bytes, three runs	160k to 192k rows/s, against 158k to 172k before the pin-renewal change.

These were measured on a single nats-server 2.14.6, not a 3-replica cluster. The 3-node measurements are tracked in #694.

Configuration

• mq.backend: nats with the mq.nats.* block (WH_MQ_NATS_*) configures:

  • urls
  • exactly one credential form, all of them file paths: creds_file, nkey_seed_file, or user + password_file
  • tls.*, including mutual TLS
  • js_domain
  • subject_prefix (default wh)
  • partitions and shards (default 32, 1–256), which must match the topology's metadata
  • ingest_consumer (default wh-ingest) and history_stream (default <PREFIX>_HISTORY)
  • connect_timeout, publish_timeout and topology_wait (5s, 5s and 60s)

• coord.backend: nats with coord.nats.bucket (WH_COORD_NATS_BUCKET, default <prefix>_coord) configures the lease bucket. It rides the mq.nats connection.

• Boot rules. Boot refuses:

  • coord.backend: nats without mq.backend: nats
  • an ingest process on mq.backend: nats with coord.backend: local
  • api without ingest, or the reverse, over a local cache
  • a process whose only role is sweeper under nats, because there is no sweeper there and the streams' retention does its work
  • any role split over the embedded broker
  • a NATS server older than 2.14
  • a partition whose duplicate_window does not cover every publish attempt and dedupe.lease
  • persist_mode: async on a partition
  • any other required topology finding

  Recommended findings, such as fewer than 3 replicas, are logged and do not stop boot. A publish retried after a lost answer reuses its Nats-Msg-Id, so it is stored once.

• Durability comes from replicas. An ingest 200 follows the partition's ack, and at 3 replicas that ack comes once a Raft quorum has stored the event. WaveHouse does not require sync_always.

• Gauges: wavehouse_mq_connected, wavehouse_mq_topology_ok, wavehouse_mq_history_behind_seconds, and the wavehouse_ingest_shard* / wavehouse_ingest_rows_held family for ownership, unowned shards, held rows and handovers.

Compatibility

• Embedded is unchanged. mq.backend: embedded remains the default: one process, per-tenant streams, the fsync-per-event 200, and the Active Sweeper under its lease. A config that sets none of the new keys runs as before.
• NATS 2.14 or later is required for external mode. 2.14 is the first release with the consumer reset that shard takeover uses, and boot refuses an older server.
• Nothing in external mode has been released, so the CHANGELOG describes it as it ships, with no migration notes.

Manual pre-production checks

The automated tests run against a single nats-server. The deployment guide's "Before production: manual checks" asks the operator to check two things once on their own cluster:

1. nack reconciliation. Apply the generated manifests with your nack release, then read back a partition and a shard durable. They should show work-queue retention, the republish, pinned_client, the group and the pinned TTL. WaveHouse's boot check reports any field that did not arrive.
2. A 3-node run at --replicas 3.
   • Kill a shard durable's leader while a worker holds its pin. Check that the worker re-pins and loses no row.
   • Kill a partition's stream leader while events flow. Check how many rows the history misses.
   • Watch server CPU and memory under load with all N×V consumers.

Tests

• Multi-process roles test (tests/integration/roles_test.go). The real binary runs as two roles=api and two roles=ingest processes over an operator-provisioned NATS (4 partitions × 8 shards), a Redis cache and a dynamodb-local dedupe table. It checks:

  • The two ingest processes split the 32 shards evenly, with none unowned.
  • Ingest lands in ClickHouse exactly once.
  • Cache invalidation and SSE reach both API processes.
  • A duplicate id is accepted once.
  • A third ingest process takes its share by handover.
  • A SIGKILLed process's shards are taken over about one lease later, and every batch is still written once.

  A second test drives the boot rules through the binary.

• Conformance suite. mqtest, the one mq.Broker contract suite, passes against the external broker connected as the shipped restricted wavehouse user. That also proves the user's permissions are enough to publish, consume and check. coordtest.Conformance passes against the NATS KV leases.

• More tests:

  • shard-claim integration tests: crash with rows in flight, handover, extras, and the cap
  • unit tests for assignment and claims
  • broker tests for units, release, reset, unowned shards, the history and the ack-stream check
  • one verifier case per topology rule, all connecting as the restricted user
  • golden manifests and permissions
  • the shipped manifests fit the shipped file store, the generator refuses an over-reservation, and the fixture reserves against the shipped store
  • the permission probe at boot and under a JetStream domain, the probe leaving the durables alone, and a consumer request refused after boot as a topology fault
  • js_domain boot, consume, ack and lease on a server in the domain and through a leafnode, and a shard's pin renewal as the restricted user in both
  • acks under both ack subject layouts, and an ack outside the shards refused
  • a blocked owner keeping its shard, a stuck shard not stalling the others, per-table order across a handover and a stop under continuous publishing, the stop order (halt, flush, NAK, release), bounded overshoot at the cap, redelivery order at the cap, and a deleted durable ending delivery
  • JetStream behaviour pinned against nats-server 2.14.6 in internal/mq/natsspike: work-queue retention with republish, pinned clients, unpin, reset, and the whole-token permission wildcard

• make ci passes locally on the final head.

Follow-ups

• bug(mq): mixed-version N/V rollout can double-pull shards; version floor only sees the connected server #684 Mixed-version N/V rollouts, and the version floor reading only the connected server
• bug(mq): history replay can skip rows after a reset, or return early during a leader election #685 History replay after a reset, or during a leader election
• bug(mq): RESET can redeliver rows a live, draining process still holds #686 A reset redelivering rows a live, draining process still holds
• bug(mq): storage-limit errors (10002/10023) map to 500 instead of 503 + Retry-After #687 Storage-limit errors mapping to 500 instead of 503 with Retry-After
• bug(mq): closing while a Consume is in progress reports a spurious failure #688 Close racing a Consume in progress
• mq: durable/partition topology checks miss backoff, pause_until, max_waiting, subject_transform, sources #689 Topology checks for hand-written durables and partitions
• bug(mq): DuplicateWindow() returns the shortest window; retention warning needs the longest #690 DuplicateWindow() and the retention warning
• bug(config): nats mq defaults duplicate mq.Default* untested, no grammar check, sweeper role is a silent no-op #691 mq.nats defaults, name grammar, and the sweeper role under nats
• bug(mq): internal/mq integration tests are selected by name pattern, so a new prefix is silently skipped #692 internal/mq integration tests selected by name pattern
• mq: follow-ups from #624 — membership cost, Unowned batching, crash-restart wait, allocations, docs, per-tenant budgets #693 Membership read cost, Unowned batching, the crash-restart wait, allocations, the --coord-bucket docs pointer, per-tenant budgets
• mq(nats): pre-production measurements on a 3-node cluster #694 Pre-production measurements on a 3-node cluster
• Related: embedded mode on work queues with a history stream (mq: one unacked event pins the embedded ingest stream's purge for every table of its tenant #653), and spreading one hot table over several shards (ingest: a hot table is capped at one worker, and a dead worker's partitions wait out the lease #649).

Part of #613.
Fixes #264.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FyrXjhR7iDg33paioLQHFq