From 0c2646f965e6d5797e0dda003173d90c7870a7fc Mon Sep 17 00:00:00 2001
From: GiHoon1123 <rlaejrqo465@naver.com>
Date: Thu, 23 Jul 2026 16:25:39 +0900
Subject: [PATCH] fix: settle pending COPY writable/duplex final on server
 error

Failed COPY FROM STDIN never invoked the stashed Writable/Duplex
final() callback when the server rejected the data with ErrorResponse
instead of CommandComplete. This left the writable hanging forever,
which meant stream.pipeline() (and any await on stream finish) never
settled. Inside sql.begin(), the transaction callback never returned,
so ROLLBACK was never sent and the connection leaked permanently as
'idle in transaction (aborted)'.

errored() already cleaned up stream/query/initial on this path but
never invoked the pending final callback. Calling final(err) settles
the stream with the real PostgresError, so pipeline() rejects,
sql.begin() can roll back, and the connection returns to the pool.

Fixes #1173
---
 src/connection.js |  1 +
 tests/index.js    | 31 +++++++++++++++++++++++++++++++
 2 files changed, 32 insertions(+)

diff --git a/src/connection.js b/src/connection.js
index 1b1cccde..230973c5 100644
--- a/src/connection.js
+++ b/src/connection.js
@@ -389,6 +389,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
 
   function errored(err) {
     stream && (stream.destroy(err), stream = null)
+    final && (final(err), final = null)
     query && queryError(query, err)
     initial && (queryError(initial, err), initial = null)
   }
diff --git a/tests/index.js b/tests/index.js
index 23e6c4d4..3f6798a9 100644
--- a/tests/index.js
+++ b/tests/index.js
@@ -2038,6 +2038,37 @@ t('Copy from abort', async() => {
   ]
 })
 
+t('Copy from settles instead of hanging when the server rejects the data', { timeout: 2 }, async() => {
+  await sql`create table test (x int, y int)`
+
+  let error
+  try {
+    await sql.begin(async sql => {
+      const writable = await sql`COPY test FROM STDIN`.writable()
+      await new Promise((resolve, reject) => {
+        writable.on('error', reject)
+        writable.on('finish', resolve)
+        // 3 columns in the row, 2 in the table -> server errors mid copy,
+        // after the writable's final() has already fired (.end() below)
+        writable.end('1\t2\t3\n')
+      })
+    })
+  } catch (err) {
+    error = err
+  }
+
+  const idleInTransaction = await sql`
+    select count(*)::int as count from pg_stat_activity
+    where state = 'idle in transaction (aborted)'
+  `
+
+  return [
+    true,
+    !!error && error.message.includes('extra data') && idleInTransaction[0].count === 0,
+    await sql`drop table test`
+  ]
+})
+
 t('multiple queries before connect', async() => {
   const sql = postgres({ ...options, max: 2 })
   const xs = await Promise.all([