From 8db0242b006a231292a5df51ae2632ba5ba88f97 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Sat, 24 Oct 2026 18:51:42 +0200 Subject: [PATCH] fix(export): bound every export call and order current days first Every Postgres and BigQuery call in the event and account exports has a client-side bound, and a Postgres client that times out is destroyed. A 28-minute watchdog in runAndExit logs the phase in flight, the days not started and the faults so far, and exits 1 before the Job's 1800 s task timeout; every other failure is logged once with its phase. The window days run before the stale backlog. Day counts come from one exact query over the table, readDay stops on a short page, and the export_rules digest covers the readDay SQL, so a change to the exported row shape reloads history. A golden test pins exported rows to ROW_SHAPE_VERSION. The request log classifies the target at request start, adds no listener for static assets, and logs a placeholder path for any target other than an origin-form path, an http(s) URL or "*". --- db/account-export.ts | 75 ++-- db/bounded.ts | 105 +++++ db/connection.ts | 3 +- db/event-export.ts | 490 ++++++++++++++-------- db/export-events.ts | 79 ++-- db/test/account-export.test.ts | 33 +- db/test/bounded.test.ts | 62 +++ db/test/event-export.test.ts | 561 ++++++++++++++++++++------ db/test/runtime-pool.test.ts | 5 +- deploy/README.md | 17 +- docs/analytics.md | 4 +- src/lib/proxy/requestLog.ts | 79 +++- src/lib/proxy/test/requestLog.test.ts | 58 ++- 13 files changed, 1196 insertions(+), 375 deletions(-) create mode 100644 db/bounded.ts create mode 100644 db/test/bounded.test.ts diff --git a/db/account-export.ts b/db/account-export.ts index 01bf132..c45757a 100644 --- a/db/account-export.ts +++ b/db/account-export.ts @@ -1,7 +1,9 @@ import fs from 'node:fs/promises'; import type { BigQuery, TableField } from '@google-cloud/bigquery'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; + +import { callWithin, loadWithin, QUERY_GRACE_MS, STATEMENT_TIMEOUT_MS } from './bounded.ts'; /** * The account export, which the event export Job (db/export-events.ts) runs @@ -56,53 +58,76 @@ export type AccountRead = { webmailDomains: boolean; }; +const readInTransaction = async (client: PoolClient): Promise => { + await client.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY'); + + const { rows } = await client.query( + `SELECT to_char(day, 'YYYY-MM-DD') AS day, organisation, accounts + FROM account_daily + ORDER BY day, organisation`, + ); + const { rows: source } = await client.query<{ + accounts: boolean; + webmailDomains: boolean; + }>(`SELECT accounts, webmail_domains AS "webmailDomains" FROM account_daily_source`); + + await client.query('COMMIT'); + + return { rows, ...source[0]! }; +}; + +/** How long the account read may take: its two queries at the statement timeout, then the grace. */ +const ACCOUNT_READ_MS = 2 * STATEMENT_TIMEOUT_MS + QUERY_GRACE_MS; + /** * Every row of the view, oldest day first, with the facts that say why it * holds none. One read-only transaction reads both, so they share one - * `now()` and one snapshot. + * `now()` and one snapshot. A read still running at `timeoutMs`, as over a + * connection that stalls past Postgres's timeout, throws. */ -export const readAccountDaily = async (pool: Pool): Promise => { +export const readAccountDaily = async ( + pool: Pool, + { timeoutMs = ACCOUNT_READ_MS }: { timeoutMs?: number } = {}, +): Promise => { const client = await pool.connect(); try { - await client.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY'); - - const { rows } = await client.query( - `SELECT to_char(day, 'YYYY-MM-DD') AS day, organisation, accounts - FROM account_daily - ORDER BY day, organisation`, + const read = await callWithin( + readInTransaction(client), + timeoutMs, + () => new Error(`account read ran past ${timeoutMs / 1000} s`), ); - const { rows: source } = await client.query<{ - accounts: boolean; - webmailDomains: boolean; - }>(`SELECT accounts, webmail_domains AS "webmailDomains" FROM account_daily_source`); - await client.query('COMMIT'); client.release(); - return { rows, ...source[0]! }; + return read; } catch (error) { - // A failed query leaves the transaction open, and a release with - // the error discards the connection and the transaction with it. + // A failed or timed-out read leaves the transaction open, or a query + // in flight, and a release with the error discards the connection + // with it. client.release(error instanceof Error ? error : true); throw error; } }; -/** Replaces the BigQuery table `account_daily` with the rows in `file`, one JSON object a line. */ +/** + * Replaces the BigQuery table `account_daily` with the rows in `file`, one + * JSON object a line. A load past its timeout is cancelled and throws + * (loadWithin). + */ export const bigQueryAccountTable = ({ datasetId, bigquery }: { datasetId: string; bigquery: BigQuery }) => async (file: string): Promise => { - // Creates the table when it does not exist. Resolves once the load - // job is done, and throws if it failed. - await bigquery - .dataset(datasetId) - .table('account_daily') - .load(file, { + // Creates the table when it does not exist. + await loadWithin({ + table: bigquery.dataset(datasetId).table('account_daily'), + file, + metadata: { sourceFormat: 'NEWLINE_DELIMITED_JSON', writeDisposition: 'WRITE_TRUNCATE', schema: { fields: ACCOUNT_SCHEMA }, - }); + }, + }); }; /** diff --git a/db/bounded.ts b/db/bounded.ts new file mode 100644 index 0000000..2721811 --- /dev/null +++ b/db/bounded.ts @@ -0,0 +1,105 @@ +import { setTimeout as sleep } from 'node:timers/promises'; + +import type { Job, JobLoadMetadata, Table } from '@google-cloud/bigquery'; + +/** + * Time bounds for the export Job's calls (db/export-events.ts) whose + * clients take none: a database query over a connection that stalls past + * Postgres's own timeout, a BigQuery call, and a BigQuery load of a local + * file and its cancel. + */ + +/** The `statement_timeout` of the export Job's pool. */ +export const STATEMENT_TIMEOUT_MS = 5 * 60_000; + +/** + * How long past its server-side bound a query may run before the Job stops + * waiting for it, for a connection that stalls where the server's timeout + * cannot reach it. + */ +export const QUERY_GRACE_MS = 5_000; + +/** How long one BigQuery load may run, upload included. */ +export const LOAD_TIMEOUT_MS = 2 * 60_000; + +/** How long the cancel of a load past its timeout is waited for. */ +export const CANCEL_TIMEOUT_MS = 10_000; + +/** What `within` resolves to when `work` is still running at `ms`. */ +export const TIMED_OUT = Symbol('timed out'); + +/** + * `work`'s outcome, or TIMED_OUT once `ms` pass with `work` left running. + * The timer is cleared when `work` settles first. + */ +export const within = async (work: Promise, ms: number): Promise => { + const timer = new AbortController(); + + try { + return await Promise.race([ + work, + sleep(ms, TIMED_OUT, { signal: timer.signal }).catch( + (): typeof TIMED_OUT => TIMED_OUT, + ), + ]); + } finally { + timer.abort(); + } +}; + +/** `call`'s result, or a throw of `timedOut()` once `ms` pass with it still running. */ +export const callWithin = async ( + call: Promise, + ms: number, + timedOut: () => Error, +): Promise => { + const result = await within(call, ms); + + if (result === TIMED_OUT) { + throw timedOut(); + } + + return result; +}; + +/** + * Loads `file` into `table` and resolves once the upload is done and the + * load job has finished, or throws if it failed. The client takes no + * timeout for an upload, so a load still running at `timeoutMs` is + * cancelled, the cancel waited for at most `cancelTimeoutMs`, and the call + * throws. A cancel that comes too late leaves the table holding the whole + * file. + */ +export const loadWithin = async ({ + table, + file, + metadata, + timeoutMs = LOAD_TIMEOUT_MS, + cancelTimeoutMs = CANCEL_TIMEOUT_MS, +}: { + table: Table; + file: string; + metadata: JobLoadMetadata; + timeoutMs?: number; + cancelTimeoutMs?: number; +}): Promise => { + let job: Job | undefined; + const load = (async () => { + [job] = await table.createLoadJob(file, metadata); + await job.promise(); + })(); + + // A load left running past its timeout settles unobserved. + load.catch(() => undefined); + + if ((await within(load, timeoutMs)) === TIMED_OUT) { + if (job) { + await within( + job.cancel().catch(() => undefined), + cancelTimeoutMs, + ); + } + + throw new Error(`load ran past ${timeoutMs / 1000} s and was cancelled`); + } +}; diff --git a/db/connection.ts b/db/connection.ts index 43c46c4..8aeb31b 100644 --- a/db/connection.ts +++ b/db/connection.ts @@ -194,7 +194,8 @@ export function getCloudSqlRuntimeStream(): DriverOptions['stream'] { if (!stream) { throw new Error( "Cloud SQL stream factory is not set on globalThis -- instrumentation.ts's " + - 'register() must run before the runtime pool connects.', + "register() or, in the export Job, db/export-events.ts's main() must set it " + + 'before the runtime pool connects.', ); } diff --git a/db/event-export.ts b/db/event-export.ts index 7b49a38..2bdef8d 100644 --- a/db/event-export.ts +++ b/db/event-export.ts @@ -20,6 +20,14 @@ import type { import type { SearchShareMode } from '../src/lib/search/searchShareRequest.ts'; import type { DeclinedReason, Known, ResolveResponse } from '../src/lib/search/types.ts'; +import { + callWithin, + loadWithin, + QUERY_GRACE_MS, + STATEMENT_TIMEOUT_MS, + TIMED_OUT, + within, +} from './bounded.ts'; import { requiredEnv } from './connection.ts'; /** @@ -292,25 +300,52 @@ const exportPage = (page: string | null): string | null => { // The version of the code that shapes a row: digest, passes, exportProps, // exportPage and exportRow. Bumped by hand with any change to them, so the -// change reaches every day (exportRules). -const ROW_SHAPE_VERSION = 1; +// change reaches every day (exportRules). READ_DAY_SQL, which shapes the +// rows they take, enters the digest as its text and needs no bump. The +// golden rows in db/test/event-export.test.ts pin each version's output. +export const ROW_SHAPE_VERSION = 1; /** - * The digest every row carries in `export_rules`: SHA-256 over PROP_RULES, - * TOKEN, PAGE_PATH, SHARE_PATH, EXPORT_SCHEMA, ROW_SHAPE_VERSION and an - * HMAC of a fixed string under `hmacKey`, so a new key changes it too. A - * day holding a row with another digest, or none, reloads (exportDays). + * One page of a UTC day's events (pgEventSource's readDay), with + * `created_at` as UTC text to the microsecond, the id as a JSON number, and + * `props` cut to the keys in $4. Parameters: the page's lower bound, the + * next midnight, the id of the last row read, the kept prop keys, the page + * size. + */ +const READ_DAY_SQL = `SELECT id::float8 AS id, + to_char(created_at AT TIME ZONE 'UTC', + 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS "createdAt", + name, user_id AS "userId", visitor_id AS "visitorId", page, + coalesce((SELECT jsonb_object_agg(key, value) + FROM jsonb_each(props) + WHERE key = ANY($4::text[])), '{}') AS props + FROM event + WHERE created_at >= $1 AND created_at < $2 AND (created_at, id) > ($1, $3) + ORDER BY created_at, id + LIMIT $5`; + +/** What shapes a row, the input exportRules digests. */ +export const ROW_SHAPE = { + PROP_RULES, + TOKEN: TOKEN.source, + PAGE_PATH: PAGE_PATH.source, + SHARE_PATH: SHARE_PATH.source, + EXPORT_SCHEMA, + READ_DAY_SQL, + ROW_SHAPE_VERSION, +}; + +/** + * The digest every row carries in `export_rules`: SHA-256 over ROW_SHAPE + * and an HMAC of a fixed string under `hmacKey`, so a new key changes it + * too. A day holding a row with another digest, or none, reloads + * (exportDays). */ export const exportRules = (hmacKey: string): string => createHash('sha256') .update( JSON.stringify({ - PROP_RULES, - TOKEN: TOKEN.source, - PAGE_PATH: PAGE_PATH.source, - SHARE_PATH: SHARE_PATH.source, - EXPORT_SCHEMA, - ROW_SHAPE_VERSION, + ...ROW_SHAPE, key: createHmac('sha256', hmacKey).update('export_rules').digest('hex'), }), ) @@ -338,18 +373,37 @@ export const exportRow = (row: SourceEvent, hmacKey: string, rules: string): Exp const DAY_MS = 86_400_000; +/** How long one BigQuery query may run. */ +const BIGQUERY_TIMEOUT_MS = 2 * 60_000; + +/** + * How long the Job waits for one BigQuery call, a query or a table read or + * create, before it gives up on it: BIGQUERY_TIMEOUT_MS plus the grace. + */ +const BIGQUERY_CALL_MS = BIGQUERY_TIMEOUT_MS + QUERY_GRACE_MS; + +/** The `connectionTimeoutMillis` of the export Job's pool (db/export-events.ts). */ +export const CONNECTION_TIMEOUT_MS = 30_000; + +/** + * How long the page queries of one day may take together, from the day's + * start, so the wait for its connection counts too. A read past it fails + * its day (pgEventSource). + */ +const DAY_READ_MS = 5 * 60_000; + /** - * How long a run starts new days for, 20 minutes, well inside the Job's - * 1800 s task timeout, so a run that has many days each waiting out a - * statement timeout ends and reports before the platform kills it. + * When runAndExit stops a Job run, 28 minutes from its start, two before + * the Job's 1800 s task timeout, so its report is out before the platform + * kills it. The days not reached by then wait for the next hourly run. */ -export const EXPORT_BUDGET_MS = 20 * 60_000; +export const JOB_DEADLINE_MS = 28 * 60_000; /** The most one BigQuery query may bill (`maximumBytesBilled`). */ const MAX_BYTES_BILLED = 1e9; -/** What the staleDays scan bills per event row, from its two columns. */ -const STALE_BYTES_PER_ROW = 74; +/** What the days query bills per event row, from its two columns. */ +const DAYS_BYTES_PER_ROW = 74; const dayOf = (ms: number): string => new Date(ms).toISOString().slice(0, 10); @@ -357,13 +411,15 @@ const dayOf = (ms: number): string => new Date(ms).toISOString().slice(0, 10); export type DayCounts = ReadonlyMap; /** - * The UTC days a run replaces, oldest first. They run from the day before - * the latest day BigQuery holds through `today`, or from the first event's - * day when BigQuery holds nothing, and add every day whose `events` count - * differs from its `partitions` count and every `stale` day. A day that - * loaded short, or failed to load under new rules, thus stays in every run - * until it matches, however many later days load. None when there is no - * event and BigQuery holds nothing. + * The UTC days a run replaces. First the window, oldest first: the day + * before the latest day BigQuery holds through `today`, or the first + * event's day through `today` when BigQuery holds nothing. Then the + * backlog, oldest first: every other day whose `events` count differs from + * its `partitions` count, and every other `stale` day. A day that loaded + * short, or failed to load under new rules, thus stays in every run until + * it matches, however many later days load, and a long backlog runs only + * after yesterday and today. None when there is no event and BigQuery holds + * nothing. */ export const exportDays = ({ today, @@ -379,21 +435,20 @@ export const exportDays = ({ const latest = [...partitions.keys()].sort().at(-1); const start = latest !== undefined ? dayOf(Date.parse(latest) - DAY_MS) : [...events.keys()].sort()[0]; - const days = new Set(stale); + const window = new Set(); if (start !== undefined) { for (let ms = Date.parse(start); ms <= Date.parse(today); ms += DAY_MS) { - days.add(dayOf(ms)); + window.add(dayOf(ms)); } } - for (const day of new Set([...events.keys(), ...partitions.keys()])) { - if ((events.get(day) ?? 0) !== (partitions.get(day) ?? 0)) { - days.add(day); - } - } + const mismatched = [...new Set([...events.keys(), ...partitions.keys()])].filter( + (day) => (events.get(day) ?? 0) !== (partitions.get(day) ?? 0), + ); + const backlog = new Set([...stale, ...mismatched].filter((day) => !window.has(day))); - return [...days].sort(); + return [...window, ...[...backlog].sort()]; }; /** The website database the rows come from. */ @@ -408,27 +463,57 @@ export type EventSource = { export type Warehouse = { /** Creates the table when it does not exist. */ ensureTable(): Promise; - /** The rows each day's partition holds. */ - dayCounts(): Promise; - /** The days holding a row whose `export_rules` is not `rules`, NULL included. */ - staleDays(rules: string): Promise>; + /** + * The rows each day's partition holds, and the days holding a row whose + * `export_rules` is not `rules`, NULL included. + */ + days(rules: string): Promise<{ counts: DayCounts; stale: ReadonlySet }>; /** Replaces the day's partition with the rows in `file`, one JSON object a line. */ replaceDay(day: string, file: string): Promise; }; const PAGE_SIZE = 5_000; -export const pgEventSource = (pool: Pool, pageSize = PAGE_SIZE): EventSource => ({ +/** + * The event table over `pool`. Each query has a client-side bound: the day + * counts the pool's `statement_timeout` plus `graceMs`, and a day's pages + * together `readMs` plus `graceMs`. A client whose query runs past its + * bound is still mid-query, so it is released with the error, which + * destroys it. + */ +export const pgEventSource = ( + pool: Pool, + { + pageSize = PAGE_SIZE, + readMs = DAY_READ_MS, + graceMs = QUERY_GRACE_MS, + }: { pageSize?: number; readMs?: number; graceMs?: number } = {}, +): EventSource => ({ async dayCounts() { - // The UTC date of created_at, the same day readDay's range bounds. - const { rows } = await pool.query<{ day: string; n: number }>( - `SELECT to_char(day, 'YYYY-MM-DD') AS day, n - FROM (SELECT (created_at AT TIME ZONE 'UTC')::date AS day, count(*)::int AS n - FROM event - GROUP BY 1) AS days`, - ); + const client = await pool.connect(); + const boundMs = Number(pool.options.statement_timeout ?? STATEMENT_TIMEOUT_MS) + graceMs; + + try { + // The UTC date of created_at, the same day readDay's range bounds. + const { rows } = await callWithin( + client.query<{ day: string; n: number }>( + `SELECT to_char(day, 'YYYY-MM-DD') AS day, n + FROM (SELECT (created_at AT TIME ZONE 'UTC')::date AS day, count(*)::int AS n + FROM event + GROUP BY 1) AS days`, + ), + boundMs, + () => new Error(`the day counts ran past ${boundMs / 1000} s`), + ); - return new Map(rows.map(({ day, n }) => [day, n])); + client.release(); + + return new Map(rows.map(({ day, n }) => [day, n])); + } catch (error) { + // Destroyed, since a query that ran past its bound is still in flight. + client.release(error instanceof Error ? error : true); + throw error; + } }, async *readDay(day) { const to = `${dayOf(Date.parse(day) + DAY_MS)}T00:00:00Z`; @@ -437,29 +522,40 @@ export const pgEventSource = (pool: Pool, pageSize = PAGE_SIZE): EventSource => // (midnight, 0) precedes the day's first row. `props` keeps only the keys some // event keeps; exportProps picks each event's own. let after = { createdAt: `${day}T00:00:00Z`, id: 0 }; + const deadline = Date.now() + readMs; + const client = await pool.connect(); + let failure: Error | undefined; - for (;;) { - const { rows } = await pool.query( - `SELECT id::float8 AS id, - to_char(created_at AT TIME ZONE 'UTC', - 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS "createdAt", - name, user_id AS "userId", visitor_id AS "visitorId", page, - coalesce((SELECT jsonb_object_agg(key, value) - FROM jsonb_each(props) - WHERE key = ANY($4::text[])), '{}') AS props - FROM event - WHERE created_at >= $1 AND created_at < $2 AND (created_at, id) > ($1, $3) - ORDER BY created_at, id - LIMIT $5`, - [after.createdAt, to, after.id, KEPT_PROPS, pageSize], - ); + try { + for (;;) { + const { rows } = await callWithin( + client.query(READ_DAY_SQL, [ + after.createdAt, + to, + after.id, + KEPT_PROPS, + pageSize, + ]), + deadline - Date.now() + graceMs, + () => new Error(`read ran past its ${readMs / 1000} s deadline`), + ); - if (rows.length === 0) { - return; - } + if (rows.length > 0) { + yield rows; + } + + // A short page is the day's last. + if (rows.length < pageSize) { + return; + } - yield rows; - after = rows[rows.length - 1]!; + after = rows[rows.length - 1]!; + } + } catch (error) { + failure = error instanceof Error ? error : new Error(String(error)); + throw error; + } finally { + client.release(failure); } }, }); @@ -476,23 +572,34 @@ export const bigQueryWarehouse = ({ const dataset = bigquery.dataset(datasetId); const table = dataset.table('event'); - // A query job fails past 1 GB billed or past 2 minutes, well inside the - // Job's 1800 s task timeout, and the run fails with it. + // A query job fails past 1 GB billed or past BIGQUERY_TIMEOUT_MS, and + // the run fails with it. + // The client takes no overall timeout for a call, so the Job stops + // waiting at BIGQUERY_CALL_MS. + const call = (work: Promise): Promise => + callWithin( + work, + BIGQUERY_CALL_MS, + () => new Error(`a BigQuery call ran past ${BIGQUERY_CALL_MS / 1000} s`), + ); + const query = async (sql: string, params?: Record): Promise => { try { - const [rows] = await bigquery.query({ - query: sql, - params, - maximumBytesBilled: String(MAX_BYTES_BILLED), - jobTimeoutMs: 120_000, - }); + const [rows] = await call( + bigquery.query({ + query: sql, + params, + maximumBytesBilled: String(MAX_BYTES_BILLED), + jobTimeoutMs: BIGQUERY_TIMEOUT_MS, + }), + ); return rows; } catch (error) { if (/bytes billed/i.test(messageOf(error))) { throw new Error( `a query exceeded the ${MAX_BYTES_BILLED / 1e9} GB maximumBytesBilled cap, ` + - `which staleDays reaches at about ${(MAX_BYTES_BILLED / STALE_BYTES_PER_ROW / 1e6).toFixed(1)} million event rows`, + `which the days query reaches at about ${(MAX_BYTES_BILLED / DAYS_BYTES_PER_ROW / 1e6).toFixed(1)} million event rows`, { cause: error }, ); } @@ -501,69 +608,57 @@ export const bigQueryWarehouse = ({ } }; - const dayCounts = async (): Promise => { - // Each partition's row count from the table's metadata, which a - // finished load updates, so the query scans no table data. - // partition_id is the UTC day as YYYYMMDD, and the pattern leaves - // out the special partitions such as __NULL__. - const rows = await query<{ partition_id: string; total_rows: unknown }>( - `SELECT partition_id, total_rows - FROM \`${projectId}.${datasetId}.INFORMATION_SCHEMA.PARTITIONS\` - WHERE table_name = 'event' AND total_rows > 0 - AND REGEXP_CONTAINS(partition_id, r'^\\d{8}$')`, - ); - - return new Map( - rows.map((row) => [ - row.partition_id.replace(/^(\d{4})(\d{2})(\d{2})$/, '$1-$2-$3'), - Number(row.total_rows), - ]), - ); - }; - return { async ensureTable() { - const [exists] = await table.exists(); + const [exists] = await call(table.exists()); if (!exists) { - await dataset.createTable('event', { - schema: { fields: EXPORT_SCHEMA }, - timePartitioning: { type: 'DAY', field: 'created_at' }, - }); + await call( + dataset.createTable('event', { + schema: { fields: EXPORT_SCHEMA }, + timePartitioning: { type: 'DAY', field: 'created_at' }, + }), + ); } }, - dayCounts, - async staleDays(rules) { - const [metadata] = await table.getMetadata(); + async days(rules) { + const [metadata] = await call(table.getMetadata()); const fields: TableField[] = metadata.schema?.fields ?? []; - // A table created without the column holds no digest, so every // day with rows is stale. The next load adds the column. - if (!fields.some((field) => field.name === 'export_rules')) { - return new Set((await dayCounts()).keys()); - } - - // Scans the created_at and export_rules columns, about - // STALE_BYTES_PER_ROW bytes a row. - const rows = await query<{ day: string }>( - `SELECT FORMAT_DATE('%F', DATE(created_at)) AS day + const stale = fields.some((field) => field.name === 'export_rules') + ? 'LOGICAL_OR(export_rules IS DISTINCT FROM @rules)' + : 'TRUE'; + + // Counts the rows themselves, one group per partition day, and + // scans the created_at and export_rules columns, about + // DAYS_BYTES_PER_ROW bytes a row. + const rows = await query<{ day: string; n: unknown; stale: boolean }>( + `SELECT FORMAT_DATE('%F', DATE(created_at)) AS day, + COUNT(*) AS n, + ${stale} AS stale FROM \`${projectId}.${datasetId}.event\` - WHERE export_rules IS DISTINCT FROM @rules GROUP BY day`, { rules }, ); - return new Set(rows.map(({ day }) => day)); + return { + counts: new Map(rows.map(({ day, n }) => [day, Number(n)])), + stale: new Set(rows.filter((row) => row.stale).map(({ day }) => day)), + }; }, async replaceDay(day, file) { - // Resolves once the load job is done, and throws if it failed. - await dataset.table(`event$${day.replaceAll('-', '')}`).load(file, { - sourceFormat: 'NEWLINE_DELIMITED_JSON', - writeDisposition: 'WRITE_TRUNCATE', - schema: { fields: EXPORT_SCHEMA }, - // A NULLABLE column added to EXPORT_SCHEMA reaches the table - // on the next load (ExportField). - schemaUpdateOptions: ['ALLOW_FIELD_ADDITION'], + await loadWithin({ + table: dataset.table(`event$${day.replaceAll('-', '')}`), + file, + metadata: { + sourceFormat: 'NEWLINE_DELIMITED_JSON', + writeDisposition: 'WRITE_TRUNCATE', + schema: { fields: EXPORT_SCHEMA }, + // A NULLABLE column added to EXPORT_SCHEMA reaches the + // table on the next load (ExportField). + schemaUpdateOptions: ['ALLOW_FIELD_ADDITION'], + }, }); }, }; @@ -572,6 +667,20 @@ export const bigQueryWarehouse = ({ const messageOf = (error: unknown): string => error instanceof Error ? error.message : String(error); +/** + * What a Job run has in flight, which runAndExit names when its deadline + * fires: the phase, the days the event export has yet to start, and the + * faults so far. + */ +export type JobProgress = { + phase: string; + unstarted: readonly string[]; + /** The failed and short days so far, as the run's report names them. */ + faults: () => readonly string[]; +}; + +const jobProgress = (): JobProgress => ({ phase: 'the start', unstarted: [], faults: () => [] }); + /** * One export run over the days exportDays picks. Each day is written to * `scratchFile` page by page before its load, so memory holds one page of @@ -593,11 +702,10 @@ const messageOf = (error: unknown): string => * days still load, and the run then rejects naming every such day with its * error. Each row carries the run's exportRules digest, so a day that * fails under new rules still holds the old digest and is the one day the - * next run reloads for them. A run starts no new day once `budgetMs` has - * passed since it began (`now` is its clock); it ends the day in hand, then - * rejects naming the days it did not reach as well as the failed and short - * ones, and the next run picks those days up. Logs each day as it is done - * and returns the rows read per day. + * next run reloads for them. A read or load past its bound fails its day + * the same way. `progress` names the phase in flight, the days not yet + * started and the faults so far, for runAndExit's deadline. Logs each day + * as it is done and returns the rows read per day. */ export const runExport = async ({ source, @@ -606,8 +714,7 @@ export const runExport = async ({ today, scratchFile, log, - budgetMs = EXPORT_BUDGET_MS, - now = Date.now, + progress = jobProgress(), }: { source: EventSource; warehouse: Warehouse; @@ -615,38 +722,41 @@ export const runExport = async ({ today: string; scratchFile: string; log: (line: string) => void; - budgetMs?: number; - now?: () => number; + progress?: JobProgress; }): Promise<{ day: string; rows: number }[]> => { - const startedAt = now(); const rules = exportRules(hmacKey); + progress.phase = 'the event table check and day counts'; await warehouse.ensureTable(); - const [events, partitions, stale] = await Promise.all([ + const [events, { counts: partitions, stale }] = await Promise.all([ source.dayCounts(), - warehouse.dayCounts(), - warehouse.staleDays(rules), + warehouse.days(rules), ]); const days = exportDays({ today, events, partitions, stale }); const read: { day: string; rows: number }[] = []; const failed: { day: string; error: unknown }[] = []; const short: string[] = []; - const unreached: string[] = []; + const faultsSoFar = () => [ + ...failed.map(({ day, error }) => `${day}: ${messageOf(error)}`), + ...(short.length > 0 ? [`short days: ${short.join(', ')}`] : []), + ]; + const earlierFaults = progress.faults; - try { - for (const day of days) { - if (now() - startedAt >= budgetMs) { - unreached.push(day); - continue; - } + progress.faults = () => [...earlierFaults(), ...faultsSoFar()]; + try { + for (const [index, day] of days.entries()) { + progress.unstarted = days.slice(index + 1); let rows = 0; const held = partitions.get(day) ?? 0; const counted = events.get(day) ?? 0; try { - await fs.writeFile(scratchFile, ''); + // Unlinked, so a load that ran past its timeout and still + // uploads keeps reading the day before's file. + await fs.rm(scratchFile, { force: true }); + progress.phase = `the read of ${day}`; for await (const page of source.readDay(day)) { const lines = page.map( @@ -658,6 +768,7 @@ export const runExport = async ({ } if (rows > 0 && rows >= held) { + progress.phase = `the load of ${day}`; await warehouse.replaceDay(day, scratchFile); } } catch (error) { @@ -685,18 +796,12 @@ export const runExport = async ({ read.push({ day, rows }); } } finally { + progress.phase = "the run's report"; + progress.unstarted = []; await fs.rm(scratchFile, { force: true }); } - const faults = [ - ...failed.map(({ day, error }) => `${day}: ${messageOf(error)}`), - ...(short.length > 0 ? [`short days: ${short.join(', ')}`] : []), - ...(unreached.length > 0 - ? [ - `days not reached in the ${budgetMs / 60_000} minute budget: ${unreached.join(', ')}`, - ] - : []), - ]; + const faults = faultsSoFar(); if (faults.length > 0) { throw new Error(`event export ${faults.join('; ')}`, { @@ -712,36 +817,99 @@ export const runExport = async ({ /** * The export Job's run (db/export-events.ts): the account export, then the - * event export, which runs whether or not the account export failed. Logs - * an account export failure as it happens and rejects with it once the - * event export is done, so a failing one leaves the other's tables - * current. An event export failure rejects as it is. + * event export, which runs whether or not the account export failed, so a + * failing one leaves the other's tables current. Rejects once both are + * done with the one that failed, or with both in one AggregateError. */ export const runExportJob = async ({ exportAccounts, exportEvents, - logError, + progress = jobProgress(), }: { exportAccounts: () => Promise; exportEvents: () => Promise; - logError: (error: unknown) => void; + progress?: JobProgress; }): Promise => { let accountFailure: Error | null = null; + progress.phase = 'the account export'; + progress.faults = () => (accountFailure !== null ? [accountFailure.message] : []); + try { await exportAccounts(); } catch (error) { - accountFailure = new Error('account export failed', { cause: error }); - logError(error); + accountFailure = new Error(`account export failed: ${messageOf(error)}`, { cause: error }); } - await exportEvents(); + try { + await exportEvents(); + } catch (error) { + throw accountFailure === null + ? error + : new AggregateError( + [accountFailure, error], + `${accountFailure.message}; ${messageOf(error)}`, + ); + } + + progress.phase = "the run's report"; if (accountFailure !== null) { throw accountFailure; } }; +/** + * Runs the Job's `main`, then exits the process with 0, or with 1 after + * logging one error that names the phase, so a socket left open, such as a + * load upload past its timeout or the pool's own connections, cannot hold + * the Job to its task timeout. A rejection is logged with the phase it + * left. A `main` still running at `deadlineMs` is stopped: the error names + * the phase in flight, the days not started and the faults so far, after + * the lines each finished day logged. Cloud Run's stdout and stderr are + * pipes, which Node writes synchronously on Linux, so every logged line is + * out before the exit. + */ +export const runAndExit = async ( + main: (progress: JobProgress) => Promise, + { + deadlineMs = JOB_DEADLINE_MS, + exit = process.exit, + logError, + }: { + deadlineMs?: number; + exit?: (code: number) => void; + logError: (error: unknown) => void; + }, +): Promise => { + const progress = jobProgress(); + let code = 0; + + try { + if ((await within(main(progress), deadlineMs)) === TIMED_OUT) { + const report = [ + `export Job stopped at its deadline, ${deadlineMs / 1000} s from its start, during ${progress.phase}`, + ...(progress.unstarted.length > 0 + ? [`days not started: ${progress.unstarted.join(', ')}`] + : []), + ...progress.faults(), + ]; + + logError(new Error(report.join('; '))); + code = 1; + } + } catch (error) { + logError( + new Error(`export Job failed during ${progress.phase}: ${messageOf(error)}`, { + cause: error, + }), + ); + code = 1; + } + + exit(code); +}; + /** The export's own settings. The database settings are the runtime's (db/connection.ts). */ export const readExportEnv = () => ({ projectId: requiredEnv('EXPORT_BQ_PROJECT'), diff --git a/db/export-events.ts b/db/export-events.ts index aaabd3e..c8c4219 100644 --- a/db/export-events.ts +++ b/db/export-events.ts @@ -2,6 +2,7 @@ import os from 'node:os'; import path from 'node:path'; import { BigQuery } from '@google-cloud/bigquery'; import { bigQueryAccountTable, readAccountDaily, runAccountExport } from './account-export'; +import { STATEMENT_TIMEOUT_MS } from './bounded'; import { buildRuntimePool, connectCloudSqlRuntime, @@ -10,8 +11,11 @@ import { } from './connection'; import { bigQueryWarehouse, + CONNECTION_TIMEOUT_MS, + type JobProgress, pgEventSource, readExportEnv, + runAndExit, runExport, runExportJob, } from './event-export'; @@ -19,8 +23,9 @@ import { // The event export Job's command (`node dist/export-events.cjs`, see // deploy/Dockerfile). It reads the database as the runtime does, through // WEBSITE_DB_* over the Cloud SQL connector, or DATABASE_URL locally, and -// runs the account export, then the event export (runExportJob). -async function main(): Promise { +// runs the account export, then the event export (runExportJob), under +// runAndExit's deadline, which names the phase in `progress` if it fires. +async function main(progress: JobProgress): Promise { const settings = readExportEnv(); const connection = resolveRuntimeConnection(); const cloudSql = @@ -31,46 +36,42 @@ async function main(): Promise { } // One connection suffices, since the run sends one query at a time. A - // query that runs past 5 minutes, the per-day count over the whole - // table, a page read or an account read, fails that read. The event - // export starts no new day after 20 minutes (EXPORT_BUDGET_MS), so - // days that each wait out the timeout still end the run, with a - // report naming the days it did not reach, inside the Job's 1800 s - // task timeout and before the next hourly run starts. - const pool = buildRuntimePool(process.env, { max: 1, statement_timeout: 300_000 }); + // query past 5 minutes fails on the server, and a connection not made + // in 30 seconds fails the query that asked for it. Each call also has a + // bound on the Job's side (db/bounded.ts). runAndExit stops the run at + // 28 minutes whatever is in flight, inside the Job's 1800 s task timeout + // and before the next hourly run starts, and its exit closes the pool + // and the connector. + const pool = buildRuntimePool(process.env, { + max: 1, + statement_timeout: STATEMENT_TIMEOUT_MS, + connectionTimeoutMillis: CONNECTION_TIMEOUT_MS, + }); const bigquery = new BigQuery({ projectId: settings.projectId }); const log = (line: string) => console.log(line); - try { - await runExportJob({ - exportAccounts: () => - runAccountExport({ - read: () => readAccountDaily(pool), - replace: bigQueryAccountTable({ datasetId: settings.datasetId, bigquery }), - scratchFile: path.join(os.tmpdir(), `account-export-${process.pid}.ndjson`), - log, - }), - exportEvents: () => - runExport({ - source: pgEventSource(pool), - warehouse: bigQueryWarehouse({ ...settings, bigquery }), - hmacKey: settings.hmacKey, - today: new Date().toISOString().slice(0, 10), - scratchFile: path.join(os.tmpdir(), `event-export-${process.pid}.ndjson`), - log, - }), - logError: (error) => - console.error('ERROR:', error instanceof Error ? error.stack : error), - }); - } finally { - // The pool first, so its connections close over the connector's tunnel. - await pool.end(); - cloudSql?.connector.close(); - } + await runExportJob({ + exportAccounts: () => + runAccountExport({ + read: () => readAccountDaily(pool), + replace: bigQueryAccountTable({ datasetId: settings.datasetId, bigquery }), + scratchFile: path.join(os.tmpdir(), `account-export-${process.pid}.ndjson`), + log, + }), + exportEvents: () => + runExport({ + source: pgEventSource(pool), + warehouse: bigQueryWarehouse({ ...settings, bigquery }), + hmacKey: settings.hmacKey, + today: new Date().toISOString().slice(0, 10), + progress, + scratchFile: path.join(os.tmpdir(), `event-export-${process.pid}.ndjson`), + log, + }), + progress, + }); } -main().catch((error: unknown) => { - console.error('ERROR:', error instanceof Error ? error.stack : error); - process.exitCode = 1; -}); +// console.error prints the error's stack and its cause's. +void runAndExit(main, { logError: (error) => console.error('ERROR:', error) }); diff --git a/db/test/account-export.test.ts b/db/test/account-export.test.ts index 7c010f7..37ad9bb 100644 --- a/db/test/account-export.test.ts +++ b/db/test/account-export.test.ts @@ -248,9 +248,9 @@ const fakeBigQuery = () => { const bigquery = { dataset: (dataset: string) => ({ table: (table: string) => ({ - load: async (file: string, metadata: Record) => { + createLoadJob: async (file: string, metadata: Record) => { loads.push({ table: `${dataset}.${table}`, file, metadata }); - return [{}]; + return [{ promise: async () => [{}] }]; }, }), }), @@ -278,6 +278,35 @@ test('bigQueryAccountTable replaces the whole table in one load', async () => { assert.ok(ACCOUNT_SCHEMA.every((field) => field.description)); }); +test( + 'readAccountDaily fails a read that hangs past its bound and destroys its client', + { timeout: 5_000 }, + async () => { + const queries: string[] = []; + const released: unknown[] = []; + const pool = { + connect: async () => ({ + // BEGIN answers, and the view's read never does, as over a stalled tunnel. + query: (text: string) => { + queries.push(text.trim().split(/\s/, 1)[0]!); + return text.startsWith('BEGIN') + ? Promise.resolve({ rows: [] }) + : new Promise(() => undefined); + }, + release: (error?: unknown) => released.push(error), + }), + } as unknown as Pool; + + await assert.rejects( + readAccountDaily(pool, { timeoutMs: 20 }), + /account read ran past 0.02 s/, + ); + assert.deepEqual(queries, ['BEGIN', 'SELECT']); + assert.equal(released.length, 1); + assert.ok(released[0] instanceof Error); + }, +); + const ROWS: AccountRow[] = [ { day: '2026-10-20', organisation: 'Personal email', accounts: 4 }, { day: '2026-10-21', organisation: 'one.org', accounts: 3 }, diff --git a/db/test/bounded.test.ts b/db/test/bounded.test.ts new file mode 100644 index 0000000..aa99f21 --- /dev/null +++ b/db/test/bounded.test.ts @@ -0,0 +1,62 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { Table } from '@google-cloud/bigquery'; +import { loadWithin } from '../bounded.ts'; + +// A table whose load job settles as `promise` does and whose cancel as `cancel` does. +const fakeTable = (promise: () => Promise, cancel: () => Promise) => { + const calls: string[] = []; + const table = { + createLoadJob: async () => { + calls.push('load'); + return [ + { + promise, + cancel: () => { + calls.push('cancel'); + return cancel(); + }, + }, + ]; + }, + } as unknown as Table; + + return { table, calls }; +}; + +const never = () => new Promise(() => undefined); + +test('loadWithin bounds a load and its cancel', { timeout: 5_000 }, async (t) => { + const cases: [string, () => Promise, () => Promise, RegExp | null][] = [ + ['a load that finishes', async () => [{}], async () => [{}], null], + [ + 'a load that fails', + async () => Promise.reject(new Error('load job failed')), + never, + /load job failed/, + ], + ['a load past its timeout', never, async () => [{}], /ran past/], + ['a load past its timeout whose cancel hangs', never, never, /ran past/], + ]; + + for (const [label, promise, cancel, error] of cases) { + await t.test(label, async () => { + const { table, calls } = fakeTable(promise, cancel); + const load = loadWithin({ + table, + file: '/tmp/day.ndjson', + metadata: {}, + timeoutMs: 20, + cancelTimeoutMs: 20, + }); + + if (error === null) { + await load; + } else { + await assert.rejects(load, error); + } + + assert.deepEqual(calls, /ran past/.test(String(error)) ? ['load', 'cancel'] : ['load']); + }); + } +}); diff --git a/db/test/event-export.test.ts b/db/test/event-export.test.ts index cd3abf2..80f596e 100644 --- a/db/test/event-export.test.ts +++ b/db/test/event-export.test.ts @@ -14,9 +14,13 @@ import { exportDays, type ExportField, exportRow, + type ExportRow, exportRules, pgEventSource, runExport, + ROW_SHAPE, + ROW_SHAPE_VERSION, + runAndExit, runExportJob, type SourceEvent, type Warehouse, @@ -98,6 +102,19 @@ test('exportRules changes with the HMAC key', () => { assert.notEqual(exportRules('rotated-key'), RULES); }); +test('exportRules changes with the SQL that reads a day', () => { + // The reader test checks that readDay sends ROW_SHAPE.READ_DAY_SQL. + const sql = ROW_SHAPE.READ_DAY_SQL; + + ROW_SHAPE.READ_DAY_SQL = `${sql} `; + + try { + assert.notEqual(exportRules(KEY), RULES); + } finally { + ROW_SHAPE.READ_DAY_SQL = sql; + } +}); + test('exportRow drops a value its rule refuses', async (t) => { const cases: [string, Record][] = [ [ @@ -184,6 +201,111 @@ test('exportRow exports a share id as its digest, so share events still join', ( assert.deepEqual(refused.props, {}); }); +// A fixed sample that reaches every rule and branch of exportRow. +const GOLDEN_SAMPLE: SourceEvent[] = [ + sourceEvent({ + id: 1, + name: 'search.query', + userId: 'user-1', + visitorId: 'visitor-1', + page: '/data?q=aid%20to%20kenya#top', + props: { + searchId: '0b7c6f1e-2d4a-4f7e-9a51-3c1d2e4f5a6b', + post: 0, + source: 'typed', + edited: false, + ms: 812, + outcome: 'definite', + kind: 'resolution', + query: 'aid to kenya', + engineMs: -1, + traceId: 'two words', + }, + }), + sourceEvent({ + id: 2, + createdAt: '2026-10-21T00:00:00.000001Z', + name: 'share.open', + visitorId: 'visitor-2', + page: '/s/aB3dE5fG7h/download/preparing', + props: { shareId: 'aB3dE5fG7h', kind: 'card', mode: 'snapshot', host: 'example.org' }, + }), + sourceEvent({ id: 3, name: 'feedback.submit', page: '/a page', props: { kind: 'widget' } }), + sourceEvent({ id: 4, name: 'retired.event', page: '/embed/x1', props: { kind: 'card' } }), +]; + +// Each ROW_SHAPE_VERSION's rows for GOLDEN_SAMPLE under KEY, with rules +// `r`. A change to the rows adds the next version with its rows, and an +// entry never changes. +const GOLDEN_ROWS: Record = { + 1: [ + { + id: 1, + created_at: '2026-10-20T13:04:05.123456Z', + name: 'search.query', + page: '/data', + signed_in: true, + visitor_hash: '77a3c0b044582bfb514a3a3ad8521316635e2256261861c83b7498ee2e420f65', + user_hash: 'e0e98314ed3ee6ea685abd3f0862870e2a07b68d95d053c7f483cda192ae40c4', + props: { + searchId: '0b7c6f1e-2d4a-4f7e-9a51-3c1d2e4f5a6b', + post: 0, + source: 'typed', + edited: false, + ms: 812, + outcome: 'definite', + kind: 'resolution', + }, + export_rules: 'r', + }, + { + id: 2, + created_at: '2026-10-21T00:00:00.000001Z', + name: 'share.open', + page: '/s/:id/download/preparing', + signed_in: false, + visitor_hash: '656d62821b3df7e629e0d8af562ac6a21302f71aeed8fa018b46c7fa00d54f88', + user_hash: null, + props: { + shareId: '7e8148cb3a5192589abc81da275c6e151ebb4c2d702234f8a68e870138326bc4', + kind: 'card', + mode: 'snapshot', + host: 'example.org', + }, + export_rules: 'r', + }, + { + id: 3, + created_at: '2026-10-20T13:04:05.123456Z', + name: 'feedback.submit', + page: null, + signed_in: false, + visitor_hash: null, + user_hash: null, + props: { kind: 'widget' }, + export_rules: 'r', + }, + { + id: 4, + created_at: '2026-10-20T13:04:05.123456Z', + name: 'retired.event', + page: '/embed/:id', + signed_in: false, + visitor_hash: null, + user_hash: null, + props: {}, + export_rules: 'r', + }, + ], +}; + +test('exportRow writes the rows pinned to the current ROW_SHAPE_VERSION', () => { + assert.deepEqual( + GOLDEN_SAMPLE.map((row) => exportRow(row, KEY, 'r')), + GOLDEN_ROWS[ROW_SHAPE_VERSION], + ); +}); + const counts = (days: Record) => new Map(Object.entries(days)); test('exportDays replaces from the day before the latest loaded day through today', async (t) => { @@ -228,7 +350,17 @@ test('exportDays replaces from the day before the latest loaded day through toda partitions: counts({ '2026-10-17': 1, '2026-10-20': 2, '2026-10-23': 1 }), stale: new Set(['2026-10-17', '2026-10-20']), }, - ['2026-10-17', '2026-10-20', '2026-10-22', '2026-10-23'], + ['2026-10-22', '2026-10-23', '2026-10-17', '2026-10-20'], + ], + [ + 'a backlog of stale and mismatched days, oldest first, after the window', + { + today: '2026-10-23', + events: counts({ '2026-10-15': 2, '2026-10-20': 2, '2026-10-23': 1 }), + partitions: counts({ '2026-10-15': 1, '2026-10-20': 2, '2026-10-23': 1 }), + stale: new Set(['2026-10-20', '2026-10-12', '2026-10-23']), + }, + ['2026-10-22', '2026-10-23', '2026-10-12', '2026-10-15', '2026-10-20'], ], [ 'a month boundary', @@ -255,25 +387,25 @@ test('exportDays adds every earlier day whose counts differ', async (t) => { 'a short day stays once later days load', { '2026-10-19': 5, ...loaded }, { '2026-10-19': 4, ...loaded }, - ['2026-10-19', '2026-10-22', '2026-10-23'], + ['2026-10-22', '2026-10-23', '2026-10-19'], ], [ 'a day missing from BigQuery', { '2026-10-15': 2, ...loaded }, loaded, - ['2026-10-15', '2026-10-22', '2026-10-23'], + ['2026-10-22', '2026-10-23', '2026-10-15'], ], [ 'a day BigQuery holds more of than the source', { '2026-10-17': 2, ...loaded }, { '2026-10-17': 3, ...loaded }, - ['2026-10-17', '2026-10-22', '2026-10-23'], + ['2026-10-22', '2026-10-23', '2026-10-17'], ], [ 'a day BigQuery holds that the source lacks', loaded, { '2026-10-16': 1, ...loaded }, - ['2026-10-16', '2026-10-22', '2026-10-23'], + ['2026-10-22', '2026-10-23', '2026-10-16'], ], ['days that match', loaded, loaded, ['2026-10-22', '2026-10-23']], ]; @@ -306,19 +438,19 @@ const fakeWarehouse = ( async ensureTable() { calls.push('ensureTable'); }, - async dayCounts() { - return new Map([...partitions].map(([day, rows]) => [day, rows.length])); - }, - async staleDays(rules) { - return new Set( - [...partitions] - .filter(([, rows]) => - rows.some( - (row) => (row as { export_rules?: string }).export_rules !== rules, - ), - ) - .map(([day]) => day), - ); + async days(rules) { + return { + counts: new Map([...partitions].map(([day, rows]) => [day, rows.length])), + stale: new Set( + [...partitions] + .filter(([, rows]) => + rows.some( + (row) => (row as { export_rules?: string }).export_rules !== rules, + ), + ) + .map(([day]) => day), + ), + }; }, async replaceDay(day, file) { if (failLoad.includes(day)) { @@ -502,9 +634,9 @@ test('runExport rechecks an earlier day whose counts differ until they match', a read: [at(1, '2026-10-19'), at(2, '2026-10-19')], calls: [ 'ensureTable', - 'replace 2026-10-19', 'replace 2026-10-22', 'replace 2026-10-23', + 'replace 2026-10-19', ], short: null, }, @@ -523,9 +655,9 @@ test('runExport rechecks an earlier day whose counts differ until they match', a counted: { '2026-10-19': 3 }, calls: [ 'ensureTable', - 'replace 2026-10-19', 'replace 2026-10-22', 'replace 2026-10-23', + 'replace 2026-10-19', ], short: /2026-10-19\D+2\D+3\b/, }, @@ -671,9 +803,9 @@ test('runExport reloads the days holding rows under other rules, and retries a f ); assert.deepEqual(first.calls, [ 'ensureTable', - 'replace 2026-10-20', 'replace 2026-10-21', 'replace 2026-10-22', + 'replace 2026-10-20', ]); assert.deepEqual(partitions.get('2026-10-20'), [exportRow(at(3, '2026-10-20'), KEY, RULES)]); @@ -689,9 +821,9 @@ test('runExport reloads the days holding rows under other rules, and retries a f }); assert.deepEqual(second.calls, [ 'ensureTable', - 'replace 2026-10-18', 'replace 2026-10-21', 'replace 2026-10-22', + 'replace 2026-10-18', ]); }); @@ -726,54 +858,111 @@ test('runExport names the short days beside a failed day', async () => { assert.equal(partitions.get('2026-10-21')?.length, 2); }); -test('runExport starts no new day past its budget and fails naming the days it did not reach', async () => { - const partitions = new Map(); - const { warehouse, calls } = fakeWarehouse(partitions, { failLoad: ['2026-10-21'] }); - const clock = { ms: 0 }; - const source = fakeSource({ - '2026-10-20': [[sourceEvent({ id: 1, createdAt: '2026-10-20T08:00:00.000000Z' })]], - '2026-10-21': [[sourceEvent({ id: 2, createdAt: '2026-10-21T08:00:00.000000Z' })]], - '2026-10-22': [[sourceEvent({ id: 3, createdAt: '2026-10-22T08:00:00.000000Z' })]], - '2026-10-23': [[sourceEvent({ id: 4, createdAt: '2026-10-23T08:00:00.000000Z' })]], - }); - - await assert.rejects( - runExport({ - source: { - ...source, - // Each day read costs 4 minutes of the injected clock. - readDay: (day) => { - clock.ms += 240_000; - return source.readDay(day); - }, +test( + 'a page query that hangs past the read deadline fails its day, destroys its client, and the next day loads', + { timeout: 5_000 }, + async () => { + const partitions = new Map(); + const { warehouse, calls } = fakeWarehouse(partitions); + // Each client's queries by text, and what it was released with. + const clients: { queries: string[]; released?: Error | 'reused' }[] = []; + const pool = { + options: {}, + connect: async () => { + const client: (typeof clients)[number] = { queries: [] }; + clients.push(client); + + return { + query: (text: string, values?: unknown[]) => { + client.queries.push(text.split(/\s/, 1)[0]!); + + if (text.includes('count(*)')) { + return Promise.resolve({ + rows: [ + { day: '2026-10-21', n: 1 }, + { day: '2026-10-22', n: 1 }, + ], + }); + } + + if (text !== ROW_SHAPE.READ_DAY_SQL) { + return Promise.resolve({ rows: [] }); + } + + // 21 October's page never answers, as over a stalled tunnel. + return String(values?.[0]).startsWith('2026-10-21') + ? new Promise(() => undefined) + : Promise.resolve({ + rows: [ + sourceEvent({ + id: 2, + createdAt: '2026-10-22T08:00:00.000000Z', + }), + ], + }); + }, + release: (error?: Error) => { + client.released = error ?? 'reused'; + }, + }; }, - warehouse, - hmacKey: KEY, - today: '2026-10-23', - scratchFile: scratch(), - log: () => undefined, - budgetMs: 600_000, - now: () => clock.ms, - }), - (error: unknown) => { - assert.ok(error instanceof Error); - assert.match(error.message, /2026-10-21: load job failed/); - assert.match(error.message, /not reached\D+10 minute budget: 2026-10-23$/); - return true; - }, - ); + } as unknown as Pool; + const started = Date.now(); - assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-20', 'replace 2026-10-22']); -}); + await assert.rejects( + runExport({ + source: pgEventSource(pool, { readMs: 50, graceMs: 20 }), + warehouse, + hmacKey: KEY, + today: '2026-10-22', + scratchFile: scratch(), + log: () => undefined, + }), + /event export 2026-10-21: read ran past its 0.05 s deadline$/, + ); + + // The 50 ms deadline plus the 20 ms grace, with room for a slow machine. + assert.ok(Date.now() - started < 1_000); + // The day counts' client, then each day's. + assert.deepEqual(clients[0], { queries: ['SELECT'], released: 'reused' }); + assert.deepEqual(clients[1]?.queries, ['SELECT']); + assert.ok(clients[1]?.released instanceof Error); + assert.deepEqual(clients[2], { queries: ['SELECT'], released: 'reused' }); + assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-22']); + }, +); + +test( + 'pgEventSource dayCounts fails a count that hangs past its bound and destroys its client', + { timeout: 5_000 }, + async () => { + const released: unknown[] = []; + const pool = { + options: { statement_timeout: 20 }, + connect: async () => ({ + // The count never answers, as over a stalled tunnel. + query: () => new Promise(() => undefined), + release: (error?: unknown) => released.push(error), + }), + } as unknown as Pool; + + await assert.rejects( + pgEventSource(pool, { graceMs: 20 }).dayCounts(), + /day counts ran past 0.04 s/, + ); + assert.equal(released.length, 1); + assert.ok(released[0] instanceof Error); + }, +); test('bigQueryWarehouse loads a day into its partition and lets the schema gain columns', async () => { const loads: { table: string; file: string; metadata: Record }[] = []; - const fake = { + const bigquery = { dataset: () => ({ table: (table: string) => ({ - load: async (file: string, metadata: Record) => { + createLoadJob: async (file: string, metadata: Record) => { loads.push({ table, file, metadata }); - return [{}]; + return [{ promise: async () => [{}] }]; }, }), }), @@ -781,7 +970,7 @@ test('bigQueryWarehouse loads a day into its partition and lets the schema gain const warehouse = bigQueryWarehouse({ projectId: 'p', datasetId: 'd', - bigquery: fake as unknown as BigQuery, + bigquery: bigquery as unknown as BigQuery, }); await warehouse.replaceDay('2026-10-22', '/tmp/day.ndjson'); @@ -801,7 +990,7 @@ test('EXPORT_SCHEMA admits a REQUIRED column only among those the table was crea assert.ok(required && nullable); }); -test('bigQueryWarehouse creates the table, finds stale days and counts from partition metadata', async (t) => { +test('bigQueryWarehouse creates the table, and counts rows and finds stale days per day', async (t) => { const fakeTable = ({ exists, fields = [] }: { exists: boolean; fields?: string[] }) => { const calls: unknown[] = []; const table = { @@ -817,15 +1006,15 @@ test('bigQueryWarehouse creates the table, finds stale days and counts from part }, }), query: async ({ query, params }: { query: string; params?: unknown }) => { - calls.push(['query', params]); - return query.includes('PARTITIONS') - ? [ - [ - { partition_id: '20261021', total_rows: 3 }, - { partition_id: '20261022', total_rows: '1434' }, - ], - ] - : [[{ day: '2026-10-21' }]]; + // The table itself, so each count is exact. + calls.push(['query', query.match(/FROM `([^`]+)`/)?.[1], params]); + const stale = fields.includes('export_rules') ? [true, false] : [true, true]; + return [ + [ + { day: '2026-10-21', n: 3, stale: stale[0] }, + { day: '2026-10-22', n: 1434, stale: stale[1] }, + ], + ]; }, }; const warehouse = bigQueryWarehouse({ @@ -836,6 +1025,10 @@ test('bigQueryWarehouse creates the table, finds stale days and counts from part return { warehouse, calls }; }; + const counts = new Map([ + ['2026-10-21', 3], + ['2026-10-22', 1434], + ]); await t.test('a missing table is created with the schema', async () => { const { warehouse, calls } = fakeTable({ exists: false }); @@ -844,18 +1037,24 @@ test('bigQueryWarehouse creates the table, finds stale days and counts from part assert.deepEqual(calls, [['createTable', 'event', { fields: EXPORT_SCHEMA }]]); }); - await t.test('staleDays queries the days whose rows carry other rules', async () => { + await t.test('one query over the table counts each day and finds the stale ones', async () => { const { warehouse, calls } = fakeTable({ exists: true, fields: ['id', 'export_rules'] }); - assert.deepEqual(await warehouse.staleDays('r1'), new Set(['2026-10-21'])); - assert.deepEqual(calls, [['query', { rules: 'r1' }]]); + assert.deepEqual(await warehouse.days('r1'), { + counts, + stale: new Set(['2026-10-21']), + }); + assert.deepEqual(calls, [['query', 'p.d.event', { rules: 'r1' }]]); }); await t.test('a table without export_rules is stale on every day', async () => { - assert.deepEqual( - await fakeTable({ exists: true, fields: ['id'] }).warehouse.staleDays('r1'), - new Set(['2026-10-21', '2026-10-22']), - ); + const { warehouse, calls } = fakeTable({ exists: true, fields: ['id'] }); + + assert.deepEqual(await warehouse.days('r1'), { + counts, + stale: new Set(['2026-10-21', '2026-10-22']), + }); + assert.deepEqual(calls, [['query', 'p.d.event', { rules: 'r1' }]]); }); await t.test('a query over the bytes-billed cap fails naming the cap', async () => { @@ -875,45 +1074,156 @@ test('bigQueryWarehouse creates the table, finds stale days and counts from part bigquery: bigquery as unknown as BigQuery, }); - await assert.rejects(warehouse.staleDays('r1'), /maximumBytesBilled cap/); - }); - - await t.test('dayCounts reads each partition_id as its day', async () => { - assert.deepEqual( - await fakeTable({ exists: true }).warehouse.dayCounts(), - new Map([ - ['2026-10-21', 3], - ['2026-10-22', 1434], - ]), - ); + await assert.rejects(warehouse.days('r1'), /maximumBytesBilled cap/); }); }); -test('runExportJob runs the event export after a failed account export, then fails', async () => { - const ran: string[] = []; - const logged: unknown[] = []; +test('runExportJob runs the event export after a failed account export, then fails with every failure', async (t) => { const accountError = new Error('bigquery down'); - - await assert.rejects( - runExportJob({ - exportAccounts: async () => { - ran.push('accounts'); - throw accountError; + const eventError = new Error('event export 2026-10-21: load job failed'); + const cases: [string, () => Promise, RegExp, number][] = [ + [ + 'the account export alone', + async () => undefined, + /^account export failed: bigquery down$/, + 1, + ], + [ + 'both', + async () => { + throw eventError; }, - exportEvents: async () => { - ran.push('events'); + /^account export failed: bigquery down; event export 2026-10-21: load job failed$/, + 2, + ], + ]; + + for (const [label, exportEvents, message, causes] of cases) { + await t.test(label, async () => { + const ran: string[] = []; + + await assert.rejects( + runExportJob({ + exportAccounts: async () => { + ran.push('accounts'); + throw accountError; + }, + exportEvents: async () => { + ran.push('events'); + await exportEvents(); + }, + }), + (error: unknown) => { + assert.ok(error instanceof Error); + assert.match(error.message, message); + assert.equal(error instanceof AggregateError ? error.errors.length : 1, causes); + return true; + }, + ); + assert.deepEqual(ran, ['accounts', 'events']); + }); + } +}); + +test( + 'runAndExit stops a run hung past its deadline, names the phase and exits 1', + { timeout: 5_000 }, + async (t) => { + const hang = () => new Promise(() => undefined); + const source = fakeSource({ + '2026-10-21': [[sourceEvent({ id: 1, createdAt: '2026-10-21T08:00:00.000000Z' })]], + '2026-10-22': [[sourceEvent({ id: 2 })]], + '2026-10-23': [[sourceEvent({ id: 3 })]], + }); + const cases: [string, Parameters[0], RegExp][] = [ + [ + 'a hung account export', + (progress) => + runExportJob({ + exportAccounts: hang, + exportEvents: async () => undefined, + progress, + }), + /deadline, 0.02 s from its start, during the account export$/, + ], + [ + "a hung day's read after a failed day", + (progress) => + runExport({ + source: { + ...source, + async *readDay(day) { + if (day === '2026-10-22') { + yield await hang(); + } + + yield* source.readDay(day); + }, + }, + warehouse: fakeWarehouse(new Map(), { failLoad: ['2026-10-21'] }).warehouse, + hmacKey: KEY, + today: '2026-10-23', + scratchFile: scratch(), + log: () => undefined, + progress, + }), + /during the read of 2026-10-22; days not started: 2026-10-23; 2026-10-21: load job failed$/, + ], + ]; + + for (const [label, main, message] of cases) { + // A run with no watchdog never ends, so each case is bounded. + await t.test(label, { timeout: 2_000 }, async () => { + const exits: number[] = []; + const logged: unknown[] = []; + + await runAndExit(main, { + deadlineMs: 20, + exit: (code) => exits.push(code), + logError: (error) => logged.push(error), + }); + + assert.deepEqual(exits, [1]); + assert.equal(logged.length, 1); + assert.match(String((logged[0] as Error).message), message); + }); + } + }, +); + +test('runAndExit exits with 0 after main resolves, and with 1 after one line naming the phase when it rejects', async (t) => { + const failure = new Error('table check failed'); + const cases: [string, Parameters[0], number, string[]][] = [ + ['a run that succeeds', async () => undefined, 0, []], + [ + 'a run that fails', + async (progress) => { + progress.phase = 'the event table check and day counts'; + throw failure; }, - logError: (error) => logged.push(error), - }), - (error: unknown) => { - assert.ok(error instanceof Error); - assert.equal(error.cause, accountError); - return true; - }, - ); + 1, + ['export Job failed during the event table check and day counts: table check failed'], + ], + ]; + + for (const [label, main, code, messages] of cases) { + await t.test(label, async () => { + const exits: number[] = []; + const logged: Error[] = []; - assert.deepEqual(ran, ['accounts', 'events']); - assert.deepEqual(logged, [accountError]); + await runAndExit(main, { + exit: (exitCode) => exits.push(exitCode), + logError: (error) => logged.push(error as Error), + }); + + assert.deepEqual(exits, [code]); + assert.deepEqual( + logged.map(({ message }) => message), + messages, + ); + assert.ok(logged.every(({ cause }) => cause === failure)); + }); + } }); /** @@ -922,7 +1232,7 @@ test('runExportJob runs the event export after a failed account export, then fai * against Postgres's own per-day count. The page size is 2, so every day * spans pages, rows sit either side of a midnight, three rows share one * time across a page boundary, and each row's props hold a key no event - * keeps. + * keeps. A short page ends a day's read without a further query. */ test('the export reader reads every row of each UTC day and no other table', async (t) => { const database = await createTestDatabase(); @@ -974,7 +1284,26 @@ test('the export reader reads every row of each UTC day and no other table', asy max: 1, }); - const source = pgEventSource(readerPool, 2); + const pool = readerPool; + // The page queries readDay sends, by text. + const pageQueries: string[] = []; + const source = pgEventSource( + { + options: pool.options, + connect: async () => { + const client = await pool.connect(); + + return { + query: (text: string, values?: unknown[]) => { + pageQueries.push(text); + return client.query(text, values); + }, + release: (error?: Error) => client.release(error), + }; + }, + } as unknown as Pool, + { pageSize: 2 }, + ); const { rows: expected } = await admin.query<{ day: string; count: number }>( `select to_char(date_trunc('day', created_at at time zone 'UTC'), 'YYYY-MM-DD') as day, count(*)::int as count @@ -997,6 +1326,10 @@ test('the export reader reads every row of each UTC day and no other table', asy } assert.deepEqual(read, expected); + // Days of 1, 3, 1 and 3 rows take 1, 2, 1 and 2 page queries, each the + // SQL the export_rules digest covers. + assert.equal(pageQueries.length, 6); + assert.ok(pageQueries.every((text) => text === ROW_SHAPE.READ_DAY_SQL)); // The three rows that share 05:00 on 22 October span two pages. assert.deepEqual(ids, [1, 2, 3, 4, 5, 6, 7, 8]); assert.deepEqual( diff --git a/db/test/runtime-pool.test.ts b/db/test/runtime-pool.test.ts index 7bd2597..5a422a8 100644 --- a/db/test/runtime-pool.test.ts +++ b/db/test/runtime-pool.test.ts @@ -76,7 +76,10 @@ class FakeCloudSqlSocket extends EventEmitter { * them. */ test('getCloudSqlRuntimeStream throws a clear error before anything is registered', () => { - assert.throws(() => getCloudSqlRuntimeStream(), /register\(\) must run/); + assert.throws( + () => getCloudSqlRuntimeStream(), + /instrumentation\.ts's register\(\).+export-events\.ts's main\(\) must set it/, + ); }); test('buildRuntimePool wires the Cloud SQL path to the registered stream factory', async (t) => { diff --git a/deploy/README.md b/deploy/README.md index 36ebeb3..07040e0 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -83,15 +83,26 @@ The Job reads the database as its own IAM user, a member of the NOLOGIN role `we | `EXPORT_BQ_DATASET` | The BigQuery dataset | | `EXPORT_ID_HMAC_KEY` | Key for the visitor and user id digests | -Each run replaces every UTC day from the latest day in the table minus one through today, so a missed run is made up by the next one. When the table is empty or missing, the Job creates it and loads every day from the first event. The run also replaces every earlier day whose count of `event` rows differs from its partition's row count. It takes both counts per day up front, one `GROUP BY` over the whole `event` table and the partitions' row counts from the dataset's `INFORMATION_SCHEMA.PARTITIONS`, which reads table metadata and scans no rows. Each day is read in pages, in `created_at` and id order with each page starting after the last row of the one before, and only with the prop keys `PROP_RULES` keeps. The pages go to a scratch file, then load into the day's partition (`event$YYYYMMDD`, `WRITE_TRUNCATE`, `ALLOW_FIELD_ADDITION`). A load adds a new NULLABLE column to the table, and fails on a new REQUIRED one, so `ExportField` in `db/event-export.ts` lets a column be REQUIRED only if the table was created with it. +Each run replaces every UTC day from the latest day in the table minus one through today, so a missed run is made up by the next one. When the table is empty or missing, the Job creates it and loads every day from the first event. The run also replaces every earlier day whose count of `event` rows differs from its partition's row count. It takes both counts per day up front: one `GROUP BY` over the whole `event` table in Postgres, and one query over the BigQuery table that counts each day's rows and finds its stale days (below). The window days load first, oldest first, then the earlier days, oldest first, so yesterday and today load before any backlog. Each day is read in pages, in `created_at` and id order with each page starting after the last row of the one before, and only with the prop keys `PROP_RULES` keeps. A page shorter than the page size ends the day's read. The pages go to a scratch file, then load into the day's partition (`event$YYYYMMDD`, `WRITE_TRUNCATE`, `ALLOW_FIELD_ADDITION`). A load adds a new NULLABLE column to the table, and fails on a new REQUIRED one, so `ExportField` in `db/event-export.ts` lets a column be REQUIRED only if the table was created with it. The scratch file sits in `os.tmpdir()`, which on Cloud Run is in memory, so a run holds one page of rows plus one day's file, about 360 bytes an event. The busiest day so far holds 1,434 events, about 0.5 MB, against the Job's 512 MiB. The file is written in full before its load starts, so a day loads only once it has been read to its end. A replace never shrinks a partition. Nothing deletes `event` rows, and they carry no foreign key to `user`, so deleting an account leaves them in place. A day that reads fewer rows than its partition holds is therefore a fault, such as a wrong database, an emptied table or a lagging replica. That day keeps its partition, the other days still load, and the run then fails naming each short day with both counts, as it does for a day that reads fewer rows than its source count from the start of the run, whatever its partition holds. Its counts still differ, so the day stays in every later run, however many later days load, and fails each one until it matches. The Job's failure alert therefore fires on every run until the export is complete. A day that reads no rows and holds none loads nothing. -A day whose read or load fails keeps its partition as it was, and the later days still load. The run then fails naming every such day with its error, and any short day. The Job logs each day as it finishes. A database query that runs past 5 minutes fails its day (`statement_timeout` on the Job's pool); a query that fails outside a day's read, such as the per-day count, fails the run. A BigQuery query that bills more than 1 GB or runs past 2 minutes fails the run (`maximumBytesBilled` and `jobTimeoutMs`). The stale-day query scans about 74 bytes an event row, so it reaches the 1 GB cap at about 13.5 million rows, and the run then fails with an error naming the cap. Days that each wait out a 5 minute timeout could outlast the Job's 1800 s task timeout, so a run starts no new day once 20 minutes have passed since it began (`EXPORT_BUDGET_MS` in `db/event-export.ts`). It finishes the day in hand, then fails naming every day it did not reach, with any failed or short day, and the next run picks those days up. +A day whose read or load fails keeps its partition as it was, and the later days still load. The run then fails naming every such day with its error, and any short day. The Job logs each day as it finishes, and a failure outside a day's read or load, such as the per-day count, fails the run. Either way the Job logs one error naming the phase it failed in, and exits 1 (`runAndExit` in `db/event-export.ts`). A BigQuery query that bills more than 1 GB fails the run (`maximumBytesBilled`). The BigQuery query that counts each day and finds its stale days scans about 74 bytes an event row, so it reaches the 1 GB cap at about 13.5 million rows, and the run then fails with an error naming the cap. -Every row carries in `export_rules` a digest of what shapes it (`exportRules` in `db/event-export.ts`): `PROP_RULES`, `PAGE_PATH`, `SHARE_PATH`, the token rule, the schema, `ROW_SHAPE_VERSION` and an HMAC of a fixed string under `EXPORT_ID_HMAC_KEY`. `ROW_SHAPE_VERSION` is bumped by hand with any change to the functions that build a row. Each run finds the days holding a row with another digest, or none, in one query over the `created_at` and `export_rules` columns, and replaces those days too. A change to the rules or the key thus reaches every partition. A day that fails keeps its old rows, so the next run retries it with the window days and the count-mismatch days, and reloads no day that is already current. +Every call the Job makes has its own time bound, and a call past it fails its day or its phase: + +- a database connection, 30 seconds (`connectionTimeoutMillis` on the Job's pool); +- the per-day count, 5 minutes on the server (`statement_timeout`), and 5 seconds more on the Job's side for a connection that stalls where the server's timeout cannot reach it; +- the account read, 5 minutes a query on the server, and 10 minutes 5 seconds on the Job's side for its two queries together; +- a day's page queries together, 5 minutes from the day's start plus 5 seconds, on the Job's side; +- a BigQuery query, 2 minutes on the server (`jobTimeoutMs`) and 5 seconds more on the Job's side, as for each table read or create; +- a load, the account export's included, 2 minutes, after which the Job cancels it and waits at most 10 seconds for the cancel (`loadWithin` in `db/bounded.ts`). + +A database call past its bound on the Job's side leaves its connection mid-query, so the Job destroys that connection and the next call opens a new one (`callWithin` in `db/bounded.ts`). The Job stops a run still going 28 minutes after its start (`JOB_DEADLINE_MS`), two minutes before the task timeout, whatever call is in flight. It logs one error naming the phase (the account export, the day counts, or a named day's read or load), the days not started, and the failed and short days so far, after the lines each finished day already logged, and exits 1. The window days run first, so a run stopped this way has loaded yesterday and today before the backlog, and the next hourly run picks up the days it did not reach. The Job also exits as soon as the run ends, which closes the pool, the connector and any upload socket a load past its timeout left open. + +Every row carries in `export_rules` a digest of what shapes it (`exportRules` in `db/event-export.ts`): `PROP_RULES`, `PAGE_PATH`, `SHARE_PATH`, the token rule, the schema, the SQL that reads a day's page, `ROW_SHAPE_VERSION` and an HMAC of a fixed string under `EXPORT_ID_HMAC_KEY`. `ROW_SHAPE_VERSION` is bumped by hand with any change to the functions that build a row, and a test pins each version's rows for a fixed sample of events, so a change to the rows without a bump fails CI. Each run finds the days holding a row with another digest, or none, in the same query over the `created_at` and `export_rules` columns that counts each day's rows, and replaces those days too. A change to the rules or the key thus reaches every partition. A day that fails keeps its old rows, so the next run retries it with the window days and the count-mismatch days, and reloads no day that is already current. The table is partitioned by day on `created_at`: diff --git a/docs/analytics.md b/docs/analytics.md index 3a00f84..c3df018 100644 --- a/docs/analytics.md +++ b/docs/analytics.md @@ -94,13 +94,13 @@ To check it on production, open the network panel and filter on `google-analytic ## Request log -The site writes one JSON line to stdout for each logged request, which Cloud Logging stores as a structured entry. `src/instrumentation.ts` registers `registerRequestLogging()` in `src/lib/proxy/requestLog.ts` at server startup. It subscribes to Node's `http.server.request.start` diagnostic channel and attaches one `close` listener to each response. Node emits `close` after `finish`, so every request writes exactly one line, when its response closes, with `finished` true for a response that ended normally. A response that closes before it ends, such as a page render the visitor abandons, an aborted RSC fetch or an `/mcp` event stream the client closes, writes its line then, with the status code it had sent. A request accepted before the subscription starts has no listener, so its line is written when its response finishes. Visitor counts come from this line, because Cloud Run's own request log records Cloudflare's edge address as the client. +The site writes one JSON line to stdout for each logged request, which Cloud Logging stores as a structured entry. `src/instrumentation.ts` registers `registerRequestLogging()` in `src/lib/proxy/requestLog.ts` at server startup. It subscribes to Node's `http.server.request.start` diagnostic channel, reads each request's target there, and attaches one `close` listener to each response it logs. A static asset gets no listener. Node emits `close` after `finish`, so every request writes exactly one line, when its response closes, with `finished` true for a response that ended normally. A response that closes before it ends, such as a page render the visitor abandons, an aborted RSC fetch or an `/mcp` event stream the client closes, writes its line then, with the status code it had sent. A request accepted before the subscription starts has no listener, so its line is written when its response finishes. Visitor counts come from this line, because Cloud Run's own request log records Cloudflare's edge address as the client. ```json {"severity":"INFO","message":"request","caller":"ip:3f9c0a…","method":"GET","path":"/embed/abc123","host":"developmentfinance.org","kind":"page","status":200,"finished":true,"bot":false,"prerender":false} ``` -`caller` is the same `ip:` the `/mcp` proxy sends upstream as `x-dfo-caller` (`src/lib/proxy/caller.ts`): an HMAC of the client address under `CALLER_HMAC_KEY`, with the address taken from `X-Forwarded-For` at the `TRUSTED_PROXY_HOPS` position. It is null when the key is unset or the request carries no usable address. `path` is the request target as it arrived, captured when the request starts so a later rewrite does not change it. It has no query string or fragment, and dot segments are resolved, so `/a/../mcp` is logged as `/mcp`. A target such as `//mcp` is a path and not a host, and is logged as `//mcp`. An absolute-form target such as `http://host/path` is logged as its pathname, and an absolute-form target whose URL does not parse is logged as the raw target cut at its first `?` or `#`. The asterisk-form target of `OPTIONS *` is logged as `*`, with kind `page`. `kind` is `api` for `/api`, `/mcp`, `/openapi.json` and `/codes` and every path below each of them (`/apiary` is not one). That covers the site's own endpoints under `/api`, such as auth, events and share, as well as the platform proxy's routes. `kind` is `prefetch` for a Next.js prefetch (`Next-Router-Prefetch`) or a browser prefetch (`Sec-Purpose: prefetch`), `rsc` for a client-side navigation (`RSC: 1`) and `page` for everything else. A `Sec-Purpose` header does not change the kind of an RSC fetch or a Next.js prefetch. Embeds under `/embed/` are logged as pages. `/_next/` files, images and fonts are not logged, and are dropped before the line's headers are read or its caller computed. `status` is the response's HTTP status code. `finished` is true when the response ended normally and false when it closed first, so a status of 200 with `finished` false is a visitor who left a stream. A response that closes before the app sent a status has none of its own, so the line carries 499 (client closed request) with `finished` false. The 499 means the app sent no status before the connection closed. It also covers a response Node answered itself, such as a 408 on a request timeout or a 400 on a client error, which never passes through the app. A response that called `writeHead` but wrote nothing to the socket before the close is logged with that status and `finished` false. Infra's parse keeps the line and counts it as a 4xx, outside the readers' 2xx count. Infra's `website_requests_daily` counts a line whose response did not finish like a finished one. `prerender` is true for a document request (no `RSC` and no `Next-Router-Prefetch` header) with `Sec-Purpose: prefetch;prerender`: the browser renders the page ahead of time and shows it on activation with no second request, so that one line is the page view. It has kind `page` and not `prefetch`, which keeps infra's `website_requests_daily` counting it as a parsed page request. The line is a page view whether or not the visitor opens the page, because the server cannot see activation. `bot` is true when the `User-Agent` header is missing or empty, or when the `isbot` package flags it: crawlers such as Googlebot, bingbot and ClaudeBot, link previews such as Slack's and Discord's, headless browsers and tools such as curl. A client that sends a browser's user agent counts as a reader. The line holds no raw IP address, query string, cookie, `Authorization` header or user agent. +`caller` is the same `ip:` the `/mcp` proxy sends upstream as `x-dfo-caller` (`src/lib/proxy/caller.ts`): an HMAC of the client address under `CALLER_HMAC_KEY`, with the address taken from `X-Forwarded-For` at the `TRUSTED_PROXY_HOPS` position. It is null when the key is unset or the request carries no usable address. `path` is the request target as it arrived, captured when the request starts so a later rewrite does not change it. It has no query string or fragment, and dot segments are resolved, so `/a/../mcp` is logged as `/mcp`. A target such as `//mcp` is read as a path and logged as `//mcp`. An http or https absolute-form target such as `http://host/path` is logged as its pathname. Any other target, such as one with another scheme (`ftp://`, `file:`, `mailto:`), an authority-form `host:port`, or an http or https URL that does not parse, is logged as the fixed path `/:unparsable`, since it can hold a user name, a password or an IP address. The asterisk-form target of `OPTIONS *` is logged as `*`, with kind `page`. `kind` is `api` for `/api`, `/mcp`, `/openapi.json` and `/codes` and every path below each of them (`/apiary` is not one). That covers the site's own endpoints under `/api`, such as auth, events and share, as well as the platform proxy's routes. `kind` is `prefetch` for a Next.js prefetch (`Next-Router-Prefetch`) or a browser prefetch (`Sec-Purpose: prefetch`), `rsc` for a client-side navigation (`RSC: 1`) and `page` for everything else. A `Sec-Purpose` header does not change the kind of an RSC fetch or a Next.js prefetch. Embeds under `/embed/` are logged as pages. `/_next/` files, images and fonts are not logged, and are dropped before the line's headers are read or its caller computed. `status` is the response's HTTP status code. `finished` is true when the response ended normally and false when it closed first, so a status of 200 with `finished` false is a visitor who left a stream. A response that closes before the app sent a status has none of its own, so the line carries 499 (client closed request) with `finished` false. The 499 means the app sent no status before the connection closed. It also covers a response Node answered itself, such as a 408 on a request timeout or a 400 on a client error, which never passes through the app. A response that called `writeHead` but wrote nothing to the socket before the close is logged with that status and `finished` false. Infra's parse keeps the line and counts it as a 4xx, outside the readers' 2xx count. Infra's `website_requests_daily` counts a line whose response did not finish like a finished one. `prerender` is true for a document request (no `RSC` and no `Next-Router-Prefetch` header) with `Sec-Purpose: prefetch;prerender`: the browser renders the page ahead of time and shows it on activation with no second request, so that one line is the page view. It has kind `page`, which keeps infra's `website_requests_daily` counting it as a parsed page request. The line is a page view whether or not the visitor opens the page, because the server cannot see activation. `bot` is true when the `User-Agent` header is missing or empty, or when the `isbot` package flags it: crawlers such as Googlebot, bingbot and ClaudeBot, link previews such as Slack's and Discord's, headless browsers and tools such as curl. A client that sends a browser's user agent counts as a reader. The line holds no raw IP address, query string, cookie, `Authorization` header or user agent. The line reads the request's raw headers, which keep the `RSC` and `Next-Router-Prefetch` headers that Next.js removes from `request.headers`. Logging leaves request bodies unread, so API and MCP payloads stream to their handlers in full. A Next.js proxy would buffer those bodies and truncate them at its default 10 MB limit. diff --git a/src/lib/proxy/requestLog.ts b/src/lib/proxy/requestLog.ts index b83b704..78d11d0 100644 --- a/src/lib/proxy/requestLog.ts +++ b/src/lib/proxy/requestLog.ts @@ -17,7 +17,9 @@ import { requestCaller } from './caller.ts'; * connection closed. * * `path` is the request target as it arrived at `http.server.request.start`, - * without its query string, with dot segments resolved. `caller` is the + * without its query string, with dot segments resolved, or `/:unparsable` for + * a target that is not origin-form, `*`, or an http or https URL that + * parses. `caller` is the * same `ip:` the /mcp proxy sends as x-dfo-caller (see caller.ts). * `bot` is true when the User-Agent header is missing, empty or one isbot * flags. `prerender` is true for a document request with `Sec-Purpose: @@ -119,31 +121,53 @@ export const requestLogEntry = ( }; }; -// The request target as it arrived, kept because Next.js can rewrite -// `request.url` before the response ends. -const arrivedTargets = new WeakMap(); +// The path of each logged request's target as it arrived, kept because +// Next.js can rewrite `request.url` before the response ends. +const arrivedPaths = new WeakMap(); // Client closed request, the status for a response whose app sent no status // before the connection closed. const CLIENT_CLOSED_REQUEST = 499; /** - * The path of a request target. An origin-form target resolves against a - * placeholder host, which reads `//mcp` as a path and not a host and resolves - * dot segments; an absolute-form target keeps its own pathname. The - * asterisk-form target of `OPTIONS *` is its own path. A target whose URL does - * not parse is cut at its first `?` or `#`, so the request still writes a line. + * The path a target is logged as when it is neither origin-form, an http or + * https absolute-form target that parses, nor `*`, since a target of any + * other form or scheme can hold userinfo or an IP address. + */ +const UNPARSABLE_TARGET_PATH = '/:unparsable'; + +const HTTP_ABSOLUTE_FORM = /^https?:\/\//i; + +/** + * The path of a request target. An origin-form target (one starting with + * `/`) resolves against a placeholder host, which reads `//mcp` as the path + * `//mcp` and resolves dot segments, and is cut at its first `?` or `#` + * if its URL does not parse. An http or https absolute-form target keeps + * its own pathname. The asterisk-form target of `OPTIONS *` is its own + * path. Every other target is UNPARSABLE_TARGET_PATH, so the request still + * writes a line. */ const targetPath = (target: string): string => { if (target === '*') { return target; } + if (target.startsWith('/')) { + try { + return new URL(`http://localhost${target}`).pathname; + } catch { + return target.split(/[?#]/, 1)[0] ?? target; + } + } + + if (!HTTP_ABSOLUTE_FORM.test(target)) { + return UNPARSABLE_TARGET_PATH; + } + try { - return new URL(/^https?:\/\//i.test(target) ? target : `http://localhost${target}`) - .pathname; + return new URL(target).pathname; } catch { - return target.split(/[?#]/, 1)[0] ?? target; + return UNPARSABLE_TARGET_PATH; } }; @@ -177,7 +201,7 @@ export const logRequest = ( response: { status: number; finished: boolean }, ): void => { try { - const path = targetPath(arrivedTargets.get(request) ?? request.url ?? '/'); + const path = arrivedPaths.get(request) ?? targetPath(request.url ?? '/'); if (isStaticAsset(path)) { return; @@ -220,7 +244,8 @@ const httpPair = ( ? { request: message.request, response: message.response } : undefined; -// The close listener logs every request that arrived after registration. +// The close listener logs every request that arrived after registration, +// apart from a static asset, which gets no listener and no entry. // ServerResponse emits `close` after `finish`, so one listener covers a // response that ends and one that does not. const onRequestStart = (message: unknown): void => { @@ -231,8 +256,13 @@ const onRequestStart = (message: unknown): void => { } const { request, response } = pair; + const path = targetPath(request.url ?? '/'); + + if (isStaticAsset(path)) { + return; + } - arrivedTargets.set(request, request.url ?? '/'); + arrivedPaths.set(request, path); response.once('close', () => { logRequest(request, { status: response.headersSent ? response.statusCode : CLIENT_CLOSED_REQUEST, @@ -242,11 +272,14 @@ const onRequestStart = (message: unknown): void => { }; // A request accepted before registration has no close listener, so its finish -// writes the line. +// writes the line. A static asset that arrived after it has no entry either, +// so logRequest reads its path from `request.url` at finish and drops it. A +// handler that rewrote it to a logged path would log it under that path; no +// rewrite in next.config.ts does that. const onResponseFinish = (message: unknown): void => { const pair = httpPair(message); - if (pair && !arrivedTargets.has(pair.request)) { + if (pair && !arrivedPaths.has(pair.request)) { logRequest(pair.request, { status: pair.response.statusCode, finished: true }); } }; @@ -254,8 +287,9 @@ const onResponseFinish = (message: unknown): void => { /** * Observe HTTP headers without reading or buffering a request body. A request * accepted before this runs is still logged when its response finishes. + * Returns a function that removes the subscribers. */ -export const registerRequestLogging = (): void => { +export const registerRequestLogging = (): (() => void) => { const previous = globalForRequestLog.dfoRequestLogSubscribers; if (previous) { @@ -269,4 +303,13 @@ export const registerRequestLogging = (): void => { start: onRequestStart, finish: onResponseFinish, }; + + return () => { + unsubscribe(REQUEST_START, onRequestStart); + unsubscribe(RESPONSE_FINISH, onResponseFinish); + + if (globalForRequestLog.dfoRequestLogSubscribers?.start === onRequestStart) { + delete globalForRequestLog.dfoRequestLogSubscribers; + } + }; }; diff --git a/src/lib/proxy/test/requestLog.test.ts b/src/lib/proxy/test/requestLog.test.ts index a59a3c2..8ebe2b6 100644 --- a/src/lib/proxy/test/requestLog.test.ts +++ b/src/lib/proxy/test/requestLog.test.ts @@ -245,7 +245,7 @@ test('logRequest: no client address logs a null caller', () => { assert.equal((line as { caller: unknown }).caller, null); }); -test('logRequest: the path is the target without its query, resolved, with //mcp kept as a path, * as itself and an unparsable URL cut at its query', () => { +test('logRequest: the path is the target without its query, resolved, with //mcp kept as a path, * as itself and any other target as a placeholder', () => { // [target, logged path or null for a dropped static asset, kind] const cases: [string, string | null, string][] = [ ['//mcp', '//mcp', 'page'], @@ -257,7 +257,13 @@ test('logRequest: the path is the target without its query, resolved, with //mcp ['http://host/_next/static/x.js', null, 'page'], ['https://host/api/v1/search?q=secret', '/api/v1/search', 'api'], ['*', '*', 'page'], - ['http://[bad/path?q=secret#frag', 'http://[bad/path', 'page'], + ['http://[bad/path?q=secret#frag', '/:unparsable', 'page'], + ['http://user:pw@203.0.113.7:99999/x?q=secret', '/:unparsable', 'page'], + ['ftp://user:pw@203.0.113.7/x?q=secret', '/:unparsable', 'page'], + ['ws://user:pw@203.0.113.7/mcp', '/:unparsable', 'page'], + ['file:///etc/passwd', '/:unparsable', 'page'], + ['mailto:a@b', '/:unparsable', 'page'], + ['example.com:443', '/:unparsable', 'page'], ]; for (const [target, path, kind] of cases) { @@ -323,7 +329,7 @@ test('server logging catches a request accepted during startup with its original // Next.js may normalise these headers before a startup request finishes. delete request.headers.rsc; delete request.headers['x-forwarded-for']; - registerRequestLogging(); + t.after(registerRequestLogging()); response.statusCode = 404; response.end('ok'); }); @@ -364,8 +370,8 @@ test( // A dev reload evaluates the module again, so its subscriber is a new function. const reloadedModule = '../requestLog.ts?reload'; const reloaded = (await import(reloadedModule)) as typeof import('../requestLog.ts'); - registerRequestLogging(); - reloaded.registerRequestLogging(); + t.after(registerRequestLogging()); + t.after(reloaded.registerRequestLogging()); const firstChunk = Promise.withResolvers(); const server = createServer(async (request, response) => { @@ -462,7 +468,7 @@ const listen = async ( test('server logging writes one line with finished true for a response that finishes', async (t) => { const lines: unknown[] = []; t.mock.method(console, 'log', (line: string) => lines.push(JSON.parse(line))); - registerRequestLogging(); + t.after(registerRequestLogging()); const closed = Promise.withResolvers(); const port = await listen(t, (_request, response) => { response.once('close', () => closed.resolve()); @@ -489,7 +495,7 @@ test( lines.push(JSON.parse(line)); logged.resolve(); }); - registerRequestLogging(); + t.after(registerRequestLogging()); const closed = Promise.withResolvers(); const port = await listen(t, (_request, response) => { response.once('close', () => closed.resolve()); @@ -534,7 +540,7 @@ test( lines.push(JSON.parse(line)); logged.resolve(); }); - registerRequestLogging(); + t.after(registerRequestLogging()); const received = Promise.withResolvers(); const port = await listen(t, () => { // The handler sends nothing, not even the headers. @@ -563,13 +569,17 @@ test( test('server logging logs the target as it arrived when a handler rewrites request.url', async (t) => { const lines: unknown[] = []; t.mock.method(console, 'log', (line: string) => lines.push(JSON.parse(line))); - registerRequestLogging(); + t.after(registerRequestLogging()); + const closed = Promise.withResolvers(); const port = await listen(t, (request, response) => { + response.once('close', () => closed.resolve()); request.url = '/_next/static/rewritten.js'; response.end('ok'); }); await (await fetch(`http://127.0.0.1:${port}/mcp?token=private-query`)).text(); + // The line is written on close, which can follow the body's arrival. + await closed.promise; assert.deepEqual( lines.map((line) => ({ @@ -579,3 +589,33 @@ test('server logging logs the target as it arrived when a handler rewrites reque [{ path: '/mcp', kind: 'api' }], ); }); + +test('server logging attaches no close listener to a static asset response', async (t) => { + const lines: unknown[] = []; + t.mock.method(console, 'log', (line: string) => lines.push(JSON.parse(line))); + t.after(registerRequestLogging()); + const listeners = new Map(); + const closed: Promise[] = []; + const port = await listen(t, (request, response) => { + listeners.set(request.url ?? '', response.listenerCount('close')); + const done = Promise.withResolvers(); + closed.push(done.promise); + response.once('close', () => done.resolve()); + response.end('ok'); + }); + + for (const path of ['/_next/static/chunks/main.js', '/studio']) { + await (await fetch(`http://127.0.0.1:${port}${path}`)).text(); + } + await Promise.all(closed); + + // The logged page carries the one listener the asset lacks. + assert.equal( + listeners.get('/studio'), + (listeners.get('/_next/static/chunks/main.js') ?? 0) + 1, + ); + assert.deepEqual( + lines.map((line) => (line as { path: string }).path), + ['/studio'], + ); +});