update canton to 20241114.14575.v1528fc50

sdk/canton/BUILD.bazel

@@ -630,7 +630,6 @@ scala_library(
         "@maven//:dev_optics_monocle_macro_2_13",
         "@maven//:io_circe_circe_core_2_13",
         "@maven//:io_grpc_grpc_api",
-        "@maven//:io_grpc_grpc_netty",
         "@maven//:io_grpc_grpc_services",
         "@maven//:io_grpc_grpc_stub",
         "@maven//:io_netty_netty_handler",

sdk/canton/community/app-base/src/main/scala/com/digitalasset/canton/admin/api/client/GrpcCtlRunner.scala

@@ -6,9 +6,8 @@ package com.digitalasset.canton.admin.api.client
 import cats.data.EitherT
 import com.daml.grpc.AuthCallCredentials
 import com.digitalasset.canton.admin.api.client.commands.GrpcAdminCommand
-import com.digitalasset.canton.lifecycle.OnShutdownRunner
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
-import com.digitalasset.canton.networking.grpc.CantonGrpcUtil
+import com.digitalasset.canton.networking.grpc.{CantonGrpcUtil, GrpcClient, GrpcManagedChannel}
 import com.digitalasset.canton.tracing.{TraceContext, TraceContextGrpc}
 import com.digitalasset.canton.util.LoggerUtil
 import io.grpc.ManagedChannel
@@ -22,7 +21,6 @@ import scala.concurrent.{ExecutionContext, Future}
 class GrpcCtlRunner(
     maxRequestDebugLines: Int,
     maxRequestDebugStringLength: Int,
-    onShutdownRunner: OnShutdownRunner,
     val loggerFactory: NamedLoggerFactory,
 ) extends NamedLogging {
 
@@ -32,27 +30,30 @@ class GrpcCtlRunner(
   def run[Req, Res, Result](
       instanceName: String,
       command: GrpcAdminCommand[Req, Res, Result],
-      channel: ManagedChannel,
+      managedChannel: GrpcManagedChannel,
       token: Option[String],
       timeout: Duration,
   )(implicit ec: ExecutionContext, traceContext: TraceContext): EitherT[Future, String, Result] = {
 
-    val baseService: command.Svc = command
-      .createServiceInternal(channel)
-      .withInterceptors(TraceContextGrpc.clientInterceptor)
-
-    val service = token.fold(baseService)(AuthCallCredentials.authorizingStub(baseService, _))
+    def service(channel: ManagedChannel): command.Svc = {
+      val baseService = command
+        .createServiceInternal(channel)
+        .withInterceptors(TraceContextGrpc.clientInterceptor)
+      token.toList.foldLeft(baseService)(AuthCallCredentials.authorizingStub)
+    }
+    val client = GrpcClient.create(managedChannel, service)
 
     for {
       request <- EitherT.fromEither[Future](command.createRequestInternal())
-      response <- submitRequest(command)(instanceName, service, request, timeout)
+      response <- submitRequest(command)(instanceName, client, request, timeout)
       result <- EitherT.fromEither[Future](command.handleResponseInternal(response))
     } yield result
   }
 
   private def submitRequest[Svc <: AbstractStub[Svc], Req, Res, Result](
       command: GrpcAdminCommand[Req, Res, Result]
-  )(instanceName: String, service: command.Svc, request: Req, timeout: Duration)(implicit
+  )(instanceName: String, service: GrpcClient[command.Svc], request: Req, timeout: Duration)(
+      implicit
       ec: ExecutionContext,
       traceContext: TraceContext,
   ): EitherT[Future, String, Res] = CantonGrpcUtil.shutdownAsGrpcErrorE(
@@ -64,8 +65,7 @@ class GrpcCtlRunner(
         ),
         timeout,
         logger,
-        onShutdownRunner,
-        CantonGrpcUtil.silentLogPolicy, // silent log policy, as the ConsoleEnvironment will log the result
+        CantonGrpcUtil.SilentLogPolicy, // silent log policy, as the ConsoleEnvironment will log the result
         _ => false, // no retry to optimize for low latency
       )
       .leftMap(_.toString)

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

@@ -16,10 +16,13 @@ import com.digitalasset.canton.config
 import com.digitalasset.canton.config.RequireTypes.Port
 import com.digitalasset.canton.config.{ClientConfig, ConsoleCommandTimeout, NonNegativeDuration}
 import com.digitalasset.canton.environment.Environment
-import com.digitalasset.canton.lifecycle.Lifecycle.CloseableChannel
 import com.digitalasset.canton.lifecycle.OnShutdownRunner
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
-import com.digitalasset.canton.networking.grpc.{CantonGrpcUtil, ClientChannelBuilder}
+import com.digitalasset.canton.networking.grpc.{
+  CantonGrpcUtil,
+  ClientChannelBuilder,
+  GrpcManagedChannel,
+}
 import com.digitalasset.canton.tracing.{Spanning, TraceContext}
 import io.opentelemetry.api.trace.Tracer
 
@@ -49,10 +52,9 @@ class GrpcAdminCommandRunner(
   private val grpcRunner = new GrpcCtlRunner(
     environment.config.monitoring.logging.api.maxMessageLines,
     environment.config.monitoring.logging.api.maxStringLength,
-    onShutdownRunner = this,
     loggerFactory,
   )
-  private val channels = TrieMap[(String, String, Port), CloseableChannel]()
+  private val channels = TrieMap[(String, String, Port), GrpcManagedChannel]()
 
   def runCommandAsync[Result](
       instanceName: String,
@@ -88,7 +90,8 @@ class GrpcAdminCommandRunner(
               CantonGrpcUtil.checkCantonApiInfo(
                 serverName = instanceName,
                 expectedName = apiName,
-                channel = ClientChannelBuilder.createChannelToTrustedServer(clientConfig),
+                channelBuilder =
+                  ClientChannelBuilder.createChannelBuilderToTrustedServer(clientConfig),
                 logger = logger,
                 timeout = commandTimeouts.bounded,
                 onShutdownRunner = this,
@@ -97,12 +100,12 @@ class GrpcAdminCommandRunner(
             )
         }
       }
-      closeableChannel = getOrCreateChannel(instanceName, clientConfig, callTimeout)
+      channel = getOrCreateChannel(instanceName, clientConfig)
       _ = logger.debug(s"Running command $command on $instanceName against $clientConfig")
       result <- grpcRunner.run(
         instanceName,
         command,
-        closeableChannel.channel,
+        channel,
         token,
         callTimeout.duration,
       )
@@ -135,16 +138,16 @@ class GrpcAdminCommandRunner(
   private def getOrCreateChannel(
       instanceName: String,
       clientConfig: ClientConfig,
-      callTimeout: config.NonNegativeDuration,
-  ): CloseableChannel =
+  ): GrpcManagedChannel =
     blocking(synchronized {
       val addr = (instanceName, clientConfig.address, clientConfig.port)
       channels.getOrElseUpdate(
         addr,
-        new CloseableChannel(
-          ClientChannelBuilder.createChannelToTrustedServer(clientConfig),
-          logger,
+        GrpcManagedChannel(
           s"ConsoleCommand",
+          ClientChannelBuilder.createChannelBuilderToTrustedServer(clientConfig).build(),
+          this,
+          logger,
         ),
       )
     })
@@ -153,10 +156,10 @@ class GrpcAdminCommandRunner(
   override def onFirstClose(): Unit =
     closeChannels()
 
-  def closeChannels(): Unit = {
+  def closeChannels(): Unit = blocking(synchronized {
     channels.values.foreach(_.close())
     channels.clear()
-  }
+  })
 }
 
 /** A console-specific version of the GrpcAdminCommandRunner that uses the console environment

sdk/canton/community/app-base/src/main/scala/com/digitalasset/canton/console/commands/ParticipantRepairAdministration.scala

@@ -6,12 +6,14 @@ package com.digitalasset.canton.console.commands
 import better.files.File
 import cats.syntax.either.*
 import cats.syntax.foldable.*
+import com.daml.nonempty.NonEmpty
 import com.digitalasset.canton.admin.api.client.commands.ParticipantAdminCommands
 import com.digitalasset.canton.admin.participant.v30.ExportAcsResponse
 import com.digitalasset.canton.config.RequireTypes.PositiveInt
 import com.digitalasset.canton.config.{ConsoleCommandTimeout, NonNegativeDuration}
 import com.digitalasset.canton.console.{
   AdminCommandRunner,
+  CommandErrors,
   ConsoleCommandResult,
   ConsoleEnvironment,
   FeatureFlag,
@@ -392,17 +394,26 @@ abstract class LocalParticipantRepairAdministration(
       skipInactive: Boolean = true,
       batchSize: Int = 100,
   ): Unit =
-    runRepairCommand(tc =>
-      access(
-        _.sync.repairService.changeAssignationAwait(
-          contractIds,
-          sourceDomain,
-          targetDomain,
-          skipInactive,
-          PositiveInt.tryCreate(batchSize),
-        )(tc)
-      )
-    )
+    NonEmpty
+      .from(contractIds.distinct) match {
+      case Some(contractIds) =>
+        runRepairCommand(tc =>
+          access(
+            _.sync.repairService.changeAssignationAwait(
+              contractIds,
+              sourceDomain,
+              targetDomain,
+              skipInactive,
+              PositiveInt.tryCreate(batchSize),
+            )(tc)
+          )
+        )
+      case None =>
+        consoleEnvironment.run(
+          CommandErrors
+            .GenericCommandError("contractIds must be non-empty")
+        )
+    }
 
   @Help.Summary("Rollback an unassignment by re-assigning the contract to the source domain.")
   @Help.Description(

sdk/canton/community/app/src/test/scala/com/digitalasset/canton/admin/api/client/GrpcCtlRunnerTest.scala

@@ -6,6 +6,7 @@ package com.digitalasset.canton.admin.api.client
 import com.digitalasset.canton.BaseTest
 import com.digitalasset.canton.admin.api.client.commands.GrpcAdminCommand
 import com.digitalasset.canton.lifecycle.OnShutdownRunner.PureOnShutdownRunner
+import com.digitalasset.canton.networking.grpc.GrpcManagedChannel
 import io.grpc.stub.AbstractStub
 import io.grpc.{CallOptions, Channel, ManagedChannel}
 import org.scalatest.wordspec.AsyncWordSpec
@@ -20,8 +21,7 @@ class GrpcCtlRunnerTest extends AsyncWordSpec with BaseTest {
       val (channel, command) = defaultMocks()
 
       "run successfully" in {
-        val onShutdownRunner = new PureOnShutdownRunner(logger)
-        new GrpcCtlRunner(1000, 1000, onShutdownRunner, loggerFactory).run(
+        new GrpcCtlRunner(1000, 1000, loggerFactory).run(
           "participant1",
           command,
           channel,
@@ -38,7 +38,7 @@ class GrpcCtlRunnerTest extends AsyncWordSpec with BaseTest {
     override def build(channel: Channel, callOptions: CallOptions): TestAbstractStub = this
   }
 
-  private def defaultMocks(): (ManagedChannel, GrpcAdminCommand[String, String, String]) = {
+  private def defaultMocks(): (GrpcManagedChannel, GrpcAdminCommand[String, String, String]) = {
     val channel = mock[ManagedChannel]
     val service = new TestAbstractStub(channel)
     val command = new GrpcAdminCommand[String, String, String] {
@@ -53,6 +53,14 @@ class GrpcCtlRunnerTest extends AsyncWordSpec with BaseTest {
       )
     }
 
-    (channel, command)
+    (
+      GrpcManagedChannel(
+        "GrcpCtrlRunnerTestChannel",
+        channel,
+        new PureOnShutdownRunner(logger),
+        logger,
+      ),
+      command,
+    )
   }
 }

sdk/canton/community/base/src/main/scala/com/digitalasset/canton/config/CacheConfig.scala

@@ -38,8 +38,11 @@ final case class CacheConfigWithTimeout(
     expireAfterTimeout: PositiveFiniteDuration = PositiveFiniteDuration.ofMinutes(10),
 ) {
 
-  def buildScaffeine(): Scaffeine[Any, Any] =
-    Scaffeine().maximumSize(maximumSize.value).expireAfterWrite(expireAfterTimeout.underlying)
+  def buildScaffeine()(implicit executionContext: ExecutionContext): Scaffeine[Any, Any] =
+    Scaffeine()
+      .maximumSize(maximumSize.value)
+      .expireAfterWrite(expireAfterTimeout.underlying)
+      .executor(executionContext.execute(_))
 
 }
 

sdk/canton/community/base/src/main/scala/com/digitalasset/canton/config/ServerConfig.scala

@@ -223,7 +223,9 @@ final case class ClientConfig(
     port: Port,
     tls: Option[TlsClientConfig] = None,
     keepAliveClient: Option[KeepAliveClientConfig] = Some(KeepAliveClientConfig()),
-)
+) {
+  def endpointAsString: String = address + ":" + port.unwrap
+}
 
 sealed trait BaseTlsArguments {
   def certChainFile: ExistingFile

sdk/canton/community/base/src/main/scala/com/digitalasset/canton/logging/LastErrorsAppender.scala

@@ -13,6 +13,7 @@ import com.digitalasset.canton.discard.Implicits.DiscardOps
 import com.github.blemale.scaffeine.Scaffeine
 
 import java.util
+import java.util.concurrent.Executors
 import scala.collection.mutable
 
 /** Logback appender that keeps a bounded queue of errors/warnings that have been logged and associated log entries with
@@ -33,13 +34,22 @@ class LastErrorsAppender()
   private var maxErrors = 128
   private var lastErrorsFileAppenderName = ""
 
+  /** Separate executor for scaffeine - to avoid using ForkJoin common pool */
+  private val scaffeineExecutor = Executors.newSingleThreadExecutor()
+
   /** An error/warn event with previous events of the same trace-id */
   private case class ErrorWithEvents(error: ILoggingEvent, events: Seq[ILoggingEvent])
 
   private val eventsCache =
-    Scaffeine().maximumSize(maxTraces.toLong).build[String, BoundedQueue[ILoggingEvent]]()
+    Scaffeine()
+      .maximumSize(maxTraces.toLong)
+      .executor(scaffeineExecutor)
+      .build[String, BoundedQueue[ILoggingEvent]]()
   private val errorsCache =
-    Scaffeine().maximumSize(maxErrors.toLong).build[String, ErrorWithEvents]()
+    Scaffeine()
+      .maximumSize(maxErrors.toLong)
+      .executor(scaffeineExecutor)
+      .build[String, ErrorWithEvents]()
 
   private def isLastErrorsFileAppender(appender: Appender[ILoggingEvent]): Boolean =
     appender.getName == lastErrorsFileAppenderName