From 4336b9ee0210d51b5f9a19eb195e995a1bd2a50e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 26 Oct 2026 22:00:01 -0500 Subject: [PATCH] fix(db): reserve transactions before checking pipeline capacity (#17831) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- doc/DATABASE.md | 10 +++ .../postgres-connection-recovery.mjs | 76 ++++++++++++++++++- .../src/postgres-connection-recovery.test.ts | 16 ++++ patches/postgres@3.4.9.patch | 24 +++++- 4 files changed, 121 insertions(+), 5 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 2032268da2d..572b3564bf9 100644 --- a/doc/DATABASE.md +++ b/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 diff --git a/packages/db/src/__fixtures__/postgres-connection-recovery.mjs b/packages/db/src/__fixtures__/postgres-connection-recovery.mjs index 64ee9db814d..15ac09f8805 100644 --- a/packages/db/src/__fixtures__/postgres-connection-recovery.mjs +++ b/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(); diff --git a/packages/db/src/postgres-connection-recovery.test.ts b/packages/db/src/postgres-connection-recovery.test.ts index ddb6bb325b8..4918fe3fc41 100644 --- a/packages/db/src/postgres-connection-recovery.test.ts +++ b/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 diff --git a/patches/postgres@3.4.9.patch b/patches/postgres@3.4.9.patch index ab27a64ba33..c45fb197a7c 100644 --- a/patches/postgres@3.4.9.patch +++ b/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