test(integration): wait for live fan-out in the NATS e2e test

CHANGELOG.md

@@ -114,6 +114,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 - **The multi-tenant docs say how a tenant folder is added by hand, where `X-Tenant-ID` is resolved, and when a stream can see the tenant `404`** (`docs/src/content/docs/{deployment,api}.md`, `docs/src/content/docs/sdk/{reference,streaming,admin}.md`, `docs/src/content/docs/settings-directory.mdx`): story 11 of [#583](https://github.com/Wave-RF/WaveHouse/issues/583). `wavehouse bootstrap` seeds one folder of a nested directory when given the folder's path, and the new folder is adopted by a whole-directory reload, not a reload by name; the API page's Authentication section now says the tenant is resolved from the header before the token is read. The SDK pages said a stream sees the `400`/`404` of tenant resolution only "when you send `X-Tenant-ID`"; a stream that sends none is tenant `0`'s, which over a nested directory with no `0` folder is the `404` ([#641](https://github.com/Wave-RF/WaveHouse/issues/641)). The DLQ example on the SDK admin page declared `data` twice in one block and showed a `0` count the server never returns, the `?table=` row said the filter narrows `total`, and the dead-letter section called the unreadable-envelope exception "new in this release" ([#676](https://github.com/Wave-RF/WaveHouse/issues/676)).
 - **`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.
+- **`TestNATSBackend_EndToEnd` no longer fails now and then with `no event with "a1" within 30s`** (`tests/integration/mq_nats_test.go`), closing [#768](https://github.com/Wave-RF/WaveHouse/issues/768): the likely cause, not reproduced, is that the test opened its two live streams and ingested at once. A stream answers `: connected` before its hub registration, and a process's hub bridge reads the history stream from now on, so an event ingested before either is in place is never delivered to that stream, however long the test waits ([#784](https://github.com/Wave-RF/WaveHouse/issues/784) tracks that in the product). The test now ingests probe rows until both streams have received one, then ingests the rows it checks.
 - **`make` pins the Go toolchain to the one `go.mod` names (its `toolchain` line, else its `go` line), so local runs and CI use the same Go** (`Makefile`, `scripts/ci/go-toolchain{,.test}.sh` (new), `.github/actions/setup-env/action.yml`, `.github/workflows/README.md`, `docs/src/content/docs/development.md`, `AGENTS.md`), closing [#580](https://github.com/Wave-RF/WaveHouse/issues/580): `GOTOOLCHAIN=auto` means the newer of the local Go and `go.mod`'s, so a machine on a newer Go built, tested and linted on a different toolchain than CI, and `make lint-go` panicked there (golangci-lint is built with an older Go and cannot type-check a newer standard library). The Makefile now exports `GOTOOLCHAIN` derived from `go.mod`, and the setup-env action exports the same value to `$GITHUB_ENV` for CI steps that run Go outside `make`, which also keeps a future runner image with a newer Go from taking over. Raising the `go` directive to a new minor version needs a golangci-lint built with that minor or newer (otherwise `make lint-go` refuses to load its config); patch bumps don't. One script, `scripts/ci/go-toolchain.sh`, derives the pin for both the Makefile and CI (with a fixture test in `make verify`): it reads `go.mod`'s `toolchain` line, else its `go` line, and fails with a message unless that is `x.y.z`, so bump with `go get go@1.N.P` rather than `go mod edit -go=1.27`. `GOTOOLCHAIN=local` in the environment is ignored; use `make GOTOOLCHAIN=local ...`.
 - **`make` shows why `pnpm install` failed** (`Makefile`), closing [#578](https://github.com/Wave-RF/WaveHouse/issues/578): the hidden `pnpm-install` target ran with `--reporter=silent`, which printed nothing at all, even on failure (a lockfile that violates the supply-chain policy, or `ERR_PNPM_OUTDATED_LOCKFILE`), leaving a bare `Error 1`. It now runs pnpm's default reporter with the output captured, and replays it when the install fails; a successful install stays quiet. It also passes `--config.confirm-modules-purge=false`, so an incompatible `node_modules` is now reinstalled without asking instead of hanging on a prompt hidden in the captured output, which also fixes the silent `ERR_PNPM_ABORTED_REMOVE_MODULES_DIR_NO_TTY` exit when stdin is not a terminal.
 - **`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.

tests/integration/mq_nats_test.go

@@ -137,19 +137,56 @@ func (p *natsProcess) sse(t *testing.T, id tenant.ID, query string) <-chan strin
 	return lines
 }
 
-// awaitEvent reads lines until a data line contains want.
-func awaitEvent(t *testing.T, lines <-chan string, want string) {
+// sawEvent reads lines until a data line contains want, and reports whether
+// one arrived within d.
+func sawEvent(t *testing.T, lines <-chan string, want string, d time.Duration) bool {
 	t.Helper()
-	deadline := time.After(30 * time.Second)
+	deadline := time.After(d)
 	for {
 		select {
 		case line, ok := <-lines:
 			require.True(t, ok, "the stream ended before an event with %q", want)
 			if strings.HasPrefix(line, "data:") && strings.Contains(line, want) {
-				return
+				return true
 			}
 		case <-deadline:
-			t.Fatalf("no event with %q within 30s", want)
+			return false
+		}
+	}
+}
+
+// awaitEvent fails the test unless a data line containing want arrives.
+func awaitEvent(t *testing.T, lines <-chan string, want string) {
+	t.Helper()
+	if !sawEvent(t, lines, want, 30*time.Second) {
+		t.Fatalf("no event with %q within 30s", want)
+	}
+}
+
+// awaitLiveFanOut returns once every stream has received an event that went
+// through the queue after it opened. It works around #784 (a stream says
+// ": connected" before it is registered for live events, and a hub bridge
+// reads the history stream from now on): once that is fixed, drop the probes
+// and assert that no event is lost after ": connected". Each probe is a row
+// of its own, so the rows the test checks are unaffected.
+func awaitLiveFanOut(t *testing.T, p *natsProcess, streams ...<-chan string) {
+	t.Helper()
+	pending := map[int]bool{}
+	for i := range streams {
+		pending[i] = true
+	}
+	deadline := time.Now().Add(30 * time.Second)
+	for attempt := 0; len(pending) > 0; attempt++ {
+		require.True(t, time.Now().Before(deadline), "%d live stream(s) never received a probe event", len(pending))
+		if attempt > 0 {
+			t.Logf("live fan-out not up after %d probe(s); %d stream(s) still waiting", attempt, len(pending))
+		}
+		status, body := p.do(t, http.MethodPost, "/v1/ingest?table=events", map[string]string{tenant.Header: "acme", "Content-Type": "application/json"}, fmt.Sprintf(`{"id":"probe-%d","page":"probe"}`, attempt))
+		require.Equal(t, http.StatusOK, status, body)
+		for i := range pending {
+			if sawEvent(t, streams[i], `"probe-`, time.Second) {
+				delete(pending, i)
+			}
 		}
 	}
 }
@@ -204,6 +241,7 @@ func TestNATSBackend_EndToEnd(t *testing.T) {
 	since := time.Now().UTC()
 	liveA := a.sse(t, "acme", "table=events")
 	liveB := b.sse(t, "acme", "table=events")
+	awaitLiveFanOut(t, a, liveA, liveB)
 	for id, row := range map[tenant.ID]string{"acme": `{"id":"a1","page":"home"}`, "globex": `{"id":"g1","page":"cart"}`} {
 		status, body := a.do(t, http.MethodPost, "/v1/ingest?table=events", map[string]string{tenant.Header: id.String(), "Content-Type": "application/json"}, row)
 		require.Equal(t, http.StatusOK, status, body)