TextDecoderStream leaks a native decoder resource if its stream errors

[h4l]

I've come across what seems to be a bug in TextDecoderStream which allows it to leak the native decoder used by its TextDecoder.

I've made a test module to demonstrate the issue (repro code is at the bottom of that page, under the output): https://gist.github.com/h4l/0199ab7cc24dd13536e01c5ea98b3ae7
The 3 tests trigger the test runner's resource leak detection (a really nice feature!).

What seems to be happening is:

• TextDecoderStream uses a TextDecoder to decode its chunks
• TextDecoder.decode() creates a native decoder resource, which it holds open when used in streaming mode. It closes the resource when a non-streaming decode() call is made.
• TextDecoderStream makes streaming decode() calls in its transform() method, and makes a final non-streaming decode() call in its flush() method.
• When a stream pipeline is errored, the flush() method of any Transformer in a TransformStream is not called, so in the case of TextDecoderStream it has no way to know its no longer in use, and keeps open its decoder.

I was looking through the streams spec when I encountered the leak (before I worked out the cause) to try to work out if I was misusing the streams in some way. It seems to me like an oversight in the spec that Transformers have no way to be told to close/clean up when a stream doesn't close cleanly. flush() is only called when the stream closes normally if I've not missed something, and there are no other lifecycle methods available for Transformers.

I can't see any idiomatic way to tell the Transformer to close, but one approach could be to wrap the readable and writable streams of the TransformerStream to watch for close/cancel/abort calls.

I've not made any previous contributions, but I'd be happy to help with a PR to fix this if it'd be useful.

[h4l]

I played around with the Streams API a bit and came up with a fairly straightforward way to implement a TransformStream whose Transformer gets notified of stream aborts. Basically two parts:

• A WritableStream can be monitored for errors by wrapping it with another WritableStream that opens a reader on the monitored stream, exposes the reader's closed promise (which rejects if the monitored stream is aborted), and forwards start/write/close/abort calls to the monitored stream.
  That looks like this: https://deno.land/x/shutdown_aware_transform_stream@1.0.0/shutdown_monitor_writable_stream.ts
• Then a TransformStream can react to stream aborts by monitoring its writable side with the monitor stream, and using the closed promise to be notified when the stream aborts.
  That looks like this: https://deno.land/x/shutdown_aware_transform_stream@1.0.0/shutdown_aware_transform_stream.ts#L98

Would you be interested in a PR for TextDecoderStream that fixes this issue by using a cut down/minimal version of the above to close the TextDecoder on aborts?

Or alternatively, it seems reasonable to me to suggest/request a change to the WHATWG Streams spec to give the controller of a TransformStream a property exposing an AbortSignal which could be used to handle stream aborts. WritableStreamDefaultController provides this already, but TransformStreamDefaultController doesn't. (And or to give Transformer objects an abort() method, like a WritableStream underlying sinks have.)

What do you think?

[crowlKats]

This seems to me like something that should be changed in the specification. It does seem to me like a good addition. My bad, flush should take care of that actually already

[h4l]

As I understand it, flush is only called when the stream closes normally, it's not called when the stream aborts: https://streams.spec.whatwg.org/#transform-stream-error
Or do you mean you think the spec should change to call flush when the stream aborts?

[h4l]

I'll bring it up with the Streams spec team.

[nkronlage]

In case this helps anyone hitting this issue when breaking out of async iteration, I was able to work around it with a custom decoder transform stream:

  const MyTextDecoderStream = () => {                                                
    const textDecoder = new TextDecoder();                                           
    return new TransformStream({                                                     
      transform(chunk : Uint8Array, controller: TransformStreamDefaultController) {  
        controller.enqueue(textDecoder.decode(chunk));                               
      },                                                                                                                          
      flush(controller: TransformStreamDefaultController) {                          
        controller.enqueue(textDecoder.decode());                                    
      }                                                                              
    });                                                                              
  }; 

[nounder]

Alternative solution with manual encoding from duplicated #19074

Deno.test("working alternative", async () => {
  const res = await fetch(
    "https://deno.land/std@0.186.0/json/testdata/test.jsonl"
  )

  const textDecoder = new TextDecoder()
  const reader = res.body!.getReader()
  const b = await reader.read()
  const t = await textDecoder.decode(b.value!)

  await reader.cancel()
})