Serialize per-target compiles (CompileScheduler + zinc per-scope lock)

sls/src/org/scala/abusers/sls/CompileScheduler.scala

@@ -0,0 +1,44 @@
+package org.scala.abusers.sls
+
+import cats.effect.std.Queue
+import cats.effect.std.Supervisor
+import cats.effect.IO
+import cats.effect.Ref
+
+/** Per-key conflating scheduler.
+  *
+  * At most one task per key runs at a time. While a key's task is running, further `schedule` calls for that key are
+  * conflated into a single pending re-run that keeps only the *newest* task; intermediate tasks are dropped. When the
+  * running task finishes, the pending one (if any) runs next.
+  *
+  * Each key has a `circularBuffer(1)` mailbox drained by one long-lived worker fiber: offering the newest task evicts
+  * any older queued one, so the buffer always holds the latest save. (`Queue.dropping` would be wrong here — it keeps
+  * the *oldest* offer and discards new ones, so the most recent save would never compile.) A worker fiber per key stays
+  * parked on `take` for the lifetime of the supervisor; the keys are the project's build targets, so this is a small,
+  * bounded number of idle fibers rather than per-save churn.
+  *
+  * Used to collapse a burst of `didSave` compiles for one build target into "the in-flight compile + at most one more"
+  * instead of one full compile per save. The zinc-cli side still serializes same-scope compiles as a safety net; this
+  * just stops us from queueing redundant work in the first place.
+  */
+class CompileScheduler private (state: Ref[IO, Map[String, Queue[IO, IO[Unit]]]], supervisor: Supervisor[IO]) {
+
+  def schedule(key: String)(task: IO[Unit]): IO[Unit] =
+    Queue.circularBuffer[IO, IO[Unit]](capacity = 1).flatMap { fresh =>
+      state.modify { m =>
+        m.get(key) match {
+          case Some(existing) => (m, existing.offer(task))
+          case None           =>
+            (m.updated(key, fresh), fresh.offer(task) *> supervisor.supervise(runLoop(fresh)).void)
+        }
+      }.flatten
+    }
+
+  private def runLoop(mailbox: Queue[IO, IO[Unit]]): IO[Unit] =
+    mailbox.take.flatMap(_.attempt.void).foreverM
+}
+
+object CompileScheduler {
+  def create(supervisor: Supervisor[IO]): IO[CompileScheduler] =
+    Ref.of[IO, Map[String, Queue[IO, IO[Unit]]]](Map.empty).map(new CompileScheduler(_, supervisor))
+}

sls/src/org/scala/abusers/sls/ServerImpl.scala

@@ -4,7 +4,6 @@ package sls
 import bsp.InitializeBuildParams
 import cats.effect.kernel.Deferred
 import cats.effect.kernel.Resource
-import cats.effect.std.Supervisor
 import cats.effect.IO
 import cats.syntax.all.*
 import fs2.io.file.Files
@@ -53,7 +52,7 @@ class ServerImpl(
     indexManager: index.IndexManager,
     symbolIndex: index.SymbolIndex,
     indexLifecycle: index.IndexLifecycle,
-    indexSupervisor: Supervisor[IO],
+    compileScheduler: CompileScheduler,
 )(using Tracer[IO], Meter[IO])
     extends SlsLanguageServer[IO] {
 
@@ -280,21 +279,21 @@ class ServerImpl(
     for {
       _    <- textDocumentSyncManager.didSave(params)
       info <- bspStateManager.get(uri)
-      _    <- indexSupervisor
-        .supervise(
-          bspStateManager
-            .compileWithCSP(uri)
-            .flatMap { res =>
-              updateOutputClasspath(info, AbsolutePath(res.outputJar)) *>
-                indexManager
-                  .onCompilationComplete(info, res)
-                  .handleError(e =>
-                    lspClient.logMessage(s"Failed to update index after compilation: ${e.getMessage()}")
-                  )
-            }
-            .timeout(60.seconds)
-        )
-        .void
+      // Conflate rapid saves of the same target: the in-flight compile finishes, only the newest pending save
+      // survives, intermediate ones are dropped. Keyed by scopeId — same as compileWithCSP / the zinc-cli lock.
+      _ <- compileScheduler.schedule(info.buildTarget.displayName.getOrElse("default")) {
+        bspStateManager
+          .compileWithCSP(uri)
+          .flatMap { res =>
+            updateOutputClasspath(info, AbsolutePath(res.outputJar)) *>
+              indexManager
+                .onCompilationComplete(info, res)
+                .handleErrorWith(e =>
+                  lspClient.logMessage(s"Failed to update index after compilation: ${e.getMessage()}")
+                )
+          }
+          .timeout(60.seconds)
+      }
     } yield ()
   }
 

sls/src/org/scala/abusers/sls/SimpleLanguageServer.scala

@@ -87,6 +87,7 @@ object SimpleScalaServer extends ProfilingIOApp {
       symbolIndex       <- index.SymbolIndex.empty.toResource
       indexLifecycle    <- index.IndexLifecycle.empty.toResource
       indexSupervisor   <- Supervisor[IO]
+      compileScheduler  <- CompileScheduler.create(indexSupervisor).toResource
       bytecodeIndexer = index.BytecodeIndexer()
       depIndexCache   = index.DepIndexCache.default
       coursierCache   = coursierapi.Cache.create()
@@ -114,6 +115,6 @@ object SimpleScalaServer extends ProfilingIOApp {
       indexManager,
       symbolIndex,
       indexLifecycle,
-      indexSupervisor,
+      compileScheduler,
     )
 }

sls/test/src/org/scala/abusers/sls/util/DidSaveCompileConcurrencySpec.scala

@@ -0,0 +1,89 @@
+package org.scala.abusers.sls.util
+
+import cats.effect.kernel.Resource
+import cats.effect.IO
+import cats.effect.Ref
+import org.scala.abusers.csp
+import org.scala.abusers.sls.index.util.Code.*
+import org.scala.abusers.sls.index.util.TestWorkspace
+import org.typelevel.otel4s.metrics.Meter
+import org.typelevel.otel4s.trace.Tracer
+import weaver.*
+
+import scala.concurrent.duration.*
+
+import TestClient.awaitUntil
+
+/** Regression guard for the rapid edit-save-edit-save bug: two `didSave`s for the same target used to fork two compile
+  * fibers that raced into the (shared, per-scope) zinc analysis store and output jar at the same time. The per-target
+  * conflating [[org.scala.abusers.sls.CompileScheduler]] must keep them from overlapping — the in-flight compile
+  * finishes before the next one starts.
+  *
+  * The CSP double here records peak concurrency instead of compiling anything: each `compile` widens its window with a
+  * sleep so an overlap would be observed if one existed.
+  */
+object DidSaveCompileConcurrencySpec extends SimpleIOSuite {
+
+  given Tracer[IO] = Tracer.noop[IO]
+  given Meter[IO]  = Meter.noop[IO]
+
+  /** A CSP double that only measures overlap. `started` lets a test await N compiles; `maxConcurrent` is the peak
+    * number seen running at once — it must stay 1 if compiles are properly serialized per scope.
+    */
+  final class ConcurrencyProbeCsp(
+      inFlight: Ref[IO, Int],
+      maxConcurrent: Ref[IO, Int],
+      started: Ref[IO, Int],
+      window: FiniteDuration,
+  ) extends csp.CspServer[IO] {
+    def compile(
+        scopeId: String,
+        classpath: List[String],
+        sourcePath: List[String],
+        scalacOptions: List[String],
+        javacOptions: List[String],
+        scalaVersion: csp.ScalaVersion,
+    ): IO[csp.CompileOutput] =
+      Resource
+        .make(
+          started.update(_ + 1) *>
+            inFlight.updateAndGet(_ + 1).flatMap(n => maxConcurrent.update(_ max n))
+        )(_ => inFlight.update(_ - 1))
+        .surround(IO.sleep(window))
+        // Nonexistent jar: updateOutputClasspath skips (no such file) and onCompilationComplete swallows the
+        // indexing error — this probe is only about the compile boundary, not downstream indexing.
+        .as(csp.CompileOutput("/nonexistent/sls-probe.jar", changedFiles = Map.empty, csp.OutputFormat.TASTY))
+  }
+
+  test("rapid successive saves of the same target never overlap compiles") {
+    val src = code"""
+      |package example
+      |
+      |class Foo {
+      |${m1}}
+    """
+    val program = for {
+      inFlight      <- Ref.of[IO, Int](0)
+      maxConcurrent <- Ref.of[IO, Int](0)
+      started       <- Ref.of[IO, Int](0)
+      probe = new ConcurrencyProbeCsp(inFlight, maxConcurrent, started, window = 300.millis)
+      result <- TestWorkspace
+        .withSources("Foo.scala" -> src.text)
+        .flatMap(ws => TestServer.resource(ws, cspServer = Some(probe)))
+        .use { h =>
+          val uri = h.workspace.uri("Foo.scala")
+          for {
+            _ <- h.client.openFile(uri, src.text)
+            // Two edits saved back-to-back: the second save lands while the first compile is still in its window.
+            _    <- h.client.insertAt(uri, src.at(m1), "  def a: Int = 1\n")
+            _    <- h.client.saveFile(uri)
+            _    <- h.client.insertAt(uri, src.at(m1), "  def b: Int = 2\n")
+            _    <- h.client.saveFile(uri)
+            _    <- awaitUntil(started.get.map(n => Option.when(n >= 2)(n)), label = "both saves to reach the CSP")
+            peak <- maxConcurrent.get
+          } yield expect(peak == 1)
+        }
+    } yield result
+    program
+  }
+}

sls/test/src/org/scala/abusers/sls/util/TestClientSpec.scala

@@ -22,12 +22,12 @@ object TestClientSpec extends SimpleIOSuite {
       |  def bar(x: Int): Int = x + 1
       |${m1}}
     """
-    TestWorkspace.withSources("Foo.scala" -> src.text).flatMap(TestServer.resource).use { h =>
+    TestWorkspace.withSources("Foo.scala" -> src.text).flatMap(TestServer.resource(_)).use { h =>
       val uri = h.workspace.uri("Foo.scala")
       for {
-        _ <- h.client.openFile(uri, src.text)
-        _ <- h.client.insertAt(uri, src.at(m1), "  def baz: String = \"b\"\n")
-        _ <- h.client.saveFile(uri)
+        _          <- h.client.openFile(uri, src.text)
+        _          <- h.client.insertAt(uri, src.at(m1), "  def baz: String = \"b\"\n")
+        _          <- h.client.saveFile(uri)
         bazSymbols <- awaitUntil(
           h.symbolIndex.getSymbolsByName("baz").map(s => Option.when(s.nonEmpty)(s)),
           label = "baz to appear in the index after save",
@@ -46,7 +46,7 @@ object TestClientSpec extends SimpleIOSuite {
       |  def bar(x: Int): Int = x + 1
       |${m1}}
     """
-    TestWorkspace.withSources("Foo.scala" -> src.text).flatMap(TestServer.resource).use { h =>
+    TestWorkspace.withSources("Foo.scala" -> src.text).flatMap(TestServer.resource(_)).use { h =>
       val uri = h.workspace.uri("Foo.scala")
       for {
         _ <- h.client.openFile(uri, src.text)
@@ -57,9 +57,9 @@ object TestClientSpec extends SimpleIOSuite {
           label = "save-driven index update",
         )
         // empty-prefix search returns every indexed symbol; the project tier is this one file
-        sessionIds  <- h.symbolIndex.searchSymbols("").map(_.map(_.id).toSet)
+        sessionIds   <- h.symbolIndex.searchSymbols("").map(_.map(_.id).toSet)
         finalContent <- h.client.bufferOf(uri)
-        freshIds <- TestWorkspace
+        freshIds     <- TestWorkspace
           .withSources("Foo.scala" -> finalContent)
           .use(_.symbolIndex.flatMap(_.searchSymbols("").map(_.map(_.id).toSet)))
       } yield expect.same(freshIds, sessionIds)
@@ -74,14 +74,14 @@ object TestClientSpec extends SimpleIOSuite {
       |  val ${m1}greeting${m2} = "hi"
       |}
     """
-    TestWorkspace.withSources("App.scala" -> src.text).flatMap(TestServer.resource).use { h =>
+    TestWorkspace.withSources("App.scala" -> src.text).flatMap(TestServer.resource(_)).use { h =>
       val uri = h.workspace.uri("App.scala")
       for {
-        _       <- h.client.openFile(uri, src.text)
-        _       <- h.client.editFile(uri, lsp.Range(src.at(m1), src.at(m2)), "farewell")
-        buffer  <- h.client.bufferOf(uri)
-        _       <- h.client.saveFile(uri)
-        onDisk  <- IO.blocking(os.read(h.workspace.sources("App.scala")))
+        _      <- h.client.openFile(uri, src.text)
+        _      <- h.client.editFile(uri, lsp.Range(src.at(m1), src.at(m2)), "farewell")
+        buffer <- h.client.bufferOf(uri)
+        _      <- h.client.saveFile(uri)
+        onDisk <- IO.blocking(os.read(h.workspace.sources("App.scala")))
       } yield expect(buffer.contains("val farewell = \"hi\"")) &&
         expect.same(buffer, onDisk)
     }

sls/test/src/org/scala/abusers/sls/util/TestServer.scala

@@ -19,11 +19,11 @@ import org.typelevel.otel4s.trace.Tracer
   *   - BSP → [[BspStubs.forTarget]] (answers inverse-sources for one synthetic Scala target; everything else raises)
   *   - CSP → [[FakeCspServer.dotcBacked]] (real in-process dotc producing real TASTy jars, no zinc)
   *   - queue → [[TestComputationQueue]] (passthrough; the `./.sls/` classpath swap is skipped)
-  *   - index lifecycle is set straight to `Ready`: `bootstrap` would index the JDK and dependency jars, which is
-  *     out of scope until the harness grows a knob for it
+  *   - index lifecycle is set straight to `Ready`: `bootstrap` would index the JDK and dependency jars, which is out of
+  *     scope until the harness grows a knob for it
   *
-  * The synthetic target reports Scala 3.0.0, which keeps `ServerImpl`'s PC-diagnostics path (gated on > 3.7.2)
-  * disabled — no presentation compiler is downloaded or loaded.
+  * The synthetic target reports Scala 3.0.0, which keeps `ServerImpl`'s PC-diagnostics path (gated on > 3.7.2) disabled
+  * — no presentation compiler is downloaded or loaded.
   */
 object TestServer {
 
@@ -36,10 +36,13 @@ object TestServer {
       lifecycle: index.IndexLifecycle,
   )
 
-  def resource(ws: TestWorkspace)(using Tracer[IO], Meter[IO]): Resource[IO, Handle] =
+  def resource(ws: TestWorkspace, cspServer: Option[CspServer[IO]] = None)(using
+      Tracer[IO],
+      Meter[IO],
+  ): Resource[IO, Handle] =
     for {
-      client            <- TestClient.create.toResource
-      steward           <- ResourceSupervisor[IO]
+      client  <- TestClient.create.toResource
+      steward <- ResourceSupervisor[IO]
       // The synthetic target's Scala version keeps every PC path off; a real provider would also drag in
       // scache, whose background machinery misbehaves under mill's test-runner classloader.
       pcProvider = new PresentationCompilerProvider {
@@ -53,18 +56,17 @@ object TestServer {
       diagnosticManager <- DiagnosticManager.instance.toResource
       cancelTokens      <- IOCancelTokens.instance
       indexSupervisor   <- Supervisor[IO]
+      compileScheduler  <- CompileScheduler.create(indexSupervisor).toResource
       symbolIndex       <- index.SymbolIndex.empty.toResource
       indexLifecycle    <- index.IndexLifecycle.empty.toResource
-      cspWorkDir        <- Resource.make(IO.blocking(os.temp.dir(prefix = "sls-test-csp")))(d =>
-        IO.blocking(os.remove.all(d))
-      )
-      cacheDir          <- Resource.make(IO.blocking(os.temp.dir(prefix = "sls-test-dep-cache")))(d =>
+      cspWorkDir <- Resource.make(IO.blocking(os.temp.dir(prefix = "sls-test-csp")))(d => IO.blocking(os.remove.all(d)))
+      cacheDir   <- Resource.make(IO.blocking(os.temp.dir(prefix = "sls-test-dep-cache")))(d =>
         IO.blocking(os.remove.all(d))
       )
 
       targetInfo = syntheticTarget(ws)
       bspStub    = BspStubs.forTarget(targetInfo.buildTarget.id)
-      fakeCsp    = FakeCspServer.dotcBacked(cspWorkDir)
+      fakeCsp    = cspServer.getOrElse(FakeCspServer.dotcBacked(cspWorkDir))
 
       bspDeferred      <- Deferred[IO, BuildServer].toResource
       cspDeferred      <- Deferred[IO, CspServer[IO]].toResource
@@ -97,7 +99,7 @@ object TestServer {
         indexManager,
         symbolIndex,
         indexLifecycle,
-        indexSupervisor,
+        compileScheduler,
       )
       _ <- client.completeServer(server).toResource
     } yield Handle(client, server, ws, symbolIndex, indexManager, indexLifecycle)