fix(db): reserve transactions before checking pipeline capacity (#17831)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Interactive requests share a PostgreSQL connection pool with
background work.
> - Prepared queries can queue behind a slow query on a busy connection.
> - Disabling pipelining lets queued work use the next free connection.
> - The driver skips its transaction reservation callback when the
pipeline limit is zero.
> - This pull request runs that callback before the capacity check.
> - Operators can test a pool without pipelining while transactions
retain their connection ownership.

## Linked Issues or Issue Description

**What happened?**

With `PGMAX_PIPELINE=0`, postgres.js 3.4.9 skips `onexecute` in its
short-circuit capacity check. A transaction then lacks its reserved
connection. Real PostgreSQL reproduces `UNSAFE_TRANSACTION` through the
app client and an undefined connection error through the raw driver.
This prevents safe use of the existing setting to reduce query queue
delays.

**Expected behavior**

A transaction must reserve its connection regardless of pipeline
capacity. Commit, rollback, savepoints, and disconnect recovery must
preserve their existing contracts.

**Steps to reproduce**

1. Use the pre-change driver patch from
`e61ee4b005f8e13f64af15f69cb62305dd869f82`.
2. Start disposable PostgreSQL and run the `no-pipeline` scenario in
`packages/db/src/__fixtures__/postgres-connection-recovery.mjs` with
`TEST_DATABASE_URL`, `DRIVER_FORMAT=esm`, and
`RECOVERY_CASE=no-pipeline`.
3. Observe the transaction failure. Apply this patch and repeat with
both `esm` and `cjs`.

Refs #13643, which introduced the existing connection recovery patch.
Searched related open and closed driver PRs. This change extends that
patch and preserves its disconnect safeguards.

## What Changed

- Move the transaction reservation callback before the pipeline capacity
check in both driver distributions.
- Add real PostgreSQL regressions for commits, concurrent transaction
reads, isolation, rollback, savepoints, reserved connections, and
progress on a freed pool connection.
- Document the existing `PGMAX_PIPELINE=0` setting and its driver
prerequisite. Keep the default pipeline limit unchanged.

## Verification

- `pnpm -r typecheck`: passed.
- `pnpm build`: passed.
- Real PostgreSQL 18.6 in an owned, disposable Docker container: the old
patch fails the new transaction scenario in both ESM and CommonJS. The
corrected patch passes all 18 scenarios, including all 12 existing
disconnect recovery cases. The container was removed afterward.
- Local service benchmark: actual `createDb` with prepared statements,
two warmed connections, two controlled advisory lock waits, and the
complete three-query `getDefaultCompanyGoal` path. Two warmups and five
alternating measured samples per setting. Pipeline 100: median 413.44
ms, range 413.33–416.10 ms. Pipeline 0 with the patch: median 12.22 ms,
range 11.86–12.57 ms. Complete SQL, bindings, and results match. Five of
five reads finish while the unrelated slow query remains blocked. These
are synthetic local queue measurements, not hosted latency claims.
- App client compatibility on real PostgreSQL: commit, concurrent
transaction reads, rollback, nested savepoint rollback, dedicated
connections, and liveness connections passed.
- `pnpm test:run`: attempted, then stopped after repeated local embedded
PostgreSQL startup failures caused by the host semaphore limit. This is
an incomplete local full-suite run, not a pass. The focused driver
checks above used disposable Docker PostgreSQL instead. The initial
sandbox attempt also stopped before test execution because tsx could not
bind its IPC socket. Final CI results will be recorded here.
- Current head `72039f71f63530300bf241475d42e857297dfb90`: Greptile and
Paperclip review are both 5/5. All three earlier findings are resolved.
[CI run
40431567516](https://github.com/paperclipai/paperclip/actions/runs/40431567516)
ran the complete validation matrix. The database lane passed all 265
tests, including all 18 real PostgreSQL driver regressions. One cleanup
timer case timed out in server shard 6/12; its isolated real-PostgreSQL
reproduction passed once and failed once, while five runs with the
pre-change driver passed. That evidence is retained and does not
establish the timer failure's cause. A single CI retry of the failed
shard passed. The final aggregate `ci / verify` gate is waiting for a
runner. Its first queued request did not start for more than 14 minutes.
A guarded queue reset confirmed that it was the only unfinished job and
that every test job had passed; attempt 3 restarts only that gate. No
test result was replaced or bypassed. The required gate must pass before
merge.

## Risks

- This changes a shared driver callback order. Both module formats and
existing disconnect recovery are covered by real database tests.
- This PR does not change pool size, enable the setting by default,
retry SQL, or change schemas. Disabling pipelining may reduce batch
throughput. Operators must compare the workload before retaining that
setting.
- The local queue benchmark proves the mechanism. Hosted request
improvement still needs acceptance after the patched image is deployed.

## Model Used

OpenAI GPT-6 in Codex, with code execution and browser developer tools.
The exact serving model identifier and context window are not exposed in
this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -580,6 +580,16 @@ PostgreSQL backends and checks rejection, pool recovery, and transaction isolati
 Remove the patch when an upstream release passes these tests. Installs of the
 unmodified `postgres` package outside this workspace do not include the patch.
 
+The patch also preserves transaction connection reservation when pipelining is
+disabled with `PGMAX_PIPELINE=0`. Without this fix, the driver's pipeline capacity
+check skips the transaction reservation callback and `BEGIN` fails with
+`UNSAFE_TRANSACTION`. Disabling pipelining makes a queued query wait for the next
+available pool connection instead of sending it behind unrelated work on a busy
+connection. This can help interactive requests that share a prepared-query pool
+with slow background work. It does not make the SQL itself faster. The default
+pipeline limit remains unchanged. Test transaction behavior before using this
+setting with a driver build that does not include the patch.
+
 Trusted-header actor synchronization retries transient connection failures,
 including `CONNECT_TIMEOUT`, at most twice. This retry applies only to the
 idempotent actor synchronization operations, not arbitrary transactions. A

packages/db/src/__fixtures__/postgres-connection-recovery.mjs

@@ -8,8 +8,9 @@ const postgres = process.env.DRIVER_FORMAT === "cjs"
 const mode = process.env.RECOVERY_CASE;
 const closed = Promise.withResolvers();
 const sql = postgres(process.env.TEST_DATABASE_URL, {
-  max: 1,
-  max_pipeline: 1,
+  max: mode === "no-pipeline" || mode === "no-pipeline-fairness" ? 2 : 1,
+  max_pipeline: mode?.startsWith("no-pipeline") ? 0 : 1,
+  connection: { application_name: `driver-recovery-${mode}` },
   backoff: 0,
   onnotice() {},
   onclose: () => closed.resolve(),
@@ -28,7 +29,76 @@ try {
   await control.unsafe("drop table if exists connection_recovery_probe");
   await control.unsafe("create table connection_recovery_probe (id integer)");
 
-  if (mode === "transaction" || mode === "transaction-closed" || mode === "savepoint") {
+  if (mode === "no-pipeline-fairness") {
+    const query = "select pg_advisory_xact_lock($1::bigint)";
+    const lookup = "select $1::int as value";
+    const one = await sql.reserve();
+    const two = await sql.reserve();
+    for (const connection of [one, two]) {
+      await connection.unsafe(query, [900001]);
+      await connection.unsafe(lookup, [42]);
+    }
+    one.release();
+    two.release();
+    let first;
+    let second;
+    let fast;
+    try {
+      await control.unsafe("select pg_advisory_lock(900001), pg_advisory_lock(900002)");
+      const waitForBlocked = async (count) => {
+        const deadline = performance.now() + 5000;
+        let observed = 0;
+        while (performance.now() < deadline) {
+          const [row] = await control.unsafe("select count(*)::int as count from pg_stat_activity where application_name=$1 and wait_event='advisory'", [`driver-recovery-${mode}`]);
+          observed = row.count;
+          if (row.count === count) return;
+          await setImmediate();
+        }
+        throw new Error(`Expected ${count} advisory lock waits; observed ${observed}`);
+      };
+      let firstFinished = false;
+      first = sql.unsafe(query, [900001]).then(() => { firstFinished = true; });
+      await waitForBlocked(1);
+      second = sql.unsafe(query, [900002]).then(() => {});
+      await waitForBlocked(2);
+      fast = sql.unsafe(lookup, [42]).then((rows) => rows);
+      await control.unsafe("select pg_advisory_unlock(900002)");
+      assert.equal((await fast)[0].value, 42);
+      assert.equal(firstFinished, false, "lookup must progress after one pool connection is released while the other stays blocked");
+    } finally {
+      await control.unsafe("select pg_advisory_unlock_all()");
+      await Promise.allSettled([first, second, fast].filter(Boolean));
+    }
+  } else if (mode === "no-pipeline" || mode === "no-pipeline-single") {
+    await sql.begin(async (tx) => {
+      await tx.unsafe("insert into connection_recovery_probe values (1)");
+      const results = await Promise.all([
+        tx.unsafe("select id from connection_recovery_probe"),
+        tx.unsafe("select count(*)::int as count from connection_recovery_probe"),
+      ]);
+      assert.deepEqual(results[0].map((row) => row.id), [1]);
+      assert.equal(results[1][0].count, 1);
+      assert.equal((await control.unsafe("select count(*)::int as count from connection_recovery_probe"))[0].count, 0);
+    });
+    await assert.rejects(sql.begin(async (tx) => {
+      await tx.unsafe("insert into connection_recovery_probe values (2)");
+      throw new Error("controlled transaction rollback");
+    }), /controlled transaction rollback/);
+    await sql.begin(async (tx) => {
+      await tx.unsafe("insert into connection_recovery_probe values (3)");
+      await assert.rejects(tx.savepoint(async (nested) => {
+        await nested.unsafe("insert into connection_recovery_probe values (4)");
+        throw new Error("controlled savepoint rollback");
+      }), /controlled savepoint rollback/);
+      assert.deepEqual((await tx.unsafe("select id from connection_recovery_probe order by id")).map((row) => row.id), [1, 3]);
+    });
+    const reserved = await sql.reserve();
+    try {
+      assert.deepEqual((await reserved.unsafe("select id from connection_recovery_probe order by id")).map((row) => row.id), [1, 3]);
+    } finally {
+      reserved.release();
+    }
+  } else if (mode === "transaction" || mode === "transaction-closed" || mode === "savepoint") {
     const started = Promise.withResolvers();
     const resume = Promise.withResolvers();
     const callbackDone = Promise.withResolvers();

packages/db/src/postgres-connection-recovery.test.ts

@@ -21,6 +21,22 @@ describe.skipIf(!support.supported)("postgres connection recovery", () => {
   afterAll(async () => { await database?.cleanup(); });
 
   for (const format of ["esm", "cjs"]) {
+    for (const scenario of ["no-pipeline-single", "no-pipeline", "no-pipeline-fairness"]) {
+      it(`${format}: ${scenario} preserves connection ownership and progress`, async () => {
+        const { stdout, stderr } = await run(process.execPath, [fixture], {
+          env: {
+            ...process.env,
+            TEST_DATABASE_URL: database.connectionString,
+            DRIVER_FORMAT: format,
+            RECOVERY_CASE: scenario,
+          },
+          timeout: 15_000,
+        });
+        expect(stdout.trim()).toBe("recovered");
+        expect(stderr).toBe("");
+      }, 20_000);
+    }
+
     for (const scenario of ["transaction", "transaction-closed", "savepoint", "reserve", "transaction-queue", "reserve-queue"]) {
       it(`${format}: rejects disconnected ${scenario} work and recovers the pool`, async () => {
         // A driver timer used to crash the whole process. A child also detects

patches/postgres@3.4.9.patch

@@ -1,7 +1,17 @@
 diff --git a/cjs/src/connection.js b/cjs/src/connection.js
-index 07f6716702ac888215c86ad6e0071aaf55ae519c..dc24963ce2dd0a219f366f80ba9847548ba23a52 100644
+index 07f6716702ac888215c86ad6e0071aaf55ae519c..85700681ccc7bf98f1fe987978db47e0097227f3 100644
 --- a/cjs/src/connection.js
 +++ b/cjs/src/connection.js
+@@ -173,8 +173,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
+       return write(toBuffer(q))
+         && !q.describeFirst
+         && !q.cursorFn
+-        && sent.length < max_pipeline
+         && (!q.options.onexecute || q.options.onexecute(connection))
++        && sent.length < max_pipeline
+     } catch (error) {
+       sent.length === 0 && write(Sync)
+       errored(error)
 @@ -438,6 +438,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
      remaining = 0
      incomings = null
@@ -84,9 +94,19 @@ index f09c61c7e998f411560b95c9a8e1ed6a97972c11..04dded5c7e2cdc34ac01b29a87cc9324
            ? queries.push(q)
            : c.execute(q) || move(c, full)
 diff --git a/src/connection.js b/src/connection.js
-index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..c95c326a2747c680c2c22b04b1c5f2e27a4e8c90 100644
+index 1b1cccde43b4570d5d071d6ffaab2669b3c2065a..81541f9cea75925256d7b02b2079c60802b2f92f 100644
 --- a/src/connection.js
 +++ b/src/connection.js
+@@ -173,8 +173,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
+       return write(toBuffer(q))
+         && !q.describeFirst
+         && !q.cursorFn
+-        && sent.length < max_pipeline
+         && (!q.options.onexecute || q.options.onexecute(connection))
++        && sent.length < max_pipeline
+     } catch (error) {
+       sent.length === 0 && write(Sync)
+       errored(error)
 @@ -438,6 +438,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
      remaining = 0
      incomings = null