diff --git a/src/jsc/bindings/webcore/streams/JSCompressionStreamShared.cpp b/src/jsc/bindings/webcore/streams/JSCompressionStreamShared.cpp index e143b7e29a57..c5cfc4154f77 100644 --- a/src/jsc/bindings/webcore/streams/JSCompressionStreamShared.cpp +++ b/src/jsc/bindings/webcore/streams/JSCompressionStreamShared.cpp @@ -187,11 +187,11 @@ static CodecOutcome stepChunkHere(JSGlobalObject* globalObject, JSTransformStrea auto& vm = getVM(globalObject); auto scope = DECLARE_THROW_SCOPE(vm); ASSERT(stream->m_nativeStateInUse); - bool continuation = !!stream->m_codecPromise; + ASSERT(stream->m_codecPromise); for (;;) { CodecStepResult step = runStepHere(globalObject, stream, coder, input, inputLen, finish); RETURN_IF_EXCEPTION(scope, CodecOutcome::Failed); - if (continuation && !stream->m_codecPromise) + if (!stream->m_codecPromise) return CodecOutcome::Pending; if (!step.thrown.isEmpty()) { thrown = step.thrown; @@ -236,7 +236,7 @@ static void settleCodecChunk(JSGlobalObject* globalObject, JSTransformStream* st resolvePromise(globalObject, promise, jsUndefined()); } -// Applies the outcome of a step that ran from a later turn to the pending chunk. +// Applies a round of stepping to the chunk's promise. static void settlePendingChunk(JSGlobalObject* globalObject, JSTransformStream* stream, CodecOutcome outcome, JSValue thrown) { auto& vm = getVM(globalObject); @@ -265,9 +265,11 @@ static JSPromise* transformChunk(JSGlobalObject* globalObject, JSTransformStream ASSERT(!stream->m_codecPromise); ASSERT(coder); + // Registered before the first step: delivering a step can run user code (an enqueue may + // resolve a read), and a terminal reached from there abandons the chunk through it. + auto* promise = JSPromise::create(vm, globalObject->promiseStructure()); + stream->m_codecPromise.set(vm, stream, promise); if (inputLen > kAsyncCodecThreshold) { - auto* promise = JSPromise::create(vm, globalObject->promiseStructure()); - stream->m_codecPromise.set(vm, stream, promise); stream->m_codecChunkOffThread = true; dispatchStepOffThread(globalObject, stream, coder, chunk, input, inputLen, finish); scope.assertNoException(); @@ -277,23 +279,8 @@ static JSPromise* transformChunk(JSGlobalObject* globalObject, JSTransformStream JSValue thrown; CodecOutcome outcome = stepChunkHere(globalObject, stream, coder, input, inputLen, finish, thrown); RETURN_IF_EXCEPTION(scope, nullptr); - switch (outcome) { - case CodecOutcome::Done: - RELEASE_AND_RETURN(scope, promiseFulfilledWith(globalObject, jsUndefined())); - case CodecOutcome::Failed: - RELEASE_AND_RETURN(scope, promiseRejectedWith(globalObject, thrown)); - case CodecOutcome::DoneSinkFull: { - auto* ready = JSPromise::create(vm, globalObject->promiseStructure()); - stream->m_nativeSinkReadyPromise.set(vm, stream, ready); - return ready; - } - case CodecOutcome::Pending: { - auto* promise = JSPromise::create(vm, globalObject->promiseStructure()); - stream->m_codecPromise.set(vm, stream, promise); - return promise; - } - } - RELEASE_ASSERT_NOT_REACHED(); + settlePendingChunk(globalObject, stream, outcome, thrown); + return promise; } void nativeCodecContinue(JSGlobalObject* globalObject, JSTransformStream* stream) diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index b70023f919bd..91acc80c2c1d 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -852,6 +852,39 @@ describe("bounded output per input chunk", () => { await expect(writer.closed).rejects.toMatchObject({ name: "AbortError" }); }); + // Delivering the first piece resolves the pending read(), and resolving a promise + // with an object looks up its `then`, so user code can abort from inside the + // chunk's very first step, before it has been parked. + test("writer.abort() re-entered from the chunk's first step settles the write", async () => { + const ds = new DecompressionStream("brotli"); + const writer = ds.writable.getWriter(); + const reader = ds.readable.getReader(); + const read = reader.read(); + + let aborted: Promise | undefined; + Object.defineProperty(Object.prototype, "then", { + configurable: true, + get() { + delete (Object.prototype as any).then; + aborted = writer.abort(new Error("stop")); + return undefined; + }, + }); + let write: Promise; + try { + write = writer.write(bombs.brotli()); + expect((await read).value!.byteLength).toBeLessThanOrEqual(kDefaultHighWaterMark); + } finally { + delete (Object.prototype as any).then; + } + expect(aborted).toBeDefined(); + + expect(await write).toBeUndefined(); + await aborted; + await expect(writer.closed).rejects.toThrow("stop"); + await expect(reader.read()).rejects.toThrow("stop"); + }); + // An abort during close() is different: the close in progress wins, so a flush // being drained keeps going and the reader still gets all of it. test("writer.abort() during a multi-step flush does not truncate it", async () => {