fix(stream): resume connection when stream errors or is destroyed (#3775)

* fix(stream): resume connection when stream errors or is destroyed

* fix: remove listeners when stream is destroyed

* ci: remove unreachable logic

* chore: remove stream listener when it is destroyed

* chore: ensure the error is emitted

* chore: TOC

* refactor: use native `close` and auto detroy

* refactor: follow ESLint rules

Author: wellwelwel

lib/commands/query.js

@@ -271,33 +271,67 @@ class Query extends Command {
   }
 
   stream(options) {
-    options = options || {};
+    options = options || Object.create(null);
     options.objectMode = true;
-    const stream = new Readable(options);
-    stream._read = () => {
-      this._connection && this._connection.resume();
-    };
-    this.on('result', (row, resultSetIndex) => {
-      if (!stream.push(row)) {
-        this._connection.pause();
-      }
-      stream.emit('result', row, resultSetIndex); // replicate old emitter
-    });
-    this.on('error', (err) => {
-      stream.emit('error', err); // Pass on any errors
+
+    const stream = new Readable({
+      ...options,
+      emitClose: true,
+      autoDestroy: true,
+      read: () => {
+        this._connection && this._connection.resume();
+      },
     });
-    this.on('end', () => {
-      stream.push(null); // pushing null, indicating EOF
+
+    // Prevent a breaking change for users that rely on `end` event
+    stream.once('close', () => {
+      if (!stream.readableEnded) {
+        stream.emit('end');
+      }
     });
-    this.on('fields', (fields) => {
+
+    const onResult = (row, index) => {
+      if (stream.destroyed) return;
+
+      if (!stream.push(row)) {
+        this._connection && this._connection.pause();
+      }
+
+      stream.emit('result', row, index); // replicate old emitter
+    };
+
+    const onFields = (fields) => {
+      if (stream.destroyed) return;
+
       stream.emit('fields', fields); // replicate old emitter
-    });
-    stream.on('end', () => {
-      stream.emit('close');
-    });
-    stream.on('error', () => {
-      this._connection && this._connection.destroy();
-    });
+    };
+
+    const onEnd = () => {
+      if (stream.destroyed) return;
+
+      stream.push(null); // pushing null, indicating EOF
+    };
+
+    const onError = (err) => {
+      stream.destroy(err);
+    };
+
+    stream._destroy = (err, cb) => {
+      this._connection && this._connection.resume();
+
+      this.removeListener('result', onResult);
+      this.removeListener('fields', onFields);
+      this.removeListener('end', onEnd);
+      this.removeListener('error', onError);
+
+      cb(err); // Pass on any errors
+    };
+
+    this.on('result', onResult);
+    this.on('fields', onFields);
+    this.on('end', onEnd);
+    this.on('error', onError);
+
     return stream;
   }
 

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

@@ -1,45 +1,92 @@
 'use strict';
 
-const process = require('node:process');
-const { test, skip } = require('poku');
+const { assert, describe, test } = 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 () => {
+describe(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;
+  await test('Ensure stream ends in case of error', async () => {
+    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;
+          }
+        }
+      );
     }
-  );
 
-  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; // forces return () -> destroy()
+  });
+
+  await test('end: Ensure stream emits error then close on server-side query error', async () => {
+    let uncaughtExceptionError;
+
+    const stream = connection
+      .query('SELECT invalid_column FROM invalid_table')
+      .stream();
+
+    stream.on('error', (error) => {
+      uncaughtExceptionError = error;
+    });
+
+    await new Promise((resolve) => stream.on('end', resolve));
+
+    assert(
+      uncaughtExceptionError instanceof Error,
+      'Expected an uncaught exception error'
+    );
+
+    assert.equal(
+      uncaughtExceptionError.message,
+      "Table 'test.invalid_table' doesn't exist"
+    );
+  });
+
+  await test('close: Ensure stream emits error then close on server-side query error', async () => {
+    let uncaughtExceptionError;
+
+    const stream = connection
+      .query('SELECT invalid_column FROM invalid_table')
+      .stream();
+
+    stream.on('error', (error) => {
+      uncaughtExceptionError = error;
     });
-  }
 
-  const rows = connection.query('SELECT * FROM items').stream();
+    await new Promise((resolve) => stream.on('close', resolve));
 
-  // eslint-disable-next-line no-unused-vars
-  for await (const _ of rows) break;
+    assert(
+      uncaughtExceptionError instanceof Error,
+      'Expected an uncaught exception error'
+    );
 
-  setTimeout(() => {
-    throw new Error('Connection remains open after stream error');
-  }, 1000).unref();
+    assert.equal(
+      uncaughtExceptionError.message,
+      "Table 'test.invalid_table' doesn't exist"
+    );
+  });
 
   connection.end();
 });