Track threads & futures spawned in integration tests to make sure logs are in chronological order (#4293)

Gedochao · web-flow · commit d85dee92d3a1 · 2026-06-01T12:04:31.000+02:00
diff --git a/modules/integration/src/test/scala/scala/cli/integration/BspSuite.scala b/modules/integration/src/test/scala/scala/cli/integration/BspSuite.scala
@@ -126,20 +126,24 @@ trait BspSuite { this: ScalaCliSuite =>
       val stdErrPathOpt: Option[os.ProcessOutput] = stdErrOpt.map(path => root / path)
       val stderr: os.ProcessOutput                = stdErrPathOpt.getOrElse(os.Inherit)
 
-      val proc = os.proc(TestUtil.cli, "bsp", bspOptions ++ extraOptionsOverride, args)
-        .spawn(cwd = root, stderr = stderr, env = bspEnvs)
+      val proc = trackSubprocess(
+        os.proc(TestUtil.cli, "bsp", bspOptions ++ extraOptionsOverride, args)
+          .spawn(cwd = root, stderr = stderr, env = bspEnvs)
+      )
       var remoteServer
         : b.BuildServer & b.ScalaBuildServer & b.JavaBuildServer & b.JvmBuildServer &
           TestBspClient.WrappedSourcesBuildServer = null
 
       val bspServerExited = Promise[Unit]()
-      val t               = new Thread("bsp-server-watcher") {
-        setDaemon(true)
-        override def run() = {
-          proc.join()
-          bspServerExited.success(())
+      val t               = trackThread(
+        new Thread("bsp-server-watcher") {
+          setDaemon(true)
+          override def run() = {
+            proc.join()
+            bspServerExited.success(())
+          }
         }
-      }
+      )
       t.start()
 
       def whileBspServerIsRunning[T](f: Future[T]): Future[T] = {
@@ -153,8 +157,9 @@ trait BspSuite { this: ScalaCliSuite =>
       }
 
       try {
-        val (localClient, remoteServer0, _) =
+        val (localClient, remoteServer0, listeningFuture) =
           TestBspClient.connect(proc.stdout, proc.stdin, pool)
+        trackFuture(listeningFuture)
         remoteServer = remoteServer0
         val initRes: b.InitializeBuildResult = Await.result(
           whileBspServerIsRunning(
@@ -179,6 +184,7 @@ trait BspSuite { this: ScalaCliSuite =>
         proc.destroy()
         proc.join(2.seconds.toMillis)
         proc.destroy(shutdownGracePeriod = 0)
+        proc.join()
       }
     }
 
diff --git a/modules/integration/src/test/scala/scala/cli/integration/ResourceTracker.scala b/modules/integration/src/test/scala/scala/cli/integration/ResourceTracker.scala
@@ -0,0 +1,67 @@
+package scala.cli.integration
+
+import java.util.concurrent.{ConcurrentLinkedQueue, TimeUnit}
+
+import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.{Await, Future}
+import scala.util.control.NonFatal
+
+private[integration] class ResourceTracker {
+  private val subprocesses                   = new ConcurrentLinkedQueue[os.SubProcess]
+  private val threads                        = new ConcurrentLinkedQueue[Thread]
+  private val futures                        = new ConcurrentLinkedQueue[Future[?]]
+  protected val drainTimeout: FiniteDuration = FiniteDuration(5, TimeUnit.SECONDS)
+  private val threadJoinTimeoutMillis        = drainTimeout.toMillis
+
+  def trackSubprocess[P <: os.SubProcess](proc: P): P = {
+    subprocesses.add(proc)
+    proc
+  }
+
+  def trackThread[T <: Thread](thread: T): T = {
+    threads.add(thread)
+    thread
+  }
+
+  def trackFuture[F <: Future[?]](future: F): F = {
+    futures.add(future)
+    future
+  }
+
+  def clear(): Unit = {
+    subprocesses.clear()
+    threads.clear()
+    futures.clear()
+  }
+
+  def drain(): Unit = {
+    drainAll(subprocesses, drainSubprocess)
+    drainAll(threads, drainThread)
+    drainAll(futures, drainFuture)
+  }
+
+  private def drainAll[A](queue: ConcurrentLinkedQueue[A], drain: A => Unit): Unit = {
+    var next = queue.poll()
+    while next != null do
+      drain(next)
+      next = queue.poll()
+  }
+
+  protected def drainSubprocess(proc: os.SubProcess): Unit =
+    try
+      if proc.isAlive() then
+        proc.destroy()
+        proc.join(drainTimeout.toMillis)
+        if proc.isAlive() then
+          proc.destroy(shutdownGracePeriod = 0)
+          proc.join(drainTimeout.toMillis)
+    catch case NonFatal(_) => ()
+
+  protected def drainThread(thread: Thread): Unit =
+    try thread.join(threadJoinTimeoutMillis)
+    catch case NonFatal(_) => ()
+
+  protected def drainFuture(future: Future[?]): Unit =
+    try Await.ready(future, drainTimeout)
+    catch case NonFatal(_) => ()
+}
diff --git a/modules/integration/src/test/scala/scala/cli/integration/ResourceTrackerTests.scala b/modules/integration/src/test/scala/scala/cli/integration/ResourceTrackerTests.scala
@@ -0,0 +1,98 @@
+package scala.cli.integration
+
+import com.eed3si9n.expecty.Expecty.expect
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.concurrent.{Future, Promise}
+import scala.jdk.CollectionConverters.*
+import scala.util.Properties
+
+class ResourceTrackerTests extends munit.FunSuite {
+
+  private final class RecordingResourceTracker(events: ConcurrentLinkedQueue[String])
+      extends ResourceTracker {
+    override protected def drainSubprocess(proc: os.SubProcess): Unit =
+      events.add("subprocess")
+
+    override protected def drainThread(thread: Thread): Unit =
+      events.add("thread")
+
+    override protected def drainFuture(future: Future[?]): Unit =
+      events.add("future")
+  }
+
+  private final class NoOpResourceTracker extends ResourceTracker {
+    override protected def drainSubprocess(proc: os.SubProcess): Unit = ()
+
+    override protected def drainThread(thread: Thread): Unit = ()
+
+    override protected def drainFuture(future: Future[?]): Unit = ()
+  }
+
+  private def eventsToList(events: ConcurrentLinkedQueue[String]): List[String] =
+    events.iterator().asScala.toList
+
+  private def shortLivedSubprocess(): os.SubProcess =
+    if Properties.isWin then os.proc("cmd", "/c", "exit", "0").spawn()
+    else os.proc("true").spawn()
+
+  test("drain order is subprocess -> thread -> future") {
+    val events  = new ConcurrentLinkedQueue[String]
+    val tracker = new RecordingResourceTracker(events)
+
+    val future1 = Promise[Unit]().future
+    val future2 = Promise[Unit]().future
+    val thread  = new Thread(() => ())
+    val proc1   = shortLivedSubprocess()
+    val proc2   = shortLivedSubprocess()
+
+    tracker.trackFuture(future1)
+    tracker.trackSubprocess(proc1)
+    tracker.trackThread(thread)
+    tracker.trackFuture(future2)
+    tracker.trackSubprocess(proc2)
+
+    tracker.drain()
+
+    expect(
+      eventsToList(events) == List("subprocess", "subprocess", "thread", "future", "future")
+    )
+  }
+
+  test("drain consumes tracked resources (idempotent on empty)") {
+    val events  = new ConcurrentLinkedQueue[String]
+    val tracker = new RecordingResourceTracker(events)
+
+    tracker.trackFuture(Promise[Unit]().future)
+    tracker.trackSubprocess(shortLivedSubprocess())
+    tracker.drain()
+    val afterFirstDrain = eventsToList(events)
+
+    tracker.drain()
+
+    expect(eventsToList(events) == afterFirstDrain)
+  }
+
+  test("clear discards without invoking drain hooks") {
+    val events  = new ConcurrentLinkedQueue[String]
+    val tracker = new RecordingResourceTracker(events)
+
+    tracker.trackFuture(Promise[Unit]().future)
+    tracker.trackSubprocess(shortLivedSubprocess())
+    tracker.trackThread(new Thread(() => ()))
+    tracker.clear()
+    tracker.drain()
+
+    expect(eventsToList(events) == Nil)
+  }
+
+  test("track* returns the tracked instance") {
+    val tracker = new NoOpResourceTracker
+    val thread  = new Thread(() => ())
+    val future  = Promise[Unit]().future
+
+    expect(tracker.trackThread(thread) == thread)
+    expect(tracker.trackFuture(future) == future)
+  }
+}
diff --git a/modules/integration/src/test/scala/scala/cli/integration/ScalaCliSuite.scala b/modules/integration/src/test/scala/scala/cli/integration/ScalaCliSuite.scala
@@ -2,10 +2,24 @@ package scala.cli.integration
 
 import java.util.concurrent.TimeUnit
 
+import scala.concurrent.Future
 import scala.concurrent.duration.{Duration, FiniteDuration}
 import scala.util.Properties
 
 abstract class ScalaCliSuite extends munit.FunSuite {
+  given scalaCliSuite: ScalaCliSuite = this
+
+  private val resourceTracker = new ResourceTracker
+
+  private[integration] def trackSubprocess[P <: os.SubProcess](proc: P): P =
+    resourceTracker.trackSubprocess(proc)
+
+  private[integration] def trackThread[T <: Thread](thread: T): T =
+    resourceTracker.trackThread(thread)
+
+  private[integration] def trackFuture[F <: Future[?]](future: F): F =
+    resourceTracker.trackFuture(future)
+
   implicit class BeforeEachOpts(munitContext: BeforeEach) {
     def locationAbsolutePath: os.Path = os.Path(munitContext.test.location.path)
   }
@@ -17,13 +31,17 @@ abstract class ScalaCliSuite extends munit.FunSuite {
     def apply(): Unit = ()
 
     override def beforeEach(context: BeforeEach): Unit = {
+      resourceTracker.clear()
       val fileName = context.locationAbsolutePath.baseName
       System.err.println(
         s">==== ${Console.CYAN}Running '${context.test.name}' from $fileName${Console.RESET}"
       )
     }
 
     override def afterEach(context: AfterEach): Unit = {
+      resourceTracker.drain()
+      System.out.flush()
+      System.err.flush()
       val fileName = context.locationAbsolutePath.baseName
       System.err.println(
         s"X==== ${Console.CYAN}Finishing '${context.test.name}' from $fileName${Console.RESET}"
diff --git a/modules/integration/src/test/scala/scala/cli/integration/TestUtil.scala b/modules/integration/src/test/scala/scala/cli/integration/TestUtil.scala
@@ -199,33 +199,35 @@ object TestUtil {
     user: String,
     password: String,
     realm: String
-  )(f: (String, Int) => T): T = {
+  )(f: (String, Int) => T)(using suite: ScalaCliSuite): T = {
     val host = "127.0.0.1"
     val port = {
       val s = new ServerSocket(0)
       try s.getLocalPort()
       finally s.close()
     }
-    val proc = os.proc(
-      cs,
-      "launch",
-      "io.get-coursier:http-server_2.12:1.0.1",
-      "--",
-      "--user",
-      user,
-      "--password",
-      password,
-      "--realm",
-      realm,
-      "--directory",
-      ".",
-      "--host",
-      host,
-      "--port",
-      port,
-      "-v"
+    val proc = suite.trackSubprocess(
+      os.proc(
+        cs,
+        "launch",
+        "io.get-coursier:http-server_2.12:1.0.1",
+        "--",
+        "--user",
+        user,
+        "--password",
+        password,
+        "--realm",
+        realm,
+        "--directory",
+        ".",
+        "--host",
+        host,
+        "--port",
+        port,
+        "-v"
+      )
+        .spawn(cwd = path, mergeErrIntoOut = true)
     )
-      .spawn(cwd = path, mergeErrIntoOut = true)
     try {
 
       // a timeout around this would be great…
@@ -245,19 +247,21 @@ object TestUtil {
       System.err.println(s"Waiting $waitFor")
       Thread.sleep(waitFor.toMillis)
 
-      val t = new Thread("test-http-server-output") {
-        setDaemon(true)
-        override def run(): Unit = {
-          var line = ""
-          while (
-            proc.isAlive() && {
-              line = proc.stdout.readLine()
-              line != null
-            }
-          )
-            System.err.println(line)
+      val t = suite.trackThread(
+        new Thread("test-http-server-output") {
+          setDaemon(true)
+          override def run(): Unit = {
+            var line = ""
+            while (
+              proc.isAlive() && {
+                line = proc.stdout.readLine()
+                line != null
+              }
+            )
+              System.err.println(line)
+          }
         }
-      }
+      )
       t.start()
       f(host, port)
     }
@@ -376,7 +380,8 @@ object TestUtil {
     threadName: String = UUID.randomUUID().toString,
     poolSize: Int = 2,
     timeout: Duration = 90.seconds
-  )(f: (os.SubProcess, Duration, ExecutionContext) => Unit): Unit =
+  )(f: (os.SubProcess, Duration, ExecutionContext) => Unit)(using suite: ScalaCliSuite): Unit =
+    suite.trackSubprocess(proc)
     try withThreadPool(threadName, poolSize) { pool =>
         f(proc, timeout, ExecutionContext.fromExecutorService(pool))
       }
