DecompressionStream generates unhandled rejections

[ChALkeR]

This is different from #6020: that one is about firing the unhandledrejection event too early for rejections that get handled, this one is about a scenario where leaking rejections don't ever get handled.

For example on the testcase from nodejs/node#58247:

  const useFrom = Array.fromAsync // switch to false and it works

  let c = 0
  globalThis.addEventListener('rejectionhandled', (e) => {
    console.log('rejections', --c)
  })
  globalThis.addEventListener('unhandledrejection', (e) => {
    console.log('rejections', ++c)
  })

  const valid = new Uint8Array([120, 156, 75, 4, 0, 0, 98, 0, 98]) // deflate('a')
  const empty = new Uint8Array(1)
  const invalid = new Uint8Array([...valid, ...empty])
  const double = new Uint8Array([...valid, ...valid])
  for (const chunks of [[valid], [invalid], [valid, empty], [valid, valid], [double]]) {
    try {
      const stream = new Blob(chunks).stream().pipeThrough(new DecompressionStream('deflate'))
      console.log(await (useFrom ? Array.fromAsync(stream) : new Response(stream).blob()))
    } catch (e) {
      console.error(e)
    }
  }

  await new Promise((resolve) => setTimeout(resolve, 200))

[ChALkeR]

Note that this is unrelated to Array.fromAsync behavior

Somewhat more obvious testcase:

let c = 0
globalThis.addEventListener('rejectionhandled', (e) => {
  console.log('rejections', --c)
})
globalThis.addEventListener('unhandledrejection', (e) => {
  console.log('rejections', ++c)
})

export const Main = {
  async test(ctrl, env, ctx) {
    try {
      const stream = ReadableStream.from([]).pipeThrough(new DecompressionStream('deflate'))
      for await (const foo of stream);
    } catch (e) {
      console.error('caught:', e)
    }

    await new Promise((resolve) => setTimeout(resolve, 200))
    console.log('finished')
  },
}

This is what happens:

workerd/server/server.c++:5889: debug: [ TEST ] main:Main
rejections 1
rejections 0
rejections 1
rejections 0
caught: [TypeError: Called close() on a decompression stream with incomplete data]
rejections 1
workerd/io/worker.c++:2324: info: uncaught exception; source = Uncaught (in promise); stack = TypeError: Called close() on a decompression stream with incomplete data
workerd/io/io-context.c++:430: info: uncaught exception; exception = workerd/jsg/_virtual_includes/iterator/workerd/jsg/value.h:1480: failed: jsg.TypeError: Called close() on a decompression stream with incomplete data
finished
workerd/server/server.c++:5903: debug: [ FAIL ] main:Main (220.933ms)
Tests failed!

[ChALkeR]

I'm also unsure yet if it's just DecompressionStream that is broken or just transform streams in general.

[jasnell]

I don't have my laptop in front of me but it seems the flush state may not be getting called in this case. Will investigate tomorrow