feat(mq): external nats broker with sharded, pinned ingest workers

.testcoverage.yml

@@ -54,6 +54,9 @@ exclude:
     # 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/
     - ^tests/
     # scripts/ holds Go helpers (cov, orchestrator) that drive the build but
     # aren't part of the shipped binary; they show up in `-coverpkg=./...`
@@ -101,3 +104,36 @@ exclude:
       # outage. The unit and integration suites cover both.
       - ^internal/config/cache_redis\.go$
       - ^internal/cache/pending\.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/nats_permprobe\.go$
+      - ^internal/mq/subject_nats\.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
+      # wiring and the retention warning under it.
+      - ^internal/app/wire_nats\.go$
+      # The ops-only listener of a process without the api role, moved
+      # out of wire.go the same way: the e2e binary runs every role.
+      - ^internal/app/wire_ops\.go$
+      # The mq.nats block's checks, as cache_redis.go above: boot adopts the
+      # embedded fixture, so e2e never reads it.
+      - ^internal/config/mq_nats\.go$
+      # The ingest processes' shard claims, which run only over the external
+      # broker's shards: the unit suite covers them with a fake queue, and
+      # the integration suite against NATS.
+      - ^internal/ingest/(claims|assign)\.go$
+      - ^internal/mq/settle\.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(_consumer)?\.go$
+      # The NATS KV leases, the same way.
+      - ^internal/mq/lease\.go$

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 carry the `//go:build integration` tag and go in `tests/integration/`, or beside the package when they test one package against its own external server (e.g. `internal/cache/redis_integration_test.go`), with the package added to the `test-integration` target.
+2. Write tests for new functionality. Unit tests go alongside the code in `internal/`. Integration tests carry the `//go:build integration` tag and go in `tests/integration/`, or beside the package when they test one package against its own external server (e.g. `internal/cache/redis_integration_test.go`), with the package added to the `test-integration` target. A test that must import NATS, which only `internal/mq` may do, goes in `internal/mq/natsspike`, or in `internal/mq` itself with the `integration` tag when it needs the package's 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`

Makefile

@@ -762,9 +762,18 @@ test: test-unit
 test-integration: go-mod-download ## Run Go integration tests + render coverage + gate threshold (requires Docker)
 	@printf "$(CYAN)==> Running Integration Tests...$(RESET)\n"
 	@rm -rf $(COV_INT)/data && mkdir -p $(COV_INT)/data
+	@# 480s: the go command kills the package at -timeout + 1m, counting
+	@# TestMain's wavehouse binary build before m.Run, and the multi-process
+	@# roles test runs its handover steps in sequence; at 240s CI killed it.
+	@GOCOVERDIR="$(CURDIR)/$(COV_INT)/data" go tool gotestsum --format $(GOTESTSUM_FMT) -- \
+		-tags="integration $(TAGS)" -timeout 480s -coverpkg=./... -race -count=1 \
+		./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 \
-		./tests/integration/... ./internal/cache/... $(ARGS) \
+		-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
 

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,8 @@ 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 mq permissions  print the wavehouse NATS user's permissions for it
   wavehouse health          liveness self-probe against the local server (container HEALTHCHECK)
   wavehouse version         print version, commit, and build time
   wavehouse help            show this help

cmd/wavehouse/mq.go

@@ -0,0 +1,182 @@
+package main
+
+import (
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/Wave-RF/WaveHouse/internal/dedupe"
+	"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
+  permissions   print the wavehouse NATS user's permissions for it
+`)
+	}
+	if len(args) == 0 {
+		usage(stderr)
+		return 2
+	}
+	switch args[0] {
+	case "manifests":
+		return runMQManifests(args[1:], stdout, stderr)
+	case "permissions":
+		return runMQPermissions(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)")
+	shards := fs.Int("shards", mq.DefaultNATSShards, "shards of every partition, one durable each (mq.nats.shards)")
+	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")
+	consumer := fs.String("ingest-consumer", mq.DefaultNATSIngestConsumer, "the shard durables' name prefix (mq.nats.ingest_consumer)")
+	history := fs.String("history-stream", "", "the history stream (mq.nats.history_stream); empty is <PREFIX>_HISTORY")
+	publishTimeout := fs.Duration("publish-timeout", mq.DefaultNATSPublishTimeout, "one publish attempt's bound (mq.nats.publish_timeout); the partitions' duplicate window covers every attempt")
+	bucket := fs.String("coord-bucket", "", "the lease KV bucket (coord.nats.bucket); empty is <prefix>_coord")
+	lease := fs.Duration("dedupe-lease", dedupe.DefaultLease, "the dedupe.lease the partitions' duplicate window must cover")
+	fileStore := fs.String("file-store", mq.FormatStoreSize(mq.DefaultNATSFileStore), "every server's JetStream max_file_store (the Helm chart's fileStore.maxSize, else its PVC size), which the streams' maxBytes must fit in")
+	partitionBytes := fs.String("partition-max-bytes", "", "each ingest partition's maxBytes; empty is 15% of --file-store")
+	fs.Usage = func() {
+		_, _ = fmt.Fprint(fs.Output(), `usage: wavehouse mq manifests [--partitions N] [--shards V] [--prefix wh] [--replicas 3] [--ingest-consumer wh-ingest] [--history-stream S] [--publish-timeout 5s] [--coord-bucket B] [--dedupe-lease 30s] [--file-store 100Gi] [--partition-max-bytes 15Gi]
+
+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.
+The streams' maxBytes are sized from --file-store: each partition 15%, the
+history 10% and the DLQ 5%, so four partitions leave a quarter of it spare.
+It refuses streams that together reserve more than --file-store.
+
+`)
+		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 *lease < 0 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --dedupe-lease must not be negative\n")
+		return 2
+	}
+	if *shards < 1 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --shards must be at least 1\n")
+		return 2
+	}
+	if *replicas < 1 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --replicas must be at least 1\n")
+		return 2
+	}
+	if *publishTimeout <= 0 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --publish-timeout must be positive\n")
+		return 2
+	}
+	store, err := positiveSize(*fileStore)
+	if err != nil {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --file-store: %v\n", err)
+		return 2
+	}
+	var partitionMax int64
+	if *partitionBytes != "" {
+		if partitionMax, err = positiveSize(*partitionBytes); err != nil {
+			_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: --partition-max-bytes: %v\n", err)
+			return 2
+		}
+	}
+	err = mq.WriteNATSManifests(stdout, mq.NATSManifestOptions{
+		Topology: mq.NATSTopology{
+			Prefix: *prefix, Partitions: *partitions, Shards: *shards, IngestConsumer: *consumer, HistoryStream: *history,
+			PublishTimeout: *publishTimeout, CoordBucket: *bucket, DedupeLease: *lease,
+		},
+		Replicas:          *replicas,
+		FileStore:         store,
+		PartitionMaxBytes: partitionMax,
+	})
+	if err != nil {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq manifests: %v\n", err)
+		return 1
+	}
+	return 0
+}
+
+// runMQPermissions implements `wavehouse mq permissions`: print the
+// wavehouse NATS user's permissions for the topology, which name every shard
+// durable, so an operator regenerates them with the shard count.
+func runMQPermissions(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("mq permissions", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	shards := fs.Int("shards", mq.DefaultNATSShards, "shards of every partition (mq.nats.shards); while a lower count's durables drain, the higher")
+	prefix := fs.String("prefix", mq.DefaultNATSSubjectPrefix, "subject prefix (mq.nats.subject_prefix)")
+	consumer := fs.String("ingest-consumer", mq.DefaultNATSIngestConsumer, "the shard durables' name prefix (mq.nats.ingest_consumer)")
+	history := fs.String("history-stream", "", "the history stream (mq.nats.history_stream); empty is <PREFIX>_HISTORY")
+	bucket := fs.String("coord-bucket", "", "the lease KV bucket (coord.nats.bucket); empty is <prefix>_coord")
+	domain := fs.String("js-domain", "", "the JetStream domain WaveHouse connects to (mq.nats.js_domain); its API requests go to $JS.<domain>.API.>")
+	fs.Usage = func() {
+		_, _ = fmt.Fprint(fs.Output(), `usage: wavehouse mq permissions [--shards V] [--prefix wh] [--ingest-consumer wh-ingest] [--history-stream S] [--coord-bucket B] [--js-domain D]
+
+Print the permissions block for the wavehouse user of a NATS Helm values
+file (config.merge.accounts.<account>.users[]), for the topology that
+wavehouse mq manifests prints with the same flags.
+
+`)
+		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 permissions: unexpected argument %q\n", fs.Arg(0))
+		fs.Usage()
+		return 2
+	}
+	if *shards < 1 {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq permissions: --shards must be at least 1\n")
+		return 2
+	}
+	err := mq.WriteNATSPermissions(stdout, mq.NATSTopology{
+		Prefix: *prefix, Shards: *shards, IngestConsumer: *consumer, HistoryStream: *history, CoordBucket: *bucket,
+	}, *domain)
+	if err != nil {
+		_, _ = fmt.Fprintf(stderr, "wavehouse mq permissions: %v\n", err)
+		return 1
+	}
+	return 0
+}
+
+// positiveSize reads a byte count of at least one byte (mq.ParseStoreSize).
+func positiveSize(s string) (int64, error) {
+	n, err := mq.ParseStoreSize(s)
+	if err == nil && n == 0 {
+		err = errors.New("must be positive")
+	}
+	return n, err
+}

cmd/wavehouse/mq_test.go

@@ -0,0 +1,116 @@
+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 of eight shards.
+// Regenerate with: go run ./cmd/wavehouse mq manifests --partitions 4 --shards 8 > deployments/nats/jetstream.yaml
+// (the file store defaults to the shipped values' 100Gi PVC).
+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", "--shards", "8"}, &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},
+		"negative lease":    {[]string{"manifests", "--dedupe-lease", "-1s"}, 2},
+		"defaults generate": {[]string{"manifests"}, 0},
+		"bad file store":    {[]string{"manifests", "--file-store", "100GB"}, 2},
+		"zero file store":   {[]string{"manifests", "--file-store", "0"}, 2},
+		"bad partition cap": {[]string{"manifests", "--partition-max-bytes", "lots"}, 2},
+		"over the store":    {[]string{"manifests", "--partitions", "6"}, 1},
+		"resized to fit":    {[]string{"manifests", "--partitions", "6", "--partition-max-bytes", "10Gi"}, 0},
+		"zero timeout":      {[]string{"manifests", "--publish-timeout", "0s"}, 2},
+		"bad consumer":      {[]string{"manifests", "--ingest-consumer", "a.b"}, 1},
+		"bad history":       {[]string{"manifests", "--history-stream", "A B"}, 1},
+		"bad js domain":     {[]string{"permissions", "--js-domain", "a.b"}, 1},
+	}
+	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")
+}
+
+// A lease past the shipped 2m window's reach widens every partition's window
+// to cover it (90s + 90s + 1s), so the manifests pass the boot check.
+func TestRunMQManifests_CoversTheDedupeLease(t *testing.T) {
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"manifests", "--partitions", "1", "--dedupe-lease", "90s"}, &out, &errOut), errOut.String())
+	assert.Contains(t, out.String(), "duplicateWindow: 3m1s\n")
+}
+
+// The permissions name every shard's durable, and match what the verifier's
+// test holds the shipped values to.
+func TestRunMQPermissions(t *testing.T) {
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"permissions", "--shards", "3", "--prefix", "acme"}, &out, &errOut), errOut.String())
+	for _, want := range []string{"$JS.API.CONSUMER.MSG.NEXT.*.wh-ingest-2", "$JS.API.CONSUMER.UNPIN.*.wh-ingest-0", "$JS.API.CONSUMER.RESET.*.wh-ingest-1", "acme.hist.>", "_INBOX_acme.>"} {
+		assert.Contains(t, out.String(), want)
+	}
+	assert.NotContains(t, out.String(), "wh-ingest-3")
+	out.Reset()
+	errOut.Reset()
+	assert.Equal(t, 2, runMQ([]string{"permissions", "--shards", "0"}, &out, &errOut))
+	assert.Contains(t, errOut.String(), "--shards must be at least 1")
+	errOut.Reset()
+	assert.Equal(t, 1, runMQ([]string{"permissions", "--shards", "257"}, &out, &errOut))
+	assert.Contains(t, errOut.String(), "shards must be from 1 to 256")
+}
+
+// The names and publish timeout the config takes are flags too, and the
+// partitions' duplicate window covers every attempt at that timeout
+// (3 × 1m + 2 × 250ms).
+func TestRunMQManifests_NamesAndPublishTimeout(t *testing.T) {
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"manifests", "--partitions", "1", "--shards", "1", "--ingest-consumer", "acme-ingest", "--history-stream", "ACME_REPLAY", "--publish-timeout", "1m"}, &out, &errOut), errOut.String())
+	for _, want := range []string{
+		"durableName: acme-ingest-0\n", "name: ACME_REPLAY\n", "duplicateWindow: 3m0.5s\n",
+		"--ingest-consumer acme-ingest --history-stream ACME_REPLAY --publish-timeout 1m0s",
+	} {
+		assert.Contains(t, out.String(), want)
+	}
+}
+
+// Under a JetStream domain the API and KV subjects come in the domain's form
+// as well as the plain one.
+func TestRunMQPermissions_JSDomain(t *testing.T) {
+	var out, errOut bytes.Buffer
+	require.Equal(t, 0, runMQ([]string{"permissions", "--shards", "1", "--js-domain", "hub"}, &out, &errOut), errOut.String())
+	for _, want := range []string{
+		"$JS.hub.API.CONSUMER.MSG.NEXT.*.wh-ingest-0", "$JS.API.CONSUMER.MSG.NEXT.*.wh-ingest-0",
+		"$JS.hub.API.$KV.wh_coord.lease.>", "$JS.hub.API.DIRECT.GET.KV_wh_coord.$KV.wh_coord.lease.>", "$JS.hub.API.STREAM.CREATE.>",
+		"JetStream domain hub",
+	} {
+		assert.Contains(t, out.String(), want)
+	}
+}