[release-line-3.4] Update 2025-12-30.21 (#418)

Reference commit: 54b52623bc

Author: canton-machine

base/util-external/src/main/scala/com/digitalasset/canton/util/Mutex.scala

@@ -0,0 +1,41 @@
+// Copyright (c) 2025 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.digitalasset.canton.util
+
+import java.util.concurrent.locks.ReentrantLock
+
+/** Lock to be used instead of blocking synchronized
+  *
+  * The fork join pool is leaking threads potentially with every invocation of blocking. Therefore,
+  * only invoke it if really necessary.
+  */
+class Mutex {
+  private val lock = new ReentrantLock()
+
+  @deprecated("use harmonize, not synchronize", since = "3.4")
+  def synchronized[T](body: => T): T = sys.error("use harmonize to distinguish")
+
+  def exclusive[T](f: => T): T = {
+    def runAndUnlock() =
+      try {
+        f
+      } finally {
+        lock.unlock()
+      }
+    if (lock.tryLock()) {
+      runAndUnlock()
+    } else {
+      // this may trigger a ForkJoinPool.managedBlock
+      scala.concurrent.blocking {
+        lock.lock()
+        runAndUnlock()
+      }
+    }
+  }
+
+}
+
+object Mutex {
+  def apply(): Mutex = new Mutex
+}

community/app-base/src/main/scala/com/digitalasset/canton/admin/api/client/commands/AdminCommand.scala

@@ -10,13 +10,14 @@ import com.digitalasset.canton.admin.api.client.commands.GrpcAdminCommand.{
 }
 import com.digitalasset.canton.config.NonNegativeDuration
 import com.digitalasset.canton.discard.Implicits.DiscardOps
+import com.digitalasset.canton.util.Mutex
 import io.grpc.stub.{AbstractStub, StreamObserver}
 import io.grpc.{Context, ManagedChannel, Status, StatusException, StatusRuntimeException}
 
 import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
 import scala.collection.mutable.ListBuffer
 import scala.concurrent.duration.FiniteDuration
-import scala.concurrent.{Future, Promise, blocking}
+import scala.concurrent.{Future, Promise}
 
 trait AdminCommand[Req, Res, Result] {
 
@@ -104,9 +105,10 @@ object GrpcAdminCommand {
   ): Future[Seq[Result]] = {
     val promise = Promise[Seq[Result]]()
     val buffer = ListBuffer[Result]()
+    val lock = new Mutex()
     val context = Context.ROOT.withCancellation()
 
-    def success(): Unit = blocking(buffer.synchronized {
+    def success(): Unit = (lock.exclusive {
       promise.trySuccess(buffer.toList).discard[Boolean]
       context.close()
     })
@@ -117,7 +119,7 @@ object GrpcAdminCommand {
         new StreamObserver[Response]() {
           override def onNext(value: Response): Unit = {
             val extracted = extract(value)
-            blocking(buffer.synchronized {
+            (lock.exclusive {
               if (buffer.lengthCompare(expected) < 0) {
                 buffer ++= extracted
                 if (buffer.lengthCompare(expected) >= 0) {

community/app-base/src/main/scala/com/digitalasset/canton/console/AmmoniteCacheLock.scala

@@ -7,11 +7,11 @@ import ammonite.runtime.Storage
 import cats.syntax.either.*
 import com.digitalasset.canton.logging.TracedLogger
 import com.digitalasset.canton.tracing.TraceContext
+import com.digitalasset.canton.util.Mutex
 
 import java.io.{File, RandomAccessFile}
 import java.nio.channels.OverlappingFileLockException
 import java.nio.file.Files
-import scala.concurrent.blocking
 import scala.util.control.NonFatal
 
 trait AmmoniteCacheLock {
@@ -22,6 +22,8 @@ trait AmmoniteCacheLock {
 
 object AmmoniteCacheLock {
 
+  private val lock = new Mutex()
+
   // Don't change this to lazy val, as the underlying InMemory storage is not thread safe.
   def InMemory: AmmoniteCacheLock = new AmmoniteCacheLock {
     override def release(): Unit = ()
@@ -56,7 +58,7 @@ object AmmoniteCacheLock {
         lock <- go(0)
       } yield lock
       attempt match {
-        case Right(lock) => lock
+        case Right(cacheLock) => cacheLock
         case Left(err) =>
           logger.warn(
             s"Failed to acquire ammonite cache directory due to $err. Will use in-memory instead."
@@ -87,7 +89,7 @@ object AmmoniteCacheLock {
 
   private def acquireLock(logger: TracedLogger, path: os.Path, isRepl: Boolean)(implicit
       traceContext: TraceContext
-  ): Either[Throwable, Option[AmmoniteCacheLock]] = blocking(synchronized {
+  ): Either[Throwable, Option[AmmoniteCacheLock]] = (lock.exclusive {
     try {
       val myLockFile = path / "lock"
       logger.debug(s"Attempting to obtain lock $myLockFile")
@@ -97,12 +99,12 @@ object AmmoniteCacheLock {
           logger.debug(s"Failed to acquire lock for $myLockFile")
           out.close()
           Right(None)
-        case Some(lock) =>
+        case Some(channelLock) =>
           Right(Some(new AmmoniteCacheLock {
             override def release(): Unit =
               try {
                 logger.debug(s"Releasing lock $myLockFile...")
-                lock.release()
+                channelLock.release()
                 out.close()
                 if (!Files.deleteIfExists(myLockFile.toNIO)) { // throws when the file exists but cannot be deleted
                   logger.warn(s"Failed to delete lock file $myLockFile because it did not exist")

community/app-base/src/main/scala/com/digitalasset/canton/console/GrpcAdminCommandRunner.scala

@@ -29,11 +29,12 @@ import com.digitalasset.canton.networking.grpc.{
   GrpcManagedChannel,
 }
 import com.digitalasset.canton.tracing.{Spanning, TraceContext}
+import com.digitalasset.canton.util.Mutex
 import io.opentelemetry.api.trace.Tracer
 
 import scala.collection.concurrent.TrieMap
 import scala.concurrent.duration.{Duration, FiniteDuration}
-import scala.concurrent.{ExecutionContextExecutor, Future, blocking}
+import scala.concurrent.{ExecutionContextExecutor, Future}
 
 /** Attempt to run a grpc admin-api command against whatever is pointed at in the config
   * @param commandTimeouts
@@ -58,6 +59,7 @@ class GrpcAdminCommandRunner(
     loggerFactory,
   )
   private val channels = TrieMap[(String, String, Port), GrpcManagedChannel]()
+  private val lock = new Mutex()
 
   def runCommandAsync[Result](
       instanceName: String,
@@ -151,7 +153,7 @@ class GrpcAdminCommandRunner(
       instanceName: String,
       clientConfig: ClientConfig,
   ): GrpcManagedChannel =
-    blocking(synchronized {
+    (lock.exclusive {
       val addr = (instanceName, clientConfig.address, clientConfig.port)
       channels.getOrElseUpdate(
         addr,
@@ -168,7 +170,7 @@ class GrpcAdminCommandRunner(
   override def onFirstClose(): Unit =
     closeChannels()
 
-  def closeChannels(): Unit = blocking(synchronized {
+  def closeChannels(): Unit = (lock.exclusive {
     channels.values.foreach(_.close())
     channels.clear()
   })

community/app-base/src/main/scala/com/digitalasset/canton/environment/Environment.scala

@@ -43,7 +43,7 @@ import com.digitalasset.canton.time.*
 import com.digitalasset.canton.tracing.TraceContext.withNewTraceContext
 import com.digitalasset.canton.tracing.{NoTracing, TraceContext, TracerProvider}
 import com.digitalasset.canton.util.FutureInstances.parallelFuture
-import com.digitalasset.canton.util.{MonadUtil, PekkoUtil, SingleUseCell}
+import com.digitalasset.canton.util.{MonadUtil, Mutex, PekkoUtil, SingleUseCell}
 import com.google.common.annotations.VisibleForTesting
 import io.circe.Encoder
 import io.circe.generic.semiauto.deriveEncoder
@@ -53,8 +53,8 @@ import org.slf4j.bridge.SLF4JBridgeHandler
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.atomic.AtomicReference
 import scala.collection.mutable.ListBuffer
+import scala.concurrent.Future
 import scala.concurrent.duration.DurationInt
-import scala.concurrent.{Future, blocking}
 import scala.util.control.NonFatal
 
 /** Holds all significant resources held by this process.
@@ -209,6 +209,8 @@ class Environment(
   private val deadlockConfig = config.monitoring.deadlockDetection
   protected def timeouts: ProcessingTimeout = config.parameters.timeouts.processing
 
+  private val lock = new Mutex()
+
   protected val futureSupervisor =
     if (config.monitoring.logging.logSlowFutures)
       new FutureSupervisor.Impl(timeouts.slowFutureWarn, loggerFactory)
@@ -567,7 +569,7 @@ class Environment(
 
   def addUserCloseable(closeable: AutoCloseable): Unit = userCloseables.append(closeable)
 
-  override def close(): Unit = blocking(this.synchronized {
+  override def close(): Unit = (lock.exclusive {
     val closeActorSystem: AutoCloseable =
       LifeCycle.toCloseableActorSystem(actorSystem, logger, timeouts)
 

community/app-base/src/main/scala/com/digitalasset/canton/environment/Nodes.scala

@@ -31,10 +31,11 @@ import com.digitalasset.canton.synchronizer.sequencer.config.{
 }
 import com.digitalasset.canton.synchronizer.sequencer.{SequencerNode, SequencerNodeBootstrap}
 import com.digitalasset.canton.tracing.TraceContext
+import com.digitalasset.canton.util.Mutex
 import com.digitalasset.canton.util.Thereafter.syntax.*
 
 import scala.collection.concurrent.TrieMap
-import scala.concurrent.{ExecutionContext, Future, Promise, blocking}
+import scala.concurrent.{ExecutionContext, Future, Promise}
 
 /** Group of CantonNodes of the same type (mediators, participants, sequencers). */
 trait Nodes[+Node <: CantonNode, +NodeBootstrap <: CantonNodeBootstrap[Node]]
@@ -128,6 +129,7 @@ class ManagedNodes[
     with HasCloseContext
     with FlagCloseableAsync {
 
+  private val lock = new Mutex()
   private val nodes = TrieMap[InstanceName, ManagedNodeStage[NodeBootstrap]]()
   override def names(): Seq[InstanceName] = configs.keys.toSeq
 
@@ -259,8 +261,8 @@ class ManagedNodes[
       params = parametersFor(name)
     } yield (config, params)
 
-  override def migrateDatabase(name: InstanceName): Either[StartupError, Unit] = blocking(
-    synchronized {
+  override def migrateDatabase(name: InstanceName): Either[StartupError, Unit] = (
+    lock.exclusive {
       for {
         cAndP <- configAndParams(name)
         (config, params) = cAndP
@@ -269,8 +271,8 @@ class ManagedNodes[
     }
   )
 
-  override def repairDatabaseMigration(name: InstanceName): Either[StartupError, Unit] = blocking(
-    synchronized {
+  override def repairDatabaseMigration(name: InstanceName): Either[StartupError, Unit] = (
+    lock.exclusive {
       for {
         cAndP <- configAndParams(name)
         (config, params) = cAndP

community/app-base/src/main/scala/com/digitalasset/canton/metrics/CsvReporter.scala

@@ -6,6 +6,7 @@ package com.digitalasset.canton.metrics
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
 import com.digitalasset.canton.tracing.NoTracing
+import com.digitalasset.canton.util.Mutex
 import io.opentelemetry.sdk.common.CompletableResultCode
 import io.opentelemetry.sdk.metrics.InstrumentType
 import io.opentelemetry.sdk.metrics.`export`.MetricExporter
@@ -15,7 +16,6 @@ import java.io.{BufferedWriter, File, FileWriter}
 import java.util
 import java.util.concurrent.atomic.AtomicBoolean
 import scala.collection.concurrent.TrieMap
-import scala.concurrent.blocking
 import scala.jdk.CollectionConverters.*
 import scala.util.{Failure, Success, Try}
 
@@ -26,7 +26,7 @@ class CsvReporter(config: MetricsReporterConfig.Csv, val loggerFactory: NamedLog
 
   private val running = new AtomicBoolean(true)
   private val files = new TrieMap[String, (FileWriter, BufferedWriter)]
-  private val lock = new Object()
+  private val lock = new Mutex()
 
   def getAggregationTemporality(instrumentType: InstrumentType): AggregationTemporality =
     AggregationTemporality.CUMULATIVE
@@ -52,8 +52,8 @@ class CsvReporter(config: MetricsReporterConfig.Csv, val loggerFactory: NamedLog
   }
 
   def `export`(metrics: util.Collection[MetricData]): CompletableResultCode = {
-    blocking {
-      lock.synchronized {
+    {
+      lock.exclusive {
         if (running.get()) {
           val ts = CantonTimestamp.now()
           MetricValue.allFromMetricData(metrics.asScala).foreach { case (value, metadata) =>

community/app/src/test/scala/com/digitalasset/canton/integration/tests/examples/ExampleIntegrationTest.scala

@@ -14,10 +14,8 @@ import com.digitalasset.canton.integration.{
   IsolatedEnvironments,
 }
 import com.digitalasset.canton.logging.NamedLogging
-import com.digitalasset.canton.util.ConcurrentBufferedLogger
 import com.digitalasset.canton.util.ShowUtil.*
-
-import scala.concurrent.blocking
+import com.digitalasset.canton.util.{ConcurrentBufferedLogger, Mutex}
 
 abstract class ExampleIntegrationTest(configPaths: File*)
     extends BaseIntegrationTest
@@ -64,6 +62,7 @@ trait HasConsoleScriptRunner { this: NamedLogging =>
 }
 
 object `ExampleIntegrationTest` {
+  private val lock = new Mutex()
   lazy val dockerImagesPath: File = "docker" / "canton" / "images"
   lazy val examplesPath: File = "community" / "app" / "src" / "pack" / "examples"
   lazy val simpleTopology: File = examplesPath / "01-simple-topology"
@@ -75,7 +74,8 @@ object `ExampleIntegrationTest` {
   lazy val advancedConfTestEnv: File =
     "community" / "app" / "src" / "test" / "resources" / "advancedConfDef.env"
   lazy val bftSequencerConfigurationFolder: File = examplesPath / "11-bft-sequencer"
-  def ensureSystemProperties(kvs: (String, String)*): Unit = blocking(synchronized {
+
+  def ensureSystemProperties(kvs: (String, String)*): Unit = (lock.exclusive {
     kvs.foreach { case (key, value) =>
       Option(System.getProperty(key)) match {
         case Some(oldValue) =>

community/app/src/test/scala/com/digitalasset/canton/util/ReleaseUtils.scala

@@ -19,9 +19,10 @@ import com.digitalasset.canton.version.{
 
 import java.nio.file.{Files, Paths}
 import scala.collection.concurrent.TrieMap
-import scala.concurrent.{ExecutionContext, Future, blocking}
+import scala.concurrent.{ExecutionContext, Future}
 
 /** A collection of small utilities for tests that have no obvious home */
+@SuppressWarnings(Array("com.digitalasset.canton.RequireBlocking"))
 object ReleaseUtils {
   final case class TestedRelease(
       releaseVersion: ReleaseVersion,
@@ -124,6 +125,7 @@ object ReleaseUtils {
     * synchronize between different requests for the same release.
     */
   private val releasesRetrieval: TrieMap[ReleaseVersion, Future[String]] = TrieMap.empty
+  private val lock = new Mutex()
 
   /** If the .tar.gz corresponding to release is not found locally, attempts to download it from
     * artifactory. Then, extract the .tar.gz file.
@@ -136,15 +138,13 @@ object ReleaseUtils {
   def retrieve(
       release: ReleaseVersion
   )(implicit elc: ErrorLoggingContext, ec: ExecutionContext): Future[String] =
-    blocking {
-      synchronized {
-        releasesRetrieval.get(release) match {
-          case Some(releaseRetrieval) => releaseRetrieval
-          case None =>
-            val releaseRetrieval = Future(downloadAndExtract(release))
-            releasesRetrieval.put(release, releaseRetrieval).discard
-            releaseRetrieval
-        }
+    lock.exclusive {
+      releasesRetrieval.get(release) match {
+        case Some(releaseRetrieval) => releaseRetrieval
+        case None =>
+          val releaseRetrieval = Future(downloadAndExtract(release))
+          releasesRetrieval.put(release, releaseRetrieval).discard
+          releaseRetrieval
       }
     }
 

community/base/src/main/scala/com/digitalasset/canton/auth/AsyncForwardingListener.scala

@@ -3,10 +3,10 @@
 
 package com.digitalasset.canton.auth
 
+import com.digitalasset.canton.util.Mutex
 import io.grpc.ServerCall
 
 import scala.collection.mutable
-import scala.concurrent.blocking
 
 /** This listener buffers all messages until `setNextListener` is called, at which point all
   * buffered messages are sent to the given listener. From then on, all future messages are sent
@@ -18,18 +18,18 @@ import scala.concurrent.blocking
 @SuppressWarnings(Array("org.wartremover.warts.Var"))
 abstract class AsyncForwardingListener[ReqT] extends ServerCall.Listener[ReqT] {
   protected type Listener = ServerCall.Listener[ReqT]
-  private[this] val lock = new Object
+  private[this] val lock = new Mutex()
   private[this] val stash: mutable.ListBuffer[Listener => Unit] = new mutable.ListBuffer
   private[this] var nextListener: Option[Listener] = None
 
-  private[this] def enqueueOrProcess(msg: Listener => Unit): Unit = blocking(lock.synchronized {
+  private[this] def enqueueOrProcess(msg: Listener => Unit): Unit = (lock.exclusive {
     val _ = nextListener.fold {
       val _ = stash.append(msg)
       ()
     }(msg)
   })
 
-  protected def setNextListener(listener: Listener): Unit = blocking(lock.synchronized {
+  protected def setNextListener(listener: Listener): Unit = (lock.exclusive {
     nextListener = Some(listener)
     stash.foreach(msg => msg(listener))
   })