Merge pull request #9502 from eed3si9n/wip/grpc

[2.x] fix: Intern GrpcActionCacheStore
2026-07-26 01:01:27 -04:00 · 2026-07-26 01:01:27 -04:00 · 9f0a459f81
parent b81265bdb5 b98c443b5c
commit 9f0a459f81
10 changed files with 173 additions and 19 deletions
--- a/.github/workflows/server-test.yml
+++ b/.github/workflows/server-test.yml
@ -14,14 +14,62 @@ jobs:
      JVM_OPTS: -Xms800M -Xmx2G -Xss6M -XX:ReservedCodeCacheSize=128M -server -Dsbt.io.virtual=false -Dfile.encoding=UTF-8
      SBT_ETC_FILE: $HOME/etc/sbt/sbtopts
    steps:
-    - uses: actions/checkout@v7
-    - name: Setup JDK
-      uses: actions/setup-java@v5
-      with:
-        distribution: "zulu"
-        java-version: "17"
-        cache: sbt
-    - uses: sbt/setup-sbt@v1
-    - name: Server test
-      shell: bash
-      run: sbt -v --client "serverTestProj/test"
+      - uses: actions/checkout@v7
+      - name: Setup JDK
+        uses: actions/setup-java@v5
+        with:
+          distribution: "zulu"
+          java-version: "17"
+          cache: sbt
+      - uses: sbt/setup-sbt@v1
+        with:
+          disk-cache: false
+      - name: Set bazel-remote version
+        id: bazel-remote-version
+        run: echo "version=2.6.2" >> "$GITHUB_OUTPUT"
+      - name: Cache bazel-remote
+        id: cache-bazel-remote
+        uses: actions/cache@v4
+        with:
+          path: $RUNNER_TOOL_CACHE/local/bazel-remote
+          key: bazel-remote-v${{ steps.bazel-remote-version.outputs.version }}-${{ runner.os }}-${{ runner.arch }}
+      - name: Download bazel-remote
+        if: steps.cache-bazel-remote.outputs.cache-hit != 'true'
+        shell: bash
+        env:
+          BAZEL_REMOTE_VERSION: ${{ steps.bazel-remote-version.outputs.version }}
+        run: |
+          mkdir -p "$RUNNER_TOOL_CACHE/local"
+          curl -sL -o "$RUNNER_TOOL_CACHE/local/bazel-remote" "https://github.com/buchgr/bazel-remote/releases/download/v${BAZEL_REMOTE_VERSION}/bazel-remote-${BAZEL_REMOTE_VERSION}-linux-amd64"
+          chmod +x "$RUNNER_TOOL_CACHE/local/bazel-remote"
+      - name: Add bazel-remote to PATH
+        run: echo "$RUNNER_TOOL_CACHE/local" >> "$GITHUB_PATH"
+      - name: Start bazel-remote
+        shell: bash
+        run: |
+          mkdir -p "$HOME/bazel-remote/temp"
+          nohup "$RUNNER_TOOL_CACHE/local/bazel-remote" --max_size 5 --dir "$HOME/bazel-remote/temp" \
+            --http_address localhost:8000 --grpc_address localhost:2024 \
+            > /tmp/bazel-remote.log 2>&1 &
+          echo $! > /tmp/bazel-remote.pid
+          for i in $(seq 1 30); do
+            curl -sf http://localhost:8000/status > /dev/null && exit 0
+            sleep 1
+          done
+          echo "bazel-remote did not start in time"
+          cat /tmp/bazel-remote.log
+          exit 1
+      - name: Remote cache scripted test
+        shell: bash
+        run: |
+          sbt -v --client "doc; publishLocal"
+          sbt -v --client "scripted remote-cache/*"
+      - name: Server test
+        shell: bash
+        run: sbt -v --client "serverTestProj/test"
+      - name: Stop bazel-remote
+        if: always()
+        run: |
+          if [ -f /tmp/bazel-remote.pid ]; then
+            kill "$(cat /tmp/bazel-remote.pid)" || true
+          fi
--- a/build.sbt
+++ b/build.sbt
@ -391,6 +391,7 @@ lazy val utilCache = project
    contrabandSettings,
    mimaSettings,
    mimaBinaryIssueFilters ++= Seq(
+      exclude[DirectMissingMethodProblem]("sbt.internal.util.CacheEventSummary#Data.*"),
    ),
    Test / fork := true,
  )
@ -558,8 +559,11 @@ lazy val remoteCacheProj = (project in file("sbt-remote-cache"))
    name := "sbt-remote-cache",
    pluginCrossBuild / sbtVersion := version.value,
    publishMavenStyle := true,
-    mimaSettings,
    libraryDependencies ++= Seq(remoteapis, scalaVerify % Test),
+    mimaSettings,
+    mimaBinaryIssueFilters ++= Seq(
+      exclude[DirectMissingMethodProblem]("sbt.internal.GrpcActionCacheStore.this"),
+    ),
  )

 // Implementation and support code for defining actions.
--- a/main/src/main/scala/sbt/ProjectExtra.scala
+++ b/main/src/main/scala/sbt/ProjectExtra.scala
@ -376,6 +376,15 @@ trait ProjectExtra extends Scoped.Syntax:
      val srvLogLevel: Option[Level.Value] = (ref / serverLog / logLevel).get(structure.data)
      val hs: Option[Seq[ServerHandler]] = get(ThisBuild / fullServerHandlers)
      val caches: Option[Seq[ActionCacheStore]] = get(cacheStores)
+      // cacheStores is recomputed on every reload; close any store dropped from the new value.
+      s.attributes.get(cacheStores.key) match
+        case Some(oldCaches) =>
+          val kept = caches.getOrElse(Nil)
+          oldCaches.foreach {
+            case store: AutoCloseable if !kept.exists(_ eq store) => store.close()
+            case _                                                => ()
+          }
+        case None => ()
      val rod: Option[NioPath] = get(rootOutputDirectory)
      val fileConverter: Option[FileConverter] = get(Keys.fileConverter)
      val commandDefs = allCommands.distinct.flatten[Command].map(_.tag(projectCommand, true))
--- a/sbt-app/src/sbt-test/remote-cache/basic/Hello.scala
+++ b/sbt-app/src/sbt-test/remote-cache/basic/Hello.scala
@ -0,0 +1,2 @@
+object Hello:
+  def main(args: Array[String]): Unit = println("Hello, world!")
--- a/sbt-app/src/sbt-test/remote-cache/basic/build.sbt
+++ b/sbt-app/src/sbt-test/remote-cache/basic/build.sbt
@ -0,0 +1,23 @@
+import sbt.internal.util.CacheEventSummary
+
+scalaVersion := "3.8.4"
+
+Global / remoteCache := Some(new java.net.URI("grpc://127.0.0.1:2024"))
+Global / localCacheDirectory := baseDirectory.value / "diskcache"
+
+val checkHit = taskKey[Unit]("asserts the previous compile was served from the remote cache")
+
+checkHit := Def.uncached {
+  val config = Def.cacheConfiguration.value
+  val prev = config.cacheEventLog.previous match
+    case data: CacheEventSummary.Data => data
+    case _                            => sys.error("empty event log")
+  streams.value.log.info(
+    s"prev hitCount=${prev.hitCount} missCount=${prev.missCount} remoteHitCount=${prev.remoteHitCount}"
+  )
+  assert(prev.missCount == 0, s"expected 100% hit rate but missCount=${prev.missCount}")
+  assert(
+    prev.remoteHitCount == prev.hitCount,
+    s"expected 100% remote hit rate but remoteHitCount=${prev.remoteHitCount} hitCount=${prev.hitCount}"
+  )
+}
--- a/sbt-app/src/sbt-test/remote-cache/basic/project/plugins.sbt
+++ b/sbt-app/src/sbt-test/remote-cache/basic/project/plugins.sbt
@ -0,0 +1 @@
+addRemoteCachePlugin
--- a/sbt-app/src/sbt-test/remote-cache/basic/test
+++ b/sbt-app/src/sbt-test/remote-cache/basic/test
@ -0,0 +1,5 @@
+> compile
+$ delete diskcache
+> clean
+> compile
+> checkHit
--- a/sbt-remote-cache/src/main/scala/sbt/internal/GrpcActionCacheStore.scala
+++ b/sbt-remote-cache/src/main/scala/sbt/internal/GrpcActionCacheStore.scala
@ -44,8 +44,10 @@ import sbt.util.{
  GetActionResultRequest,
  UpdateActionResultRequest,
 }
+import scala.collection.concurrent.TrieMap
 import scala.concurrent.{ Await, ExecutionContext, Future, Promise, TimeoutException }
 import scala.concurrent.duration.*
+import scala.ref.WeakReference
 import scala.util.Using
 import scala.util.control.NonFatal
 import scala.jdk.CollectionConverters.*
@ -57,6 +59,16 @@ object GrpcActionCacheStore:
  val remoteTimeoutInSec = 60
  val remoteTimeout = (remoteTimeoutInSec + 2).second

+  private case class CacheValue(
+      rootCerts: Option[Path],
+      clientCertChain: Option[Path],
+      clientPrivateKey: Option[Path],
+      remoteHeaders: List[String],
+      store: WeakReference[GrpcActionCacheStore],
+  )
+
+  private val instances: TrieMap[URI, CacheValue] = TrieMap.empty
+
  def apply(
      uri: URI,
      rootCerts: Option[Path],
@ -64,6 +76,36 @@ object GrpcActionCacheStore:
      clientPrivateKey: Option[Path],
      remoteHeaders: List[String],
      disk: DiskActionCacheStore,
+  ): GrpcActionCacheStore =
+    def mkStore(): GrpcActionCacheStore =
+      val store = build(uri, rootCerts, clientCertChain, clientPrivateKey, remoteHeaders, disk)
+      instances.put(
+        uri,
+        CacheValue(
+          rootCerts,
+          clientCertChain,
+          clientPrivateKey,
+          remoteHeaders,
+          WeakReference(store)
+        )
+      )
+      store
+    instances.get(uri) match
+      case Some(v)
+          if v.rootCerts == rootCerts && v.clientCertChain == clientCertChain
+            && v.clientPrivateKey == clientPrivateKey && v.remoteHeaders == remoteHeaders =>
+        v.store.get match
+          case Some(existing) => existing
+          case None           => mkStore()
+      case _ => mkStore()
+
+  private def build(
+      uri: URI,
+      rootCerts: Option[Path],
+      clientCertChain: Option[Path],
+      clientPrivateKey: Option[Path],
+      remoteHeaders: List[String],
+      disk: DiskActionCacheStore,
  ): GrpcActionCacheStore =
    val b: ManagedChannelBuilder[?] = uri.getScheme() match
      case "grpc" =>
@ -96,7 +138,7 @@ object GrpcActionCacheStore:
      case Some(x) if x.startsWith("/") => x.drop(1)
      case Some(x)                      => x
      case None                         => ""
-    new GrpcActionCacheStore(channel, instanceName, remoteHeaders, disk)
+    new GrpcActionCacheStore(channel, instanceName, remoteHeaders, disk, uri)

  class AuthCallCredentials(remoteHeaders: List[String]) extends CallCredentials:
    val pairs = remoteHeaders.map: h =>
@ -132,12 +174,14 @@ end GrpcActionCacheStore
 * https://github.com/bazelbuild/remote-apis/blob/main/build/bazel/remote/execution/v2/remote_execution.proto
 * https://github.com/googleapis/googleapis/blob/ff15be54722218705740b9fc6223d264c4cdb6dd/google/bytestream/bytestream.proto
 */
-class GrpcActionCacheStore(
+class GrpcActionCacheStore private (
    channel: ManagedChannel,
    instanceName: String,
    remoteHeaders: List[String],
    disk: DiskActionCacheStore,
-) extends AbstractActionCacheStore:
+    cacheKey: URI,
+) extends AbstractActionCacheStore
+    with AutoCloseable:
  import GrpcActionCacheStore.*

  lazy val creds = GrpcActionCacheStore.AuthCallCredentials(remoteHeaders)
@ -167,6 +211,17 @@ class GrpcActionCacheStore(
  val fixedThreadPool = Executors.newFixedThreadPool(100)
  given ExecutionContext = ExecutionContext.fromExecutor(fixedThreadPool)

+  override def close(): Unit =
+    instances.get(cacheKey).foreach { v =>
+      if v.store.get.contains(this) then instances.remove(cacheKey, v)
+    }
+    try
+      try
+        channel.shutdown()
+        if !channel.awaitTermination(5, TimeUnit.SECONDS) then channel.shutdownNow()
+      catch case NonFatal(_) => channel.shutdownNow()
+    finally fixedThreadPool.shutdown()
+
  /**
   * https://github.com/bazelbuild/remote-apis/blob/9ff14cecffe5287ba337f857731ceadfc2d80de9/build/bazel/remote/execution/v2/remote_execution.proto#L170
   */
--- a/util-cache/src/main/scala/sbt/internal/util/CacheEventLog.scala
+++ b/util-cache/src/main/scala/sbt/internal/util/CacheEventLog.scala
@ -18,7 +18,8 @@ enum CacheEventSummary:
      missCount: Long,
      hitRate: Double,
      onsiteCount: Option[Long],
-      errorCount: Option[Long]
+      errorCount: Option[Long],
+      remoteHitCount: Long
  )
  override def toString(): String = this match
    case Empty => ""
@ -28,7 +29,8 @@ enum CacheEventSummary:
          missCount,
          hitRate,
          onsiteCount,
-          errorCount
+          errorCount,
+          remoteHitCount
        ) =>
      val hitDescs = hits.map {
        case (id, 1) => s"1 $id cache hit"
@ -74,12 +76,14 @@ class CacheEventLog:
      val hitRate = if total > 0 then (hitCount.toDouble / total.toDouble) else 0.0
      val onsiteCount = events.get(ActionCacheEvent.OnsiteTask)
      val errorCount = events.get(ActionCacheEvent.Error)
+      val remoteHitCount = hits.view.collect { case (id, v) if id.startsWith("remote") => v }.sum
      CacheEventSummary.Data(
        hits.toSeq,
        hitCount,
        missCount,
        hitRate,
        onsiteCount,
-        errorCount
+        errorCount,
+        remoteHitCount
      )
 end CacheEventLog
--- a/util-cache/src/test/scala/sbt/util/CacheEventLogTest.scala
+++ b/util-cache/src/test/scala/sbt/util/CacheEventLogTest.scala
@ -1,6 +1,6 @@
 package sbt.util

-import sbt.internal.util.{ ActionCacheEvent, CacheEventLog }
+import sbt.internal.util.{ ActionCacheEvent, CacheEventLog, CacheEventSummary }
 import verify.BasicTestSuite

 object CacheEventLogTest extends BasicTestSuite:
@ -59,6 +59,9 @@ object CacheEventLogTest extends BasicTestSuite:
    logger.append(ActionCacheEvent.OnsiteTask)
    val expectedSummary = "cache 75%, 1 disk cache hit, 2 remote cache hits, 1 onsite task"
    assertEquals(logger.summary.toString(), expectedSummary)
+    logger.summary match
+      case data: CacheEventSummary.Data => assert(data.remoteHitCount == 2L)
+      case _                            => sys.error("expected CacheEventSummary.Data")
  }

  test("summary of 1 disk event after clear") {