HttpClientResponse leaks if embedded into a stream that is canceled before the connection succeeded

Dart SDK version: 2.12.0-29.7.beta (beta) (Fri Nov 13 11:35:21 2020 +0100) on "macos_x64"

Look at the following example that receives video streams via HTTP.

import 'dart:io';

// synthetic delay before executing the http request
const httpDelay = 0;

final httpClient = HttpClient();

Stream<List<int>> foo() {
  const urls = [
    'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4',
    'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/ElephantsDream.mp4'
  ];
  return Stream.fromIterable(urls).asyncMap((url) async {
    print('foo: Wait $httpDelay seconds');
    await Future<void>.delayed(const Duration(seconds: httpDelay));

    print('foo: Opening request to $url');

    // create request
    final req = await httpClient.getUrl(Uri.parse(url));

    // wait 3 seconds and print connection status
    Future.delayed(const Duration(seconds: 3), () => print('foo: Connected to ${req.connectionInfo?.remoteAddress}'));

    print('foo: Return response of $url');

    // get response stream
    return req.close();
  }).asyncExpand((data) => data);
}

Future<void> main() async {
  print('main: Start the stream');

  final bar = foo().listen((event) {});

  print('main: Wait 1 second');
  await Future<void>.delayed(const Duration(seconds: 1));

  print('main: Cancel the stream');
  bar.cancel();
}

main: It iterates over a list of URLs and produces a consecutive stream of the received data, but cancels the subscription after 1 second.

foo: For each URL, we wait some synthetic delay given by httpDelay, then start the request and return it's response. In parallel, after 3 seconds, we print the connection status.

EXPECTED BEHAVIOUR

For httpDelay=0 and httpDelay=2, the program should terminate after a few seconds.

ACTUAL BEHAVIOUR

For httpDelay=2, the program hangs forever.

ANALYSIS

For httpDelay=0, the output is:

main: Start the stream
main: Wait 1 second
foo: Wait 0 seconds
foo: Opening request to http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
foo: Return response of http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
main: Cancel the stream

For httpDelay=2, the output is

main: Start the stream
main: Wait 1 second
foo: Wait 2 seconds
main: Cancel the stream
foo: Opening request to http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
foo: Return response of http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
foo: Connected to InternetAddress('172.217.19.80', IPv4)

So, if the stream is cancelled BEFORE the connection was made, the connection is open and hence leaks.

SEVERITY

Found in Flutter app production code resulting in weird behaviour. Actually, the root cause is very fundamental and will result in leaks of streams whenever an inner stream is expanded to an outer stream and the outer stream is cancelled early enough.

UGLY WORKAROUND

If we change foo to:

Stream<List<int>> foo() {
  const urls = [
    'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4',
    'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/ElephantsDream.mp4'
  ];
  return Stream.fromIterable(urls).asyncExpand((url) async* {
    print('foo: Wait $httpDelay seconds');
    await Future<void>.delayed(const Duration(seconds: httpDelay));

    print('foo: Opening request to $url');

    // create request
    final req = await httpClient.getUrl(Uri.parse(url));

    // wait 3 seconds and print connection status
    Future.delayed(const Duration(seconds: 3), () => print('foo: Connected to ${req.connectionInfo?.remoteAddress}'));

    print('foo: Return response of $url');

    // get response stream
    final stream = await req.close();

    // safeguarded yield*
    try {
      yield* stream;
    } finally {
      print("foo: yield* finished");
      try {
        // bruteforce cancelation
        stream.listen((event) {}).cancel();
      } catch (err) {/**/}
    }
  });
}

for httpDelay=3, the program terminates with the following output:

main: Start the stream
main: Wait 1 second
foo: Wait 3 seconds
main: Cancel the stream
foo: Opening request to http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
foo: Return response of http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4
foo: yield* finished
foo: Connected to null

dart-lang / sdk

HttpClientResponse leaks if embedded into a stream that is canceled before the connection succeeded #44514