diff --git a/src/sql_jsc/postgres/PostgresSQLConnection.rs b/src/sql_jsc/postgres/PostgresSQLConnection.rs index f42d7c179971..1522ce8998fb 100644 --- a/src/sql_jsc/postgres/PostgresSQLConnection.rs +++ b/src/sql_jsc/postgres/PostgresSQLConnection.rs @@ -1807,12 +1807,18 @@ impl PostgresSQLConnection { match item.status.get() { QueryStatus::Running | QueryStatus::Binding | QueryStatus::PartialResponse => { let flags = item.flags.get(); + if !flags.counted { + return; + } + item.update_flags(|f| f.counted = false); if flags.simple { - self.nonpipelinable_requests - .set(self.nonpipelinable_requests.get() - 1); + let n = self.nonpipelinable_requests.get(); + debug_assert!(n > 0, "nonpipelinable_requests underflow"); + self.nonpipelinable_requests.set(n.saturating_sub(1)); } else if flags.pipelined { - self.pipelined_requests - .set(self.pipelined_requests.get() - 1); + let n = self.pipelined_requests.get(); + debug_assert!(n > 0, "pipelined_requests underflow"); + self.pipelined_requests.set(n.saturating_sub(1)); } } QueryStatus::Pending => { @@ -1938,6 +1944,7 @@ impl PostgresSQLConnection { } self.nonpipelinable_requests .set(self.nonpipelinable_requests.get() + 1); + req.update_flags(|f| f.counted = true); self.update_flags(|f| f.remove(ConnectionFlags::IS_READY_FOR_QUERY)); req.status.set(QueryStatus::Running); defer_cleanup!(self); @@ -2070,7 +2077,10 @@ impl PostgresSQLConnection { f.remove(ConnectionFlags::IS_READY_FOR_QUERY) }); req.status.set(QueryStatus::Binding); - req.update_flags(|f| f.pipelined = true); + req.update_flags(|f| { + f.pipelined = true; + f.counted = true; + }); self.pipelined_requests .set(self.pipelined_requests.get() + 1); @@ -2233,7 +2243,10 @@ impl PostgresSQLConnection { }); req.status.set(QueryStatus::Binding); statement.status = StatementStatus::Parsing; - req.update_flags(|f| f.pipelined = true); + req.update_flags(|f| { + f.pipelined = true; + f.counted = true; + }); self.pipelined_requests .set(self.pipelined_requests.get() + 1); self.flush_data_and_reset_timeout(); diff --git a/src/sql_jsc/postgres/PostgresSQLQuery.rs b/src/sql_jsc/postgres/PostgresSQLQuery.rs index 4b0509fa8ea1..9dffe40e12fd 100644 --- a/src/sql_jsc/postgres/PostgresSQLQuery.rs +++ b/src/sql_jsc/postgres/PostgresSQLQuery.rs @@ -92,6 +92,11 @@ pub struct Flags { pub bigint: bool, pub simple: bool, pub pipelined: bool, + /// Set when this request's dispatch incremented the connection's + /// `pipelined_requests` / `nonpipelinable_requests` counter; cleared when + /// `finish_request` consumes that contribution. Makes the decrement + /// idempotent across the three `finish_request` call sites. + pub counted: bool, pub result_mode: PostgresSQLQueryResultMode, } @@ -103,6 +108,7 @@ impl Default for Flags { bigint: false, simple: false, pipelined: false, + counted: false, result_mode: PostgresSQLQueryResultMode::Objects, } } @@ -542,6 +548,7 @@ impl PostgresSQLQuery { connection .nonpipelinable_requests .set(connection.nonpipelinable_requests.get() + 1); + this.update_flags(|f| f.counted = true); this.status.set(Status::Running); } else { this.status.set(Status::Pending); @@ -675,7 +682,10 @@ impl PostgresSQLQuery { connection.flags.set(f); } this.status.set(Status::Binding); - this.update_flags(|f| f.pipelined = true); + this.update_flags(|f| { + f.pipelined = true; + f.counted = true; + }); connection .pipelined_requests .set(connection.pipelined_requests.get() + 1); diff --git a/test/js/sql/postgres-finish-request-underflow-fixture.ts b/test/js/sql/postgres-finish-request-underflow-fixture.ts new file mode 100644 index 000000000000..3b997ae375ef --- /dev/null +++ b/test/js/sql/postgres-finish-request-underflow-fixture.ts @@ -0,0 +1,109 @@ +// Fixture for postgres-finish-request-underflow.test.ts. Runs in a subprocess +// so a debug_assert panic on the counter invariant is observable as a nonzero +// exit code instead of taking down the test runner. +// +// Drives a single connection through every finish_request call site: +// - the ReadyForQuery 'Z' arm, for a simple query that completes normally +// - the ErrorResponse 'E' arm, for a simple query the server rejects +// - the ErrorResponse arm followed by CommandComplete + ReadyForQuery for the +// same exchange (the sequence whose second CommandComplete used to flip the +// request back to PartialResponse and double-decrement the counter) +// +// After each exchange a fresh query must still dispatch: if the per-class +// request counter leaked high or wrapped past zero, advance() would refuse to +// write it and the subprocess would sit idle until the watchdog fires. +import { SQL } from "bun"; +import { + listeningServer, + pgAuthenticationOk, + pgCommandComplete, + pgErrorResponse, + pgReadFrontendMessages, + pgReadyForQuery, +} from "./wire-frames"; + +const watchdog = setTimeout(() => { + console.error("WATCHDOG: a later query never dispatched (request counter leaked or wrapped)"); + process.exit(1); +}, 15_000); + +type ConnState = { buf: Buffer; sawStartup: boolean; simpleCount: number }; + +const { port, server } = await listeningServer(socket => { + const state: ConnState = { buf: Buffer.alloc(0), sawStartup: false, simpleCount: 0 }; + socket.on("data", data => { + state.buf = Buffer.concat([state.buf, data]); + if (!state.sawStartup) { + if (state.buf.length < 4) return; + const len = state.buf.readInt32BE(0); + if (state.buf.length < len) return; + state.buf = state.buf.subarray(len); + state.sawStartup = true; + socket.write(Buffer.concat([pgAuthenticationOk(), pgReadyForQuery()])); + return; + } + state.buf = pgReadFrontendMessages(state.buf, (type, body) => { + if (type !== 0x51 /* 'Q' simple Query */) return; + const q = body.toString("utf8", 0, body.indexOf(0)); + state.simpleCount++; + if (q.includes("reject_once")) { + socket.write( + Buffer.concat([pgErrorResponse({ S: "ERROR", C: "XX000", M: "boom" }), pgReadyForQuery()]), + ); + } else if (q.includes("reject_then_late_result")) { + // ErrorResponse first, then a late CommandComplete + ReadyForQuery for + // the same exchange. The late CommandComplete must be discarded; the + // ReadyForQuery that follows must not decrement a second time. + socket.write( + Buffer.concat([ + pgErrorResponse({ S: "ERROR", C: "XX000", M: "boom" }), + pgCommandComplete("SELECT 0"), + pgReadyForQuery(), + ]), + ); + } else { + socket.write( + Buffer.concat([pgCommandComplete(`SELECT ${state.simpleCount}`), pgReadyForQuery()]), + ); + } + }); + }); + socket.on("error", () => {}); +}); + +const opts = { url: `postgres://u@127.0.0.1:${port}/db`, max: 1, idleTimeout: 5, connectionTimeout: 5 } as const; + +// 1. normal simple-query completion: finish_request via the 'Z' arm. +{ + const sql = new SQL(opts); + await sql.unsafe("select ok").simple(); + await sql.unsafe("select still_dispatches").simple(); + await sql.close({ timeout: 0 }); +} + +// 2. ErrorResponse: finish_request via the 'E' arm, then a follow-up query +// must still dispatch on the same connection. +{ + const sql = new SQL(opts); + const err: any = await sql.unsafe("select reject_once").simple().catch(e => e); + if (err?.code !== "ERR_POSTGRES_SERVER_ERROR") { + throw new Error(`expected ERR_POSTGRES_SERVER_ERROR, got ${err?.code ?? err}`); + } + await sql.unsafe("select still_dispatches").simple(); + await sql.close({ timeout: 0 }); +} + +// 3. ErrorResponse + late CommandComplete + ReadyForQuery for the same request. +{ + const sql = new SQL(opts); + const err: any = await sql.unsafe("select reject_then_late_result").simple().catch(e => e); + if (err?.code !== "ERR_POSTGRES_SERVER_ERROR") { + throw new Error(`expected ERR_POSTGRES_SERVER_ERROR, got ${err?.code ?? err}`); + } + await sql.unsafe("select still_dispatches").simple(); + await sql.close({ timeout: 0 }); +} + +clearTimeout(watchdog); +await new Promise(resolve => server.close(() => resolve())); +console.log("DONE"); diff --git a/test/js/sql/postgres-finish-request-underflow.test.ts b/test/js/sql/postgres-finish-request-underflow.test.ts new file mode 100644 index 000000000000..baae27461952 --- /dev/null +++ b/test/js/sql/postgres-finish-request-underflow.test.ts @@ -0,0 +1,50 @@ +// https://github.com/oven-sh/bun/issues/32004 +// +// PostgresSQLConnection::finish_request decremented the per-class in-flight +// counter (nonpipelinable_requests / pipelined_requests) from three call +// sites (ReadyForQuery, ErrorResponse, connection-close cleanup) with no +// per-request idempotence guard. Under connection-failure timing a request +// could be finished twice, driving the u32 past zero: a debug build panics +// with `attempt to subtract with overflow`; a release build silently wraps +// to u32::MAX, after which advance() treats the connection as permanently +// busy and queued queries never dispatch. +// +// The double-decrement reproduces under syscall fault injection on the +// Postgres socket but not from a scripted server alone (the known +// server-driven path was closed by the status==Fail skip in +// CommandComplete/DataRow). This test is a regression guard over the +// per-request `counted` bookkeeping: it drives one connection through every +// finish_request call site back-to-back and asserts a follow-up query still +// dispatches. A leaked-high or wrapped counter would wedge the follow-up +// query until the fixture's watchdog fires, and a violated counter invariant +// would trip the debug_assert in finish_request. +import { expect, test } from "bun:test"; +import { bunEnv, bunExe } from "harness"; +import path from "node:path"; + +test("postgres: per-class request counter is balanced across every finish_request call site", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), path.join(import.meta.dir, "postgres-finish-request-underflow-fixture.ts")], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect({ + stdout: stdout.trim(), + overflowPanic: /attempt to subtract with overflow|_requests underflow/.test(stderr), + watchdog: /WATCHDOG/.test(stderr), + exitCode, + signalCode: proc.signalCode, + // not asserted; included so a panic backtrace shows up in the diff + stderr, + }).toEqual({ + stdout: "DONE", + overflowPanic: false, + watchdog: false, + exitCode: 0, + signalCode: null, + stderr: expect.any(String), + }); +}, 30_000);