CompressionStream: register a chunk's promise before its first codec step

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)

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<void> | 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<void>;
+    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 () => {