fix: only apply our pipeTo/pipeThrough optimisations to TransformStreams who have no transformers (IdentityStreams). (#556)

Author: JakeChampion

.github/workflows/main.yml

@@ -255,6 +255,7 @@ jobs:
           - streaming-close
           - tee
           - timers
+          - transform-stream
           - urlsearchparams
     steps:
     - name: Checkout fastly/js-compute-runtime
@@ -404,6 +405,7 @@ jobs:
           - 'secret-store'
           - 'status'
           - 'timers'
+          - 'transform-stream'
           - 'urlsearchparams'
     steps:
     - uses: actions/checkout@v3

integration-tests/js-compute/fixtures/transform-stream/bin/index.js

@@ -0,0 +1,26 @@
+/// <reference path="../../../../../types/index.d.ts" />
+/* eslint-env serviceworker */
+/* global TransformStream */
+
+import { routes } from "../../../test-harness.js";
+
+function upperCase() {
+    const decoder = new TextDecoder()
+    const encoder = new TextEncoder()
+    return new TransformStream({
+        transform(chunk, controller) {
+            controller.enqueue(encoder.encode(decoder.decode(chunk).toUpperCase()));
+        },
+    });
+}
+
+routes.set("/identity", () => {
+    return fetch('https://http-me.glitch.me/test?body=hello', { backend: 'http-me' }).then(response => {
+        return new Response(response.body.pipeThrough(new TransformStream));
+    })
+});
+routes.set("/uppercase", () => {
+    return fetch('https://http-me.glitch.me/test?body=hello', { backend: 'http-me' }).then(response => {
+        return new Response(response.body.pipeThrough(upperCase()));
+    })
+});

integration-tests/js-compute/fixtures/transform-stream/fastly.toml.in

@@ -0,0 +1,27 @@
+# This file describes a Fastly Compute@Edge package. To learn more visit:
+# https://developer.fastly.com/reference/fastly-toml/
+
+authors = ["[email protected]"]
+description = ""
+language = "other"
+manifest_version = 2
+name = "transform-stream"
+service_id = ""
+
+[scripts]
+  build = "node ../../../../js-compute-runtime-cli.js bin/index.js"
+
+[local_server]
+
+  [local_server.backends]
+
+    [local_server.backends.http-me]
+      url = "https://http-me.glitch.me"
+
+[setup]
+
+  [setup.backends]
+
+    [setup.backends.http-me]
+      address = "http-me.glitch.me"
+      port = 443

integration-tests/js-compute/fixtures/transform-stream/tests.json

@@ -0,0 +1,28 @@
+{
+  "GET /identity": {
+    "environments": [
+      "c@e", "viceroy"
+    ],
+    "downstream_request": {
+      "method": "GET",
+      "pathname": "/identity"
+    },
+    "downstream_response": {
+      "status": 200,
+      "body": "hello"
+    }
+  },
+  "GET /uppercase": {
+    "environments": [
+      "c@e", "viceroy"
+    ],
+    "downstream_request": {
+      "method": "GET",
+      "pathname": "/uppercase"
+    },
+    "downstream_response": {
+      "status": 200,
+      "body": "HELLO"
+    }
+  }
+}

runtime/js-compute-runtime/builtins/transform-stream.cpp

@@ -42,7 +42,14 @@ bool pipeTo(JSContext *cx, unsigned argc, JS::Value *vp) {
   JS::RootedObject target(cx, args[0].isObject() ? &args[0].toObject() : nullptr);
   if (target && builtins::NativeStreamSource::stream_has_native_source(cx, self) &&
       JS::IsWritableStream(target) && builtins::TransformStream::is_ts_writable(cx, target)) {
-    builtins::NativeStreamSource::set_stream_piped_to_ts_writable(cx, self, target);
+    if (auto ts = builtins::TransformStream::ts_from_writable(cx, target)) {
+      auto streamHasTransformer =
+          JS::GetReservedSlot(ts, builtins::TransformStream::Slots::HasTransformer).toBoolean();
+      // We only want to apply the optimisation on identity-streams
+      if (!streamHasTransformer) {
+        builtins::NativeStreamSource::set_stream_piped_to_ts_writable(cx, self, target);
+      }
+    }
   }
 
   return JS::Call(cx, args.thisv(), original_pipeTo, JS::HandleValueArray(args), args.rval());
@@ -318,6 +325,15 @@ JSObject *TransformStream::writable(JSObject *self) {
   return &JS::GetReservedSlot(self, TransformStream::Slots::Writable).toObject();
 }
 
+JSObject *TransformStream::ts_from_writable(JSContext *cx, JS::HandleObject writable) {
+  MOZ_ASSERT(is_ts_writable(cx, writable));
+  JSObject *sink = builtins::NativeStreamSink::get_stream_sink(cx, writable);
+  if (!sink || !builtins::NativeStreamSink::is_instance(sink)) {
+    return nullptr;
+  }
+  return builtins::NativeStreamSink::owner(sink);
+}
+
 bool TransformStream::is_ts_writable(JSContext *cx, JS::HandleObject writable) {
   JSObject *sink = builtins::NativeStreamSink::get_stream_sink(cx, writable);
   if (!sink || !builtins::NativeStreamSink::is_instance(sink)) {
@@ -466,6 +482,9 @@ bool TransformStream::constructor(JSContext *cx, unsigned argc, JS::Value *vp) {
   if (!self)
     return false;
 
+  JS::SetReservedSlot(self, TransformStream::Slots::HasTransformer,
+                      JS::BooleanValue(JS::ToBoolean(transformer)));
+
   args.rval().setObject(*self);
   return true;
 }

runtime/js-compute-runtime/builtins/transform-stream.h

@@ -22,6 +22,7 @@ class TransformStream : public BuiltinImpl<TransformStream> {
                  // used as a body.
     UsedAsMixin, // `true` if the TransformStream is used in another transforming
                  // builtin, such as CompressionStream.
+    HasTransformer,
     Count
   };
   static const JSFunctionSpec static_methods[];
@@ -39,6 +40,7 @@ class TransformStream : public BuiltinImpl<TransformStream> {
                                         JS::HandleObject target);
   static JSObject *writable(JSObject *self);
   static bool is_ts_writable(JSContext *cx, JS::HandleObject writable);
+  static JSObject *ts_from_writable(JSContext *cx, JS::HandleObject writable);
   static JSObject *controller(JSObject *self);
   static bool backpressure(JSObject *self);
   static JSObject *backpressureChangePromise(JSObject *self);