From a87aa2fafe05c358455173ea65877be5eefcbd63 Mon Sep 17 00:00:00 2001
From: Ciro Spaciari <ciro.spaciari@gmail.com>
Date: Sat, 14 Oct 2023 20:16:49 -0300
Subject: fix(net/tls) fix pg hang on end + hanging on query (#6487)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* fix pg hang on end + hanging on query

* remove dummy function

* fix node-stream

* add test

* fix test

* return error in test

* fix test use once instead of on

* fix OOM

* generated

* 💅

* 💅
---
 src/js/node/net.js | 24 ++++++++++++++++++++----
 1 file changed, 20 insertions(+), 4 deletions(-)

(limited to 'src/js/node/net.js')

diff --git a/src/js/node/net.js b/src/js/node/net.js
index fc0aba548..f8da33d53 100644
--- a/src/js/node/net.js
+++ b/src/js/node/net.js
@@ -66,6 +66,9 @@ const bunSocketServerOptions = Symbol.for("::bunnetserveroptions::");
 const bunSocketInternal = Symbol.for("::bunnetsocketinternal::");
 const bunTLSConnectOptions = Symbol.for("::buntlsconnectoptions::");
 
+function closeNT(self) {
+  self.emit("close");
+}
 function endNT(socket, callback, err) {
   socket.end();
   callback(err);
@@ -320,7 +323,7 @@ const Socket = (function (InternalSocket) {
       this._parent = this;
       this._parentWrap = this;
       this.#pendingRead = undefined;
-      this.#upgraded = false;
+      this.#upgraded = null;
       if (socket instanceof Socket) {
         this.#socket = socket;
       }
@@ -355,6 +358,14 @@ const Socket = (function (InternalSocket) {
       Socket.#Drain(socket);
     }
 
+    #closeRawConnection() {
+      const connection = this.#upgraded;
+      connection[bunSocketInternal] = null;
+      connection.unref();
+      connection.destroy();
+      process.nextTick(closeNT, connection);
+    }
+
     connect(port, host, connectListener) {
       var path;
       var connection = this.#socket;
@@ -455,7 +466,7 @@ const Socket = (function (InternalSocket) {
 
           if (socket) {
             this.connecting = true;
-            this.#upgraded = true;
+            this.#upgraded = connection;
             const result = socket.upgradeTLS({
               data: this,
               tls,
@@ -466,6 +477,7 @@ const Socket = (function (InternalSocket) {
               // replace socket
               connection[bunSocketInternal] = raw;
               raw.timeout(raw.timeout);
+              this.once("end", this.#closeRawConnection);
               raw.connecting = false;
               this[bunSocketInternal] = tls;
             } else {
@@ -479,7 +491,7 @@ const Socket = (function (InternalSocket) {
               if (!socket) return;
 
               this.connecting = true;
-              this.#upgraded = true;
+              this.#upgraded = connection;
               const result = socket.upgradeTLS({
                 data: this,
                 tls,
@@ -491,6 +503,7 @@ const Socket = (function (InternalSocket) {
                 // replace socket
                 connection[bunSocketInternal] = raw;
                 raw.timeout(raw.timeout);
+                this.once("end", this.#closeRawConnection);
                 raw.connecting = false;
                 this[bunSocketInternal] = tls;
               } else {
@@ -537,6 +550,7 @@ const Socket = (function (InternalSocket) {
     _final(callback) {
       this[bunSocketInternal]?.end();
       callback();
+      process.nextTick(closeNT, this);
     }
 
     get localAddress() {
@@ -559,8 +573,10 @@ const Socket = (function (InternalSocket) {
       const queue = this.#readQueue;
       let chunk;
       while ((chunk = queue.peek())) {
-        if (!this.push(chunk)) return;
+        const can_continue = !this.push(chunk);
+        // always remove from queue push will queue it internally if needed
         queue.shift();
+        if (!can_continue) break;
       }
     }
 
-- 
cgit v1.2.3