fix: destroy connection when stream errors (#3769)

* Destroy connection when stream is destroyed

* Only destroy connection on stream error

* Add test case

* Shorten timeout

* Update test/integration/connection/test-stream-error-destroy-connection.test.cjs

Co-authored-by: Weslley Araújo <[email protected]>

---------

Co-authored-by: Weslley Araújo <[email protected]>

Author: fenichelar

lib/commands/query.js

@@ -295,6 +295,9 @@ class Query extends Command {
     stream.on('end', () => {
       stream.emit('close');
     });
+    stream.on('error', () => {
+      this._connection && this._connection.destroy();
+    });
     return stream;
   }
 

test/integration/connection/test-stream-error-destroy-connection.test.cjs

@@ -0,0 +1,45 @@
+'use strict';
+
+const process = require('node:process');
+const { test, skip } = require('poku');
+const common = require('../../common.test.cjs');
+
+if (process.env.MYSQL_USE_TLS === '1') skip('Skipping for SSL=1');
+
+test('Ensure stream ends in case of error', async () => {
+  const connection = common.createConnection();
+
+  connection.query(
+    [
+      'CREATE TEMPORARY TABLE `items` (',
+      '`id` int(11) NOT NULL AUTO_INCREMENT,',
+      '`text` varchar(255) DEFAULT NULL,',
+      'PRIMARY KEY (`id`)',
+      ') ENGINE=InnoDB DEFAULT CHARSET=utf8',
+    ].join('\n'),
+    (err) => {
+      if (err) {
+        throw err;
+      }
+    }
+  );
+
+  for (let i = 0; i < 100; i++) {
+    connection.execute('INSERT INTO items(text) VALUES(?)', ['test'], (err) => {
+      if (err) {
+        throw err;
+      }
+    });
+  }
+
+  const rows = connection.query('SELECT * FROM items').stream();
+
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of rows) break;
+
+  setTimeout(() => {
+    throw new Error('Connection remains open after stream error');
+  }, 1000).unref();
+
+  connection.end();
+});