WIP: Remove StreamSink

DO NOT MERGE until the autogate introduced in #5738 is fully rolled out and working.

Until then I'm putting this up just to show what we'll be able to remove.

Author: kentonv

src/workerd/api/basics.c++

@@ -572,56 +572,6 @@ class AbortTriggerRpcClient final {
   rpc::AbortTrigger::Client client;
 };
 
-namespace {
-// The jsrpc handler that receives aborts from the remote and triggers them locally
-//
-// TODO(cleanup): This class has been copied to external-pusher.c++. The copy here can be
-//   deleted as soon as we've switched from StreamSink to ExternalPusher and can delete all the
-//   StreamSink-related code. For now I'm not trying to avoid duplication.
-class AbortTriggerRpcServer final: public rpc::AbortTrigger::Server {
- public:
-  AbortTriggerRpcServer(kj::Own<kj::PromiseFulfiller<void>> fulfiller,
-      kj::Own<AbortSignal::PendingReason>&& pendingReason)
-      : fulfiller(kj::mv(fulfiller)),
-        pendingReason(kj::mv(pendingReason)) {}
-
-  kj::Promise<void> abort(AbortContext abortCtx) override {
-    auto params = abortCtx.getParams();
-    auto reason = params.getReason().getV8Serialized();
-
-    pendingReason->getWrapped() = kj::heapArray(reason.asBytes());
-    fulfiller->fulfill();
-    return kj::READY_NOW;
-  }
-
-  kj::Promise<void> release(ReleaseContext releaseCtx) override {
-    released = true;
-    return kj::READY_NOW;
-  }
-
-  ~AbortTriggerRpcServer() noexcept(false) {
-    if (pendingReason->getWrapped() != nullptr) {
-      // Already triggered
-      return;
-    }
-
-    if (!released) {
-      pendingReason->getWrapped() = JSG_KJ_EXCEPTION(FAILED, DOMAbortError,
-          "An AbortSignal received over RPC was implicitly aborted because the connection back to "
-          "its trigger was lost.");
-    }
-
-    // Always fulfill the promise in case the AbortSignal was waiting
-    fulfiller->fulfill();
-  }
-
- private:
-  kj::Own<kj::PromiseFulfiller<void>> fulfiller;
-  kj::Own<AbortSignal::PendingReason> pendingReason;
-  bool released = false;
-};
-}  // namespace
-
 AbortSignal::AbortSignal(kj::Maybe<kj::Exception> exception,
     jsg::Optional<jsg::JsRef<jsg::JsValue>> maybeReason,
     Flag flag)
@@ -862,28 +812,19 @@ void AbortSignal::serialize(jsg::Lock& js, jsg::Serializer& serializer) {
     return;
   }
 
-  auto triggerCap = [&]() -> rpc::AbortTrigger::Client {
-    KJ_IF_SOME(pusher, externalHandler->getExternalPusher()) {
-      auto pipeline = pusher.pushAbortSignalRequest(capnp::MessageSize{2, 0}).sendForPipeline();
-
-      externalHandler->write(
-          [signal = pipeline.getSignal()](rpc::JsValue::External::Builder builder) mutable {
-        builder.setAbortSignal(kj::mv(signal));
-      });
+  auto pipeline = externalHandler->getExternalPusher()
+                      .pushAbortSignalRequest(capnp::MessageSize{2, 0})
+                      .sendForPipeline();
 
-      return pipeline.getTrigger();
-    } else {
-      return externalHandler
-          ->writeStream([&](rpc::JsValue::External::Builder builder) mutable {
-        builder.setAbortTrigger();
-      }).castAs<rpc::AbortTrigger>();
-    }
-  }();
+  externalHandler->write(
+      [signal = pipeline.getSignal()](rpc::JsValue::External::Builder builder) mutable {
+    builder.setAbortSignal(kj::mv(signal));
+  });
 
   auto& ioContext = IoContext::current();
   // Keep track of every AbortSignal cloned from this one.
   // If this->triggerAbort(...) is called, each rpcClient will be informed.
-  rpcClients.add(ioContext.addObject(kj::heap<AbortTriggerRpcClient>(kj::mv(triggerCap))));
+  rpcClients.add(ioContext.addObject(kj::heap<AbortTriggerRpcClient>(pipeline.getTrigger())));
 }
 
 jsg::Ref<AbortSignal> AbortSignal::deserialize(
@@ -914,24 +855,12 @@ jsg::Ref<AbortSignal> AbortSignal::deserialize(
   auto& ioctx = IoContext::current();
 
   auto reader = externalHandler->read();
-  if (reader.isAbortTrigger()) {
-    // Old-style StreamSink.
-    // TODO(cleanup): Remove this once the ExternalPusher autogate has rolled out.
-    auto paf = kj::newPromiseAndFulfiller<void>();
-    auto pendingReason = ioctx.addObject(kj::refcounted<PendingReason>());
-
-    externalHandler->setLastStream(
-        kj::heap<AbortTriggerRpcServer>(kj::mv(paf.fulfiller), kj::addRef(*pendingReason)));
-    signal->rpcAbortPromise = ioctx.addObject(kj::heap(kj::mv(paf.promise)));
-    signal->pendingReason = kj::mv(pendingReason);
-  } else {
-    KJ_REQUIRE(reader.isAbortSignal(), "external table slot type does't match serialization tag");
+  KJ_REQUIRE(reader.isAbortSignal(), "external table slot type does't match serialization tag");
 
-    auto resolvedSignal = ioctx.getExternalPusher()->unwrapAbortSignal(reader.getAbortSignal());
+  auto resolvedSignal = ioctx.getExternalPusher()->unwrapAbortSignal(reader.getAbortSignal());
 
-    signal->rpcAbortPromise = ioctx.addObject(kj::heap(kj::mv(resolvedSignal.signal)));
-    signal->pendingReason = ioctx.addObject(kj::mv(resolvedSignal.reason));
-  }
+  signal->rpcAbortPromise = ioctx.addObject(kj::heap(kj::mv(resolvedSignal.signal)));
+  signal->pendingReason = ioctx.addObject(kj::mv(resolvedSignal.reason));
 
   return signal;
 }

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

@@ -519,93 +519,6 @@ jsg::Optional<uint32_t> ByteLengthQueuingStrategy::size(
 
 namespace {
 
-// TODO(cleanup): These classes have been copied to external-pusher.c++. The copies here can be
-//   deleted as soon as we've switched from StreamSink to ExternalPusher and can delete all the
-//   StreamSink-related code. For now I'm not trying to avoid duplication.
-
-// HACK: We need as async pipe, like kj::newOneWayPipe(), except supporting explicit end(). So we
-//   wrap the two ends of the pipe in special adapters that track whether end() was called.
-class ExplicitEndOutputPipeAdapter final: public capnp::ExplicitEndOutputStream {
- public:
-  ExplicitEndOutputPipeAdapter(
-      kj::Own<kj::AsyncOutputStream> inner, kj::Own<kj::RefcountedWrapper<bool>> ended)
-      : inner(kj::mv(inner)),
-        ended(kj::mv(ended)) {}
-
-  kj::Promise<void> write(kj::ArrayPtr<const byte> buffer) override {
-    return KJ_REQUIRE_NONNULL(inner)->write(buffer);
-  }
-  kj::Promise<void> write(kj::ArrayPtr<const kj::ArrayPtr<const byte>> pieces) override {
-    return KJ_REQUIRE_NONNULL(inner)->write(pieces);
-  }
-
-  kj::Maybe<kj::Promise<uint64_t>> tryPumpFrom(
-      kj::AsyncInputStream& input, uint64_t amount) override {
-    return KJ_REQUIRE_NONNULL(inner)->tryPumpFrom(input, amount);
-  }
-
-  kj::Promise<void> whenWriteDisconnected() override {
-    return KJ_REQUIRE_NONNULL(inner)->whenWriteDisconnected();
-  }
-
-  kj::Promise<void> end() override {
-    // Signal to the other side that end() was actually called.
-    ended->getWrapped() = true;
-    inner = kj::none;
-    return kj::READY_NOW;
-  }
-
- private:
-  kj::Maybe<kj::Own<kj::AsyncOutputStream>> inner;
-  kj::Own<kj::RefcountedWrapper<bool>> ended;
-};
-
-class ExplicitEndInputPipeAdapter final: public kj::AsyncInputStream {
- public:
-  ExplicitEndInputPipeAdapter(kj::Own<kj::AsyncInputStream> inner,
-      kj::Own<kj::RefcountedWrapper<bool>> ended,
-      kj::Maybe<uint64_t> expectedLength)
-      : inner(kj::mv(inner)),
-        ended(kj::mv(ended)),
-        expectedLength(expectedLength) {}
-
-  kj::Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) override {
-    size_t result = co_await inner->tryRead(buffer, minBytes, maxBytes);
-
-    KJ_IF_SOME(l, expectedLength) {
-      KJ_ASSERT(result <= l);
-      l -= result;
-      if (l == 0) {
-        // If we got all the bytes we expected, we treat this as a successful end, because the
-        // underlying KJ pipe is not actually going to wait for the other side to drop. This is
-        // consistent with the behavior of Content-Length in HTTP anyway.
-        ended->getWrapped() = true;
-      }
-    }
-
-    if (result < minBytes) {
-      // Verify that end() was called.
-      if (!ended->getWrapped()) {
-        JSG_FAIL_REQUIRE(Error, "ReadableStream received over RPC disconnected prematurely.");
-      }
-    }
-    co_return result;
-  }
-
-  kj::Maybe<uint64_t> tryGetLength() override {
-    return inner->tryGetLength();
-  }
-
-  kj::Promise<uint64_t> pumpTo(kj::AsyncOutputStream& output, uint64_t amount) override {
-    return inner->pumpTo(output, amount);
-  }
-
- private:
-  kj::Own<kj::AsyncInputStream> inner;
-  kj::Own<kj::RefcountedWrapper<bool>> ended;
-  kj::Maybe<uint64_t> expectedLength;
-};
-
 // Wrapper around ReadableStreamSource that prevents deferred proxying. We need this for RPC
 // streams because although they are "system streams", they become disconnected when the IoContext
 // is destroyed, due to the JsRpcCustomEvent being canceled.
@@ -680,37 +593,21 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) {
   StreamEncoding encoding = controller.getPreferredEncoding();
   auto expectedLength = controller.tryGetLength(encoding);
 
-  capnp::ByteStream::Client streamCap = [&]() {
-    KJ_IF_SOME(pusher, externalHandler->getExternalPusher()) {
-      auto req = pusher.pushByteStreamRequest(capnp::MessageSize{2, 0});
-      KJ_IF_SOME(el, expectedLength) {
-        req.setLengthPlusOne(el + 1);
-      }
-      auto pipeline = req.sendForPipeline();
-
-      externalHandler->write([encoding, expectedLength, source = pipeline.getSource()](
-                                 rpc::JsValue::External::Builder builder) mutable {
-        auto rs = builder.initReadableStream();
-        rs.setStream(kj::mv(source));
-        rs.setEncoding(encoding);
-      });
-
-      return pipeline.getSink();
-    } else {
-      return externalHandler
-          ->writeStream(
-              [encoding, expectedLength](rpc::JsValue::External::Builder builder) mutable {
-        auto rs = builder.initReadableStream();
-        rs.setEncoding(encoding);
-        KJ_IF_SOME(l, expectedLength) {
-          rs.getExpectedLength().setKnown(l);
-        }
-      }).castAs<capnp::ByteStream>();
-    }
-  }();
+  auto req = externalHandler->getExternalPusher().pushByteStreamRequest(capnp::MessageSize{2, 0});
+  KJ_IF_SOME(el, expectedLength) {
+    req.setLengthPlusOne(el + 1);
+  }
+  auto pipeline = req.sendForPipeline();
+
+  externalHandler->write([encoding, expectedLength, source = pipeline.getSource()](
+                             rpc::JsValue::External::Builder builder) mutable {
+    auto rs = builder.initReadableStream();
+    rs.setStream(kj::mv(source));
+    rs.setEncoding(encoding);
+  });
 
   kj::Own<capnp::ExplicitEndOutputStream> kjStream =
-      ioctx.getByteStreamFactory().capnpToKjExplicitEnd(kj::mv(streamCap));
+      ioctx.getByteStreamFactory().capnpToKjExplicitEnd(pipeline.getSink());
 
   auto sink = newSystemStream(kj::mv(kjStream), encoding, ioctx);
 
@@ -741,25 +638,7 @@ jsg::Ref<ReadableStream> ReadableStream::deserialize(
 
   auto& ioctx = IoContext::current();
 
-  kj::Own<kj::AsyncInputStream> in;
-  if (rs.hasStream()) {
-    in = ioctx.getExternalPusher()->unwrapStream(rs.getStream());
-  } else {
-    kj::Maybe<uint64_t> expectedLength;
-    auto el = rs.getExpectedLength();
-    if (el.isKnown()) {
-      expectedLength = el.getKnown();
-    }
-
-    auto pipe = kj::newOneWayPipe(expectedLength);
-
-    auto endedFlag = kj::refcounted<kj::RefcountedWrapper<bool>>(false);
-
-    auto out = kj::heap<ExplicitEndOutputPipeAdapter>(kj::mv(pipe.out), kj::addRef(*endedFlag));
-    in = kj::heap<ExplicitEndInputPipeAdapter>(kj::mv(pipe.in), kj::mv(endedFlag), expectedLength);
-
-    externalHandler->setLastStream(ioctx.getByteStreamFactory().kjToCapnp(kj::mv(out)));
-  }
+  auto in = ioctx.getExternalPusher()->unwrapStream(rs.getStream());
 
   return js.alloc<ReadableStream>(ioctx,
       kj::heap<NoDeferredProxyReadableStream>(newSystemStream(kj::mv(in), encoding, ioctx), ioctx));

src/workerd/api/tests/BUILD.bazel

@@ -157,15 +157,6 @@ wd_test(
     data = ["abortsignal-test.js"],
 )
 
-wd_test(
-    src = "abortsignal-external-pusher-test.wd-test",
-    args = ["--experimental"],
-    data = [
-        "abortsignal-test.js",
-        "abortsignal-test.wd-test",
-    ],
-)
-
 wd_test(
     src = "actor-stub-test.wd-test",
     args = ["--experimental"],
@@ -284,15 +275,6 @@ wd_test(
     data = ["js-rpc-test.js"],
 )
 
-wd_test(
-    src = "js-rpc-external-pusher-test.wd-test",
-    args = ["--experimental"],
-    data = [
-        "js-rpc-test.js",
-        "js-rpc-test.wd-test",
-    ],
-)
-
 wd_test(
     src = "js-rpc-params-ownership-test.wd-test",
     args = ["--experimental"],