Fixup TransformStream backpressure

[jasnell]

The TransformStream would originally start with backpressure enabled as it is supposed to but we failed to reapply backpressure when items were enqueued. This meant that after the first write to the TransformStream, no backpressure signaling would occur. Doh! This fixes it.

Unfortunately a compat flag is necessary because suddenly enforcing backpressure correctly could break existing workers that failed to realize it was broken. I noticed this because making the change broke one of our tests.

export default {
  async fetch(req, env, ctx) {

    async function sleep(ns) {
      await new Promise((res) => setTimeout(res, ns));
    }

    const strategy = new ByteLengthQueuingStrategy({ highWaterMark: 2024 });

    const { writable, readable } = new TransformStream(undefined, strategy);

    let writes = 0, reads = 0;

    // Start writing chunks to the stream
    async function write(writable) {
      const writer = writable.getWriter();
      for (let n = 0; n < 10000; n++) {
        console.log('waiting on backpressure', writer.ready, writer.desiredSize);
        // To correctly await on backpressure, wait on the writer.ready promise
        await writer.ready;
        console.log('backpressure cleared. writing a chunk');
        // You can also wait on the write.write but doing so is unnecessary if you're awaiting
        // on the writer.ready and can actually hurt performance by reducing writer throughput
        writer.write(crypto.getRandomValues(new Uint8Array(1024)));
        console.log('chunk written', ++writes, writer.desiredSize);
      }
      await writer.close();
      console.log('done writing');
    }

    async function read(readable) {
      console.log('reading pending...');
      await sleep(5_000);

      console.log('reading....');
      for await (const chunk of readable) {
        console.log(`read ${chunk.length}`, ++reads);
        await sleep(100);
      }
      console.log('done reading');
    }

    await Promise.all([
      write(writable),
      read(readable)
    ]);

    return new Response("Hello World\n");
  }
};

[byule]

A few questions - can you elaborate on what broke in the example? What change did we need to make for the test to pass?

Also why isn't there any updates to tests if we broke a test? Is the test not run with the compat flag?

[jasnell]

... Is the test not run with the compat flag?

Correct. The test doesn't use the compat flag. I'm separately working up a set of new tests to verify that backpressure is fully operational.

In the example given above, without the flag, the first write to the writable side of the transform would block as expected waiting on the first read. Once the first read occurred, however, the internal backpressure flag would be set to false and would never be set back to true, meaning the write loop would continue unchecked until all 10k writes occurred. Then all of the reads would happen. With the fix, each write would appropriately block for backpressure so we'd see a mix of writes and reads in step with each other, each read releasing backpressure and each write triggering backpressure.

[byule]

Right - everything you mention here makes sense, but I still don't understand why the test would fail. Does the test check for ordering? Regardless of whether back-pressure is working (data is written to stream, and data is read out of stream), I expect this test to functionally pass.

[jasnell]

The test fails with the compat flag because it relies on the write and close promises resolving before a transform stream is read from, which those should not because backpressure is being properly applied.