bedrock - npm Package Compare versions

Comparing version 4.4.1 to 4.4.2

CHANGELOG.md

 # bedrock ChangeLog
 
+## 4.4.2 - 2021-11-04
+
+### Fixed
+- Signal handlers that prematurely terminated the primary or its workers
+  have been refactored to allow an orderly exit. The primary and worker
+  internal messaging and exit process has been cleaned up and simplified to
+  help ensure more consistent outcomes and eliminate a number of bugs or
+  unexpected states. Additionally, some unused IPC messages were removed.
+  The expectation is that these changes do not affect existing applications
+  except in those cases where those applications may have been exiting in
+  inconsistent ways; the aim is for those applications to now exit the same
+  way regardless of IPC message delivery order.
+
 ## 4.4.1 - 2021-09-21
@@ -4,0 +17,0 @@

lib/bedrock.js

@@ -29,2 +29,18 @@ /*!
 
+// primary process state
+let WORKERS_EXITED;
+const PRIMARY_STATE = {
+  switchedUser: false,
+  runOnce: {},
+  workersExited: new Promise(resolve => {
+    WORKERS_EXITED = resolve;
+  })
+};
+
+// exiting state
+let EXITING = false;
+
+// worker bedrock started state
+let BEDROCK_STARTED = false;
+
 // read package.json fields
@@ -132,3 +148,3 @@ pkginfo(module, 'version');
     console.error('Failed to initialize logging system:', err);
-    process.exit(1);
+    await _exit(1);
   }
@@ -138,3 +154,3 @@ // run
     _runPrimary(startTime, options);
-    // don't call callback in primary process
+    // don't emit `bedrock.error` in primary process
     return;
@@ -169,3 +185,3 @@ }
   // send message to primary
-  process.send({type: 'bedrock.switchProcessUser'});
+  process.send({type: 'bedrock.switchProcessUser'}, _exitOnError);
 };
@@ -189,3 +205,3 @@
   // notify primary to schedule work (if not already scheduled/run)
-  process.send({type, id, options});
+  process.send({type, id, options}, _exitOnError);
 
@@ -214,3 +230,3 @@ // wait for scheduling result
   // notify other workers that work is complete
-  process.send(msg);
+  process.send(msg, _exitOnError);
 
@@ -337,7 +353,2 @@ if(error) {
 
-  // exit on terminate
-  process.on('SIGTERM', function() {
-    process.exit();
-  });
-
   if(cluster.isMaster) {
@@ -370,6 +381,6 @@ cluster.setupMaster({
   if(config.cli.command.name() !== 'test') {
-    process.on('uncaughtException', function(error) {
+    process.on('uncaughtException', async function(error) {
       process.removeAllListeners('uncaughtException');
       logger.critical(`${logPrefix} uncaught error`, {error});
-      process.exit(1);
+      await _exit(1);
     });
@@ -382,6 +393,6 @@ }
   if(config.cli.command.name() !== 'test') {
-    process.on('unhandledRejection', function(error) {
+    process.on('unhandledRejection', async function(error) {
       process.removeAllListeners('unhandledRejection');
       logger.critical(`${logPrefix} unhandled promise rejection`, {error});
-      process.exit(1);
+      await _exit(1);
     });
@@ -414,6 +425,5 @@ }
   Object.keys(SIGNALS).forEach(signal => {
-    process.on(signal, async () => {
+    process.on(signal, async function exitProcess() {
       logger.info(`${logPrefix} received signal.`, {signal});
-      await events.emit('bedrock.exit');
-      process.exit();
+      await _exit();
     });
@@ -458,43 +468,49 @@ });
 
-  // keep track of primary state
-  const primaryState = {
-    switchedUser: false,
-    runOnce: {}
-  };
-
-  // notify workers to exit if primary exits
-  process.on('exit', function() {
-    for(const id in cluster.workers) {
-      cluster.workers[id].send({type: 'bedrock.core', message: 'exit'});
-    }
+  // the first time `process.exit()` is called on the primary, attempt to do
+  // an orderly exit
+  process.once('exit', async function() {
+    await _exit();
   });
 
   // handle worker exit
-  cluster.on('exit', function(worker, code, signal) {
+  cluster.on('exit', async function(worker, code, signal) {
     // if the worker called kill() or disconnect(), it was intentional, so exit
     // the process
-    if(worker.exitedAfterDisconnect) {
+    const shouldExit = worker.exitedAfterDisconnect;
+    if(shouldExit) {
       logger.info(
         `${logPrefix} worker "${worker.process.pid}" exited on purpose ` +
         `with code "${code}" and signal "${signal}"; exiting primary process.`);
-      process.exit(code);
+    } else {
+      // accidental worker exit (crash)
+      logger.critical(
+        `${logPrefix} worker "${worker.process.pid}" exited with code ` +
+        `"${code}" and signal "${signal}".`);
     }
 
-    // accidental worker exit (crash)
-    logger.critical(
-      `${logPrefix} worker "${worker.process.pid}" exited with code ` +
-      `"${code}" and signal "${signal}".`);
-
     // if configured, fork a replacement worker
-    if(config.core.worker.restart) {
+    if(!shouldExit && !EXITING && config.core.worker.restart) {
       // clear any runOnce records w/allowOnRestart option set
-      for(const id in primaryState.runOnce) {
-        if(primaryState.runOnce[id].worker === worker.id &&
-          primaryState.runOnce[id].options.allowOnRestart) {
-          delete primaryState.runOnce[id];
+      for(const id in PRIMARY_STATE.runOnce) {
+        if(PRIMARY_STATE.runOnce[id].worker === worker.id &&
+          PRIMARY_STATE.runOnce[id].options.allowOnRestart) {
+          delete PRIMARY_STATE.runOnce[id];
         }
       }
-      _startWorker(primaryState, script);
+      _startWorker(script);
     } else {
-      process.exit(1);
+      // if all workers have exited, resolve `WORKERS_EXITED`
+      let allDead = true;
+      for(const id in cluster.workers) {
+        if(!cluster.workers[id].isDead()) {
+          allDead = false;
+          break;
+        }
+      }
+      if(allDead) {
+        WORKERS_EXITED();
+      }
+      // exit primary (will wait on `WORKERS_EXITED` if not all workers have
+      // quit yet to allow an orderly exit)
+      await _exit(code);
     }
@@ -506,3 +522,3 @@ });
   for(let i = 0; i < workers; ++i) {
-    _startWorker(primaryState, script);
+    _startWorker(script);
   }
@@ -527,22 +543,2 @@ logger.info(`${logPrefix} started`, {timeMs: Date.now() - startTime});
 
-  // listen for primary process exit
-  let bedrockStarted = false;
-  process.on('message', function(msg) {
-    if(!_isMessageType(msg, 'bedrock.core') || msg.message !== 'exit') {
-      return;
-    }
-
-    if(!bedrockStarted) {
-      return events.emit('bedrock-cli.exit').finally(() => {
-        process.exit();
-      });
-    }
-
-    return events.emit('bedrock.stop').then(() => {
-      return events.emit('bedrock-cli.exit').finally(() => {
-        process.exit();
-      });
-    });
-  });
-
   const cliReady = await events.emit('bedrock-cli.ready');
@@ -554,4 +550,5 @@ // skip default behavior if cancelled (do not emit bedrock core events)
   }
-  bedrockStarted = true;
 
+  BEDROCK_STARTED = true;
+
   // snapshot the values of the fields that must be changed
@@ -585,3 +582,3 @@ // during the `bedrock.configure` event
 
-function _startWorker(state, script) {
+function _startWorker(script) {
   const worker = cluster.fork();
@@ -591,11 +588,4 @@ loggers.attach(worker);
   // listen to start requests from workers
-  worker.on('message', initWorker);
-
-  // TODO: simplify with cluster.on('online')?
-  function initWorker(msg) {
-    if(!_isMessageType(msg, 'bedrock.worker.started')) {
-      return;
-    }
+  worker.once('online', function initWorker() {
     // notify worker to initialize and provide the cwd and script to run
-    worker.removeListener('message', initWorker);
     worker.send({
@@ -605,10 +595,8 @@ type: 'bedrock.worker.init',
       script
+    }, err => {
+      if(err) {
+        // failure to send init message should hard terminate the worker
+        worker.process.kill();
+      }
     });
-  }
-
-  // listen to exit requests from workers
-  worker.on('message', function(msg) {
-    if(_isMessageType(msg, 'bedrock.core') && msg.message === 'exit') {
-      process.exit(msg.status);
-    }
   });
@@ -618,3 +606,3 @@
   // from a worker indicating to do so
-  if(!state.switchedUser) {
+  if(!PRIMARY_STATE.switchedUser) {
     worker.on('message', switchProcessUserListener);
@@ -628,4 +616,4 @@ }
     worker.removeListener('message', switchProcessUserListener);
-    if(!state.switchedUser) {
-      state.switchedUser = true;
+    if(!PRIMARY_STATE.switchedUser) {
+      PRIMARY_STATE.switchedUser = true;
       // switch group
@@ -642,3 +630,6 @@ if(config.core.running.groupId && process.setgid) {
 
-  // listen to schedule run once functions
+  // listen to schedule run once functions; when this message is received, it
+  // means a worker has asked the primary to schedule a function to run just
+  // once on the first worker that requests it to be run, causing the others
+  // to wait
   worker.on('message', function(msg) {
@@ -652,6 +643,6 @@ if(!_isMessageType(msg, 'bedrock.runOnce')) {
     if(msg.done) {
-      state.runOnce[msg.id].done = true;
-      state.runOnce[msg.id].error = msg.error || null;
+      PRIMARY_STATE.runOnce[msg.id].done = true;
+      PRIMARY_STATE.runOnce[msg.id].error = msg.error || null;
       // notify workers to call callback
-      const notify = state.runOnce[msg.id].notify;
+      const notify = PRIMARY_STATE.runOnce[msg.id].notify;
       while(notify.length > 0) {
@@ -665,3 +656,3 @@ const id = notify.shift();
             error: msg.error
-          });
+          }, _exitOnError);
         }
@@ -672,4 +663,4 @@ }
 
-    if(msg.id in state.runOnce) {
-      if(state.runOnce[msg.id].done) {
+    if(msg.id in PRIMARY_STATE.runOnce) {
+      if(PRIMARY_STATE.runOnce[msg.id].done) {
         // already ran, notify worker immediately
@@ -680,7 +671,7 @@ worker.send({
           done: true,
-          error: state.runOnce[msg.id].error
-        });
+          error: PRIMARY_STATE.runOnce[msg.id].error
+        }, _exitOnError);
       } else {
         // still running, add worker ID to notify queue for later notification
-        state.runOnce[msg.id].notify.push(worker.id);
+        PRIMARY_STATE.runOnce[msg.id].notify.push(worker.id);
       }
@@ -691,3 +682,3 @@ return;
     // run in this worker
-    state.runOnce[msg.id] = {
+    PRIMARY_STATE.runOnce[msg.id] = {
       worker: worker.id,
@@ -699,3 +690,3 @@ notify: [],
     };
-    worker.send({type, id: msg.id, done: false});
+    worker.send({type, id: msg.id, done: false}, _exitOnError);
   });
@@ -731,1 +722,92 @@ }
 }
+
+async function _preparePrimaryExit() {
+  let allDead = true;
+  for(const id in cluster.workers) {
+    const worker = cluster.workers[id];
+    if(!worker.isDead()) {
+      // if any error occurs when telling the worker to terminate, hard
+      // terminate it instead and ignore any errors that occur as a result
+      // of that
+      worker.once('error', () => {
+        worker.once('error', () => {});
+        worker.process.kill();
+      });
+      cluster.workers[id].kill();
+      allDead = false;
+    }
+  }
+  if(allDead) {
+    // all workers are dead
+    WORKERS_EXITED();
+  }
+}
+
+async function _exit(code) {
+  /* Notes on exiting:
+
+  When the primary does an orderly exit, it must notify all workers
+  to exit, wait for them to do so and then finally exit itself.
+
+  When a worker does an orderly exit, it emits and awaits bedrock events based
+  on the current state of the application and then exits. Upon exiting, a
+  message will be sent to the primary with the exit status code.
+
+  Regardless of whether a worker or the primary exit first, the primary will
+  be notified of a worker exiting and it will decide whether or not to kill
+  all workers and exit itself.
+
+  If any worker or the primary are asked to exit while an orderly exit is in
+  progress, this request will be ignored and the process will exit once the
+  orderly exit completes. */
+
+  // orderly exit already in progress
+  if(EXITING) {
+    return;
+  }
+
+  EXITING = true;
+
+  try {
+    if(cluster.isMaster) {
+      await _preparePrimaryExit();
+      await PRIMARY_STATE.workersExited;
+    } else {
+      // these events are only emitted in workers
+      if(BEDROCK_STARTED) {
+        await events.emit('bedrock.stop');
+      }
+      await events.emit('bedrock-cli.exit');
+      await events.emit('bedrock.exit');
+    }
+  } finally {
+    await _logExit(code);
+    process.exit(code);
+  }
+}
+
+async function _exitOnError(err) {
+  if(err) {
+    // a failure to send an IPC message is fatal, exit
+    await _exit(1);
+  }
+}
+
+async function _logExit(code = 0) {
+  if(!cluster.isMaster) {
+    return;
+  }
+  // log final message and wait for logger to flush
+  const logger = loggers.get('app');
+  const logPrefix = '[bedrock/primary]';
+  try {
+    const p = new Promise(resolve => {
+      logger.info(
+        `${logPrefix} primary process exiting with code "${code}".`, {code},
+        () => {
+          resolve();
+        });
+    });
+    await p;
+  } finally {}
+}

lib/loggers/WorkerTransport.js

@@ -35,3 +35,4 @@ /*!
 
-    // send logger message to primary
+    // send logger message to primary and ignore EPIPE errors that prevent the
+    // message from reaching the primary because it has disconnected
     process.send({
@@ -41,3 +42,3 @@ type: 'bedrock.logger',
       category: this.category
-    });
+    }, _noop);
 
@@ -47,1 +48,4 @@ callback();
 };
+
+function _noop() {
+}

lib/worker.js

 /*!
- * Copyright (c) 2012-2019 Digital Bazaar, Inc. All rights reserved.
+ * Copyright (c) 2012-2021 Digital Bazaar, Inc. All rights reserved.
  */
@@ -9,5 +9,2 @@ 'use strict';
 
-// notify primary to send initialization options
-process.send({type: 'bedrock.worker.started'});
-
 function init(msg) {
@@ -14,0 +11,0 @@ if(!(typeof msg === 'object' && msg.type === 'bedrock.worker.init')) {

package.json

 {
   "name": "bedrock",
-  "version": "4.4.1",
+  "version": "4.4.2",
   "description": "A core foundation for rich Web applications.",
@@ -5,0 +5,0 @@ "license": "SEE LICENSE IN LICENSE.md",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

157156

increased by2.13%

Lines of code

2398

increased by3.27%

No dependency changes