From 6f7aca47160b94279b7b041b233281a5675c1e2b Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Sat, 10 Oct 2026 13:35:38 -0400 Subject: [PATCH] test(unit): run internal/app's tests in parallel to fit the unit budget internal/app's 77 tests ran one after another and took 6 to 8s of test time alone under -race -cover, against the 15s make test-unit gives each package. Most of that was waiting, not work: the binary used about 2.5s of CPU over 8s of wall time. - TestMain silences the default logger once, and newApp no longer swaps it or the OTel providers per test. guardGlobals stays only in the serial tests that turn Prometheus on or replace the logger themselves. - Every test that leaves process-wide state alone calls t.Parallel, and so do its subtests where they are independent. The tests that capture the logger, turn on Prometheus, set the environment or signal the process stay serial, as does TestNew_NestedDirectory, whose subtests run before the reload it then checks. - Two serial tests waited out a production retry. A moved tenant's schema discovery retried after a random wait of up to 2s (more when an attempt came before the fix), and the DynamoDB table check after 1s. discoveries.backoff and App.dynamoRetry now hold those first waits, at their production values, and each test shortens its own. The package now takes 2.2 to 3.1s of test time alone, from 5.9 to 7.7s interleaved on the same machine, with coverage unchanged at 93.0%. internal/mq already fit after #745 (2.6s alone). Closes #739 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + internal/app/app.go | 5 +- internal/app/app_test.go | 93 +++++++++++++++++++++------- internal/app/dedupe_dynamodb_test.go | 1 + internal/app/discoveries.go | 8 ++- internal/app/main_test.go | 5 +- internal/app/mq_nats_test.go | 10 +-- internal/app/roles_test.go | 9 ++- internal/app/wire_dynamodb.go | 2 +- 9 files changed, 99 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 547d84d8f..e597028f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -146,6 +146,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - **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. - **External NATS: the shipped manifests fit the shipped volume, boot refuses permissions narrower than the shards, and the tooling covers `js_domain`** (`internal/mq/{nats_manifests,nats_topology,external}.go`, `internal/mq/nats_permprobe.go` (new), `internal/mq/natstest/natstest.go`, `cmd/wavehouse/mq.go` (+ tests; `internal/mq/{nats_manifests,external_perms}_test.go` new), `deployments/nats/{jetstream.yaml,values.yaml}`, `docs/src/content/docs/{deployment,architecture}.md`, `configuration.mdx`): part of [#613](https://github.com/Wave-RF/WaveHouse/issues/613). The shipped streams reserved 225 GiB against the shipped 100 Gi volume, whose size the NATS Helm chart makes each server's `max_file_store`, so the server refused the third partition. `wavehouse mq manifests --file-store` (default `100Gi`) now sizes each partition at 15% of it, the history at 10% and the dead-letter stream at 5%, so the shipped four partitions reserve 75 GiB; a partition's size does not follow N, and the generator refuses streams that reserve more than the store (`--partition-max-bytes` resizes them). The test server takes its file store from the shipped values as the chart does, where it had a limit no test could reach. Boot now probes every shard durable as the connecting user, with a pull and an unpin the server rejects on their merits, and a `required` finding names each durable the user may not pull, where permissions generated for fewer shards passed boot and left those shards' rows unpulled; a consumer request the server refuses later sets `wavehouse_mq_topology_ok` to `0` at once. `wavehouse mq permissions --js-domain` allows and denies the `$JS..API` subjects a client in a JetStream domain sends, beside the plain ones a server in the domain checks after mapping them; `wavehouse mq manifests` takes `--ingest-consumer`, `--history-stream` and `--publish-timeout`, the duplicate window following the timeout; the `wavehouse` user may ack only for its shard durables, under either ack subject layout, where it could ack for any consumer; and an `mq.nats.ingest_consumer` or `history_stream` outside `[a-zA-Z0-9_-]` refuses boot. - **The topology tests no longer fail on a nats-server newer than 2.14, and `internal/mq` runs in about a third less time** (`internal/mq/nats_topology_test.go`, `internal/mq/natstest/natstest.go`, `internal/mq/natsspike/*_test.go`): boot adds an advisory finding when the server is off the recommended 2.14 line, which five tests that assert the exact findings list did not expect, so a server upgrade turned the suite red ([#707](https://github.com/Wave-RF/WaveHouse/issues/707)). They now set that one finding aside, and `TestVerifyNATSTopology_ServerVersionFinding` holds it to the pinned server's own version. The `Findings` cases ran one after another on one server (8.8s alone); they now run in parallel across four, and `natstest`'s `Manifests.Create` creates a stream's consumers concurrently, which shortens every test that boots the shipped topology ([#739](https://github.com/Wave-RF/WaveHouse/issues/739)). The `natsspike` servers now empty their store after shutdown, because a deleted stream's directory is removed by a goroutine `Shutdown` does not wait for, which made `t.TempDir`'s cleanup fail with "directory not empty". +- **`internal/app` fits the unit budget with room to spare** (`internal/app/*_test.go`, `internal/app/{app,discoveries,wire_dynamodb}.go` (a retry wait tests can shorten)): closes [#739](https://github.com/Wave-RF/WaveHouse/issues/739). Its tests ran one after another and took 6 to 8s alone under `-race -cover`, of the 15s `make test-unit` gives each package, most of it spent waiting rather than working. `TestMain` now silences the default logger once, so most of the tests run in parallel; the ones that capture the logger, turn on Prometheus, set the environment or signal the process stay serial. Two of those waited out a production retry: a moved tenant's schema discovery (a random wait of up to 2s, at times more) and the DynamoDB table check (1s). Each now shortens it through a field whose production value is unchanged. The package takes about 2 to 3s alone; `internal/mq` already fit after the entry above. - **One tenant's failed queue join no longer ends ingest for every tenant** (`internal/mq/{mq,embedded}.go` (+ tests), `docs/src/content/docs/{architecture,durability,ingest-pipeline}.md`, `settings-directory.mdx`): closes [#675](https://github.com/Wave-RF/WaveHouse/issues/675). When a tenant's queue opened while the server ran and the ingest worker's consumer could not join it, the broker reported that as the consumer's delivery ending, so the worker failed and the process restarted — one tenant's failure taking ingest down for all of them — while the stream hub's consumer failing the same way was only logged, and the tenant's `GET /v1/stream` connections got no live rows until a restart. A queue a consumer cannot join is now left unrecorded as open, whichever consumer it is: `SetMaxBytes` returns the error and `MaxBytes` keeps not reporting the budget, so the next reload tries the consumers that had not joined again, and the tenant's publishes answer `503` and try them too, paced as a queue that cannot open already is (one shared attempt, then refusals for five seconds). No row is accepted that a consumer would not read, no other tenant is touched, and nothing restarts. A queue reopened after its ingest stream went missing has its consumers' delivery started again on the new stream, where the stopped delivery on the old one used to stand in for it. A consumer that had joined and whose delivery then ends on its own (a deleted durable, a closed connection) still fails the worker, and boot is unchanged: a flat directory whose queue cannot open refuses boot. - **Dedupe claims an id, publishes, then commits it — and keys it by tenant, table and id** (`internal/dedupe/{dedupe,key,embedded,managed}.go` (+ tests), `internal/dedupe/dedupetest/` (new), `internal/api/ingest.go` (+ tests), `internal/settings/validate_test.go`, `internal/keyenc/keyenc.go`, `internal/testutil/mocks.go`, `internal/app/app_test.go`, `AGENTS.md`, `docs/src/content/docs/{architecture,api,deployment,development}.md`, `settings-directory.mdx`, `sdk/reference.md`): `CheckAndMark` is replaced by a two-phase `Reserve` → `Commit` / `Release` contract with a lease on the pending claim, and every backend now runs one conformance suite. Four bugs go with it. Two concurrent requests carrying one id no longer both publish it: Pebble's check and claim happen under one lock, and the loser answers `503` with `Retry-After` while the winner is still publishing ([#390](https://github.com/Wave-RF/WaveHouse/issues/390)). A publish that fails gives its id back, so the retry a `503` asks for is published instead of skipped as a duplicate of a record that never reached the queue ([#384](https://github.com/Wave-RF/WaveHouse/issues/384)) — the residual case is a publish that fails after it already reached the broker (a timeout, a disconnect), where the released id lets the retry through but that retry publishes a genuine second copy; [#629](https://github.com/Wave-RF/WaveHouse/pull/629) closes that with an idempotency key. The same id in two tables is two ids ([#222](https://github.com/Wave-RF/WaveHouse/issues/222)'s keyspace half). An explicit `null` id is a missing id — rejected under `require_id`, published un-deduped otherwise — instead of the one id `""` that made every null record after the first a duplicate ([#370](https://github.com/Wave-RF/WaveHouse/issues/370)). **Upgrade:** the key layout changes, so an id seen before the upgrade is accepted once more after it; nothing is migrated, and the old keys are left in `/pebble`, unread, deleted by the retention sweep (see Added) ([#220](https://github.com/Wave-RF/WaveHouse/issues/220)) ([Deployment → Upgrading across the dedupe key change](https://github.com/Wave-RF/WaveHouse/blob/main/docs/src/content/docs/deployment.md#upgrading-across-the-dedupe-key-change)). The key is readable text, `//` (for example `acme/clicks/evt-123`), with the table and id escaped and joined by `internal/keyenc`, the escaping NATS subject tokens already use, so any table name gets a keyspace of its own, including one holding a NUL byte or a `/`. New metrics: `wavehouse_ingest_dedupe_commit_failed_total` (a published record whose id failed to commit; the claim lapses with its lease) and `wavehouse_dedupe_hashed_id_total` (an id over 1,024 bytes once escaped, stored as its SHA-256). - **Ingest runs in windows of 256 records over the dedupe contract, and a dedupe store that cannot answer is a `503`** (`internal/api/ingest.go` (+ tests), `internal/mq/{mq,embedded}.go` (+ tests), `internal/dedupe/key.go` (+ tests), `internal/testutil/mocks.go`, `AGENTS.md`, `docs/src/content/docs/{api,architecture,durability}.md`, `settings-directory.mdx`, `sdk/reference.md`): each window of a request is prepared, then reserved in one dedupe call, published in order, and committed in one call, so a batch costs one dedupe round trip per phase per window rather than per record — on Pebble, one commit `fsync` per window (a 1,000-record batch: four syncs instead of a thousand, 24 ms against 5.7 s of dedupe time measured with the queue stubbed). Every deduped record is published under an idempotency key (`mq.WithIdempotencyKey`, JetStream's message id, derived by `dedupe.IdempotencyKey`), and each tenant's ingest stream now keeps an explicit two-minute duplicate window, sized to `2 × the 30-second lease + 1s`: an uncertain publish's `503` sends the *full* lease as `Retry-After`, so an obedient client's retry can land up to ~2×lease after the original request, and the `+1s` covers a backend whose claim expiry itself rounds up by that much. That closes the last path of [#384](https://github.com/Wave-RF/WaveHouse/issues/384): a publish that fails with an unknown outcome (anything but a full queue) keeps its record's claim until the lease lapses instead of releasing it, and a retry after the lease but within two minutes of the first publish is dropped by the queue if the first copy was stored (a later one is stored again). A dedupe store that is not open or that reports `dedupe.ErrUnavailable` now answers `503 {"error":"dedupe store unavailable"}` with `Retry-After: 5`, which the SDK retries, rather than `500 dedupe failed`. A mid-body read error or dedupe failure now drops the open window unpublished, where records before it used to be published; `wavehouse_ingest_dedupe_commit_failed_total` and `wavehouse_ingest_dedupe_disabled_total` count records, as before, now added a window at a time. diff --git a/internal/app/app.go b/internal/app/app.go index 6762a3030..cd8b0a765 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -105,6 +105,9 @@ type App struct { // and dedupeStats the figures of the one Pebble instance they share. dedup *dedupe.Stores dedupeStats func() map[string]int64 + // dynamoRetry is the DynamoDB table check's first retry wait, doubling + // up to 30s; tests shorten it. + dynamoRetry time.Duration mq mq.Broker cache cache.Cache coord coord.Coordinator @@ -146,7 +149,7 @@ const ( // start in Run. A failure releases whatever was already opened and returns the // error, so the caller never holds a half-built App. func New(ctx context.Context, opts Options) (app *App, err error) { - a := &App{cfg: opts.Config, build: opts.Build, logLevel: opts.LogLevel, listener: opts.Listener} + a := &App{cfg: opts.Config, build: opts.Build, logLevel: opts.LogLevel, listener: opts.Listener, dynamoRetry: time.Second} if a.logLevel == nil { a.logLevel = &slog.LevelVar{} } diff --git a/internal/app/app_test.go b/internal/app/app_test.go index fa8a5441c..a3922b041 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -45,11 +45,12 @@ import ( "github.com/Wave-RF/WaveHouse/internal/testutil/storedir" ) -// None of these tests run in parallel: New installs a process-wide default -// logger and, with Prometheus on, the global OTel providers. Every app boots -// against a ClickHouse address that is guaranteed closed, so the boot-time -// schema discovery fails fast and deterministically (the degraded path) no -// matter what is listening on the developer's :9000. +// A test that touches process-wide state stays serial: the default logger +// (capturing it, or New replacing it with otel logs on), the global OTel +// providers (Prometheus or otel on), the environment, or signals. Every app +// boots against a ClickHouse address that is guaranteed closed, so the +// boot-time schema discovery fails fast and deterministically (the degraded +// path) no matter what is listening on the developer's :9000. // closedAddr returns a 127.0.0.1 address nothing listens on: bind an // ephemeral port, then release it. @@ -124,8 +125,9 @@ func testConfig(t *testing.T, settingsDir string) *config.Config { } } -// guardGlobals restores the process-wide state New may replace: the default -// logger, and the OTel providers when Prometheus/OTLP is on. +// guardGlobals restores the process-wide state New may replace — the default +// logger, and the OTel providers when Prometheus/OTLP is on — for a serial +// test that turns them on or swaps the logger itself. func guardGlobals(t *testing.T) { t.Helper() savedLogger := slog.Default() @@ -138,12 +140,10 @@ func guardGlobals(t *testing.T) { otel.SetTracerProvider(savedTP) otel.SetMeterProvider(savedMP) }) - slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil))) } func newApp(t *testing.T, cfg *config.Config, opts Options) *App { t.Helper() - guardGlobals(t) opts.Config = cfg a, err := New(t.Context(), opts) require.NoError(t, err) @@ -159,6 +159,7 @@ func get(t *testing.T, h http.Handler, path string) *httptest.ResponseRecorder { } func TestNew_DegradedBootServesDiagnostics(t *testing.T) { + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) a := newApp(t, cfg, Options{Build: BuildInfo{Version: "1.2.3", GitCommit: "abc", BuildTime: "now"}}) @@ -182,6 +183,7 @@ func TestNew_DegradedBootServesDiagnostics(t *testing.T) { // answer — boot is degraded without ClickHouse — so it proves the tenant // resolved. func TestNew_TenantHeaderResolvesAgainstTheRegistry(t *testing.T) { + t.Parallel() a := newApp(t, testConfig(t, writeSettings(t, nil)), Options{}) tests := []struct { @@ -196,6 +198,7 @@ func TestNew_TenantHeaderResolvesAgainstTheRegistry(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, tt.path, nil) if tt.header != "" { req.Header.Set(tenant.Header, tt.header) @@ -220,6 +223,7 @@ func TestAsyncGetters_RegistryMiss(t *testing.T) { } func TestNew_DedupeFollowsSettings(t *testing.T) { + t.Parallel() tests := []struct { name string enabled bool @@ -229,6 +233,7 @@ func TestNew_DedupeFollowsSettings(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() dir := writeSettings(t, map[string]any{"dedupe": map[string]any{ "enabled": tt.enabled, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}, }}) @@ -252,6 +257,7 @@ func rewriteSettings(t *testing.T, dir string, patch map[string]any) { } func TestReload_DrivesTheRegisteredHooks(t *testing.T) { + t.Parallel() // Hooks are registered in New and fired by the reload triggers Run // starts; a direct Reload stands in for any of the three triggers and // pins that the relocated hooks still follow the adopted document. @@ -353,6 +359,7 @@ func TestNew_NestedDirectory(t *testing.T) { // admin token cannot — over a nested directory the ops routes reach every // tenant, so the operator key alone opens them. func TestNew_NestedOperatorReloadsOneTenant(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "broken": invalidQuery}) cfg := testConfig(t, root) cfg.Auth.OperatorKey = "unit-test-operator-key" @@ -427,6 +434,7 @@ func TestNew_NestedWithoutAnOperatorKeyWarnsTheOpsTreeIsClosed(t *testing.T) { // request, so a lost 0 folder is felt at once on the routes that read tenant // 0's list. func TestReload_NestedHooksFollowEachTenant(t *testing.T) { + t.Parallel() dedupeOn := map[string]any{"enabled": true, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}} grown := map[string]any{"dedupe": dedupeOn, "mq": map[string]any{"max_bytes_gb": 2}} root := writeNestedSettings(t, map[string]map[string]any{ @@ -500,6 +508,7 @@ func TestReload_NestedHooksFollowEachTenant(t *testing.T) { // reopened over the same seen ids when the folder is back. The instance is // open while some tenant's store is, and Close releases it. func TestNew_NestedDedupeStoreFollowsEachTenant(t *testing.T) { + t.Parallel() dedupeOn := map[string]any{"dedupe": map[string]any{"enabled": true, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}}} root := writeNestedSettings(t, map[string]map[string]any{"acme": dedupeOn, "globex": nil, "broken": invalidQuery}) cfg := testConfig(t, root) @@ -562,6 +571,7 @@ func TestNew_NestedDedupeStoreFollowsEachTenant(t *testing.T) { // is reached only by a Config built by hand; it must refuse boot, not wire // nothing. func TestNew_RefusesALayerWithoutABackend(t *testing.T) { + t.Parallel() for _, tc := range []struct { key string unset func(*config.Config) @@ -572,7 +582,7 @@ func TestNew_RefusesALayerWithoutABackend(t *testing.T) { {"coord.backend", func(c *config.Config) { c.Coord.Backend = "" }}, } { t.Run(tc.key, func(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) tc.unset(cfg) _, err := New(t.Context(), Options{Config: cfg}) @@ -587,6 +597,7 @@ func TestNew_RefusesALayerWithoutABackend(t *testing.T) { // instance — their ingest answers 503 until a reload or a restart opens it — // while the process, and every tenant with dedupe off, carries on. func TestNew_DedupeOpenFailure(t *testing.T) { + t.Parallel() dedupeOn := map[string]any{"dedupe": map[string]any{"enabled": true, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}}} // A regular file where the instance's directory should be is what Pebble // refuses to open. @@ -595,13 +606,14 @@ func TestNew_DedupeOpenFailure(t *testing.T) { require.NoError(t, os.WriteFile(filepath.Join(dataDir, "pebble"), nil, 0o600)) } t.Run("flat refuses boot", func(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, writeSettings(t, dedupeOn)) block(t, cfg.DataDir) _, err := New(t.Context(), Options{Config: cfg}) require.ErrorContains(t, err, "dedupe open") }) t.Run("nested fails closed", func(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": dedupeOn, "globex": dedupeOn, "initech": nil}) cfg := testConfig(t, root) block(t, cfg.DataDir) @@ -625,6 +637,7 @@ func TestNew_DedupeOpenFailure(t *testing.T) { // force the failure, as TestNew_DedupeOpenFailure does Pebble's. The failed // open clears it, so the next publish opens the queue: each one tries again. func TestNew_QueueOpenFailure(t *testing.T) { + t.Parallel() block := func(t *testing.T, dataDir, stream string) { t.Helper() p := filepath.Join(dataDir, "nats", "jetstream", "$G", "streams", stream) @@ -632,13 +645,14 @@ func TestNew_QueueOpenFailure(t *testing.T) { require.NoError(t, os.WriteFile(p, nil, 0o600)) } t.Run("flat refuses boot", func(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) block(t, cfg.DataDir, "DLQ_0") _, err := New(t.Context(), Options{Config: cfg}) require.ErrorContains(t, err, "mq open") }) t.Run("nested costs the tenant alone", func(t *testing.T) { + t.Parallel() // globex, not acme: opened first, acme's streams keep the streams // directory occupied through globex's failed open, which the server // would otherwise remove on a goroutine of its own while the next @@ -657,7 +671,7 @@ func TestNew_QueueOpenFailure(t *testing.T) { // Boot opens each served tenant's queue under New's context, as New's doc // says: a stop signaled during boot is not held up by one open per tenant. func TestNew_QueueSetupHonorsTheBootContext(t *testing.T) { - guardGlobals(t) + t.Parallel() ctx, cancel := context.WithCancel(t.Context()) cancel() _, err := New(ctx, Options{Config: testConfig(t, writeSettings(t, nil))}) @@ -704,6 +718,7 @@ func TestSharedTables_InvalidatesTheTenantsSharingTheTables(t *testing.T) { // the wiring orphans its cache as it comes back; a tenant that stayed is // never touched, and a reload that changes nothing bumps nobody. func TestReload_ReadmittedTenantCacheIsOrphaned(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil}) a := newApp(t, testConfig(t, root), Options{}) // The hooks read a.cache at reload time: a recording cache from here on. @@ -755,6 +770,7 @@ func (p *pruneRecorder) last() map[tenant.ID]bool { // Every reload prunes the cache's version index down to the tenants served, // so a tenant rejected or removed stops holding it (#262). func TestReload_PrunesCacheIndexToServedTenants(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil}) a := newApp(t, testConfig(t, root), Options{}) _, ok := a.cache.(pruner) @@ -792,6 +808,7 @@ func redisTestConfig(t *testing.T, settingsDir, addr string) *config.Config { // reached does not refuse boot: the cache starts bypassed, and the reload // hook that prunes an in-process index leaves it alone. func TestNew_RedisCacheBootsBypassedWhenUnreachable(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil}) a := newApp(t, redisTestConfig(t, root, closedAddr(t)), Options{}) _, ok := a.cache.(*cache.RedisCache) @@ -810,7 +827,7 @@ func TestNew_RedisCacheBootsBypassedWhenUnreachable(t *testing.T) { // A TLS file that went missing between validation and wiring refuses boot, // naming the key. func TestNew_RedisCacheRefusesAnUnreadableTLSFile(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := redisTestConfig(t, writeSettings(t, nil), closedAddr(t)) cfg.Cache.Redis.TLS = config.CacheRedisTLS{Enabled: true, CAFile: filepath.Join(t.TempDir(), "gone.pem")} _, err := New(t.Context(), Options{Config: cfg}) @@ -912,7 +929,7 @@ func nudges(sub *stream.Subscriber) int { // where it was due, and a reload of a tenant's own interval reshapes its // wheel in place with its open stream still on it. func TestKeepalives_EachTenantAtItsOwnInterval(t *testing.T) { - guardGlobals(t) + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": keepalive(30, 3), "globex": keepalive(10, 2)}) synctest.Test(t, func(t *testing.T) { tenants, k := hookedKeepalives(t, root) @@ -969,7 +986,7 @@ func TestKeepalives_EachTenantAtItsOwnInterval(t *testing.T) { // directory's interval, its wheel reshaped in place by a reload that changes // the pair, and left as it is by a reload the directory rejects. func TestKeepalives_FlatDirectory(t *testing.T) { - guardGlobals(t) + t.Parallel() dir := writeSettings(t, keepalive(30, 3)) synctest.Test(t, func(t *testing.T) { tenants, k := hookedKeepalives(t, dir) @@ -1006,7 +1023,7 @@ func TestKeepalives_FlatDirectory(t *testing.T) { // over on another when it is served again; the component's stop leaves no // wheel turning, and none starts after it. func TestKeepalives_FollowTheTenantsServed(t *testing.T) { - guardGlobals(t) + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": keepalive(5, 1), "globex": keepalive(5, 1)}) synctest.Test(t, func(t *testing.T) { tenants, k := hookedKeepalives(t, root) @@ -1074,7 +1091,7 @@ func TestKeepalives_FollowTheTenantsServed(t *testing.T) { // and some of them wait on a queue or a cache. It is handed a wheel all the // same, the one the hook then keeps. func TestKeepalives_LookupAheadOfTheReloadHook(t *testing.T) { - guardGlobals(t) + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": keepalive(30, 3)}) tenants, findings := settings.Open(root) require.NotNil(t, tenants, "findings: %v", findings) @@ -1160,6 +1177,7 @@ func turningWheel(t *testing.T, a *App, id tenant.ID) *tenantWheel { // none ever arrives on one whose tenant asks for a day, which only another // tenant's cadence could send. func TestRun_KeepsEachTenantsStreamsAliveAtItsOwnInterval(t *testing.T) { + t.Parallel() const day = 24 * 60 * 60 root := writeNestedSettings(t, map[string]map[string]any{"acme": keepalive(2, 1), "globex": keepalive(day, 1)}) var lc net.ListenConfig @@ -1234,19 +1252,21 @@ func gapWindow(minutes int) map[string]any { // (mq.Purger.PurgeAcked). A flat directory's single tenant gets exactly its // own window. func TestGapWindows(t *testing.T) { + t.Parallel() open := func(t *testing.T, dir string) *settings.Registry { t.Helper() - guardGlobals(t) tenants, findings := settings.Open(dir) require.NotNil(t, tenants, "findings: %v", findings) return tenants } t.Run("flat directory", func(t *testing.T) { + t.Parallel() assert.Equal(t, map[tenant.ID]time.Duration{tenant.Default: 45 * time.Minute}, gapWindows(open(t, writeSettings(t, gapWindow(45))))) }) t.Run("nested directory", func(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": gapWindow(15), "globex": gapWindow(60), "initech": gapWindow(30)}) tenants := open(t, root) assert.Equal(t, map[tenant.ID]time.Duration{"acme": 15 * time.Minute, "globex": 60 * time.Minute, "initech": 30 * time.Minute}, gapWindows(tenants)) @@ -1263,6 +1283,7 @@ func TestGapWindows(t *testing.T) { }) t.Run("a folder rejected since boot keeps everything", func(t *testing.T) { + t.Parallel() // globex is what lets boot open the directory at all. root := writeNestedSettings(t, map[string]map[string]any{"acme": invalidQuery, "globex": gapWindow(30)}) tenants := open(t, root) @@ -1278,7 +1299,7 @@ func TestGapWindows(t *testing.T) { // A finding about a nested directory itself — a loose file beside the tenant // folders — refuses boot, like an invalid flat directory. func TestNew_NestedLooseFileRefusesBoot(t *testing.T) { - guardGlobals(t) + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil}) require.NoError(t, os.WriteFile(filepath.Join(root, "notes.txt"), []byte("scratch"), 0o600)) a, err := New(t.Context(), Options{Config: testConfig(t, root)}) @@ -1288,7 +1309,7 @@ func TestNew_NestedLooseFileRefusesBoot(t *testing.T) { } func TestNew_RefusesInvalidSettingsDirectory(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, t.TempDir()) // empty: every required file is missing a, err := New(t.Context(), Options{Config: cfg}) require.Error(t, err) @@ -1347,7 +1368,7 @@ func analystPipe(t *testing.T, dir string) { // place) once the dedupe store is already open, and a second New on the same // data_dir must find the Pebble lock released. func TestNew_LateBootFailureReleasesEverything(t *testing.T) { - guardGlobals(t) + t.Parallel() dir := writeSettings(t, map[string]any{"dedupe": map[string]any{ "enabled": true, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}, }}) @@ -1372,6 +1393,7 @@ func TestNew_LateBootFailureReleasesEverything(t *testing.T) { // than evaluated under the default_role, and the process serves everything // else. func TestNew_UnreachableJWKSBootsFailClosed(t *testing.T) { + t.Parallel() dir := writeSettings(t, authPatch("http://"+closedAddr(t)+"/jwks.json")) analystPipe(t, dir) started := time.Now() @@ -1402,6 +1424,7 @@ func TestNew_UnreachableJWKSBootsFailClosed(t *testing.T) { // is refused under globex's header, and pointing acme's folder at another // provider and reloading it swaps acme's verifier alone. func TestNew_VerifierPerTenant(t *testing.T) { + t.Parallel() acme, acmeKey, _ := jwksServer(t, "acme-1") globex, globexKey, globexFetches := jwksServer(t, "globex-1") root := writeNestedSettings(t, map[string]map[string]any{ @@ -1478,6 +1501,7 @@ func TestNew_VerifierPerTenant(t *testing.T) { } func TestNew_PrometheusInlineMountsOnRouter(t *testing.T) { + guardGlobals(t) cfg := testConfig(t, writeSettings(t, nil)) cfg.Prometheus = config.Prometheus{Enabled: true, Path: "/metrics"} a := newApp(t, cfg, Options{}) @@ -1550,6 +1574,7 @@ func TestRun_ServesUntilCancelled(t *testing.T) { } func TestRun_SweeperRunsUnderItsLease(t *testing.T) { + t.Parallel() var lc net.ListenConfig ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) @@ -1572,6 +1597,7 @@ func TestRun_SweeperRunsUnderItsLease(t *testing.T) { } func TestRun_PrometheusSidecar(t *testing.T) { + guardGlobals(t) var lc net.ListenConfig ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) @@ -1604,6 +1630,7 @@ func TestRun_PrometheusSidecar(t *testing.T) { } func TestRun_ListenFailureStopsEverything(t *testing.T) { + t.Parallel() // Hold the wildcard address the server binds (":port"), not loopback: a // process already on the port holds the same address, and macOS allows a // wildcard bind while only 127.0.0.1:port is held (Linux refuses both), @@ -1651,6 +1678,7 @@ func (c dyingConsumer) Consume(func(*mq.Message), int) (func(), <-chan error, er } func TestRun_DeadIngestWorkerStopsEverything(t *testing.T) { + t.Parallel() a := newApp(t, testConfig(t, writeSettings(t, nil)), Options{}) // The worker takes its consumer from a.mq when Run starts it. a.mq = dyingConsumerBroker{Broker: a.mq, reason: fmt.Errorf("%w: consumer deleted", mq.ErrDeliveryEnded)} @@ -1666,6 +1694,7 @@ func TestRun_DeadIngestWorkerStopsEverything(t *testing.T) { } func TestClose_AbandonsAStuckCloseAtTheDeadline(t *testing.T) { + t.Parallel() release := make(chan struct{}) t.Cleanup(func() { close(release) }) a := &App{} @@ -1738,6 +1767,7 @@ func endsCleanly(t *testing.T, stream io.Reader) { // reconnect then meets the tenant's 503 or 404. A flat directory never stops // serving tenant 0, so a reload it rejects leaves the stream open. func TestRun_StopEndsOpenStreams(t *testing.T) { + t.Parallel() start := func(t *testing.T, settingsDir string) (a *App, baseURL string, stop func() error) { t.Helper() var lc net.ListenConfig @@ -1756,6 +1786,7 @@ func TestRun_StopEndsOpenStreams(t *testing.T) { } t.Run("the stop", func(t *testing.T) { + t.Parallel() _, baseURL, stop := start(t, writeSettings(t, nil)) resp := openStream(t, baseURL, "") started := time.Now() @@ -1765,6 +1796,7 @@ func TestRun_StopEndsOpenStreams(t *testing.T) { }) t.Run("a reload that stops serving the tenant", func(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil}) a, baseURL, stop := start(t, root) defer func() { assert.NoError(t, stop()) }() @@ -1796,6 +1828,7 @@ func TestRun_StopEndsOpenStreams(t *testing.T) { }) t.Run("a reload the flat directory rejects", func(t *testing.T) { + t.Parallel() dir := writeSettings(t, nil) a, baseURL, stop := start(t, dir) resp := openStream(t, baseURL, "") @@ -1825,7 +1858,7 @@ func poolSettings(addr string, open int) map[string]any { // config and the settings pool must fit under it, so an impossible pair // refuses to boot naming both numbers (#530). func TestNew_RefusesAPoolAboveTheCeiling(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, writeSettings(t, poolSettings(closedAddr(t), 10))) cfg.ClickHouse.MaxTotalConns = 4 _, err := New(t.Context(), Options{Config: cfg}) @@ -1879,6 +1912,7 @@ func chSettings(addr, user string, open int) map[string]any { // its own and leaves the other on the very same Manager, resized to its own // ask — the worked example of the story. func TestNew_NestedPoolsFollowEachTenantsTuple(t *testing.T) { + t.Parallel() shared, other := closedAddr(t), closedAddr(t) root := writeNestedSettings(t, map[string]map[string]any{ "acme": chSettings(shared, "default", 10), @@ -1911,7 +1945,7 @@ func TestNew_NestedPoolsFollowEachTenantsTuple(t *testing.T) { // A nested directory's pools must fit the ceiling together: boot is refused // naming the sum and the ceiling, like a flat directory's one pool. func TestNew_NestedRefusesPoolsAboveTheCeiling(t *testing.T) { - guardGlobals(t) + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{ "acme": chSettings(closedAddr(t), "default", 10), "globex": chSettings(closedAddr(t), "default", 10), @@ -1977,6 +2011,7 @@ func TestReload_CeilingRefusesAThirdTupleThenOpensIt(t *testing.T) { // fresh pool, registry and verifier, and an id it sent before the removal is // still a duplicate. Its open streams end too: TestRun_StopEndsOpenStreams. func TestReload_TenantGoneReleasesItsPoolAndRegistry(t *testing.T) { + t.Parallel() jwks, _, fetches := jwksServer(t, "acme-1") acmeSettings := authPatch(jwks.URL) acmeSettings["dedupe"] = map[string]any{"enabled": true, "id_field": "event_id", "require_id": false, "retention": "0", "tables": map[string]any{}} @@ -2057,6 +2092,7 @@ func TestReload_TenantGoneReleasesItsPoolAndRegistry(t *testing.T) { // another tenant's discovery does afterwards; /readyz then pings every open // pool and names each one that does not answer. func TestNew_NestedProbesFollowTheFirstTenantToLoad(t *testing.T) { + t.Parallel() root := writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil}) a := newApp(t, testConfig(t, root), Options{}) @@ -2093,6 +2129,7 @@ func TestNew_NestedProbesFollowTheFirstTenantToLoad(t *testing.T) { // Retry-After, not a 404: in a flat directory during the degraded boot, and // in a nested one per tenant. func TestNew_SchemaNotLoadedIs503(t *testing.T) { + t.Parallel() ingest := func(t *testing.T, a *App, id string) *httptest.ResponseRecorder { t.Helper() req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/v1/ingest?table=clicks", strings.NewReader(`{"page": "/"}`)) @@ -2105,6 +2142,7 @@ func TestNew_SchemaNotLoadedIs503(t *testing.T) { return rec } t.Run("flat, degraded boot", func(t *testing.T) { + t.Parallel() a := newApp(t, testConfig(t, writeSettings(t, nil)), Options{}) rec := ingest(t, a, "") assert.Equal(t, http.StatusServiceUnavailable, rec.Code, "body: %s", rec.Body.String()) @@ -2112,6 +2150,7 @@ func TestNew_SchemaNotLoadedIs503(t *testing.T) { assert.Contains(t, rec.Body.String(), "schema not loaded yet") }) t.Run("nested, per tenant", func(t *testing.T) { + t.Parallel() a := newApp(t, testConfig(t, writeNestedSettings(t, map[string]map[string]any{"acme": nil})), Options{}) rec := ingest(t, a, "acme") assert.Equal(t, http.StatusServiceUnavailable, rec.Code, "body: %s", rec.Body.String()) @@ -2192,6 +2231,7 @@ func databaseSettings(addr, database string) map[string]any { // schema.refresh_interval (60 seconds here). The tenant beside it, and a // reload that moves nobody, keep the registry and the loop they had. func TestReload_MovedTenantDiscoversTheNewDatabase(t *testing.T) { + t.Parallel() addr := closedAddr(t) root := writeNestedSettings(t, map[string]map[string]any{ "acme": databaseSettings(addr, "default"), @@ -2233,6 +2273,7 @@ func TestReload_MovedTenantDiscoversTheNewDatabase(t *testing.T) { // A flat directory's tenant 0 moves the same way. func TestReload_MovedTenantDiscoversTheNewDatabase_Flat(t *testing.T) { + t.Parallel() addr := closedAddr(t) dir := writeSettings(t, databaseSettings(addr, "default")) a := newApp(t, testConfig(t, dir), Options{}) @@ -2260,6 +2301,7 @@ func TestReload_MovedTenantFailedDiscoveryIsRetried(t *testing.T) { root := writeNestedSettings(t, map[string]map[string]any{"acme": databaseSettings(addr, "default")}) a := newApp(t, testConfig(t, root), Options{}) fake := newFakeClickHouse(t, a, map[string][]string{"default": {"events"}}) + a.discoveries.backoff = 10 * time.Millisecond logs := logtest.Capture(t, slog.LevelWarn) // No fake names moved_db: its discovery dials the closed port. @@ -2309,6 +2351,7 @@ func (c *registryAtInvalidation) InvalidateTenant(ctx context.Context, id tenant // pool, and a request arriving meanwhile must not find the previous // database's schema. func TestReload_MovedTenantRegistryIsDroppedBeforeTheCacheInvalidation(t *testing.T) { + t.Parallel() addr := closedAddr(t) root := writeNestedSettings(t, map[string]map[string]any{"acme": databaseSettings(addr, "default")}) a := newApp(t, testConfig(t, root), Options{}) @@ -2347,6 +2390,7 @@ func (c *stuckConn) Query(context.Context, string, ...any) (driver.Rows, error) // the reload returns: Close waits for it with the loops of the served // tenants, and names its tenant when the release budget ends first. func TestClose_WaitsForALoopAReloadStopped(t *testing.T) { + t.Parallel() conn := &stuckConn{entered: make(chan struct{}), release: make(chan struct{})} // Released on every way out, so a failed assertion leaves no loop stuck. release := sync.OnceFunc(func() { close(conn.release) }) @@ -2377,6 +2421,7 @@ func TestClose_WaitsForALoopAReloadStopped(t *testing.T) { // Close stops every tenant's discovery loop within the release budget, and // the pools after them. func TestClose_StopsTheDiscoveryLoops(t *testing.T) { + t.Parallel() a := newApp(t, testConfig(t, writeNestedSettings(t, map[string]map[string]any{"acme": nil, "globex": nil})), Options{}) loops := *a.discoveries.cur.Load() require.Len(t, loops, 2) diff --git a/internal/app/dedupe_dynamodb_test.go b/internal/app/dedupe_dynamodb_test.go index e9dd5290c..56e447ed3 100644 --- a/internal/app/dedupe_dynamodb_test.go +++ b/internal/app/dedupe_dynamodb_test.go @@ -344,6 +344,7 @@ func TestRun_DynamoDBDedupeRetriesTheTableCheck(t *testing.T) { ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) a := newApp(t, cfg, Options{Listener: ln}) + a.dynamoRetry = 10 * time.Millisecond acme := a.dedup.For("acme") require.False(t, acme.Open()) diff --git a/internal/app/discoveries.go b/internal/app/discoveries.go index f056b470b..4cf189801 100644 --- a/internal/app/discoveries.go +++ b/internal/app/discoveries.go @@ -37,6 +37,9 @@ type discoveries struct { // and onLoaded the first success: what /livez is driven by. onAttempt func(tenant.ID, error) onLoaded func(tenant.ID) + // backoff is the boot retry's first backoff, doubling up to 60s; tests + // shorten it. + backoff time.Duration mu sync.Mutex // serializes reconcile, drop, adopt and close cur atomic.Pointer[map[tenant.ID]*tenantDiscovery] @@ -55,7 +58,7 @@ type tenantDiscovery struct { } func newDiscoveries(ctx context.Context, build func(tenant.ID, *settings.Store) *discovery.SchemaRegistry, onAttempt func(tenant.ID, error), onLoaded func(tenant.ID)) *discoveries { - d := &discoveries{ctx: ctx, build: build, onAttempt: onAttempt, onLoaded: onLoaded} + d := &discoveries{ctx: ctx, build: build, onAttempt: onAttempt, onLoaded: onLoaded, backoff: 2 * time.Second} d.cur.Store(&map[tenant.ID]*tenantDiscovery{}) return d } @@ -139,10 +142,11 @@ func (d *discoveries) adopt(id tenant.ID, reg *discovery.SchemaRegistry) { func (d *discoveries) start(id tenant.ID, reg *discovery.SchemaRegistry) *tenantDiscovery { ctx, cancel := context.WithCancel(d.ctx) //nolint:gosec // G118: held on the tenantDiscovery, called by reconcile or close td := &tenantDiscovery{id: id, registry: reg, cancel: cancel, done: make(chan struct{})} + backoff := d.backoff go func() { defer close(td.done) if !reg.Loaded() { - err := reg.RetryRefresh(ctx, 2*time.Second, 60*time.Second, func(err error) { d.onAttempt(id, err) }) + err := reg.RetryRefresh(ctx, backoff, 60*time.Second, func(err error) { d.onAttempt(id, err) }) if err != nil { // ctx cancelled before success — the process is stopping, or // the tenant is no longer served. diff --git a/internal/app/main_test.go b/internal/app/main_test.go index 2f21403b0..731c4ef91 100644 --- a/internal/app/main_test.go +++ b/internal/app/main_test.go @@ -4,11 +4,14 @@ import ( "testing" "github.com/Wave-RF/WaveHouse/internal/mq" + "github.com/Wave-RF/WaveHouse/internal/testutil/logtest" ) // TestMain turns off the embedded broker's fsync per write, which every boot -// here pays for opening its queues. +// here pays for opening its queues, and silences the default logger once, so +// the tests that leave it alone can run in parallel. func TestMain(m *testing.M) { mq.EmbeddedSyncAlways = false + logtest.Silence() m.Run() } diff --git a/internal/app/mq_nats_test.go b/internal/app/mq_nats_test.go index ba5199937..66d30e96d 100644 --- a/internal/app/mq_nats_test.go +++ b/internal/app/mq_nats_test.go @@ -38,6 +38,7 @@ func natsConfig(t *testing.T, url string) *config.Config { // consumes the operator's durable; the operator deleting it ends the worker, // and with it Run, naming the component. func TestNew_NATSBackend(t *testing.T) { + t.Parallel() srv := natstest.Start(t) cfg := natsConfig(t, srv.URL()) cfg.Roles = []config.Role{config.RoleAPI, config.RoleIngest} @@ -81,9 +82,8 @@ func TestNew_NATSWiresNoSweeper(t *testing.T) { //nolint:paralleltest // capture a := newApp(t, cfg, Options{}) assert.NotContains(t, componentNames(a), "sweeper") - // New's observability wiring replaced the default logger; capture after. - // The boot check already warned for this window, so reloading it again - // warns nobody: only a changed window does. + // Captured after New: the boot check already warned for this window, so + // reloading it again warns nobody: only a changed window does. logs := logtest.Capture(t, slog.LevelWarn) warned := func() int { return strings.Count(logs.String(), "keeps less than this tenant's gap window") } _, adopted := a.tenants.Reload("test") @@ -99,7 +99,7 @@ func TestNew_NATSWiresNoSweeper(t *testing.T) { //nolint:paralleltest // capture // A cluster never reached within topology_wait refuses boot as unavailable. func TestNew_NATSUnreachable(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := natsConfig(t, "nats://"+closedAddr(t)) cfg.MQ.NATS.TopologyWait = time.Millisecond _, err := New(t.Context(), Options{Config: cfg}) @@ -109,9 +109,9 @@ func TestNew_NATSUnreachable(t *testing.T) { // The operator's topology missing a piece refuses boot with the finding. func TestNew_NATSTopologyMissing(t *testing.T) { + t.Parallel() srv := natstest.Start(t) require.NoError(t, srv.Operator.JetStream().DeleteStream(t.Context(), "WH_DLQ")) - guardGlobals(t) cfg := natsConfig(t, srv.URL()) cfg.MQ.NATS.TopologyWait = time.Millisecond _, err := New(t.Context(), Options{Config: cfg}) diff --git a/internal/app/roles_test.go b/internal/app/roles_test.go index f3efd4c03..4b9fa4c75 100644 --- a/internal/app/roles_test.go +++ b/internal/app/roles_test.go @@ -26,6 +26,7 @@ import ( // process's. New does not validate, so the embedded MQ stands in for the // shared one a split needs (config.Validate refuses it outside tests). func TestNew_RolesChooseTheComponents(t *testing.T) { + t.Parallel() for _, tc := range []struct { name string roles []config.Role @@ -58,6 +59,7 @@ func TestNew_RolesChooseTheComponents(t *testing.T) { }}, } { t.Run(tc.name, func(t *testing.T) { + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) cfg.Roles = tc.roles a := newApp(t, cfg, Options{}) @@ -67,7 +69,7 @@ func TestNew_RolesChooseTheComponents(t *testing.T) { } func TestNew_RefusesAConfigWithoutRoles(t *testing.T) { - guardGlobals(t) + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) cfg.Roles = nil _, err := New(t.Context(), Options{Config: cfg}) @@ -88,6 +90,7 @@ func hs256(t *testing.T, secret, role string) string { // token verifier runs there, so even an admin token the API would admit is // refused. Every tenant route, and the rest of /v1/ops, is not there. func TestNew_OpsOnlyRouter(t *testing.T) { + t.Parallel() dir := writeSettings(t, nil) require.NoError(t, os.WriteFile(filepath.Join(dir, settings.FileRoles), []byte(`{"roles": ["admin"]}`), 0o600)) require.NoError(t, os.WriteFile(filepath.Join(dir, settings.FilePolicies), []byte(`{"admin_role": "admin", "tables": {}}`), 0o600)) @@ -147,6 +150,7 @@ func TestNew_OpsOnlyRouter(t *testing.T) { // ClickHouse pool answers (here none can), a sweeper-only one once booted. // Liveness never waits on schema discovery, which only the API runs. func TestNew_OpsOnlyReadiness(t *testing.T) { + t.Parallel() cfg := testConfig(t, writeSettings(t, nil)) cfg.Roles = []config.Role{config.RoleIngest} a := newApp(t, cfg, Options{}) @@ -159,6 +163,7 @@ func TestNew_OpsOnlyReadiness(t *testing.T) { // A reload that moves the tenant repoints the pool of a process without the // api role, which has no schema registry to start over. func TestReload_OpsOnlyMovedTenant(t *testing.T) { + t.Parallel() addr := closedAddr(t) dir := writeSettings(t, databaseSettings(addr, "default")) cfg := testConfig(t, dir) @@ -173,6 +178,7 @@ func TestReload_OpsOnlyMovedTenant(t *testing.T) { } func TestNew_OpsOnlyPrometheusInline(t *testing.T) { + guardGlobals(t) cfg := testConfig(t, writeSettings(t, nil)) cfg.Roles = []config.Role{config.RoleIngest} cfg.Prometheus = config.Prometheus{Enabled: true, Path: "/metrics"} @@ -185,6 +191,7 @@ func TestNew_OpsOnlyPrometheusInline(t *testing.T) { // A sweeper-only process serves its listener and runs the sweeper under the // lease, as the all-roles one does. func TestRun_SweeperOnlyProcess(t *testing.T) { + t.Parallel() var lc net.ListenConfig ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) diff --git a/internal/app/wire_dynamodb.go b/internal/app/wire_dynamodb.go index 68f68ce45..9111e31d3 100644 --- a/internal/app/wire_dynamodb.go +++ b/internal/app/wire_dynamodb.go @@ -108,7 +108,7 @@ func (a *App) wireDynamoDedupe(ctx context.Context) error { "table", c.Table, "error", err) } a.add(component{name: "dedupe table check", run: func(ctx context.Context) error { - for wait := time.Second; ready() != nil; wait = min(2*wait, 30*time.Second) { + for wait := a.dynamoRetry; ready() != nil; wait = min(2*wait, 30*time.Second) { select { case <-ctx.Done(): return nil