use abort signal on http.ClientRequest (#2222)

* use abort signal on http.ClientRequest

* fix edge case and add test

2 files changed, 67 insertions, 18 deletions

diff --git a/src/bun.js/http.exports.js b/src/bun.js/http.exports.js
index 61bb77e04..0cc51af62 100644
--- a/src/bun.js/http.exports.js
+++ b/src/bun.js/http.exports.js
@@ -11,6 +11,9 @@ const debug = process.env.BUN_JS_DEBUG ? (...args) => console.log("node:http", .
 const kEmptyObject = Object.freeze(Object.create(null));
 const kOutHeaders = Symbol.for("kOutHeaders");
 const kEndCalled = Symbol.for("kEndCalled");
+const kAbortController = Symbol.for("kAbortController");
+const kClearTimeout = Symbol("kClearTimeout");
+
 const kCorked = Symbol.for("kCorked");
 const searchParamsSymbol = Symbol.for("query"); // This is the symbol used in Node
 
@@ -546,7 +549,6 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
   return true;
 }
 
-const kClearTimeout = Symbol("kClearTimeout");
 
 export class OutgoingMessage extends Writable {
   #headers;
@@ -559,6 +561,7 @@ export class OutgoingMessage extends Writable {
 
   #fakeSocket;
   #timeoutTimer = null;
+  [kAbortController] = null;
 
   // For compat with IncomingRequest
   get headers() {
@@ -652,7 +655,6 @@ export class OutgoingMessage extends Writable {
     }
   }
 
-  // TODO: Use fetch AbortSignal when implemented
   setTimeout(msecs, callback) {
     if (this.#timeoutTimer) return this;
     if (callback) {
@@ -660,8 +662,9 @@ export class OutgoingMessage extends Writable {
     }
 
     this.#timeoutTimer = setTimeout(async () => {
-      this.emit("timeout");
       this.#timeoutTimer = null;
+      this[kAbortController]?.abort();
+      this.emit("timeout");
     }, msecs);
 
     return this;
@@ -885,8 +888,6 @@ export class ServerResponse extends Writable {
 export class ClientRequest extends OutgoingMessage {
   #timeout;
   #res = null;
-  #aborted = false;
-  #timeoutCb = null;
   #upgradeOrConnect = false;
   #parser = null;
   #maxHeadersCount = null;
@@ -903,6 +904,8 @@ export class ClientRequest extends OutgoingMessage {
 
   #body = null;
   #fetchRequest;
+  #signal = null;
+  [kAbortController] = null;
 
   #options;
   #finished;
@@ -947,12 +950,20 @@ export class ClientRequest extends OutgoingMessage {
 
   _final(callback) {
     this.#finished = true;
+    this[kAbortController] = new AbortController();
+    this[kAbortController].signal.addEventListener("abort", ()=> {
+        this[kClearTimeout]();
+    });
+    if(this.#signal?.aborted){
+      this[kAbortController].abort();
+    }
     this.#fetchRequest = fetch(`${this.#protocol}//${this.#host}:${this.#port}${this.#path}`, {
       method: this.#method,
       headers: this.getHeaders(),
       body: this.#body,
       redirect: "manual",
       verbose: Boolean(process.env.BUN_JS_DEBUG),
+      signal: this[kAbortController].signal
     })
       .then(response => {
         var res = (this.#res = new IncomingMessage(response, {
@@ -970,19 +981,15 @@ export class ClientRequest extends OutgoingMessage {
       });
 
     callback();
-
-    // TODO: Clear timeout here
   }
 
   get aborted() {
-    return this.#aborted;
+    return this.#signal?.aborted || !!this[kAbortController]?.signal.aborted;
   }
 
-  // TODO: Use fetch AbortSignal when implemented
   abort() {
-    if (this.#aborted) return;
-    this.#aborted = true;
-    this[kClearTimeout]();
+    if (this.aborted) return;
+    this[kAbortController].abort();
     // TODO: Close stream if body streaming
   }
 
@@ -1038,13 +1045,16 @@ export class ClientRequest extends OutgoingMessage {
 
     this.#socketPath = options.socketPath;
 
-    // if (options.timeout !== undefined)
-    //   this.timeout = getTimerDuration(options.timeout, "timeout");
+    if (options.timeout !== undefined)
+      this.setTimeout(options.timeout, null);
 
     const signal = options.signal;
     if (signal) {
-      // TODO: Implement this when AbortSignal binding is available from Zig (required for fetch)
-      // addAbortSignal(signal, this);
+      //We still want to control abort function and timeout so signal call our AbortController
+      signal.addEventListener("abort", ()=> {
+        this[kAbortController]?.abort();
+      });
+      this.#signal = signal;
     }
     let method = options.method;
     const methodIsString = typeof method === "string";
@@ -1108,8 +1118,6 @@ export class ClientRequest extends OutgoingMessage {
 
     this.#finished = false;
     this.#res = null;
-    this.#aborted = false;
-    this.#timeoutCb = null;
     this.#upgradeOrConnect = false;
     this.#parser = null;
     this.#maxHeadersCount = null;
diff --git a/test/bun.js/node-http.test.ts b/test/bun.js/node-http.test.ts
index 3ba383c2e..dc26230a3 100644
--- a/test/bun.js/node-http.test.ts
+++ b/test/bun.js/node-http.test.ts
@@ -457,4 +457,45 @@ describe("node:http", () => {
       expect(globalAgent instanceof Agent).toBe(true);
     });
   });
+
+  describe("ClientRequest.signal", () => {
+    let server;
+    let server_port;
+    let server_host;
+    beforeAll(() => {
+      server = createServer((req, res) => {
+        Bun.sleep(10).then(()=> {
+          res.writeHead(200, { "Content-Type": "text/plain" });
+          res.end("Hello World");  
+        })
+      });
+      server.listen({ port: 0}, (_err,host, port)=> {
+        server_port = port;
+        server_host = host;
+      });
+    });
+    afterAll(() => {
+      server.close();
+    });
+    it("should attempt to make a standard GET request and abort", done => {
+      get(`http://127.0.0.1:${server_port}`,{ signal: AbortSignal.timeout(5)  }, res => {
+        let data = "";
+        res.setEncoding("utf8");
+        res.on("data", chunk => {
+          data += chunk;
+        });
+        res.on("end", () => {
+          expect(true).toBeFalsy();
+          done();
+        });
+        res.on("error", _ => {
+          expect(true).toBeFalsy();
+          done();
+        });
+      }).on("error", (err)=>{
+        expect(err?.name).toBe("AbortError");
+        done();
+      });
+    });
+  });
 });