Fix GC visitation in stream readers and writers

src/workerd/api/streams/common.h

@@ -742,6 +742,12 @@ class ReaderLocked {
     return canceler;
   }
 
+  void clear() {
+    reader = kj::none;
+    closedFulfiller = kj::none;
+    canceler = kj::none;
+  }
+
 private:
   kj::Maybe<ReadableStreamController::Reader&> reader;
   kj::Maybe<jsg::Promise<void>::Resolver> closedFulfiller;
@@ -788,6 +794,12 @@ class WriterLocked {
     }
   }
 
+  void clear() {
+    writer = kj::none;
+    closedFulfiller = kj::none;
+    readyFulfiller = kj::none;
+  }
+
 private:
   kj::Maybe<WritableStreamController::Writer&> writer;
   kj::Maybe<jsg::Promise<void>::Resolver> closedFulfiller;

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

@@ -472,8 +472,7 @@ kj::Maybe<kj::Promise<DeferredProxy<void>>> WritableStreamSink::tryPumpFrom(
 // =======================================================================================
 
 ReadableStreamInternalController::~ReadableStreamInternalController() noexcept(false) {
-  KJ_IF_SOME(locked, readState.tryGet<ReaderLocked>()) {
-    auto lock = kj::mv(locked);
+  if (readState.is<ReaderLocked>()) {
     readState.init<Unlocked>();
   }
 }
@@ -801,12 +800,12 @@ void ReadableStreamInternalController::releaseReader(
           locked.getClosedFulfiller(),
           js.v8TypeError("This ReadableStream reader has been released."_kj));
     }
-    auto lock = kj::mv(locked);
+    locked.clear();
 
     // When maybeJs is nullptr, that means releaseReader was called when the reader is
     // being deconstructed and not as the result of explicitly calling releaseLock. In
     // that case, we don't want to change the lock state itself because we do not have
-    // an isolate lock. Moving the lock above will free the lock state while keeping the
+    // an isolate lock. Clearing the lock above will free the lock state while keeping the
     // ReadableStream marked as locked.
     if (maybeJs != kj::none) {
       readState.template init<Unlocked>();
@@ -815,8 +814,7 @@ void ReadableStreamInternalController::releaseReader(
 }
 
 WritableStreamInternalController::~WritableStreamInternalController() noexcept(false) {
-  KJ_IF_SOME(locked, writeState.tryGet<WriterLocked>()) {
-    auto lock = kj::mv(locked);
+  if (writeState.is<WriterLocked>()) {
     writeState.init<Unlocked>();
   }
 }
@@ -1288,12 +1286,12 @@ void WritableStreamInternalController::releaseWriter(
           locked.getClosedFulfiller(),
           js.v8TypeError("This WritableStream writer has been released."_kj));
     }
-    auto lock = kj::mv(locked);
+    locked.clear();
 
     // When maybeJs is nullptr, that means releaseWriter was called when the writer is
     // being deconstructed and not as the result of explicitly calling releaseLock and
     // we do not have an isolate lock. In that case, we don't want to change the lock
-    // state itself. Moving the lock above will free the lock state while keeping the
+    // state itself. Clearing the lock above will free the lock state while keeping the
     // WritableStream marked as locked.
     if (maybeJs != kj::none) {
       writeState.template init<Unlocked>();
@@ -1838,7 +1836,7 @@ void WritableStreamInternalController::visitForGc(jsg::GcVisitor& visitor) {
   for (auto& event : queue) {
     KJ_SWITCH_ONEOF(event.event) {
       KJ_CASE_ONEOF(write, Write) {
-        visitor.visit(write.promise, write.ref);
+        visitor.visit(write.promise);
       }
       KJ_CASE_ONEOF(close, Close) {
         visitor.visit(close.promise);

src/workerd/api/streams/internal.h

@@ -283,7 +283,6 @@ class WritableStreamInternalController: public WritableStreamController {
     kj::Maybe<jsg::Promise<void>::Resolver> promise;
     std::shared_ptr<v8::BackingStore> ownBytes;
     kj::ArrayPtr<const kj::byte> bytes;
-    kj::Maybe<jsg::Ref<WritableStream>> ref;
   };
   struct Close {
     kj::Maybe<jsg::Promise<void>::Resolver> promise;

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

@@ -155,6 +155,9 @@ void ReaderImpl::releaseLock(jsg::Lock& js) {
 }
 
 void ReaderImpl::visitForGc(jsg::GcVisitor& visitor) {
+  KJ_IF_SOME(readable, state.tryGet<Attached>()) {
+    visitor.visit(readable);
+  }
   visitor.visit(closedPromise);
 }
 

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

@@ -197,6 +197,9 @@ bool WritableStream::inspectExpectsBytes() {
 }
 
 void WritableStreamDefaultWriter::visitForGc(jsg::GcVisitor& visitor) {
+  KJ_IF_SOME(writable, state.tryGet<Attached>()) {
+    visitor.visit(writable);
+  }
   visitor.visit(closedPromise, readyPromise);
 }