lib: add error handling for input stream

This adds support for error handling in readline.createInterface()
for cases where the input object is not supplied, the input stream
is invalid, or the underlying buffer emits an error.

Now, the 'error' emissions by the readline module are thrown but
in order to log those in the specific case of await for loops,
we still need to fix silent rejections (TODO added there) inside
async iterators for the thenables to work.

Fixes: #30831

PR-URL: #31603
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

Author: rexagod

lib/readline.js

@@ -190,6 +190,10 @@ function Interface(input, output, completer, terminal) {
     this._ttyWrite = _ttyWriteDumb.bind(this);
   }
 
+  function onerror(err) {
+    self.emit('error', err);
+  }
+
   function ondata(data) {
     self._normalWrite(data);
   }
@@ -227,9 +231,12 @@ function Interface(input, output, completer, terminal) {
 
   this[kLineObjectStream] = undefined;
 
+  input.on('error', onerror);
+
   if (!this.terminal) {
     function onSelfCloseWithoutTerminal() {
       input.removeListener('data', ondata);
+      input.removeListener('error', onerror);
       input.removeListener('end', onend);
     }
 
@@ -240,6 +247,7 @@ function Interface(input, output, completer, terminal) {
   } else {
     function onSelfCloseWithTerminal() {
       input.removeListener('keypress', onkeypress);
+      input.removeListener('error', onerror);
       input.removeListener('end', ontermend);
       if (output !== null && output !== undefined) {
         output.removeListener('resize', onresize);
@@ -1098,12 +1106,17 @@ Interface.prototype[SymbolAsyncIterator] = function() {
     });
     const lineListener = (input) => {
       if (!readable.push(input)) {
+        // TODO(rexagod): drain to resume flow
         this.pause();
       }
     };
     const closeListener = () => {
       readable.push(null);
     };
+    const errorListener = (err) => {
+      readable.destroy(err);
+    };
+    this.on('error', errorListener);
     this.on('line', lineListener);
     this.on('close', closeListener);
     this[kLineObjectStream] = readable;

test/parallel/test-readline-input-onerror.js

@@ -0,0 +1,40 @@
+'use strict';
+const common = require('../common');
+const fs = require('fs');
+const readline = require('readline');
+const path = require('path');
+
+async function processLineByLine_SymbolAsyncError(filename) {
+  const fileStream = fs.createReadStream(filename);
+  const rl = readline.createInterface({
+    input: fileStream,
+    crlfDelay: Infinity
+  });
+  // eslint-disable-next-line no-unused-vars
+  for await (const line of rl) {
+    /* check SymbolAsyncIterator `errorListener` */
+  }
+}
+
+const f = path.join(__dirname, 'file.txt');
+
+// catch-able SymbolAsyncIterator `errorListener` error
+processLineByLine_SymbolAsyncError(f).catch(common.expectsError({
+  code: 'ENOENT',
+  message: `ENOENT: no such file or directory, open '${f}'`
+}));
+
+async function processLineByLine_InterfaceErrorEvent(filename) {
+  const fileStream = fs.createReadStream(filename);
+  const rl = readline.createInterface({
+    input: fileStream,
+    crlfDelay: Infinity
+  });
+  rl.on('error', common.expectsError({
+    code: 'ENOENT',
+    message: `ENOENT: no such file or directory, open '${f}'`
+  }));
+}
+
+// check Interface 'error' event
+processLineByLine_InterfaceErrorEvent(f);