test(unit): run internal/app's tests in parallel to fit the unit budget

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.<domain>.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 `"<nil>"` 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 `<data_dir>/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, `<tenant>/<table>/<id>` (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.

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{}
 	}