Issue 1:
Describe the bug

I want to wait until a subprocess(like a web server) is ready.

So I'm tried to write the code that reads the stdout of the process in async iteration and when I get a specific string, break it.

But it seems to be not releasing a TextDecoder if it hasn't completed the async iteration loop.

Example:

readablestream_test.ts

import { TextLineStream, } from "https://deno.land/std@0.129.0/streams/delimiter.ts";

Deno.test("ReadableStream leak memory", async () => {
const process = Deno.run({
cmd: ["ls"],
stdout: "piped",
});

const lines = process.stdout.readable
.pipeThrough(new TextDecoderStream())
.pipeThrough(new TextLineStream());

for await (const _line of lines) {
break;
}
await lines.cancel();

process.kill("SIGTERM");
process.close();

console.log(Deno.resources());
});

We can get the following result.

$ deno run -A readablestream_test.ts
Check file:///Users/ioriikeda/Projects/ts-scraps/leak-error/readablestream_test.ts
running 1 test from file:///Users/ioriikeda/Projects/ts-scraps/leak-error/readablestream_test.ts
test ReadableStream leak memory ...{ "0": "stdin", "1": "stdout", "2": "stderr", "5": "textDecoder" }

FAILED (45ms)

failures:

ReadableStream leak memory
AssertionError: Test case is leaking 1 resource:

- A text decoder (rid 5) was created during the test, but not finsihed during the test. Close the text decoder by calling `textDecoder.decode('')` or `await textDecoderStream.readable.cancel()`.

at assert (deno:runtime/js/06_util.js:46:13)
at resourceSanitizer (deno:runtime/js/40_testing.js:406:7)
at async Object.exitSanitizer [as fn] (deno:runtime/js/40_testing.js:424:9)
at async runTest (deno:runtime/js/40_testing.js:784:7)
at async Object.runTests (deno:runtime/js/40_testing.js:982:22)

failures:

ReadableStream leak memory

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

error: Test failed

Expected behavior

Release the resource even if an async iteration has not completed when we break the iteration.

Environment

OS: MacOS 12.1

deno: deno 1.20.1 (release, x86_64-apple-darwin)

std: 0.129.0

Issue 2:
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.