diff --git a/src/runtime/socket/UpgradedDuplex.rs b/src/runtime/socket/UpgradedDuplex.rs index 6782c6c1b5ee..c64cb4acb3b2 100644 --- a/src/runtime/socket/UpgradedDuplex.rs +++ b/src/runtime/socket/UpgradedDuplex.rs @@ -64,9 +64,8 @@ pub(crate) struct UpgradedDuplex { /// Replayed by [`Self::drain_pending`] after the staged bytes, preserving /// the original data-then-EOF order. pub pending_end: Cell, - /// 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, + /// [`Self::pause_stream`] called `origin.pause()`; [`Self::on_close`] undoes it. + pub reads_paused: Cell, } bun_event_loop::impl_timer_owner!(UpgradedDuplex; from_timer_ptr => event_loop_timer); @@ -209,6 +208,10 @@ impl UpgradedDuplex { js_wrapper.ensure_still_alive(); (this.handlers.on_close)(this.handlers.ctx); + // Left paused, a net.Socket transport never reads its peer's FIN and stays open. + if this.reads_paused.get() { + this.resume_stream(); + } // closes the underlying duplex this.call_write_or_end(None, false); @@ -217,6 +220,51 @@ impl UpgradedDuplex { js_wrapper.ensure_still_alive(); } + /// node's `JSStreamSocket.readStop()`: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/js_stream_socket.js#L117-L125 + #[uws_callback(export = "UpgradedDuplex__pause_stream")] + pub(crate) fn pause_stream(&self) -> bool { + // Before `start_tls` the handshake still needs the reads, and `on_open` clears the owner's paused flag. + if self.wrapper_ref().is_none() { + return false; + } + // Set first and kept on failure: `pause()` is user code that can close this socket, or throw after it paused. + self.reads_paused.set(true); + self.call_origin("pause") + } + + #[uws_callback(export = "UpgradedDuplex__resume_stream")] + pub(crate) fn resume_stream(&self) -> bool { + if !self.call_origin("resume") { + return false; + } + self.reads_paused.set(false); + true + } + + /// Calls `origin[name]()`. A throw goes to `on_error`; false when the call did not complete. + fn call_origin(&self, name: &str) -> bool { + let duplex = self.origin.get(); + if duplex.is_empty() { + return false; + } + let Some(global) = self.global else { + return false; + }; + let method = match duplex.get(&global, name) { + Ok(Some(f)) if f.is_callable() => f, + Ok(_) => return false, + Err(err) => { + (self.handlers.on_error)(self.handlers.ctx, global.take_error(err)); + return false; + } + }; + if let Err(err) = method.call(&global, duplex, &[]) { + (self.handlers.on_error)(self.handlers.ctx, global.take_error(err)); + return false; + } + true + } + fn call_write_or_end(&self, data: Option<&[u8]>, msg_more: bool) { // No JS duplex to talk to: the zeroed placeholder, or the owning // socket's finalizer abandoned it (`abandon_js_side`). @@ -239,8 +287,15 @@ impl UpgradedDuplex { // throws writeAfterFIN (EPIPE). The trailing end() is not a write // and still goes through the writableEnded probe below, so a // half-open transport sees our FIN. - if data.is_some() && self.transport_eof.get() { - return; + if data.is_some() { + match Self::readable_got_eof(duplex, &global) { + Ok(false) => {} + Ok(true) => return, + Err(err) => { + (self.handlers.on_error)(self.handlers.ctx, global.take_error(err)); + return; + } + } } match duplex.get(&global, "writableEnded") { Ok(Some(ended)) if ended.to_boolean() => return, @@ -276,6 +331,19 @@ impl UpgradedDuplex { } } + /// `_readableState.ended`, not the 'end' event: a paused transport holds 'end' back. + fn readable_got_eof(duplex: JSValue, global: &JSGlobalObject) -> JsResult { + let Some(state) = duplex.get(global, "_readableState")? else { + return Ok(false); + }; + if !state.is_object() { + return Ok(false); + } + Ok(state + .get(global, "ended")? + .is_some_and(|ended| ended.to_boolean())) + } + fn internal_write(this: *mut Self, encoded_data: &[u8]) { // SAFETY: see handler note above. unsafe { &*this }.write_encrypted(encoded_data); @@ -408,7 +476,7 @@ impl UpgradedDuplex { current_timeout: Cell::new(0), pending_data: JsCell::new(Vec::new()), pending_end: Cell::new(false), - transport_eof: Cell::new(false), + reads_paused: Cell::new(false), } } @@ -679,7 +747,7 @@ impl UpgradedDuplex { self.ssl_error.set(CertError::default()); self.pending_data.set(Vec::new()); self.pending_end.set(false); - self.transport_eof.set(false); + self.reads_paused.set(false); } } @@ -738,7 +806,6 @@ fn on_end(_global: &JSGlobalObject, frame: &CallFrame) -> JsResult { // SAFETY: see host-fn note above. let this = unsafe { &*self_ptr.cast::() }; - this.transport_eof.set(true); if this.wrapper_ref().is_some() { (this.handlers.on_end)(this.handlers.ctx); } else { diff --git a/src/uws_sys/lib.rs b/src/uws_sys/lib.rs index ad9170bca813..f0ff0b1338ab 100644 --- a/src/uws_sys/lib.rs +++ b/src/uws_sys/lib.rs @@ -221,6 +221,8 @@ unsafe extern "C" { safe fn UpgradedDuplex__shutdown_read(this: &mut UpgradedDuplex); safe fn UpgradedDuplex__close(this: &mut UpgradedDuplex); safe fn UpgradedDuplex__abandon_js_side(this: &mut UpgradedDuplex); + safe fn UpgradedDuplex__pause_stream(this: &mut UpgradedDuplex) -> bool; + safe fn UpgradedDuplex__resume_stream(this: &mut UpgradedDuplex) -> bool; } impl UpgradedDuplex { #[inline] @@ -282,6 +284,14 @@ impl UpgradedDuplex { pub(crate) fn abandon_js_side(&mut self) { UpgradedDuplex__abandon_js_side(self) } + #[inline] + pub(crate) fn pause_stream(&mut self) -> bool { + UpgradedDuplex__pause_stream(self) + } + #[inline] + pub(crate) fn resume_stream(&mut self) -> bool { + UpgradedDuplex__resume_stream(self) + } } // ── WindowsNamedPipe (cycle-break shim) ───────────────────────────────────── diff --git a/src/uws_sys/socket.rs b/src/uws_sys/socket.rs index 016e77932502..7e85301b21c9 100644 --- a/src/uws_sys/socket.rs +++ b/src/uws_sys/socket.rs @@ -523,7 +523,7 @@ impl NewSocketHandler { connected s => if s.is_established() { s.pause(); true } else { false }, connecting _c => false, detached => true, - duplex _d => false, // TODO: pause/resume upgraded duplex + duplex d => d.pause_stream(), pipe p => p.pause_stream(), ) } @@ -533,7 +533,7 @@ impl NewSocketHandler { connected s => if s.is_established() { s.resume(); true } else { false }, connecting _c => false, detached => true, - duplex _d => false, // TODO: pause/resume upgraded duplex + duplex d => d.resume_stream(), pipe p => p.resume_stream(), ) } diff --git a/test/js/node/tls/node-tls-connect.test.ts b/test/js/node/tls/node-tls-connect.test.ts index cbb1c842a4ba..0310b3779dcc 100644 --- a/test/js/node/tls/node-tls-connect.test.ts +++ b/test/js/node/tls/node-tls-connect.test.ts @@ -843,6 +843,269 @@ it("a client and a server TLSSocket connected through a synchronous in-memory du }); }); +describe("a TLS socket over a Duplex transport reads it with backpressure", () => { + // Node reads such a transport through a JSStreamSocket, whose readStop() and + // readStart() pause and resume it, so it only flows while the TLS socket + // takes more data: + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/js_stream_socket.js#L117-L125 + // Each side's _write pushes straight into the other side and never waits, + // so only the reading TLS socket can slow the transport down. + function inMemoryPair() { + 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 serverContext = () => ({ isServer: true, secureContext: tls.createSecureContext(COMMON_CERT_) }); + const payload = Buffer.alloc(1024 * 1024, "x"); + // Rejects with the first 'error' any of `sockets` emits. Meant to be raced. + function firstErrorOf(...sockets: TLSSocket[]) { + const { promise, reject } = Promise.withResolvers(); + promise.catch(() => {}); + for (const socket of sockets) socket.on("error", reject); + return promise; + } + // Destroys every socket in `sockets` when the test leaves its scope. + function destroyedOnExit(sockets: { destroy(): unknown }[]) { + return { + [Symbol.dispose]() { + for (const socket of sockets) socket.destroy(); + }, + }; + } + + // Both wraps run the same engine; the reader is the side under test. + function securePair(reader: "client" | "server") { + const { clientSide, serverSide } = inMemoryPair(); + const server = new TLSSocket(serverSide, serverContext()); + const client = tls.connect({ socket: clientSide, rejectUnauthorized: false }); + const failed = firstErrorOf(server, client); + const secured = Promise.all([once(server, "secure"), once(client, "secureConnect")]); + const [readerSocket, transport, writerSocket] = + reader === "client" ? [client, clientSide, server] : [server, serverSide, client]; + return { readerSocket, transport, writerSocket, failed, secured, cleanup: destroyedOnExit([client, server]) }; + } + + describe.each(["client", "server"] as const)("%s reader", reader => { + // The writer does not end: what happens to unread data once the peer has + // closed is a separate matter from how much the socket takes in. + it("a paused socket pauses the transport once its buffer is full", async () => { + const { readerSocket, transport, writerSocket, failed, secured, cleanup } = securePair(reader); + using _ = cleanup; + await Promise.race([secured, failed]); + readerSocket.pause(); + // The pair is synchronous, so the callback runs with every byte handed + // to the reader's transport. + const written = Promise.withResolvers(); + writerSocket.write(payload, err => (err ? written.reject(err) : written.resolve())); + await Promise.race([written.promise, failed]); + // Without backpressure the transport keeps flowing and the paused socket + // holds the whole payload. + expect({ + transportFlowing: transport.readableFlowing, + socketIsFull: readerSocket.readableLength >= readerSocket.readableHighWaterMark, + mostOfItWaitsInTheTransport: transport.readableLength > payload.length / 2, + }).toEqual({ transportFlowing: false, socketIsFull: true, mostOfItWaitsInTheTransport: true }); + + const chunks: Buffer[] = []; + let total = 0; + const received = Promise.withResolvers(); + readerSocket.on("data", (chunk: Buffer) => { + chunks.push(chunk); + if ((total += chunk.length) >= payload.length) received.resolve(); + }); + readerSocket.resume(); + await Promise.race([received.promise, failed]); + expect(Buffer.concat(chunks).equals(payload)).toBe(true); + }); + + it("a slow reader gets the whole payload while the transport is paused and resumed under it", async () => { + const { readerSocket, transport, writerSocket, failed, secured, cleanup } = securePair(reader); + using _ = cleanup; + await Promise.race([secured, failed]); + let pauses = 0; + transport.on("pause", () => pauses++); + writerSocket.write(payload); + + const read = (async () => { + let total = 0; + let mostHeld = 0; + for await (const chunk of readerSocket) { + total += chunk.length; + mostHeld = Math.max(mostHeld, readerSocket.readableLength); + if (total >= payload.length) break; + // Yield a macrotask per chunk so the socket's buffer fills up. + await new Promise(resolve => setImmediate(resolve)); + } + return { total, heldLessThanHalfAtAnyTime: mostHeld < payload.length / 2 }; + })(); + expect({ ...(await Promise.race([read, failed])), pausedMoreThanOnce: pauses > 1 }).toEqual({ + total: payload.length, + heldLessThanHalfAtAnyTime: true, + pausedMoreThanOnce: true, + }); + }); + }); + + it("does not answer a close_notify that it reads after the transport's EOF", async () => { + // A net.Socket that has read its peer's FIN fails a later write with EPIPE + // (writeAfterFIN in net.ts), and so does this transport. A paused reader + // makes the engine read the peer's close_notify long after that EOF, and + // the transport's 'end' event, held back by the pause, comes later still. + const transportErrors: string[] = []; + const gotEOF = new WeakSet(); + const makeSide = (peer: () => Duplex) => + new Duplex({ + read() {}, + write(chunk, _encoding, callback) { + if (gotEOF.has(this)) + return callback(Object.assign(new Error("write after the peer's FIN"), { code: "EPIPE" })); + peer().push(chunk); + callback(); + }, + final(callback) { + gotEOF.add(peer()); + peer().push(null); + callback(); + }, + }); + const clientSide: Duplex = makeSide(() => serverSide); + const serverSide: Duplex = makeSide(() => clientSide); + clientSide.on("error", (err: NodeJS.ErrnoException) => transportErrors.push(`${err.code}`)); + const server = new TLSSocket(serverSide, serverContext()); + const client = tls.connect({ socket: clientSide, rejectUnauthorized: false }); + using _ = destroyedOnExit([client, server]); + const failed = firstErrorOf(server, client); + await Promise.race([Promise.all([once(server, "secure"), once(client, "secureConnect")]), failed]); + + client.pause(); + // Like a TLS server over TCP: the payload, the close_notify, then the FIN. + server.end(payload); + await Promise.race([once(server, "finish"), failed]); + serverSide.end(); + + let total = 0; + client.on("data", (chunk: Buffer) => (total += chunk.length)); + const closed = once(client, "close"); + client.resume(); + await Promise.race([closed, failed]); + expect({ total, transportErrors }).toEqual({ total: payload.length, transportErrors: [] }); + }); + + it("a pause() issued before the engine exists does not stall the handshake", async () => { + // The engine is created on a later event-loop turn. A transport paused + // ahead of it would never deliver the server's flight. + const { clientSide, serverSide } = inMemoryPair(); + const server = new TLSSocket(serverSide, serverContext()); + const client = tls.connect({ + socket: clientSide, + rejectUnauthorized: false, + // Only an onread socket stops its handle from pause(). + onread: { buffer: Buffer.alloc(64), callback: () => {} }, + }); + using _ = destroyedOnExit([client, server]); + client.pause(); + const secured = Promise.all([once(server, "secure"), once(client, "secureConnect")]); + await Promise.race([secured, firstErrorOf(server, client)]); + expect({ server: server.getProtocol(), client: client.getProtocol() }).toEqual({ + server: "TLSv1.3", + client: "TLSv1.3", + }); + }); + + describe("a server wrap over a net.Socket with unflushed plain writes", () => { + // Such a wrap cannot take the fd over, so it reads the net.Socket like any + // other Duplex. Nothing destroys that net.Socket along with the TLS + // socket, so one left paused would never read the client's FIN and would + // stay open for good. + async function connectedWrap() { + const sockets: { destroy(): unknown }[] = []; + const accepted = Promise.withResolvers<{ transport: net.Socket; wrapped: TLSSocket }>(); + const listener = net.createServer(transport => { + transport.cork(); + transport.write("!"); + const wrapped = new TLSSocket(transport, serverContext()); + transport.uncork(); + sockets.push(wrapped, transport); + accepted.resolve({ transport, wrapped }); + }); + // close() waits for the connections, so they go first. + const dispose = async () => { + for (const socket of sockets) socket.destroy(); + await listener[Symbol.asyncDispose](); + }; + try { + await once(listener.listen(0, "127.0.0.1"), "listening"); + const clientTransport = net.connect((listener.address() as AddressInfo).port, "127.0.0.1"); + sockets.push(clientTransport); + const [{ transport, wrapped }, [greeting]] = await Promise.all([ + accepted.promise, + once(clientTransport, "data"), + ]); + expect(String(greeting)).toBe("!"); + const client = tls.connect({ socket: clientTransport, rejectUnauthorized: false }); + sockets.push(client); + const failed = firstErrorOf(wrapped, client); + await Promise.race([Promise.all([once(wrapped, "secure"), once(client, "secureConnect")]), failed]); + return { transport, wrapped, client, failed, [Symbol.asyncDispose]: dispose }; + } catch (err) { + await dispose(); + throw err; + } + } + + it("destroying the paused socket lets the transport read its peer's close", async () => { + await using wrap = await connectedWrap(); + const { transport, wrapped, client, failed } = wrap; + wrapped.pause(); + // Without backpressure the engine takes the whole payload and the + // client's close_notify, answers it, and the client closes. + const outcome = Promise.race([ + once(transport, "pause").then(() => "transport paused"), + once(client, "close").then(() => "client closed"), + failed, + ]); + client.end(payload); + expect(await outcome).toBe("transport paused"); + + const closed = once(transport, "close"); + wrapped.destroy(); + await Promise.race([closed, failed]); + expect(transport.destroyed).toBe(true); + }); + + it("a destroy from the transport's 'pause' listener does the same", async () => { + await using wrap = await connectedWrap(); + const { transport, wrapped, client, failed } = wrap; + wrapped.pause(); + // The engine is still inside the transport's pause() when the socket closes. + let destroyedFromPause = false; + transport.once("pause", () => { + destroyedFromPause = true; + wrapped.destroy(); + }); + const closed = once(transport, "close"); + client.end(payload); + await Promise.race([closed, failed]); + expect({ destroyedFromPause, transportDestroyed: transport.destroyed }).toEqual({ + destroyedFromPause: true, + transportDestroyed: true, + }); + }); + }); +}); + describe("application data written over a Duplex transport before the handshake completes", () => { // Node parks such a write (TLSWrap's pending cleartext) and sends it right // after the handshake: the write is still pending when 'secureConnect' /