[2.0.x] fix: Read all forked test output before reporting the fork's exit (#9860)

**Problem**
When a forked test JVM exits, WorkerProxy.watch calls notifyExit right
away, and on exit code 0 that completes the run. If the response reader
has not consumed the rest of the stream yet, the remaining events, logs and
the final response are dropped, and the build passes with failing tests
unreported. Slow listeners, or several forks at once, make this likely.

**Solution**
Track when the response stream reaches end of input (the socket reader for
Tcp/Ipc, stdout for Stdio), and have the watch thread wait for it, with a
bound, before calling notifyExit. This is what sbt 1.x did by joining the
acceptor thread. The new scripted test uses a listener that sleeps per
event, and fails without this change.

Fixes #9859

Generated-by: Claude Opus 5.5

Co-authored-by: Eugene <[email protected]>

main-actions/src/main/scala/sbt/internal/WorkerExchange.scala

@@ -24,8 +24,9 @@ import scala.sys.process.{ BasicIO, Process, ProcessIO }
 import scala.collection.mutable
 import scala.collection.concurrent.TrieMap
 import scala.collection.mutable.ListBuffer
-import scala.concurrent.{ Await, Promise }
+import scala.concurrent.{ Await, Future, Promise }
 import scala.concurrent.duration.*
+import scala.util.Try
 import scala.util.control.NonFatal
 
 object WorkerExchange:
@@ -83,11 +84,13 @@ object WorkerExchange:
       IO.classLocationPath(classOf[Gson]).toFile,
     )
     val inputRef = Promise[OutputStream]()
+    val responsesRead = Promise[Unit]()
     def runAccepter(out: OutputStream, in: InputStream): Unit =
       inputRef.success(out)
       val scanner = Scanner(in, "UTF-8")
       while scanner.hasNextLine() do notifyListeners(scanner.nextLine())
-    val (connArgs, closer): (Seq[String], Option[AutoCloseable]) = connectionType match
+      responsesRead.trySuccess(())
+    val (connArgs, closer) = connectionType match
       case WorkerConnection.Tcp =>
         val serverSocket = Retry(ServerSocket(0, 1, loopback))
         val accepter = Thread(() => {
@@ -130,20 +133,24 @@ object WorkerExchange:
     val onStdoutLine: String => Unit = connectionType match
       case WorkerConnection.Stdio => notifyListeners
       case _                      => (line) => scala.Console.out.println(line)
+    def readStdout(stdout: InputStream): Unit =
+      BasicIO.processFully(onStdoutLine)(stdout)
+      if connectionType == WorkerConnection.Stdio then responsesRead.trySuccess(())
     val processIo = ProcessIO(
       in = (input) =>
         (connectionType match
           case WorkerConnection.Stdio => inputRef.success(input)
           case _                      => ()
         ),
-      out = BasicIO.processFully(onStdoutLine),
+      out = readStdout,
       err = BasicIO.processFully((line) => scala.Console.err.println(line)),
     )
     val forkWithIo = fo.withOutputStrategy(OutputStrategy.CustomInputOutput(processIo))
     val p = Fork.java.fork(forkWithIo, options)
     val forkTimeout = fo.connectionTimeout.getOrElse(30.seconds)
     val input = Await.result(inputRef.future, forkTimeout)
-    WorkerProxy(input, p, options, closer)
+    WorkerProxy(input, p, options, closer, responsesRead.future)
+  end startWorker
 
   /** Generates a fresh path suitable for binding a `WorkerConnection.Ipc` socket. */
   def newIpcSocketPath(): NioPath =
@@ -178,6 +185,7 @@ class WorkerProxy(
     val process: Process,
     val options: Seq[String],
     closer: Option[AutoCloseable],
+    responsesRead: Future[Unit],
 ) extends AutoCloseable:
   lazy val inputStream = PrintStream(input)
   def close(): Unit =
@@ -194,11 +202,21 @@ class WorkerProxy(
 
   val watch = Thread(() => {
     while process.isAlive() do Thread.sleep(100)
+    Try(Await.ready(responsesRead, WorkerProxy.responseDrainTimeout))
     WorkerExchange.listeners.foreach(_.notifyExit(process))
   })
   watch.start()
 end WorkerProxy
 
+object WorkerProxy:
+  /**
+   * How long to keep reading a worker's responses after it exits, before reporting the exit.
+   * The reader reaches end of stream as soon as it has consumed everything the worker wrote;
+   * the bound only matters for a worker that died before connecting.
+   */
+  private val responseDrainTimeout = 10.seconds
+end WorkerProxy
+
 abstract class WorkerResponseListener extends Function1[String, Unit]:
   def notifyExit(p: Process): Unit
 

sbt-app/src/sbt-test/tests/fork-slow-listener/build.sbt

@@ -0,0 +1,11 @@
+scalaVersion := "3.9.0"
+Test / fork := true
+libraryDependencies += "org.scalameta" %% "munit" % "1.0.4" % Test
+
+Test / testListeners += new TestsListener:
+  def doInit(): Unit = ()
+  def startGroup(name: String): Unit = ()
+  def testEvent(event: TestEvent): Unit = Thread.sleep(50)
+  def endGroup(name: String, t: Throwable): Unit = ()
+  def endGroup(name: String, result: TestResult): Unit = ()
+  def doComplete(finalResult: TestResult): Unit = ()

sbt-app/src/sbt-test/tests/fork-slow-listener/src/test/scala/ManyTests.scala

@@ -0,0 +1,2 @@
+class ManyTests extends munit.FunSuite:
+  (1 to 40).foreach(i => test(s"test $i")(assert(i < 40)))

sbt-app/src/sbt-test/tests/fork-slow-listener/test

@@ -0,0 +1 @@
+-> test