fix: drop bus subscribers that stopped reading

constants.js

@@ -105,6 +105,10 @@ var csts = {
   KILL_SIGNAL             : process.env.PM2_KILL_SIGNAL || 'SIGINT',
   KILL_USE_MESSAGE        : process.env.PM2_KILL_USE_MESSAGE || false,
 
+  // Drop a pub.sock subscriber once this many bytes wait for it, otherwise a `pm2 logs`
+  // suspended with Ctrl+Z or a dead ssh session grows the daemon 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,12 @@ Daemon.prototype.innerStart = function(cb) {
    */
   this.pub    = axon.socket('pub-emitter');
 
+  // A subscriber that stopped reading must not make the daemon queue the whole log stream
+  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,8 @@ module.exports = PubSocket;
 
 function PubSocket() {
   Socket.call(this);
+  // bytes a peer may have queued before it is dropped, Infinity never drops
+  this.set('max pending bytes', Infinity);
 }
 
 /**
@@ -28,6 +30,38 @@ function PubSocket() {
 
 PubSocket.prototype.__proto__ = Socket.prototype;
 
+/**
+ * Write `buf` to `sock` unless the peer stopped reading.
+ *
+ * A peer that does not drain its socket fills the kernel buffer, after
+ * which every message is queued in this process memory without limit.
+ * Above `max pending bytes` the peer is destroyed (it may reconnect) and
+ * `slow subscriber` is emitted with the socket and the pending byte count.
+ *
+ * @param {net.Socket} sock
+ * @param {Buffer} buf
+ * @param {Function} [cb]
+ * @return {Boolean} true if `buf` was written
+ * @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 before emit: a listener may publish again (pm2 logs through
+    // the bus) and must see this peer as 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 +71,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 +92,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,58 @@
+
+var axon = require('..')
+  , net = require('net')
+  , should = require('should');
+
+// a subscriber that stops reading (like `pm2 logs` suspended with Ctrl+Z) is
+// dropped once `max pending bytes` is exceeded, a reading one 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(); // never reads, like a process stopped with SIGSTOP
+
+    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);
+  });
+});