From 8850e4bdb2c86520d3623b0b382c5bcf7ba550e8 Mon Sep 17 00:00:00 2001
From: James M Snell <jasnell@gmail.com>
Date: Mon, 26 Aug 2024 17:13:30 -0700
Subject: [PATCH] Account for the queue already having been drained

When using the new proactive non-lazy drain compat flag, we
missed a few necessary checks during the write loop
---
 src/workerd/api/streams/internal.c++ | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/src/workerd/api/streams/internal.c++ b/src/workerd/api/streams/internal.c++
index de892b1d3c2..f0e19f33084 100644
--- a/src/workerd/api/streams/internal.c++
+++ b/src/workerd/api/streams/internal.c++
@@ -1590,6 +1590,8 @@ jsg::Promise<void> WritableStreamInternalController::writeLoopAfterFrontOutputLo
           .then(js,
               ioContext.addFunctor(
                   [this, check, maybeAbort, amountToWrite](jsg::Lock& js) -> jsg::Promise<void> {
+        // Under some conditions, the clean up has already happened.
+        if (queue.empty()) return js.resolvedPromise();
         auto& request = check();
         maybeResolvePromise(js, request.promise);
         decreaseCurrentWriteBufferSize(js, amountToWrite);
@@ -1599,6 +1601,8 @@ jsg::Promise<void> WritableStreamInternalController::writeLoopAfterFrontOutputLo
       }),
               ioContext.addFunctor([this, check, maybeAbort, amountToWrite](
                                        jsg::Lock& js, jsg::Value reason) -> jsg::Promise<void> {
+        // Under some conditions, the clean up has already happened.
+        if (queue.empty()) return js.resolvedPromise();
         auto handle = reason.getHandle(js);
         auto& request = check();
         auto& writable = state.get<IoOwn<Writable>>();
@@ -1742,12 +1746,16 @@ jsg::Promise<void> WritableStreamInternalController::writeLoopAfterFrontOutputLo
 
       return ioContext.awaitIo(js, writable->canceler.wrap(writable->sink->end()))
           .then(js, ioContext.addFunctor([this, check](jsg::Lock& js) {
+        // Under some conditions, the clean up has already happened.
+        if (queue.empty()) return;
         auto& request = check();
         maybeResolvePromise(js, request.promise);
         queue.pop_front();
         finishClose(js);
       }),
               ioContext.addFunctor([this, check](jsg::Lock& js, jsg::Value reason) {
+        // Under some conditions, the clean up has already happened.
+        if (queue.empty()) return;
         auto handle = reason.getHandle(js);
         auto& request = check();
         maybeRejectPromise<void>(js, request.promise, handle);