From 311df9893c9a4097e2fce8631ecba9cdb6d16ed0 Mon Sep 17 00:00:00 2001 From: DeamonMV Date: Fri, 24 Jul 2026 17:19:29 +0300 Subject: [PATCH] feat: drain sender and receiver children before shutdown On SIGTERM/SIGINT the master closes the queue server, but forked sender and receiver children only find out when their queue-server socket drops -- logged as an unexpected connection-closed error, and for senders, in-flight deliveries are cut off mid-send. Before tearing down the queue server, the master now notifies every child over the fork IPC channel. Each child marks itself closing (so the socket drop is expected, not logged as an error) and drains before exiting: senders finish their current delivery and stop fetching new work, receivers let open SMTP sessions complete via smtp-server's close(). The queue server (and the Mongo connection behind it) is not torn down until every notified child has actually exited. Server.close() (lib/transport/ server.js) doesn't just stop listening -- it immediately force-closes every connected child socket, so closing it right away would sever an in-flight sender's only channel back to the queue mid-delivery, before it can report the outcome, leaving it hung forever instead of exiting. There is still no per-child drain timeout beyond that: a child that never finishes is caught by the master's existing ~10s force-exit. The sender's shutdown handler captures its drain target up front: a sender spawned by the staggered startup delay could otherwise be added to the set after the drain loop already started, and never be waited for. --- app.js | 39 +++++++-- lib/receiver/smtp-proxy.js | 14 ++++ lib/sender.js | 4 + lib/sending-zone.js | 17 ++++ services/receiver.js | 21 +++++ services/sender.js | 44 ++++++++++ test/graceful-shutdown-test.js | 147 +++++++++++++++++++++++++++++++++ 7 files changed, 281 insertions(+), 5 deletions(-) create mode 100644 test/graceful-shutdown-test.js diff --git a/app.js b/app.js index 14078e5..f594a10 100644 --- a/app.js +++ b/app.js @@ -228,6 +228,12 @@ let stop = code => { log.info('Process', 'Server closing down...'); queue.closing = true; + // Notify sender and receiver children over the fork IPC channel before the queue server + // is torn down, so they drain in-flight work and exit quietly instead of each logging an + // unexpected queue-connection-closed error when their socket to it drops. + sendingZone.closeSenders(); + smtpInterfaces.forEach(smtpInterface => smtpInterface.closeChildren()); + let closed = 0; let checkClosed = () => { if (++closed === 2 + smtpInterfaces.length) { @@ -254,12 +260,35 @@ let stop = code => { checkClosed(); }); - queueServer.close(() => { - // wait until all connections to the API HTTP are closed - log.info('QS', 'Service closed'); - checkClosed(); + // Server.close() (lib/transport/server.js) does not just stop listening -- it immediately + // force-closes every currently connected child socket. Closing the queue server (and the + // Mongo connection behind it via queue.stop()) before a notified child has actually exited + // would cut off its only channel back to the queue mid-delivery, so it can never report the + // outcome and hangs forever instead of exiting. Wait for every child to close on its own + // first; the forceExitTimer below is still the hard ceiling if one never does. + let waitForChildrenDrain = onDrained => { + let remaining = () => { + let total = 0; + sendingZone.sendingZonelist.forEach(zone => (total += zone.children.size)); + smtpInterfaces.forEach(smtpInterface => (total += smtpInterface.children.size)); + return total; + }; + let check = () => { + if (!remaining()) { + return onDrained(); + } + setTimeout(check, 100).unref(); + }; + check(); + }; + + waitForChildrenDrain(() => { + queueServer.close(() => { + log.info('QS', 'Service closed'); + checkClosed(); + }); + queue.stop(); }); - queue.stop(); // If we were not able to stop other stuff by 10 sec. force close let forceExitTimer = setTimeout(() => forceStop(code), 10 * 1000); diff --git a/lib/receiver/smtp-proxy.js b/lib/receiver/smtp-proxy.js index b49577d..9cddb1e 100644 --- a/lib/receiver/smtp-proxy.js +++ b/lib/receiver/smtp-proxy.js @@ -368,6 +368,20 @@ class SMTPProxy { } this.server.close(callback); } + + // Tell receiver children to shut down over the fork IPC channel before the proxy itself + // closes. Sets `closing` first so the 'close' handler above treats their exit as expected + // instead of logging it and respawning. + closeChildren() { + this.closing = true; + this.children.forEach(child => { + try { + child.send({ shutdown: true }); + } catch (err) { + // IPC channel already gone; the child will exit on its own + } + }); + } } module.exports = SMTPProxy; diff --git a/lib/sender.js b/lib/sender.js index 752ca7e..9ad678c 100644 --- a/lib/sender.js +++ b/lib/sender.js @@ -176,6 +176,10 @@ class Sender extends EventEmitter { sendNext() { if (this.closing) { + // Draining: the current delivery (if any) already completed and we hit this gate + // instead of fetching more work. Emit 'closed' so a caller waiting to shut down + // this process knows this sender is done. Safe to emit more than once. + this.emit('closed'); return; } diff --git a/lib/sending-zone.js b/lib/sending-zone.js index cc1a286..254555e 100644 --- a/lib/sending-zone.js +++ b/lib/sending-zone.js @@ -208,6 +208,19 @@ class SendingZone { return pool[index]; } + // Tell every sender child to shut down over the fork IPC channel. Called from app.js + // stop() before the queue server closes, so children set `closing` and exit cleanly + // instead of logging an unexpected queue-connection-closed error when the socket drops. + close() { + this.children.forEach(child => { + try { + child.send({ shutdown: true }); + } catch (err) { + // IPC channel already gone; the child will exit on its own + } + }); + } + spawnSenders(callback) { if (!this.queue || this.queue.closing || this.disabled) { return setImmediate(callback); @@ -519,6 +532,10 @@ class DomainConfig { module.exports.sendingZonelist = sendingZonelist; module.exports.SendingZone = SendingZone; +// Notify every zone's sender children that the server is shutting down. +module.exports.closeSenders = () => { + sendingZonelist.forEach(zone => zone.close()); +}; module.exports.routingHeaders = routingHeaders; module.exports.init = (queue, callback) => { diff --git a/services/receiver.js b/services/receiver.js index 10991cf..b403292 100644 --- a/services/receiver.js +++ b/services/receiver.js @@ -179,6 +179,27 @@ queueClient.connect(err => { // start accepting sockets process.on('message', (m, socket) => { + if (m && m.shutdown) { + // The master (lib/receiver/smtp-proxy.js closeChildren()) sends this over the fork IPC + // channel when the whole server is stopping, since only the master receives SIGTERM/ + // SIGINT directly. Setting `closing` here makes the queue-connection 'close' handler + // above treat the impending socket close as expected instead of logging it as an error. + if (closing) { + return; + } + closing = true; + + if (!smtpServer || !smtpServer.server) { + return process.exit(0); + } + + log.info('SMTP/' + currentInterface + '/' + process.pid, 'Received shutdown from master, draining SMTP sessions'); + return smtpServer.close(() => { + log.info('SMTP/' + currentInterface + '/' + process.pid, 'Graceful shutdown, draining complete, exiting'); + process.exit(0); + }); + } + if (m === 'socket') { if (!socket) { log.verbose('SMTP/' + currentInterface + '/' + process.pid, 'Null Socket'); diff --git a/services/sender.js b/services/sender.js index 0fffb63..4c01c92 100644 --- a/services/sender.js +++ b/services/sender.js @@ -64,6 +64,50 @@ log.info(logName, '[%s] Starting sending for %s', clientId, zone.name); process.title = config.ident + ': sender/' + currentZone; +// The master (lib/sending-zone.js close()) sends { shutdown: true } over the fork IPC channel +// when the whole server is stopping, since only the master receives SIGTERM/SIGINT directly. +// Setting `closing` here makes the queue-connection 'close' handler above treat the impending +// socket close as expected instead of logging it as an error. +process.on('message', m => { + if (!m || !m.shutdown || closing) { + return; + } + closing = true; + + let finished = false; + let finish = () => { + if (finished) { + return; + } + finished = true; + log.info(logName, 'Graceful shutdown, draining complete, exiting'); + process.exit(0); + }; + + if (!senders.size) { + return finish(); + } + + log.info(logName, 'Received shutdown from master, draining %s sender(s)', senders.size); + + // Capture the drain target now: a sender spawned by the staggered setTimeout in + // spawnConnections (up to 1500ms after startup) could otherwise be added to `senders` + // after this loop but before every 'closed' fires, growing senders.size past the drained + // count so finish() would never run. + let totalToDrain = senders.size; + let drained = new Set(); + senders.forEach(sender => { + sender.removeAllListeners('error'); + sender.once('closed', () => { + drained.add(sender); + if (drained.size >= totalToDrain) { + finish(); + } + }); + sender.close(); + }); +}); + let sendCommand = (cmd, callback) => { let id = ++cmdId; let data = { diff --git a/test/graceful-shutdown-test.js b/test/graceful-shutdown-test.js new file mode 100644 index 0000000..5f9625d --- /dev/null +++ b/test/graceful-shutdown-test.js @@ -0,0 +1,147 @@ +'use strict'; + +// Covers the notify + drain primitives the graceful-shutdown wiring relies on: +// - SendingZone.close() / closeSenders() -> IPC { shutdown: true } to sender children +// - SMTPProxy.closeChildren() -> IPC { shutdown: true } to receiver children +// - Sender.close() + sendNext() -> the 'closed' drain handshake +// The actual process.on('message', ...) handlers in services/sender.js and +// services/receiver.js run as forked child processes and can't be required in-process, +// so this only exercises the building blocks they call. + +const EventEmitter = require('events'); +const sendingZoneModule = require('../lib/sending-zone'); +const { SendingZone } = sendingZoneModule; +const SMTPProxy = require('../lib/receiver/smtp-proxy'); +const Sender = require('../lib/sender'); + +// A fake forked child: records what was sent over the IPC channel. throwOnSend simulates +// a channel that is already gone (child exited), so we can assert the notify loop keeps +// going instead of stopping at the first dead child. +function makeChild(throwOnSend) { + let sent = []; + return { + sent, + send(msg) { + if (throwOnSend) { + throw new Error('channel closed'); + } + sent.push(msg); + } + }; +} + +module.exports['SendingZone.close() sends shutdown to every sender child'] = test => { + let zone = Object.create(SendingZone.prototype); + let c1 = makeChild(); + let c2 = makeChild(); + zone.children = new Set([c1, c2]); + + zone.close(); + + test.deepEqual(c1.sent, [{ shutdown: true }]); + test.deepEqual(c2.sent, [{ shutdown: true }]); + test.done(); +}; + +module.exports['SendingZone.close() swallows a dead IPC channel and still notifies the rest'] = test => { + let zone = Object.create(SendingZone.prototype); + let dead = makeChild(true); + let alive = makeChild(); + // Insertion order is iteration order: the dead child goes first, so if the throw were + // not caught the alive child would never be notified. + zone.children = new Set([dead, alive]); + + test.doesNotThrow(() => zone.close()); + test.deepEqual(alive.sent, [{ shutdown: true }]); + test.done(); +}; + +module.exports['closeSenders() calls close() on every registered zone'] = test => { + let goodClosed = 0; + let badClosed = 0; + sendingZoneModule.sendingZonelist.set('good', { close: () => goodClosed++ }); + sendingZoneModule.sendingZonelist.set('bad', { close: () => badClosed++ }); + + sendingZoneModule.closeSenders(); + + test.equal(goodClosed, 1); + test.equal(badClosed, 1); + sendingZoneModule.sendingZonelist.clear(); + test.done(); +}; + +module.exports['SMTPProxy.closeChildren() sets closing and notifies every receiver child'] = test => { + let proxy = Object.create(SMTPProxy.prototype); + proxy.closing = false; + let c1 = makeChild(); + let c2 = makeChild(); + proxy.children = new Set([c1, c2]); + + proxy.closeChildren(); + + // `closing` must be set so the 'close' handler treats the exit as expected (no error log, + // no respawn). + test.equal(proxy.closing, true); + test.deepEqual(c1.sent, [{ shutdown: true }]); + test.deepEqual(c2.sent, [{ shutdown: true }]); + test.done(); +}; + +module.exports['SMTPProxy.closeChildren() swallows a dead IPC channel and still notifies the rest'] = test => { + let proxy = Object.create(SMTPProxy.prototype); + proxy.closing = false; + let dead = makeChild(true); + let alive = makeChild(); + proxy.children = new Set([dead, alive]); + + test.doesNotThrow(() => proxy.closeChildren()); + test.deepEqual(alive.sent, [{ shutdown: true }]); + test.done(); +}; + +function makeSender() { + let sender = Object.create(Sender.prototype); + EventEmitter.call(sender); + sender.closing = false; + sender.zone = { name: 'good' }; + sender.logName = 'Sender/good/test'; + return sender; +} + +module.exports['Sender.close() flips closing and the next sendNext() emits "closed"'] = test => { + let sender = makeSender(); + let closedCount = 0; + sender.on('closed', () => closedCount++); + + sender.close(); + test.equal(sender.closing, true); + + // The next loop cycle hits the drain gate and signals it has stopped, letting + // services/sender.js know this sender is done. + sender.sendNext(); + test.equal(closedCount, 1); + test.done(); +}; + +module.exports['Sender.sendNext() while draining does not fetch more work'] = test => { + let sender = makeSender(); + sender.closing = true; + let sendCommandCalled = false; + sender.sendCommand = () => { + sendCommandCalled = true; + }; + sender.on('closed', () => {}); + + sender.sendNext(); + + test.equal(sendCommandCalled, false); + test.done(); +}; + +module.exports['Sender.close() is idempotent'] = test => { + let sender = makeSender(); + sender.close(); + test.doesNotThrow(() => sender.close()); + test.equal(sender.closing, true); + test.done(); +}; \ No newline at end of file