chore: ♻️ lift changes from #5694 to fetch package so I can build on top of them

Author: sestinj

packages/fetch/src/fetch.ts

@@ -107,19 +107,30 @@ export async function fetchwithRequestOptions(
   }
 
   // fetch the request with the provided options
-  const resp = await fetch(url, {
-    ...init,
-    body: updatedBody ?? init?.body,
-    headers: headers,
-    agent: agent,
-  });
-
-  if (!resp.ok) {
-    const requestId = resp.headers.get("x-request-id");
-    if (requestId) {
-      console.log(`Request ID: ${requestId}, Status: ${resp.status}`);
+  try {
+    const resp = await fetch(url, {
+      ...init,
+      body: updatedBody ?? init?.body,
+      headers: headers,
+      agent: agent,
+    });
+
+    if (!resp.ok) {
+      const requestId = resp.headers.get("x-request-id");
+      if (requestId) {
+        console.log(`Request ID: ${requestId}, Status: ${resp.status}`);
+      }
     }
-  }
 
-  return resp;
+    return resp;
+  } catch (error) {
+    if (error instanceof Error && error.name === "AbortError") {
+      // Return a Response object that streamResponse etc can handle
+      return new Response(null, {
+        status: 499, // Client Closed Request
+        statusText: "Client Closed Request",
+      });
+    }
+    throw error;
+  }
 }

packages/fetch/src/stream.ts

@@ -10,6 +10,9 @@ export async function* toAsyncIterable(
 export async function* streamResponse(
   response: Response,
 ): AsyncGenerator<string> {
+  if (response.status === 499) {
+    return; // In case of client-side cancellation, just return
+  }
   if (response.status !== 200) {
     throw new Error(await response.text());
   }
@@ -21,22 +24,29 @@ export async function* streamResponse(
   // Get the major version of Node.js
   const nodeMajorVersion = parseInt(process.versions.node.split(".")[0], 10);
 
-  if (nodeMajorVersion >= 20) {
-    // Use the new API for Node 20 and above
-    const stream = (ReadableStream as any).from(response.body);
-    for await (const chunk of stream.pipeThrough(
-      new TextDecoderStream("utf-8"),
-    )) {
-      yield chunk;
+  try {
+    if (nodeMajorVersion >= 20) {
+      // Use the new API for Node 20 and above
+      const stream = (ReadableStream as any).from(response.body);
+      for await (const chunk of stream.pipeThrough(
+        new TextDecoderStream("utf-8"),
+      )) {
+        yield chunk;
+      }
+    } else {
+      // Fallback for Node versions below 20
+      // Streaming with this method doesn't work as version 20+ does
+      const decoder = new TextDecoder("utf-8");
+      const nodeStream = response.body as unknown as NodeJS.ReadableStream;
+      for await (const chunk of toAsyncIterable(nodeStream)) {
+        yield decoder.decode(chunk, { stream: true });
+      }
     }
-  } else {
-    // Fallback for Node versions below 20
-    // Streaming with this method doesn't work as version 20+ does
-    const decoder = new TextDecoder("utf-8");
-    const nodeStream = response.body as unknown as NodeJS.ReadableStream;
-    for await (const chunk of toAsyncIterable(nodeStream)) {
-      yield decoder.decode(chunk, { stream: true });
+  } catch (e) {
+    if (e instanceof Error && e.name.startsWith("AbortError")) {
+      return; // In case of client-side cancellation, just return
     }
+    throw e;
   }
 }