sql(postgres): make finish_request's per-class counter decrement idempotent

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();

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);

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);