[2.x] fix: Don't freeze the server when a terminal-properties response is malformed or slow (#9526)

One attached client answering the sbt/terminalpropertiesquery request badly, or
slower than 5 seconds, could freeze the whole server for every client:

- The response handler dropped malformed responses (response.foreach(buffer.put))
  instead of falling back to a default like every sibling handler, so the updater
  thread waiting on the queue timed out with the properties reference still null.
- getProperties(block = true) waits while properties is null, but nothing
  completes it after the updater's one-shot 5-second poll times out: waiters woke
  from the notify, saw null, and waited again with no updater outstanding. The
  1-second lastUpdate throttle also let a caller start waiting with no query in
  flight at all, and the wait condition was checked outside the pending monitor,
  losing wakeups that fired between the check and the wait.
- A response arriving after the poll timeout was delivered into a queue that was
  never deregistered, so it neither set properties nor woke anyone.

The threads that block here include the command loop iterating channels and the
fast-track thread handling attach and cancel, so one bad or briefly-stalled
client wedged prompts, Ctrl-C, and command dispatch server-wide until that
client disconnected.

The properties response handler now falls back to a default like its siblings;
the updater expires its query on timeout, deregistering it (rescuing a response
that raced in) and completing properties with the empty default so waiters
always make progress; and both wait sites hold the pending monitor and gate on
the query in flight. waitForPending gets the same treatment, since it seeds
lazy vals whose initialization otherwise parks every thread touching them.

Regression test: a raw-protocol session that answers every server request with
a result of the wrong shape attaches interactively; a well-behaved batch client
must then still be served twice. Fails on develop with a three-minute timeout
(the server is frozen), passes with this change. A unit spec pins the expiry
semantics, including the late-response rescue.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

main/src/main/scala/sbt/internal/server/NetworkChannel.scala

@@ -754,21 +754,31 @@ final class NetworkChannel(
           pending.set(true)
           val queue = VirtualTerminal.sendTerminalPropertiesQuery(term.name, jsonRpcRequest)
           val update: Runnable = () => {
-            queue.poll(5, java.util.concurrent.TimeUnit.SECONDS) match {
+            try {
-              case null =>
+              queue.poll(5, java.util.concurrent.TimeUnit.SECONDS) match {
-              case t    => properties.set(t)
+                case null =>
-            }
+                  VirtualTerminal.expireTerminalPropertiesQuery(term.name, queue) match {
-            pending.synchronized {
+                    case Some(late) => properties.set(late)
-              lastUpdate.set(Deadline.now)
+                    case None       => Util.ignoreResult(properties.compareAndSet(null, empty))
-              pending.set(false)
+                  }
-              pending.notifyAll()
+                case t => properties.set(t)
+              }
+            } finally {
+              pending.synchronized {
+                lastUpdate.set(Deadline.now)
+                pending.set(false)
+                pending.notifyAll()
+              }
             }
           }
           new Thread(update, s"network-terminal-${term.name}-update") {
             setDaemon(true)
           }.start()
         }
-        while (block && properties.get == null) pending.synchronized(pending.wait())
+        // The updater clears pending inside this monitor before notifying.
+        pending.synchronized {
+          while (block && properties.get == null && pending.get) pending.wait()
+        }
         ()
       } else throw new InterruptedException
     }
@@ -798,7 +808,7 @@ final class NetworkChannel(
       else
         withThread(
           {
-            if (pending.get) pending.synchronized(pending.wait())
+            pending.synchronized { while (pending.get) pending.wait() }
             Option(properties.get).map(f).getOrElse(false)
           },
           false

main/src/main/scala/sbt/internal/server/VirtualTerminal.scala

@@ -84,6 +84,19 @@ object VirtualTerminal {
     jsonRpcRequest(id, terminalCapabilities, query)
     queue
   }
+  private[sbt] def expireTerminalPropertiesQuery(
+      channelName: String,
+      queue: ArrayBlockingQueue[TerminalPropertiesResponse],
+  ): Option[TerminalPropertiesResponse] = {
+    import scala.jdk.CollectionConverters.*
+    pendingTerminalProperties.asScala.collectFirst {
+      case (k @ (`channelName`, _), q) if q eq queue => k
+    } match {
+      case Some(k) if pendingTerminalProperties.remove(k) != null => Option(queue.poll())
+      // The response handler won the removal: its put is imminent, wait it out briefly.
+      case _ => Option(queue.poll(100, java.util.concurrent.TimeUnit.MILLISECONDS))
+    }
+  }
   private[sbt] def cancelRequests(name: String): Unit = {
     import scala.jdk.CollectionConverters.*
     pendingTerminalCapabilities.asScala.foreach {
@@ -191,7 +204,10 @@ object VirtualTerminal {
           r.result.flatMap(Converter.fromJson[TerminalPropertiesResponse](_).toOption)
         pendingTerminalProperties.remove((callback.name, r.id)) match {
           case null   =>
-          case buffer => response.foreach(buffer.put)
+          case buffer =>
+            buffer.put(
+              response.getOrElse(TerminalPropertiesResponse(0, 0, false, false, false, false))
+            )
         }
       case r if pendingTerminalCapabilities.get((callback.name, r.id)) != null =>
         val response =

main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala

@@ -0,0 +1,27 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.server
+
+import sbt.protocol.TerminalPropertiesResponse
+import verify.BasicTestSuite
+
+object VirtualTerminalSpec extends BasicTestSuite:
+  test("expiring an unanswered properties query deregisters it"):
+    val queue = VirtualTerminal.sendTerminalPropertiesQuery("expire-test", (_, _, _) => ())
+    assert(VirtualTerminal.expireTerminalPropertiesQuery("expire-test", queue).isEmpty)
+    // Once expired, the registration is gone: a channel-wide cancel must not touch the queue.
+    VirtualTerminal.cancelRequests("expire-test")
+    assert(queue.poll() == null)
+
+  test("expiring rescues a response that raced in before deregistration"):
+    val queue = VirtualTerminal.sendTerminalPropertiesQuery("expire-race", (_, _, _) => ())
+    val r = TerminalPropertiesResponse(80, 24, true, true, true, true)
+    queue.put(r)
+    assert(VirtualTerminal.expireTerminalPropertiesQuery("expire-race", queue) == Some(r))
+end VirtualTerminalSpec

notes/2.0.0/terminal-properties-freeze.md

@@ -0,0 +1,10 @@
+### One client can no longer freeze the sbt server for every client
+
+An attached client that answered the server's terminal-properties query with a
+malformed or error response, or slower than five seconds, left the channel's
+terminal permanently uninitialized: threads that render prompts and progress,
+including the command loop and the thread handling Ctrl-C, blocked on it
+forever, freezing the server for every connected client until the offending
+client disconnected. Such responses now fall back to default terminal
+properties, unanswered queries expire, and the waits are bounded by the query
+in flight.

server-test/src/test/scala/sbt/protocol/FaultyTerminalSession.scala

@@ -0,0 +1,33 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.protocol
+
+import java.io.File
+import java.net.Socket
+import sbt.internal.protocol.JsonRpcRequestMessage
+import sbt.internal.util.Util
+import sjsonnew.BasicJsonProtocol.*
+
+/**
+ * A client session that answers every server-to-client request with a result of the
+ * wrong shape, for tests exercising the server's handling of malformed responses.
+ */
+final class FaultyTerminalSession(socket: Socket)
+    extends ServerSessionImpl(socket, "faulty-terminal-session-read-thread"):
+  val propertiesQueried = new java.util.concurrent.CountDownLatch(1)
+  override protected def onRequest(msg: JsonRpcRequestMessage): Unit =
+    if msg.method == Serialization.terminalPropertiesQuery then propertiesQueried.countDown()
+    Util.ignoreResult(sendJsonRpcResponse(msg.id, "bogus"))
+end FaultyTerminalSession
+
+object FaultyTerminalSession:
+  def connect(portfile: File): FaultyTerminalSession =
+    val (socket, _) = ClientSocket.socket(portfile, false)
+    new FaultyTerminalSession(socket)
+end FaultyTerminalSession

server-test/src/test/scala/testpkg/AbstractServerTest.scala

@@ -7,9 +7,12 @@
 
 package testpkg
 
-import java.io.File
+import java.io.{ File, InputStream, PrintStream }
 import java.nio.file.{ Files, Path }
+import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit, TimeoutException }
 import scala.concurrent.duration.*
+import sbt.internal.client.NetworkClient
+import sbt.internal.util.Util
 import sbt.io.IO
 import sbt.io.syntax.*
 import sbt.protocol.ServerSession
@@ -95,6 +98,47 @@ trait AbstractServerTest extends AnyFunSuite with BeforeAndAfterAll {
     svr = new SbtServer(session, buildDir, process)
   }
 
+  private object BlockingInputStream extends InputStream {
+    override def read(): Int = {
+      try Thread.sleep(Long.MaxValue)
+      catch { case _: InterruptedException => }
+      -1
+    }
+  }
+  private val nullPrintStream = new PrintStream(_ => {}, false)
+
+  private def background[R](f: => R): R = {
+    val result = new LinkedBlockingQueue[Either[Throwable, R]]
+    val thread = new Thread("server-test-batch-client") {
+      setDaemon(true)
+      override def run(): Unit =
+        try Util.ignoreResult(result.put(Right(f)))
+        catch { case e: Throwable => Util.ignoreResult(result.put(Left(e))) }
+    }
+    thread.start()
+    result.poll(3, TimeUnit.MINUTES) match {
+      case null =>
+        thread.interrupt()
+        thread.join(10000)
+        throw new TimeoutException("client did not complete within 3 minutes")
+      case Left(e)  => throw e
+      case Right(r) => r
+    }
+  }
+
+  /** Runs the thin client in batch mode against this suite's server; returns its exit code. */
+  protected def runBatchClient(args: String*): Int =
+    background(
+      NetworkClient.client(
+        testPath.toFile,
+        args.toArray,
+        BlockingInputStream,
+        nullPrintStream,
+        nullPrintStream,
+        false
+      )
+    )
+
   override protected def afterAll(): Unit = {
     svr.close()
     svr = null

server-test/src/test/scala/testpkg/RebootTest.scala

@@ -8,11 +8,6 @@
 
 package testpkg
 
-import java.io.{ InputStream, PrintStream }
-import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit, TimeoutException }
-import sbt.internal.client.NetworkClient
-import sbt.internal.util.Util
-
 /**
  * Regression for https://github.com/sbt/sbt/issues/9095: `reboot` from a client must bring the
  * server back and complete instead of leaving a zombie server that drops the client.
@@ -20,48 +15,11 @@ import sbt.internal.util.Util
 class RebootTest extends AbstractServerTest {
   override val testDirectory: String = "client"
 
-  private object BlockingInputStream extends InputStream {
-    override def read(): Int = {
-      try Thread.sleep(Long.MaxValue)
-      catch { case _: InterruptedException => }
-      -1
-    }
-  }
-  private val nullPrintStream = new PrintStream(_ => {}, false)
-
-  private def background[R](f: => R): R = {
-    val result = new LinkedBlockingQueue[Either[Throwable, R]]
-    val thread = new Thread("reboot-test-client") {
-      setDaemon(true)
-      override def run(): Unit =
-        try Util.ignoreResult(result.put(Right(f)))
-        catch { case e: Throwable => Util.ignoreResult(result.put(Left(e))) }
-    }
-    thread.start()
-    result.poll(3, TimeUnit.MINUTES) match {
-      case null =>
-        thread.interrupt()
-        thread.join(10000)
-        throw new TimeoutException("client did not complete within 3 minutes")
-      case Left(e)  => throw e
-      case Right(r) => r
-    }
-  }
-
-  private def client(args: String*): Int =
-    background(
-      NetworkClient.client(
-        testPath.toFile,
-        args.toArray,
-        BlockingInputStream,
-        nullPrintStream,
-        nullPrintStream,
-        false
-      )
-    )
-
   test("reboot completes and the rebooted server serves the next command") {
-    assert(client("reboot") == 0, "reboot from a client must complete with exit 0")
+    assert(runBatchClient("reboot") == 0, "reboot from a client must complete with exit 0")
-    assert(client("willSucceed") == 0, "the rebooted server must serve a new client connection")
+    assert(
+      runBatchClient("willSucceed") == 0,
+      "the rebooted server must serve a new client connection"
+    )
   }
 }

server-test/src/test/scala/testpkg/TerminalPropertiesFreezeTest.scala

@@ -0,0 +1,40 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package testpkg
+
+import java.util.concurrent.TimeUnit
+import scala.concurrent.duration.*
+import sbt.internal.util.Util
+import sbt.protocol.{ Attach, FaultyTerminalSession, Serialization }
+import sbt.protocol.codec.JsonProtocol.given
+
+/**
+ * Regression: one attached client that answers the terminal-properties query with a
+ * malformed response must not freeze the server for every other client.
+ */
+class TerminalPropertiesFreezeTest extends AbstractServerTest {
+  override val testDirectory: String = "client"
+
+  test("a client with a broken terminal-properties response does not freeze the server") {
+    val portfile = new java.io.File(testPath.toFile, "project/target/active.json")
+    val faulty = FaultyTerminalSession.connect(portfile)
+    try {
+      faulty.initialize(10.seconds, false).get
+      Util.ignoreResult(
+        faulty.sendJsonRpc(faulty.nextId(), Serialization.attach, Attach(interactive = true))
+      )
+      assert(
+        faulty.propertiesQueried.await(30, TimeUnit.SECONDS),
+        "server never sent the terminal-properties query"
+      )
+      assert(runBatchClient("willSucceed") == 0, "a well-behaved client must still be served")
+      assert(runBatchClient("willSucceed") == 0, "the server must stay serviceable")
+    } finally faulty.close()
+  }
+}