diff --git a/main/src/main/scala/sbt/internal/server/NetworkChannel.scala b/main/src/main/scala/sbt/internal/server/NetworkChannel.scala index 2b3c2ae96..5e57878a5 100644 --- a/main/src/main/scala/sbt/internal/server/NetworkChannel.scala +++ b/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 { - case null => - case t => properties.set(t) - } - pending.synchronized { - lastUpdate.set(Deadline.now) - pending.set(false) - pending.notifyAll() + try { + queue.poll(5, java.util.concurrent.TimeUnit.SECONDS) match { + case null => + VirtualTerminal.expireTerminalPropertiesQuery(term.name, queue) match { + case Some(late) => properties.set(late) + case None => Util.ignoreResult(properties.compareAndSet(null, empty)) + } + 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 diff --git a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala b/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala index 21efe9d7d..d8eb17b4b 100644 --- a/main/src/main/scala/sbt/internal/server/VirtualTerminal.scala +++ b/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 = diff --git a/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala b/main/src/test/scala/sbt/internal/server/VirtualTerminalSpec.scala new file mode 100644 index 000000000..ddd8db129 --- /dev/null +++ b/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 diff --git a/notes/2.0.0/terminal-properties-freeze.md b/notes/2.0.0/terminal-properties-freeze.md new file mode 100644 index 000000000..c42180a86 --- /dev/null +++ b/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. diff --git a/server-test/src/test/scala/sbt/protocol/FaultyTerminalSession.scala b/server-test/src/test/scala/sbt/protocol/FaultyTerminalSession.scala new file mode 100644 index 000000000..69e64d79c --- /dev/null +++ b/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 diff --git a/server-test/src/test/scala/testpkg/AbstractServerTest.scala b/server-test/src/test/scala/testpkg/AbstractServerTest.scala index 11f438db7..0ea4ad509 100644 --- a/server-test/src/test/scala/testpkg/AbstractServerTest.scala +++ b/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 diff --git a/server-test/src/test/scala/testpkg/RebootTest.scala b/server-test/src/test/scala/testpkg/RebootTest.scala index 23a7c60db..47f55c406 100644 --- a/server-test/src/test/scala/testpkg/RebootTest.scala +++ b/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(client("willSucceed") == 0, "the rebooted server must serve a new client connection") + assert(runBatchClient("reboot") == 0, "reboot from a client must complete with exit 0") + assert( + runBatchClient("willSucceed") == 0, + "the rebooted server must serve a new client connection" + ) } } diff --git a/server-test/src/test/scala/testpkg/TerminalPropertiesFreezeTest.scala b/server-test/src/test/scala/testpkg/TerminalPropertiesFreezeTest.scala new file mode 100644 index 000000000..2093b0815 --- /dev/null +++ b/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() + } +}