update our Canton fork to latest release-line-3.5

build.sbt

@@ -35,7 +35,6 @@ lazy val `canton-pekko-fork` = BuildCommon.`canton-pekko-fork`
 lazy val `canton-magnolify-addon` = BuildCommon.`canton-magnolify-addon`
 lazy val `canton-scalatest-addon` = BuildCommon.`canton-scalatest-addon`
 lazy val `canton-ledger-common` = BuildCommon.`canton-ledger-common`
-lazy val `canton-ledger-api-core` = BuildCommon.`canton-ledger-api-core`
 lazy val `canton-ledger-api-value` = BuildCommon.`canton-ledger-api-value`
 lazy val `canton-ledger-json-api` = BuildCommon.`canton-ledger-json-api`
 lazy val `canton-daml-adjustable-clock` = BuildCommon.`canton-daml-adjustable-clock`
@@ -47,6 +46,7 @@ lazy val `canton-sequencer-driver-api` = BuildCommon.`canton-sequencer-driver-ap
 lazy val `canton-kms-driver-api` = BuildCommon.`canton-kms-driver-api`
 lazy val `canton-community-reference-driver` = BuildCommon.`canton-community-reference-driver`
 lazy val `canton-observability-metrics-testing` = BuildCommon.`canton-observability-metrics-testing`
+lazy val `canton-traffic-enforcement-component` = BuildCommon.`canton-traffic-enforcement-component`
 
 lazy val `splice-wartremover-extension` = Wartremover.`splice-wartremover-extension`
 
@@ -149,7 +149,6 @@ lazy val root: Project = (project in file("."))
     `canton-community-synchronizer`,
     `canton-community-participant`,
     `canton-ledger-common`,
-    `canton-ledger-api-core`,
     `canton-ledger-api-value`,
     `canton-google-common-protos-scala`,
     `canton-observability-metrics-testing`,
@@ -2502,10 +2501,7 @@ updateTestConfigForParallelRuns := {
   val allTestNames =
     definedTests
       .all(
-        ScopeFilter(inAggregates(root), inConfigurations(Test)) -- ScopeFilter(
-          inProjects(`canton-ledger-api-core`),
-          inConfigurations(Test),
-        )
+        ScopeFilter(inAggregates(root), inConfigurations(Test))
       )
       .value
       .flatten

canton/CODEOWNERS

@@ -48,3 +48,6 @@ CODEOWNERS @DACH-NY/canton-change-owners
 
 # Inform CN on docker image changes but does not require their approval (canton-change-owners can approve)
 /docker/canton/images @DACH-NY/canton-network-upstream-notifications @DACH-NY/canton-change-owners
+
+# Docs should be updated in the cf-docs repo instead
+/docs-open @soren-da @rgugliel-da

canton/VERSION

@@ -1 +1 @@
-3.5.1-SNAPSHOT
+3.5.7-SNAPSHOT

canton/base/testing-utils/src/main/resources/logback-test.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration debug="false">
+  <!-- propagate logback changes to jul handlers -->
+  <contextListener class="ch.qos.logback.classic.jul.LevelChangePropagator">
+    <resetJUL>true</resetJUL>
+  </contextListener>
+
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <!-- attempt to place the trace-id with a trailing space, however replace with an empty string if it's empty -->
+      <pattern>%highlight(%-5level %logger{10} %replace(tid:%mdc{trace-id} ){'tid: ', ''}- %msg%replace(, context: %marker){', context: $', ''}%replace( err-context:%mdc{err-context} ){' err-context: ', ''}%n)</pattern>
+    </encoder>
+    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+      <level>WARN</level>
+    </filter>
+  </appender>
+
+  <appender name="FILE" class="ch.qos.logback.core.FileAppender">
+    <file>${LOG_FILE_NAME:-log/canton_test.log}</file>
+    <append>${LOG_APPEND:-true}</append>
+    <encoder>
+      <!-- attempt to place the trace-id with a trailing space, however replace with an empty string if it's empty -->
+      <pattern>%date [%thread] %-5level %logger{10} %replace(tid:%mdc{trace-id} ){'tid: ', ''}- %msg%replace(, context: %marker){', context: $', ''}%replace( err-context:%mdc{err-context} ){' err-context: ', ''}%n</pattern>
+    </encoder>
+  </appender>
+
+    <logger name="slick.jdbc.JdbcBackend.statement" level="DEBUG" />
+
+  <!-- Include general rewrite rules for messages we also want to rewrite in production
+       By setting the REWRITE_TESTING variable we can include the warning suppression of testing messages.
+  -->
+  <property name="REWRITE_TESTING" value="true" />
+  <include resource="rewrite-appender.xml"/>
+
+  <root level="INFO">
+    <appender-ref ref="REWRITE_LOG_LEVEL_SYNC"/>
+  </root>
+
+  <!-- Ensure everything is logged before the JVM exits -->
+  <shutdownHook class="ch.qos.logback.core.hook.DefaultShutdownHook" />
+
+  <!-- Note that com.daml logs valuable info at TRACE level.
+       Set to TRACE for troubleshooting.
+  -->
+  <logger name="com.daml" level="${LOG_LEVEL_CANTON:-DEBUG}"/>
+  <logger name="com.digitalasset" level="${LOG_LEVEL_CANTON:-DEBUG}"/>
+  <!-- The TransactionsReader is quite noisy. -->
+  <logger name="com.digitalasset.canton.platform.store.dao.events.TransactionsReaderCanton" level="${LOG_LEVEL_CANTON:-INFO}"/>
+  <!-- netty reports ssl errors on DEBUG level ... -->
+  <logger name="io.grpc.netty.shaded.io.netty.handler.ssl" level="${LOG_LEVEL_CANTON:-DEBUG}"/>
+
+  <!-- Set the logging level for Pekko so that if someone configures pekko to actually log something, this will end up in the log -->
+  <logger name="pekko" level="${LOG_LEVEL_CANTON:-DEBUG}"/>
+  <!-- janino can be too verbose in tests -->
+  <logger name="org.codehaus.janino" level="WARN"/>
+
+  <!-- Enable for database-relevant debug logs -->
+<!--  <logger name="slick" level="DEBUG"/>-->
+<!--  <logger name="com.zaxxer.hikari" level="DEBUG"/>-->
+<!--  <logger name="org.testcontainers" level="DEBUG"/>-->
+</configuration>

canton/base/testing-utils/src/main/scala/com/daml/testing/utils/GrpcServerResource.scala

@@ -6,6 +6,7 @@ package com.daml.testing.utils
 import io.grpc.*
 
 import java.net.SocketAddress
+import java.util.UUID
 import java.util.concurrent.TimeUnit
 
 final class GrpcServerResource(
@@ -17,7 +18,11 @@ final class GrpcServerResource(
 
   override protected def construct(): ServerWithChannelProvider = {
     boundServices = services()
-    ServerWithChannelProvider.fromServices(boundServices, port, "server")
+    ServerWithChannelProvider.fromServices(
+      boundServices,
+      port,
+      s"server_${UUID.randomUUID().toString}",
+    )
   }
 
   override protected def destruct(resource: ServerWithChannelProvider): Unit = {

canton/build.sbt

@@ -464,6 +464,10 @@ lazy val `model-based-testing-drivers` =
   CommunityProjects.`model-based-testing-drivers`
 lazy val `model-based-testing-integration-tests` =
   CommunityProjects.`model-based-testing-integration-tests`
+lazy val `traffic-enforcement-api` =
+  CommunityProjects.`traffic-enforcement-api`
+lazy val `traffic-enforcement-component` =
+  CommunityProjects.`traffic-enforcement-component`
 
 lazy val `scalatest-utils` = DamlProjects.`scalatest-utils`
 lazy val `scala-utils` = DamlProjects.`scala-utils`

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

@@ -190,10 +190,18 @@ import com.digitalasset.canton.networking.grpc.ForwardingStreamObserver
 import com.digitalasset.canton.platform.apiserver.execution.CommandStatus
 import com.digitalasset.canton.protocol.LfContractId
 import com.digitalasset.canton.serialization.ProtoConverter
+import com.digitalasset.canton.tea.v1.TrafficServiceGrpc.TrafficServiceStub
+import com.digitalasset.canton.tea.v1.{
+  GetAccountRequest,
+  GetAccountResponse,
+  TrafficServiceGrpc,
+  UpdateAccountRequest,
+  UpdateAccountResponse,
+}
 import com.digitalasset.canton.topology.transaction.TopologyTransaction.GenericTopologyTransaction
 import com.digitalasset.canton.topology.{ParticipantId, Party, PartyId, SynchronizerId}
 import com.digitalasset.canton.util.BinaryFileUtil
-import com.digitalasset.canton.{LfPackageId, LfPackageName, LfPartyId}
+import com.digitalasset.canton.{LfPackageId, LfPackageName, LfPartyId, config}
 import com.google.protobuf.empty.Empty
 import com.google.protobuf.field_mask.FieldMask
 import io.grpc.*
@@ -1747,6 +1755,7 @@ object LedgerApiCommands {
         minLedgerTimeAbs: Option[Instant],
         deduplicationPeriod: Option[DeduplicationPeriod],
         hashingSchemeVersion: HashingSchemeVersion,
+        optTimeout: Option[config.NonNegativeDuration],
     ) extends BaseCommand[
           ExecuteSubmissionAndWaitRequest,
           ExecuteSubmissionAndWaitResponse,
@@ -1776,7 +1785,8 @@ object LedgerApiCommands {
       ): Either[String, ExecuteSubmissionAndWaitResponse] =
         Right(response)
 
-      override def timeoutType: TimeoutType = DefaultUnboundedTimeout
+      override def timeoutType: TimeoutType =
+        optTimeout.map(CustomClientTimeout(_)).getOrElse(DefaultUnboundedTimeout)
     }
 
     final case class ExecuteAndWaitForTransactionCommand(
@@ -1790,6 +1800,7 @@ object LedgerApiCommands {
         transactionShape: Option[TransactionShape],
         includeCreatedEventBlob: Boolean,
         customEventFormat: Option[EventFormat],
+        optTimeout: Option[config.NonNegativeDuration],
     ) extends BaseCommand[
           ExecuteSubmissionAndWaitForTransactionRequest,
           ExecuteSubmissionAndWaitForTransactionResponse,
@@ -1851,7 +1862,8 @@ object LedgerApiCommands {
       ): Either[String, ExecuteSubmissionAndWaitForTransactionResponse] =
         Right(response)
 
-      override def timeoutType: TimeoutType = DefaultUnboundedTimeout
+      override def timeoutType: TimeoutType =
+        optTimeout.map(CustomClientTimeout(_)).getOrElse(DefaultUnboundedTimeout)
     }
 
     final case class PreferredPackageVersion(
@@ -2450,4 +2462,48 @@ object LedgerApiCommands {
 
     }
   }
+  object Traffic {
+
+    abstract class BaseCommand[Req, Res] extends GrpcAdminCommand[Req, Res, Res] {
+      override type Svc = TrafficServiceStub
+
+      override def createService(channel: ManagedChannel): TrafficServiceStub =
+        TrafficServiceGrpc.stub(channel)
+
+      override protected def handleResponse(response: Res): Either[String, Res] = Right(response)
+    }
+
+    final case class GetAccount(accountId: String)
+        extends BaseCommand[
+          GetAccountRequest,
+          GetAccountResponse,
+        ] {
+      override protected def createRequest(): Either[String, GetAccountRequest] =
+        Right(GetAccountRequest(accountId))
+
+      override protected def submitRequest(
+          service: TrafficServiceStub,
+          request: GetAccountRequest,
+      ): Future[GetAccountResponse] =
+        service.getAccount(request)
+    }
+
+    final case class UpdateAccount(
+        accountId: String,
+        balance: Option[Long],
+        deduplicationId: String,
+    ) extends BaseCommand[
+          UpdateAccountRequest,
+          UpdateAccountResponse,
+        ] {
+      override protected def createRequest(): Either[String, UpdateAccountRequest] =
+        Right(UpdateAccountRequest(accountId, balance, deduplicationId))
+
+      override protected def submitRequest(
+          service: TrafficServiceStub,
+          request: UpdateAccountRequest,
+      ): Future[UpdateAccountResponse] =
+        service.updateAccount(request)
+    }
+  }
 }

canton/community/app-base/src/main/scala/com/digitalasset/canton/admin/api/client/data/SynchronizerConnectionConfig.scala

@@ -243,7 +243,10 @@ final case class SynchronizerConnectionConfig(
     )
 
   def toInternal: SynchronizerConnectionConfigInternal =
-    this.transformInto[SynchronizerConnectionConfigInternal]
+    this
+      .into[SynchronizerConnectionConfigInternal]
+      .withFieldRenamed(_.synchronizerId, _.psid)
+      .transform
 }
 
 object SynchronizerConnectionConfig {
@@ -320,5 +323,5 @@ object SynchronizerConnectionConfig {
   private[canton] def fromInternal(
       internal: SynchronizerConnectionConfigInternal
   ): SynchronizerConnectionConfig =
-    internal.transformInto[SynchronizerConnectionConfig]
+    internal.into[SynchronizerConnectionConfig].withFieldRenamed(_.psid, _.synchronizerId).transform
 }