more streams debugging

Author: Guy Bedford

integration-tests/js-compute/fixtures/app/src/index.js

@@ -55,7 +55,7 @@ if (env('FASTLY_DEBUG_LOGGING') === '1') {
   if (fastly.debugMessages) {
     const { debug: consoleDebug } = console;
     console.debug = function debug(...args) {
-      fastly.debugMessages.push(...args);
+      fastly.debugLog(...args);
       consoleDebug(...args);
     };
   }

integration-tests/js-compute/fixtures/module-mode/src/http-cache.js

@@ -564,13 +564,54 @@ const getTestUrl = (path = `/${Math.random().toString().slice(2)}`) =>
     );
   });
 
+  routes.set('/http-cache/body-stream', async () => {
+    const url = getTestUrl();
+
+    const cacheOverride = new CacheOverride({
+      afterSend(res) {
+        // Create a transform that uppercases the response
+        const transformer = new TransformStream({
+          start(controller) {
+            console.debug('transform start');
+          },
+          flush(controller) {
+            console.debug('transform flush');
+          },
+          transform(chunk, controller) {
+            console.debug('transform', chunk.byteLength);
+            const text = new TextDecoder().decode(chunk);
+            const upperText = text.toUpperCase();
+            const upperChunk = new TextEncoder().encode(upperText);
+            console.debug('enqueue', upperChunk.byteLength);
+            controller.enqueue(upperChunk);
+          },
+        });
+
+        return {
+          bodyTransform: transformer,
+          cache: true,
+        };
+      },
+    });
+
+    const res = await fetch(url, { cacheOverride });
+    const text = await res.text();
+    strictEqual(text, text.toUpperCase());
+  });
+
   routes.set('/http-cache/body-transform', async () => {
     const url = getTestUrl();
 
     const cacheOverride = new CacheOverride({
       afterSend(res) {
         // Create a transform that uppercases the response
         const transformer = new TransformStream({
+          start(controller) {
+            console.debug('transform start');
+          },
+          flush(controller) {
+            console.debug('transform flush');
+          },
           transform(chunk, controller) {
             console.debug('transform', chunk.byteLength);
             const text = new TextDecoder().decode(chunk);
@@ -591,6 +632,7 @@ const getTestUrl = (path = `/${Math.random().toString().slice(2)}`) =>
     const res = await fetch(url, { cacheOverride });
     const text = await res.text();
     strictEqual(text, text.toUpperCase());
+    throw new Error('wow');
   });
 
   // Test transform that throws an error

integration-tests/js-compute/fixtures/module-mode/src/index.js

@@ -22,7 +22,7 @@ if (env('FASTLY_DEBUG_LOGGING') === '1') {
   if (fastly.debugMessages) {
     const { debug: consoleDebug } = console;
     console.debug = function debug(...args) {
-      fastly.debugMessages.push(...args);
+      fastly.debugLog(...args);
       consoleDebug(...args);
     };
   }

integration-tests/js-compute/fixtures/module-mode/tests.json

@@ -214,7 +214,6 @@
     "features": ["http-cache"]
   },
   "GET /http-cache/body-transform": {
-    "skip": true,
     "environments": ["compute"],
     "features": ["http-cache"]
   },

runtime/fastly/builtins/fastly.cpp

@@ -95,6 +95,23 @@ bool Fastly::enableDebugLogging(JSContext *cx, unsigned argc, JS::Value *vp) {
   return true;
 }
 
+bool debugLog(JSContext *cx, unsigned argc, JS::Value *vp) {
+  JS::CallArgs args = CallArgsFromVp(argc, vp);
+  if (!args.requireAtLeast(cx, __func__, 1))
+    return false;
+  JS::RootedString msg_str(cx, JS::ToString(cx, args[0]));
+  if (!msg_str) {
+    return false;
+  }
+  auto msg_host_str = core::encode(cx, msg_str);
+  if (!msg_host_str) {
+    return false;
+  }
+  debug_messages.push_back(std::string(msg_host_str));
+  args.rval().setUndefined();
+  return true;
+}
+
 bool Fastly::getGeolocationForIpAddress(JSContext *cx, unsigned argc, JS::Value *vp) {
   JS::CallArgs args = CallArgsFromVp(argc, vp);
   REQUEST_HANDLER_ONLY("fastly.getGeolocationForIpAddress");
@@ -523,6 +540,7 @@ bool install(api::Engine *engine) {
   const JSFunctionSpec methods[] = {
       JS_FN("dump", Fastly::dump, 1, 0),
       JS_FN("enableDebugLogging", Fastly::enableDebugLogging, 1, JSPROP_ENUMERATE),
+      JS_FN("debugLog", debugLog, 1, JSPROP_ENUMERATE),
       JS_FN("getGeolocationForIpAddress", Fastly::getGeolocationForIpAddress, 1, JSPROP_ENUMERATE),
       JS_FN("getLogger", Fastly::getLogger, 1, JSPROP_ENUMERATE),
       JS_FN("includeBytes", Fastly::includeBytes, 1, JSPROP_ENUMERATE),

runtime/fastly/builtins/fetch/fetch.cpp

@@ -461,25 +461,26 @@ template <InternalMethod then_or_catch_handler>
 bool apply_transform_stream(JSContext *cx, JS::HandleObject response,
                             host_api::HttpBody into_body) {
   DEBUG_LOG("Apply transform stream")
-  // Create the stream source from the response body (in turn queues the stream task)
+  // Create the stream source from the response body
   JS::RootedObject body_stream_in(cx, RequestOrResponse::create_body_stream(cx, response));
   if (!body_stream_in) {
     return false;
   }
 
-  // Pipe the stream source through the transform stream
-  JSObject *transform_stream =
-      &JS::GetReservedSlot(response, static_cast<uint32_t>(Response::Slots::CacheBodyTransform))
-           .toObject();
-  JS::RootedValueArray<1> pipeThroughArgs(cx);
-  pipeThroughArgs[0].setObject(*transform_stream);
-  JS::RootedValue transform_readable_out(cx);
-  DEBUG_LOG("calling pipe through")
-  if (!JS::Call(cx, body_stream_in, "pipeThrough", pipeThroughArgs, &transform_readable_out)) {
-    DEBUG_LOG("pipe through ERR")
+  // Create the transform stream, and extract its readable and writable end objects
+  JS::RootedValue transform_stream(
+      cx,
+      JS::GetReservedSlot(response, static_cast<uint32_t>(Response::Slots::CacheBodyTransform)));
+  JS::RootedObject transform_stream_obj(cx, &transform_stream.toObject());
+  JS::RootedValue transform_readable(cx);
+  if (!JS_GetProperty(cx, transform_stream_obj, "readable", &transform_readable)) {
+    return false;
+  }
+
+  JS::RootedValue transform_writable(cx);
+  if (!JS_GetProperty(cx, transform_stream_obj, "writable", &transform_writable)) {
     return false;
   }
-  DEBUG_LOG("pipe through OK")
 
   // Pipe the readable end of the transform stream into the into_body, with the completion handler
   // on error or success
@@ -496,7 +497,7 @@ bool apply_transform_stream(JSContext *cx, JS::HandleObject response,
   if (!sink) {
     return false;
   }
-  JS::RootedObject transform_readable_obj(cx, &transform_readable_out.toObject());
+  JS::RootedObject transform_readable_obj(cx, &transform_readable.toObject());
   JS::RootedValueArray<1> pipeToArgs(cx);
   pipeToArgs[0].setObject(*sink);
   JS::RootedValue pipe_promise(cx);
@@ -521,6 +522,18 @@ bool apply_transform_stream(JSContext *cx, JS::HandleObject response,
     }
   }
 
+  // Pipe the incoming body stream into the writable end of the transform stream
+  JS::RootedValueArray<1> args(cx);
+  args[0].set(transform_writable);
+  JS::RootedValue pipe_ret(cx);
+  if (!JS::Call(cx, body_stream_in, "pipeTo", args, &pipe_ret)) {
+    return false;
+  }
+
+  // We now add a task for the incoming handle
+  // ENGINE->queue_async_task(
+  //     new FastlyAsyncTask(RequestOrResponse::body_handle(response).async_handle(), nullptr,
+  //                         JS::UndefinedHandleValue, nullptr));
   return true;
 }
 

runtime/fastly/builtins/fetch/request-response.cpp

@@ -141,36 +141,36 @@ bool process_body_read(JSContext *cx, host_api::HttpBody::Handle handle, JS::Han
   return true;
 }
 
+enum StreamState { Complete, Wait, Error };
+
 struct ReadResult {
   JS::UniqueChars buffer;
   size_t length;
-  bool end_of_stream;
+  StreamState state;
 };
 
 // Returns a UniqueChars and the length of that string. The UniqueChars value is not
 // null-terminated.
-template <bool read_async> ReadResult read_from_handle_all(JSContext *cx, host_api::HttpBody body) {
+ReadResult read_from_handle_all(JSContext *cx, host_api::HttpBody body) {
   std::vector<host_api::HostString> chunks;
   size_t bytes_read = 0;
   bool end_of_stream = true;
   while (true) {
     DEBUG_LOG("read from handle all loop")
-    if (read_async) {
-      auto ready_res = body.is_ready();
-      if (auto *err = ready_res.to_err()) {
-        HANDLE_ERROR(cx, *err);
-        return {nullptr, 0, true};
-      }
-      if (!ready_res.unwrap()) {
-        DEBUG_LOG("async break")
-        end_of_stream = false;
-        break;
-      }
+    auto ready_res = body.is_ready();
+    if (auto *err = ready_res.to_err()) {
+      HANDLE_ERROR(cx, *err);
+      return {nullptr, 0, StreamState::Error};
+    }
+    if (!ready_res.unwrap()) {
+      DEBUG_LOG("async break")
+      end_of_stream = false;
+      break;
     }
     auto res = body.read(HANDLE_READ_CHUNK_SIZE);
     if (auto *err = res.to_err()) {
       HANDLE_ERROR(cx, *err);
-      return {nullptr, 0, true};
+      return {nullptr, 0, StreamState::Error};
     }
 
     auto &chunk = res.unwrap();
@@ -184,7 +184,7 @@ template <bool read_async> ReadResult read_from_handle_all(JSContext *cx, host_a
 
   JS::UniqueChars buf;
   if (chunks.size() == 0) {
-    return {nullptr, 0, end_of_stream};
+    return {nullptr, 0, end_of_stream ? StreamState::Complete : StreamState::Wait};
   } else if (chunks.size() == 1) {
     // If there was only one chunk read, reuse that allocation.
     auto &chunk = chunks.back();
@@ -194,7 +194,7 @@ template <bool read_async> ReadResult read_from_handle_all(JSContext *cx, host_a
     buf.reset(static_cast<char *>(JS_string_malloc(cx, bytes_read)));
     if (!buf) {
       JS_ReportOutOfMemory(cx);
-      return {nullptr, 0};
+      return {nullptr, 0, StreamState::Error};
     }
 
     char *end = buf.get();
@@ -203,7 +203,7 @@ template <bool read_async> ReadResult read_from_handle_all(JSContext *cx, host_a
     }
   }
 
-  return {std::move(buf), bytes_read, end_of_stream};
+  return {std::move(buf), bytes_read, end_of_stream ? StreamState::Complete : StreamState::Wait};
 }
 
 } // namespace
@@ -1383,9 +1383,9 @@ bool async_process_body_handle_for_bodyAll(JSContext *cx, uint32_t handle, JS::H
   auto body = RequestOrResponse::body_handle(self);
   auto *parse_body = reinterpret_cast<RequestOrResponse::ParseBodyCB *>(body_parser.toPrivate());
   DEBUG_LOG("Consume read from handle all task")
-  auto [buf, bytes_read, end_of_stream] = read_from_handle_all<true>(cx, body);
+  auto [buf, bytes_read, state] = read_from_handle_all(cx, body);
   DEBUG_LOG("Got handle all task")
-  if (!buf && end_of_stream) {
+  if (state == StreamState::Error) {
     DEBUG_LOG("stream error")
     JS::RootedObject result_promise(cx);
     result_promise =
@@ -1396,7 +1396,7 @@ bool async_process_body_handle_for_bodyAll(JSContext *cx, uint32_t handle, JS::H
     return RejectPromiseWithPendingError(cx, result_promise);
   }
 
-  if (end_of_stream) {
+  if (state == StreamState::Complete) {
     DEBUG_LOG("stream complete")
     return parse_body(cx, self, std::move(buf), bytes_read);
   }
@@ -1414,9 +1414,9 @@ bool RequestOrResponse::consume_body_handle_for_bodyAll(JSContext *cx, JS::Handl
   auto body = body_handle(self);
   auto *parse_body = reinterpret_cast<ParseBodyCB *>(body_parser.toPrivate());
   DEBUG_LOG("Consume read from handle all")
-  auto [buf, bytes_read, end_of_stream] = read_from_handle_all<true>(cx, body);
+  auto [buf, bytes_read, state] = read_from_handle_all(cx, body);
   DEBUG_LOG("Got handle all")
-  if (!buf && end_of_stream) {
+  if (state == StreamState::Error) {
     DEBUG_LOG("stream error")
     JS::RootedObject result_promise(cx);
     result_promise =
@@ -1425,7 +1425,7 @@ bool RequestOrResponse::consume_body_handle_for_bodyAll(JSContext *cx, JS::Handl
     return RejectPromiseWithPendingError(cx, result_promise);
   }
 
-  if (end_of_stream) {
+  if (state == StreamState::Complete) {
     DEBUG_LOG("stream complete")
     return parse_body(cx, self, std::move(buf), bytes_read);
   }

runtime/fastly/host-api/host_api.cpp

@@ -899,8 +899,7 @@ Result<HttpBody> HttpBody::make() {
 }
 
 Result<HostString> HttpBody::read(uint32_t chunk_size) const {
-  auto handle = std::to_string(reinterpret_cast<uint32_t>(this->handle));
-  TRACE_CALL_ARGS(TSV(handle))
+  TRACE_CALL_ARGS(TSV(std::to_string(handle)))
   Result<HostString> res;
 
   fastly::fastly_world_list_u8 ret;
@@ -912,6 +911,7 @@ Result<HostString> HttpBody::read(uint32_t chunk_size) const {
     cabi_free(ret.ptr);
     res.emplace_err(err);
   } else {
+    TRACE_CALL_RET(TSV(std::to_string(ret.len)))
     res.emplace(JS::UniqueChars(reinterpret_cast<char *>(ret.ptr)), ret.len);
   }