deno TextDecoderStream resource leak reproduction

output

$ deno --version
deno 1.17.0+3db18bf (canary, x86_64-apple-darwin)
v8 9.7.106.15
typescript 4.5.2

$ deno test text_decoder_stream_behaviour_test.ts
Check file:///private/tmp/text_decoder_stream_behaviour_test.ts
running 3 tests from file:///private/tmp/text_decoder_stream_behaviour_test.ts
test 1: TextDecoderStream leaks its TextDecoder's native decoder when a stream in its pipeline errors ... FAILED (17ms)
test 2: TransformStream using TextDecoder directly leaks a native decoder when a stream in its pipeline errors ... FAILED (5ms)
test 3: TextDecoderStream leaks its TextDecoder's native decoder when its downstream pipe is aborted ... FAILED (9ms)

failures:

1: TextDecoderStream leaks its TextDecoder's native decoder when a stream in its pipeline errors
AssertionError: Test case is leaking resources.
Before: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr"
}
After: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr",
  "3": "textDecoder"
}

Make sure to close all open resource handles returned from Deno APIs before
finishing test case.
    at assert (deno:runtime/js/06_util.js:41:13)
    at resourceSanitizer (deno:runtime/js/40_testing.js:154:7)
    at async Object.exitSanitizer [as fn] (deno:runtime/js/40_testing.js:170:9)
    at async runTest (deno:runtime/js/40_testing.js:428:7)
    at async Object.runTests (deno:runtime/js/40_testing.js:541:22)

2: TransformStream using TextDecoder directly leaks a native decoder when a stream in its pipeline errors
AssertionError: Test case is leaking resources.
Before: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr",
  "3": "textDecoder"
}
After: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr",
  "3": "textDecoder",
  "5": "textDecoder"
}

Make sure to close all open resource handles returned from Deno APIs before
finishing test case.
    at assert (deno:runtime/js/06_util.js:41:13)
    at resourceSanitizer (deno:runtime/js/40_testing.js:154:7)
    at async Object.exitSanitizer [as fn] (deno:runtime/js/40_testing.js:170:9)
    at async runTest (deno:runtime/js/40_testing.js:428:7)
    at async Object.runTests (deno:runtime/js/40_testing.js:541:22)

3: TextDecoderStream leaks its TextDecoder's native decoder when its downstream pipe is aborted
AssertionError: Test case is leaking resources.
Before: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr",
  "3": "textDecoder",
  "5": "textDecoder"
}
After: {
  "0": "stdin",
  "1": "stdout",
  "2": "stderr",
  "3": "textDecoder",
  "5": "textDecoder",
  "7": "textDecoder"
}

Make sure to close all open resource handles returned from Deno APIs before
finishing test case.
    at assert (deno:runtime/js/06_util.js:41:13)
    at resourceSanitizer (deno:runtime/js/40_testing.js:154:7)
    at async Object.exitSanitizer [as fn] (deno:runtime/js/40_testing.js:170:9)
    at async runTest (deno:runtime/js/40_testing.js:428:7)
    at async Object.runTests (deno:runtime/js/40_testing.js:541:22)

failures:

        1: TextDecoderStream leaks its TextDecoder's native decoder when a stream in its pipeline errors
        2: TransformStream using TextDecoder directly leaks a native decoder when a stream in its pipeline errors
        3: TextDecoderStream leaks its TextDecoder's native decoder when its downstream pipe is aborted

test result: FAILED. 0 passed; 3 failed; 0 ignored; 0 measured; 0 filtered out (71ms)

error: Test failed

text_decoder_stream_behaviour_test.ts

import {
  assert,
  assertEquals,
  assertMatch,
  assertRejects,
} from "https://deno.land/std@0.118.0/testing/asserts.ts";

/*
 * These tests triggers a leaked resource error from the test runner for a
 * "textDecoder". What's 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.
 */

Deno.test(
  "1: TextDecoderStream leaks its TextDecoder's native decoder when a stream in its pipeline errors",
  async () => {
    const src = new ReadableStream({
      pull(controller) {
        controller.enqueue(encode("foo"));
      },
    });
    const dst = new WritableStream({
      write(chunk) {
        assertEquals(chunk, "foo");
        throw new Error("example");
      },
    });

    await assertRejects(
      async () => await src.pipeThrough(new TextDecoderStream()).pipeTo(dst),
      Error,
      "example",
    );
  },
);

// This demonstrates what's happening with TextDecoderStream by using
// TextDecoder directly in roughly the same way it does.
Deno.test(
  "2: TransformStream using TextDecoder directly leaks a native decoder when a stream in its pipeline errors",
  async () => {
    let flushCalled = false;

    const decoder = new TextDecoder();
    const transformer: Transformer<Uint8Array, string> = {
      transform(chunk, controller) {
        controller.enqueue(decoder.decode(chunk, { stream: true }));
      },
      flush(controller) {
        flushCalled = true;
        controller.enqueue(
          decoder.decode(undefined, { stream: false }),
        );
      },
    };
    const src = new ReadableStream({
      pull(controller) {
        controller.enqueue(encode("foo"));
      },
    });
    const dst = new WritableStream({
      write(chunk) {
        assertEquals(chunk, "foo");
        throw new Error("example");
      },
    });

    await assertRejects(
      async () =>
        await src.pipeThrough(new TransformStream(transformer)).pipeTo(dst),
      Error,
      "example",
    );

    assert(!flushCalled);
  },
);

Deno.test(
  "3: TextDecoderStream leaks its TextDecoder's native decoder when its downstream pipe is aborted",
  async () => {
    const src = new ReadableStream({
      pull(controller) {
        controller.enqueue(encode("foo"));
      },
    });
    const aborter = new AbortController();
    const dst = new WritableStream({
      write(chunk) {
        assertEquals(chunk, "foo");
        // break the pipe between the TextDecoderStream and this WritableStream
        aborter.abort();
      },
    });

    try {
      await src.pipeThrough(new TextDecoderStream()).pipeTo(dst, {
        signal: aborter.signal,
      });
    } catch (e) {
      assertMatch(`${e}`, /abort/i);
    }
  },
);

function encode(str: string): Uint8Array {
  return new TextEncoder().encode(str);
}