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