[EricAndrechek]

Part of #613. Fixes two defects in ExternalNATS that D4 (#639) found. Base: feat/mq-nats-wiring (#639), so review after #636 and #639.

1. A deliberate close logged a WARN

ExternalNATS.Close logged mq: disconnected from nats; reconnecting at WARN with error=null. This was measured in D4's integration run. The cause is that nats.go calls the disconnect callback from Close/Drain with a nil error. Close now sets a closing flag before it drains, and the callback does not warn once that flag is set. A real disconnect still warns.

2. Rows in removed partitions were never drained

D2's verifier said that streams outside the N partitions were "drained until the operator deletes them". D3's CreateConsumer only consumed partitions 0..N-1, and deployment.md told operators to stop ingest before lowering N.

Choice: (a), drain them. CreateConsumer now also binds wh-ingest on every stream that holds <prefix>.ingest.> subjects outside the N partitions, and consumes it beside the partitions.

Why (a) and not (b), refusing to boot while an extra partition holds rows:

• (b) cannot be operated without downtime. While the fleet rolls out, the processes still on the old N keep publishing into the partition being removed. A new process that refuses to boot until that partition is empty waits on processes it is meant to replace. The only way out is to stop ingest, which is what the old doc said to do.
• (a) needs no new procedure. The operator applies the regenerated manifests and rolls out. The new processes drain the removed partition while the old ones still feed it. The generated resources set preventDelete, so the stream and its durable outlive their removal from the manifests. Nothing is lost unless the operator deletes a partition that still holds rows, and deleting any partition that holds rows already loses them.
• (a) is also what the verifier already promised. Only the code and the docs had to change.

Details:

• Deleting a drained extra stream is not a failure. The end of that stream's delivery is logged at INFO and is not reported on failed, so the worker keeps running. Client errors on an extra's delivery are logged at INFO too, because they are expected once the operator deletes it. An extra with no pull wh-ingest durable, or one already gone, is skipped.
• The verifier's finding (extraPartitions, still recommended) now counts the stream's rows. It also says when the stream has no durable, so nothing drains it. With interest retention, rows with no durable have already been dropped by the server, unless some other consumer still holds them (inferred from nats-server's cleanup when a consumer is deleted, not tested).
• Prefetch is split over the N partitions only. Each extra gets a quarter share, so a stream that has been drained and deleted does not keep cutting the partitions' fetch-ahead until the next restart. Total fetch-ahead runs slightly over the budget while extras exist.
• SSE limit: the regenerated history no longer sources the removed partition. Rows that the old processes publish there after the apply reach ClickHouse, but not live SSE or replay. deployment.md says so.

Tests

• TestExternalNATS_CloseLogsNoWarning (integration, serial because it captures the default logger). It closes a broker that has a running consumer and a hub subscription, and asserts nothing was logged at WARN or above. It then stops the server and asserts that another broker still warns. It fails on the base with the nil-error WARN (measured).

• TestExternalNATS_LoweringNDrainsTheRemovedPartition (integration, serial). Steps:

  • Apply wavehouse mq manifests --partitions 2.
  • Publish 5 rows to partition 1 from an N=2 broker.
  • Apply the N=1 manifests with the new natstest.Manifests.Apply (create-or-update, as nack does).
  • Boot an N=1 broker. The finding names WH_INGEST_1 with drains its 5 rows.
  • All 5 rows arrive through the worker's consumer.
  • The N=2 broker, still up, publishes again, and the new worker gets that row too. WH_INGEST_1 then empties.
  • The operator deletes the stream. The INFO line "stopped draining" appears, failed stays quiet, and partition 0 keeps delivering.

  The earlier version of this test fails on the base with a 15s receive timeout (measured).

• TestVerifyNATSTopology_Findings: the extra-partition case checks the new text, plus a new case for an extra partition without its durable.

Evidence

• make ci (queued, GOTOOLCHAIN=go1.26.6) was green on 6d42f5a. Coverage: unit 93.5%, integration 56.9%, e2e 60.8%, Go total 94.3%.
  • The first run, on 18d47de, failed with the internal/mq unit package's 15s timeout under heavy load; the running test was 2s in, so it was not a hang.
• make build-docs passes.
• Pre-existing flake, not from this PR: TestEmbeddedNATS_PacesTheRetriesOfAQueueThatCannotOpen failed 9 of 10 runs with -count=10 on a clean origin/feat/mq-nats-wiring checkout (measured). The error is open …/streams/DLQ_acme: invalid argument / no such file or directory. It passed on its own inside the green make ci.
• pre-push-reviewer (opus):
  • Round 1: iterate, with 0 MUST.
    • SHOULD: the shrink test did not cover an old-N process publishing during the rollout, or the extra's delivery actually ending. Both are now covered.
    • SHOULD: the stream delete step was unclear. It is now named.
    • MAY: the prefetch split. Done.
  • Round 2: ship_it.
• docs-reviewer (opus):
  • Round 1: iterate, with 2 SHOULD.
    • The delete step under preventDelete. Now fixed.
    • Two unqualified "durable deleted ends the worker" claims in ingest-pipeline.md. Both now fixed.
  • Round 2: ship_it.
• Known gate gap (fix(agents): pre-push review gate can attest to unreviewed code (marker inherited across commits; wrong worktree checked) #454): reviewer markers land in the main checkout, so the verdicts are recorded here. No marker was hand-written.

Expected conflicts

• B2 (feat/coord-nats-kv) adds a KV bucket to the topology spec, the verifier and the manifests. This PR's nats_topology.go edit is confined to the body of extraPartitions and its doc comment. A conflict is only likely if B2 touches extraPartitions or the lines right around it. B2 may also add a stream that holds subjects under <prefix>.ingest.>; it should not, or the new drain would treat it as an extra partition.
• natstest/natstest.go: a new Manifests.Apply method, added between Create and AwaitSources. It merges trivially unless B2 edits the same spot.
• CHANGELOG: a new first bullet under Unreleased → Fixed.

Left to later PRs

• D5 (Sharded, partition claiming): extras need an owner there too. The simplest option is that every claimant drains every extra, since extras are temporary.
• Nothing in this PR re-scans for extras after CreateConsumer. That is fine, because an extra appears only when N changes, which requires a restart.

🤖 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: 3aa415b8-40b1-4421-8f61-cb8894f6617a

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