From 2b96255d5e8aa0a5cf77756b27960fa12760db31 Mon Sep 17 00:00:00 2001 From: Diyorbek Rustamjonov Date: Wed, 9 Sep 2026 20:32:04 +0500 Subject: [PATCH] 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 --- constants.js | 4 ++ lib/Daemon.js | 6 ++ modules/pm2-axon/lib/sockets/pub.js | 58 ++++++++++++++----- .../pm2-axon/test/test.pub.slow-subscriber.js | 58 +++++++++++++++++++ 4 files changed, 112 insertions(+), 14 deletions(-) create mode 100644 modules/pm2-axon/test/test.pub.slow-subscriber.js diff --git a/constants.js b/constants.js index ca5c5bf493..781cf8bf84 100644 --- a/constants.js +++ b/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' diff --git a/lib/Daemon.js b/lib/Daemon.js index f46cbf0d45..bf135a3314 100644 --- a/lib/Daemon.js +++ b/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() { diff --git a/modules/pm2-axon/lib/sockets/pub.js b/modules/pm2-axon/lib/sockets/pub.js index 568f347480..e25729a7bc 100644 --- a/modules/pm2-axon/lib/sockets/pub.js +++ b/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; diff --git a/modules/pm2-axon/test/test.pub.slow-subscriber.js b/modules/pm2-axon/test/test.pub.slow-subscriber.js new file mode 100644 index 0000000000..bc722af064 --- /dev/null +++ b/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); + }); +});