build(test): keep make ci's time limits from killing healthy runs

CHANGELOG.md

@@ -97,6 +97,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 ### Fixed
 
+- **`make ci` no longer kills healthy test runs at the `go test` time limits** (`Makefile`, `docs/src/content/docs/development.md`): its parallel phase runs the unit suite beside the lint and build jobs on every core, where `internal/app` and `internal/mq`, about 10s each alone, run past the suite's 15s per-package budget, so that phase now gives the suite 60s (`CI_UNIT_TIMEOUT`). `make test-unit` and the unit job in CI, which run the suite on its own, keep the 15s (`UNIT_TIMEOUT`). The integration suite's limit goes from 480s to 900s (`INTEGRATION_TIMEOUT`): `tests/integration` takes about 6m on a CI runner and up to 8m under Docker Desktop, which 480s left half a minute. Each limit is a Makefile variable, so one can be raised for a run (`make ci CI_UNIT_TIMEOUT=90s`).
+- **Four `internal/mq` tests no longer fail when the embedded NATS server removes its streams directory under them** (`internal/mq/embedded_test.go`): after a stream is deleted, nats-server 2.14.6 removes its streams directory and the account's, once they are empty, on a goroutine of its own. A test that deleted every stream and then opened one raced that removal ([nats-io/nats-server#8725](https://github.com/nats-io/nats-server/issues/8725)), and the open was refused as `error creating store for stream`: rare (one failure in 6,400 local runs), but enough to fail the unit job on a pull request that changed no Go. Each of the four now keeps another stream on the server, so the directory is never empty, as `TestEmbeddedNATS_PacesTheRetriesOfAQueueThatCannotOpen` already did.
+- **`TestExternalNATS_AcksUnderBothAckSubjectLayouts` no longer fails on an ack that races the end of the test** (`internal/mq/external_perms_test.go`): the test published a second row to read the ack subject's layout off the history stream, and the worker's consumer, left running by the drain before it, took that row from the work queue and acked it as the test's context ended (`context canceled`, 9 of 40 runs alone). It now reads the layout off the history stream's copy of the row already drained.
+- **The Redis Cluster integration tests pick another host port when Docker refuses the first** (`internal/cache/redis_integration_test.go`): the one-node cluster publishes a host port the test picks, and Docker Desktop refuses the ports Windows has reserved (`ports are not available`), which lie in the range the pick is drawn from: about one pick in twenty on the machine this was seen on, failing `make ci` in its integration phase. A refused or already-taken port is now answered with another pick, up to five.
 - **A settings directory holding only folders that define no tenant refuses boot again** (`internal/settings/{registry,tree}.go` (+ tests), `internal/app/wire.go` (comment; + tests), `docs/src/content/docs/{deployment,architecture}.md`, `settings-directory.mdx`, `AGENTS.md`): closes [#599](https://github.com/Wave-RF/WaveHouse/issues/599), part of [#583](https://github.com/Wave-RF/WaveHouse/issues/583). A root holding folders and none of the four files is read as nested, and a nested root whose every folder was rejected, or none of whose folders was named by a tenant id — a fresh volume whose only entry is `lost+found`, a parent directory mounted by mistake — opened as a registry serving no tenant, so the server came up and answered every tenant route `unknown tenant`. Boot now refuses such a root with the findings, as a flat invalid root is refused and as this root was before the nested shape existed. The rule is boot's alone: a whole-directory reload that finds every folder gone or broken still drops every tenant and keeps running, and `wavehouse validate` is unchanged. A root with one valid folder beside a rejected or stray one still boots and serves that tenant, and an empty or missing root still refuses boot as the four files, missing.
 - **A tenant moved to another ClickHouse address or database discovers its new schema with the reload** (`internal/app/{wire,discoveries}.go` (+ tests), `internal/discovery/discovery.go` (comment), `internal/testutil/testutil.go`, `docs/src/content/docs/architecture.md`, `settings-directory.mdx`, `AGENTS.md`): closes [#638](https://github.com/Wave-RF/WaveHouse/issues/638). A reload that changed a tenant's `clickhouse.addr` or `clickhouse.database` orphaned the tenant's cache and left its schema registry as it was, so until the tenant's loop fired at `schema.refresh_interval` its queries and inserts were validated against the previous database's schema and run against the new one. The tenants the pools reconcile reports stale now have their registry dropped in the same hook, and the discovery reconcile that follows builds each a fresh one over the pool it is on, as it does for a tenant back after a rejection or removal: the first discovery runs at once in the tenant's own loop, so the reload never waits on ClickHouse, and until it succeeds the tenant's table lookups answer `503` with `Retry-After: 5`, as before any first discovery. A discovery that fails is logged with its tenant (`schema discovery retry failed`), counted in `wavehouse_schema_refresh_failures_total`, and retried with backoff from two seconds to sixty. A tenant whose address and database did not change keeps its registry and its loop, a flat directory's tenant `0` moves the same way, and a process without the api role, which discovers no schema, only repoints its pool.
 - **A shard's owner keeps it while its ClickHouse is slow, one stuck shard no longer stalls the process's others, and a handover or clean stop keeps each table's rows in order** (`internal/mq/external_consumer.go` (new, from `external.go`), `internal/mq/{external,mq}.go`, `internal/ingest/{claims,worker}.go`, `.testcoverage.yml`, tests in `internal/mq/external_test.go`, `internal/ingest/{claims,worker}_test.go` and `tests/integration/shard_order_test.go` (new), `docs/src/content/docs/{deployment,ingest-pipeline,architecture}.md`, `AGENTS.md`): part of the external-NATS workstream of [#613](https://github.com/Wave-RF/WaveHouse/issues/613), fixing the shard ownership of the entries above before they ship. The pull consumer ran the worker's handler on its delivery goroutine and pulled again only when it returned, and the server renews a shard's pin only on a pull, so a process at its cap of held rows for longer than the pinned TTL (10 seconds) lost its pins: `wavehouse_ingest_shards_unowned` counted its shards, and a peer that judged it dead reset them, redelivering rows it still held — two writers per table. Each shard now has a puller that never runs the handler, fetches only what the shard's cap leaves room for, and at the cap renews the pin every 5 seconds with a one-row fetch, at most one `ack_wait` of such rows past the cap, 12 at the generated 1 minute (past that, and once halted, with a pull of `max_bytes` 1, which delivers nothing); measured, an owner blocked 13 seconds on a hung insert kept one pin throughout and took 2 rows past its share, and a unit drains as fast as before (160,000–192,000 rows a second, against 158,000–172,000). `ResetOrphaned` and `Unowned` also require that the shard delivered nothing and had no ack for its pinned TTL. The process-wide cap of 10,000 held rows let one stuck table take every slot and stall every shard the process owned; each shard now holds its own share, plus at most the 12 rows its renewals take (see the entry above), and with one shard's share full a healthy shard's rows kept flowing. A clean stop now halts every shard first (`mq.Halter`), keeping the pins, so the worker writes what it holds and what the shards still deliver, and releases them after its final flush; rows delivered during that flush were dropped before, and came back only once they had been waited out. A handover keeps the pin until the rows delivered are written, bounded by the worker's 60-second ack wait instead of 15 seconds, and a row the worker drops while stopping is NAKed before the release. Checked under continuous publishing, with the first process's ClickHouse slower than a batch window: every table's rows arrived exactly once and in order across a handover and a clean stop, and the next owner wrote its first rows about 2 seconds after the stop, against 9 seconds when the stop leaves rows behind. A shard fetches in 1-second pulls, so a shard stops fetching within a second of a halt, and an idle shard costs one pull a second. A shard durable deleted while no pull of it was waiting stalled silently until the five-minute topology check, since the next pull only got no responders: two such pulls in a row, or one renewal, now look the durable up, and one that is gone, or whose stream is, ends the worker (measured: 5 seconds after the delete, with the handler busy). A bind whose durable or stream is gone, or whose durable no longer fits (`mq.ErrConsumerMismatch`: `ack_wait` shorter than asked, or no `max_ack_pending`), ends the worker instead of retrying every tick with a warning, and a bind failure that lasts a minute logs an error. `wavehouse_ingest_rows_held_waits_total` is gone: nothing waits at the cap any more. A renewal at the cap takes a row rather than holding it back: a held-back redelivery goes to the back of the server's redelivery queue, and with max_bytes-1 renewals rows NAKed as 0, 1, 2, 3 came back as 2, 3, 0, 1 (measured); now in order, until a stuck shard is 12 rows past its cap. The server requeues the same way while another process's pull waits for the pin, so rows redelivered across a handover or a stop may come back out of order among themselves, though still ahead of newer rows. The unpin on release names no pin id, so it is preceded by a check that the pin is still this process's; the unit renews its pin until the unpin returns (for up to `ack_wait` and one 5-second renewal after its halt has handed on what it fetched, longer than a handover waits), so the pin cannot lapse between the two, and only a consumer leader change there could move it.

Makefile

@@ -744,12 +744,32 @@ install-playwright-docs: pnpm-install
 COV_DEFER ?=
 export COV_DEFER
 
+# UNIT_TIMEOUT: go test's limit for each package of the unit suite. 15s is the
+# budget a unit package has to fit when the suite runs on its own (#617), as
+# it does in CI's unit job.
+UNIT_TIMEOUT ?= 15s
+
+# CI_UNIT_TIMEOUT: the limit `make ci` passes as UNIT_TIMEOUT for its parallel
+# phase. There the unit suite shares every core with the lint and build jobs,
+# where a package that takes 10s alone runs past its 15s budget, and healthy
+# runs were killed.
+CI_UNIT_TIMEOUT ?= 60s
+
+# INTEGRATION_TIMEOUT: go test's limit for each package of the integration
+# suite (internal/mq's tagged tests, run on their own, keep a shorter one).
+# 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. tests/integration takes about 6m on a CI
+# runner and up to 8m under Docker Desktop: at 240s CI killed it, and 480s
+# left Docker Desktop half a minute.
+INTEGRATION_TIMEOUT ?= 900s
+
 .PHONY: test-unit
 test-unit: go-mod-download ## Run Go unit tests + render coverage + gate threshold
 	@printf "$(CYAN)==> Running Unit Tests...$(RESET)\n"
 	@rm -rf $(COV_UNIT)/data && mkdir -p $(COV_UNIT)/data
 	@GOCOVERDIR="$(CURDIR)/$(COV_UNIT)/data" go tool gotestsum --format $(GOTESTSUM_FMT) -- \
-		-tags="$(TAGS)" -cover -race -timeout 15s ./internal/... ./cmd/... $(ARGS) \
+		-tags="$(TAGS)" -cover -race -timeout $(UNIT_TIMEOUT) ./internal/... ./cmd/... $(ARGS) \
 		-args -test.gocoverdir="$(CURDIR)/$(COV_UNIT)/data"
 	@if [ -z "$(COV_DEFER)" ]; then go run ./scripts/cov render unit; fi
 
@@ -762,11 +782,8 @@ 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 \
+		-tags="integration $(TAGS)" -timeout $(INTEGRATION_TIMEOUT) -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
@@ -858,7 +875,7 @@ ci-parallel: verify-parallel build build-cover build-ts build-docs test test-ts
 .PHONY: ci
 ci: ## Full pipeline — parallel checks, then sequential heavy suites + coverage
 	@echo "$(CYAN)==> Phase 1: Parallel Build & Static Checks$(RESET)"
-	@$(MAKE) -j $(JOBS) ci-parallel COV_DEFER=1
+	@$(MAKE) -j $(JOBS) ci-parallel COV_DEFER=1 UNIT_TIMEOUT=$(CI_UNIT_TIMEOUT)
 	@echo "$(CYAN)==> Phase 2: Sequential Heavy Tests$(RESET)"
 	@$(MAKE) test-integration COV_DEFER=1
 	@$(MAKE) test-e2e COV_DEFER=1

docs/src/content/docs/development.md

@@ -333,6 +333,8 @@ Each test target writes `covdata` to `tmp/coverage/<suite>/data/`, renders a tex
 
 **Extra flags**: All test targets accept `ARGS="..."` for additional `go test` flags (e.g., `-run`, `-count`, `-timeout`).
 
+**Time limits**: `make test-unit` gives each unit package 15s (`UNIT_TIMEOUT`), the budget it has to fit when the suite runs on its own, as it does in the unit job in CI. `make ci` gives it 60s (`CI_UNIT_TIMEOUT`) in its parallel phase, where the suite shares every core with the lint and build jobs, so a package over the budget can pass `make ci` and still fail in CI. `tests/integration` gets 900s (`INTEGRATION_TIMEOUT`). Each is a Makefile variable, so a slower machine can raise one for a run: `make ci CI_UNIT_TIMEOUT=90s`.
+
 **Note on timing**: gotestsum's `DONE ... in X.XXXs` reports pure test execution time. The total wall time includes Go compiling all packages — the first run compiles everything (~15s), subsequent runs use the build cache (~1s).
 
 ### Test Structure

internal/cache/redis_integration_test.go

@@ -61,12 +61,7 @@ func noPersistence(hc *container.HostConfig) {
 func startContainer(t *testing.T, req testcontainers.ContainerRequest, port string) (testcontainers.Container, string) {
 	t.Helper()
 	ctx := context.Background()
-	if req.HostConfigModifier == nil {
-		req.HostConfigModifier = noPersistence
-	}
-	req.WaitingFor = wait.ForListeningPort(port).WithStartupTimeout(90 * time.Second)
-	ctr, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: req, Started: true})
-	testcontainers.CleanupContainer(t, ctr)
+	ctr, err := runContainer(ctx, t, req, port)
 	require.NoError(t, err)
 	host, err := ctr.Host(ctx)
 	require.NoError(t, err)
@@ -75,6 +70,19 @@ func startContainer(t *testing.T, req testcontainers.ContainerRequest, port stri
 	return ctr, net.JoinHostPort(host, mapped.Port())
 }
 
+// runContainer starts req and waits for port to listen, leaving a failed
+// start to its caller.
+func runContainer(ctx context.Context, t *testing.T, req testcontainers.ContainerRequest, port string) (testcontainers.Container, error) {
+	t.Helper()
+	if req.HostConfigModifier == nil {
+		req.HostConfigModifier = noPersistence
+	}
+	req.WaitingFor = wait.ForListeningPort(port).WithStartupTimeout(90 * time.Second)
+	ctr, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: req, Started: true})
+	testcontainers.CleanupContainer(t, ctr)
+	return ctr, err
+}
+
 func startStandalone(t *testing.T, image string, cmd ...string) *server {
 	t.Helper()
 	ctr, addr := startContainer(t, testcontainers.ContainerRequest{
@@ -104,23 +112,44 @@ func startDragonfly(t *testing.T) *server {
 // MOVED. The node announces 127.0.0.1 on a host port bound to the same
 // number, so the address the client learns from CLUSTER SLOTS is dialable
 // from the test.
+//
+// The host port is the test's pick, and the host may refuse to publish it:
+// Docker Desktop refuses the ports Windows has reserved, which lie in the
+// range freePort draws from (about one pick in twenty on the machine this
+// was seen on), and any host refuses one taken since freePort let it go. A
+// refused pick is answered with another.
 func startCluster(t *testing.T) *server {
 	t.Helper()
-	port := freePort(t)
-	p := network.MustParsePort(port + "/tcp")
-	ctr, _ := startContainer(t, testcontainers.ContainerRequest{
-		Image: redisImage,
-		Cmd: []string{
-			"redis-server", "--port", port, "--cluster-enabled", "yes", "--cluster-port", "16379",
-			"--cluster-announce-ip", "127.0.0.1", "--save", "", "--appendonly", "no",
-		},
-		ExposedPorts: []string{port + "/tcp"},
-		HostConfigModifier: func(hc *container.HostConfig) {
-			noPersistence(hc)
-			hc.PortBindings = network.PortMap{p: {{HostPort: port}}}
-		},
-	}, port+"/tcp")
-	code, out, err := ctr.Exec(context.Background(), []string{"redis-cli", "-p", port, "cluster", "addslotsrange", "0", "16383"})
+	ctx := context.Background()
+	var (
+		ctr  testcontainers.Container
+		port string
+	)
+	for attempt := 1; ; attempt++ {
+		port = freePort(t)
+		p := network.MustParsePort(port + "/tcp")
+		var err error
+		ctr, err = runContainer(ctx, t, testcontainers.ContainerRequest{
+			Image: redisImage,
+			Cmd: []string{
+				"redis-server", "--port", port, "--cluster-enabled", "yes", "--cluster-port", "16379",
+				"--cluster-announce-ip", "127.0.0.1", "--save", "", "--appendonly", "no",
+			},
+			ExposedPorts: []string{port + "/tcp"},
+			HostConfigModifier: func(hc *container.HostConfig) {
+				noPersistence(hc)
+				hc.PortBindings = network.PortMap{p: {{HostPort: port}}}
+			},
+		}, port+"/tcp")
+		if err == nil {
+			break
+		}
+		if attempt == clusterPortPicks || !portRefused(err) {
+			require.NoError(t, err)
+		}
+		t.Logf("host port %s refused, picking another: %v", port, err)
+	}
+	code, out, err := ctr.Exec(ctx, []string{"redis-cli", "-p", port, "cluster", "addslotsrange", "0", "16383"})
 	require.NoError(t, err)
 	require.Zero(t, code, "%v", out)
 	s := &server{ctr: ctr, addr: net.JoinHostPort("127.0.0.1", port), mode: cache.RedisCluster}
@@ -134,6 +163,21 @@ func startCluster(t *testing.T) *server {
 	return s
 }
 
+// clusterPortPicks is how many host ports startCluster tries before it gives
+// up.
+const clusterPortPicks = 5
+
+// portRefused reports whether a container failed to start because the host
+// would not publish its port: reserved (Docker Desktop's "ports are not
+// available") or taken (the engine's "port is already allocated" and
+// "address already in use").
+func portRefused(err error) bool {
+	msg := err.Error()
+	return strings.Contains(msg, "ports are not available") ||
+		strings.Contains(msg, "port is already allocated") ||
+		strings.Contains(msg, "address already in use")
+}
+
 func freePort(t *testing.T) string {
 	t.Helper()
 	var lc net.ListenConfig