Recover from watcher restart.

Author: davidmorgan

build_runner/lib/src/daemon/daemon_builder.dart

@@ -277,6 +277,9 @@ class BuildRunnerDaemonBuilder implements DaemonBuilder {
     Stream<List<WatchEvent>> graphEvents() =>
         PackageGraphWatcher(packageGraph, watch: PackageNodeWatcher.new)
             .watch()
+            // TODO(davidmorgan): do something with the nulls, which indicate
+            // "watcher dropped events and restarted".
+            .whereNotNull()
             .asyncWhere(
               (change) => shouldProcess(
                 change,

build_runner/lib/src/entrypoint/watch.dart

@@ -49,25 +49,30 @@ class WatchCommand extends BuildRunnerCommand {
   }
 
   Future<int> _run(WatchOptions options) async {
-    var handler = await watch(
-      builderApplications,
-      enableLowResourcesMode: options.enableLowResourcesMode,
-      configKey: options.configKey,
-      buildDirs: options.buildDirs,
-      outputSymlinksOnly: options.outputSymlinksOnly,
-      packageGraph: packageGraph,
-      trackPerformance: options.trackPerformance,
-      skipBuildScriptCheck: options.skipBuildScriptCheck,
-      verbose: options.verbose,
-      builderConfigOverrides: options.builderConfigOverrides,
-      isReleaseBuild: options.isReleaseBuild,
-      logPerformanceDir: options.logPerformanceDir,
-      buildFilters: options.buildFilters,
-    );
+    while (true) {
+      final handler = await watch(
+        builderApplications,
+        enableLowResourcesMode: options.enableLowResourcesMode,
+        configKey: options.configKey,
+        buildDirs: options.buildDirs,
+        outputSymlinksOnly: options.outputSymlinksOnly,
+        packageGraph: packageGraph,
+        trackPerformance: options.trackPerformance,
+        skipBuildScriptCheck: options.skipBuildScriptCheck,
+        verbose: options.verbose,
+        builderConfigOverrides: options.builderConfigOverrides,
+        isReleaseBuild: options.isReleaseBuild,
+        logPerformanceDir: options.logPerformanceDir,
+        buildFilters: options.buildFilters,
+      );
 
-    final completer = Completer<int>();
-    handleBuildResultsStream(handler.buildResults, completer);
-    return completer.future;
+      final completer = Completer<int>();
+      handleBuildResultsStream(handler.buildResults, completer);
+      final result = await completer.future;
+      if (result != ExitCode.tempFail.code) {
+        return result;
+      }
+    }
   }
 
   /// Listens to [buildResults], handling certain types of errors and completing
@@ -83,6 +88,9 @@ class WatchCommand extends BuildRunnerCommand {
           completer.completeError(const BuildScriptChangedException());
         } else if (result.failureType == FailureType.buildConfigChanged) {
           completer.completeError(const BuildConfigChangedException());
+        } else if (result.failureType == FailureType.watcherRestarted) {
+          // TODO(davidmorgan): don't communicate using errors.
+          completer.complete(ExitCode.tempFail.code);
         }
       }
     });

build_runner/lib/src/generate/watch_impl.dart

@@ -291,6 +291,22 @@ class WatchImpl implements BuildState {
     );
     graphWatcher
         .watch()
+        .where((change) {
+          if (change == null) {
+            controller.add(
+              BuildResult(
+                BuildStatus.failure,
+                [],
+                failureType: FailureType.watcherRestarted,
+              ),
+            );
+            _terminateCompleter.complete();
+            return false;
+          } else {
+            return true;
+          }
+        })
+        .whereNotNull()
         .asyncMap<AssetChange>((change) {
           // Delay any events until the first build is completed.
           if (firstBuildCompleter.isCompleted) return change;

build_runner/lib/src/watcher/graph_watcher.dart

@@ -33,31 +33,42 @@ class PackageGraphWatcher {
   }) : _strategy = watch ?? _default;
 
   /// Returns a stream of records for assets that changed in the package graph.
-  Stream<AssetChange> watch() {
+  ///
+  /// A `null` event indicates that the watcher failed: watching will continue
+  /// but some events are missing.
+  Stream<AssetChange?> watch() {
     assert(!_isWatching);
     _isWatching = true;
     return LazyStream(_watch);
   }
 
-  Stream<AssetChange> _watch() {
+  Stream<AssetChange?> _watch() {
     final allWatchers =
         _graph.allPackages.values
             .where((node) => node.dependencyType == DependencyType.path)
             .map(_strategy)
             .toList();
+    final restartsController = StreamController<AssetChange?>();
     final filteredEvents =
         allWatchers
             .map(
               (w) => w.watch().where(_nestedPathFilter(w.node)).handleError((
                 Object e,
                 StackTrace s,
               ) {
-                buildLog.error(
-                  buildLog.renderThrowable(
-                    'Failed to watch files in package:${w.node.name}.',
-                    e,
-                  ),
-                );
+                if (e is FileSystemException &&
+                    e.message.startsWith(
+                      'Directory watcher closed unexpectedly',
+                    )) {
+                  restartsController.add(null);
+                } else {
+                  buildLog.error(
+                    buildLog.renderThrowable(
+                      'Failed to watch files in package:${w.node.name}.',
+                      e,
+                    ),
+                  );
+                }
               }),
             )
             .toList();
@@ -69,12 +80,13 @@ class PackageGraphWatcher {
       );
       _readyCompleter.complete();
     }();
-    return StreamGroup.merge(filteredEvents);
+    return StreamGroup.merge([...filteredEvents, restartsController.stream]);
   }
 
-  bool Function(AssetChange) _nestedPathFilter(PackageNode rootNode) {
+  bool Function(AssetChange?) _nestedPathFilter(PackageNode rootNode) {
     final ignorePaths = _nestedPaths(rootNode);
-    return (change) => !ignorePaths.any(change.id.path.startsWith);
+    return (change) =>
+        change == null || !ignorePaths.any(change.id.path.startsWith);
   }
 
   // Returns a set of all package paths that are "nested" within a node.

build_runner/lib/src/watcher/node_watcher.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:io';
 
 import 'package:build_runner_core/build_runner_core.dart';
 import 'package:watcher/watcher.dart';
@@ -29,9 +30,22 @@ class PackageNodeWatcher {
     : _strategy = watch ?? _default;
 
   /// Returns a stream of records for assets that change recursively.
-  Stream<AssetChange> watch() {
+  /*Stream<AssetChange> watch() {
     _watcher = _strategy(node.path);
     final events = _watcher.events;
     return events.map((e) => AssetChange.fromEvent(node, e));
+  }*/
+
+  Stream<AssetChange> watch() {
+    _watcher = _strategy(node.path);
+    final events = _watcher.events;
+    final result = StreamController<AssetChange>();
+    events.listen((e) => result.add(AssetChange.fromEvent(node, e)));
+    Timer.periodic(const Duration(seconds: 20), (_) {
+      result.addError(
+        const FileSystemException('Directory watcher closed unexpectedly'),
+      );
+    });
+    return result.stream;
   }
 }

build_runner/test/watcher/graph_watcher_test.dart

@@ -60,7 +60,7 @@ void main() {
         },
       );
 
-      final events = <AssetChange>[];
+      final events = <AssetChange?>[];
       unawaited(watcher.watch().forEach(events.add));
       await watcher.ready;
 

build_runner_core/lib/src/generate/build_result.dart

@@ -59,6 +59,7 @@ class FailureType {
   static final cantCreate = FailureType._(73);
   static final buildConfigChanged = FailureType._(75);
   static final buildScriptChanged = FailureType._(75);
+  static final watcherRestarted = FailureType._(75);
   final int exitCode;
   FailureType._(this.exitCode);
 }