Reduce commit overhead in bulk import: async commit + bigger batches

rockyou import was ~195s of actual CPU work spread across 42 minutes wall
clock: Postgres was fsync-bound, not CPU/parallelism-bound (each batch was
its own implicit-commit transaction). Wrapping each batch in an explicit
transaction with synchronous_commit=off, and raising the default batch size
20000 (also used for wordlist worker chunking, previously hardcoded to 2000),
brought the same import down to ~25 minutes wall clock / ~186s user time.

Next bottleneck to address: B-tree index maintenance cost as it outgrows
shared_buffers (confirmed via pg_stat_user_tables: 14.3M live tuples, zero
dead tuples on a fresh DB, so it's not autovacuum/bloat). Random-order
inserts into the (prefix, suffix) index cause disk-bound page faults once the
index no longer fits in cache, which matched the observed end-of-run
slowdown and the "N-at-a-time" stalls lining up with --jobs concurrency.

packages/import/src/index.ts

@@ -8,7 +8,6 @@ import { createLimiter } from "./concurrencyLimit.ts";
 import { WorkerPool } from "./workerPool.ts";
 
 const HASH_LINE = /^([0-9A-Fa-f]{40})(?::(\d+))?$/;
-const CHUNK_SIZE = 2000;
 
 function usage(): never {
   console.error(
@@ -17,7 +16,8 @@ function usage(): never {
       `  hashes          file contains one SHA-1 hash per line, optionally "HASH:COUNT"\n` +
       `  file may be gzip-compressed (.gz)\n\n` +
       `Options:\n` +
-      `  --batch-size    rows per upsert statement (default 5000)\n` +
+      `  --batch-size    rows per upsert transaction (default 20000); bulk imports are\n` +
+      `                  bound by commit rate, so bigger batches mean fewer commits\n` +
       `  --jobs, -j      parallelism: worker threads for wordlist hashing, or\n` +
       `                  concurrent DB upserts for hashes (default: all CPU cores)\n`,
   );
@@ -75,7 +75,7 @@ async function importWordlist(file: string, batchSize: number, jobs: number) {
 
   for await (const line of readLines(file)) {
     chunk.push(line);
-    if (chunk.length >= CHUNK_SIZE) {
+    if (chunk.length >= batchSize) {
       inFlight.push(submit(chunk));
       chunk = [];
     }
@@ -124,7 +124,7 @@ async function main() {
     allowPositionals: true,
     options: {
       format: { type: "string" },
-      "batch-size": { type: "string", default: "5000" },
+      "batch-size": { type: "string", default: "20000" },
       jobs: { type: "string", short: "j" },
     },
   });

packages/import/src/upsertBatch.ts

@@ -25,11 +25,30 @@ export async function upsertBatch(pool: Pool, entries: Entry[]): Promise<void> {
   // statements; a consistent lock order avoids deadlocks between them.
   const rows = [...merged.values()].sort((a, b) => (a.prefix + a.suffix < b.prefix + b.suffix ? -1 : 1));
 
-  await pool.query(
+  // Bulk imports are commit-rate bound, not CPU bound: every pool.query() is
-    `INSERT INTO pwned_passwords (prefix, suffix, count)
+  // its own implicit transaction, and by default Postgres fsyncs WAL on each
-     SELECT * FROM UNNEST($1::char(5)[], $2::char(35)[], $3::bigint[])
+  // commit. At import volumes that serializes everything on disk latency
-     ON CONFLICT (prefix, suffix) DO UPDATE
+  // regardless of how many connections are writing concurrently. Since a
-       SET count = pwned_passwords.count + EXCLUDED.count`,
+  // failed/interrupted import is just re-run, trading a small durability
-    [rows.map((r) => r.prefix), rows.map((r) => r.suffix), rows.map((r) => r.count)],
+  // window (an OS crash could lose the last few commits) for throughput is
-  );
+  // the right tradeoff here; this setting only affects this pool, not the
+  // backend's read-only connection.
+  const client = await pool.connect();
+  try {
+    await client.query("BEGIN");
+    await client.query("SET LOCAL synchronous_commit = OFF");
+    await client.query(
+      `INSERT INTO pwned_passwords (prefix, suffix, count)
+       SELECT * FROM UNNEST($1::char(5)[], $2::char(35)[], $3::bigint[])
+       ON CONFLICT (prefix, suffix) DO UPDATE
+         SET count = pwned_passwords.count + EXCLUDED.count`,
+      [rows.map((r) => r.prefix), rows.map((r) => r.suffix), rows.map((r) => r.count)],
+    );
+    await client.query("COMMIT");
+  } catch (err) {
+    await client.query("ROLLBACK");
+    throw err;
+  } finally {
+    client.release();
+  }
 }