diff --git a/test/docker/prestart-map.mjs b/test/docker/prestart-map.mjs
index a8c0c0d9b5bd..e377ea4ba9ad 100644
--- a/test/docker/prestart-map.mjs
+++ b/test/docker/prestart-map.mjs
@@ -19,6 +19,7 @@ export const prestartMap = {
   "js/sql/postgres-binary-numeric": ["postgres_plain"],
   "js/sql/postgres-multi-statement-fields": ["postgres_plain"],
   "js/sql/postgres-simple-query-pipeline": ["postgres_plain"],
+  "js/sql/postgres-listen-notify": ["postgres_plain"],
   "js/sql/sql-onconnect-onclose-throw": ["postgres_plain", "mysql_plain"],
   "js/sql/sql-prepare-false": ["postgres_plain"],
   "js/valkey/": ["redis_unified"],
diff --git a/test/js/sql/postgres-listen-notify.test.ts b/test/js/sql/postgres-listen-notify.test.ts
index 731ace6d4957..40fab43b1ccb 100644
--- a/test/js/sql/postgres-listen-notify.test.ts
+++ b/test/js/sql/postgres-listen-notify.test.ts
@@ -6,6 +6,11 @@
 // real server cannot be made to produce on demand. Behavior only observable
 // at process level (exit, uncaught exceptions, RSS) runs in a subprocess.
 // End-to-end behavior runs against the docker-compose postgres service.
+//
+// Every test owns its mock server and client (or, against the shared postgres
+// service, its channel names), so the suites run concurrently: each reconnect
+// test sits out at least one 250ms backoff, and each subprocess test pays a
+// debug-build startup.
 
 import { SQL } from "bun";
 import { describe, expect, test } from "bun:test";
@@ -143,7 +148,246 @@ function gate() {
   });
 }
 
-describe("listen", () => {
+// The rejection of a promise that may still be pending, awaited plainly.
+// `expect(promise).rejects` on a pending promise re-enters the event loop
+// (#33261); from inside a socket callback, which is where a continuation
+// after `await sql.listen()` runs, the inner run drops the outer dispatch's
+// remaining events, and a subprocess test whose exit event is dropped hangs.
+async function rejection(promise: Promise<unknown>): Promise<unknown> {
+  try {
+    await promise;
+  } catch (err) {
+    return err;
+  }
+  throw new Error("expected a rejection");
+}
+
+const connectionClosed = {
+  name: "PostgresError",
+  code: "ERR_POSTGRES_CONNECTION_CLOSED",
+  message: "Connection closed",
+};
+/** What the mock's ErrorResponse (SQLSTATE 42601) becomes on the client. */
+const serverError = (message: string) => ({ name: "PostgresError", code: "ERR_POSTGRES_SYNTAX_ERROR", message });
+const invalidChannel = (channel: string, reason: string) => ({
+  name: "TypeError",
+  code: "ERR_INVALID_ARG_VALUE",
+  message: `The argument 'channel' ${reason}. Received '${channel}'`,
+});
+const invalidCallback = (name: string) => ({
+  name: "TypeError",
+  code: "ERR_INVALID_ARG_TYPE",
+  message: `The "${name}" argument must be of type function. Received type number (1)`,
+});
+
+// Declared first: the runner hands out concurrency slots in declaration
+// order (5 under ASAN, 20 otherwise) and these are the slowest tests.
+describe.concurrent("in a subprocess", () => {
+  const wireFrames = path.join(import.meta.dir, "wire-frames.ts");
+  // The mock runs inside the child and is unref'd, so only the subscription
+  // under test can hold the child open. `notifyMany`, `dropConnections`,
+  // `sockets`, `pgNotificationResponse` and `sql` are in scope for `body`.
+  async function run(body: string, env: Record<string, string> = {}) {
+    using dir = tempDir("pg-listen-subprocess", {
+      "fixture.ts": `
+        import { SQL } from "bun";
+        import { listeningServer, pgAuthenticationOk, pgReadyForQuery, pgCommandComplete, pgNotificationResponse, pgReadFrontendMessages, pgRaw, pgInt32 } from ${JSON.stringify(wireFrames)};
+        const sockets = new Set();
+        const { port, server } = await listeningServer(socket => {
+          sockets.add(socket);
+          socket.unref();
+          let buffered = Buffer.alloc(0);
+          socket.once("data", () => {
+            socket.write(Buffer.concat([pgAuthenticationOk(), pgRaw("K", Buffer.concat([pgInt32(1), pgInt32(2)])), pgReadyForQuery()]));
+            socket.on("data", data => {
+              buffered = pgReadFrontendMessages(Buffer.concat([buffered, data]), (type, body) => {
+                if (type !== 0x51) return;
+                socket.write(Buffer.concat([pgCommandComplete(body.toString("utf8", 0, body.indexOf(" "))), pgReadyForQuery()]));
+              });
+            });
+          });
+          socket.on("close", () => sockets.delete(socket));
+        });
+        server.unref();
+        const notifyMany = frames => {
+          const blob = Buffer.concat(frames.map(([channel, payload]) => pgNotificationResponse(1, channel, payload)));
+          for (const socket of sockets) socket.write(blob);
+        };
+        const dropConnections = () => { for (const socket of sockets) socket.destroy(); };
+        const sql = new SQL("postgres://u@127.0.0.1:" + port + "/db", { max: 1 });
+        ${body}
+      `,
+    });
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "fixture.ts"],
+      cwd: String(dir),
+      env: { ...bunEnv, ...env },
+      stderr: "pipe",
+      timeout: 30_000,
+      killSignal: "SIGKILL",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    return {
+      stdout,
+      stderr: stderr
+        .split("\n")
+        .filter(line => line && !line.startsWith("WARNING: ASAN"))
+        .join("\n"),
+      exitCode,
+      signal: proc.signalCode,
+    };
+  }
+  const clean = (stdout: string) => ({ stdout, stderr: "", exitCode: 0, signal: null });
+
+  test("a subscription keeps the process alive until it is removed", async () => {
+    const result = await run(`
+      const subscription = await sql.listen("ch", async payload => {
+        console.log("got " + payload);
+        await subscription.unlisten();
+        console.log("unlistened");
+      });
+      console.log("subscribed");
+      // Sent from an unref'd timer after top-level code finishes, so the
+      // subscription is the only thing keeping the process here.
+      setTimeout(() => notifyMany([["ch", "wake"]]), 20).unref();
+    `);
+    expect(result).toEqual(clean("subscribed\ngot wake\nunlistened\n"));
+  });
+
+  test("a dropped connection keeps the process alive through the backoff and reconnects", async () => {
+    const result = await run(`
+      let subscribes = 0;
+      const subscription = await sql.listen("ch", () => {}, () => {
+        if (++subscribes === 2) { console.log("reconnected"); subscription.unlisten(); }
+      });
+      console.log("subscribed");
+      dropConnections();
+    `);
+    expect(result).toEqual(clean("subscribed\nreconnected\n"));
+  });
+
+  test("sql.close() releases a subscription so the process exits", async () => {
+    const result = await run(`
+      await sql.listen("ch", () => {});
+      await sql.close();
+      console.log("closed");
+    `);
+    expect(result).toEqual(clean("closed\n"));
+  });
+
+  test("a throwing listener surfaces as uncaughtException; the channel's other listeners and later notifications are unaffected", async () => {
+    const result = await run(`
+      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
+      const first = await sql.listen("ch", payload => {
+        console.log("first got " + payload);
+        if (payload === "bad") throw new Error("listener failed");
+      });
+      const second = await sql.listen("ch", payload => {
+        console.log("second got " + payload);
+        if (payload === "good") { first.unlisten(); second.unlisten(); }
+      });
+      notifyMany([["ch", "bad"], ["ch", "good"]]);
+    `);
+    expect(result).toEqual(
+      clean(
+        ["first got bad", "uncaught: listener failed", "second got bad", "first got good", "second got good", ""].join(
+          "\n",
+        ),
+      ),
+    );
+  });
+
+  test("a lone throwing listener surfaces as uncaughtException and later notifications still arrive", async () => {
+    const result = await run(`
+      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
+      const subscription = await sql.listen("ch", payload => {
+        console.log("got " + payload);
+        if (payload === "bad") throw new Error("listener failed");
+        subscription.unlisten();
+      });
+      notifyMany([["ch", "bad"], ["ch", "good"]]);
+    `);
+    expect(result).toEqual(clean("got bad\nuncaught: listener failed\ngot good\n"));
+  });
+
+  test("a throwing onlisten surfaces as uncaughtException; listen() resolves and the reconnect is not retried", async () => {
+    // An empty stderr is the assertion that the reconnect sweep did not take
+    // the second throw for a failed LISTEN (which it warns about and retries).
+    const result = await run(`
+      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
+      let calls = 0;
+      const subscription = await sql.listen("ch", () => {}, () => {
+        console.log("onlisten " + ++calls);
+        if (calls === 2) subscription.unlisten().then(() => console.log("unlistened"));
+        throw new Error("onlisten failed " + calls);
+      });
+      console.log("subscribed to " + subscription.channel);
+      dropConnections();
+    `);
+    expect(result).toEqual(
+      clean(
+        [
+          "onlisten 1",
+          "uncaught: onlisten failed 1",
+          "subscribed to ch",
+          "onlisten 2",
+          "uncaught: onlisten failed 2",
+          "unlistened",
+          "",
+        ].join("\n"),
+      ),
+    );
+  });
+
+  test("delivering many notifications retains nothing", async () => {
+    // Measured by RSS so a leaked native string backing (invisible to the JS
+    // heap) is caught. Each phase delivers its volume twice and reports the
+    // growth across the second pass: the first pass takes RSS to its steady
+    // state (the allocator keeps the GC's between-collection high-water mark
+    // of payload strings as free memory), after which a correct
+    // implementation shows roughly no growth and leaking one string per
+    // notification shows the pass's full payload volume.
+    //
+    // Bun.shrink() returns the heap's freed blocks to the OS at the next idle
+    // point; without it the reading drifts by up to 8 MiB from pass to pass.
+    const result = await run(
+      `
+      const channels = ["a", "b", "c", "d"];
+      let listener;
+      for (const channel of channels) await sql.listen(channel, payload => listener(payload));
+      const segment = (count, payload) => [count, Buffer.concat(Array.from({ length: count }, (_, i) => pgNotificationResponse(1, channels[i % 4], payload)))];
+      const deliver = ([count, blob]) => new Promise(done => {
+        let remaining = count;
+        listener = () => { if (--remaining === 0) done(); };
+        for (const socket of sockets) socket.write(blob);
+      });
+      const rss = async () => { Bun.gc(true); Bun.shrink(); await Bun.sleep(0); return process.memoryUsage.rss(); };
+      const measure = async (seg, rounds) => {
+        const pass = async () => { for (let i = 0; i < rounds; i++) await deliver(seg); };
+        await pass();
+        const base = await rss();
+        await pass();
+        return Math.round(((await rss()) - base) / 1024 / 1024);
+      };
+      // 256 KiB per segment, 96 segments = 24 MiB per pass.
+      const small = await measure(segment(256, Buffer.alloc(1024, 0x61).toString()), 96);
+      const large = await measure(segment(4, Buffer.alloc(64 * 1024, 0x62).toString()), 96);
+      console.log(JSON.stringify({ small, large }));
+      await sql.close();
+    `,
+      { ASAN_OPTIONS: "quarantine_size_mb=4:detect_leaks=0" },
+    );
+    expect(result.stderr).toBe("");
+    expect(result.exitCode).toBe(0);
+    const growth = JSON.parse(result.stdout);
+    // Steady state moves by a few MiB between passes; a leak adds ~24 MiB.
+    expect(growth).toEqual({ small: expect.any(Number), large: expect.any(Number) });
+    expect(growth.small, JSON.stringify(growth)).toBeLessThan(6);
+    expect(growth.large, JSON.stringify(growth)).toBeLessThan(6);
+  }, 30_000);
+});
+
+describe.concurrent("listen", () => {
   test("routes notifications to the channel's listener, in order", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
@@ -214,9 +458,10 @@ describe("listen", () => {
     const ascii63 = Buffer.alloc(63, "c").toString();
     const cjk63 = Buffer.alloc(63, "字").toString(); // 21 characters
     const cjk66 = Buffer.alloc(66, "字").toString(); // 22 characters, 66 bytes
-    await expect(sql.listen(ascii63 + "c", () => {})).rejects.toThrow(/63 bytes/);
-    await expect(sql.listen(cjk66, () => {})).rejects.toThrow(/63 bytes/);
-    expect(() => sql.notify(cjk66)).toThrow(/63 bytes/);
+    const tooLong = (channel: string) => invalidChannel(channel, "must be at most 63 bytes");
+    expect(await rejection(sql.listen(ascii63 + "c", () => {}))).toMatchObject(tooLong(ascii63 + "c"));
+    expect(await rejection(sql.listen(cjk66, () => {}))).toMatchObject(tooLong(cjk66));
+    expect(() => sql.notify(cjk66)).toThrow(expect.objectContaining(tooLong(cjk66)));
 
     await sql.listen(ascii63, () => {});
     await sql.listen(cjk63, () => {});
@@ -244,14 +489,23 @@ describe("listen", () => {
   test("concurrent listen() calls on a new channel share its round trip", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
+    const got: string[] = [];
     const all = gate();
     const count = all.after(3);
-    const subscriptions = await Promise.all([1, 2, 3].map(() => sql.listen("ch", () => count())));
+    const subscriptions = await Promise.all(
+      [1, 2, 3].map(n =>
+        sql.listen("ch", payload => {
+          got.push(`${n}:${payload}`);
+          count();
+        }),
+      ),
+    );
     expect(server.queries).toEqual(['LISTEN "ch"']);
     expect(subscriptions.map(subscription => subscription.channel)).toEqual(["ch", "ch", "ch"]);
     expect(new Set(subscriptions).size).toBe(3);
     server.notify("ch", "x");
     await all;
+    expect(got).toEqual(["1:x", "2:x", "3:x"]);
   });
 
   test("a listener that unlistens itself mid-dispatch does not starve the others", async () => {
@@ -319,10 +573,10 @@ describe("listen", () => {
   test("registering the same callback twice is two subscriptions, each removed by its own handle", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
-    let calls = 0;
+    const got: string[] = [];
     let count = () => {};
-    const callback = () => {
-      calls++;
+    const callback = (payload: string) => {
+      got.push(payload);
       count();
     };
     const first = await sql.listen("ch", callback);
@@ -333,7 +587,7 @@ describe("listen", () => {
     count = both.after(2);
     server.notify("ch", "x");
     await both;
-    expect(calls).toBe(2);
+    expect(got).toEqual(["x", "x"]);
 
     await first.unlisten();
     const one = gate();
@@ -342,7 +596,7 @@ describe("listen", () => {
     await one;
     // This round trip is ordered after any second delivery of "y".
     await sql.listen("barrier", () => {});
-    expect(calls).toBe(3);
+    expect(got).toEqual(["x", "x", "y"]);
     expect(server.queries).toEqual(['LISTEN "ch"', 'LISTEN "barrier"']);
 
     await second.unlisten();
@@ -378,7 +632,7 @@ describe("listen", () => {
   });
 });
 
-describe("unlisten", () => {
+describe.concurrent("unlisten", () => {
   test("removing one of several listeners is local; removing the last sends UNLISTEN", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
@@ -456,7 +710,9 @@ describe("unlisten", () => {
     const unlistening = subscription.unlisten();
     await server.untilQuery(queries => queries.includes('UNLISTEN "ch"'));
     server.dropConnections();
-    await unlistening;
+    // Plain await: this continuation is inside the mock's data callback (see the reconnect suite).
+    expect(await unlistening).toBeUndefined();
+    expect(server.queries).toEqual(['LISTEN "keep"', 'LISTEN "ch"', 'UNLISTEN "ch"']);
   });
 
   test("a handle from before close() is a no-op afterwards, also for a re-opened client", async () => {
@@ -476,7 +732,7 @@ describe("unlisten", () => {
   });
 });
 
-describe("listen/unlisten interleavings", () => {
+describe.concurrent("listen/unlisten interleavings", () => {
   test("a re-listen() while the UNLISTEN is in flight gets its own LISTEN, ordered after it", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
@@ -498,7 +754,7 @@ describe("listen/unlisten interleavings", () => {
     await using sql = client(server.url);
     await sql.listen("keep", () => {});
     server.failNextListen("bad");
-    await expect(sql.listen("bad", () => {})).rejects.toThrow("cannot LISTEN bad");
+    expect(await rejection(sql.listen("bad", () => {}))).toMatchObject(serverError("cannot LISTEN bad"));
     await sql.listen("bad", () => {});
     expect(server.queries).toEqual(['LISTEN "keep"', 'LISTEN "bad"', 'LISTEN "bad"']);
   });
@@ -512,7 +768,11 @@ describe("listen/unlisten interleavings", () => {
     const b = sql.listen("ch", () => {});
     server.release();
     const results = await Promise.allSettled([a, b]);
-    expect(results.map(result => result.status)).toEqual(["rejected", "rejected"]);
+    expect(results).toEqual([
+      { status: "rejected", reason: expect.objectContaining(serverError("cannot LISTEN ch")) },
+      { status: "rejected", reason: expect.objectContaining(serverError("cannot LISTEN ch")) },
+    ]);
+    expect(server.queries).toEqual(['LISTEN "ch"']);
     await server.untilClosed(1);
   });
 
@@ -520,14 +780,18 @@ describe("listen/unlisten interleavings", () => {
     const { port, server } = await listeningServer(socket => socket.destroy());
     try {
       await using sql = client(`postgres://u@127.0.0.1:${port}/db`);
-      await expect(sql.listen("ch", () => {})).rejects.toThrow();
+      expect(await rejection(sql.listen("ch", () => {}))).toMatchObject({
+        name: "PostgresError",
+        code: "ERR_POSTGRES_CONNECTION_FAILED",
+        message: "Connection closed before the connection was established",
+      });
     } finally {
       server.close();
     }
   });
 });
 
-describe("reconnect", () => {
+describe.concurrent("reconnect", () => {
   test("re-subscribes every channel, runs onlisten again, resumes delivery", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
@@ -592,9 +856,15 @@ describe("reconnect", () => {
   test("a channel whose re-LISTEN is rejected is retried with backoff and warns", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
+    // console.warn is process-wide and this suite is concurrent, so keep only
+    // the warnings that name this test's channel.
     const warnings: string[] = [];
     const originalWarn = console.warn;
-    console.warn = (...args: unknown[]) => void warnings.push(args.join(" "));
+    console.warn = (...args: unknown[]) => {
+      const line = args.join(" ");
+      if (line.includes('"flaky"')) warnings.push(line);
+      else originalWarn(...args);
+    };
     try {
       const resubscribed = gate();
       await sql.listen("flaky", () => {}, resubscribed.after(2));
@@ -602,7 +872,7 @@ describe("reconnect", () => {
       server.dropConnections();
       await resubscribed;
       expect(server.connections.at(-1)!.filter(query => query === 'LISTEN "flaky"')).toHaveLength(2);
-      expect(warnings).toEqual([expect.stringContaining('LISTEN "flaky" failed, retrying: cannot LISTEN flaky')]);
+      expect(warnings).toEqual(['bun:sql LISTEN "flaky" failed, retrying: cannot LISTEN flaky']);
     } finally {
       console.warn = originalWarn;
     }
@@ -637,8 +907,9 @@ describe("reconnect", () => {
     await sql.listen("ch", got.resolve, repaired.after(2));
 
     // Plain awaits rather than expect().rejects: that one runs the event loop
-    // re-entrantly, and this code is still inside the dropped connection's
-    // data callback; on Windows the inner run frees the socket under it.
+    // re-entrantly (#33261), and this code is still inside the dropped
+    // connection's data callback; on Windows the inner run frees the socket
+    // under it (#39643).
     const outcome = (promise: Promise<unknown>) =>
       promise.then(
         () => "resolved",
@@ -649,7 +920,7 @@ describe("reconnect", () => {
     // and rejects once the client has processed the drop, which is the moment
     // the client considers "ch" unsubscribed and has armed its backoff.
     server.dropConnections();
-    expect(await outcome(sql.listen("probe", () => {}))).toMatch(/closed/i);
+    expect(await outcome(sql.listen("probe", () => {}))).toBe("Connection closed");
 
     // This listen() brings the connection back (cancelling the backoff) and is
     // the one sending LISTEN "ch"; its rejection must not strand the first listener.
@@ -663,7 +934,7 @@ describe("reconnect", () => {
   });
 });
 
-describe("close()", () => {
+describe.concurrent("close()", () => {
   test("tears down the listen connection and rejects an in-flight listen()", async () => {
     await using server = await mockServer();
     const sql = client(server.url);
@@ -673,10 +944,11 @@ describe("close()", () => {
     await server.untilQuery(queries => queries.includes('LISTEN "pending"'));
 
     await sql.close();
-    await expect(pending).rejects.toThrow();
+    expect(await rejection(pending)).toMatchObject(connectionClosed);
     await server.untilClosed(1);
-    await expect(sql.listen("ch", () => {})).rejects.toThrow("Connection closed");
-    await subscription.unlisten();
+    expect(await rejection(sql.listen("ch", () => {}))).toMatchObject(connectionClosed);
+    expect(await subscription.unlisten()).toBeUndefined();
+    expect(server.queries).toEqual(['LISTEN "ch"', 'LISTEN "pending"']);
   });
 
   test("in the same tick as a listen() on the live connection rejects it before any LISTEN is sent", async () => {
@@ -702,7 +974,7 @@ describe("close()", () => {
       const sql = new SQL(`postgres://u@127.0.0.1:${port}/db`, { max: 1, connectionTimeout: 60 });
       const listening = sql.listen("ch", () => {});
       await sql.close();
-      await expect(listening).rejects.toThrow();
+      expect(await rejection(listening)).toMatchObject(connectionClosed);
     } finally {
       server.close();
     }
@@ -713,30 +985,44 @@ describe("close()", () => {
     await using sql = client(server.url);
     const got = Promise.withResolvers<string>();
     await sql.listen("ch", got.resolve);
-    await expect(sql.close({ timeout: -1 })).rejects.toThrow();
+    expect(await rejection(sql.close({ timeout: -1 }))).toMatchObject({
+      name: "TypeError",
+      code: "ERR_INVALID_ARG_VALUE",
+      message: "The property 'options.timeout' must be a non-negative integer less than 2^31. Received -1",
+    });
     server.notify("ch", "still subscribed");
     expect(await got.promise).toBe("still subscribed");
+    expect(server.liveConnections).toBe(1);
   });
 });
 
-describe("arguments", () => {
+describe.concurrent("arguments", () => {
   test.each(["postgres", "sqlite"] as const)("%s: invalid arguments fail before any I/O", async adapter => {
     await using sql = adapter === "postgres" ? client("postgres://u@127.0.0.1:1/db") : new SQL("sqlite://:memory:");
     const callback = () => {};
-    await expect(sql.listen("", callback)).rejects.toThrow(/non-empty/);
-    await expect(sql.listen("a\0b", callback)).rejects.toThrow(/null bytes/);
-    await expect(sql.listen("ch", 1 as any)).rejects.toThrow(/onnotify/);
-    await expect(sql.listen("ch", callback, 1 as any)).rejects.toThrow(/onlisten/);
-    expect(() => sql.notify("", "p")).toThrow(/non-empty/);
-    expect(() => sql.notify("ch", null as any)).toThrow(/payload/);
-    expect(() => sql.notify("ch", 1 as any)).toThrow(/payload/);
+    const emptyChannel = invalidChannel("", "must be a non-empty string");
+    const invalidPayload = (received: string) => ({
+      name: "TypeError",
+      code: "ERR_INVALID_ARG_TYPE",
+      message: `The "payload" argument must be of type string. Received ${received}`,
+    });
+    expect(await rejection(sql.listen("", callback))).toMatchObject(emptyChannel);
+    expect(await rejection(sql.listen("a\0b", callback))).toMatchObject(
+      invalidChannel("a\\x00b", "must not contain null bytes"),
+    );
+    expect(await rejection(sql.listen("ch", 1 as any))).toMatchObject(invalidCallback("onnotify"));
+    expect(await rejection(sql.listen("ch", callback, 1 as any))).toMatchObject(invalidCallback("onlisten"));
+    expect(() => sql.notify("", "p")).toThrow(expect.objectContaining(emptyChannel));
+    expect(() => sql.notify("ch", null as any)).toThrow(expect.objectContaining(invalidPayload("null")));
+    expect(() => sql.notify("ch", 1 as any)).toThrow(expect.objectContaining(invalidPayload("type number (1)")));
   });
 
   test("non-Postgres adapters reject with a clear error", async () => {
     await using sql = new SQL("sqlite://:memory:");
-    await expect(sql.listen("ch", () => {})).rejects.toThrow("PostgreSQL only");
-    await expect(sql.notify("ch", "p")).rejects.toThrow("PostgreSQL only");
-    await expect(sql.notify("ch")).rejects.toThrow("PostgreSQL only");
+    const unsupported = { name: "Error", message: "LISTEN/NOTIFY is not supported by this adapter (PostgreSQL only)" };
+    expect(await rejection(sql.listen("ch", () => {}))).toMatchObject(unsupported);
+    expect(await rejection(sql.notify("ch", "p"))).toMatchObject(unsupported);
+    expect(await rejection(sql.notify("ch"))).toMatchObject(unsupported);
   });
 
   test("reserved connections expose listen() on the client's shared listen connection; unlisten lives on the subscription only", async () => {
@@ -757,7 +1043,7 @@ describe("arguments", () => {
   });
 });
 
-describe("notify()", () => {
+describe.concurrent("notify()", () => {
   test("is sent through the pool without being awaited, and opens no listen connection", async () => {
     await using server = await mockServer();
     await using sql = client(server.url);
@@ -766,215 +1052,20 @@ describe("notify()", () => {
     // to end in the docker suite.
     const settled = Promise.allSettled([sql.notify("ch", "payload"), sql.notify("signal")]);
     await server.untilQuery(queries => queries.some(query => query.includes("pg_notify($1, $2)")));
-    expect((await settled).map(result => result.status)).toEqual(["rejected", "rejected"]);
-    expect(server.liveConnections).toBe(1);
-  });
-});
-
-describe("in a subprocess", () => {
-  const wireFrames = path.join(import.meta.dir, "wire-frames.ts");
-  // The mock runs inside the child and is unref'd, so only the subscription
-  // under test can hold the child open. `notifyMany`, `dropConnections`,
-  // `sockets`, `pgNotificationResponse` and `sql` are in scope for `body`.
-  async function run(body: string, env: Record<string, string> = {}) {
-    using dir = tempDir("pg-listen-subprocess", {
-      "fixture.ts": `
-        import { SQL } from "bun";
-        import { listeningServer, pgAuthenticationOk, pgReadyForQuery, pgCommandComplete, pgNotificationResponse, pgReadFrontendMessages, pgRaw, pgInt32 } from ${JSON.stringify(wireFrames)};
-        const sockets = new Set();
-        const { port, server } = await listeningServer(socket => {
-          sockets.add(socket);
-          socket.unref();
-          let buffered = Buffer.alloc(0);
-          socket.once("data", () => {
-            socket.write(Buffer.concat([pgAuthenticationOk(), pgRaw("K", Buffer.concat([pgInt32(1), pgInt32(2)])), pgReadyForQuery()]));
-            socket.on("data", data => {
-              buffered = pgReadFrontendMessages(Buffer.concat([buffered, data]), (type, body) => {
-                if (type !== 0x51) return;
-                socket.write(Buffer.concat([pgCommandComplete(body.toString("utf8", 0, body.indexOf(" "))), pgReadyForQuery()]));
-              });
-            });
-          });
-          socket.on("close", () => sockets.delete(socket));
-        });
-        server.unref();
-        const notifyMany = frames => {
-          const blob = Buffer.concat(frames.map(([channel, payload]) => pgNotificationResponse(1, channel, payload)));
-          for (const socket of sockets) socket.write(blob);
-        };
-        const dropConnections = () => { for (const socket of sockets) socket.destroy(); };
-        const sql = new SQL("postgres://u@127.0.0.1:" + port + "/db", { max: 1 });
-        ${body}
-      `,
-    });
-    await using proc = Bun.spawn({
-      cmd: [bunExe(), "fixture.ts"],
-      cwd: String(dir),
-      env: { ...bunEnv, ...env },
-      stderr: "pipe",
-      timeout: 30_000,
-      killSignal: "SIGKILL",
-    });
-    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
-    return {
-      stdout,
-      stderr: stderr
-        .split("\n")
-        .filter(line => line && !line.startsWith("WARNING: ASAN"))
-        .join("\n"),
-      exitCode,
-      signal: proc.signalCode,
+    const rejected = {
+      status: "rejected",
+      reason: expect.objectContaining(serverError("mock rejects extended-protocol queries")),
     };
-  }
-  const clean = (stdout: string) => ({ stdout, stderr: "", exitCode: 0, signal: null });
-
-  test("a subscription keeps the process alive until it is removed", async () => {
-    const result = await run(`
-      const subscription = await sql.listen("ch", async payload => {
-        console.log("got " + payload);
-        await subscription.unlisten();
-        console.log("unlistened");
-      });
-      console.log("subscribed");
-      // Sent from an unref'd timer after top-level code finishes, so the
-      // subscription is the only thing keeping the process here.
-      setTimeout(() => notifyMany([["ch", "wake"]]), 20).unref();
-    `);
-    expect(result).toEqual(clean("subscribed\ngot wake\nunlistened\n"));
-  });
-
-  test("a dropped connection keeps the process alive through the backoff and reconnects", async () => {
-    const result = await run(`
-      let subscribes = 0;
-      const subscription = await sql.listen("ch", () => {}, () => {
-        if (++subscribes === 2) { console.log("reconnected"); subscription.unlisten(); }
-      });
-      console.log("subscribed");
-      dropConnections();
-    `);
-    expect(result).toEqual(clean("subscribed\nreconnected\n"));
-  });
-
-  test("sql.close() releases a subscription so the process exits", async () => {
-    const result = await run(`
-      await sql.listen("ch", () => {});
-      await sql.close();
-      console.log("closed");
-    `);
-    expect(result).toEqual(clean("closed\n"));
-  });
-
-  test("a throwing listener surfaces as uncaughtException; the channel's other listeners and later notifications are unaffected", async () => {
-    const result = await run(`
-      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
-      const first = await sql.listen("ch", payload => {
-        console.log("first got " + payload);
-        if (payload === "bad") throw new Error("listener failed");
-      });
-      const second = await sql.listen("ch", payload => {
-        console.log("second got " + payload);
-        if (payload === "good") { first.unlisten(); second.unlisten(); }
-      });
-      notifyMany([["ch", "bad"], ["ch", "good"]]);
-    `);
-    expect(result).toEqual(
-      clean(
-        ["first got bad", "uncaught: listener failed", "second got bad", "first got good", "second got good", ""].join(
-          "\n",
-        ),
-      ),
-    );
-  });
-
-  test("a lone throwing listener surfaces as uncaughtException and later notifications still arrive", async () => {
-    const result = await run(`
-      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
-      const subscription = await sql.listen("ch", payload => {
-        console.log("got " + payload);
-        if (payload === "bad") throw new Error("listener failed");
-        subscription.unlisten();
-      });
-      notifyMany([["ch", "bad"], ["ch", "good"]]);
-    `);
-    expect(result).toEqual(clean("got bad\nuncaught: listener failed\ngot good\n"));
-  });
-
-  test("a throwing onlisten surfaces as uncaughtException; listen() resolves and the reconnect is not retried", async () => {
-    // An empty stderr is the assertion that the reconnect sweep did not take
-    // the second throw for a failed LISTEN (which it warns about and retries).
-    const result = await run(`
-      process.on("uncaughtException", err => console.log("uncaught: " + err.message));
-      let calls = 0;
-      const subscription = await sql.listen("ch", () => {}, () => {
-        console.log("onlisten " + ++calls);
-        if (calls === 2) subscription.unlisten().then(() => console.log("unlistened"));
-        throw new Error("onlisten failed " + calls);
-      });
-      console.log("subscribed to " + subscription.channel);
-      dropConnections();
-    `);
-    expect(result).toEqual(
-      clean(
-        [
-          "onlisten 1",
-          "uncaught: onlisten failed 1",
-          "subscribed to ch",
-          "onlisten 2",
-          "uncaught: onlisten failed 2",
-          "unlistened",
-          "",
-        ].join("\n"),
-      ),
-    );
+    expect(await settled).toEqual([rejected, rejected]);
+    expect(server.queries).toEqual(["SELECT pg_notify($1, $2)"]);
+    expect(server.liveConnections).toBe(1);
   });
-
-  test("delivering many notifications retains nothing", async () => {
-    // Measured by RSS so a leaked native string backing (invisible to the JS
-    // heap) is caught. Each phase delivers its volume twice and reports the
-    // growth across the second pass: the first pass takes RSS to its steady
-    // state (the allocator keeps the GC's between-collection high-water mark
-    // of payload strings as free memory), after which a correct
-    // implementation shows roughly no growth and leaking one string per
-    // notification shows the pass's full payload volume.
-    const result = await run(
-      `
-      const channels = ["a", "b", "c", "d"];
-      let listener;
-      for (const channel of channels) await sql.listen(channel, payload => listener(payload));
-      const segment = (count, payload) => [count, Buffer.concat(Array.from({ length: count }, (_, i) => pgNotificationResponse(1, channels[i % 4], payload)))];
-      const deliver = ([count, blob]) => new Promise(done => {
-        let remaining = count;
-        listener = () => { if (--remaining === 0) done(); };
-        for (const socket of sockets) socket.write(blob);
-      });
-      const rss = () => { Bun.gc(true); return process.memoryUsage.rss(); };
-      const measure = async (seg, rounds) => {
-        const pass = async () => { for (let i = 0; i < rounds; i++) await deliver(seg); };
-        await pass();
-        const base = rss();
-        await pass();
-        return Math.round((rss() - base) / 1024 / 1024);
-      };
-      // 256 KiB per segment, 96 segments = 24 MiB per pass.
-      const small = await measure(segment(256, Buffer.alloc(1024, 0x61).toString()), 96);
-      const large = await measure(segment(4, Buffer.alloc(64 * 1024, 0x62).toString()), 96);
-      console.log(JSON.stringify({ small, large }));
-      await sql.close();
-    `,
-      { ASAN_OPTIONS: "quarantine_size_mb=4:detect_leaks=0" },
-    );
-    expect(result.stderr).toBe("");
-    expect(result.exitCode).toBe(0);
-    const growth = JSON.parse(result.stdout);
-    // Steady state drifts by a few MiB between passes; a leak adds ~24 MiB.
-    expect(growth).toEqual({ small: expect.any(Number), large: expect.any(Number) });
-    expect(growth.small, JSON.stringify(growth)).toBeLessThan(6);
-    expect(growth.large, JSON.stringify(growth)).toBeLessThan(6);
-  }, 30_000);
 });
 
 if (isDockerEnabled()) {
-  describeWithContainer("postgres", { image: "postgres_plain" }, container => {
+  // One postgres service for every test, so each test has its own channel
+  // names: a NOTIFY reaches every session that listens on that channel.
+  describeWithContainer("postgres", { image: "postgres_plain", concurrent: true }, container => {
     const connect = () =>
       new SQL(`postgres://bun_sql_test@${container.host}:${container.port}/bun_sql_test`, { max: 2 });
     // The listening backend is the session whose last statement is our LISTEN.
@@ -1013,12 +1104,14 @@ if (isDockerEnabled()) {
         if (payload === "barrier") barrier.open();
       });
 
-      await expect(
-        sql.begin(async tx => {
-          await tx.notify("e2e_tx", "rolled back");
-          throw new Error("abort");
-        }),
-      ).rejects.toThrow("abort");
+      expect(
+        await rejection(
+          sql.begin(async tx => {
+            await tx.notify("e2e_tx", "rolled back");
+            throw new Error("abort");
+          }),
+        ),
+      ).toMatchObject({ message: "abort" });
       await sql.begin(tx => tx.notify("e2e_tx", "committed"));
       await sql.notify("e2e_tx", "barrier");
       await barrier;
@@ -1031,28 +1124,33 @@ if (isDockerEnabled()) {
     // so a later delivery proves an earlier notification was or was not
     // delivered.
     describe("ported from postgres.js", () => {
-      test("listen and notify", async () => {
+      test.each(["pgjs_hello", "withUpperChar"])("listen and notify: %s", async channel => {
         await container.ready;
         await using sql = connect();
         const result = Promise.withResolvers<string>();
-        await sql.listen("pgjs_hello", result.resolve);
-        await sql.notify("pgjs_hello", "works");
+        const subscription = await sql.listen(channel, result.resolve);
+        expect(subscription.channel).toBe(channel);
+        await sql.notify(channel, "works");
         expect(await result.promise).toBe("works");
       });
 
       test("double listen", async () => {
         await container.ready;
         await using sql = connect();
-        let count = 0;
-        for (let i = 0; i < 2; i++) {
-          const received = Promise.withResolvers<string>();
-          await sql.listen("pgjs_hello", received.resolve);
-          await sql.notify("pgjs_hello", "world");
-          await received.promise;
-          count++;
+        const got: string[] = [];
+        let count = () => {};
+        for (const n of [1, 2]) {
+          const delivered = gate();
+          count = delivered.after(n); // every listener so far receives it
+          await sql.listen("pgjs_double", payload => {
+            got.push(`${n}:${payload}`);
+            count();
+          });
+          await sql.notify("pgjs_double", "world");
+          await delivered;
         }
         await sql.listen("pgjs_weee", () => {});
-        expect(count).toBe(2);
+        expect(got).toEqual(["1:world", "1:world", "2:world"]);
       });
 
       test("multiple listeners work after a reconnect", async () => {
@@ -1084,7 +1182,7 @@ if (isDockerEnabled()) {
         count = b.after(2);
         await sql.notify("pgjs_reconnect_multi", "b");
         await b;
-        expect(xs.join("")).toBe("1a2a1b2b");
+        expect(xs).toEqual(["1a", "2a", "1b", "2b"]);
       });
 
       test("listen and notify with weird name", async () => {
@@ -1102,35 +1200,31 @@ if (isDockerEnabled()) {
         await subscription.unlisten();
 
         const barrier = gate();
-        await sql.listen("pgjs_barrier", barrier.open);
+        await sql.listen("pgjs_weird_barrier", barrier.open);
         await sql.notify(channel, "after unlisten");
-        await sql.notify("pgjs_barrier", "");
+        await sql.notify("pgjs_weird_barrier", "");
         await barrier;
         expect(got).toEqual(["works"]);
       });
 
-      test("listen and notify with upper case", async () => {
-        await container.ready;
-        await using sql = connect();
-        const result = Promise.withResolvers<string>();
-        await sql.listen("withUpperChar", result.resolve);
-        await sql.notify("withUpperChar", "works");
-        expect(await result.promise).toBe("works");
-      });
-
       test("listen reconnects", async () => {
         await container.ready;
         await using sql = connect();
+        const events: string[] = [];
         const a = gate();
         const b = gate();
-        const resolvers: Record<string, () => void> = { a: a.open, b: b.open };
-        let connects = 0;
+        const arrivals: Record<string, () => void> = { a: a.open, b: b.open };
         const reconnected = gate();
+        const onlisten = reconnected.after(2);
         await sql.listen(
           "pgjs_reconnect",
-          x => resolvers[x]?.(),
+          payload => {
+            events.push("got " + payload);
+            arrivals[payload]?.();
+          },
           () => {
-            if (++connects === 2) reconnected.open();
+            events.push("onlisten");
+            onlisten();
           },
         );
         await sql.notify("pgjs_reconnect", "a");
@@ -1139,7 +1233,7 @@ if (isDockerEnabled()) {
         await reconnected;
         await sql.notify("pgjs_reconnect", "b");
         await b;
-        expect(connects).toBe(2);
+        expect(events).toEqual(["onlisten", "got a", "onlisten", "got b"]);
       });
 
       test("listen result reports correct connection state after reconnection", async () => {
@@ -1165,20 +1259,20 @@ if (isDockerEnabled()) {
         await using sql = connect();
         const xs: string[] = [];
         const a = gate();
-        const subscription = await sql.listen("pgjs_test", x => {
+        const subscription = await sql.listen("pgjs_unlisten", x => {
           xs.push(x);
           a.open();
         });
-        await sql.notify("pgjs_test", "a");
+        await sql.notify("pgjs_unlisten", "a");
         await a;
         await subscription.unlisten();
 
         const barrier = gate();
-        await sql.listen("pgjs_barrier", barrier.open);
-        await sql.notify("pgjs_test", "b");
-        await sql.notify("pgjs_barrier", "");
+        await sql.listen("pgjs_unlisten_barrier", barrier.open);
+        await sql.notify("pgjs_unlisten", "b");
+        await sql.notify("pgjs_unlisten_barrier", "");
         await barrier;
-        expect(xs.join("")).toBe("a");
+        expect(xs).toEqual(["a"]);
       });
 
       test("listen after unlisten", async () => {
@@ -1193,18 +1287,18 @@ if (isDockerEnabled()) {
 
         const a = gate();
         received = a.open;
-        const subscription = await sql.listen("pgjs_test", listener);
-        await sql.notify("pgjs_test", "a");
+        const subscription = await sql.listen("pgjs_relisten", listener);
+        await sql.notify("pgjs_relisten", "a");
         await a;
         await subscription.unlisten();
-        await sql.notify("pgjs_test", "b");
+        await sql.notify("pgjs_relisten", "b");
 
         const c = gate();
         received = c.open;
-        await sql.listen("pgjs_test", listener);
-        await sql.notify("pgjs_test", "c");
+        await sql.listen("pgjs_relisten", listener);
+        await sql.notify("pgjs_relisten", "c");
         await c;
-        expect(xs.join("")).toBe("ac");
+        expect(xs).toEqual(["a", "c"]);
       });
 
       test("multiple listeners and unlisten one", async () => {
@@ -1212,26 +1306,26 @@ if (isDockerEnabled()) {
         await using sql = connect();
         const xs: string[] = [];
         let count = () => {};
-        await sql.listen("pgjs_test", x => {
+        await sql.listen("pgjs_multi", x => {
           xs.push("1" + x);
           count();
         });
-        const s2 = await sql.listen("pgjs_test", x => {
+        const s2 = await sql.listen("pgjs_multi", x => {
           xs.push("2" + x);
           count();
         });
 
         const a = gate();
         count = a.after(2);
-        await sql.notify("pgjs_test", "a");
+        await sql.notify("pgjs_multi", "a");
         await a;
         await s2.unlisten();
 
         const b = gate();
         count = b.open;
-        await sql.notify("pgjs_test", "b");
+        await sql.notify("pgjs_multi", "b");
         await b;
-        expect(xs.join("")).toBe("1a2a1b");
+        expect(xs).toEqual(["1a", "2a", "1b"]);
       });
     });
   });