node:tls: keep unread data and still emit 'close' when a wrapped transport closes

src/js/node/net.ts

@@ -2019,10 +2019,7 @@ Socket.prototype.connect = function connect(...args) {
             tls,
             socket: this[khandlers],
           });
-          connection.on("data", events[0]);
-          connection.on("end", events[1]);
-          connection.on("drain", events[2]);
-          connection.on("close", events[3]);
+          attachUpgradedDuplex(this, connection, events);
           this._handle = result;
         } else {
           // upgradeTLS requires an established socket; a socket that is still
@@ -2070,10 +2067,7 @@ Socket.prototype.connect = function connect(...args) {
                   tls,
                   socket: this[khandlers],
                 });
-                connection.on("data", events[0]);
-                connection.on("end", events[1]);
-                connection.on("drain", events[2]);
-                connection.on("close", events[3]);
+                attachUpgradedDuplex(this, connection, events);
                 this._handle = result;
               } else {
                 this[kupgraded] = connection;
@@ -2303,6 +2297,23 @@ function hasUnflushedWrites(connection) {
   return connection.writableLength > 0 || connection[kwriteCallback] != null;
 }
 
+// `events` are the stream-level TLS engine's data/end/drain/close thunks.
+// https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L739-L741
+function attachUpgradedDuplex(self, connection, events) {
+  // Ahead of events[3], so the engine's close finds the socket already
+  // destroyed and reports neither the aborted handshake nor a second 'close'.
+  connection.on("close", () => {
+    // The engine reads the transport without backpressure, so after a clean
+    // EOF the transport can close while decrypted data is still unread. The
+    // stream's own end finishes the socket then; a destroy would drop the data.
+    if (!self[kended] && !self[kOnreadPendingEnd]) self.destroy();
+  });
+  connection.on("data", events[0]);
+  connection.on("end", events[1]);
+  connection.on("drain", events[2]);
+  connection.on("close", events[3]);
+}
+
 function drainOnreadTail(self, fromRead?) {
   if (self[kOnreadTail] === undefined) return false;
   if (fromRead) self[kOnreadReadRequested] = true;
@@ -2372,10 +2383,7 @@ Socket.prototype[Symbol.for("::bunUpgradeServerTLS::")] = function (connection,
       socket: serverHandlersFor(this),
       isServer: true,
     });
-    connection.on("data", events[0]);
-    connection.on("end", events[1]);
-    connection.on("drain", events[2]);
-    connection.on("close", events[3]);
+    attachUpgradedDuplex(this, connection, events);
     this[kupgraded] = connection;
     this._handle = result;
     return;
@@ -2401,10 +2409,7 @@ Socket.prototype[Symbol.for("::bunUpgradeServerTLS::")] = function (connection,
         socket: serverHandlersFor(this),
         isServer: true,
       });
-      connection.on("data", events[0]);
-      connection.on("end", events[1]);
-      connection.on("drain", events[2]);
-      connection.on("close", events[3]);
+      attachUpgradedDuplex(this, connection, events);
       this._handle = result;
       this.emit(kUpgradeAttached);
       return;

src/runtime/socket/UpgradedDuplex.rs

@@ -64,6 +64,9 @@ pub(crate) struct UpgradedDuplex {
     /// Replayed by [`Self::drain_pending`] after the staged bytes, preserving
     /// the original data-then-EOF order.
     pub pending_end: Cell<bool>,
+    /// The transport closed before the TLS engine existed (same window as
+    /// [`Self::pending_data`]). Consumed by the queued `StartTLS` task.
+    pub pending_close: Cell<bool>,
     /// The transport delivered EOF (its 'end' event fired). Teardown payloads
     /// (close_notify) are dropped after this; see [`Self::call_write_or_end`].
     pub transport_eof: Cell<bool>,
@@ -408,6 +411,7 @@ impl UpgradedDuplex {
             current_timeout: Cell::new(0),
             pending_data: JsCell::new(Vec::new()),
             pending_end: Cell::new(false),
+            pending_close: Cell::new(false),
             transport_eof: Cell::new(false),
         }
     }
@@ -542,9 +546,12 @@ impl UpgradedDuplex {
 
     #[uws_callback(export = "UpgradedDuplex__close")]
     pub(crate) fn close(&self) {
-        if let Some(w) = self.wrapper_ref() {
-            let _ = w.shutdown(true);
-        }
+        let Some(w) = self.wrapper_ref() else {
+            // `start_tls` is still queued: no SSL, and no close callback yet.
+            self.pending_close.set(true);
+            return;
+        };
+        let _ = w.shutdown(true);
     }
 
     #[uws_callback(export = "UpgradedDuplex__shutdown")]
@@ -679,6 +686,7 @@ impl UpgradedDuplex {
         self.ssl_error.set(CertError::default());
         self.pending_data.set(Vec::new());
         self.pending_end.set(false);
+        self.pending_close.set(false);
         self.transport_eof.set(false);
     }
 }

src/runtime/socket/socket_body.rs

@@ -4399,6 +4399,12 @@ impl DuplexUpgradeContext {
                     unsafe { Self::deinit(this.as_ptr()) };
                     return;
                 }
+                // The transport closed while this task was queued: an engine
+                // started now could never handshake, and nothing would free it.
+                if this.upgrade.pending_close.replace(false) {
+                    Self::on_close(this);
+                    return;
+                }
                 log!(
                     "DuplexUpgradeContext.startTLS mode={}",
                     <&'static str>::from(this.mode)

test/js/node/tls/node-tls-connect.test.ts

@@ -1031,6 +1031,165 @@ describe("application data written over a Duplex transport before the handshake
   });
 });
 
+describe("a TLS socket over a Duplex transport follows that transport's teardown", () => {
+  // Node wraps a stream that has no fd in a JSStreamSocket and destroys the
+  // TLS socket when that wrap closes:
+  // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L739-L741
+  // So a destroyed transport always reaches the TLS socket as 'close' with
+  // hadError === false. Bun runs a TLS engine over the stream instead, and
+  // that engine is created on a later event-loop turn, so the cases below
+  // cover both sides of that turn.
+  const makeTransport = () =>
+    new Duplex({
+      read() {},
+      write(_chunk, _encoding, callback) {
+        callback();
+      },
+    });
+  const serverContext = () => ({ isServer: true, secureContext: tls.createSecureContext(COMMON_CERT_) });
+  // Resolves with the 'error' and 'close' events the socket emitted, in order.
+  const recordTeardown = (socket: TLSSocket) => {
+    const events: string[] = [];
+    const closed = Promise.withResolvers<string[]>();
+    socket.on("error", (err: NodeJS.ErrnoException) => events.push(`error:${err.code}`));
+    socket.on("close", hadError => {
+      events.push(`close:${hadError}`);
+      closed.resolve(events);
+    });
+    return closed.promise;
+  };
+
+  it("a server wrap closes when the transport is destroyed in the same tick", async () => {
+    // The engine does not exist yet, so the close has to be held and reported
+    // when it starts. Without that the socket never closes at all.
+    const transport = makeTransport();
+    const wrapped = new TLSSocket(transport, serverContext());
+    const teardown = recordTeardown(wrapped);
+    transport.destroy();
+    expect(await teardown).toEqual(["close:false"]);
+    expect(wrapped.destroyed).toBe(true);
+  });
+
+  it("tls.connect({ socket }) closes when the transport is destroyed in the same tick", async () => {
+    const transport = makeTransport();
+    const client = tls.connect({ socket: transport, rejectUnauthorized: false });
+    const teardown = recordTeardown(client);
+    transport.destroy();
+    expect(await teardown).toEqual(["close:false"]);
+    expect(client.destroyed).toBe(true);
+  });
+
+  it("a server wrap closes when the transport is destroyed after the engine started", async () => {
+    // The engine is up and its handshake is still pending. The aborted
+    // handshake must not surface as an ECONNRESET instead of the close.
+    const transport = makeTransport();
+    const wrapped = new TLSSocket(transport, serverContext());
+    const teardown = recordTeardown(wrapped);
+    // 'secure' never fires without a peer, but the upgrade task that creates
+    // the engine is queued ahead of this one.
+    await new Promise<void>(resolve => setImmediate(resolve));
+    transport.destroy();
+    expect(await teardown).toEqual(["close:false"]);
+    expect(wrapped.destroyed).toBe(true);
+  });
+
+  it("a server wrap closes when the transport of an established connection is destroyed", async () => {
+    // Each side's _write pushes into the other, so this handshakes for real
+    // over two in-memory duplexes (what Playwright's client-certificate proxy
+    // does with its browser-facing transport).
+    const makeSide = (peer: () => Duplex) =>
+      new Duplex({
+        read() {},
+        write(chunk, _encoding, callback) {
+          peer().push(chunk);
+          callback();
+        },
+        final(callback) {
+          peer().push(null);
+          callback();
+        },
+      });
+    const clientSide: Duplex = makeSide(() => serverSide);
+    const serverSide: Duplex = makeSide(() => clientSide);
+    const wrapped = new TLSSocket(serverSide, serverContext());
+    const client = tls.connect({ socket: clientSide, rejectUnauthorized: false });
+    try {
+      const secured = Promise.withResolvers<void>();
+      client.on("error", secured.reject);
+      wrapped.on("error", secured.reject);
+      wrapped.on("secure", () => secured.resolve());
+      await secured.promise;
+      const teardown = recordTeardown(wrapped);
+      serverSide.destroy();
+      expect(await teardown).toEqual(["close:false"]);
+      expect(wrapped.destroyed).toBe(true);
+    } finally {
+      client.destroy();
+      wrapped.destroy();
+    }
+  });
+
+  // The engine reads the transport with no backpressure, so a peer that ends
+  // cleanly leaves the whole payload decrypted inside the TLS socket while its
+  // transport is already closing. A consumer that is not flowing reads it
+  // after that close, so the close must not destroy the socket there.
+  describe.each(["paused", "for-await"] as const)("a clean peer close keeps unread data (%s reader)", reader => {
+    it("delivers the whole payload and 'end'", async () => {
+      const pair = () => {
+        const makeSide = (peer: () => Duplex) =>
+          new Duplex({
+            read() {},
+            write(chunk, _encoding, callback) {
+              peer().push(chunk);
+              callback();
+            },
+            final(callback) {
+              peer().push(null);
+              callback();
+            },
+          });
+        const clientSide: Duplex = makeSide(() => serverSide);
+        const serverSide: Duplex = makeSide(() => clientSide);
+        return { clientSide, serverSide };
+      };
+      const { clientSide, serverSide } = pair();
+      const payload = Buffer.alloc(256 * 1024, "bun");
+      const server = new TLSSocket(serverSide, serverContext());
+      server.on("secure", () => server.end(payload));
+      const client = tls.connect({ socket: clientSide, rejectUnauthorized: false });
+      const failed = Promise.withResolvers<never>();
+      server.on("error", failed.reject);
+      client.on("error", failed.reject);
+
+      const read = (async () => {
+        if (reader === "for-await") {
+          let total = 0;
+          for await (const chunk of client) {
+            total += chunk.length;
+            // Yield, so the transport's close lands between two reads.
+            await new Promise<void>(resolve => setImmediate(resolve));
+          }
+          return total;
+        }
+        client.pause();
+        // Resume only once the transport is gone, which is the case this
+        // covers: the socket holds every decrypted byte at that point.
+        await once(clientSide, "close");
+        let total = 0;
+        client.on("data", chunk => (total += chunk.length));
+        const ended = once(client, "end");
+        client.resume();
+        await ended;
+        return total;
+      })();
+
+      expect(await Promise.race([read, failed.promise])).toBe(payload.length);
+      client.destroy();
+      server.destroy();
+    });
+  });
+});
+
 it("delivers 'session' even when the data handler destroys the socket immediately", async () => {
   // The TLS1.3 NewSessionTickets ride in the same read pass as the response
   // bytes. If the parked session were only flushed after the data dispatch,