fix(telemetry): exclude pre-schema rows from latency rollup (#137)

## What

Follow-up to #132. Adds a schema-version marker to every latency row and
filters on it in the rollup query, so pre-#132 rows are excluded from
`/telemetry/v1/stats/latency`.

## Why (the AE default-0 gotcha)

The Analytics Engine SQL API has **no NULLs**: any `double` a row never
wrote is materialized as **`0`** at query time. Rows written by the
pre-latency Worker (before #132) have no `double7..10`, so they read
`scan_p50 == 0` — a materialized 0, not a real sample. That passed the
rollup's `WHERE <p50col> >= 0` sentinel filter and showed up as legit `0
ms` samples. Caught in the live smoke of the endpoint: a day that
predates the deploy reported `samples > 0, avg 0`.

The `-1` sentinel can't fix this on its own — it only distinguishes
empty from present *within* a row that actually wrote the doubles; a
legacy row never wrote them, so there's no sentinel to read.

## Fix

- **Write** (`writePing`): append a schema-version marker `double11 = 1`
(`LATENCY_SCHEMA_VERSION`) on every row that carries the latency
doubles.
- **Query** (`queryLatencyMetric`): add `AND double11 >= 1` to each
per-metric `WHERE` (in addition to the existing `>= 0` sentinel). A
legacy row materializes the marker as `0` and is excluded; a new
empty-latency row is marked (`double11 = 1`) but still excluded by its
`-1` sentinel; a real `0 ms` still counts.
- Both sites comment the AE missing-double=0 behavior. README dataset
table + rollup section updated (`double11` row, dual-filter
explanation).

## Tests (worker vitest)

- `writePing` marks every new latency-schema row with `double11 = 1`.
- A new-but-empty-latency row is marked (`double11 = 1`) yet still
carries the `-1` sentinels — proving the marker and sentinel filters are
independent (the sentinel is what excludes it).
- Both per-metric rollup queries include `double11 >= 1` (the mechanism
that excludes pre-latency rows, whose marker materializes as 0).

Gates green: worker `typecheck` + `lint` + `test` (58, +2 new).

## Notes

Cut from latest `origin/main` (which already includes #132). Endpoint
still needs the same wrangler secrets as #132; the live smoke that
surfaced this bug already has them set. Do not merge — for review.

Refs #34

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: pmaxhogan

telemetry-worker/README.md

@@ -36,13 +36,22 @@ One data point per ping (`writePing`):
 | `blob1..6` | `os`, `arch`, `channel`, `version`, `os_version` (`""` if absent), `errors_by_class` JSON |
 | `double1..6` | `files_uploaded`, `bytes_uploaded`, `deep_verify_runs`, `update_applied` (0/1), `total_errors`, `ts` (epoch ms) |
 | `double7..10` | `scan_p50`, `scan_p95`, `upload_per_mb_p50`, `upload_per_mb_p95` (ms) |
+| `double11` | `latency_schema_version` (schema marker, `1`) |
 
-The 4 latency doubles (DESIGN s13) are **appended** so the original columns keep
+The latency doubles (DESIGN s13) are **appended** so the original columns keep
 their positions. When the client had no samples for a metric this window (its
 array is empty), the pair is written as the sentinel **`-1`** so the rollup query
 can tell "no samples" apart from a legitimate `0 ms` (a sub-millisecond per-file
 scan rounds to 0).
 
+`double11` is a **schema marker** (`1`) written on every row that carries the
+latency doubles. It exists because the Analytics Engine SQL API has no NULLs and
+materializes any double a row never wrote as `0`: rows written by the pre-latency
+Worker have `scan_p50 == 0` (a materialized 0, not a real sample) and would
+otherwise pass the `>= 0` sentinel filter and pollute the rollup as fake `0 ms`
+samples. The rollup filters `WHERE double11 >= 1`, so pre-latency rows (marker
+materializes as `0`) are excluded.
+
 ## `GET /telemetry/v1/stats/latency`
 
 Per-day aggregates of the client-reported percentiles. **Authenticated** - it
@@ -68,10 +77,11 @@ Authorization: Bearer <QUERY_TOKEN>
 
 Per metric, per UTC day: `avg_p50_ms` (mean of the pinged p50s), `avg_p95_ms`
 (mean of the pinged p95s), `max_p95_ms` (worst pinged p95), and `samples` (number
-of pings that reported the metric). Empty-latency pings (the `-1` sentinel) are
-excluded per metric via `WHERE <p50col> >= 0`, so a real `0 ms` still counts. The
-two metrics are queried separately (each filters its own sentinel column) via the
-Analytics Engine SQL API.
+of pings that reported the metric). Each metric query excludes two kinds of
+non-sample rows (`WHERE double11 >= 1 AND <p50col> >= 0`): pre-latency rows (the
+schema marker materializes as `0`) and empty-latency pings (the `-1` sentinel),
+while keeping a real `0 ms`. The two metrics are queried separately (each filters
+its own sentinel column) via the Analytics Engine SQL API.
 
 Status codes: `200` success; `401` missing/wrong bearer; `405` non-GET;
 `502` upstream AE SQL query failed; `503` the endpoint is not configured

telemetry-worker/src/index.ts

@@ -386,17 +386,36 @@ function latencyPair(arr: number[]): [number, number] {
   return [NO_LATENCY, NO_LATENCY];
 }
 
+/// Schema-version marker written on EVERY row that carries the latency doubles
+/// (DESIGN s13). Load-bearing for the rollup: the Analytics Engine SQL API has NO
+/// NULLs and materializes any double a row never wrote as `0`, so rows written by
+/// the PRE-latency Worker have `scan_p50 == 0` (a materialized 0, not a real
+/// sample) and would otherwise pass the `>= 0` sentinel filter and pollute the
+/// rollup as fake `0 ms` samples. A row with this marker `>= 1` provably has the
+/// latency schema; a legacy row materializes the marker as `0`, so the rollup's
+/// `WHERE <marker> >= 1` excludes it. Bump this only if the latency-double LAYOUT
+/// changes (append-only, like the doubles themselves).
+const LATENCY_SCHEMA_VERSION = 1;
+
+/// The AE double column the schema marker occupies (double11, appended after the
+/// 4 latency percentiles). The rollup filters `>= 1` on this so pre-latency rows
+/// (which materialize it as 0) are excluded.
+const LATENCY_SCHEMA_COL = "double11";
+
 /// Write one validated ping to Analytics Engine (SPEC s16, DESIGN s13). Schema:
 /// - indexes: [install_id]   (the sampling/grouping key - anonymous)
 /// - blobs:   [os, arch, channel, version, os_version, errors_by_class JSON]
 ///            (low-card dims; os_version is "" when the client did not send one)
 /// - doubles: [files_uploaded, bytes_uploaded, deep_verify_runs, update_applied,
 ///             total_errors, ts,                              // double1..double6
-///             scan_p50, scan_p95, upload_per_mb_p50, upload_per_mb_p95]
+///             scan_p50, scan_p95, upload_per_mb_p50, upload_per_mb_p95,
 ///                                                            // double7..double10
-///   The 4 latency doubles are appended (never reordered) so existing columns keep
-///   their positions; an absent metric writes the NO_LATENCY (-1) sentinel.
-/// Writes are non-blocking (no await / waitUntil needed per the CF docs).
+///             latency_schema_version]                        // double11
+///   The latency doubles are appended (never reordered) so existing columns keep
+///   their positions; an absent metric writes the NO_LATENCY (-1) sentinel, and
+///   double11 marks the row as carrying the latency schema (LATENCY_SCHEMA_VERSION)
+///   so the rollup can exclude pre-latency rows (whose missing doubles AE
+///   materializes as 0). Writes are non-blocking (no await / waitUntil per CF docs).
 export function writePing(env: Env, p: PingPayload): void {
   const [scanP50, scanP95] = latencyPair(p.latency_p50_p95_ms.scan);
   const [upP50, upP95] = latencyPair(p.latency_p50_p95_ms.upload_per_mb);
@@ -427,6 +446,10 @@ export function writePing(env: Env, p: PingPayload): void {
       scanP95,
       upP50,
       upP95,
+      // DESIGN s13: schema-version marker (double11). Present (>= 1) on every row
+      // that carries the latency doubles above; a pre-latency row has no double11
+      // so AE materializes it as 0, letting the rollup exclude such rows.
+      LATENCY_SCHEMA_VERSION,
     ],
   });
 }
@@ -548,10 +571,13 @@ function clampStatsDays(raw: string | null): number {
 
 /// Query one latency metric's per-day aggregates over the AE SQL API. `p50Col` /
 /// `p95Col` are the AE double column names for this metric (e.g. `double7` /
-/// `double8`). Rows carrying the NO_LATENCY sentinel (`< 0`, an empty-latency
-/// ping) are excluded via `WHERE p50 >= 0`, so a legit `0 ms` still counts. The
-/// response is the CF `{ meta, data }` JSON (NOT ndjson); each `data[]` row's
-/// numeric columns arrive as strings, so they are coerced with `Number`.
+/// `double8`). Two `WHERE` filters exclude non-samples (both needed because AE has
+/// no NULLs - a never-written double reads as 0): the schema marker (excludes
+/// PRE-latency rows whose latency doubles materialize as 0) and the NO_LATENCY
+/// sentinel (`< 0`, excludes latency-schema rows with no samples this window, while
+/// keeping a legit `0 ms`). The response is the CF `{ meta, data }` JSON (NOT
+/// ndjson); each `data[]` row's numeric columns arrive as strings, coerced with
+/// `Number`.
 async function queryLatencyMetric(
   env: Env,
   accountId: string,
@@ -561,14 +587,24 @@ async function queryLatencyMetric(
 ): Promise<LatencyDayRow[]> {
   // `days` is a validated integer (clampStatsDays) and the column names are
   // internal constants, so this interpolation carries no injection surface.
+  //
+  // Two filters, both load-bearing (the AE SQL API has NO NULLs - a double a row
+  // never wrote is materialized as 0):
+  //   - `${LATENCY_SCHEMA_COL} >= 1`: exclude PRE-latency rows. Those rows never
+  //     wrote the latency doubles, so their `${p50Col}` materializes as 0 and
+  //     would otherwise pass the sentinel filter below as a fake `0 ms` sample.
+  //     Only rows carrying the latency schema wrote the marker (>= 1).
+  //   - `${p50Col} >= 0`: exclude latency-schema rows whose metric had NO samples
+  //     this window (written as the -1 sentinel), while KEEPING a legit 0 ms.
   const sql =
     `SELECT toDate(timestamp) AS day, ` +
     `AVG(${p50Col}) AS avg_p50, ` +
     `AVG(${p95Col}) AS avg_p95, ` +
     `MAX(${p95Col}) AS max_p95, ` +
     `COUNT() AS samples ` +
     `FROM ${DATASET} ` +
-    `WHERE timestamp > NOW() - INTERVAL '${days}' DAY AND ${p50Col} >= 0 ` +
+    `WHERE timestamp > NOW() - INTERVAL '${days}' DAY ` +
+    `AND ${LATENCY_SCHEMA_COL} >= 1 AND ${p50Col} >= 0 ` +
     `GROUP BY day ORDER BY day`;
 
   const resp = await fetch(

telemetry-worker/test/handler.test.ts

@@ -572,6 +572,8 @@ const SCAN_P50 = 6;
 const SCAN_P95 = 7;
 const UP_P50 = 8;
 const UP_P95 = 9;
+/// The schema-version marker double (double11), appended after the 4 percentiles.
+const SCHEMA_MARKER = 10;
 
 describe("telemetry worker latency doubles (DESIGN s13)", () => {
   it("writes the client-reported [p50, p95] latency doubles", async () => {
@@ -589,6 +591,31 @@ describe("telemetry worker latency doubles (DESIGN s13)", () => {
     expect(dp.doubles[UP_P95]).toBe(110);
   });
 
+  it("marks every new latency-schema row with the schema-version marker (double11)", async () => {
+    // The marker (>= 1) is what lets the rollup exclude pre-latency rows: AE
+    // materializes a missing double as 0, so a legacy row reads double11 == 0.
+    const { env, writes } = mockEnv();
+    const p = validPayload();
+    p.latency_p50_p95_ms = { scan: [3, 12], upload_per_mb: [40, 110] };
+    const res = await handle(postPing(JSON.stringify(p)), env);
+    expect(res.status).toBe(204);
+    const dp = writes[0] as { doubles: number[] };
+    expect(dp.doubles[SCHEMA_MARKER]).toBe(1);
+  });
+
+  it("marks a new row EVEN WHEN its latency is empty (marker present, sentinels -1)", async () => {
+    // A new-but-empty row is marked as latency-schema (double11 == 1) yet still
+    // carries the -1 sentinels, so the rollup's sentinel filter (not the marker)
+    // is what excludes it - the two filters are independent.
+    const { env, writes } = mockEnv();
+    const res = await handle(postPing(JSON.stringify(validPayload())), env);
+    expect(res.status).toBe(204);
+    const dp = writes[0] as { doubles: number[] };
+    expect(dp.doubles[SCHEMA_MARKER]).toBe(1); // marked new...
+    expect(dp.doubles[SCAN_P50]).toBe(-1); // ...but sentinel-excluded from the rollup
+    expect(dp.doubles[UP_P50]).toBe(-1);
+  });
+
   it("writes the -1 sentinel for a metric with no samples (empty array)", async () => {
     const { env, writes } = mockEnv();
     const p = validPayload();
@@ -701,6 +728,9 @@ describe("telemetry worker GET /stats/latency (DESIGN s13)", () => {
     expect(fetchMock).toHaveBeenCalledTimes(2);
     expect(sqls.some((s) => s.includes("double7") && s.includes("double7 >= 0"))).toBe(true);
     expect(sqls.some((s) => s.includes("double9") && s.includes("double9 >= 0"))).toBe(true);
+    // BOTH queries also filter the schema marker (excludes pre-latency rows whose
+    // missing latency doubles AE materializes as 0).
+    expect(sqls.every((s) => s.includes("double11 >= 1"))).toBe(true);
     // The default 7-day window is in the SQL.
     expect(sqls.every((s) => s.includes("INTERVAL '7' DAY"))).toBe(true);
   });