From c407f377398321c8a51c878b67f98b7619ba3530 Mon Sep 17 00:00:00 2001 From: BrianHotopp Date: Sun, 2 Aug 2026 02:35:39 -0400 Subject: [PATCH] [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) --- .../scala/sbt/internal/util/Terminal.scala | 1 + .../util/WriteableInputStreamSpec.scala | 28 ++++++ main/src/main/scala/sbt/Main.scala | 2 +- .../sbt/internal/server/VirtualTerminal.scala | 33 ++++--- .../internal/server/VirtualTerminalSpec.scala | 39 +++++++- notes/2.0.0/terminal-client-disconnect.md | 10 ++ .../sbt/protocol/ServerSessionImpl.scala | 6 +- .../sbt/protocol/ServerSessionImplSpec.scala | 36 ++++++++ .../sbt/protocol/SilentTerminalSession.scala | 54 +++++++++++ .../scala/testpkg/FailedLoadPromptTest.scala | 92 +++++++++++++++++++ .../scala/testpkg/TerminalMapsDrainTest.scala | 59 ++++++++++++ 11 files changed, 343 insertions(+), 17 deletions(-) create mode 100644 internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala create mode 100644 notes/2.0.0/terminal-client-disconnect.md create mode 100644 protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala create mode 100644 server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala create mode 100644 server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala create mode 100644 server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala diff --git a/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala b/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala index 0fe75095c..bcbb63698 100644 --- a/internal/util-logging/src/main/scala/sbt/internal/util/Terminal.scala +++ b/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)) () } } diff --git a/internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala b/internal/util-logging/src/test/scala/sbt/internal/util/WriteableInputStreamSpec.scala new file mode 100644 index 000000000..79cbb59a8 --- /dev/null +++ b/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 diff --git a/main/src/main/scala/sbt/Main.scala b/main/src/main/scala/sbt/Main.scala index 741fd040a..8ded3c871 100644 --- a/main/src/main/scala/sbt/Main.scala +++ b/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 } diff --git a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala b/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala index d8eb17b4b..9cc12a910 100644 --- a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala +++ 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 } diff --git a/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala b/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala index ddd8db129..440f3cc89 100644 --- a/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala +++ b/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 diff --git a/notes/2.0.0/terminal-client-disconnect.md b/notes/2.0.0/terminal-client-disconnect.md new file mode 100644 index 000000000..d83cb99a2 --- /dev/null +++ b/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 diff --git a/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala b/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala index 10d2c96c0..bd20bd0e1 100644 --- a/protocol/src/main/scala/sbt/protocol/ServerSessionImpl.scala +++ b/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) diff --git a/protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala b/protocol/src/test/scala/sbt/protocol/ServerSessionImplSpec.scala new file mode 100644 index 000000000..9f0acbc47 --- /dev/null +++ b/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 diff --git a/server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala b/server-test/src/test/scala/sbt/protocol/SilentTerminalSession.scala new file mode 100644 index 000000000..564abf304 --- /dev/null +++ b/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 diff --git a/server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala b/server-test/src/test/scala/testpkg/FailedLoadPromptTest.scala new file mode 100644 index 000000000..02efc40b1 --- /dev/null +++ b/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}") + } +} diff --git a/server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala b/server-test/src/test/scala/testpkg/TerminalMapsDrainTest.scala new file mode 100644 index 000000000..93b4ba9ab --- /dev/null +++ b/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)") + } +}