Make unsuccessful tasks always run the subsequent time when --watch is active (#6263)

Fixes #6258. We keep track of
the unsuccessful tasks in `SelectiveExecution.Metadata#forceRunTasks`
during `--watch` and force those to re-run next evaluation

This ensures that if a user does `--watch` and makes edits to code and
config, they do not see a successful run until all the tasks selected by
the `--watch` are successful. Previously, a user editing a single file
will result in only a subset of tasks running, potentially resulting in
previously-skipped tasks now being selected out and ending up never
being run

Also tested manually by setting up a `--watch __.compile` and then doing
a find-and-replacec `new ([a-zA-Z]+)\(` to `$1(` with IntelliJ. Without
this PR, after fixing the first few failures, the `--watch` appears to
succeed, but if I press Enter to re-run everything I seem more failures.
With this PR, the `--watch` fails consistently until I fix all failures,
and then pressing `Enter` to re-run does not show any new failures

Author: lihaoyi

core/api/src/mill/api/SelectiveExecution.scala

@@ -40,7 +40,8 @@ object SelectiveExecution {
   case class Metadata(
       inputHashes: Map[String, Int],
       codeSignatures: Map[String, Int],
-      @com.lihaoyi.unroll buildOverrideSignatures: Map[String, Int] = Map()
+      @com.lihaoyi.unroll buildOverrideSignatures: Map[String, Int] = Map(),
+      @com.lihaoyi.unroll forceRunTasks: Set[String] = Set()
   )
   object Metadata {
     case class Computed(

core/eval/src/mill/eval/EvaluatorImpl.scala

@@ -271,54 +271,46 @@ final class EvaluatorImpl(
       serialCommandExec = serialCommandExec
     )
 
-    @scala.annotation.nowarn("msg=cannot be checked at runtime")
-    val watched = (evaluated.transitiveResults.iterator ++ selectiveResults)
-      .collect {
-        case (_: Task.Sources, ExecResult.Success(Val(ps: Seq[PathRef]))) =>
-          ps.map(r => Watchable.Path(r.path.toNIO, r.quick, r.sig))
-        case (_: Task.Source, ExecResult.Success(Val(p: PathRef))) =>
-          Seq(Watchable.Path(p.path.toNIO, p.quick, p.sig))
-        case (t: Task.Input[_], result) =>
-
-          val ctx = new mill.api.TaskCtx.Impl(
-            args = Vector(),
-            dest0 = () => null,
-            log = logger,
-            env = this.execution.env,
-            reporter = reporter,
-            testReporter = testReporter,
-            workspace = workspace,
-            _systemExitWithReason = (reason, exitCode) =>
-              throw Exception(s"systemExit called: reason=$reason, exitCode=$exitCode"),
-            fork = null,
-            jobs = execution.effectiveThreadCount,
-            offline = offline
-          )
-          val pretty = t.ctx0.fileName + ":" + t.ctx0.lineNum
-          Seq(Watchable.Value(
-            () => t.evaluate(ctx).hashCode(),
-            result.map(_.value).hashCode(),
-            pretty
-          ))
-      }
-      .flatten
-      .toSeq
-
-    maybeNewMetadata.foreach { newMetadata =>
-      val enclosingModules = PlanImpl
-        .plan(tasks)
-        .transitive
-        .collect { case n: Task.Named[_] => n.ctx.enclosingModule }
-        .distinct
-
-      val scriptBuildOverrides = enclosingModules.flatMap(_.moduleDynamicBuildOverrides)
+    val allResults = evaluated.transitiveResults ++ selectiveResults
 
-      val allBuildOverrides = (staticBuildOverrides ++ scriptBuildOverrides)
-        .map { case (k, v) => (k, v.##) }
+    @scala.annotation.nowarn("msg=cannot be checked at runtime")
+    val watched = allResults.collect {
+      case (_: Task.Sources, ExecResult.Success(Val(ps: Seq[PathRef]))) =>
+        ps.map(r => Watchable.Path(r.path.toNIO, r.quick, r.sig))
+      case (_: Task.Source, ExecResult.Success(Val(p: PathRef))) =>
+        Seq(Watchable.Path(p.path.toNIO, p.quick, p.sig))
+      case (t: Task.Input[_], result) =>
+
+        val ctx = new mill.api.TaskCtx.Impl(
+          args = Vector(),
+          dest0 = () => null,
+          log = logger,
+          env = this.execution.env,
+          reporter = reporter,
+          testReporter = testReporter,
+          workspace = workspace,
+          _systemExitWithReason = (reason, exitCode) =>
+            throw Exception(s"systemExit called: reason=$reason, exitCode=$exitCode"),
+          fork = null,
+          jobs = execution.effectiveThreadCount,
+          offline = offline
+        )
+        val pretty = t.ctx0.fileName + ":" + t.ctx0.lineNum
+        Seq(Watchable.Value(
+          () => t.evaluate(ctx).hashCode(),
+          result.map(_.value).hashCode(),
+          pretty
+        ))
+    }.flatten.toVector
+
+    for (newMetadata <- maybeNewMetadata) {
+      val failingTaskNames = allResults
+        .collect { case (t: Task.Named[_], r) if r.asSuccess.isEmpty => t.ctx.segments.render }
+        .toSet
 
-      this.selective.saveMetadata(
-        SelectiveExecution.Metadata(newMetadata.inputHashes, codeSignatures, allBuildOverrides)
-      )
+      // For tasks that were not successful, force them to re-run next time even
+      // if not changed so the user can see that there are still failures remaining
+      selective.saveMetadata(newMetadata.copy(forceRunTasks = failingTaskNames))
     }
 
     val errorStr = ExecutionResultsApi.formatFailing(evaluated)

core/eval/src/mill/eval/SelectiveExecutionImpl.scala

@@ -63,8 +63,9 @@ class SelectiveExecutionImpl(evaluator: Evaluator)
       newHashes.buildOverrideSignatures
     )
 
-    val changedRootTasks = (changedInputNames ++ changedCodeNames ++ changedBuildOverrides)
-      .flatMap(namesToTasks.get(_): Option[Task[?]])
+    val changedRootTasks =
+      (changedInputNames ++ changedCodeNames ++ changedBuildOverrides ++ oldHashes.forceRunTasks)
+        .flatMap(namesToTasks.get(_): Option[Task[?]])
 
     val allNodes = breadthFirst(transitiveNamed.map(t => t: Task[?]))(_.inputs)
     val downstreamEdgeMap = SpanningForest.reverseEdges(allNodes.map(t => (t, t.inputs)))
@@ -117,7 +118,7 @@ class SelectiveExecutionImpl(evaluator: Evaluator)
       val transitiveNamed = PlanImpl.transitiveNamed(tasks)
       val oldMetadata = upickle.read[SelectiveExecution.Metadata](oldMetadataTxt)
       val (changedRootTasks, downstreamTasks) =
-        evaluator.selective.computeDownstream(
+        computeDownstream(
           transitiveNamed,
           oldMetadata,
           computedMetadata.metadata

integration/invalidation/selective-execution/src/SelectiveExecutionChangedCodeTests.scala

@@ -0,0 +1,52 @@
+package mill.integration
+import mill.testkit.UtestIntegrationTestSuite
+import scala.concurrent.duration._
+
+import utest._
+import utest.asserts.{RetryMax, RetryInterval}
+
+object SelectiveExecutionChangedCodeTests extends UtestIntegrationTestSuite {
+  implicit val retryMax: RetryMax = RetryMax(120.seconds)
+  implicit val retryInterval: RetryInterval = RetryInterval(1.seconds)
+  val tests: Tests = Tests {
+    test("changed-code") - integrationTest { tester =>
+      import tester._
+
+      // Check method body code changes correctly trigger downstream evaluation
+      eval(
+        ("selective.prepare", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+      modifyFile(workspacePath / "build.mill", _.replace("\"barHelper \"", "\"barHelper! \""))
+      val cached1 = eval(
+        ("selective.run", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(!cached1.out.contains("Computing fooCommand"))
+      assert(cached1.out.contains("Computing barCommand"))
+
+      // Check module body code changes correctly trigger downstream evaluation
+      eval(
+        ("selective.prepare", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+      modifyFile(
+        workspacePath / "build.mill",
+        _.replace("object foo extends Module {", "object foo extends Module { println(123)")
+      )
+      val cached2 = eval(
+        ("selective.run", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(cached2.out.contains("Computing fooCommand"))
+      assert(!cached2.out.contains("Computing barCommand"))
+    }
+
+  }
+}

integration/invalidation/selective-execution/src/SelectiveExecutionChangedInputsTests.scala

@@ -0,0 +1,82 @@
+package mill.integration
+import mill.testkit.UtestIntegrationTestSuite
+import scala.concurrent.duration._
+
+import utest._
+import utest.asserts.{RetryMax, RetryInterval}
+
+object SelectiveExecutionChangedInputsTests extends UtestIntegrationTestSuite {
+  implicit val retryMax: RetryMax = RetryMax(120.seconds)
+  implicit val retryInterval: RetryInterval = RetryInterval(1.seconds)
+  val tests: Tests = Tests {
+    test("changed-inputs") - integrationTest { tester =>
+      import tester._
+
+      eval(
+        ("selective.prepare", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      // no op
+      val noOp = eval(
+        ("selective.run", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(!noOp.out.contains("Computing fooCommand"))
+      assert(!noOp.out.contains("Computing barCommand"))
+
+      // After one input changed
+      modifyFile(workspacePath / "bar/bar.txt", _ + "!")
+
+      val resolve = eval(("selective.resolve", "{foo.fooCommand,bar.barCommand}"), check = true)
+      assert(resolve.out == "bar.barCommand")
+
+      val cached = eval(
+        ("selective.run", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(!cached.out.contains("Computing fooCommand"))
+      assert(cached.out.contains("Computing barCommand"))
+
+      // zero out the `mill-selective-execution.json` file to run all tasks
+      os.write.over(workspacePath / "out/mill-selective-execution.json", "")
+      val runAll = eval(
+        ("selective.run", "{foo.fooCommand,bar.barCommand}"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(runAll.out.contains("Computing fooCommand"))
+      assert(runAll.out.contains("Computing barCommand"))
+    }
+
+    test("changed-inputs-generic") - integrationTest { tester =>
+      // Make sure you can run `selective.prepare` on a broader set of tasks than
+      // `selective.resolve` or `selective.run` and thingsstill work
+      import tester._
+
+      // `selective.prepare` defaults to `__` if no selector is passed
+      eval(("selective.prepare"), check = true)
+      modifyFile(workspacePath / "bar/bar.txt", _ + "!")
+
+      val resolve = eval(("selective.resolve", "bar.barCommand"), check = true)
+      assert(resolve.out == "bar.barCommand")
+      val resolve2 = eval(("selective.resolve", "foo.fooCommand"), check = true)
+      assert(resolve2.out == "")
+
+      val cached = eval(
+        ("selective.run", "bar.barCommand"),
+        check = true,
+        stderr = os.Inherit
+      )
+
+      assert(!cached.out.contains("Computing fooCommand"))
+      assert(cached.out.contains("Computing barCommand"))
+    }
+  }
+}