release: website dev to main

db/account-export.ts

@@ -1,16 +1,18 @@
 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
  * before the event export: copies the Postgres view `account_daily`
- * (migration 0020) into the BigQuery table `account_daily`, replacing the
- * whole table in one load job, so a failed run leaves the previous table.
- * The view counts new accounts per UTC day and organisation, for every day
- * from the first account's through yesterday's, and the rows hold no email
- * address or user id.
+ * (migrations 0020 and 0021) into the BigQuery table `account_daily`,
+ * replacing the whole table in one load job, so a failed run leaves the
+ * previous table. The view counts new accounts per UTC day and
+ * organisation, for every day from the first account's through
+ * yesterday's, and the rows hold no email address or user id.
  */
 
 /** One `account_daily` row. */
@@ -45,55 +47,120 @@ export const ACCOUNT_SCHEMA: TableField[] = [
     },
 ];
 
-/** Every row of the view, oldest day first. */
-export const readAccountDaily = async (pool: Pool): Promise<AccountRow[]> => {
-    const { rows } = await pool.query<AccountRow>(
+/**
+ * What one read of the view returns. `accounts` and `webmailDomains` come
+ * from the view `account_daily_source`: whether the user table holds any
+ * account, and whether `webmail_domain` holds any domain.
+ */
+export type AccountRead = {
+    rows: AccountRow[];
+    accounts: boolean;
+    webmailDomains: boolean;
+};
+
+const readInTransaction = async (client: PoolClient): Promise<AccountRead> => {
+    await client.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');
+
+    const { rows } = await client.query<AccountRow>(
         `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`);
 
-    return rows;
+    await client.query('COMMIT');
+
+    return { rows, ...source[0]! };
 };
 
-/** Replaces the BigQuery table `account_daily` with the rows in `file`, one JSON object a line. */
+/** 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. A read still running at `timeoutMs`, as over a
+ * connection that stalls past Postgres's timeout, throws.
+ */
+export const readAccountDaily = async (
+    pool: Pool,
+    { timeoutMs = ACCOUNT_READ_MS }: { timeoutMs?: number } = {},
+): Promise<AccountRead> => {
+    const client = await pool.connect();
+
+    try {
+        const read = await callWithin(
+            readInTransaction(client),
+            timeoutMs,
+            () => new Error(`account read ran past ${timeoutMs / 1000} s`),
+        );
+
+        client.release();
+
+        return read;
+    } catch (error) {
+        // 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. A load past its timeout is cancelled and throws
+ * (loadWithin).
+ */
 export const bigQueryAccountTable =
     ({ datasetId, bigquery }: { datasetId: string; bigquery: BigQuery }) =>
     async (file: string): Promise<void> => {
-        // 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 },
-            });
+            },
+        });
     };
 
 /**
  * One account export run. Writes every row to `scratchFile`, then replaces
- * the table with it. The view holds a row for every day since the first
- * account's, so a read with no rows is a fault (a wrong database, an
- * emptied table) unless every account is from today, and rejects with the
- * table kept.
+ * the table with it. The view holds a row for every day from the first
+ * account's through yesterday's while `webmail_domain` holds any domain,
+ * so a read with no rows means one of three things. While `webmail_domain`
+ * is empty, or while the user table holds no account (a wrong database, an
+ * emptied table), the run rejects with the table kept. When every account
+ * is from today, the table is replaced with no rows, so an account deleted
+ * since the last run, and the organisation it named, leave the export.
  */
 export const runAccountExport = async ({
     read,
     replace,
     scratchFile,
     log,
 }: {
-    read: () => Promise<AccountRow[]>;
+    read: () => Promise<AccountRead>;
     replace: (file: string) => Promise<void>;
     scratchFile: string;
     log: (line: string) => void;
 }): Promise<void> => {
-    const rows = await read();
+    const { rows, accounts, webmailDomains } = await read();
+
+    if (rows.length === 0 && !webmailDomains) {
+        throw new Error(
+            'account export: account_daily read no rows, webmail_domain is empty, table kept',
+        );
+    }
 
-    if (rows.length === 0) {
-        throw new Error('account export: account_daily read no rows, table kept');
+    if (rows.length === 0 && !accounts) {
+        throw new Error('account export: account_daily read no accounts, table kept');
     }
 
     try {

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 <T>(work: Promise<T>, ms: number): Promise<T | typeof TIMED_OUT> => {
+    const timer = new AbortController();
+
+    try {
+        return await Promise.race([
+            work,
+            sleep<typeof TIMED_OUT>(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 <T>(
+    call: Promise<T>,
+    ms: number,
+    timedOut: () => Error,
+): Promise<T> => {
+    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<void> => {
+    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`);
+    }
+};

db/connection.ts

@@ -97,20 +97,20 @@ type EnvLike = Record<string, string | undefined>;
 /**
  * Non-throwing check for which pool kind the runtime should build. Shared
  * by `resolveRuntimeConnection` below (full validation, safe to throw --
- * called from instrumentation.ts's `register()` at server start) and
- * src/lib/db/index.ts (must never throw at import, including during
- * `next build` with no env at all).
+ * called at process start) and src/lib/db/index.ts (must never throw at
+ * import, including during `next build` with no env at all).
  */
 export function isCloudSqlConfigured(env: EnvLike = process.env): boolean {
     return !!env.WEBSITE_DB_INSTANCE_CONNECTION_NAME;
 }
 
 /**
  * Resolves which database the runtime pool should connect to. Cloud SQL
- * (deployed services) wins whenever its instance var is set; a local
- * Postgres via DATABASE_URL is the fallback. Throws a clear error naming
- * the missing variable(s) otherwise -- safe here because this is only
- * called from instrumentation.ts's `register()` at server start, never at
+ * (deployed services and the event export Job) wins whenever its instance
+ * var is set; a local Postgres via DATABASE_URL is the fallback. Throws a
+ * clear error naming the missing variable(s) otherwise -- safe here
+ * because its callers, instrumentation.ts's `register()` at server start
+ * and db/export-events.ts's `main()`, call it at process start, never at
  * module import time.
  */
 export function resolveRuntimeConnection(env: EnvLike = process.env): RuntimeConnection {
@@ -145,10 +145,11 @@ export function resolveRuntimeConnection(env: EnvLike = process.env): RuntimeCon
  * Opens one `Connector` for the runtime pool's lifetime and resolves the
  * stream factory pg's `Pool` uses to dial Cloud SQL over PSC with IAM auth
  * -- the same path `buildMigratorPool` uses for the migrator role, just
- * authenticating as the runtime IAM database user instead. Called once
- * from instrumentation.ts's `register()`; the `Connector` is intentionally
- * never closed here (SIGTERM exits the process, same as the migrator's
- * shutdown note above but there is no request to await first).
+ * authenticating as the connection's IAM database user instead. Called
+ * once per process, from instrumentation.ts's `register()` and from
+ * db/export-events.ts's `main()`. The caller owns the `Connector`: the
+ * server leaves it open until SIGTERM exits the process, and the export
+ * Job closes it after ending its pool.
  */
 export async function connectCloudSqlRuntime(
     connection: CloudSqlRuntimeConnection,
@@ -170,18 +171,19 @@ interface CloudSqlRuntimeGlobal {
 const globalForCloudSqlRuntime = globalThis as unknown as CloudSqlRuntimeGlobal;
 
 /**
- * Stores the stream factory `instrumentation.ts`'s `register()` resolved,
- * so src/lib/db/index.ts's pool can read it lazily at connect time without
- * importing the connector itself (index.ts must stay import-safe during
- * `next build`, and the connector is only ever constructed once, in
- * `register()`).
+ * Stores the stream factory `connectCloudSqlRuntime` resolved, so a pool
+ * from `buildRuntimePool` (src/lib/db/index.ts's, the export Job's) can
+ * read it lazily at connect time without importing the connector itself
+ * (index.ts must stay import-safe during `next build`, and the connector
+ * is constructed once per process, in `register()` or the export Job's
+ * `main()`).
  */
 export function setCloudSqlRuntimeStream(stream: DriverOptions['stream']): void {
     globalForCloudSqlRuntime.cloudSqlRuntimeStream = stream;
 }
 
 /**
- * Reads the stream factory `register()` stored. Throws only when actually
+ * Reads the stored stream factory. Throws only when actually
  * called -- i.e. when pg attempts a connection -- never at import, so a
  * misconfigured deploy still builds and starts; it just fails the first
  * request that touches the database.
@@ -192,31 +194,32 @@ 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.',
         );
     }
 
     return stream;
 }
 
 /**
- * Builds the one runtime `pg.Pool`, shared by src/lib/db/index.ts and
- * db/test/runtime-pool.test.ts so both exercise the same construction
- * path. Picks the pool kind from `isCloudSqlConfigured` alone -- a
+ * Builds a runtime `pg.Pool`, shared by src/lib/db/index.ts, the export
+ * Job (db/export-events.ts) and db/test/runtime-pool.test.ts so all
+ * exercise the same construction path. Picks the pool kind from `isCloudSqlConfigured` alone -- a
  * non-throwing check -- rather than the full, throwing
  * `resolveRuntimeConnection`, because this must never throw when called at
  * module import time (`next build` runs with no DB env at all):
  *
  * - Cloud SQL: `stream` is a function pg only calls when it actually opens
- *   a connection. It reads the factory `instrumentation.ts`'s `register()`
- *   stored on globalThis at that point, not now -- a clear error surfaces
- *   on the first connection attempt if `register()` never ran, not at
- *   import or build.
+ *   a connection. It reads the factory `setCloudSqlRuntimeStream` stored
+ *   on globalThis at that point, not now -- a clear error surfaces on the
+ *   first connection attempt if none was stored, not at import or build.
  * - URL: `connectionString` may be `undefined` here (no env at build); pg
  *   only fails to connect with it later.
  *
- * `sizing` overrides the pool size and adds timeouts, for a pool that must
- * give up fast (src/lib/db/index.ts's counter pool).
+ * `sizing` overrides the pool size and adds timeouts: src/lib/db/index.ts's
+ * counter pool gives up fast, and the export Job's pool holds one
+ * connection with a 5-minute `statement_timeout`.
  *
  * Every pool gets an `'error'` listener. pg-pool emits `'error'` when an
  * idle client fails (Cloud SQL or a proxy dropping the connection), and an