ci: fuzz the parser, the query modes and pg-native on a random seed

Author: nigrosimone

.github/workflows/ci.yml

@@ -93,3 +93,47 @@ jobs:
           cache: yarn
       - run: yarn install --frozen-lockfile
       - run: yarn test
+
+  # A few fuzz rounds on a seed nobody chose, every push and every pull request. One run proves
+  # nothing; a few hundred runs walk over ground no hand written test reaches, and the seed
+  # changes every time so the coverage accumulates across pushes. A divergence turns this job
+  # red and prints the seed that reproduces it, with the case shrunk to the lines worth pasting
+  # into a test: replay it locally with `node fuzz/<tool>.js --seed N --rounds 1`.
+  fuzz:
+    timeout-minutes: 15
+    needs: lint
+    services:
+      postgres:
+        image: ghcr.io/railwayapp-templates/postgres-ssl:18
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_HOST_AUTH_METHOD: 'md5'
+          POSTGRES_DB: ci_db_test
+          PGDATA: /var/lib/postgresql/data
+        ports:
+          - 5432:5432
+        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+    runs-on: ubuntu-latest
+    env:
+      PGUSER: postgres
+      PGPASSWORD: postgres
+      PGHOST: localhost
+      PGDATABASE: ci_db_test
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Setup node
+        uses: actions/setup-node@v4
+        with:
+          node-version: 26
+          cache: yarn
+      - run: yarn install --frozen-lockfile
+      - run: yarn build
+      - name: Fuzz the protocol parser against the bytes it was given, cut at random points
+        run: node fuzz/wire.js --rounds 500
+      - name: Fuzz pg against itself in every query mode
+        run: node fuzz/modes.js --rounds 100
+      - name: Fuzz pg against pg-native
+        run: node fuzz/native.js --rounds 100

fuzz/README.md

@@ -0,0 +1,24 @@
+# Fuzzing
+
+Three differential fuzzers. Each draws random cases from a seed, runs them two ways that must
+agree, and on a divergence prints the seed that reproduces it and the case shrunk to the lines
+worth pasting into a test. The CI runs a few hundred rounds of each on a seed nobody chose, on
+every push and pull request.
+
+| Tool | Arms | Needs a server |
+| --- | --- | --- |
+| `node fuzz/wire.js` | the pg-protocol parser on random backend messages, in one buffer and cut at random points, against what was written | no |
+| `node fuzz/modes.js` | `client.query` against the extended protocol forced, a named statement run twice, rowMode array, the binary result format, pipeline mode, pg-cursor and pg-query-stream | yes, `PG*` variables |
+| `node fuzz/native.js` | pg against pg-native, as a plain query, a named statement and rowMode array | yes, and pg-native built |
+
+```bash
+node fuzz/modes.js --rounds 200        # longer
+node fuzz/modes.js --seed 12345 --rounds 1   # replay what a past run printed
+node fuzz/modes.js --keep-going        # do not stop at the first divergence
+node fuzz/modes.js --no-shrink         # print the round as drawn
+```
+
+The queries come from `queries.js`: selects over `generate_series` with a column per type family,
+sometimes as a parameter instead of a literal, writes on a temp table, and statements that fail
+on purpose, some of them only after rows were sent. What the binary arm can compare is limited to
+the types pg-types has a binary parser for, the `binary` flag of each type says which.

fuzz/lib.js

@@ -0,0 +1,131 @@
+'use strict'
+
+// What the three fuzzers share: a seeded generator, so a failure prints the seed that reproduces
+// it; the loop over rounds; and the shrinking, which drops the parts of a failing case one at a
+// time for as long as the failure survives, so what gets printed is the few lines worth pasting
+// into a test rather than the whole round.
+
+/** @param {number} seed @returns {() => number} the same sequence for the same seed */
+const mulberry32 = (seed) => {
+  let a = seed >>> 0
+  return () => {
+    a = (a + 0x6d2b79f5) >>> 0
+    let t = Math.imul(a ^ (a >>> 15), 1 | a)
+    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t
+    return ((t ^ (t >>> 14)) >>> 0) / 4294967296
+  }
+}
+
+const int = (rng, min, max) => min + Math.floor(rng() * (max - min + 1))
+const pick = (rng, items) => items[Math.floor(rng() * items.length)]
+const chance = (rng, p) => rng() < p
+
+// strings a server can send: no NUL, since the protocol delimits them with one, and a spread
+// from empty to long with multibyte and awkward characters in between
+const ALPHABET = ['a', 'Z', '0', ' ', '_', "'", '"', '\\', '{', '}', ',', 'é', '€', '😀', '\n', '\t', 'ÿ']
+const string = (rng, max = 12) => {
+  const length = chance(rng, 0.1) ? 0 : chance(rng, 0.05) ? int(rng, 100, max * 40) : int(rng, 1, max)
+  let out = ''
+  for (let i = 0; i < length; i++) out += chance(rng, 0.7) ? pick(rng, ALPHABET.slice(0, 3)) : pick(rng, ALPHABET)
+  return out
+}
+
+const bytes = (rng, max = 16) => {
+  const length = chance(rng, 0.1) ? 0 : int(rng, 1, max)
+  const out = Buffer.alloc(length)
+  for (let i = 0; i < length; i++) out[i] = chance(rng, 0.2) ? 0 : int(rng, 0, 255)
+  return out
+}
+
+// one canonical text for a value, so two results compare as strings: a Buffer by its bytes, a
+// Date by its instant, and the rest by JSON with the keys in insertion order
+const canon = (value) => {
+  const replacer = (_, v) => {
+    if (Buffer.isBuffer(v)) return `<${v.toString('hex')}>`
+    if (v && v.type === 'Buffer' && Array.isArray(v.data)) return `<${Buffer.from(v.data).toString('hex')}>`
+    if (typeof v === 'bigint') return `${v}n`
+    if (v === undefined) return '<undefined>'
+    if (typeof v === 'number' && !Number.isFinite(v)) return `<${v}>`
+    return v
+  }
+  return JSON.stringify(value instanceof Date ? value.toISOString() : value, replacer)
+}
+
+const parseArgs = (argv) => {
+  const flag = (name, fallback) => {
+    const at = argv.indexOf(`--${name}`)
+    return at === -1 ? fallback : Number(argv[at + 1])
+  }
+  return {
+    rounds: flag('rounds', 100),
+    seed: flag('seed', (Date.now() ^ (process.pid << 16)) >>> 0),
+    keepGoing: argv.includes('--keep-going'),
+    noShrink: argv.includes('--no-shrink'),
+  }
+}
+
+/**
+ * Drops items from a failing case one at a time, keeping every drop the failure survives.
+ *
+ * @template T
+ * @param {T} plan
+ * @param {(plan: T) => T[]} variants the smaller plans to try, each with one thing removed
+ * @param {(plan: T) => Promise<boolean>} fails
+ */
+const shrink = async (plan, variants, fails) => {
+  let current = plan
+  let progress = true
+  while (progress) {
+    progress = false
+    for (const smaller of variants(current)) {
+      if (await fails(smaller)) {
+        current = smaller
+        progress = true
+        break
+      }
+    }
+  }
+  return current
+}
+
+/**
+ * The loop every fuzzer runs: draw a case per round, run it, and on a divergence print the seed,
+ * shrink the case and print it as source.
+ *
+ * @param {object} fuzzer
+ * @param {string} fuzzer.name
+ * @param {(rng: () => number) => any} fuzzer.draw
+ * @param {(plan: any) => Promise<string|null>} fuzzer.run a description of the divergence, or null
+ * @param {(plan: any) => any[]} fuzzer.variants
+ * @param {(plan: any) => string} fuzzer.source
+ * @param {() => Promise<void>} [fuzzer.close]
+ */
+const main = async (fuzzer) => {
+  const args = parseArgs(process.argv.slice(2))
+  console.log(`${fuzzer.name}: ${args.rounds} rounds from seed ${args.seed}`)
+  let found = 0
+  for (let round = 0; round < args.rounds; round++) {
+    const seed = (args.seed + round) >>> 0
+    const plan = fuzzer.draw(mulberry32(seed))
+    const divergence = await fuzzer.run(plan)
+    if (!divergence) {
+      if (round % 25 === 24) console.log(`  ${round + 1} rounds, no divergence`)
+      continue
+    }
+    found++
+    console.log(`\n=== divergence in round ${round}, seed ${seed} (replay: --seed ${seed} --rounds 1)`)
+    console.log(divergence)
+    if (!args.noShrink) {
+      console.log('\nshrinking...')
+      const small = await shrink(plan, fuzzer.variants, async (p) => Boolean(await fuzzer.run(p)))
+      console.log(`\n${fuzzer.source(small)}`)
+      console.log(await fuzzer.run(small))
+    }
+    if (!args.keepGoing) break
+  }
+  if (fuzzer.close) await fuzzer.close()
+  console.log(`\n${found} divergence${found === 1 ? '' : 's'}`)
+  process.exit(found ? 1 : 0)
+}
+
+module.exports = { mulberry32, int, pick, chance, string, bytes, canon, shrink, main }

fuzz/modes.js

@@ -0,0 +1,144 @@
+'use strict'
+
+// pg against itself: the same statements, on one connection per mode, and every mode must
+// answer what a plain query on a plain client answered.
+//
+// The reference arm is `client.query(text, values)`. The others are the paths a program can
+// take to the same rows: the extended protocol forced on a query that would have used the
+// simple one, a named prepared statement run twice, rowMode array, the binary result format,
+// pipeline mode with the whole round in flight at once, pg-cursor reading in batches and
+// pg-query-stream. Each keeps its own connection and runs the whole round in order, so the
+// temp table and the transaction state are the same on every arm.
+//
+//   node fuzz/modes.js                 fifty rounds, PG* variables say where the server is
+//   node fuzz/modes.js --seed 12345    replay what a past run did
+//   node fuzz/modes.js --keep-going    do not stop at the first divergence
+
+const { createHash } = require('crypto')
+const pg = require('../packages/pg')
+const Cursor = require('../packages/pg-cursor')
+const QueryStream = require('../packages/pg-query-stream')
+const { int, mulberry32, main } = require('./lib')
+const { draw, render, summarize, QUIET, rowsOnly, source, variants, SETUP } = require('./queries')
+
+// the outcome of a query, whichever way it went
+const outcome = (promise) =>
+  promise.then(
+    (result) => ({ result }),
+    (error) => ({ error })
+  )
+const settle = (promise) => outcome(promise).then(({ result, error }) => summarize(result, error))
+const settleRows = (promise) => outcome(promise).then(({ result, error }) => rowsOnly(result, error))
+
+const name = (text) => `fuzz_${createHash('sha1').update(text).digest('hex')}`
+
+// each arm runs one query of the round on its own connection and answers the canonical text.
+// `rng` is the round's, so a batch size is drawn the same on a replay
+const ARMS = {
+  extended: (client, q) => settle(client.query({ text: q.text, values: q.values, queryMode: 'extended' })),
+  // twice, so the second run goes through the statement cache; only a select, since running a
+  // write twice would leave this arm's table different from the others
+  prepared: async (client, q) => {
+    const first = await settle(client.query({ text: q.text, values: q.values, name: name(q.text) }))
+    // and not after a failure either, which would have aborted an open transaction
+    if (q.kind !== 'select' || first.startsWith('{"error"')) return first
+    const second = await settle(client.query({ text: q.text, values: q.values, name: name(q.text) }))
+    return first === second ? first : `first run: ${first}\n  second run: ${second}`
+  },
+  array: async (client, q, rng, reference) => {
+    const got = await settle(client.query({ text: q.text, values: q.values, rowMode: 'array' }))
+    // the reference rows as arrays, which is the only thing this mode changes
+    const { result, error } = reference
+    const expected = summarize(result && { ...result, rows: result.rows.map(Object.values) }, error)
+    return got === expected ? summarize(result, error) : got
+  },
+  binary: (client, q) =>
+    q.binary
+      ? settle(client.query({ text: q.text, values: q.values, binary: true }))
+      : settle(client.query(q.text, q.values)),
+  cursor: async (client, q, rng) => {
+    if (q.kind !== 'select') return settleRows(client.query(q.text, q.values))
+    const cursor = client.query(new Cursor(q.text, q.values))
+    const rows = []
+    try {
+      for (;;) {
+        const batch = await cursor.read(int(rng, 1, 40))
+        if (batch.length === 0) break
+        rows.push(...batch)
+      }
+      await cursor.close()
+      return rowsOnly({ rows })
+    } catch (error) {
+      return rowsOnly(null, error)
+    }
+  },
+  stream: async (client, q, rng) => {
+    if (q.kind !== 'select') return settleRows(client.query(q.text, q.values))
+    const rows = []
+    try {
+      const stream = client.query(
+        new QueryStream(q.text, q.values, { batchSize: int(rng, 1, 40), highWaterMark: int(rng, 1, 40) })
+      )
+      for await (const row of stream) rows.push(row)
+      return rowsOnly({ rows })
+    } catch (error) {
+      return rowsOnly(null, error)
+    }
+  },
+}
+
+const clients = {}
+const connect = async () => {
+  clients.reference = new pg.Client()
+  clients.pipeline = new pg.Client({ pipeline: true })
+  for (const arm of Object.keys(ARMS)) clients[arm] = new pg.Client()
+  for (const client of Object.values(clients)) {
+    await client.connect()
+    await client.query(QUIET)
+  }
+}
+
+// every round starts from the same state on every arm: no transaction open, an empty table
+const reset = async (client) => {
+  await client.query('ROLLBACK').catch(() => {})
+  await client.query('DROP TABLE IF EXISTS fuzz_rows')
+  await client.query(SETUP)
+}
+
+const run = async (plan) => {
+  if (!clients.reference) await connect()
+  for (const client of Object.values(clients)) await reset(client)
+  const rng = mulberry32(plan.queries.length)
+  const queries = plan.queries.map(render)
+  const references = []
+  for (const q of queries) references.push(await outcome(clients.reference.query(q.text, q.values)))
+  const full = references.map(({ result, error }) => summarize(result, error))
+
+  // the whole round in flight at once on the pipelined connection
+  const pipelined = await Promise.all(queries.map((q) => settle(clients.pipeline.query(q.text, q.values))))
+  for (let i = 0; i < queries.length; i++) {
+    if (pipelined[i] !== full[i])
+      return `pipeline, query ${i}
+  reference: ${full[i]}
+  pipeline:  ${pipelined[i]}`
+  }
+
+  for (const [arm, runArm] of Object.entries(ARMS)) {
+    for (let i = 0; i < queries.length; i++) {
+      const got = await runArm(clients[arm], queries[i], rng, references[i])
+      const { result, error } = references[i]
+      const expected = arm === 'cursor' || arm === 'stream' ? rowsOnly(result, error) : full[i]
+      if (got !== expected)
+        return `${arm}, query ${i}
+  reference: ${expected}
+  ${arm}: ${got}`
+    }
+  }
+  return null
+}
+
+const close = async () => {
+  for (const client of Object.values(clients)) await client.end()
+}
+
+main({ name: 'modes', draw, run, variants, source, close })