From 7f8bf2537db0ae596a2c1baabd4011a190666ca6 Mon Sep 17 00:00:00 2001
From: Matt Mastracci <matthew@mastracci.com>
Date: Thu, 3 Aug 2023 14:27:25 -0600
Subject: refactor(ext/fetch): refactor fetch to use new write_error method
 (#20029)

This is a prerequisite for fast streams work -- this particular resource
used a custom `mpsc`-style stream, and this work will allow us to unify
it with the streams in `ext/http` in time.

Instead of using Option as an internal semaphore for "correctly
completed EOF", we allow code to propagate errors into the channel which
can be picked up by downstream sinks like Hyper. EOF is signalled using
a more standard sender drop.
---
 ext/fetch/26_fetch.js | 32 +++++++++++++++++++++-----------
 1 file changed, 21 insertions(+), 11 deletions(-)

(limited to 'ext/fetch/26_fetch.js')

diff --git a/ext/fetch/26_fetch.js b/ext/fetch/26_fetch.js
index 5084fab34..6be63d077 100644
--- a/ext/fetch/26_fetch.js
+++ b/ext/fetch/26_fetch.js
@@ -201,6 +201,23 @@ async function mainFetch(req, recursive, terminator) {
 
   let requestSendError;
   let requestSendErrorSet = false;
+
+  async function propagateError(err, message) {
+    // TODO(lucacasonato): propagate error into response body stream
+    try {
+      await core.writeTypeError(requestBodyRid, message);
+    } catch (err) {
+      if (!requestSendErrorSet) {
+        requestSendErrorSet = true;
+        requestSendError = err;
+      }
+    }
+    if (!requestSendErrorSet) {
+      requestSendErrorSet = true;
+      requestSendError = err;
+    }
+  }
+
   if (requestBodyRid !== null) {
     if (
       reqBody === null ||
@@ -220,9 +237,7 @@ async function mainFetch(req, recursive, terminator) {
           val = res.value;
         } catch (err) {
           if (terminator.aborted) break;
-          // TODO(lucacasonato): propagate error into response body stream
-          requestSendError = err;
-          requestSendErrorSet = true;
+          await propagateError(err, "failed to read");
           break;
         }
         if (done) break;
@@ -231,9 +246,7 @@ async function mainFetch(req, recursive, terminator) {
             "Item in request body ReadableStream is not a Uint8Array",
           );
           await reader.cancel(error);
-          // TODO(lucacasonato): propagate error into response body stream
-          requestSendError = error;
-          requestSendErrorSet = true;
+          await propagateError(error, error.message);
           break;
         }
         try {
@@ -241,9 +254,7 @@ async function mainFetch(req, recursive, terminator) {
         } catch (err) {
           if (terminator.aborted) break;
           await reader.cancel(err);
-          // TODO(lucacasonato): propagate error into response body stream
-          requestSendError = err;
-          requestSendErrorSet = true;
+          await propagateError(err, "failed to write");
           break;
         }
       }
@@ -252,8 +263,7 @@ async function mainFetch(req, recursive, terminator) {
           await core.shutdown(requestBodyRid);
         } catch (err) {
           if (!terminator.aborted) {
-            requestSendError = err;
-            requestSendErrorSet = true;
+            await propagateError(err, "failed to flush");
           }
         }
       }
-- 
cgit v1.2.3