[2.x] fix: Don't strand the server when a client dies with a terminal query unanswered (#9527)

A client disconnecting with a terminal control query outstanding could park a
server thread forever, wedging prompts and command dispatch for every client
(#6841, #6840):

- VirtualTerminal.cancelRequests drained only 2 of the 8 pending terminal
  maps, so waiters on the set-echo, raw-mode, attributes, and size queues were
  never woken. It now drains all of them, with offer instead of put so the
  shutdown path itself cannot block on a full queue.
- Raw-mode requests were registered in the set-echo map, so their waiters were
  invisible to any raw-mode-specific handling.
- Closing a channel terminal did not wake readers parked on its input stream;
  close now delivers EOF so a prompt blocked on a dead client's input unwinds.
- The failed-load prompt read its answer byte from System.in, which under
  non-virtual IO is the process's own stdin and never carries client input; it
  now reads the active terminal's input stream.
- ServerSessionImpl.close() could not deliver EOF to the peer while its read
  thread was parked in a native read (the native close is never delivered), so
  the server never noticed orderly client disconnects at all. It now shuts
  down socket input first, which wakes the reader and lets the close through.

Regression test: a raw-protocol client that attaches, triggers the failed-load
prompt without answering the raw-mode query, and disconnects; the server must
shut down cleanly (EOF at the prompt maps to 'q') instead of staying parked
forever. Fails on develop with the server still alive and the command loop
parked in setRawMode; passes with this change. VirtualTerminalSpec pins the
drain across all eight maps and that other channels are untouched.

Generated-by: kimi-code/k3 (Oh My Pi)

internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala

@@ -640,6 +640,7 @@ object Terminal {
     }
     override def close(): Unit = if (closed.compareAndSet(false, true)) {
       executor.shutdownNow()
+      buffer.synchronized(buffer.put(-1: Integer))
       ()
     }
   }

internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.util
+
+import java.io.PipedInputStream
+import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit }
+import verify.BasicTestSuite
+
+object WriteableInputStreamSpec extends BasicTestSuite:
+  test("close wakes a parked reader with EOF"):
+    // an underlying stream that never delivers a byte, so the reader stays parked
+    val stream = Terminal.WriteableInputStream(new PipedInputStream(), "test")
+    val result = new LinkedBlockingQueue[Integer]
+    val reader = new Thread(() => result.put(stream.read()))
+    reader.setDaemon(true)
+    reader.start()
+    // let the reader park inside the buffer take
+    Thread.sleep(500)
+    stream.close()
+    assert(result.poll(5, TimeUnit.SECONDS) == -1)
+    assert(stream.read() == -1)
+end WriteableInputStreamSpec

main/src/main/scala/sbt/Main.scala

@@ -939,7 +939,7 @@ object BuiltinCommands {
       s.log.warn("Project loading failed: (r)etry, (q)uit, (l)ast, or (i)gnore? (default: r)")
       val result: Int =
         try
-          ITerminal.get.withRawInput(System.in.read) match {
+          ITerminal.get.withRawInput(ITerminal.get.inputStream.read) match {
             case -1 => 'q'.toInt
             case b  => b
           }

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

@@ -30,6 +30,7 @@ import sbt.protocol.Serialization.{
   terminalSetRawMode,
 }
 import sjsonnew.support.scalajson.unsafe.Converter
+import sbt.internal.util.Util
 import sbt.protocol.{
   Attach,
   TerminalAttributesQuery,
@@ -99,18 +100,24 @@ object VirtualTerminal {
   }
   private[sbt] def cancelRequests(name: String): Unit = {
     import scala.jdk.CollectionConverters.*
-    pendingTerminalCapabilities.asScala.foreach {
-      case (k @ (`name`, _), q) =>
-        pendingTerminalCapabilities.remove(k)
-        q.put(TerminalCapabilitiesResponse(None, None, None))
-      case _ =>
-    }
-    pendingTerminalProperties.asScala.foreach {
-      case (k @ (`name`, _), q) =>
-        pendingTerminalProperties.remove(k)
-        q.put(TerminalPropertiesResponse(0, 0, false, false, false, false))
-      case _ =>
-    }
+    def drain[A](
+        map: ConcurrentHashMap[(String, String), ArrayBlockingQueue[A]],
+        default: A
+    ): Unit =
+      map.asScala.foreach {
+        case (k @ (`name`, _), q) =>
+          map.remove(k)
+          Util.ignoreResult(q.offer(default))
+        case _ =>
+      }
+    drain(pendingTerminalCapabilities, TerminalCapabilitiesResponse(None, None, None))
+    drain(pendingTerminalProperties, TerminalPropertiesResponse(0, 0, false, false, false, false))
+    drain(pendingTerminalAttributes, TerminalAttributesResponse("", "", "", "", ""))
+    drain(pendingTerminalSetAttributes, ())
+    drain(pendingTerminalSetSize, ())
+    drain(pendingTerminalGetSize, TerminalGetSizeResponse(1, 1))
+    drain(pendingTerminalSetEcho, ())
+    drain(pendingTerminalSetRawMode, ())
   }
   private[sbt] def sendTerminalAttributesQuery(
       channelName: String,
@@ -177,7 +184,7 @@ object VirtualTerminal {
   ): ArrayBlockingQueue[Unit] = {
     val id = UUID.randomUUID.toString
     val queue = new ArrayBlockingQueue[Unit](1)
-    pendingTerminalSetEcho.put((channelName, id), queue)
+    pendingTerminalSetRawMode.put((channelName, id), queue)
     jsonRpcRequest(id, terminalSetRawMode, query)
     queue
   }

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

@@ -8,7 +8,14 @@
 
 package sbt.internal.server
 
-import sbt.protocol.TerminalPropertiesResponse
+import sbt.protocol.{
+  TerminalCapabilitiesQuery,
+  TerminalPropertiesResponse,
+  TerminalSetAttributesCommand,
+  TerminalSetEchoCommand,
+  TerminalSetRawModeCommand,
+  TerminalSetSizeCommand,
+}
 import verify.BasicTestSuite
 
 object VirtualTerminalSpec extends BasicTestSuite:
@@ -24,4 +31,34 @@ object VirtualTerminalSpec extends BasicTestSuite:
     val r = TerminalPropertiesResponse(80, 24, true, true, true, true)
     queue.put(r)
     assert(VirtualTerminal.expireTerminalPropertiesQuery("expire-race", queue) == Some(r))
+
+  test("cancelRequests wakes waiters on every pending terminal map"):
+    val name = "drain-test"
+    val props = VirtualTerminal.sendTerminalPropertiesQuery(name, (_, _, _) => ())
+    val caps = VirtualTerminal.sendTerminalCapabilitiesQuery(
+      name,
+      (_, _, _) => (),
+      TerminalCapabilitiesQuery(None, None, None)
+    )
+    val attrs = VirtualTerminal.sendTerminalAttributesQuery(name, (_, _, _) => ())
+    val setAttrs = VirtualTerminal.setTerminalAttributesCommand(
+      name,
+      (_, _, _) => (),
+      TerminalSetAttributesCommand("", "", "", "", "")
+    )
+    val setSize =
+      VirtualTerminal.setTerminalSize(name, (_, _, _) => (), TerminalSetSizeCommand(80, 24))
+    val getSize = VirtualTerminal.getTerminalSize(name, (_, _, _) => ())
+    val echo = VirtualTerminal.setTerminalEcho(name, (_, _, _) => (), TerminalSetEchoCommand(true))
+    val raw =
+      VirtualTerminal.setTerminalRawMode(name, (_, _, _) => (), TerminalSetRawModeCommand(true))
+    VirtualTerminal.cancelRequests(name)
+    List(props, caps, attrs, setAttrs, setSize, getSize, echo, raw).foreach { q =>
+      assert(q.poll() != null)
+    }
+
+  test("cancelRequests leaves other channels' waiters parked"):
+    val queue = VirtualTerminal.sendTerminalPropertiesQuery("other-channel", (_, _, _) => ())
+    VirtualTerminal.cancelRequests("drain-test-2")
+    assert(queue.poll() == null)
 end VirtualTerminalSpec

notes/2.0.0/terminal-client-disconnect.md

@@ -0,0 +1,10 @@
+### Fixes
+
+- Don't strand server threads when a client disconnects with a terminal control query unanswered. `VirtualTerminal.cancelRequests` now wakes waiters on all pending terminal maps (set-echo, raw-mode, attributes, size), not just properties and capabilities. Fixes the server-wide freeze where one dying client wedges prompts and command dispatch for every client. ([#6841][6841], [#6840][6840])
+- Register raw-mode requests in the raw-mode map (they were registered in the set-echo map).
+- Wake terminal input readers with EOF when a channel's terminal closes, so a prompt blocked on a dead client's input unwinds instead of parking forever.
+- Read the failed-load prompt byte from the active terminal's input stream instead of `System.in`, which under non-virtual IO is the process's own stdin and never carries client input.
+- Deliver EOF reliably on client close: `ServerSessionImpl.close()` now shuts down socket input before closing, so a client disconnect is noticed by the server even while the client's read thread is parked.
+
+[6841]: https://github.com/sbt/sbt/issues/6841
+[6840]: https://github.com/sbt/sbt/issues/6840

protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala

@@ -135,10 +135,12 @@ private[sbt] class ServerSessionImpl(
    */
   override def close(): Unit = if (closed.compareAndSet(false, true)) {
     running.set(false)
-    try
+    try {
+      // close() cannot deliver EOF while the read thread is parked in a native read
+      socket.shutdownInput()
       out.close()
       socket.close()
-    catch case _: IOException => ()
+    } catch case _: IOException => ()
     onClose()
     if Thread.currentThread() != readThread then
       try readThread.joinFor(ServerSessionImpl.ReadThreadDestroyTimeout)

protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala

@@ -0,0 +1,36 @@
+/*
+ * 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.nio.file.Files
+import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit }
+import org.scalasbt.ipcsocket.{ UnixDomainServerSocket, UnixDomainSocket }
+import verify.BasicTestSuite
+
+object ServerSessionImplSpec extends BasicTestSuite:
+  private val isWin = System.getProperty("os.name").toLowerCase.contains("win")
+  test("close delivers EOF to the peer while the read thread is parked"):
+    // named pipes have different close semantics; the swallowed-close mechanism is unix-specific
+    if isWin then ()
+    else
+      val path = Files.createTempDirectory("session-eof").resolve("sock")
+      val server = UnixDomainServerSocket(path.toString, false)
+      val peerResult = new LinkedBlockingQueue[Integer]
+      val accepted = new Thread(() => {
+        val conn = server.accept()
+        peerResult.put(conn.getInputStream.read())
+      })
+      accepted.setDaemon(true)
+      accepted.start()
+      val session = new ServerSessionImpl(UnixDomainSocket(path.toString, false))
+      // let the session's read thread park in its native read
+      Thread.sleep(500)
+      session.close()
+      assert(peerResult.poll(10, TimeUnit.SECONDS) == -1)
+end ServerSessionImplSpec

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

@@ -0,0 +1,54 @@
+/*
+ * 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 java.util.concurrent.CountDownLatch
+import java.util.concurrent.atomic.AtomicReference
+import sbt.internal.protocol.JsonRpcRequestMessage
+import sbt.internal.protocol.JsonRpcNotificationMessage
+import sbt.internal.util.Util
+import sjsonnew.BasicJsonProtocol.*
+
+/**
+ * A client session that never answers the terminal control queries (set echo, set raw
+ * mode, attributes, size), for tests exercising the server's handling of a client that
+ * stalls or dies with such a query outstanding.
+ */
+final class SilentTerminalSession(socket: Socket)
+    extends ServerSessionImpl(socket, "silent-terminal-session-read-thread"):
+  import SilentTerminalSession.silentMethods
+  val silentQuery = new CountDownLatch(1)
+  val firstSilentQuery = new AtomicReference[String]
+  val inputRequested = new CountDownLatch(1)
+  override protected def onRequest(msg: JsonRpcRequestMessage): Unit =
+    msg.method match
+      case m if silentMethods(m) =>
+        Util.ignoreResult(firstSilentQuery.compareAndSet(null, m))
+        silentQuery.countDown()
+      case _ => Util.ignoreResult(sendJsonRpcResponse(msg.id, "bogus"))
+  override protected def onNotification(msg: JsonRpcNotificationMessage): Unit =
+    if msg.method == Serialization.readSystemIn then inputRequested.countDown()
+    super.onNotification(msg)
+end SilentTerminalSession
+
+object SilentTerminalSession:
+  private val silentMethods = Set(
+    Serialization.terminalSetEcho,
+    Serialization.terminalSetRawMode,
+    Serialization.getTerminalAttributes,
+    Serialization.setTerminalAttributes,
+    Serialization.terminalGetSize,
+    Serialization.terminalSetSize,
+  )
+  def connect(portfile: File): SilentTerminalSession =
+    val (socket, _) = ClientSocket.socket(portfile, false)
+    new SilentTerminalSession(socket)
+end SilentTerminalSession

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

@@ -0,0 +1,92 @@
+/*
+ * 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.io.{ InputStream, OutputStream, PrintStream }
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.atomic.AtomicReference
+import scala.concurrent.duration.*
+import sbt.internal.client.NetworkClient
+import sbt.internal.langserver.SbtExecParams
+import sbt.internal.langserver.codec.JsonProtocol.given
+import sbt.internal.util.Util
+import scala.collection.mutable
+
+/**
+ * The failed-load prompt must reach the interactive client that triggered the reload,
+ * and its answer must get back to the server: typing 'r' retries the load.
+ */
+class FailedLoadPromptTest extends AbstractServerTest {
+  override val testDirectory: String = "client"
+
+  private class CachingOutputStream extends OutputStream {
+    private val byteBuffer = new mutable.ArrayBuffer[Byte]
+    override def write(i: Int) = Util.ignoreResult(synchronized(byteBuffer += i.toByte))
+    def text: String = new String(synchronized(byteBuffer.toArray), "UTF-8")
+  }
+  private class CachingPrintStream(val cos: CachingOutputStream = new CachingOutputStream)
+      extends PrintStream(cos, true) {
+    def text: String = cos.text
+  }
+  private class QueueInputStream extends InputStream {
+    private val queue = new LinkedBlockingQueue[Integer]
+    def push(s: String): Unit = s.getBytes("UTF-8").foreach(b => queue.put(b.toInt))
+    override def read(): Int = queue.take()
+  }
+
+  private def awaitUntil(deadlineSeconds: Int)(condition: => Boolean): Boolean = {
+    val deadline = System.nanoTime + deadlineSeconds * 1000000000L
+    var met = condition
+    while (!met && System.nanoTime < deadline) {
+      Thread.sleep(500)
+      met = condition
+    }
+    met
+  }
+
+  test("an interactive client can answer the failed-load prompt") {
+    val buildFile = testPath.resolve("build.sbt")
+    val goodBuild = java.nio.file.Files.readString(buildFile)
+    val in = new QueueInputStream
+    val out = new CachingPrintStream
+    val err = new CachingPrintStream
+    val exitCode = new AtomicReference[Option[Int]](None)
+    val clientThread = new Thread("failed-load-prompt-test-client") {
+      setDaemon(true)
+      override def run(): Unit = {
+        val code = NetworkClient.client(testPath.toFile, Array.empty[String], in, out, err, false)
+        exitCode.set(Some(code))
+      }
+    }
+    clientThread.start()
+    assert(awaitUntil(30)(out.text.contains("sbt:")), s"client never attached: ${out.text}")
+
+    java.nio.file.Files.writeString(buildFile, "val = =\n")
+    in.push("reload\r")
+    // let the failed-load prompt start reading, then repair the build and answer 'r'
+    Thread.sleep(20000)
+    java.nio.file.Files.writeString(buildFile, goodBuild)
+    in.push("r")
+    // a successful retry returns the client to the command prompt; run a task to prove it.
+    // the server-side [success] is observed through the suite session's log notifications,
+    // since the in-process client does not render exec results on the provided stream.
+    // the prompt parks the command loop until answered: on develop (reads process stdin)
+    // any further exec starves; with the fix the client's 'r' resolves it and execs flow.
+    val id = svr.session.nextId()
+    svr.session.sendJsonRpc(id, "sbt/exec", SbtExecParams("show name")).get
+    val served = svr.session.waitForNotificationMsg(90.seconds)(_.method == "build/logMessage")
+    assert(
+      served.isSuccess,
+      s"server did not serve another client after the prompt: exit=${exitCode.get}\nout=${out.text}\nerr=${err.text}"
+    )
+
+    in.push("exit\r")
+    assert(awaitUntil(60)(exitCode.get.isDefined), s"client did not exit: ${out.text}")
+  }
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.langserver.SbtExecParams
+import sbt.internal.util.Util
+import sbt.protocol.{ Attach, Serialization, SilentTerminalSession }
+import sbt.protocol.codec.JsonProtocol.given
+import sbt.internal.langserver.codec.JsonProtocol.given
+
+/**
+ * Regression: a client that dies while the server waits on a terminal control answer
+ * must not strand the parked server-side thread.
+ */
+class TerminalMapsDrainTest extends AbstractServerTest {
+  override val testDirectory: String = "client"
+
+  test("a client dying at a failed-reload prompt does not strand the server") {
+    val portfile = new java.io.File(testPath.toFile, "project/target/active.json")
+    val buildFile = testPath.resolve("build.sbt")
+    val goodBuild = java.nio.file.Files.readString(buildFile)
+    val silent = SilentTerminalSession.connect(portfile)
+    var parkedOn = "none"
+    try {
+      silent.initialize(10.seconds, false).get
+      Util.ignoreResult(
+        silent.sendJsonRpc(silent.nextId(), Serialization.attach, Attach(interactive = true))
+      )
+      java.nio.file.Files.writeString(buildFile, "val = =\n")
+      silent.sendJsonRpc(silent.nextId(), "sbt/exec", SbtExecParams("reload")).get
+      val queried = silent.silentQuery.await(60, TimeUnit.SECONDS)
+      val input = silent.inputRequested.await(5, TimeUnit.SECONDS)
+      assert(queried || input, "server never queried the terminal nor requested input")
+      // the command loop is now parked waiting for an answer that never comes
+      parkedOn =
+        Option(silent.firstSilentQuery.get).getOrElse(if (input) "readSystemIn" else "unknown")
+    } finally {
+      silent.close()
+      java.nio.file.Files.writeString(buildFile, goodBuild)
+    }
+    // EOF at the failed-load prompt maps to 'q' by design: the server must shut down
+    // cleanly rather than stay parked on the dead client's unanswered query.
+    def serverAlive: Boolean =
+      ProcessHandle.current.descendants.anyMatch { ph =>
+        ph.info.command.orElse("").contains("java")
+      }
+    val deadline = 90.seconds.fromNow
+    while (serverAlive && deadline.hasTimeLeft()) Thread.sleep(500)
+    assert(!serverAlive, s"server must exit after the prompting client dies (parked on $parkedOn)")
+  }
+}