Fix startup race condition (#3367)

Changes.md

@@ -96,6 +96,7 @@
 - fix(outbound): replace empty Message-ID header #3288
 - fix(outbound): don't send SNI servername when connecting to an IP
 - fix(outbound): chown queue dir after creation #3291
+- fix(server): async endpoint.bind() and await in server.js #3366
 
 ### [3.0.3] - 2024-02-07
 

endpoint.js

@@ -1,7 +1,7 @@
 'use strict';
 // Socket address parser/formatter and server binding helper
 
-const fs = require('node:fs');
+const fs = require('node:fs/promises');
 const sockaddr = require('sockaddr');
 
 module.exports = function endpoint (addr, defaultPort) {
@@ -46,21 +46,24 @@ class Endpoint {
     }
 
     // Make server listen on this endpoint, w/optional options
-    bind (server, opts) {
-        let done;
-        opts = Object.assign({}, opts || {});
+    async bind (server, opts) {
+        opts = {...opts};
+
+        const mode = this.mode ? parseInt(this.mode, 8) : false;
         if (this.path) {
-            const path = opts.path = this.path;
-            const mode = this.mode ? parseInt(this.mode, 8) : false;
-            if (mode) {
-                done = () => fs.chmodSync(path, mode);
-            }
-            if (fs.existsSync(path)) fs.unlinkSync(path);
-        }
-        else {
+            opts.path = this.path;
+            await fs.rm(this.path, { force: true }); // errors are ignored when force is true
+        } else {
             opts.host = this.host;
             opts.port = this.port;
         }
-        server.listen(opts, done);
+
+        return new Promise((resolve, reject) => {
+            server.listen(opts, async (err) => {
+                if(err) return reject(err);
+                if (mode) await fs.chmod(opts.path, mode);
+                resolve()
+            });
+        });
     }
 }

server.js

@@ -446,7 +446,7 @@ Server.setup_smtp_listeners = async (plugins2, type, inactivity_timeout) => {
                 Server.logerror(e)
             })
 
-        ep.bind(server, {backlog: 0});
+        await ep.bind(server, {backlog: 0});
     }
 
     if (errors.length) {
@@ -459,7 +459,7 @@ Server.setup_smtp_listeners = async (plugins2, type, inactivity_timeout) => {
     plugins2.run_hooks(`init_${type}`, Server);
 }
 
-Server.setup_http_listeners = () => {
+Server.setup_http_listeners = async () => {
     if (!Server.http?.cfg?.listen) return;
 
     const listeners = Server.get_listen_addrs(Server.http.cfg, 80);
@@ -505,7 +505,7 @@ Server.setup_http_listeners = () => {
             Server.logerror(e);
         })
 
-        ep.bind(Server.http.server, {backlog: 0});
+        await ep.bind(Server.http.server, {backlog: 0});
     }
 
     Server.plugins.run_hooks('init_http', Server);

test/endpoint.js

@@ -45,70 +45,49 @@ describe('endpoint', () => {
             }
 
             this.mockfs = {
-                existsSync (path, ...args) {
-                    log.push(['existsSync', path, ...args]);
-                    return ('undefined' !== typeof modes[path]);
-                },
-                chmodSync (path, mode, ...args) {
-                    log.push(['chmodSync', path, mode, ...args]);
+                chmod (path, mode, ...args) {
+                    log.push(['chmod', path, mode, ...args]);
                     modes[path] = mode;
                 },
-                unlinkSync (path, ...args) {
-                    log.push(['unlinkSync', path, ...args]);
-                    if ('undefined' !== typeof modes[path]) {
-                        delete modes[path];
-                    }
-                    else {
-                        log.push(['unlink without existing socket']);
-                    }
+                rm (path, ...args) {
+                    log.push(['rm', path, ...args]);
                 },
             };
 
-            mock('node:fs', this.mockfs);
+            mock('node:fs/promises', this.mockfs);
             this.endpoint = mock.reRequire('../endpoint');
             done();
         })
 
         afterEach((done) => {
-            mock.stop('node:fs');
+            mock.stop('node:fs/promises');
             done();
         })
 
-        it('IP socket', () => {
-            this.endpoint('10.0.0.3:42').bind(this.server, {backlog:19});
+        it('IP socket', async () => {
+            await this.endpoint('10.0.0.3:42').bind(this.server, {backlog:19});
             assert.deepEqual(
                 this.log, [
                     ['listen', {host: '10.0.0.3', port: 42, backlog: 19}],
                 ]);
         })
 
-        it('Unix socket', () => {
-            this.endpoint('/foo/bar.sock').bind(this.server, {readableAll:true});
+        it('Unix socket', async () => {
+            await this.endpoint('/foo/bar.sock').bind(this.server, {readableAll:true});
             assert.deepEqual(
                 this.log, [
-                    ['existsSync', '/foo/bar.sock'],
+                    ['rm', '/foo/bar.sock', {force:true}],
                     ['listen', {path: '/foo/bar.sock', readableAll: true}],
                 ]);
         })
 
-        it('Unix socket (pre-existing)', () => {
-            this.modes['/foo/bar.sock'] = 0o755;
-            this.endpoint('/foo/bar.sock').bind(this.server);
-            assert.deepEqual(
-                this.log, [
-                    ['existsSync', '/foo/bar.sock'],
-                    ['unlinkSync', '/foo/bar.sock'],
-                    ['listen', {path: '/foo/bar.sock'}],
-                ]);
-        })
-
-        it('Unix socket w/mode', () => {
-            this.endpoint('/foo/bar.sock:764').bind(this.server);
+        it('Unix socket w/mode', async () => {
+            await this.endpoint('/foo/bar.sock:764').bind(this.server);
             assert.deepEqual(
                 this.log, [
-                    ['existsSync', '/foo/bar.sock'],
+                    ['rm', '/foo/bar.sock', {force:true}],
                     ['listen', {path: '/foo/bar.sock'}],
-                    ['chmodSync', '/foo/bar.sock', 0o764],
+                    ['chmod', '/foo/bar.sock', 0o764],
                 ]);
         })
     })