feat(mq): a Broker over an external NATS cluster

[EricAndrechek]

Part of #613. This is PR D3 of the external-NATS workstream.

Stacking: the base is feat/mq-nats-topology (#624, D2). This branch also merges feat/mq-conformance (#623, D1), so the diff contains #623's commits too. Review it after #623 and #624. This PR's own change is everything after the merge commit aa9a6553.

What

mq.ExternalNATS (internal/mq/external.go) is an mq.Broker over an operator-owned NATS cluster: the topology in #624 (N interest-retention ingest partitions, a history stream that sources them, and one dead-letter stream). It never creates, changes, purges or deletes a stream or a durable. It creates only auto-expiring consumers on the history stream: one per Subscribe (the hub) and one per replay.

Nothing selects it yet. The config (mq.backend: nats, mq.nats.*), the wireMQ case, the integration test through app, and the user-facing docs are D4, on G1. NewNATS can be constructed and is tested. Only internal/mq imports NATS, as before.

• Boot: NewNATS connects and then runs feat(mq): external nats broker with sharded, pinned ingest workers #624's awaitNATSTopology until TopologyWait runs out. The wait also bounds each check's requests. Recommended findings are logged. If a required finding is still there when the wait ends, boot fails with a *TopologyError that lists every finding. A cluster never reached is ErrUnavailable (with passwords in the URLs redacted). Partition and dead-letter streams are then found by subject.
• Auth: user plus password file, nkey seed file, creds file, TLS, and mutual TLS. Secrets are file paths only. Conflicting auth, or half a key pair, is refused before any dial.
• Publish: goes to wh.ingest.<fnv32a(tenant)%N>.<tenant>.<table>[.<scope>] with WithExpectStream(partition), and each attempt is bounded by PublishTimeout.
  • A lost answer is retried at most twice with the same Nats-Msg-Id, so the partition stores it once.
  • maximum bytes exceeded and maximum messages per subject exceeded → ErrQueueFull.
  • No answer, a lost or closed connection, or a partition stream the operator deleted → ErrUnavailable. A deleted stream is also logged at Error with its name and drops wavehouse_mq_topology_ok at once.
• Worker: CreateConsumer maps buffer-consumer (or the operator's own name) to wh-ingest. It finds that durable on every partition, checks ack_wait ≥ cfg.AckWait and max_ack_pending > 0, and never creates it. Any other name is ErrConsumerNotFound. Consume runs one pull per partition and splits the prefetch between them. A delivery the client ends on its own (durable deleted, connection closed for good) is reported on failed exactly once, and never after stop (mq: propagate terminal JetStream consumer errors to the ingest worker #587).
• Hub: Subscribe uses a per-pod ordered, ack-less consumer on the history stream with DeliverNew. consumerName is ignored. A handler error is logged, because there is no redelivery to ask for.
• Replay: ReplaySince uses an ack-less consumer on the history stream, filtered to the topic, starting at since. It pulls in batches of 256 and counts down from the events the server counted at the start, so it never chases the live tail: the SSE handler has already subscribed live before gap-fill, so tail events would be duplicates. A pull or connection failure before it finishes is an error. The consumer is deleted afterwards, and its 1-minute inactive threshold outlasts sending one batch to a slow client.
• DLQ: DeadLetter publishes to wh.dlq.<topic key> with a fresh id. DeadLetterCounts is one subject-filtered info call on the shared stream. A tenant that has parked nothing gets zero counts.
• PurgeAcked: returns (false, nil) with no I/O. It warns once per tenant whose gap window is longer than the history's max_age. A consumer name that does not map is ErrConsumerNotFound.
• SetMaxBytes: records the budget and logs once that budgets are not enforced per tenant. Enforcing them is D6.
• Close: stops every consumer (so nothing reports failed) and the watch loop, then drains the connection.
• Re-check and gauges: the topology is verified again every 5 minutes and the history's sources are read every 30s. The gauges are wavehouse_mq_connected, wavehouse_mq_topology_ok, and per source wavehouse_mq_history_source_lag and wavehouse_mq_history_source_last_active_seconds. The source lag matters beyond SSE, because a stalled source holds rows on every partition (S1).

Additions and deviations from the design

• The source gauge is "seconds since last contact", not "attached" (measured). After a NATS restart the server does not report a source as detached (active: -1). Active keeps counting up from the last contact until the source re-attaches. In one probe, one partition's source re-attached at ~11s and the other three were still unattached at 14s. An "attached" gauge would have read 1 throughout. So a source re-attaching is not a topology fault, as S1 asked. The verifier's one transient finding ("not attached yet", active < 0, which is only seen before a source's first attach) is marked transient in feat(mq): external nats broker with sharded, pinned ingest workers #624's Finding and kept off topology_ok.
• The re-check skips while disconnected (measured). A re-check that ran while the connection was down read the server version as "", which gave a false required finding, "cannot read server version". The connection has its own gauge.
• Close cancels an in-flight re-check. Without this, a re-check started during an outage could hold Close for up to 2 × 30s.
• Config shape: NATSConfig embeds feat(mq): external nats broker with sharded, pinned ingest workers #624's NATSTopology instead of repeating its fields flat. D4 fills both halves from mq.nats.*.
• Test placement: the tests are //go:build integration in internal/mq itself. They need feat(mq): external nats broker with sharded, pinned ingest workers #624's fixture, which lives in package mq's test files, and the conformance run has to be in mq_test because mqtest imports mq. So external_export_test.go exposes the fixture to it. make test-integration runs these tests by name (^Test(ExternalNATS|NewNATS|NATSPermissions_Refuse)) in a second gotestsum call, so the package's untagged tests stay in the unit suite only. .testcoverage.yml excludes external.go from the unit and e2e per-suite gates. The merged total still counts it.
• The duplicate-window floor now covers every retry. feat(mq): external nats broker with sharded, pinned ingest workers #624's verifier asked for ≥ 2 × publish_timeout, but the last retry arrives 3·T + 2·250ms after the first attempt. minDuplicateWindow() is now shared by the verifier and the manifest generator. The shipped jetstream.yaml is unchanged (2m is more than 15.5s).
• testutil.StoreDir (refs fix(test): ingest dispatch-loop test races t.TempDir cleanup, reddens main #442): NewEmbeddedMQ now removes its store with retries, the same pattern mqtest uses. The fix(test): ingest dispatch-loop test races t.TempDir cleanup, reddens main #442 cleanup race failed internal/ingest in 4 of 5 make ci runs on this branch under heavy load (measured; 0/8 in isolation on both base and HEAD). This is a mitigation, not the drain-before-return fix fix(test): ingest dispatch-loop test races t.TempDir cleanup, reddens main #442 asks for, so fix(test): ingest dispatch-loop test races t.TempDir cleanup, reddens main #442 stays open.
• Small edits to feat(mq): external nats broker with sharded, pinned ingest workers #624's files: Finding.transient (unexported) and the fixture keeping its Options, so a test can restart the server.

Tests (all integration-tagged; ~6s together)

• Conformance: mqtest.Run against ExternalNATS, connected as the restricted wavehouse user from deployments/nats/values.yaml, with each case on its own server and the shipped N=4 manifests. Caps: PerTenantBudget, PurgesAcked, UnbudgetedNotFound and ConfiguresDurables are all false. EndDelivery deletes wh-ingest on every partition as the operator. Fill shrinks the tenant's partition to 4 KiB. 18/18 applicable cases pass. AckWaitRedelivers is skipped by its cap. This proves the published permission set for publish, consume, ack, DLQ, replay and the hub, beyond feat(mq): external nats broker with sharded, pinned ingest workers #624's read-only checks (measured). The run was stable over -count=8 under heavy load.
• Reconnect: the server is stopped mid-consume. A publish while it is down is ErrUnavailable and not ErrQueueFull. After a restart, publishing resumes, the consumer delivers again, and failed stays quiet.
• Idempotent retry: the answers to the first two attempts are lost. All three attempts carry the same Nats-Msg-Id and the partition holds one message. A new publish gets a new id. Losing every answer is ErrUnavailable.
• A topic at max_msgs_per_subject is ErrQueueFull, and only that topic.
• A deleted partition is ErrUnavailable naming the stream, and the gauge drops. The stream is not recreated.
• Boot: a missing DLQ is ErrTopology naming it. An unreachable server is ErrUnavailable within the wait. Conflicting options are refused.
• Auth: nkey seed, creds file (operator mode, in-process), and mutual TLS with a CA generated in the test. Missing credentials are ErrUnavailable.
• Re-check: a DLQ deleted after boot drops topology_ok, and recreating it restores it. After a server restart the sources read as silent and topology_ok stays 1.
• Gauges read through an OTel manual reader. PurgeAcked warns once per tenant. CreateConsumer checks the durable and never creates one.
• Replay: a slow client (25ms/event) over more than one batch gets the whole replay (this fails on the old 5s threshold with "no responders"). Events published during a replay are not sent by it.
• Duplicate window: a window of exactly 2 × PublishTimeout is refused.
• Permissions: the wavehouse user is refused stream create, update, purge and delete, a durable on a partition, and deleting wh-ingest.
• Measurements (WH_MQ_MEASURE=1; in-process single-node server, file storage, under heavy load; inferred to be optimistic for an R3 cluster):
  • publish-to-hub latency over 2000 events: p50 62µs, p99 180µs, max 525µs. That is far under design risk 3's 50ms threshold, so the hub stays on the history.
  • one tenant's publish throughput: 16,000 × 256 B from 32 callers in 0.20s, ≈80k/s (design risk 4).

Evidence

• make ci: green on 2757e64 (unit 93.3%, integration 51.6%, e2e 61.2%, Go total 94.0%). Earlier runs on c8ad3f0 failed only in internal/mq's 15s unit budget (test(app): internal/app unit tests use 8–17 s of the 15 s budget, time out under load #617, under heavy load) and in the fix(test): ingest dispatch-loop test races t.TempDir cleanup, reddens main #442 ingest cleanup race, which the last commit mitigates.
• Reviewers (markers land in the main checkout, fix(agents): pre-push review gate can attest to unreviewed code (marker inherited across commits; wrong worktree checked) #454, so the gate cannot attest to them there; verdicts are recorded here):
  • pre-push-reviewer (opus), four rounds. Round 1 raised three [SHOULD]s: the source gauge read -1e-9 not -1; the duplicate-window floor was below the retry schedule; ReplaySince made a round trip per event and chased the tail. Round 2 raised one [SHOULD]: batching under a 5s inactive threshold lost a slow client's gap-fill (measured). Rounds 3 and 4 (the StoreDir commit) were ship_it.
  • docs-reviewer (opus): ship_it in round 1; one [MAY] in round 2 ("purges" in AGENTS.md); ship_it in round 3 (c8ad3f0). It was skipped on record for 2757e64, a test-helper-only commit.

Deliberately left to later PRs

• D4: mq.backend/mq.nats.* config and validation, the wireMQ case, tests/integration/mq_nats_test.go through app, and the docs pages (configuration, deployment "External NATS (Kubernetes)", architecture, ingest-pipeline, durability, api).
• D5: Sharded / ConsumerConfig.Shards, so a worker consumes only the partitions it claims.
• D6: per-tenant budgets via the backlog KV. Until then SetMaxBytes only records.

Follow-up: dead letters counted per table (7847c14, 50a1170)

DeadLetterCounts on both ExternalNATS and EmbeddedNATS folded a scoped topic into the name table + "." + scope, so a table literally named a.b could share one count with table a scoped b, and the ?table= filter matched only a table's unscoped subject. Both brokers now key their per-subject counts through one deadLetterTables (internal/mq/deadletter.go): every scope of a table counts under the table itself, and the ?table= filter keeps all of that table's scopes. deadletter.go and its test are byte-identical to the same files on #655, so they merge without conflict (measured with git merge-tree; the branches as a whole still conflict, since this one predates main's #612). Total is unchanged on each broker (external sums every subject of the tenant; embedded uses the stream's message count). Scope is always empty today, so GET /v1/ops/dlq/stats does not change visibly; the mqtest conformance suite (deadLetterCounts, deadLetterKeepsTheTopicAndDoesNotAck) pins the new behavior on both brokers, and #623's copy of the suite has to flip the same way (noted on #623).

Evidence for this follow-up:

• make ci green locally at 50a1170 with JOBS=1. At higher parallelism internal/mq's 15 s unit budget trips on this branch, which predates main's test(mq,app): fit the unit budget; fail fast on an uncreatable store #647 speed-up; the mq unit tests took 12.6 s before this change and 12.3 s after it, unloaded, so the change does not cause it.
• Pre-push reviewers: pre-push-reviewer and docs-reviewer ship_it at 50a1170 (round 1 at 7847c14 found the code correct; its findings — this addition's wording and the changelog's file list — are fixed).

🤖 Generated with Claude Code

https://claude.ai/code/session_01EJr5tY4WQUy2sc4MbW67vL

[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: f4d95978-4299-4445-a951-e0b2dfce5ba3

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

---

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.}