fix(export): bound every export call and order current days first

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<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`);
+
+    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<AccountRead> => {
+export const readAccountDaily = async (
+    pool: Pool,
+    { timeoutMs = ACCOUNT_READ_MS }: { timeoutMs?: number } = {},
+): Promise<AccountRead> => {
     const client = await pool.connect();
 
     try {
-        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 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<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 },
-            });
+            },
+        });
     };
 
 /**

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

@@ -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.',
         );
     }