Skip to content

Commit

Permalink
Account for the queue already having been drained
Browse files Browse the repository at this point in the history
When using the new proactive non-lazy drain compat flag, we
missed a few necessary checks during the write loop
  • Loading branch information
jasnell committed Aug 27, 2024
1 parent 8c02685 commit 8850e4b
Showing 1 changed file with 8 additions and 0 deletions.
8 changes: 8 additions & 0 deletions src/workerd/api/streams/internal.c++
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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>>();
Expand Down Expand Up @@ -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);
Expand Down

0 comments on commit 8850e4b

Please sign in to comment.