fix(http2): finish() and terminate() must not hang after a failed socket write

pkgs/http2/CHANGELOG.md

@@ -27,6 +27,9 @@
   now extends `StreamTransportException`, as documented for `TransportException`
   subclasses. `MultiProtocolHttpServer.startServing` forwards an unexpected ALPN
   protocol to `onError` instead of throwing inside the socket listener.
+- `finish()` and `terminate()` no longer hang when a write to the socket fails
+  (for example because the peer closed the connection); previously they waited
+  for a `Socket.done` that never completes after a failed write.
 
 ## 3.1.0
 

pkgs/http2/lib/src/async_utils/async_utils.dart

@@ -68,10 +68,19 @@ class BufferedSink {
         // Currently `_doneFuture` will just complete normally if the sink
         // cancelled.
       };
-    _doneFuture = Future.wait([
-      _controller.stream.pipe(dataSink),
-      dataSink.done,
-    ]);
+    // `pipe` completes once `dataSink.close()` has completed, which for
+    // `dart:io` sinks and `StreamController`s is once `dataSink.done` has.
+    //
+    // It must not additionally wait for `dataSink.done`: when a write fails
+    // (e.g. the peer reset the connection), `addStream` completes with the
+    // error but `Stream.pipe` does not close the sink, so `dataSink.done` never
+    // completes - and neither would this future, nor `Connection.finish()` and
+    // `Connection.terminate()`, which wait for it.
+    //
+    // A failed write means the connection is dead; the owner learns about that
+    // through the incoming side, so like a cancelled sink it just completes
+    // this future normally.
+    _doneFuture = _controller.stream.pipe(dataSink).catchError((Object _) {});
   }
 
   /// The underlying sink.

pkgs/http2/test/client_test.dart

@@ -14,6 +14,7 @@ import 'package:http2/src/settings/settings.dart';
 import 'package:http2/transport.dart';
 import 'package:test/test.dart';
 
+import 'src/async_utils/async_utils_test.dart' show FailingSink;
 import 'src/hpack/hpack_test.dart' show isHeader;
 
 void main() {
@@ -1693,6 +1694,27 @@ void main() {
           await Future.wait([serverFun(), clientFun()], eagerError: true);
         },
       );
+
+      test(
+        'finish-and-terminate-complete-when-the-socket-write-fails',
+        () async {
+          // Before any of the client's bytes reach the wire, the peer has gone
+          // away: every write fails and the sink's `done` never completes (as
+          // with a `dart:io` Socket whose `addStream` failed).
+          for (final close in [
+            (ClientTransportConnection c) => c.finish(),
+            (ClientTransportConnection c) => c.terminate(),
+          ]) {
+            final incoming = StreamController<List<int>>();
+            final connection = ClientTransportConnection.viaStreams(
+              incoming.stream,
+              FailingSink(),
+            );
+            await close(connection).timeout(const Duration(seconds: 2));
+            await incoming.close();
+          }
+        },
+      );
     });
   });
 }

pkgs/http2/test/src/async_utils/async_utils_test.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:io';
 
 import 'package:http2/src/async_utils/async_utils.dart';
 import 'package:test/test.dart';
@@ -67,6 +68,15 @@ void main() {
       );
     });
 
+    test('buffered-sink-done-after-failed-write', () async {
+      final bs = BufferedSink(FailingSink());
+      bs.sink.add([1, 2, 3]);
+      await bs.sink.close();
+      // Must not hang on `FailingSink.done`, which (like `Socket.done` after a
+      // failed `addStream`) never completes.
+      await bs.doneFuture.timeout(const Duration(seconds: 2));
+    });
+
     test('buffered-bytes-writer', () async {
       var c = StreamController<List<int>>();
       var writer = BufferedBytesWriter(c);
@@ -89,3 +99,32 @@ void main() {
     });
   });
 }
+
+/// Behaves like a `dart:io` [Socket] whose peer has reset the connection:
+/// [addStream] fails on the first write and [done] only ever completes through
+/// an explicit [close], which `Stream.pipe` does not call after an error.
+class FailingSink implements StreamSink<List<int>> {
+  final _done = Completer<void>();
+
+  @override
+  void add(List<int> data) {}
+
+  @override
+  void addError(Object error, [StackTrace? stackTrace]) {}
+
+  @override
+  Future<void> addStream(Stream<List<int>> stream) async {
+    await for (final _ in stream) {
+      throw const SocketException('Connection reset by peer');
+    }
+  }
+
+  @override
+  Future<void> close() {
+    if (!_done.isCompleted) _done.complete();
+    return _done.future;
+  }
+
+  @override
+  Future<void> get done => _done.future;
+}