Wave-RF · EricAndrechek · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026
diff --git a/.github/labeler.yml b/.github/labeler.yml
@@ -34,6 +34,11 @@
     - any-glob-to-any-file:
       - "internal/cache/**"
 
+"area/coord":
+  - changed-files:
+    - any-glob-to-any-file:
+      - "internal/coord/**"
+
 "area/dedupe":
   - changed-files:
     - any-glob-to-any-file:

diff --git a/.testcoverage.yml b/.testcoverage.yml
@@ -48,6 +48,15 @@ exclude:
     # HTTP assertions). It is imported only from *_test.go files, never from
     # production code, so there's nothing meaningful to cover.
     - ^internal/testutil/
+    # internal/mq/mqtest/ is the Broker conformance suite: test code that
+    # lives outside *_test.go only so each backend's tests can import it.
+    - ^internal/mq/mqtest/
+    # internal/mq/natstest/ stands up NATS as an operator deploys it, for
+    # tests outside internal/mq; test code, like mqtest.
+    - ^internal/mq/natstest/
+    # The coord conformance suite: test helpers every Coordinator's tests
+    # run, imported only from *_test.go like testutil.
+    - ^internal/coord/coordtest/
     - ^tests/
     # scripts/ holds Go helpers (cov, orchestrator) that drive the build but
     # aren't part of the shipped binary; they show up in `-coverpkg=./...`
@@ -73,3 +82,37 @@ exclude:
       - ^internal/settings/
       - ^cmd/wavehouse/validate\.go$
       - ^cmd/wavehouse/bootstrap\.go$
+      # The external-NATS topology spec, verifier and manifest generator
+      # (and the `mq manifests` CLI) run against an operator's NATS, which
+      # the e2e stack (embedded broker) never has: unit territory, covered
+      # there by a fixture server. Excluding them keeps e2e at ~61%. The
+      # external broker is the same, covered by the integration suite.
+      - ^internal/mq/nats_topology\.go$
+      - ^internal/mq/nats_manifests\.go$
+      - ^internal/mq/subject_nats\.go$
+      - ^internal/mq/external\.go$
+      - ^internal/mq/lease\.go$
+      - ^cmd/wavehouse/mq\.go$
+      # The in-process cache backend: the e2e stack runs cache.backend=redis
+      # (#613), so the binary carries LocalCache and its version index but e2e
+      # never reaches them. The unit suite and the integration suite's main
+      # app (cache.backend=local) cover them; the merged total still counts them.
+      - ^internal/cache/(local|version_manager)\.go$
+      # The DynamoDB dedupe backend: the e2e binary runs Pebble dedupe, so
+      # this file measured 0% there and pulled e2e to 58.6%. The unit
+      # (fake API) and integration (dynamodb-local) suites cover it, and the
+      # merged total still counts it.
+      - ^internal/dedupe/dynamodb\.go$
+      # Their wiring, moved out of wire.go for this: the NATS queue and lease
+      # wiring and the retention warning under it, the DynamoDB dedupe
+      # wiring, and the ops-only listener of a process without the api role
+      # (the e2e binary runs every role). Integration and unit suites cover
+      # them; the merged total still counts them.
+      - ^internal/app/wire_(nats|dynamodb|ops)\.go$
+    unit:
+      # 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$
+      # 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/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -39,7 +39,7 @@ Open a [feature request issue](https://github.com/Wave-RF/WaveHouse/issues/new?t
 
    The pre-push hook (installed by `make tools`) blocks a push until the tree has been validated locally: a code change needs `make ci`, a docs/prose-only change needs only `make verify` (the same split CI makes). `make lint` / `make test` / `make build` are fast inner-loop subsets.
 
-2. Write tests for new functionality. Unit tests go alongside the code in `internal/`. Integration tests go in `tests/` with the `//go:build integration` tag.
+2. Write tests for new functionality. Unit tests go alongside the code in `internal/`. Integration tests go in `tests/` with the `//go:build integration` tag. The exception is a test that must import NATS, which only `internal/mq` may do; such tests go in `internal/mq/natsspike`, or in `internal/mq` itself with the `integration` tag when they need its internals (the external NATS broker's tests, which `make test-integration` selects by name).
 
 3. Update documentation if your change affects:
    - API endpoints → update `docs/src/content/docs/api.md`

diff --git a/Makefile b/Makefile
@@ -764,7 +764,13 @@ test-integration: go-mod-download ## Run Go integration tests + render coverage
 	@rm -rf $(COV_INT)/data && mkdir -p $(COV_INT)/data
 	@GOCOVERDIR="$(CURDIR)/$(COV_INT)/data" go tool gotestsum --format $(GOTESTSUM_FMT) -- \
 		-tags="integration $(TAGS)" -timeout 240s -coverpkg=./... -race -count=1 \
-		./tests/integration/... $(ARGS) \
+		./tests/integration/... ./internal/mq/natsspike/... ./internal/cache/... $(ARGS) \
+		-args -test.gocoverdir="$(CURDIR)/$(COV_INT)/data"
+	@# internal/mq's integration-tagged tests (the external NATS broker) run
+	@# alone: its untagged tests are the unit suite's.
+	@GOCOVERDIR="$(CURDIR)/$(COV_INT)/data" go tool gotestsum --format $(GOTESTSUM_FMT) -- \
+		-tags="integration $(TAGS)" -timeout 240s -coverpkg=./... -race -count=1 \
+		-run '^Test(ExternalNATS|NewNATS|NATSPermissions_Refuse|Leases)' ./internal/mq $(ARGS) \
 		-args -test.gocoverdir="$(CURDIR)/$(COV_INT)/data"
 	@if [ -z "$(COV_DEFER)" ]; then go run ./scripts/cov render integration; fi
 

diff --git a/README.md b/README.md
@@ -71,8 +71,8 @@ ClickHouse is a phenomenal OLAP database, but pointing a frontend right at it le
 
 If you're building user-facing analytics, WaveHouse is like **Supabase for ClickHouse**. Or an **open-source Tinybird** that pushes data to the frontend in real time over SSE, not just pull-based REST.
 
-- **Ingest** — async durable WAL (embedded NATS JetStream), `200 OK` instantly, background batch-flush; schema-validated against `system.columns`; optional ID-based dedup (idempotent ingest); dead-letter queue for failed inserts.
-- **Query** — in-process Ristretto cache + `singleflight` coalescing; type-safe structured query AST; Tinybird-style named pipes (parameterized SQL endpoints).
+- **Ingest** — async durable WAL (embedded NATS JetStream), `200 OK` instantly, background batch-flush; schema-validated against `system.columns`; optional ID-based dedup (idempotent ingest); dead-letter queue for rows ClickHouse rejects (an unavailable ClickHouse is retried with backoff, not dead-lettered).
+- **Query** — result cache (in-process Ristretto, or a Redis shared by every instance) + `singleflight` coalescing; type-safe structured query AST; Tinybird-style named pipes (parameterized SQL endpoints).
 - **Real-time** — native SSE push, broadcast *before* the ClickHouse flush, with JetStream gap-fill for late/reconnecting clients.
 - **Security** — Hasura-style per-table, per-role column + row policies with JWT claim templating, defined in the hot-reloadable settings directory.
 - **Client** — `@wavehouse/sdk`: TypeScript client with query builder, live queries, streaming, and schema codegen; one runtime dependency (an SSE frame parser, ~1.4 KB gzipped).

diff --git a/clients/ts/src/errors.test.ts b/clients/ts/src/errors.test.ts
@@ -63,6 +63,44 @@ describe("parseErrorResponse", () => {
     expect(e.retryable).toBe(true);
   });
 
+  it("takes the server's code and retryable when it sends them", async () => {
+    const res = new Response(
+      JSON.stringify({
+        error: "Code: 62. Syntax error",
+        code: "clickhouse.rejected",
+        retryable: false,
+      }),
+      { status: 400, statusText: "Bad Request" },
+    );
+    const e = await parseErrorResponse(res);
+    expect(e.code).toBe("clickhouse.rejected");
+    expect(e.retryable).toBe(false);
+  });
+
+  it("lets the server mark a 5xx not retryable", async () => {
+    const res = new Response(
+      JSON.stringify({
+        error: "Authentication failed",
+        code: "clickhouse.misconfigured",
+        retryable: false,
+      }),
+      { status: 502, statusText: "Bad Gateway" },
+    );
+    const e = await parseErrorResponse(res);
+    expect(e.code).toBe("clickhouse.misconfigured");
+    expect(e.retryable).toBe(false);
+  });
+
+  it("ignores a non-string code and a non-boolean retryable", async () => {
+    const res = new Response(JSON.stringify({ error: "x", code: 123, retryable: "no" }), {
+      status: 500,
+      statusText: "Internal Server Error",
+    });
+    const e = await parseErrorResponse(res);
+    expect(e.code).toBe("HTTP_500");
+    expect(e.retryable).toBe(true);
+  });
+
   it("marks 4xx as not retryable", async () => {
     const res = new Response(JSON.stringify({ error: "forbidden" }), {
       status: 403,

diff --git a/clients/ts/src/errors.ts b/clients/ts/src/errors.ts
@@ -16,10 +16,14 @@ export async function parseErrorResponse(res: Response): Promise<WaveHouseError>
         ? body.message
         : res.statusText;
 
-  const retryable = res.status === 503 || res.status >= 500;
+  // The server's own `code` and `retryable` win where it sends them (a
+  // failed ClickHouse query, for one); the status decides otherwise.
+  const code =
+    typeof body?.code === "string" && body.code !== "" ? body.code : `HTTP_${res.status}`;
+  const retryable = typeof body?.retryable === "boolean" ? body.retryable : res.status >= 500;
   return {
     status: res.status,
-    code: `HTTP_${res.status}`,
+    code,
     message,
     details: body,
     retryable,

diff --git a/clients/ts/src/http.test.ts b/clients/ts/src/http.test.ts
@@ -108,6 +108,29 @@ describe("request", () => {
     expect(result.error?.retryable).toBe(false);
   });
 
+  it("does not retry a 5xx the server marks not retryable", async () => {
+    fetchSpy.mockResolvedValue(
+      new Response(
+        JSON.stringify({
+          error: "Authentication failed",
+          code: "clickhouse.misconfigured",
+          retryable: false,
+        }),
+        { status: 502 },
+      ),
+    );
+
+    const result = await request(makeCtx({ options: { maxRetries: 2 } }), {
+      method: "POST",
+      path: "/v1/query?table=clicks",
+      body: {},
+    });
+
+    expect(fetchSpy).toHaveBeenCalledOnce();
+    expect(result.error?.code).toBe("clickhouse.misconfigured");
+    expect(result.error?.retryable).toBe(false);
+  });
+
   it("returns error for 500 without retry when maxRetries=0", async () => {
     fetchSpy.mockResolvedValue(
       new Response(JSON.stringify({ error: "internal" }), { status: 500 }),

diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go
@@ -97,6 +97,8 @@ func main() {
 			os.Exit(runValidate(os.Args[2:]))
 		case "bootstrap":
 			os.Exit(runBootstrap(os.Args[2:]))
+		case "mq":
+			os.Exit(runMQ(os.Args[2:], os.Stdout, os.Stderr))
 		case "version", "--version", "-v":
 			fmt.Printf("wavehouse %s (commit %s, built %s)\n", Version, GitCommit, BuildTime)
 			os.Exit(0)
@@ -138,6 +140,7 @@ func printUsage(w io.Writer) {
   wavehouse                 start the server
   wavehouse validate [dir]  validate a settings directory (dir falls back to %[1]s)
   wavehouse bootstrap [dir] write a starter settings directory, every key at its default (dir falls back to %[1]s)
+  wavehouse mq manifests    print the nack resources for an external NATS JetStream
   wavehouse health          liveness self-probe against the local server (container HEALTHCHECK)
   wavehouse version         print version, commit, and build time
   wavehouse help            show this help
@@ -171,14 +174,17 @@ func run(ctx context.Context) int {
 		return 1
 	}
 
-	// data_dir must be writable before anything dials out, so the refusal
-	// (and, for the typical cause — a bind mount owned by root rather than
-	// UID 65532 — the remediation) lands at the top of the log rather than
-	// after ClickHouse discovery. NATS and Pebble still fail loud on their
-	// own if the directory changes underneath us.
-	if err := config.CheckDataDir(cfg.DataDir); err != nil {
-		logger.Error("check data_dir", "error", err)
-		return 1
+	// data_dir, when a selected backend keeps state there, must be writable
+	// before anything dials out, so the refusal (and, for the typical cause —
+	// a bind mount owned by root rather than UID 65532 — the remediation)
+	// lands at the top of the log rather than after ClickHouse discovery.
+	// NATS and Pebble still fail loud on their own if the directory changes
+	// underneath us.
+	if cfg.NeedsDataDir() {
+		if err := config.CheckDataDir(cfg.DataDir); err != nil {
+			logger.Error("check data_dir", "error", err)
+			return 1
+		}
 	}
 
 	a, err := app.New(ctx, app.Options{

diff --git a/cmd/wavehouse/mq.go b/cmd/wavehouse/mq.go
@@ -0,0 +1,83 @@
+package main
+
+import (
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/Wave-RF/WaveHouse/internal/mq"
+)
+
+// runMQ implements `wavehouse mq <command>`: tooling for the message queue.
+// Exit codes: 0 ok, 1 failed, 2 usage.
+func runMQ(args []string, stdout, stderr io.Writer) int {
+	usage := func(w io.Writer) {
+		_, _ = fmt.Fprint(w, `usage: wavehouse mq <command>
+
+commands:
+  manifests   print the nack resources for an external NATS JetStream
+`)
+	}
+	if len(args) == 0 {
+		usage(stderr)
+		return 2
+	}
+	switch args[0] {
+	case "manifests":
+		return runMQManifests(args[1:], stdout, stderr)
+	case "help", "-h", "--help":
+		usage(stdout)
+		return 0
+	default:
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq: unknown command %q\n\n", args[0])
+		usage(stderr)
+		return 2
+	}
+}
+
+// runMQManifests implements `wavehouse mq manifests`: print the nack
+// Stream, Consumer and KeyValue resources for the topology WaveHouse checks at boot
+// under mq.backend: nats, for the operator to apply.
+func runMQManifests(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("mq manifests", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	partitions := fs.Int("partitions", mq.DefaultNATSPartitions, "number of ingest partition streams (mq.nats.partitions)")
+	prefix := fs.String("prefix", mq.DefaultNATSSubjectPrefix, "subject prefix (mq.nats.subject_prefix)")
+	replicas := fs.Int("replicas", 3, "replicas for every stream and the lease bucket")
+	bucket := fs.String("coord-bucket", "", "the lease KV bucket (coord.nats.bucket); empty is <prefix>_coord")
+	fs.Usage = func() {
+		_, _ = fmt.Fprint(fs.Output(), `usage: wavehouse mq manifests [--partitions N] [--prefix wh] [--replicas 3] [--coord-bucket B]
+
+Print the nack (jetstream.nats.io/v1beta2) Stream, Consumer and KeyValue
+resources for the JetStream topology WaveHouse needs under mq.backend: nats
+and coord.backend: nats, as YAML for kubectl apply. WaveHouse never creates these itself; it checks them at boot.
+
+`)
+		fs.PrintDefaults()
+	}
+	if err := fs.Parse(args); err != nil {
+		if errors.Is(err, flag.ErrHelp) {
+			return 0
+		}
+		return 2
+	}
+	if fs.NArg() > 0 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: unexpected argument %q\n", fs.Arg(0))
+		fs.Usage()
+		return 2
+	}
+	if *replicas < 1 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --replicas must be at least 1\n")
+		return 2
+	}
+	err := mq.WriteNATSManifests(stdout, mq.NATSManifestOptions{
+		Topology: mq.NATSTopology{Prefix: *prefix, Partitions: *partitions, CoordBucket: *bucket},
+		Replicas: *replicas,
+	})
+	if err != nil {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: %v\n", err)
+		return 1
+	}
+	return 0
+}
diff --git a/cmd/wavehouse/mq_test.go b/cmd/wavehouse/mq_test.go
@@ -0,0 +1,51 @@
+package main
+
+import (
+	"bytes"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The shipped manifests are the generator's output for four partitions.
+// Regenerate with: go run ./cmd/wavehouse mq manifests --partitions 4 > deployments/nats/jetstream.yaml
+func TestRunMQManifests_MatchesShipped(t *testing.T) {
+	want, err := os.ReadFile("../../deployments/nats/jetstream.yaml")
+	require.NoError(t, err)
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"manifests", "--partitions", "4"}, &out, &errOut), errOut.String())
+	assert.Equal(t, string(want), out.String(), "deployments/nats/jetstream.yaml is stale; regenerate it")
+}
+
+func TestRunMQ_ExitCodes(t *testing.T) {
+	cases := map[string]struct {
+		args []string
+		code int
+	}{
+		"no command":        {nil, 2},
+		"unknown command":   {[]string{"frobnicate"}, 2},
+		"help":              {[]string{"help"}, 0},
+		"manifests help":    {[]string{"manifests", "-h"}, 0},
+		"stray argument":    {[]string{"manifests", "extra"}, 2},
+		"unknown flag":      {[]string{"manifests", "--nope"}, 2},
+		"zero replicas":     {[]string{"manifests", "--replicas", "0"}, 2},
+		"bad prefix":        {[]string{"manifests", "--prefix", "a.b"}, 1},
+		"bad partitions":    {[]string{"manifests", "--partitions", "-1"}, 1},
+		"bad coord bucket":  {[]string{"manifests", "--coord-bucket", "a.b"}, 1},
+		"defaults generate": {[]string{"manifests"}, 0},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			var out, errOut bytes.Buffer
+			assert.Equal(t, tc.code, runMQ(tc.args, &out, &errOut), errOut.String())
+		})
+	}
+}
+
+func TestRunMQManifests_NamesTheLeaseBucket(t *testing.T) {
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"manifests", "--prefix", "acme", "--coord-bucket", "acme_leases"}, &out, &errOut), errOut.String())
+	assert.Contains(t, out.String(), "kind: KeyValue\nmetadata:\n  name: acme-coord\nspec:\n  bucket: acme_leases\n")
+}
diff --git a/cmd/wavehouse/validate_test.go b/cmd/wavehouse/validate_test.go
@@ -19,7 +19,7 @@ func writeSettingsDir(t *testing.T, policies string) string {
 		"roles.json":    `{"roles": ["public"]}`,
 		"policies.json": policies,
 		"pipes.json":    `{}`,
-		"config.json":   `{"clickhouse": {"addr": "localhost:9000", "http_port": 8123, "http_scheme": "http", "database": "default", "username": "default", "query_timeout": 30, "tls": {"enabled": false, "ca_file": "", "cert_file": "", "key_file": "", "insecure_skip_verify": false, "server_name": ""}, "headers": {}, "max_open_conns": 10, "max_idle_conns": 5}, "auth": {"jwks_url": "", "role_claim": "role"}, "dedupe": {"enabled": false, "id_field": "event_id", "require_id": false}, "dlq": {"enabled": true}, "query": {"default_max_rows": 10000, "timestamp_bucket_seconds": 60}, "schema": {"refresh_interval": 60}, "stream": {"keepalive_interval": 30, "keepalive_buckets": 3, "gap_window_minutes": 15}, "mq": {"max_bytes_gb": 1}, "cors": {"allowed_origins": ["*"]}}`,
+		"config.json":   `{"clickhouse": {"addr": "localhost:9000", "http_port": 8123, "http_scheme": "http", "database": "default", "username": "default", "query_timeout": 30, "tls": {"enabled": false, "ca_file": "", "cert_file": "", "key_file": "", "insecure_skip_verify": false, "server_name": ""}, "headers": {}, "max_open_conns": 10, "max_idle_conns": 5}, "auth": {"jwks_url": "", "role_claim": "role"}, "dedupe": {"enabled": false, "id_field": "event_id", "require_id": false, "retention": "0"}, "dlq": {"enabled": true}, "query": {"default_max_rows": 10000, "timestamp_bucket_seconds": 60}, "schema": {"refresh_interval": 60}, "stream": {"keepalive_interval": 30, "keepalive_buckets": 3, "gap_window_minutes": 15}, "mq": {"max_bytes_gb": 1}, "cors": {"allowed_origins": ["*"]}}`,
 	}
 	for name, content := range files {
 		require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600))