message:store not handling async/callback correctly

[skadefro]

I truly hope this is not me missing something obvious again. I have tested both with callback syntax and promises, f.eks this will easily reproduce the issue:

app.addHook('message:headers', (envelope, messageInfo) => {
    return new Promise(async (resolve, reject) => {
        var err = new Error("hi mom");
       err.responseCode = 554;
        return reject(app.reject(envelope, 'ai-spam-check', messageInfo, `${err.responseCode} ${err.message}`));
    });
});

and

app.addHook('message:headers', (envelope, messageInfo, next) => {
    var err = new Error("hi mom");
    err.responseCode = 554;
    return next(app.reject(envelope, 'ai-spam-check', messageInfo, `${err.responseCode} ${err.message}`));
});

doing this with any of the 3 events message:headers and message:queue will make the process throw and unhandled exception in mail-drop.js at https://github.com/zone-eu/zone-mta/blob/a4d2258d52d8c7a87cef12a77844729428578319/lib/mail-drop.js#L145 resulting in the process dying and the client getting disconnected. Sure the process starts up again, but will just die the next time the mail is tried delivered. The error is from the fact, dkimStream.once('error', err => { is not called until further down but i cannot see why. i tried moving my plugin registration in config/default.js to the bottom in case plugins are handled in order, but that made no difference. What am i missing here, are we not suppose to be able to decline messages from within these 3 events or am I doing it wrongly ? If not, then I guess I could add a custom property to envelope and then trow the error in a later event, but what event would come after message:queue that could handle it ?

[skadefro]

ffs, why do you always see the error right after posting. Sorry for the noise. I can do it from message:queue, i did not test that the right way, it works fine from message:queue

[skadefro]

Ah, i removed all events from my code, to make sure, it was easily reproducible, and it was not. Suddenly everything seemed to work. so slowly re-added code, and i found the issue, but not a solution. Reopening, i believe this is a bug, we should be allowed to do some kind of processing from within the message:store event, that can take more than 2 seconds. ( I need this, to scan the mail content i think the correct way is to use AnalyzerHook, like the anti virus plugins are doing. So I am going to use that instead, but I still believe there is an issue with message:store )

There is some kind of timing issue with DKIM and message:store event. If you resolve or throw and error quickly from message:store it works. But if you add any kind of delay to message:store it fails. I suspect that means dkimStream.once('error', err => needs to be moved further up in the code, or maybe the problem is events are not processed in sequence ? Reproducible code, works with both callback and promises

    app.addHook('message:headers', (envelope, messageInfo, next) => {
        return new Promise(async (resolve, reject) => {
            var err = new Error("hi mom");
            err.responseCode = 554;
            return reject(app.reject(envelope, 'ai-spam-check', messageInfo, `${err.responseCode} ${err.message}`));
        });
    });
    app.addHook('message:store2', (envelope, body) => {
        return new Promise(async (resolve, reject) => {
            setTimeout(() => {
                resolve();
            }, 2000);
        });        
    });
    app.addHook('message:store', (envelope, body, next) => {
        setTimeout(() => {
            next();
        }, 2000);        
    });

This wil throw and unhandled exception in mail-drop.js at https://github.com/zone-eu/zone-mta/blob/a4d2258d52d8c7a87cef12a77844729428578319/lib/mail-drop.js#L145 Same even when reading the body, like this, will still give the above unhandled exception

app.addHook('message:store', (envelope, body, next) => {
    let chunks = [];
    body.on('data', chunk => {
        chunks.push(chunk);
    });
    body.on('end', async () => {
        setTimeout(() => {
            next();
        }, 2000);        
    });
    body.resume();
});

[andris9]

It's a stream pipeline. If you exhaust the stream in your handler, then the pipeline breaks, as there would be nothing left to consume. This hook should be as fast as possible and not touch the stream itself.