bsp: a task finishing between two state reads could be admitted twice

.scalafmt.conf

@@ -1,4 +1,4 @@
-version=3.11.1
+version=3.11.5
 maxColumn = 160
 rewrite.rules = [SortImports, RedundantBraces, RedundantParens, PreferCurlyFors]
 project.excludePaths = [

bleep-bsp-tests/src/scala/bleep/bsp/ServerRunInterruptTest.scala

@@ -22,11 +22,12 @@ class ServerRunInterruptTest extends AnyFunSuite with Matchers {
     val parked = new CountDownLatch(1)
     val thrown = new AtomicReference[Option[Throwable]](None)
 
-    val t = new Thread(
-      () =>
-        try body(parked)
-        catch { case e: Throwable => thrown.set(Some(e)) }, "interrupt-under-test"
-    )
+    // scalafmt's parser (any version/dialect as of 3.11.5) cannot parse a lambda whose body is an
+    // indented try/catch followed by `, arg` in the same call — bind the Runnable first.
+    val runnable: Runnable = () =>
+      try body(parked)
+      catch { case e: Throwable => thrown.set(Some(e)) }
+    val t = new Thread(runnable, "interrupt-under-test")
     t.setDaemon(true)
     t.start()
 

bleep-bsp/src/scala/bleep/bsp/TaskDag.scala

@@ -1173,8 +1173,14 @@ object TaskDag {
           supervisor: cats.effect.std.Supervisor[IO]
       ): IO[Unit] =
         for {
-          dag <- dagRef.get
+          // Read `running` BEFORE `dag`. A completing task writes in the opposite order — finished
+          // into dagRef (end of executeTask), then removed from runningRef (its guarantee) — so a
+          // task that finishes between the two reads is visible in at least one snapshot. Read the
+          // other way around, it is visible in neither: not finished in the stale dag, not running
+          // in the fresh set — so `ready.filterNot(running)` admits it a second time and the task
+          // runs twice. Caught by LinkDagIntegrationTest emitting two LinkStarted events 1ms apart.
           running <- runningRef.get
+          dag <- dagRef.get
           maybeKilled <- isKilled
           _ <-
             if (dag.isComplete) {
@@ -1230,8 +1236,10 @@ object TaskDag {
                 }
                 // Re-read state. If nothing is running, the DAG is either complete, in a
                 // transient gap (skips just opened up new ready tasks), or genuinely stuck.
-                newDag <- dagRef.get
+                // Same read order as the loop top: running before dag, so a task finishing
+                // between the reads is seen by at least one of them.
                 newRunning <- runningRef.get
+                newDag <- dagRef.get
                 _ <-
                   if (newDag.isComplete) IO.unit
                   else if (newRunning.isEmpty && newDag.ready.isEmpty && newDag.toSkip.isEmpty) {

bleep-cli/src/scala/bleep/commands/CompileServerStopAll.scala

@@ -7,6 +7,7 @@ import cats.effect.unsafe.implicits.global
 import ryddig.Logger
 
 import java.nio.file.{Files, Path}
+import scala.jdk.CollectionConverters.IteratorHasAsScala
 import scala.jdk.StreamConverters.StreamHasToScala
 
 case class CompileServerStopAll(logger: Logger, userPaths: UserPaths) extends BleepCommand {
@@ -29,6 +30,7 @@ case class CompileServerStopAll(logger: Logger, userPaths: UserPaths) extends Bl
 
       // Give the OS time to release file handles after process kill
       Thread.sleep(200)
+      val sizeMb = dirSizeBytes(socketDir) / (1024L * 1024L)
       try FileUtils.deleteDirectory(socketDir)
       catch {
         case _: java.nio.file.DirectoryNotEmptyException | _: java.nio.file.FileSystemException =>
@@ -37,7 +39,18 @@ case class CompileServerStopAll(logger: Logger, userPaths: UserPaths) extends Bl
           Thread.sleep(2000)
           FileUtils.deleteDirectory(socketDir)
       }
+      // These dirs can hold multi-GB diagnostic artifacts (heap dumps from pre-M11 servers, output
+      // logs). Deleting gigabytes without a word once cost a user the evidence for an OOM report —
+      // say what went away, and stand out when it was big.
+      val msg = s"deleted $socketDir (${sizeMb}MB)"
+      if (sizeMb >= 1024) logger.warn(msg) else logger.info(msg)
     }
     Right(())
   }
+
+  private def dirSizeBytes(dir: Path): Long = {
+    val stream = Files.walk(dir)
+    try stream.iterator().asScala.filter(Files.isRegularFile(_)).map(Files.size).sum
+    finally stream.close()
+  }
 }

bleep-core/src/scala/bleep/commands/ReactiveBsp.scala

@@ -457,7 +457,7 @@ case class ReactiveBsp(
       _ <- display.printSummary(filterContext)
       // Update previousRunState from collected events (only in DiffWatch mode)
       _ <- if (isDiffWatch) IO(previousRunState.set(PreviousRunState.fromEvents(collectedBuildEvents.get().reverse))) else IO.unit
-      _ <- IO.delay(started.logger.info(s"  BSP server log: ${BspRifle.getOutputFile(config)}"))
+      _ <- IO.delay(started.logger.info(serverLogLine(config)))
       _ <-
         if (flamegraph)
           IO.delay(started.logger.info(s"  Flamegraph: ${started.buildPaths.dotBleepDir.resolve("trace.json")} (open in chrome://tracing or ui.perfetto.dev)"))
@@ -498,12 +498,21 @@ case class ReactiveBsp(
               case None      => printCancelledSummary(started, durationMs)
             }
         }
-        started.logger.info(s"  BSP server log: ${BspRifle.getOutputFile(config)}")
+        started.logger.info(serverLogLine(config))
         if (flamegraph)
           started.logger.info(s"  Flamegraph: ${started.buildPaths.dotBleepDir.resolve("trace.json")} (open in chrome://tracing or ui.perfetto.dev)")
       }
   }
 
+  /** The failure path can point at a log that is not there — the server never started, or the socket dir was deleted (`compile-server stop-all`, per-invocation
+    * shutdown hook). Saying so beats sending the reader to a file that does not exist.
+    */
+  private def serverLogLine(config: BspRifleConfig): String = {
+    val outputFile = BspRifle.getOutputFile(config)
+    if (java.nio.file.Files.exists(outputFile)) s"  BSP server log: $outputFile"
+    else s"  BSP server log: $outputFile (missing — the server never started, or its directory was deleted)"
+  }
+
   private def printErrorSummary(started: Started, err: Throwable, durationMs: Long): Unit = {
     import bleep.testing.BleepConsole as C
     val logger = started.logger