diff --git a/backend/__tests__/services/emailWebhookTransport.test.js b/backend/__tests__/services/emailWebhookTransport.test.js index 64e2edac..f13505a5 100644 --- a/backend/__tests__/services/emailWebhookTransport.test.js +++ b/backend/__tests__/services/emailWebhookTransport.test.js @@ -48,7 +48,6 @@ beforeEach(() => { process.env.EMAIL_WEBHOOK_SECRET = SECRET; transport.__testing.setAllowPrivateUrls(false); transport.__testing.resetSecretWarning(); - transport.__testing.setReadTimeout(10000); validateExternalUrlAsync.mockResolvedValue({ valid: true, reason: 'ok' }); axios.post.mockResolvedValue({ status: 200, data: streamOf('') }); }); @@ -212,13 +211,16 @@ describe('attachments', () => { }); describe('response handling', () => { - it('caps how much of a receiver response it will buffer', async () => { + it('streams the response and discards it unread', async () => { + const body = streamOf('{"messageId":"ignored"}'); + axios.post.mockResolvedValue({ status: 200, data: body }); await transport.send(MAIL); + const opts = axios.post.mock.calls[0][2]; - // Only the status and an optional messageId are read; an unbounded body - // from a faulty or hostile receiver must not be buffered into memory. + // 'stream' stops axios buffering; destroying it means not a byte is read, + // so a hostile or unbounded body costs neither memory nor time. expect(opts.responseType).toBe('stream'); - expect(opts.maxBodyLength).toEqual(expect.any(Number)); + expect(body.destroyed).toBe(true); }); it('sizes the request cap in UTF-8 bytes, so non-ASCII mail is not rejected', async () => { @@ -282,23 +284,29 @@ describe('delivery result', () => { expect(caught.message).toContain('ECONNREFUSED'); }); - it('gives up on a response stream that never closes, instead of hanging', async () => { - // axios' timeout covers the headers; with responseType 'stream' it has - // already resolved. Without a deadline this await hangs, the queue row - // stays pending, and the next pass sends the same email again. + it('cannot hang on a response stream that never closes', async () => { + // The body is never read, so there is nothing to wait on. This used to + // need a deadline: without one the await hung, the queue row stayed + // pending, and the next pass sent the same email again. const { PassThrough } = require('stream'); const neverEnds = new PassThrough(); // written to by nobody, never ended axios.post.mockResolvedValue({ status: 200, data: neverEnds }); - transport.__testing.setReadTimeout(25); const result = await transport.send(MAIL); expect(result.messageId).toEqual(expect.any(String)); - // The stream is torn down rather than left dangling. expect(neverEnds.destroyed).toBe(true); }); - it('prefers a messageId the receiver reports', async () => { - axios.post.mockResolvedValue({ status: 200, data: streamOf(JSON.stringify({ messageId: 'from-n8n-123' })) }); - expect((await transport.send(MAIL)).messageId).toBe('from-n8n-123'); + it('survives a stream that errors on destroy', async () => { + // destroy() can emit on a socket-backed stream, and an unhandled 'error' + // on a stream throws. The listener attached before destroying is what + // stops a receiver's teardown from failing a delivered send. + const { PassThrough } = require('stream'); + const angry = new PassThrough(); + const originalDestroy = angry.destroy.bind(angry); + angry.destroy = () => { angry.emit('error', new Error('socket hang up')); originalDestroy(); }; + axios.post.mockResolvedValue({ status: 200, data: angry }); + + await expect(transport.send(MAIL)).resolves.toBeTruthy(); }); }); diff --git a/backend/src/services/emailWebhookTransport.js b/backend/src/services/emailWebhookTransport.js index 90989834..28703c15 100644 --- a/backend/src/services/emailWebhookTransport.js +++ b/backend/src/services/emailWebhookTransport.js @@ -29,10 +29,6 @@ const { validateExternalUrlAsync } = require('../utils/networkValidation'); const SIGNATURE_HEADER = 'X-PicPeak-Signature'; const HTTP_TIMEOUT_MS = 15000; -// Wall clock for reading the response body. Separate from HTTP_TIMEOUT_MS, -// which axios applies to the headers only. Mutable so a test can shrink it -// rather than actually waiting ten seconds. -let READ_TIMEOUT_MS = 10000; // Attachments are base64 in the JSON body, which inflates them by a third. // Invoices and quotes are the real users of this and run to a few hundred KB; @@ -141,57 +137,6 @@ async function encodeAttachments(attachments) { return encoded; } -/** - * Read at most MAX_RESPONSE_BYTES from a response stream, then stop. - * - * Only a messageId is wanted, so the rest is dropped on the floor rather than - * buffered — a faulty or hostile receiver must not be able to grow this - * process's memory on every queue attempt. Any read failure resolves empty: - * the delivery verdict is the status code, which is already known by the time - * this runs, so a broken body must never turn a delivered message into a retry. - */ -function readBounded(stream) { - const MAX_RESPONSE_BYTES = 10 * 1024; - if (!stream || typeof stream.on !== 'function') return Promise.resolve(''); - return new Promise((resolve) => { - const chunks = []; - let size = 0; - let done = false; - // A deadline, not just a size cap. axios' `timeout` covers the response - // HEADERS; with responseType 'stream' it has already resolved by the time - // we get here, so a receiver that answers 2xx and then never closes its - // body — or trickles bytes forever — would leave this await hanging. The - // queue row would stay pending and the next processor pass would POST the - // same message again, so an unclosed stream becomes duplicate email. - const timer = setTimeout(() => { - stream.destroy(); - finish(); - }, READ_TIMEOUT_MS); - // Node's unref keeps a hung read from holding the process open at exit. - if (typeof timer.unref === 'function') timer.unref(); - - function finish() { - if (done) return; - done = true; - clearTimeout(timer); - resolve(Buffer.concat(chunks).toString('utf8')); - } - - stream.on('data', (chunk) => { - size += chunk.length; - if (size <= MAX_RESPONSE_BYTES) { - chunks.push(chunk); - } else { - stream.destroy(); - finish(); - } - }); - stream.on('end', finish); - stream.on('error', finish); - stream.on('close', finish); - }); -} - /** * Recipients as a flat list of single addresses. * @@ -304,21 +249,29 @@ async function send(mail) { throw new Error(`email webhook request failed (${detail})`); } - // Status first: it is the delivery verdict, and it is known before a single - // byte of the body is read. - const delivered = response.status >= 200 && response.status < 300; - const body = await readBounded(response.data); - if (!delivered) { + // The body is discarded unread. Only the status matters — it is the delivery + // verdict — and the id below is synthesised either way. + // + // Reading it used to be worth 41 lines of bounded-read-with-deadline, to + // recover a messageId a receiver MIGHT return. That value was only ever + // logged: nothing persists it, there is no email_queue.message_id column. It + // was not worth its bugs — the size cap made a delivered message retry, and + // the missing deadline let an unclosed stream hang the queue and resend. Not + // reading is how those stop being reachable rather than defended against. + // + // An error listener first: destroy() can emit on a socket-backed stream, and + // an unhandled 'error' on a stream throws. + if (response.data && typeof response.data.destroy === 'function') { + response.data.on('error', () => {}); + response.data.destroy(); + } + + if (response.status < 200 || response.status >= 300) { throw new Error(`email webhook returned ${response.status}`); } - // The queue stores a messageId for the record. There is no SMTP id here, so - // synthesise one that is obviously not from a mail server. - let reported = null; - try { - reported = body ? JSON.parse(body).messageId : null; - } catch { /* a receiver is not obliged to answer JSON */ } - return { messageId: reported || `webhook-${signature.slice(0, 16)}` }; + // Deterministic, traceable, and obviously not from a mail server. + return { messageId: `webhook-${signature.slice(0, 16)}` }; } module.exports = { @@ -330,7 +283,6 @@ module.exports = { MAX_ATTACHMENT_BYTES, encodeAttachments, setAllowPrivateUrls(value) { allowPrivateUrls = !!value; }, - setReadTimeout(ms) { READ_TIMEOUT_MS = ms; }, resetSecretWarning() { warnedAboutMissingSecret = false; }, }, };