Aggregate watch events

It is possible for sbt to get into a weird state when in a continuous
build when the auto reload feature is on and a source file and a build
file are changed in a small window of time. If sbt detects the source
file first, it will start running the command but then it will
autoreload when it runs the command because of the build file change.
This causes the watch to get into a broken state because it is necessary
to completely restart the watch after sbt exits.

To fix this, we can aggregate the detected events in a 100ms window. The
idea is to handle bursts of file events so we poll in 5ms increments and
as soon as no events are detected, we trigger a build.

main/src/main/scala/sbt/internal/Continuous.scala

@@ -465,16 +465,41 @@ private[sbt] object Continuous extends DeprecatedContinuous {
         }
       }
 
+      private[this] val antiEntropyWindow = configs.map(_.watchSettings.antiEntropy).max
       private[this] val monitor = FileEventMonitor.antiEntropy(
         eventMonitorObservers,
-        configs.map(_.watchSettings.antiEntropy).max,
+        antiEntropyWindow,
         logger,
         quarantinePeriod,
         retentionPeriod
       )
 
-      override def poll(duration: Duration, filter: Event => Boolean): Seq[Event] =
-        monitor.poll(duration, filter)
+      private[this] val antiEntropyPollPeriod =
+        configs.map(_.watchSettings.antiEntropyPollPeriod).max
+      override def poll(duration: Duration, filter: Event => Boolean): Seq[Event] = {
+        monitor.poll(duration, filter) match {
+          case s if s.nonEmpty =>
+            val limit = antiEntropyWindow.fromNow
+            /*
+             * File events may come in bursts so we poll for a short time to see if there
+             * are other changes detected in the burst. As soon as no changes are detected
+             * during the polling window, we return all of the detected events. The polling
+             * period is by default 5 milliseconds which is short enough to detect bursts
+             * induced by commands like git rebase but fast enough to not lead to a noticable
+             * increase in latency.
+             */
+            @tailrec def aggregate(res: Seq[Event]): Seq[Event] =
+              if (limit.isOverdue) res
+              else {
+                monitor.poll(antiEntropyPollPeriod) match {
+                  case s if s.nonEmpty => aggregate(res ++ s)
+                  case _               => res
+                }
+              }
+            aggregate(s)
+          case s => s
+        }
+      }
 
       override def close(): Unit = {
         configHandle.close()
@@ -865,6 +890,8 @@ private[sbt] object Continuous extends DeprecatedContinuous {
     // alternative would be SettingKey[() => InputStream], but that doesn't feel right because
     // one might want the InputStream to depend on other tasks.
     val inputStream: Option[TaskKey[InputStream]] = key.get(watchInputStream)
+    val antiEntropyPollPeriod: FiniteDuration =
+      key.get(watchAntiEntropyPollPeriod).getOrElse(Watch.defaultAntiEntropyPollPeriod)
   }
 
   /**

main/src/main/scala/sbt/nio/Keys.scala

@@ -67,6 +67,9 @@ object Keys {
   val watchAntiEntropyRetentionPeriod = settingKey[FiniteDuration](
     "Wall clock Duration for which a FileEventMonitor will store anti-entropy events. This prevents spurious triggers when a task takes a long time to run. Higher values will consume more memory but make spurious triggers less likely."
   ).withRank(BMinusSetting)
+  val watchAntiEntropyPollPeriod = settingKey[FiniteDuration](
+    "The duration for which sbt will poll for file events during the window in which sbt is buffering file events"
+  )
   val onChangedBuildSource = settingKey[WatchBuildSourceOption](
     "Determines what to do if the sbt meta build sources have changed"
   ).withRank(DSetting)

main/src/main/scala/sbt/nio/Watch.scala

@@ -450,6 +450,14 @@ object Watch {
    */
   final val defaultAntiEntropy: FiniteDuration = 500.milliseconds
 
+  /**
+   * The duration for which we will poll for new file events when we are buffering events
+   * after an initial event has been detected to avoid spurious rebuilds.
+   *
+   * If this value is ever updated, please update the comment in Continuous.getFileEvents.
+   */
+  final val defaultAntiEntropyPollPeriod: FiniteDuration = 5.milliseconds
+
   /**
    * The duration in wall clock time for which a FileEventMonitor will retain anti-entropy
    * events for files. This is an implementation detail of the FileEventMonitor. It should
@@ -613,5 +621,6 @@ object Watch {
     watchForceTriggerOnAnyChange :== false,
     watchPersistFileStamps := (sbt.Keys.turbo in ThisBuild).value,
     watchTriggers :== Nil,
+    watchAntiEntropyPollPeriod := Watch.defaultAntiEntropyPollPeriod,
   )
 }