feat: drain sender and receiver children before shutdown

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);

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;

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;
         }
 

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) => {

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');

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 = {

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();
+};