fix() manually close source stream (#5466)

lib/execution/runner.js

@@ -72,6 +72,24 @@ class Runner {
       this.client.releaseConnection(this.connection);
     });
 
+    // If the stream is manually destroyed, the close event is not
+    // propagated to the top of the pipe chain. We need to manually verify
+    // that the source stream is closed and if not, manually destroy it.
+    stream.on('pipe', (sourceStream) => {
+      const cleanSourceStream = () => {
+        if (!sourceStream.closed) {
+          sourceStream.destroy();
+        }
+      };
+
+      // Stream already closed, cleanup immediately
+      if (stream.closed) {
+        cleanSourceStream();
+      } else {
+        stream.on('close', cleanSourceStream);
+      }
+    });
+
     const connectionAcquirePromise = this.ensureConnection(
       ensureConnectionStreamCallback,
       {

test/integration2/query/misc/additional.spec.js

@@ -128,7 +128,7 @@ describe('Additional', function () {
           });
         });
 
-        it('should close the connection when a stream query iteration errors', async function () {
+        it('should release the connection when a stream query iteration errors', async function () {
           const spy = sinon.spy(knex.client, 'releaseConnection');
 
           const stream = knex.raw('VALUES (1), (2), (3)').stream();
@@ -146,6 +146,26 @@ describe('Additional', function () {
           spy.restore();
         });
 
+        it('should close the db connection when prematurely closing a stream', async function () {
+          await knex('accounts').truncate();
+          await insertAccounts(knex, 'accounts');
+
+          // We limit to only one row at a time, to keep the cursor open
+          const stream = knex('accounts').stream({ highWaterMark: 1 });
+
+          // eslint-disable-next-line no-unused-vars
+          for await (const _ of stream) {
+            stream.destroy();
+            break;
+          }
+
+          await new Promise((res) => setTimeout(res, 50));
+
+          // Will timeout if the connection is in the pool, but not closed
+          this.timeout(1000);
+          await knex('accounts').limit(1);
+        });
+
         it('should process response done through a stream', async () => {
           await knex('accounts').truncate();
           await insertAccounts(knex, 'accounts');