lib,src: implement WebAssembly Web API

Refs: nodejs#41749
Fixes: nodejs#21130

PR-URL: nodejs#42701
Reviewed-By: Gus Caplan <me@gus.host>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

doc/api/errors.md

@@ -2890,6 +2890,17 @@ The WASI instance has already started.
 
 The WASI instance has not been started.
 
+<a id="ERR_WEBASSEMBLY_RESPONSE"></a>
+
+### `ERR_WEBASSEMBLY_RESPONSE`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The `Response` that has been passed to `WebAssembly.compileStreaming` or to
+`WebAssembly.instantiateStreaming` is not a valid WebAssembly response.
+
 <a id="ERR_WORKER_INIT_FAILED"></a>
 
 ### `ERR_WORKER_INIT_FAILED`

lib/internal/bootstrap/pre_execution.js

@@ -5,6 +5,7 @@ const {
   ObjectDefineProperties,
   ObjectDefineProperty,
   ObjectGetOwnPropertyDescriptor,
+  PromiseResolve,
   SafeMap,
   SafeWeakMap,
   StringPrototypeStartsWith,
@@ -24,7 +25,11 @@ const {
 } = require('internal/util');
 
 const { Buffer } = require('buffer');
-const { ERR_MANIFEST_ASSERT_INTEGRITY } = require('internal/errors').codes;
+const {
+  ERR_INVALID_ARG_TYPE,
+  ERR_MANIFEST_ASSERT_INTEGRITY,
+  ERR_WEBASSEMBLY_RESPONSE,
+} = require('internal/errors').codes;
 const assert = require('internal/assert');
 
 function prepareMainThreadExecution(expandArgv1 = false,
@@ -215,6 +220,44 @@ function setupFetch() {
     Request: lazyInterface('Request'),
     Response: lazyInterface('Response'),
   });
+
+  // The WebAssembly Web API: https://webassembly.github.io/spec/web-api
+  internalBinding('wasm_web_api').setImplementation((streamState, source) => {
+    (async () => {
+      const response = await PromiseResolve(source);
+      if (!(response instanceof lazyUndici().Response)) {
+        throw new ERR_INVALID_ARG_TYPE(
+          'source', ['Response', 'Promise resolving to Response'], response);
+      }
+
+      const contentType = response.headers.get('Content-Type');
+      if (contentType !== 'application/wasm') {
+        throw new ERR_WEBASSEMBLY_RESPONSE(
+          `has unsupported MIME type '${contentType}'`);
+      }
+
+      if (!response.ok) {
+        throw new ERR_WEBASSEMBLY_RESPONSE(
+          `has status code ${response.status}`);
+      }
+
+      if (response.bodyUsed !== false) {
+        throw new ERR_WEBASSEMBLY_RESPONSE('body has already been used');
+      }
+
+      // Pass all data from the response body to the WebAssembly compiler.
+      for await (const chunk of response.body) {
+        streamState.push(chunk);
+      }
+    })().then(() => {
+      // No error occurred. Tell the implementation that the stream has ended.
+      streamState.finish();
+    }, (err) => {
+      // An error occurred, either because the given object was not a valid
+      // and usable Response or because a network error occurred.
+      streamState.abort(err);
+    });
+  });
 }
 
 // TODO(aduh95): move this to internal/bootstrap/browser when the CLI flag is

lib/internal/errors.js

@@ -1650,6 +1650,7 @@ E('ERR_VM_MODULE_NOT_MODULE',
   'Provided module is not an instance of Module', Error);
 E('ERR_VM_MODULE_STATUS', 'Module status %s', Error);
 E('ERR_WASI_ALREADY_STARTED', 'WASI instance has already started', Error);
+E('ERR_WEBASSEMBLY_RESPONSE', 'WebAssembly response %s', TypeError);
 E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error);
 E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') =>
   `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`,

node.gyp

@@ -543,6 +543,7 @@
         'src/node_util.cc',
         'src/node_v8.cc',
         'src/node_wasi.cc',
+        'src/node_wasm_web_api.cc',
         'src/node_watchdog.cc',
         'src/node_worker.cc',
         'src/node_zlib.cc',

src/api/environment.cc

@@ -3,8 +3,10 @@
 #include "node_errors.h"
 #include "node_internals.h"
 #include "node_native_module_env.h"
+#include "node_options-inl.h"
 #include "node_platform.h"
 #include "node_v8_platform-inl.h"
+#include "node_wasm_web_api.h"
 #include "uv.h"
 
 #if HAVE_INSPECTOR
@@ -252,6 +254,13 @@ void SetIsolateMiscHandlers(v8::Isolate* isolate, const IsolateSettings& s) {
     s.allow_wasm_code_generation_callback : AllowWasmCodeGenerationCallback;
   isolate->SetAllowWasmCodeGenerationCallback(allow_wasm_codegen_cb);
 
+  Mutex::ScopedLock lock(node::per_process::cli_options_mutex);
+  if (per_process::cli_options->get_per_isolate_options()
+          ->get_per_env_options()
+          ->experimental_fetch) {
+    isolate->SetWasmStreamingCallback(wasm_web_api::StartStreamingCompilation);
+  }
+
   if ((s.flags & SHOULD_NOT_SET_PROMISE_REJECTION_CALLBACK) == 0) {
     auto* promise_reject_cb = s.promise_reject_callback ?
       s.promise_reject_callback : PromiseRejectCallback;

src/env.h

@@ -550,7 +550,9 @@ constexpr size_t kFsStatsBufferLength =
   V(tls_wrap_constructor_function, v8::Function)                               \
   V(trace_category_state_function, v8::Function)                               \
   V(udp_constructor_function, v8::Function)                                    \
-  V(url_constructor_function, v8::Function)
+  V(url_constructor_function, v8::Function)                                    \
+  V(wasm_streaming_compilation_impl, v8::Function)                             \
+  V(wasm_streaming_object_constructor, v8::Function)
 
 class Environment;
 struct AllocatedBuffer;

src/node_binding.cc

@@ -87,6 +87,7 @@
   V(uv)                                                                        \
   V(v8)                                                                        \
   V(wasi)                                                                      \
+  V(wasm_web_api)                                                              \
   V(watchdog)                                                                  \
   V(worker)                                                                    \
   V(zlib)

src/node_wasm_web_api.cc

@@ -0,0 +1,196 @@
+#include "node_wasm_web_api.h"
+
+#include "memory_tracker-inl.h"
+#include "node_errors.h"
+
+namespace node {
+namespace wasm_web_api {
+
+using v8::ArrayBuffer;
+using v8::ArrayBufferView;
+using v8::Context;
+using v8::Function;
+using v8::FunctionCallbackInfo;
+using v8::FunctionTemplate;
+using v8::Local;
+using v8::MaybeLocal;
+using v8::Object;
+using v8::Value;
+using v8::WasmStreaming;
+
+Local<Function> WasmStreamingObject::Initialize(Environment* env) {
+  Local<Function> templ = env->wasm_streaming_object_constructor();
+  if (!templ.IsEmpty()) {
+    return templ;
+  }
+
+  Local<FunctionTemplate> t = env->NewFunctionTemplate(New);
+  t->Inherit(BaseObject::GetConstructorTemplate(env));
+  t->InstanceTemplate()->SetInternalFieldCount(
+      WasmStreamingObject::kInternalFieldCount);
+
+  env->SetProtoMethod(t, "push", Push);
+  env->SetProtoMethod(t, "finish", Finish);
+  env->SetProtoMethod(t, "abort", Abort);
+
+  auto function = t->GetFunction(env->context()).ToLocalChecked();
+  env->set_wasm_streaming_object_constructor(function);
+  return function;
+}
+
+void WasmStreamingObject::RegisterExternalReferences(
+    ExternalReferenceRegistry* registry) {
+  registry->Register(Push);
+  registry->Register(Finish);
+  registry->Register(Abort);
+}
+
+void WasmStreamingObject::MemoryInfo(MemoryTracker* tracker) const {
+  // v8::WasmStreaming is opaque. We assume that the size of the WebAssembly
+  // module that is being compiled is roughly what V8 allocates (as in, off by
+  // only a small factor).
+  tracker->TrackFieldWithSize("streaming", wasm_size_);
+}
+
+MaybeLocal<Object> WasmStreamingObject::Create(
+    Environment* env, std::shared_ptr<WasmStreaming> streaming) {
+  Local<Function> ctor = Initialize(env);
+  Local<Object> obj;
+  if (!ctor->NewInstance(env->context(), 0, nullptr).ToLocal(&obj)) {
+    return MaybeLocal<Object>();
+  }
+
+  CHECK(streaming);
+
+  WasmStreamingObject* ptr = Unwrap<WasmStreamingObject>(obj);
+  CHECK_NOT_NULL(ptr);
+  ptr->streaming_ = streaming;
+  ptr->wasm_size_ = 0;
+  return obj;
+}
+
+void WasmStreamingObject::New(const FunctionCallbackInfo<Value>& args) {
+  CHECK(args.IsConstructCall());
+  Environment* env = Environment::GetCurrent(args);
+  new WasmStreamingObject(env, args.This());
+}
+
+void WasmStreamingObject::Push(const FunctionCallbackInfo<Value>& args) {
+  WasmStreamingObject* obj;
+  ASSIGN_OR_RETURN_UNWRAP(&obj, args.Holder());
+  CHECK(obj->streaming_);
+
+  CHECK_EQ(args.Length(), 1);
+  Local<Value> chunk = args[0];
+
+  // The start of the memory section backing the ArrayBuffer(View), the offset
+  // of the ArrayBuffer(View) within the memory section, and its size in bytes.
+  const void* bytes;
+  size_t offset;
+  size_t size;
+
+  if (LIKELY(chunk->IsArrayBufferView())) {
+    Local<ArrayBufferView> view = chunk.As<ArrayBufferView>();
+    bytes = view->Buffer()->GetBackingStore()->Data();
+    offset = view->ByteOffset();
+    size = view->ByteLength();
+  } else if (LIKELY(chunk->IsArrayBuffer())) {
+    Local<ArrayBuffer> buffer = chunk.As<ArrayBuffer>();
+    bytes = buffer->GetBackingStore()->Data();
+    offset = 0;
+    size = buffer->ByteLength();
+  } else {
+    return node::THROW_ERR_INVALID_ARG_TYPE(
+        Environment::GetCurrent(args),
+        "chunk must be an ArrayBufferView or an ArrayBuffer");
+  }
+
+  // Forward the data to V8. Internally, V8 will make a copy.
+  obj->streaming_->OnBytesReceived(static_cast<const uint8_t*>(bytes) + offset,
+                                   size);
+  obj->wasm_size_ += size;
+}
+
+void WasmStreamingObject::Finish(const FunctionCallbackInfo<Value>& args) {
+  WasmStreamingObject* obj;
+  ASSIGN_OR_RETURN_UNWRAP(&obj, args.Holder());
+  CHECK(obj->streaming_);
+
+  CHECK_EQ(args.Length(), 0);
+  obj->streaming_->Finish();
+}
+
+void WasmStreamingObject::Abort(const FunctionCallbackInfo<Value>& args) {
+  WasmStreamingObject* obj;
+  ASSIGN_OR_RETURN_UNWRAP(&obj, args.Holder());
+  CHECK(obj->streaming_);
+
+  CHECK_EQ(args.Length(), 1);
+  obj->streaming_->Abort(args[0]);
+}
+
+void StartStreamingCompilation(const FunctionCallbackInfo<Value>& info) {
+  // V8 passes an instance of v8::WasmStreaming to this callback, which we can
+  // use to pass the WebAssembly module bytes to V8 as we receive them.
+  // Unfortunately, our fetch() implementation is a JavaScript dependency, so it
+  // is difficult to implement the required logic here. Instead, we create a
+  // a WasmStreamingObject that encapsulates v8::WasmStreaming and that we can
+  // pass to the JavaScript implementation. The JavaScript implementation can
+  // then push() bytes from the Response and eventually either finish() or
+  // abort() the operation.
+
+  // Create the wrapper object.
+  std::shared_ptr<WasmStreaming> streaming =
+      WasmStreaming::Unpack(info.GetIsolate(), info.Data());
+  Environment* env = Environment::GetCurrent(info);
+  Local<Object> obj;
+  if (!WasmStreamingObject::Create(env, streaming).ToLocal(&obj)) {
+    // A JavaScript exception is pending. Let V8 deal with it.
+    return;
+  }
+
+  // V8 always passes one argument to this callback.
+  CHECK_EQ(info.Length(), 1);
+
+  // Prepare the JavaScript implementation for invocation. We will pass the
+  // WasmStreamingObject as the first argument, followed by the argument that we
+  // received from V8, i.e., the first argument passed to compileStreaming (or
+  // instantiateStreaming).
+  Local<Function> impl = env->wasm_streaming_compilation_impl();
+  CHECK(!impl.IsEmpty());
+  Local<Value> args[] = {obj, info[0]};
+
+  // Hand control to the JavaScript implementation. It should never throw an
+  // error, but if it does, we leave it to the calling V8 code to handle that
+  // gracefully. Otherwise, we assert that the JavaScript function does not
+  // return anything.
+  MaybeLocal<Value> maybe_ret =
+      impl->Call(env->context(), info.This(), 2, args);
+  Local<Value> ret;
+  CHECK_IMPLIES(maybe_ret.ToLocal(&ret), ret->IsUndefined());
+}
+
+// Called once by JavaScript during initialization.
+void SetImplementation(const FunctionCallbackInfo<Value>& info) {
+  Environment* env = Environment::GetCurrent(info);
+  env->set_wasm_streaming_compilation_impl(info[0].As<Function>());
+}
+
+void Initialize(Local<Object> target,
+                Local<Value>,
+                Local<Context> context,
+                void*) {
+  Environment* env = Environment::GetCurrent(context);
+  env->SetMethod(target, "setImplementation", SetImplementation);
+}
+
+void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
+  registry->Register(SetImplementation);
+}
+
+}  // namespace wasm_web_api
+}  // namespace node
+
+NODE_MODULE_CONTEXT_AWARE_INTERNAL(wasm_web_api, node::wasm_web_api::Initialize)
+NODE_MODULE_EXTERNAL_REFERENCE(wasm_web_api,
+                               node::wasm_web_api::RegisterExternalReferences)

src/node_wasm_web_api.h

@@ -0,0 +1,54 @@
+#ifndef SRC_NODE_WASM_WEB_API_H_
+#define SRC_NODE_WASM_WEB_API_H_
+
+#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
+
+#include "base_object-inl.h"
+#include "v8.h"
+
+namespace node {
+namespace wasm_web_api {
+
+// Wrapper for interacting with a v8::WasmStreaming instance from JavaScript.
+class WasmStreamingObject final : public BaseObject {
+ public:
+  static v8::Local<v8::Function> Initialize(Environment* env);
+
+  static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
+
+  void MemoryInfo(MemoryTracker* tracker) const override;
+  SET_MEMORY_INFO_NAME(WasmStreamingObject)
+  SET_SELF_SIZE(WasmStreamingObject)
+
+  static v8::MaybeLocal<v8::Object> Create(
+      Environment* env, std::shared_ptr<v8::WasmStreaming> streaming);
+
+ private:
+  WasmStreamingObject(Environment* env, v8::Local<v8::Object> object)
+      : BaseObject(env, object) {
+    MakeWeak();
+  }
+
+  ~WasmStreamingObject() override {}
+
+ private:
+  static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Push(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Finish(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Abort(const v8::FunctionCallbackInfo<v8::Value>& args);
+
+  std::shared_ptr<v8::WasmStreaming> streaming_;
+  size_t wasm_size_;
+};
+
+// This is a v8::WasmStreamingCallback implementation that must be passed to
+// v8::Isolate::SetWasmStreamingCallback when setting up the isolate in order to
+// enable the WebAssembly.(compile|instantiate)Streaming APIs.
+void StartStreamingCompilation(const v8::FunctionCallbackInfo<v8::Value>& args);
+
+}  // namespace wasm_web_api
+}  // namespace node
+
+#endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
+
+#endif  // SRC_NODE_WASM_WEB_API_H_