[flutter_tool] Don't crash when writing to pub stdin fails (#49323)

packages/flutter_tools/lib/src/dart/pub.dart

@@ -308,13 +308,25 @@ class _DefaultPub implements Pub {
     );
 
     // Pipe the Flutter tool stdin to the pub stdin.
-    unawaited(process.stdin.addStream(io.stdin));
-
-    // Pipe the put stdout and stderr to the tool stdout and stderr.
-    await Future.wait<dynamic>(<Future<dynamic>>[
-      io.stdout.addStream(process.stdout),
-      io.stderr.addStream(process.stderr),
-    ]);
+    unawaited(process.stdin.addStream(io.stdin)
+      // If pub exits unexpectedly with an error, that will be reported below
+      // by the tool exit after the exit code check.
+      .catchError((dynamic err, StackTrace stack) {
+        globals.printTrace('Echoing stdin to the pub subprocess failed:');
+        globals.printTrace('$err\n$stack');
+      }
+    ));
+
+    // Pipe the pub stdout and stderr to the tool stdout and stderr.
+    try {
+      await Future.wait<dynamic>(<Future<dynamic>>[
+        io.stdout.addStream(process.stdout),
+        io.stderr.addStream(process.stderr),
+      ]);
+    } catch (err, stack) {
+      globals.printTrace('Echoing stdout or stderr from the pub subprocess failed:');
+      globals.printTrace('$err\n$stack');
+    }
 
     // Wait for pub to exit.
     final int code = await process.exitCode;

packages/flutter_tools/test/commands.shard/permeable/packages_test.dart

@@ -424,6 +424,27 @@ void main() {
       Pub: () => const Pub(),
     });
 
+    testUsingContext('pub publish input fails', () async {
+      final PromptingProcess process = PromptingProcess(stdinError: true);
+      mockProcessManager.processFactory = (List<String> commands) => process;
+      final Future<void> runPackages = createTestCommandRunner(PackagesCommand()).run(<String>['pub', 'publish']);
+      final Future<void> runPrompt = process.showPrompt('Proceed (y/n)? ', <String>['hello', 'world']);
+      final Future<void> simulateUserInput = Future<void>(() {
+        mockStdio.simulateStdin('y');
+      });
+      await Future.wait<void>(<Future<void>>[runPackages, runPrompt, simulateUserInput]);
+      final List<String> commands = mockProcessManager.commands;
+      expect(commands, hasLength(2));
+      expect(commands[0], matches(r'dart-sdk[\\/]bin[\\/]pub'));
+      expect(commands[1], 'publish');
+      // We get a trace message about the write to stdin failing.
+      expect(testLogger.traceText, contains('Echoing stdin to the pub subprocess failed'));
+    }, overrides: <Type, Generator>{
+      ProcessManager: () => mockProcessManager,
+      Stdio: () => mockStdio,
+      Pub: () => const Pub(),
+    });
+
     testUsingContext('publish', () async {
       await createTestCommandRunner(PackagesCommand()).run(<String>['pub', 'publish']);
       final List<String> commands = mockProcessManager.commands;

packages/flutter_tools/test/src/mocks.dart

@@ -310,21 +310,28 @@ class FakeProcess implements Process {
 /// A process that prompts the user to proceed, then asynchronously writes
 /// some lines to stdout before it exits.
 class PromptingProcess implements Process {
+  PromptingProcess({
+    bool stdinError = false,
+  }) : _stdin = CompleterIOSink(throwOnAdd: stdinError);
+
   Future<void> showPrompt(String prompt, List<String> outputLines) async {
-    _stdoutController.add(utf8.encode(prompt));
-    final List<int> bytesOnStdin = await _stdin.future;
-    // Echo stdin to stdout.
-    _stdoutController.add(bytesOnStdin);
-    if (bytesOnStdin[0] == utf8.encode('y')[0]) {
-      for (final String line in outputLines) {
-        _stdoutController.add(utf8.encode('$line\n'));
+    try {
+      _stdoutController.add(utf8.encode(prompt));
+      final List<int> bytesOnStdin = await _stdin.future;
+      // Echo stdin to stdout.
+      _stdoutController.add(bytesOnStdin);
+      if (bytesOnStdin.isNotEmpty && bytesOnStdin[0] == utf8.encode('y')[0]) {
+        for (final String line in outputLines) {
+          _stdoutController.add(utf8.encode('$line\n'));
+        }
       }
+    } finally {
+      await _stdoutController.close();
     }
-    await _stdoutController.close();
   }
 
   final StreamController<List<int>> _stdoutController = StreamController<List<int>>();
-  final CompleterIOSink _stdin = CompleterIOSink();
+  final CompleterIOSink _stdin;
 
   @override
   Stream<List<int>> get stdout => _stdoutController.stream;
@@ -347,6 +354,12 @@ class PromptingProcess implements Process {
 
 /// An IOSink that completes a future with the first line written to it.
 class CompleterIOSink extends MemoryIOSink {
+  CompleterIOSink({
+    this.throwOnAdd = false,
+  });
+
+  final bool throwOnAdd;
+
   final Completer<List<int>> _completer = Completer<List<int>>();
 
   Future<List<int>> get future => _completer.future;
@@ -354,7 +367,12 @@ class CompleterIOSink extends MemoryIOSink {
   @override
   void add(List<int> data) {
     if (!_completer.isCompleted) {
-      _completer.complete(data);
+      // When throwOnAdd is true, complete with empty so any expected output
+      // doesn't appear.
+      _completer.complete(throwOnAdd ? <int>[] : data);
+    }
+    if (throwOnAdd) {
+      throw 'CompleterIOSink Error';
     }
     super.add(data);
   }
@@ -375,9 +393,20 @@ class MemoryIOSink implements IOSink {
   @override
   Future<void> addStream(Stream<List<int>> stream) {
     final Completer<void> completer = Completer<void>();
-    stream.listen((List<int> data) {
-      add(data);
-    }).onDone(() => completer.complete());
+    StreamSubscription<List<int>> sub;
+    sub = stream.listen(
+      (List<int> data) {
+        try {
+          add(data);
+        } catch (err, stack) {
+          sub.cancel();
+          completer.completeError(err, stack);
+        }
+      },
+      onError: completer.completeError,
+      onDone: completer.complete,
+      cancelOnError: true,
+    );
     return completer.future;
   }