child_process: emit IPC messages on next tick

This commit fixes a regression related to IPC 'message'
events. When messages are not emitted in the next tick,
a 'message' handler that throws can break the IPC read
loop.

Refs: #6909
Refs: #13459
Refs: #13648
PR-URL: #13856
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Santiago Gimeno <santiago.gimeno@gmail.com>

lib/internal/child_process.js

@@ -456,7 +456,6 @@ function setupChannel(target, channel) {
       }
       chunks[0] = jsonBuffer + chunks[0];
 
-      var nextTick = false;
       for (var i = 0; i < numCompleteChunks; i++) {
         var message = JSON.parse(chunks[i]);
 
@@ -465,12 +464,11 @@ function setupChannel(target, channel) {
         // that we deliver the handle with the right message however.
         if (isInternal(message)) {
           if (message.cmd === 'NODE_HANDLE')
-            handleMessage(message, recvHandle, true, false);
+            handleMessage(message, recvHandle, true);
           else
-            handleMessage(message, undefined, true, false);
+            handleMessage(message, undefined, true);
         } else {
-          handleMessage(message, undefined, false, nextTick);
-          nextTick = true;
+          handleMessage(message, undefined, false);
         }
       }
       jsonBuffer = incompleteChunk;
@@ -532,7 +530,7 @@ function setupChannel(target, channel) {
 
     // Convert handle object
     obj.got.call(this, message, handle, function(handle) {
-      handleMessage(message.msg, handle, isInternal(message.msg), false);
+      handleMessage(message.msg, handle, isInternal(message.msg));
     });
   });
 
@@ -742,15 +740,13 @@ function setupChannel(target, channel) {
     target.emit(event, message, handle);
   }
 
-  function handleMessage(message, handle, internal, nextTick) {
+  function handleMessage(message, handle, internal) {
     if (!target.channel)
       return;
 
     var eventName = (internal ? 'internalMessage' : 'message');
-    if (nextTick)
-      process.nextTick(emit, eventName, message, handle);
-    else
-      target.emit(eventName, message, handle);
+
+    process.nextTick(emit, eventName, message, handle);
   }
 
   channel.readStart();

test/parallel/test-child-process-ipc-next-tick.js

@@ -0,0 +1,39 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const cp = require('child_process');
+const NUM_MESSAGES = 10;
+const values = [];
+
+for (let i = 0; i < NUM_MESSAGES; ++i) {
+  values[i] = i;
+}
+
+if (process.argv[2] === 'child') {
+  const received = values.map(() => { return false; });
+
+  process.on('uncaughtException', common.mustCall((err) => {
+    received[err] = true;
+    const done = received.every((element) => { return element === true; });
+
+    if (done)
+      process.disconnect();
+  }, NUM_MESSAGES));
+
+  process.on('message', (msg) => {
+    // If messages are handled synchronously, throwing should break the IPC
+    // message processing.
+    throw msg;
+  });
+
+  process.send('ready');
+} else {
+  const child = cp.fork(__filename, ['child']);
+
+  child.on('message', common.mustCall((msg) => {
+    assert.strictEqual(msg, 'ready');
+    values.forEach((value) => {
+      child.send(value);
+    });
+  }));
+}