From f652e80efc598c4fc28bb581a4b9e2d6c5fba68e Mon Sep 17 00:00:00 2001 From: productdevbook Date: Wed, 2 Sep 2026 02:10:10 +0200 Subject: [PATCH] fix(drivers): a socket dying with a full write buffer no longer hangs the send MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #131. `sendInternal` waited for backpressure to clear with a promise that had one way to settle: the `drain` event. If the peer vanished while the buffer was still full, `drain` never fired and nothing else reached that promise — `failAll` only rejects waiters queued for a reply, and `commandTimeoutMs` belongs to those same waiters. `sendMessage()` never returned, the pooled connection was never released, and `withRetry` could not help because nothing had failed. Reproduced with a 4 MB message to a server that stops reading and then closes: the promise stayed pending indefinitely. It now races `drain` against `error` and `close`, removing all three listeners whichever wins, so a pooled connection does not accumulate one per write. Same shape as #117, and the second half of it: a promise with a single settlement path and a peer that can disappear before it is taken. While in there: `tls.connect({ socket })` leaves the pre-upgrade listeners on the socket underneath — measured, the `data` handler is still attached and the socket still flowing after wrapping. Anything delivered from there would be TLS records going into the reply parser as noise, and the old close handler would report the failure a second time. They are detached on upgrade now. Whether those bytes actually arrive depends on Node internals I could not exercise without a TLS-capable fake server, so this is written as hardening rather than as a fix for an observed failure. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GQAPDWLnDrdYHKhou3Ntvb --- src/drivers/_smtp/connection.ts | 71 +++++++++++++++++--------- test/drivers/smtp-backpressure.test.ts | 71 ++++++++++++++++++++++++++ 2 files changed, 119 insertions(+), 23 deletions(-) create mode 100644 test/drivers/smtp-backpressure.test.ts diff --git a/src/drivers/_smtp/connection.ts b/src/drivers/_smtp/connection.ts index 2664713..a307b46 100644 --- a/src/drivers/_smtp/connection.ts +++ b/src/drivers/_smtp/connection.ts @@ -71,17 +71,21 @@ export async function createConnection(opts: ConnectionOptions): Promise - parser.push(typeof chunk === "string" ? chunk : chunk.toString("utf8")), - ) - const onCloseReason = { value: undefined as Error | undefined } - socket.on("error", (err: Error) => failAll(pending, wrapNetworkError(err))) - socket.on("close", () => { - const err = onCloseReason.value ?? wrapNetworkError(new Error("connection closed")) - failAll(pending, err) - }) + + /** Point the reply parser and the failure handlers at a socket. */ + const listen = (target: Socket | TLSSocket) => { + target.setEncoding("utf8") + target.on("data", (chunk: string | Buffer) => + parser.push(typeof chunk === "string" ? chunk : chunk.toString("utf8")), + ) + target.on("error", (err: Error) => failAll(pending, wrapNetworkError(err))) + target.on("close", () => { + failAll(pending, onCloseReason.value ?? wrapNetworkError(new Error("connection closed"))) + }) + } + + listen(socket) const setup = async () => { await waitForConnect(socket, opts.connectionTimeoutMs, opts.secure) @@ -103,16 +107,18 @@ export async function createConnection(opts: ConnectionOptions): Promise - parser.push(typeof chunk === "string" ? chunk : chunk.toString("utf8")), - ) - socket.on("error", (err: Error) => failAll(pending, wrapNetworkError(err))) - socket.on("close", () => { - const err = onCloseReason.value ?? wrapNetworkError(new Error("connection closed")) - failAll(pending, err) - }) + const plaintext = socket as Socket + socket = await upgradeTls(plaintext, tls, opts) + // `tls.connect({ socket })` leaves our listeners on the socket + // underneath — measured: the `data` handler is still attached and the + // socket is still flowing. Anything it delivers from here on is TLS + // records, which would go into the reply parser as noise, and its + // close handler would report the failure a second time. Detach them + // and listen to the encrypted socket instead. + plaintext.removeAllListeners("data") + plaintext.removeAllListeners("error") + plaintext.removeAllListeners("close") + listen(socket) caps = await ehlo(pending, socket, opts.localName, opts.commandTimeoutMs) } else if (opts.requireTLS && !opts.secure) { throw new Error(`[unemail] [smtp] STARTTLS required but not offered by ${opts.host}`) @@ -207,9 +213,28 @@ function failAll(pending: PendingReply[], err: Error): void { async function sendInternal(socket: Socket | TLSSocket, line: string): Promise { const payload = `${line}\r\n` - if (!socket.write(payload)) { - await new Promise((resolve) => socket.once("drain", () => resolve())) - } + if (socket.write(payload)) return + + // The write buffer is full. Waiting for `drain` alone would never settle + // if the peer vanishes first: this promise is not a reply waiter, so + // `failAll` does not reach it and `commandTimeoutMs` does not cover it — + // the send would hang forever and the pooled connection never come back. + await new Promise((resolve, reject) => { + const done = (settle: () => void) => () => { + socket.removeListener("drain", onDrain) + socket.removeListener("error", onError) + socket.removeListener("close", onClose) + settle() + } + const onDrain = done(resolve) + const onError = (error: Error) => done(() => reject(wrapNetworkError(error)))() + const onClose = done(() => + reject(wrapNetworkError(new Error("connection closed while sending"))), + ) + socket.once("drain", onDrain) + socket.once("error", onError) + socket.once("close", onClose) + }) } function recvInternal( diff --git a/test/drivers/smtp-backpressure.test.ts b/test/drivers/smtp-backpressure.test.ts new file mode 100644 index 0000000..ff62b53 --- /dev/null +++ b/test/drivers/smtp-backpressure.test.ts @@ -0,0 +1,71 @@ +import { afterEach, describe, expect, it } from "vitest" +import net from "node:net" +import { createConnection } from "../../src/drivers/_smtp/connection.ts" + +let server: net.Server | null = null + +afterEach(() => { + server?.close() + server = null +}) + +/** + * Completes the handshake, then stops reading once DATA begins so the + * client's write buffer fills, and kills the connection while it is full. + */ +async function stallingServer(): Promise { + server = net.createServer((sock) => { + sock.setEncoding("utf8") + sock.write("220 fake ESMTP\r\n") + let inData = false + sock.on("data", (chunk: string) => { + if (inData) { + sock.pause() + setTimeout(() => sock.destroy(), 50) + return + } + for (const line of chunk.split("\r\n").filter(Boolean)) { + if (/^EHLO/i.test(line)) sock.write("250-fake\r\n250 SIZE 100000000\r\n") + else if (/^DATA/i.test(line)) { + sock.write("354 go\r\n") + inData = true + } else sock.write("250 ok\r\n") + } + }) + sock.on("error", () => {}) + }) + await new Promise((resolve) => server!.listen(0, "127.0.0.1", () => resolve())) + return (server!.address() as net.AddressInfo).port +} + +describe("a socket that dies while the write buffer is full", () => { + it("rejects rather than hanging", async () => { + const port = await stallingServer() + const conn = await createConnection({ + host: "127.0.0.1", + port, + secure: false, + authMethod: "AUTO", + rejectUnauthorized: true, + localName: "test.local", + connectionTimeoutMs: 2000, + commandTimeoutMs: 2000, + }) + + // Larger than the socket's 16 KB write buffer, so `write()` returns + // false and the send has to wait for a drain that never comes. + const body = `Subject: s\r\n\r\n${"x".repeat(4 * 1024 * 1024)}` + + const outcome = await Promise.race([ + conn.sendMessage({ from: "a@b.com", rcpt: ["c@d.com"] }, body).then( + () => "resolved", + (error: Error) => `rejected: ${error.message}`, + ), + new Promise((resolve) => setTimeout(() => resolve("hung"), 4000)), + ]) + + expect(outcome).not.toBe("hung") + expect(outcome).toMatch(/^rejected:/) + conn.destroy() + }, 15_000) +})