Every view kind's output is pinned, and output may not depend on which query finished first

CLAUDE.md

@@ -695,6 +695,16 @@ they can't tell you:
   `check:splitby-perf`) are stages of `scripts/gate.sh` of their own, and the boundary suite is in
   `npm test`.
   `npm run report:coverage` — the unit suites' coverage (CI shows it on the reports job).
+  **Every view kind's output is pinned** by `test/view-snapshots.test.ts` (in `npm test`, no
+  database): one gallery example per kind runs through the whole engine against warehouse answers
+  recorded in `test/snapshots/answers/`, and its compiled SQL and queries, computed result, figure
+  and results text are compared with `test/snapshots/<kind>.{sql,json,svg,txt}`. A diff there is a
+  change to review, not noise: `UPDATE_SNAPSHOTS=1` accepts output changes, and a changed query
+  needs `RECORD_SNAPSHOTS=1` with `.env.local` (public warehouse only; the answers hold public
+  studies' patient-level rows). A new view kind adds its case there. Each case also runs with its
+  answers delivered in reverse, and must not change: a runner combines what it asked for after
+  every answer is in, in the view's order (`askEachSite` in `lib/views/shared.ts`), never inside
+  the callback as each one lands.
   Secrets come from `.env.local` at runtime; never hardcode them anywhere, including tests.
 - Lint: `npm run lint` (~2 min, the gate's first stage) — see "Code standards" above for what it
   holds code to. While iterating, `npx eslint <files you touched>` is enough.

lib/ch.ts

@@ -226,21 +226,37 @@ export function directConnections(): ChConn[] {
   return connections().filter((c) => c.id !== peer?.id);
 }
 
+/** One statement as the database would receive it: its text, and the tables that ride along with it. */
+export interface ChStatement { conn: ChConn; format: string; query: string; externals: { name: string; structure: string; blob: Blob }[] }
+/**
+ * Answers statements instead of the database, or watches them on their way to it (`send`, which
+ * takes extra ClickHouse settings for that statement, e.g. "max_threads=1"). Only
+ * test/view-snapshots.test.ts sets one: it records the warehouse's answers, then replays them.
+ */
+export type ChTransport = (st: ChStatement, send: (settings?: string) => Promise<string>) => Promise<string>;
+let transport: ChTransport | null = null;
+export function setChTransport(t: ChTransport | null): void { transport = t; }
+
 async function post(sql: string, conn: ChConn, format: string): Promise<string> {
   assertNotLoopbackDirect(sql, conn);
-  if (!conn.url || !conn.user || !conn.password || !conn.db)
-    throw new Error("ClickHouse instance '" + conn.id + "' is missing configuration (url/password/db)");
   const ctx = ctxStore.getStore();
   const query = sql + "\nFORMAT " + format;
   // uploaded custom data and stored id sets ride along only with a statement that names them
   // (lib/custom-data.ts, lib/id-sets.ts)
   const externals = [...externalsFor(query),
     ...[...customExternalsFor(query, conn.id), ...idSetExternalsFor(query, conn.id)]
       .map((e) => ({ name: e.name, structure: e.structure, blob: new Blob([e.data]) }))];
+  if (transport) return transport({ conn, format, query, externals }, (settings) => postToDatabase(conn, query, externals, ctx, settings));
+  return postToDatabase(conn, query, externals, ctx);
+}
+
+async function postToDatabase(conn: ChConn, query: string, externals: ChStatement["externals"], ctx: ChCtx | undefined, settings?: string): Promise<string> {
+  if (!conn.url || !conn.user || !conn.password || !conn.db)
+    throw new Error("ClickHouse instance '" + conn.id + "' is missing configuration (url/password/db)");
   // Tell ClickHouse to stop a query whose client has gone away. Without it an abandoned
   // request keeps its share of the database until it finishes, which is most of what an
   // abandoned request costs.
-  const url = conn.url.replace(/\/+$/, "") + "/?cancel_http_readonly_queries_on_client_close=1";
+  const url = conn.url.replace(/\/+$/, "") + "/?cancel_http_readonly_queries_on_client_close=1" + (settings ? "&" + settings : "");
   const send = async (): Promise<string> => {
     const auth = "Basic " + Buffer.from(conn.user + ":" + conn.password).toString("base64");
     let init: RequestInit;

lib/views/assays.ts

@@ -4,7 +4,7 @@ import { assayProfileSql, assayMomentsSql, assayValueCountsSql, assayEntityNames
 import { chiSquareTest, benjaminiHochberg, welchT, welchAnovaTest, varianceFromMoments } from "../stats.ts";
 import { assaysMethod } from "../method.ts";
 import type { ViewRun } from "./run.ts";
-import { overLimit, answering, refuse, niceCeil, PROFILE_NAME } from "./shared.ts";
+import { overLimit, answering, refuse, niceCeil, PROFILE_NAME, askEachSite } from "./shared.ts";
 
 const ASSAY_ROWS = 300;
 /** past this many entities a profile is refused rather than compared: the per-cohort answer
@@ -54,35 +54,35 @@ export async function runAssays(view: AssaysView, ctx: ViewRun): Promise<ViewRes
   const names = new Map<string, string>();
   let missing = 0;
   const perSite: { cohort: string; study: string; rows: number }[] = [];
-  await Promise.all([
-    ...holders.map(async (st) => {
-      for (const r of await ok[st].viewMeta("assays:names", (s, d) => assayEntityNamesSql(s, d, pt))) if (!names.has(String(r.entity))) names.set(String(r.entity), String(r.name));
-    }),
-    ...vNames.flatMap((n) => cohorts[n].studies.filter((st) => holders.includes(st)).map(async (st) => {
-      const def = doc.cohorts[n];
-      if (isCat) {
-        const rows = await ok[st].viewAggregate("assays:values:" + n, def, (c, s, d) => assayValueCountsSql(c, s, d, pt));
-        for (const r of rows) {
-          const v = String(r.value ?? "").trim(), c = Number(r.n);
-          if (MISSING_VALUES.has(v)) { missing += c; continue; }
-          const k = n + "\t" + r.entity;
-          entities.add(String(r.entity));
-          const m = counts.get(k) || new Map<string, number>();
-          m.set(v, (m.get(v) || 0) + c); counts.set(k, m);
-        }
-        perSite.push({ cohort: n, study: st, rows: rows.length });
-      } else {
-        const rows = await ok[st].viewAggregate("assays:moments:" + n, def, (c, s, d) => assayMomentsSql(c, s, d, pt));
-        for (const r of rows) {
-          const k = n + "\t" + r.entity, cur = mom.get(k) || { n: 0, sx: 0, sxx: 0 };
-          cur.n += Number(r.n); cur.sx += Number(r.sx); cur.sxx += Number(r.sxx); mom.set(k, cur);
-          missing += Number(r.other);
-          entities.add(String(r.entity));
-        }
-        perSite.push({ cohort: n, study: st, rows: rows.length });
-      }
-    }))
+  const holding = Object.fromEntries(holders.map((st) => [st, ok[st]]));
+  const [nameRows, answers] = await Promise.all([
+    Promise.all(holders.map((st) => ok[st].viewMeta("assays:names", (s, d) => assayEntityNamesSql(s, d, pt)))),
+    askEachSite(ctx, holding, (n, _st, r) => isCat
+      ? r.viewAggregate("assays:values:" + n, doc.cohorts[n], (c, s, d) => assayValueCountsSql(c, s, d, pt))
+      : r.viewAggregate("assays:moments:" + n, doc.cohorts[n], (c, s, d) => assayMomentsSql(c, s, d, pt)))
   ]);
+  // the first study (in the view's order) to name an entity names it
+  for (const rows of nameRows) for (const r of rows) if (!names.has(String(r.entity))) names.set(String(r.entity), String(r.name));
+  for (const { cohort: n, study: st, answer: rows } of answers) {
+    if (isCat) {
+      for (const r of rows) {
+        const v = String(r.value ?? "").trim(), c = Number(r.n);
+        if (MISSING_VALUES.has(v)) { missing += c; continue; }
+        const k = n + "\t" + r.entity;
+        entities.add(String(r.entity));
+        const m = counts.get(k) || new Map<string, number>();
+        m.set(v, (m.get(v) || 0) + c); counts.set(k, m);
+      }
+    } else {
+      for (const r of rows) {
+        const k = n + "\t" + r.entity, cur = mom.get(k) || { n: 0, sx: 0, sxx: 0 };
+        cur.n += Number(r.n); cur.sx += Number(r.sx); cur.sxx += Number(r.sxx); mom.set(k, cur);
+        missing += Number(r.other);
+        entities.add(String(r.entity));
+      }
+    }
+    perSite.push({ cohort: n, study: st, rows: rows.length });
+  }
   if (!entities.size) return refuse("assays", ctx, "none of these cohorts' " + unit + " has a value in “" + pt + "”");
   if (entities.size > ASSAY_MAX_ENTITIES) return refuse("assays", ctx, "“" + pt + "” has " + entities.size.toLocaleString("en-US") + " entities with values here; an assays view compares at most " + ASSAY_MAX_ENTITIES.toLocaleString("en-US"));
 

lib/views/diffexpr.ts

@@ -4,7 +4,7 @@ import { profileMomentsAllSql, genesetMembersSql } from "../federation-views.ts"
 import { benjaminiHochberg, varianceFromMoments, welchT, preRankedGsea } from "../stats.ts";
 import { diffexprMethod } from "../method.ts";
 import type { ViewRun } from "./run.ts";
-import { overLimit, answering, refuse, niceCeil } from "./shared.ts";
+import { overLimit, answering, refuse, niceCeil, askEachSite } from "./shared.ts";
 
 const DE_MIN_N = 3, DE_POINT_CAP = 1500, DE_PERMUTATIONS = 1000;
 const SET_NAME = /^[A-Z0-9_]{3,120}$/;
@@ -51,15 +51,15 @@ export async function runDiffexpr(view: DiffexprView, ctx: ViewRun): Promise<Vie
       " profile — available: " + ([...offered].sort().join(", ") || "none"));
 
   const moments = new Map<string, { n: number; sx: number; sxx: number }>(); // cohort|gene
-  await Promise.all(vNames.flatMap((n) => cohorts[n].studies.filter((st) => ok[st]).map(async (st) => {
-    const rows = await ok[st].viewAggregate("diffexpr:moments:" + n, doc.cohorts[n],
-      (c, s, d) => profileMomentsAllSql(c, s, d, profileType, log2));
+  const answers = await askEachSite(ctx, ok, (n, _st, r) => r.viewAggregate("diffexpr:moments:" + n, doc.cohorts[n],
+    (c, s, d) => profileMomentsAllSql(c, s, d, profileType, log2)));
+  for (const { cohort: n, answer: rows } of answers) {
     for (const r of rows) {
       const k = n + "|" + String(r.gene), cur = moments.get(k) || { n: 0, sx: 0, sxx: 0 };
       cur.n += Number(r.n); cur.sx += Number(r.sx); cur.sxx += Number(r.sxx);
       moments.set(k, cur);
     }
-  })));
+  }
   const [A, B] = vNames;
   const genes = [...new Set([...moments.keys()].map((k) => k.slice(k.indexOf("|") + 1)))];
   const tested: DiffexprGene[] = [];

lib/views/enrichment.ts

@@ -4,7 +4,7 @@ import { enrichAlteredSql, enrichPanelProfiledSql, enrichWesSql, type EnrichAlte
 import { fisherExact, chiSquareTest, benjaminiHochberg } from "../stats.ts";
 import { enrichmentMethod } from "../method.ts";
 import type { ViewRun } from "./run.ts";
-import { overLimit, cytobands, answering, refuse, niceCeil } from "./shared.ts";
+import { overLimit, cytobands, answering, refuse, niceCeil, askEachSite } from "./shared.ts";
 
 const ENRICH_ROWS = 300;
 const ALTERATION_WORD: Record<EnrichAlteration, string> = {
@@ -32,19 +32,18 @@ export async function runEnrichment(view: EnrichmentView, ctx: ViewRun): Promise
   const altered = new Map<string, number>(), panel = new Map<string, number>(), wes = new Map<string, number>();
   const genes = new Set<string>();
   const perSite: { cohort: string; study: string; genes: number; wes: number }[] = [];
-  await Promise.all(vNames.flatMap((n) => cohorts[n].studies.filter((st) => ok[st]).map(async (st) => {
-    const def = doc.cohorts[n];
-    const [a, pr, w] = await Promise.all([
-      ok[st].viewAggregate("enrichment:altered:" + n, def, (c, s, d) => enrichAlteredSql(c, s, d, alteration, pin, measure === "driver")),
-      ok[st].viewAggregate("enrichment:panels:" + n, def, (c, s, d) => enrichPanelProfiledSql(c, s, d, alteration)),
-      ok[st].viewAggregate("enrichment:wes:" + n, def, (c, s, d) => enrichWesSql(c, s, d, alteration))
-    ]);
+  const answers = await askEachSite(ctx, ok, (n, st, r) => Promise.all([
+    r.viewAggregate("enrichment:altered:" + n, doc.cohorts[n], (c, s, d) => enrichAlteredSql(c, s, d, alteration, pin, measure === "driver")),
+    r.viewAggregate("enrichment:panels:" + n, doc.cohorts[n], (c, s, d) => enrichPanelProfiledSql(c, s, d, alteration)),
+    r.viewAggregate("enrichment:wes:" + n, doc.cohorts[n], (c, s, d) => enrichWesSql(c, s, d, alteration))
+  ]));
+  for (const { cohort: n, study: st, answer: [a, pr, w] } of answers) {
     for (const r of a) { const k = n + "\t" + r.gene; altered.set(k, (altered.get(k) || 0) + Number(r.altered)); genes.add(String(r.gene)); }
     for (const r of pr) { const k = n + "\t" + r.gene; panel.set(k, (panel.get(k) || 0) + Number(r.profiled)); }
     const wn = Number(w[0]?.n || 0);
     wes.set(n, (wes.get(n) || 0) + wn);
     perSite.push({ cohort: n, study: st, genes: a.length, wes: wn });
-  })));
+  }
   const two = vNames.length === 2;
   const tested: EnrichmentGene[] = [];
   let untested = 0;

lib/views/shared.ts

@@ -67,3 +67,16 @@ export async function cytobands(ok: Record<string, ViewResolver>, genes: string[
 export async function studyCatalogs(c: ViewRun, studies: string[]): Promise<StudyCatalog[]> {
   return Promise.all(studies.map(async (st) => ({ study: st, catalog: (await c.studyProfiles(st)).catalog || [] })));
 }
+
+/**
+ * Asks every (cohort, study) the view draws that `resolvers` holds, concurrently, and returns the
+ * answers in the view's order (its cohorts, then each cohort's studies), never in the order they
+ * finish. Combine them after this returns: combining inside `ask` as each answer lands makes the
+ * result depend on which query finished first (test/view-snapshots.test.ts checks for that).
+ */
+export async function askEachSite<R, T>(ctx: ViewRun, resolvers: Record<string, R>,
+  ask: (cohort: string, study: string, r: R) => Promise<T>): Promise<{ cohort: string; study: string; answer: T }[]> {
+  const sites = ctx.vNames.flatMap((cohort) => ctx.cohorts[cohort].studies.filter((st) => resolvers[st]).map((study) => ({ cohort, study })));
+  const answers = await Promise.all(sites.map((s) => ask(s.cohort, s.study, resolvers[s.study])));
+  return sites.map((s, i) => ({ ...s, answer: answers[i] }));
+}

lib/views/survival.ts

@@ -5,6 +5,7 @@ import type { ViewResult, LogRank, LogRankK, SurvivalView } from "../types.ts";
 import { expandSurvivalCounts, type SurvivalCount, type EndpointColumns } from "../federation.ts";
 import { survivalMethod } from "../method.ts";
 import { untimed, type Untimed, type ViewRun, type DocView } from "./run.ts";
+import { askEachSite } from "./shared.ts";
 
 export async function runSurvival(view: DocView, c: ViewRun): Promise<ViewResult | Untimed> {
   const { doc, cohorts, resolvers, vNames, vStudies, participationFor } = c;
@@ -60,10 +61,8 @@ export async function runSurvival(view: DocView, c: ViewRun): Promise<ViewResult
     : null;
   const perSite = new Map<string, Map<string, SurvivalCount[]>>(); // cohort -> study -> triples
   for (const n of vNames) perSite.set(n, new Map());
-  await Promise.all(vNames.flatMap((n) =>
-    cohorts[n].studies.map(async (st) => {
-      perSite.get(n)!.set(st, await resolvers[st].survivalCounts(doc.cohorts[n], endpoint, anchor, columns));
-    })));
+  for (const { cohort: n, study: st, answer } of await askEachSite(c, resolvers, (n, _st, r) => r.survivalCounts(doc.cohorts[n], endpoint, anchor, columns)))
+    perSite.get(n)!.set(st, answer);
   // An EMPTY COHORT also yields no rows, and must not be reported as a
   // missing endpoint — that sends the reader (or a model) off to conclude the
   // study lacks OS when it does not. Separate the two causes before blaming