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: nodejs#30831

lib/internal/streams/async_iterator.js

@@ -189,6 +189,7 @@ const createReadableStreamAsyncIterator = (stream) => {
         } else {
           iterator[kLastResolve] = resolve;
           iterator[kLastReject] = reject;
+          // TODO(rexagod): catch readline silent rejects
         }
       },
       writable: true,

lib/readline.js

@@ -43,7 +43,9 @@ const {
 const {
   ERR_INVALID_CALLBACK,
   ERR_INVALID_CURSOR_POS,
-  ERR_INVALID_OPT_VALUE
+  ERR_INVALID_OPT_VALUE,
+  ERR_MISSING_OPTION,
+  ERR_STREAM_PREMATURE_CLOSE
 } = require('internal/errors').codes;
 const { validateString } = require('internal/validators');
 const {
@@ -134,6 +136,8 @@ function Interface(input, output, completer, terminal) {
     }
     crlfDelay = input.crlfDelay;
     input = input.input;
+  } else {
+    throw new ERR_MISSING_OPTION('options.input');
   }
 
   if (completer !== undefined && typeof completer !== 'function') {
@@ -182,6 +186,12 @@ function Interface(input, output, completer, terminal) {
     this._ttyWrite = _ttyWriteDumb.bind(this);
   }
 
+  function onerror(err) {
+    const debug = require('internal/util/debuglog').debuglog('stream');
+    debug(err);
+    throw new ERR_STREAM_PREMATURE_CLOSE();
+  }
+
   function ondata(data) {
     self._normalWrite(data);
   }
@@ -219,9 +229,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);
     }
 
@@ -232,6 +245,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);
@@ -1082,6 +1096,7 @@ Interface.prototype[SymbolAsyncIterator] = function() {
     });
     const lineListener = (input) => {
       if (!readable.push(input)) {
+        // TODO(rexagod): drain to resume flow
         this.pause();
       }
     };

test/parallel/test-readline-interface.js

@@ -147,7 +147,9 @@ function isWarned(emitter) {
   // Sending a single character with no newline
   {
     const fi = new FakeInput();
-    const rli = new readline.Interface(fi, {});
+    const rli = new readline.Interface(
+      { input: fi, output: fi, terminal: terminal }
+    );
     let called = false;
     rli.on('line', function(line) {
       called = true;

test/parallel/test-tty-stdin-pipe.js

@@ -28,7 +28,9 @@ require('../common');
 const assert = require('assert');
 const readline = require('readline');
 
-const rl = readline.createInterface(process.stdin, process.stdout);
+const rl = readline.createInterface(
+  { input: process.stdin, output: process.stdout }
+);
 rl.resume();
 
 let hasPaused = false;