[ DDS ] Update DDS launch sites to assume DDS process closes stderr

Removes risk of DDS connection information being split across two stream
events, causing JSON decoding to fail.

Also updates DDS to close stderr, even in the error case.

TEST=Existing service and dartdev tests

Change-Id: I5cceab899aac1fa63bd7578dd658b34096722bd3
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/371000
Reviewed-by: Derek Xu <derekx@google.com>

pkg/dartdev/lib/src/dds_runner.dart

@@ -46,12 +46,17 @@ class DDSRunner {
       ],
       mode: ProcessStartMode.detachedWithStdio,
     );
-    final completer = Completer<void>();
+
+    // NOTE: update pkg/dartdev/lib/src/commands/run.dart if this message
+    // is changed to ensure consistency.
     const devToolsMessagePrefix =
         'The Dart DevTools debugger and profiler is available at:';
     if (debugDds) {
       late final StreamSubscription stdoutSub;
-      stdoutSub = process.stdout.transform(utf8.decoder).listen((event) {
+      stdoutSub = process.stdout
+          .transform(utf8.decoder)
+          .transform(const LineSplitter())
+          .listen((event) {
         if (event.startsWith(devToolsMessagePrefix)) {
           final ddsDebuggingUri = event.split(' ').last;
           print(
@@ -61,21 +66,27 @@ class DDSRunner {
         }
       });
     }
-    late final StreamSubscription stderrSub;
-    stderrSub = process.stderr.transform(utf8.decoder).listen((event) {
-      final result = json.decode(event) as Map<String, dynamic>;
-      final state = result['state'];
-      if (state == 'started') {
-        if (result.containsKey('devToolsUri')) {
-          final devToolsUri = result['devToolsUri'];
+
+    // DDS will close stderr once it's finished launching.
+    final launchResult = await process.stderr.transform(utf8.decoder).join();
+
+    void printError(String details) => stderr.writeln(
+          'Could not start the VM service:\n$details',
+        );
+
+    try {
+      final result = json.decode(launchResult);
+      if (result
+          case {
+            'state': 'started',
+            'ddsUri': final String ddsUriStr,
+          }) {
+        ddsUri = Uri.parse(ddsUriStr);
+        if (result case {'devToolsUri': String devToolsUri}) {
           print('$devToolsMessagePrefix $devToolsUri');
         }
-        ddsUri = Uri.parse(result['ddsUri']);
-        stderrSub.cancel();
-        completer.complete();
       } else {
-        stderrSub.cancel();
-        final error = result['error'] ?? event;
+        final error = result['error'] ?? result;
         final stacktrace = result['stacktrace'] ?? '';
         String message = 'Could not start the VM service: ';
         if (error.contains('Failed to create server socket')) {
@@ -83,15 +94,15 @@ class DDSRunner {
         } else {
           message += '$error\n$stacktrace\n';
         }
-        completer.completeError(message);
+        printError(message);
+        return false;
       }
-    });
-    try {
-      await completer.future;
-      return true;
-    } catch (e) {
-      stderr.write(e);
+    } catch (_) {
+      // Malformed JSON was likely encountered, so output the entirety of
+      // stderr in the error message.
+      printError(launchResult);
       return false;
     }
+    return true;
   }
 }

pkg/dds/bin/dds.dart

@@ -131,6 +131,10 @@ ${argParser.usage}
     stderr.close();
   } catch (e, st) {
     writeErrorResponse(e, st);
+  } finally {
+    // Always close stderr to notify tooling that DDS has finished writing
+    // launch details.
+    await stderr.close();
   }
 }
 

sdk/lib/_internal/vm/bin/vmservice_io.dart

@@ -114,12 +114,20 @@ class _DebuggingSession {
       ],
       mode: ProcessStartMode.detachedWithStdio,
     );
-    final completer = Completer<void>();
-    late StreamSubscription<String> stderrSub;
-    stderrSub = _process!.stderr.transform(utf8.decoder).listen((event) {
-      final result = json.decode(event) as Map<String, dynamic>;
-      final state = result['state'];
-      if (state == 'started') {
+
+    // DDS will close stderr once it's finished launching.
+    final launchResult = await _process!.stderr.transform(utf8.decoder).join();
+
+    void printError(String details) => stderr.writeln(
+          'Could not start the VM service:\n$details',
+        );
+
+    try {
+      final result = json.decode(launchResult);
+      if (result
+          case {
+            'state': 'started',
+          }) {
         if (result case {'devToolsUri': String devToolsUri}) {
           // NOTE: update pkg/dartdev/lib/src/commands/run.dart if this message
           // is changed to ensure consistency.
@@ -135,21 +143,17 @@ class _DebuggingSession {
             } when _printDtd) {
           print('The Dart Tooling Daemon (DTD) is available at: $dtdUri');
         }
-        stderrSub.cancel();
-        completer.complete();
       } else {
-        final error = result['error'] ?? event;
-        stderrSub.cancel();
-        completer.completeError('Could not start the VM service:\n$error\n');
+        printError(result['error'] ?? result);
+        return false;
       }
-    });
-    try {
-      await completer.future;
-      return true;
-    } catch (e) {
-      stderr.write(e);
+    } catch (_) {
+      // Malformed JSON was likely encountered, so output the entirety of
+      // stderr in the error message.
+      printError(launchResult);
       return false;
     }
+    return true;
   }
 
   void shutdown() => _process!.kill();