[2.0.x] fix: Keep accepting after a client fails

**Problem**
The accept loop did not catch an exception from onIncomingSocket, so
its thread ended. Nothing closed the server socket, so the path stayed
bound: the server took no further client, and the process kept running.
The socket of the client that failed stayed open too.

**Solution**
Log that client, close its socket, and take the next one. An exception
from accept itself keeps the handling it had, so a socket that is
really broken still ends the loop.

The callback now takes a holder rather than the socket, and calls
AtomicCloseable.release to keep it. The loop closes whatever the
callback left behind.

Co-authored-by: Claude Opus 5 (1M context) <[email protected]>

internal/util-core/src/main/scala/sbt/internal/util/Util.scala

@@ -13,7 +13,7 @@ import java.util.Locale
 
 import scala.collection.concurrent.TrieMap
 import scala.reflect.Selectable.reflectiveSelectable
-import scala.util.Properties
+import scala.util.{ Properties, Try }
 
 object Util:
   def makeList[T](size: Int, value: T): List[T] = List.fill(size)(value)
@@ -50,6 +50,8 @@ object Util:
     ()
   }
 
+  def ignoreTry[A](f: => A): Unit = ignoreResult(Try(f))
+
   lazy val isMac: Boolean =
     System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("mac")
 

main-command/src/main/scala/sbt/internal/AtomicCloseable.scala

@@ -0,0 +1,47 @@
+/*
+ * 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
+package internal
+
+import java.util.concurrent.atomic.AtomicReference
+
+import sbt.internal.util.Util
+
+/** Holds a closeable that a caller replaces, and closes the one it replaces. */
+private[sbt] class AtomicCloseable[A >: Null <: AutoCloseable](val ref: AtomicReference[A])
+    extends AnyVal:
+  def get: A = ref.get
+  def set(c: A): Unit = AtomicCloseable.close(ref.getAndSet(c))
+  def close(): Unit = AtomicCloseable.close(AtomicCloseable.release(ref))
+
+  /** Keeps the value another caller put here, and closes the one this caller built. */
+  def setIfEmpty(ctor: => A): A =
+    var obj = ref.get
+    if obj eq null then
+      val value = ctor
+      require(value ne null, "AtomicCloseable.setIfEmpty: `ctor` must not return null")
+      while obj eq null do obj = if ref.compareAndSet(null, value) then value else ref.get
+      if obj ne value then AtomicCloseable.close(value)
+    obj
+end AtomicCloseable
+
+private[sbt] object AtomicCloseable:
+  def apply[A >: Null <: AutoCloseable](): AtomicCloseable[A] =
+    new AtomicCloseable(new AtomicReference[A])
+
+  def apply[A >: Null <: AutoCloseable](c: A): AtomicCloseable[A] =
+    new AtomicCloseable(new AtomicReference[A](c))
+
+  def close(obj: AutoCloseable): Unit =
+    if obj ne null then Util.ignoreTry(obj.close())
+
+  def release[A >: Null <: AutoCloseable](ref: AtomicReference[A]): A =
+    ref.getAndSet(null)
+
+end AtomicCloseable

main-command/src/main/scala/sbt/internal/server/Server.scala

@@ -46,7 +46,7 @@ private[sbt] object Server {
 
   def start(
       connection: ServerConnection,
-      onIncomingSocket: (Socket, ServerInstance) => Unit,
+      onIncomingSocket: (AtomicReference[Socket], ServerInstance) => Unit,
       log: Logger
   ): ServerInstance =
     new ServerInstance { self =>
@@ -109,14 +109,19 @@ private[sbt] object Server {
               running.set(true)
               p.success(())
               while (running.get()) {
+                val clientSocket = AtomicCloseable[Socket]()
                 try {
-                  val socket = serverSocket.accept()
-                  onIncomingSocket(socket, self)
+                  clientSocket.set(serverSocket.accept())
+                  onIncomingSocket(clientSocket.ref, self)
                 } catch {
+                  case scala.util.control.NonFatal(e) if clientSocket.get ne null =>
+                    log.error(s"sbt server failed to serve a client: $e")
+                    log.trace(e)
                   case e: IOException if Option(e.getMessage).exists(_.contains("connect")) =>
                   case _: SocketTimeoutException          => // its ok
                   case _: SocketException if !running.get => // the server is shutting down
                 }
+                clientSocket.close()
               }
               serverSocketHolder.get match {
                 case null =>

main-command/src/test/scala/sbt/internal/AtomicCloseableSpec.scala

@@ -0,0 +1,72 @@
+/*
+ * 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
+package internal
+
+import verify.BasicTestSuite
+
+object AtomicCloseableSpec extends BasicTestSuite:
+  class Probe extends AutoCloseable:
+    var closed: Boolean = false
+    override def close(): Unit = closed = true
+  end Probe
+
+  test("a holder built around a value"):
+    val probe = new Probe
+    val holder = AtomicCloseable(probe)
+    assert(holder.get == probe)
+    assert(!probe.closed)
+    holder.close()
+    assert(probe.closed)
+
+  test("a value that replaces another closes it"):
+    val holder = AtomicCloseable[Probe]()
+    val first, second = new Probe
+    holder.set(first)
+    holder.set(second)
+    assert(first.closed)
+    assert(!second.closed)
+    assert(holder.get == second)
+
+  test("closing empties the holder"):
+    val holder = AtomicCloseable[Probe]()
+    val probe = new Probe
+    holder.set(probe)
+    holder.close()
+    assert(probe.closed)
+    assert(holder.get == null)
+
+  test("closing an empty holder does nothing"):
+    AtomicCloseable[Probe]().close()
+
+  test("a holder that has a value keeps it"):
+    val holder = AtomicCloseable[Probe]()
+    val first = new Probe
+    holder.set(first)
+    val second = new Probe
+    assert(holder.setIfEmpty(second) == first)
+    assert(!first.closed)
+    assert(!second.closed)
+
+  test("a holder that has no value takes the one it is given"):
+    val holder = AtomicCloseable[Probe]()
+    val probe = new Probe
+    assert(holder.setIfEmpty(probe) == probe)
+    assert(holder.get == probe)
+    assert(!probe.closed)
+
+  test("a value that loses a race is closed"):
+    val holder = AtomicCloseable[Probe]()
+    val winner, loser = new Probe
+    // the holder fills while this caller builds its own value
+    val result = holder.setIfEmpty { holder.set(winner); loser }
+    assert(result == winner)
+    assert(loser.closed)
+    assert(!winner.closed)
+end AtomicCloseableSpec

main/src/main/scala/sbt/internal/CommandExchange.scala

@@ -205,19 +205,20 @@ private[sbt] final class CommandExchange {
     lazy val enableBsp = s.get(bspEnabled).getOrElse(true)
     lazy val portfile = s.baseDir / "project" / "target" / "active.json"
 
-    def onIncomingSocket(socket: Socket, instance: ServerInstance): Unit = {
+    def onIncomingSocket(socket: AtomicReference[Socket], instance: ServerInstance): Unit = {
       val name = newNetworkName
       Terminal.consoleLog(s"new client connected: $name")
       val channel =
         new NetworkChannel(
           name,
-          socket,
+          socket.get,
           auth,
           instance,
           handlers,
           mkAskUser(name),
         )
       subscribe(channel)
+      AtomicCloseable.release(socket) // i took over
     }
     if (server.isEmpty && firstInstance.get) {
       val h = Hash.halfHashString(IO.toURI(portfile).toString)