fix: drop bus subscribers that stopped reading

A `pm2 logs` or `pm2 monit` client that stops reading pub.sock (suspended
with Ctrl+Z, dead ssh session) fills the kernel socket buffer, after which
PubSocket.send queues every log line of every app in the daemon memory
without any limit. The God Daemon then dies with "Reached heap limit" and
its fork children are left behind as orphans.

PubSocket gets a `max pending bytes` setting: once a peer has more than
that waiting in its write queue it is destroyed (it may reconnect) and a
`slow subscriber` event is emitted. The daemon enables it with 16 MB by
default, configurable with PM2_PUB_MAX_PENDING_BYTES (0 disables), and
logs the drop in pm2.log.

Refs #5145 #6113 #5802 #5917 #4737 #4647 #3049

Author: diyorbekrustamjonov

constants.js

@@ -105,6 +105,11 @@ var csts = {
   KILL_SIGNAL             : process.env.PM2_KILL_SIGNAL || 'SIGINT',
   KILL_USE_MESSAGE        : process.env.PM2_KILL_USE_MESSAGE || false,
 
+  // Drop a bus (pub.sock) subscriber once this many bytes are waiting for it,
+  // otherwise a client that stopped reading (`pm2 logs` suspended with Ctrl+Z,
+  // dead ssh session) makes the daemon buffer every log line until OOM. 0 disables.
+  PUB_MAX_PENDING_BYTES   : isNaN(parseInt(process.env.PM2_PUB_MAX_PENDING_BYTES)) ? 16 * 1024 * 1024 : parseInt(process.env.PM2_PUB_MAX_PENDING_BYTES),
+
   PM2_PROGRAMMATIC        : typeof(process.env.pm_id) !== 'undefined' || process.env.PM2_PROGRAMMATIC,
   PM2_LOG_DATE_FORMAT     : process.env.PM2_LOG_DATE_FORMAT !== undefined ? process.env.PM2_LOG_DATE_FORMAT : 'YYYY-MM-DDTHH:mm:ss'
 

lib/Daemon.js

@@ -104,6 +104,13 @@ Daemon.prototype.innerStart = function(cb) {
    */
   this.pub    = axon.socket('pub-emitter');
 
+  // A subscriber that stopped reading (`pm2 logs` suspended with Ctrl+Z, dead
+  // ssh session) must not make the daemon queue the whole log stream in memory
+  this.pub.sock.set('max pending bytes', cst.PUB_MAX_PENDING_BYTES);
+  this.pub.sock.on('slow subscriber', function(sock, pending) {
+    console.log('Bus subscriber is not reading (%d bytes pending), dropping it', pending);
+  });
+
   this.pub_socket = this.pub.bind(this.pub_socket_file);
 
   this.pub_socket.once('bind', function() {

modules/pm2-axon/lib/sockets/pub.js

@@ -20,6 +20,7 @@ module.exports = PubSocket;
 
 function PubSocket() {
   Socket.call(this);
+  this.set('max pending bytes', Infinity);
 }
 
 /**
@@ -28,6 +29,40 @@ function PubSocket() {
 
 PubSocket.prototype.__proto__ = Socket.prototype;
 
+/**
+ * Write `buf` to `sock`, unless the peer stopped reading.
+ *
+ * A subscriber that does not drain its socket (a `pm2 logs` suspended
+ * with Ctrl+Z, a dead ssh session) fills the kernel buffer, after which
+ * every message is queued in this process memory without any limit.
+ * Once more than `max pending bytes` are waiting for a peer, the peer is
+ * dropped instead (it may reconnect) and a `slow subscriber` event is
+ * emitted with the socket and the number of pending bytes.
+ *
+ * @param {net.Socket} sock
+ * @param {Buffer} buf
+ * @param {Function} [cb]
+ * @return {Boolean} true if `buf` was handed to `sock`
+ * @api private
+ */
+
+PubSocket.prototype.writeTo = function(sock, buf, cb){
+  if (!sock.writable) return false;
+
+  var max = this.get('max pending bytes');
+  if (max > 0 && sock.writableLength > max) {
+    var pending = sock.writableLength;
+    // destroy first: listeners of the event may publish again (the pm2
+    // daemon logs through the bus) and must find this peer not writable
+    sock.destroy();
+    this.emit('slow subscriber', sock, pending);
+    return false;
+  }
+
+  sock.write(buf, cb);
+  return true;
+};
+
 /**
  * Send `msg` to all established peers.
  *
@@ -37,20 +72,19 @@ PubSocket.prototype.__proto__ = Socket.prototype;
 
 PubSocket.prototype.send = function(msg){
   var socks = this.socks;
-  var len = socks.length;
   var buf = this.pack(arguments);
 
   for (var sock of socks) {
-      if (sock.writable) sock.write(buf);
+    this.writeTo(sock, buf);
   }
 
   return this;
 };
 
 PubSocket.prototype.sendv2 = function(data, cb){
+  var self = this;
   var socks = this.socks;
   var len = socks.length;
-  var sock;
 
   if (len == 0)
     return process.nextTick(cb);
@@ -59,18 +93,15 @@ PubSocket.prototype.sendv2 = function(data, cb){
 
   var i = 0;
 
+  function done() {
+    i++;
+    if (i == len)
+      process.nextTick(cb);
+  }
+
   socks.forEach(function(sock) {
-    if (sock.writable)
-      sock.write(buf, function() {
-        i++;
-        if (i == len)
-          process.nextTick(cb);
-      });
-    else {
-      i++;
-      if (i == len)
-        process.nextTick(cb);
-    }
+    if (!self.writeTo(sock, buf, done))
+      done();
   });
 
   return this;

modules/pm2-axon/test/test.pub.slow-subscriber.js

@@ -0,0 +1,59 @@
+
+var axon = require('..')
+  , net = require('net')
+  , should = require('should');
+
+// A subscriber that stops reading (like `pm2 logs` suspended with Ctrl+Z)
+// must be dropped once `max pending bytes` is exceeded, while a subscriber
+// that keeps reading stays connected.
+
+var LIMIT = 1024 * 1024;
+
+var pub = axon.socket('pub');
+pub.set('max pending bytes', LIMIT);
+
+var payload = Buffer.alloc(64 * 1024, 'x');
+
+pub.bind(4200, function(){
+  var reader = net.connect(4200);
+  reader.on('data', function(){});
+
+  var stuck = net.connect(4200);
+  stuck.on('connect', function(){
+    stuck.pause();
+
+    var dropped = null;
+    pub.on('slow subscriber', function(sock, pending){
+      pending.should.be.above(LIMIT);
+      sock.remotePort.should.equal(stuck.localPort);
+      dropped = sock;
+    });
+
+    // peers are registered on the server side asynchronously
+    setTimeout(function(){
+      pub.socks.length.should.equal(2);
+
+      var sent = 0;
+      var timer = setInterval(function(){
+        for (var i = 0; i < 8 && !dropped; i++) {
+          pub.send(payload);
+          sent++;
+        }
+
+        if (dropped) {
+          clearInterval(timer);
+          // the stuck peer is gone, the reading one is kept
+          setTimeout(function(){
+            pub.socks.length.should.equal(1);
+            pub.socks[0].remotePort.should.equal(reader.localPort);
+            reader.destroy();
+            stuck.destroy();
+            pub.close();
+          }, 100);
+        } else if (sent > 20000) {
+          throw new Error('slow subscriber was not dropped after ' + sent + ' messages');
+        }
+      }, 5);
+    }, 50);
+  });
+});