Add readable-source/writable-sink adapters

This is the first step towards a significant refactoring of the
streams implementation with the goal of making it safer and more
performance.

This commits adds four new classes:

* `ReadableStreamSourceKjAdapter`: Adapts a ReadableStream to the
  `ReadableStreamSource` API, fully encapsulating the JS interactions
  behind the KJ-friendly interface.
* `ReadableStreamSourceJsAdapter`: Adapts a `ReadableStreamSource`
  to a more friendly JS interface.
* `WritableStreamSinkKjAdapter`: Adapts a WritableStream to the
  `WritableStreamSink` API, fully encapsulating the JS interactions
  behind the KJ-friendly interface.
* `WritableStreamSinkJsAdapter`: Adapts a `WritableStreamSink`
  to a more friendly JS interface.

Also adds tests.

This commit does not update any existing code to use these adapters
yet, that will come in follow-up commits.

There are several key goals for this refactoring:

* Make the streams implementation safer by using the new checked
  Queue for buffering operations and removing direct reliance on
  raw references.
* Reducing the overall complexity of the streams implementation
  through better encapsulation and separation of concerns.

These require very careful review to ensure the behavior is correct
before we start switching existing code over to use these adapters.
In future commits I will be incrementally removing the old code
and switching to use these adapters.

Based on Harris' feedback, I've updated the implementation
of the ReadableStreamSourceKjAdapter to include a minimum
read policy. When the policy is IMMEDIATE, the adapter will
always return immediately after minimally fulfilling the
minimum read, even if more data is available and there is
more room in buffer we're reading into. If the policy is
OPPORTUNISTIC, then the adapter will attempt to fill the
buffer as much as possible (up until there is <= 512
bytes remaining in the buffer or we hit maxBytes) while
we are holding the isolate lock. The default policy is
OPPORTUNISTIC, with the idea being that we should do as
much work as possible while we have the isolate lock
before returning to the kj event loop.

Author: jasnell

src/workerd/api/BUILD.bazel

@@ -347,6 +347,24 @@ kj_test(
     deps = ["//src/workerd/tests:test-fixture"],
 )
 
+kj_test(
+    src = "streams/writable-sink-adapter-test.c++",
+    deps = [
+        "//src/workerd/io",
+        "//src/workerd/jsg",
+        "//src/workerd/tests:test-fixture",
+    ],
+)
+
+kj_test(
+    src = "streams/readable-source-adapter-test.c++",
+    deps = [
+        "//src/workerd/io",
+        "//src/workerd/jsg",
+        "//src/workerd/tests:test-fixture",
+    ],
+)
+
 wd_test(
     src = "actor-alarms-delete-test.wd-test",
     args = ["--experimental"],

src/workerd/api/streams/common.h

@@ -593,7 +593,8 @@ class WritableStreamController {
     //
     // The controller is guaranteed to either outlive the Writer or will detach the Writer so the
     // WritableStreamController& reference should always remain valid.
-    virtual void attach(WritableStreamController& controller,
+    virtual void attach(jsg::Lock& js,
+        WritableStreamController& controller,
         jsg::Promise<void> closedPromise,
         jsg::Promise<void> readyPromise) = 0;
 
@@ -603,7 +604,7 @@ class WritableStreamController {
 
     // The ready promise can be replaced whenever backpressure is signaled by the underlying
     // controller.
-    virtual void replaceReadyPromise(jsg::Promise<void> readyPromise) = 0;
+    virtual void replaceReadyPromise(jsg::Lock& js, jsg::Promise<void> readyPromise) = 0;
   };
 
   struct PendingAbort {
@@ -821,10 +822,10 @@ class WriterLocked {
     return readyFulfiller;
   }
 
-  void setReadyFulfiller(jsg::PromiseResolverPair<void>& pair) {
+  void setReadyFulfiller(jsg::Lock& js, jsg::PromiseResolverPair<void>& pair) {
     KJ_IF_SOME(w, writer) {
       readyFulfiller = kj::mv(pair.resolver);
-      w.replaceReadyPromise(kj::mv(pair.promise));
+      w.replaceReadyPromise(js, kj::mv(pair.promise));
     }
   }
 

src/workerd/api/streams/internal.c++

@@ -1023,7 +1023,7 @@ void WritableStreamInternalController::updateBackpressure(jsg::Lock& js, bool ba
       // the existing one is resolved or not.
       auto prp = js.newPromiseAndResolver<void>();
       prp.promise.markAsHandled(js);
-      writerLock.setReadyFulfiller(prp);
+      writerLock.setReadyFulfiller(js, prp);
       return;
     }
 
@@ -1406,7 +1406,7 @@ bool WritableStreamInternalController::lockWriter(jsg::Lock& js, Writer& writer)
   }
 
   writeState = kj::mv(lock);
-  writer.attach(*this, kj::mv(closedPrp.promise), kj::mv(readyPrp.promise));
+  writer.attach(js, *this, kj::mv(closedPrp.promise), kj::mv(readyPrp.promise));
   return true;
 }