From 44a8c416c751aa98d8a9aeba510606880a0b2a24 Mon Sep 17 00:00:00 2001 From: Paul Nothaft <53005142+the-luap@users.noreply.github.com> Date: Sat, 29 Aug 2026 13:06:19 +0200 Subject: [PATCH] refactor(email): stop reading the webhook response body at all (#1225) (#1239) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transport carried 41 lines of bounded-read-with-deadline to recover a messageId a receiver MIGHT return. That value is only ever logged — nothing persists it, there is no email_queue.message_id column — and the code to get it produced two of the last four review findings: the size cap made a DELIVERED message retry (axios throws while reading), and the missing deadline let an unclosed stream hang the queue and resend. Not reading the body is how that whole class stops being reachable rather than defended against. responseType 'stream' still keeps axios from buffering; the stream is destroyed immediately and the id is synthesised as before. The status was always the delivery verdict, and it is known before any of this. An 'error' listener goes on before destroy(): destroy can emit on a socket-backed stream, and an unhandled 'error' on a stream throws — which would have turned a receiver's teardown into a failed send. Net: -45 lines of service code, one fewer constant, one fewer test seam, and three of the hardest cases in the suite replaced by two simpler ones. 23 tests, 63 across the email suites, eslint clean. Co-authored-by: Paul Nothaft --- .../services/emailWebhookTransport.test.js | 36 +++++--- backend/src/services/emailWebhookTransport.js | 88 +++++-------------- 2 files changed, 42 insertions(+), 82 deletions(-) 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; }, }, };