From 65e8f34696ffe1c84ba9b159505b48a5286022e9 Mon Sep 17 00:00:00 2001 From: Eugene Yokota Date: Sat, 25 Jul 2026 18:50:17 -0400 Subject: [PATCH] [2.x] fix: Intern GrpcActionCacheStore **Problem** GrpcActionCacheStore gets recreated per reload. **Solution** This interns GrpcActionCacheStore based on the parameters. --- build.sbt | 5 +- main/src/main/scala/sbt/ProjectExtra.scala | 9 +++ .../sbt/internal/GrpcActionCacheStore.scala | 61 ++++++++++++++++++- 3 files changed, 71 insertions(+), 4 deletions(-) diff --git a/build.sbt b/build.sbt index 50a47303b..fc666c5f9 100644 --- a/build.sbt +++ b/build.sbt @@ -558,8 +558,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. diff --git a/main/src/main/scala/sbt/ProjectExtra.scala b/main/src/main/scala/sbt/ProjectExtra.scala index 4985203b4..de0a13e2b 100755 --- 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)) diff --git a/sbt-remote-cache/src/main/scala/sbt/internal/GrpcActionCacheStore.scala b/sbt-remote-cache/src/main/scala/sbt/internal/GrpcActionCacheStore.scala index bb22de175..defd96654 100644 --- 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 */