Remove randomness from port assignment during coverage collection. (#7548)

Also, defer to test package for throttling (this will require a test
package update as well).

Also, add a lot more instrumentation to --verbose mode for tests, and
fix minor trivial things here and there, and add error handling in
more places.

Also, refactor how coverage works to be simpler and not use statics.

bin/flutter

@@ -36,13 +36,16 @@ DART_SDK_PATH="$FLUTTER_ROOT/bin/cache/dart-sdk"
 
 DART="$DART_SDK_PATH/bin/dart"
 
+# To debug the tool, you can uncomment the following line to enable checked mode and set an observatory port:
+# FLUTTER_TOOL_ARGS="--observe=65432 --checked"
+
 (
   if hash flock 2>/dev/null; then
     flock 3 # ensures that we don't simultaneously update Dart in multiple parallel instances
     # some platforms (e.g. Mac) don't have flock or any reliable alternative
   fi
   REVISION=`(cd "$FLUTTER_ROOT"; git rev-parse HEAD)`
-  if [ ! -f "$SNAPSHOT_PATH" ] || [ ! -f "$STAMP_PATH" ] || [ `cat "$STAMP_PATH"` != "$REVISION" ] || [ "$FLUTTER_TOOLS_DIR/pubspec.yaml" -nt "$FLUTTER_TOOLS_DIR/pubspec.lock" ]; then
+  if [ ! -f "$SNAPSHOT_PATH" ] || [ ! -s "$STAMP_PATH" ] || [ `cat "$STAMP_PATH"` != "$REVISION" ] || [ "$FLUTTER_TOOLS_DIR/pubspec.yaml" -nt "$FLUTTER_TOOLS_DIR/pubspec.lock" ]; then
     mkdir -p "$FLUTTER_ROOT/bin/cache"
     touch "$FLUTTER_ROOT/bin/cache/.dartignore"
     "$FLUTTER_ROOT/bin/internal/update_dart_sdk.sh"
@@ -56,7 +59,7 @@ DART="$DART_SDK_PATH/bin/dart"
 ) 3< "$PROG_NAME"
 
 set +e
-"$DART" "$SNAPSHOT_PATH" "$@"
+"$DART" $FLUTTER_TOOL_ARGS "$SNAPSHOT_PATH" "$@"
 
 # The VM exits with code 253 if the snapshot version is out-of-date.
 # If it is, we need to snapshot it again.
@@ -67,4 +70,4 @@ fi
 
 set -e
 "$DART" --snapshot="$SNAPSHOT_PATH" --packages="$FLUTTER_TOOLS_DIR/.packages" "$SCRIPT_PATH"
-"$DART" "$SNAPSHOT_PATH" "$@"
+"$DART" $FLUTTER_TOOL_ARGS "$SNAPSHOT_PATH" "$@"

packages/flutter_tools/lib/src/commands/test.dart

@@ -32,7 +32,7 @@ class TestCommand extends FlutterCommand {
     argParser.addFlag('merge-coverage',
       defaultsTo: false,
       negatable: false,
-      help: 'Whether to merge converage data with "coverage/lcov.base.info". '
+      help: 'Whether to merge converage data with "coverage/lcov.base.info".\n'
             'Implies collecting coverage data. (Requires lcov)'
     );
     argParser.addOption('coverage-path',
@@ -93,6 +93,7 @@ class TestCommand extends FlutterCommand {
       timeout: new Duration(seconds: 30),
     );
     status.stop();
+    printTrace('coverage information collection complete');
     if (coverageData == null)
       return false;
 
@@ -146,40 +147,42 @@ class TestCommand extends FlutterCommand {
 
   @override
   Future<Null> runCommand() async {
-    List<String> testArgs = argResults.rest.map((String testPath) => path.absolute(testPath)).toList();
+    List<String> testArgs = <String>[];
 
     commandValidator();
 
-    Directory testDir;
+    if (!terminal.supportsColor)
+      testArgs.addAll(<String>['--no-color', '-rexpanded']);
+
+    CoverageCollector collector;
+    if (argResults['coverage'] || argResults['merge-coverage']) {
+      collector = new CoverageCollector();
+      testArgs.add('--concurrency=1');
+    }
+
+    testArgs.add('--');
 
-    if (testArgs.isEmpty) {
+    Directory testDir;
+    List<String> files = argResults.rest.map((String testPath) => path.absolute(testPath)).toList();
+    if (files.isEmpty) {
       testDir = _currentPackageTestDir;
       if (!testDir.existsSync())
         throwToolExit("Test directory '${testDir.path}' not found.");
-
       testArgs.addAll(_findTests(testDir));
+    } else {
+      testArgs.addAll(files);
     }
 
-    testArgs.insert(0, '--');
-    if (!terminal.supportsColor)
-      testArgs.insertAll(0, <String>['--no-color', '-rexpanded']);
-
-    if (argResults['coverage'])
-      testArgs.insert(0, '--concurrency=1');
-
     final String shellPath = tools.getHostToolPath(HostTool.SkyShell) ?? Platform.environment['SKY_SHELL'];
     if (!fs.isFileSync(shellPath))
       throwToolExit('Cannot find Flutter shell at $shellPath');
-    loader.installHook(shellPath: shellPath);
+    loader.installHook(shellPath: shellPath, collector: collector);
 
     Cache.releaseLockEarly();
 
-    CoverageCollector collector = CoverageCollector.instance;
-    collector.enabled = argResults['coverage'] || argResults['merge-coverage'];
-
     int result = await _runTests(testArgs, testDir);
 
-    if (collector.enabled) {
+    if (collector != null) {
       if (!await _collectCoverageData(collector, mergeCoverageData: argResults['merge-coverage']))
         throwToolExit(null);
     }

packages/flutter_tools/lib/src/test/coverage_collector.dart

@@ -14,43 +14,6 @@ import '../globals.dart';
 
 /// A class that's used to collect coverage data during tests.
 class CoverageCollector {
-  CoverageCollector._();
-
-  /// The singleton instance of the coverage collector.
-  static final CoverageCollector instance = new CoverageCollector._();
-
-  /// By default, coverage collection is not enabled. Set [enabled] to true
-  /// to turn on coverage collection.
-  bool enabled = false;
-  int observatoryPort;
-
-  /// Adds a coverage collection tasks to the pending queue. The task will not
-  /// begin collecting coverage data until [CoverageCollectionTask.start] is
-  /// called.
-  ///
-  /// When a process is spawned to accumulate code coverage data, this method
-  /// should be called before the process terminates so that this collector
-  /// knows to wait for the coverage data in [finalizeCoverage].
-  ///
-  /// If this collector is not [enabled], the task will still be added to the
-  /// pending queue. Only when the task is started will the enabled state of
-  /// the collector be consulted.
-  CoverageCollectionTask addTask({
-    String host,
-    int port,
-    Process processToKill,
-  }) {
-    final CoverageCollectionTask task = new CoverageCollectionTask._(
-      this,
-      host,
-      port,
-      processToKill,
-    );
-    _tasks.add(task._future);
-    return task;
-  }
-
-  List<Future<Null>> _tasks = <Future<Null>>[];
   Map<String, dynamic> _globalHitmap;
 
   void _addHitmap(Map<String, dynamic> hitmap) {
@@ -60,36 +23,43 @@ class CoverageCollector {
       mergeHitmaps(hitmap, _globalHitmap);
   }
 
-  /// Returns a future that completes once all tasks have finished.
-  /// This will not start any tasks that were not already started.
+  /// Collects coverage for the given [Process] using the given `port`.
   ///
-  /// If [timeout] is specified, the future will timeout (with a
-  /// [TimeoutException]) after the specified duration.
-  Future<Null> finishPendingTasks({ Duration timeout }) {
-    Future<dynamic> future = Future.wait(_tasks, eagerError: true);
-    if (timeout != null)
-      future = future.timeout(timeout);
-    return future;
+  /// This should be called when the code whose coverage data is being collected
+  /// has been run to completion so that all coverage data has been recorded.
+  ///
+  /// The returned [Future] completes when the coverage is collected.
+  Future<Null> collectCoverage(Process process, InternetAddress host, int port) async {
+    assert(process != null);
+    assert(port != null);
+
+    int pid = process.pid;
+    int exitCode;
+    process.exitCode.then((int code) {
+      exitCode = code;
+    });
+
+    printTrace('pid $pid (port $port): collecting coverage data...');
+    final Map<dynamic, dynamic> data = await collect(host.address, port, false, false);
+    printTrace('pid $pid (port $port): ${ exitCode != null ? "process terminated prematurely with exit code $exitCode; aborting" : "collected coverage data; merging..." }');
+    if (exitCode != null)
+      throw new Exception('Failed to collect coverage, process terminated prematurely.');
+    _addHitmap(createHitmap(data['coverage']));
+    printTrace('pid $pid (port $port): done merging coverage data into global coverage map.');
   }
 
   /// Returns a future that will complete with the formatted coverage data
   /// (using [formatter]) once all coverage data has been collected.
   ///
-  /// This will not start any collection tasks. It us up to the caller of
-  /// [addTask] to maintain a reference to the [CoverageCollectionTask] and
-  /// call `start` on the task once the code in question has run. Failure to do
-  /// so will cause this method to wait indefinitely for the task.
+  /// This will not start any collection tasks. It us up to the caller of to
+  /// call [collectCoverage] for each process first.
   ///
   /// If [timeout] is specified, the future will timeout (with a
   /// [TimeoutException]) after the specified duration.
-  ///
-  /// This must only be called if this collector is [enabled].
   Future<String> finalizeCoverage({
     Formatter formatter,
     Duration timeout,
   }) async {
-    assert(enabled);
-    await finishPendingTasks(timeout: timeout);
     printTrace('formating coverage data');
     if (_globalHitmap == null)
       return null;
@@ -99,68 +69,8 @@ class CoverageCollector {
       List<String> reportOn = <String>[path.join(packagePath, 'lib')];
       formatter = new LcovFormatter(resolver, reportOn: reportOn, basePath: packagePath);
     }
-    return await formatter.format(_globalHitmap);
-  }
-}
-
-/// A class that represents a pending task of coverage data collection.
-/// Instances of this class are obtained when a process starts running code
-/// by calling [CoverageCollector.addTask]. Then, when the code has run to
-/// completion (all the coverage data has been recorded), the task is started
-/// to actually collect the coverage data.
-class CoverageCollectionTask {
-  final Completer<Null> _completer = new Completer<Null>();
-  final CoverageCollector _collector;
-  final String _host;
-  final int _port;
-  final Process _processToKill;
-
-  CoverageCollectionTask._(
-    this._collector,
-    this._host,
-    this._port,
-    this._processToKill,
-  );
-
-  bool _started = false;
-
-  Future<Null> get _future => _completer.future;
-
-  /// Starts the task of collecting coverage.
-  ///
-  /// This should be called when the code whose coverage data is being collected
-  /// has been run to completion so that all coverage data has been recorded.
-  /// Failure to do so will cause [CoverageCollector.finalizeCoverage] to wait
-  /// indefinitely for the task to complete.
-  ///
-  /// Each task may only be started once.
-  void start() {
-    assert(!_started);
-    _started = true;
-
-    if (!_collector.enabled) {
-      _processToKill.kill();
-      _completer.complete();
-      return;
-    }
-
-    int pid = _processToKill.pid;
-    printTrace('collecting coverage data from pid $pid on port $_port');
-    collect(_host, _port, false, false).then(
-      (Map<dynamic, dynamic> data) {
-        printTrace('done collecting coverage data from pid $pid');
-        _processToKill.kill();
-        try {
-          _collector._addHitmap(createHitmap(data['coverage']));
-          printTrace('done merging data from pid $pid into global coverage map');
-          _completer.complete();
-        } catch (error, stackTrace) {
-          _completer.completeError(error, stackTrace);
-        }
-      },
-      onError: (dynamic error, StackTrace stackTrace) {
-        _completer.completeError(error, stackTrace);
-      },
-    );
+    String result = await formatter.format(_globalHitmap);
+    _globalHitmap = null;
+    return result;
   }
 }

packages/flutter_tools/lib/src/test/flutter_platform.dart

@@ -4,8 +4,8 @@
 
 import 'dart:async';
 import 'dart:convert';
-import 'dart:math' as math;
 
+import 'package:meta/meta.dart';
 import 'package:path/path.dart' as path;
 import 'package:stream_channel/stream_channel.dart';
 
@@ -38,23 +38,28 @@ const Duration _kTestProcessTimeout = const Duration(minutes: 5);
 /// hold that against the test.
 const String _kStartTimeoutTimerMessage = 'sky_shell test process has entered main method';
 
-/// The address at which our WebSocket server resides.
+/// The address at which our WebSocket server resides and at which the sky_shell
+/// processes will host the Observatory server.
 final InternetAddress _kHost = InternetAddress.LOOPBACK_IP_V4;
 
-void installHook({ String shellPath }) {
-  hack.registerPlatformPlugin(<TestPlatform>[TestPlatform.vm], () => new FlutterPlatform(shellPath: shellPath));
+void installHook({ @required String shellPath, CoverageCollector collector }) {
+  hack.registerPlatformPlugin(
+    <TestPlatform>[TestPlatform.vm],
+    () => new FlutterPlatform(shellPath: shellPath, collector: collector),
+  );
 }
 
 enum _InitialResult { crashed, timedOut, connected }
-enum _TestResult { crashed, harnessBailed, completed }
+enum _TestResult { crashed, harnessBailed, testBailed }
 typedef Future<Null> _Finalizer();
 
 class FlutterPlatform extends PlatformPlugin {
-  FlutterPlatform({ this.shellPath }) {
+  FlutterPlatform({ this.shellPath, this.collector }) {
     assert(shellPath != null);
   }
 
   final String shellPath;
+  final CoverageCollector collector;
 
   // Each time loadChannel() is called, we spin up a local WebSocket server,
   // then spin up the engine in a subprocess. We pass the engine a Dart file
@@ -65,13 +70,33 @@ class FlutterPlatform extends PlatformPlugin {
 
   @override
   StreamChannel<dynamic> loadChannel(String testPath, TestPlatform platform) {
-    final StreamChannelController<dynamic> controller = new StreamChannelController<dynamic>(allowForeignErrors: false);
-    _startTest(testPath, controller.local);
-    return controller.foreign;
+    StreamController<dynamic> localController = new StreamController<dynamic>();
+    StreamController<dynamic> remoteController = new StreamController<dynamic>();
+    Completer<Null> testCompleteCompleter = new Completer<Null>();
+    _FlutterPlatformStreamSinkWrapper<dynamic> remoteSink = new _FlutterPlatformStreamSinkWrapper<dynamic>(
+      remoteController.sink,
+      testCompleteCompleter.future,
+    );
+    StreamChannel<dynamic> localChannel = new StreamChannel<dynamic>.withGuarantees(
+      remoteController.stream,
+      localController.sink,
+    );
+    StreamChannel<dynamic> remoteChannel = new StreamChannel<dynamic>.withGuarantees(
+      localController.stream,
+      remoteSink,
+    );
+    _startTest(testPath, localChannel).whenComplete(() {
+      testCompleteCompleter.complete();
+    });
+    return remoteChannel;
   }
 
+  int _testCount = 0;
+
   Future<Null> _startTest(String testPath, StreamChannel<dynamic> controller) async {
-    printTrace('starting test: $testPath');
+    int ourTestCount = _testCount;
+    _testCount += 1;
+    printTrace('test $ourTestCount: starting test $testPath');
 
     final List<_Finalizer> finalizers = <_Finalizer>[];
     bool subprocessActive = false;
@@ -81,15 +106,34 @@ class FlutterPlatform extends PlatformPlugin {
 
       // Prepare our WebSocket server to talk to the engine subproces.
       HttpServer server = await HttpServer.bind(_kHost, 0);
-      finalizers.add(() async { await server.close(force: true); });
+      finalizers.add(() async {
+        printTrace('test $ourTestCount: shutting down test harness socket server');
+        await server.close(force: true);
+      });
       Completer<WebSocket> webSocket = new Completer<WebSocket>();
-      server.listen((HttpRequest request) {
+      server.listen(
+        (HttpRequest request) {
           webSocket.complete(WebSocketTransformer.upgrade(request));
-      });
+        },
+        onError: (dynamic error, dynamic stack) {
+          // If you reach here, it's unlikely we're going to be able to really handle this well.
+          printTrace('test $ourTestCount: test harness socket server experienced an unexpected error: $error');
+          if (!controllerSinkClosed) {
+            controller.sink.addError(error, stack);
+            controller.sink.close();
+          } else {
+            printError('unexpected error from test harness socket server: $error');
+          }
+        },
+        cancelOnError: true,
+      );
 
       // Prepare a temporary directory to store the Dart file that will talk to us.
       Directory temporaryDirectory = fs.systemTempDirectory.createTempSync('dart_test_listener');
-      finalizers.add(() async { temporaryDirectory.deleteSync(recursive: true); });
+      finalizers.add(() async {
+        printTrace('test $ourTestCount: deleting temporary directory');
+        temporaryDirectory.deleteSync(recursive: true);
+      });
 
       // Prepare the Dart file that will talk to us and start the test.
       File listenerFile = fs.file('${temporaryDirectory.path}/listener.dart');
@@ -99,52 +143,51 @@ class FlutterPlatform extends PlatformPlugin {
         encodedWebsocketUrl: Uri.encodeComponent("ws://${_kHost.address}:${server.port}"),
       ));
 
-      // If we are collecting coverage data, then set that up now.
-      int observatoryPort;
-      if (CoverageCollector.instance.enabled) {
-        // TODO(ianh): the random number on the next line is a landmine that will eventually
-        // cause a hard-to-find bug...
-        observatoryPort = CoverageCollector.instance.observatoryPort ?? new math.Random().nextInt(30000) + 2000;
-        await CoverageCollector.instance.finishPendingTasks();
-      }
-
       // Start the engine subprocess.
+      printTrace('test $ourTestCount: starting shell process');
       Process process = await _startProcess(
         shellPath,
         listenerFile.path,
         packages: PackageMap.globalPackagesPath,
-        observatoryPort: observatoryPort,
+        enableObservatory: collector != null,
       );
       subprocessActive = true;
       finalizers.add(() async {
-        if (subprocessActive)
+        if (subprocessActive) {
+          printTrace('test $ourTestCount: ensuring end-of-process for shell');
           process.kill();
-        int exitCode = await process.exitCode;
+          final int exitCode = await process.exitCode;
           subprocessActive = false;
-        if (!controllerSinkClosed && exitCode != 0) {
+          if (!controllerSinkClosed && exitCode != -15) {
             String message = _getErrorMessage(_getExitCodeMessage(exitCode, 'after tests finished'), testPath, shellPath);
             controller.sink.addError(message);
           }
+        }
       });
 
-      CoverageCollectionTask coverageTask = CoverageCollector.instance.addTask(
-        host: _kHost.address,
-        port: observatoryPort,
-        processToKill: process, // This kills the subprocess whether coverage is enabled or not.
-      );
-
       Completer<Null> timeout = new Completer<Null>();
 
       // Pipe stdout and stderr from the subprocess to our printStatus console.
-      _pipeStandardStreamsToConsole(process, startTimeoutTimer: () {
+      // We also keep track of what observatory port the engine used, if any.
+      int processObservatoryPort;
+      _pipeStandardStreamsToConsole(
+        process,
+        storeObservatoryPort: (int observatoryPort) {
+          assert(processObservatoryPort == null);
+          printTrace('test $ourTestCount: using observatory port $observatoryPort from pid ${process.pid} to collect coverage');
+          processObservatoryPort = observatoryPort;
+        },
+        startTimeoutTimer: () {
           new Future<_InitialResult>.delayed(_kTestStartupTimeout, () => timeout.complete());
-      });
+        },
+      );
 
       // At this point, three things can happen next:
       // The engine could crash, in which case process.exitCode will complete.
       // The engine could connect to us, in which case webSocket.future will complete.
       // The local test harness could get bored of us.
 
+      printTrace('test $ourTestCount: awaiting initial result for pid ${process.pid}');
       _InitialResult initialResult = await Future.any(<Future<_InitialResult>>[
         process.exitCode.then<_InitialResult>((int exitCode) => _InitialResult.crashed),
         timeout.future.then<_InitialResult>((Null _) => _InitialResult.timedOut),
@@ -154,40 +197,69 @@ class FlutterPlatform extends PlatformPlugin {
 
       switch (initialResult) {
         case _InitialResult.crashed:
+          printTrace('test $ourTestCount: process with pid ${process.pid} crashed before connecting to test harness');
           int exitCode = await process.exitCode;
+          subprocessActive = false;
           String message = _getErrorMessage(_getExitCodeMessage(exitCode, 'before connecting to test harness'), testPath, shellPath);
           controller.sink.addError(message);
           controller.sink.close();
+          printTrace('test $ourTestCount: waiting for controller sink to close');
           await controller.sink.done;
           break;
         case _InitialResult.timedOut:
+          printTrace('test $ourTestCount: timed out waiting for process with pid ${process.pid} to connect to test harness');
           String message = _getErrorMessage('Test never connected to test harness.', testPath, shellPath);
           controller.sink.addError(message);
           controller.sink.close();
+          printTrace('test $ourTestCount: waiting for controller sink to close');
           await controller.sink.done;
           break;
         case _InitialResult.connected:
+          printTrace('test $ourTestCount: process with pid ${process.pid} connected to test harness');
           WebSocket testSocket = await webSocket.future;
 
           Completer<Null> harnessDone = new Completer<Null>();
           StreamSubscription<dynamic> harnessToTest = controller.stream.listen(
             (dynamic event) { testSocket.add(JSON.encode(event)); },
             onDone: () { harnessDone.complete(); },
+            onError: (dynamic error, dynamic stack) {
+              // If you reach here, it's unlikely we're going to be able to really handle this well.
+              printError('test harness controller stream experienced an unexpected error\ntest: $testPath\nerror: $error');
+              if (!controllerSinkClosed) {
+                controller.sink.addError(error, stack);
+                controller.sink.close();
+              } else {
+                printError('unexpected error from test harness controller stream: $error');
+              }
+            },
+            cancelOnError: true,
           );
 
           Completer<Null> testDone = new Completer<Null>();
           StreamSubscription<dynamic> testToHarness = testSocket.listen(
-            (dynamic event) {
-              assert(event is String); // we shouldn't ever get binary messages
-              controller.sink.add(JSON.decode(event));
+            (dynamic encodedEvent) {
+              assert(encodedEvent is String); // we shouldn't ever get binary messages
+              controller.sink.add(JSON.decode(encodedEvent));
             },
             onDone: () { testDone.complete(); },
+            onError: (dynamic error, dynamic stack) {
+              // If you reach here, it's unlikely we're going to be able to really handle this well.
+              printError('test socket stream experienced an unexpected error\ntest: $testPath\nerror: $error');
+              if (!controllerSinkClosed) {
+                controller.sink.addError(error, stack);
+                controller.sink.close();
+              } else {
+                printError('unexpected error from test socket stream: $error');
+              }
+            },
+            cancelOnError: true,
           );
 
+          printTrace('test $ourTestCount: awaiting test result for pid ${process.pid}');
           _TestResult testResult = await Future.any(<Future<_TestResult>>[
             process.exitCode.then<_TestResult>((int exitCode) { return _TestResult.crashed; }),
-            testDone.future.then<_TestResult>((Null _) { return _TestResult.completed; }),
             harnessDone.future.then<_TestResult>((Null _) { return _TestResult.harnessBailed; }),
+            testDone.future.then<_TestResult>((Null _) { return _TestResult.testBailed; }),
           ]);
 
           harnessToTest.cancel();
@@ -195,40 +267,60 @@ class FlutterPlatform extends PlatformPlugin {
 
           switch (testResult) {
             case _TestResult.crashed:
+              printTrace('test $ourTestCount: process with pid ${process.pid} crashed');
               int exitCode = await process.exitCode;
               subprocessActive = false;
               String message = _getErrorMessage(_getExitCodeMessage(exitCode, 'before test harness closed its WebSocket'), testPath, shellPath);
               controller.sink.addError(message);
               controller.sink.close();
+              printTrace('test $ourTestCount: waiting for controller sink to close');
               await controller.sink.done;
               break;
-            case _TestResult.completed:
-              break;
             case _TestResult.harnessBailed:
+              printTrace('test $ourTestCount: process with pid ${process.pid} no longer needed by test harness');
+              break;
+            case _TestResult.testBailed:
+              printTrace('test $ourTestCount: process with pid ${process.pid} no longer needs test harness');
               break;
           }
           break;
       }
 
-      coverageTask.start();
-      subprocessActive = false;
-    } catch (e, stack) {
+      if (subprocessActive && collector != null) {
+        printTrace('test $ourTestCount: collecting coverage');
+        await collector.collectCoverage(process, _kHost, processObservatoryPort);
+      }
+    } catch (error, stack) {
+      printTrace('test $ourTestCount: error caught during test; ${controllerSinkClosed ? "reporting to console" : "sending to test framework"}');
       if (!controllerSinkClosed) {
-        controller.sink.addError(e, stack);
+        controller.sink.addError(error, stack);
       } else {
-        printError('unhandled error during test:\n$e\n$stack');
+        printError('unhandled error during test:\n$testPath\n$error');
       }
     } finally {
-      for (_Finalizer finalizer in finalizers)
+      printTrace('test $ourTestCount: cleaning up...');
+      for (_Finalizer finalizer in finalizers) {
+        try {
           await finalizer();
+        } catch (error, stack) {
+          printTrace('test $ourTestCount: error while cleaning up; ${controllerSinkClosed ? "reporting to console" : "sending to test framework"}');
+          if (!controllerSinkClosed) {
+            controller.sink.addError(error, stack);
+          } else {
+            printError('unhandled error during finalization of test:\n$testPath\n$error');
+          }
+        }
+      }
       if (!controllerSinkClosed) {
         controller.sink.close();
+        printTrace('test $ourTestCount: waiting for controller sink to close');
         await controller.sink.done;
       }
     }
     assert(!subprocessActive);
     assert(controllerSinkClosed);
-    printTrace('ending test: $testPath');
+    printTrace('test $ourTestCount: finished');
+    return null;
   }
 
   String _generateTestMain({
@@ -286,14 +378,11 @@ void main() {
   }
 
 
-  Future<Process> _startProcess(String executable, String testPath, { String packages, int observatoryPort }) {
+  Future<Process> _startProcess(String executable, String testPath, { String packages, bool enableObservatory: false }) {
     assert(executable != null); // Please provide the path to the shell in the SKY_SHELL environment variable.
     List<String> arguments = <String>[];
-    if (observatoryPort != null) {
-      arguments.add('--observatory-port=$observatoryPort');
-    } else {
+    if (!enableObservatory)
       arguments.add('--disable-observatory');
-    }
     arguments.addAll(<String>[
       '--enable-dart-profiling',
       '--non-interactive',
@@ -309,22 +398,46 @@ void main() {
     return processManager.start(executable, arguments, environment: environment);
   }
 
+  String get observatoryPortString => 'Observatory listening on http://${_kHost.address}:';
+  String get diagnosticPortString => 'Diagnostic server listening on http://${_kHost.address}:';
+
   void _pipeStandardStreamsToConsole(
     Process process, {
     void startTimeoutTimer(),
+    void storeObservatoryPort(int port)
   }) {
     for (Stream<List<int>> stream in
         <Stream<List<int>>>[process.stderr, process.stdout]) {
       stream.transform(UTF8.decoder)
         .transform(const LineSplitter())
-        .listen((String line) {
-          if (line == _kStartTimeoutTimerMessage && startTimeoutTimer != null)
+        .listen(
+          (String line) {
+            if (line == _kStartTimeoutTimerMessage) {
+              if (startTimeoutTimer != null)
                 startTimeoutTimer();
-          else if (line.startsWith('error: Unable to read Dart source \'package:test/'))
+            } else if (line.startsWith('error: Unable to read Dart source \'package:test/')) {
+              printTrace('Shell: $line');
               printError('\n\nFailed to load test harness. Are you missing a dependency on flutter_test?\n');
-          else if (line != null)
+            } else if (line.startsWith(observatoryPortString)) {
+              printTrace('Shell: $line');
+              try {
+                int port = int.parse(line.substring(observatoryPortString.length, line.length - 1)); // last character is a slash
+                if (storeObservatoryPort != null)
+                  storeObservatoryPort(port);
+              } catch (error) {
+                printError('Could not parse shell observatory port message: $error');
+              }
+            } else if (line.startsWith(diagnosticPortString)) {
+              printTrace('Shell: $line');
+            } else if (line != null) {
               printStatus('Shell: $line');
-        });
+            }
+          },
+          onError: (dynamic error) {
+            printError('shell console stream for process pid ${process.pid} experienced an unexpected error: $error');
+          },
+          cancelOnError: true,
+        );
     }
   }
 
@@ -334,6 +447,8 @@ void main() {
 
   String _getExitCodeMessage(int exitCode, String when) {
     switch (exitCode) {
+      case 1:
+        return 'Shell subprocess cleanly reported an error $when. Check the logs above for an error message.';
       case 0:
         return 'Shell subprocess ended cleanly $when. Did main() call exit()?';
       case -0x0f: // ProcessSignal.SIGTERM
@@ -342,8 +457,43 @@ void main() {
         return 'Shell subprocess crashed with segmentation fault $when.';
       case -0x06: // ProcessSignal.SIGABRT
         return 'Shell subprocess crashed with SIGABRT ($exitCode) $when.';
+      case -0x02: // ProcessSignal.SIGINT
+        return 'Shell subprocess terminated by ^C (SIGINT, $exitCode) $when.';
       default:
         return 'Shell subprocess crashed with unexpected exit code $exitCode $when.';
     }
   }
 }
+
+class _FlutterPlatformStreamSinkWrapper<S> implements StreamSink<S> {
+  _FlutterPlatformStreamSinkWrapper(this._parent, this._shellProcessClosed);
+  final StreamSink<S> _parent;
+  final Future<Null> _shellProcessClosed;
+
+  @override
+  Future<Null> get done => _done.future;
+  final Completer<Null> _done = new Completer<Null>();
+
+  @override
+  Future<dynamic> close() {
+   Future.wait(<Future<dynamic>>[
+      _parent.close(),
+      _shellProcessClosed,
+    ]).then(
+      (List<dynamic> value) {
+        _done.complete();
+      },
+      onError: (dynamic error, StackTrace stack) {
+        _done.completeError(error, stack);
+      },
+    );
+    return done;
+  }
+
+  @override
+  void add(S event) => _parent.add(event);
+  @override
+  void addError(dynamic errorEvent, [ StackTrace stackTrace ]) => _parent.addError(errorEvent, stackTrace);
+  @override
+  Future<dynamic> addStream(Stream<S> stream) => _parent.addStream(stream);
+}

packages/flutter_tools/lib/src/usage.dart

@@ -104,11 +104,11 @@ class Usage {
 
   /// Returns when the last analytics event has been sent, or after a fixed
   /// (short) delay, whichever is less.
-  Future<Null> ensureAnalyticsSent() {
+  Future<Null> ensureAnalyticsSent() async {
     // TODO(devoncarew): This may delay tool exit and could cause some analytics
     // events to not be reported. Perhaps we could send the analytics pings
     // out-of-process from flutter_tools?
-    return _analytics.waitForLastPing(timeout: new Duration(milliseconds: 250));
+    await _analytics.waitForLastPing(timeout: new Duration(milliseconds: 250));
   }
 
   void printUsage() {