Harden the streaming-ingest tier's bounds and add a kill switch

[jodeleeuw]

This branch includes the merge of PR #228's branch (harden-staging-sweep) to resolve overlapping conflicts during development; its diff will shrink to just this PR's own changes once #228 lands, and this PR should be merged after it.

A review of the built streaming-ingest tier (docs/streaming-ingest-design.md) found that one anonymous POST /api/session bought a capability to force up to 10,000 trials × 64 KiB = 640 MB into RTDB for a single session id, that the sweep's assembleSession read a whole session into memory before its byte cap applied (and compared that cap against UTF-16 .length rather than real bytes), and that nothing bounded how many sessions one experiment id — public by construction — could have open at once. This PR closes each of those, plus adds an operational kill switch.

Rules caps. database.rules.json's $seq pattern is now 1-3 digits (1,000 trials) and the per-trial .length cap is 16384, both hand-copies of MAX_TRIALS_PER_SESSION / MAX_TRIAL_BYTES in a shared constant module, functions/src/staging-assembly.ts, which api-session-start.ts already reads to answer the client. A new __tests__/rules-constants.test.js parses the rules file and asserts its literals still match those constants, so the rules, the endpoint's response, and the docs can't silently drift apart. The rules' own comments now say plainly that RTDB's .length counts UTF-16 code units, not bytes — a trial full of multibyte content can cost up to ~3x as many real bytes for the same .length — and that MAX_ASSEMBLED_BYTES (measured with Buffer.byteLength) is the byte-accurate backstop.

lastFlushAt was newData.isNumber(), letting a client write any timestamp, including one far in the future — which made disconnectedSince() treat every later disconnect stamp as already answered, so such a session could never be swept as abandoned. It now requires newData.val() == now, like the other server-stamped timestamps in the tree.

Bounded assembly read. assembleSession no longer does one .get() of a session's whole trials node. It pages through orderByKey().startAfter(lastKey) via a new pure accumulator, assembleTrialsPaged, which stops asking for another page the moment the accumulated byte size would cross MAX_ASSEMBLED_BYTES — so an over-cap session is truncated without ever being pulled into memory in full. Both assembleTrials and assembleTrialsPaged now cost each trial with Buffer.byteLength instead of .length.

Per-experiment concurrency cap. openSessionCounts/{experimentId} is a transactional counter, incremented in openSession() and decremented in discardSession(), capping an experiment at MAX_OPEN_SESSIONS_PER_EXPERIMENT (500) concurrently open sessions — generously above any lecture-hall or online-panel study DataPipe hosts today. It self-heals negative drift in the transaction itself (a stored value below zero is treated as zero) and the sweep runs reconcileOpenSessionCounts each pass, scoped to that run's candidate experiments, to correct positive drift (a missed decrement). Deliberately not a per-IP counter: pages/docs/privacy.js states DataPipe's own code never reads or stores a participant's IP address, and this cap counts sessions per experiment — the same unit maxSessions already limits — carrying no information about who is opening them. A refusal returns the existing 503 SESSION_START_ERROR shape, which the plugin already falls back from.

Kill switch. STREAMING_ENABLED=false makes POST /api/session return that same 503 shape without touching Firestore or RTDB, checked before anything else in the handler. Default (unset) is enabled, so no existing deployment's behavior changes. Documented as a commented example in functions/.env.datapipe-test.

Docs. pages/docs/api.js's example response and refusal-cause text updated to match (16384/1000, and the new refusal reasons). docs/streaming-ingest-design.md gets a new "Hardening pass" section covering all of the above.

Tests. __tests__/database-rules.test.js: updated caps, new lastFlushAt stale/future/server-time cases, new openSessionCounts read/write denial cases. functions/src/__tests__/staging-assembly.test.js: assembleTrialsPaged paging/truncation/ordering cases (including one asserting the fetcher stops being called once the byte cap is crossed) and streamingEnabled cases. functions/src/__tests__/rules-constants.test.js (new): the JSON-literal-to-constant pin described above. functions/src/__tests__/staging-emulator.test.js: the per-experiment cap (refusal at the cap, admission once a slot frees up via direct discard and via the sweep), the kill switch's enabled-by-default behavior, and a >1-page real RTDB assembly-paging recovery.

Also added, in response to two rounds of emulator-run feedback:

• The first two concurrency-cap tests were rewritten after a real emulator run: "lets a new session in once a prior one completes" now calls discardSession() directly rather than routing through a full /api/data completion, whose pass/fail depended on the mocked provider round trip rather than the slot-release mechanism under test. "lets a new session in once an abandoned one is swept" seeded openSessionCounts with no matching openSessions entries; reconcileOpenSessionCounts correctly recomputes an experiment's counter from the real entries under openSessions on every sweep run (the counter has no legitimate reason to differ from what is actually open), so it rightly overwrote the unsupported seeded value — the fixture was wrong, not the reconcile. A new seedOpenSessions() helper seeds matching real entries, parked a year out on expiresAt so they never crowd a same-run real session out of the sweep's oldest-30 candidate window.
• A new "releases the concurrency-cap slot on a genuine, non-refused completion" test closes the resulting gap: nothing end-to-end proved a real /api/data completion reaches discardSession. Building it surfaced the actual root cause of the original 500-instead-of-499 reading: this file's shared staging-testuser owner fixture sets connectedAccounts.gdrive = { accessToken, refreshToken }, but providers/gdrive.ts's resolveToken() reads .encryptedToken and .tokenExpiresAt — neither of which that fixture sets. Every "completion" attempt against it therefore failed at token resolution, before ever reaching discardStaging — the "DataPipe itself fails" branch, which by design leaves the staged copy and its counter slot in place for the sweep to recover. That was correct, intentional behavior, not a bug; the test believed it was exercising a real completion when it was exercising a token failure. The new test uses its own owner with a token shape resolveToken() can actually resolve, so the request clears every gate and reaches the provider; it gets a 202 rather than a literal 201 (this file has no mock Google Drive server on the fixed GDRIVE_API_BASE port that gdrive-emulator.test.js owns for the whole run), but that queued branch calls discardStaging on the way to responding exactly as a real 201 would, and the test asserts the counter is released afterward.

cd functions && npm run build and npm run lint (repo root) are both clean.

Test plan

• Pure tests (repo root): staging-assembly, rules-constants, concurrency-limit, live-sessions, staging-session-id — all passing
• Emulator suites: database-rules, session-id-validation-emulator, staging-emulator — 95 tests, all passing
• cd functions && npm run build clean
• npm run lint (repo root) clean

🤖 Generated with Claude Code

https://claude.ai/code/session_01Q8xf16ov88M3KojPVfQwJT