Update 2023-04-25.23 (#87)

Reference commit: d286f23461

Co-authored-by: Canton <canton@digitalasset.com>

Author: canton-machine

community/app-base/src/main/scala/com/digitalasset/canton/config/CantonConfig.scala

@@ -412,7 +412,6 @@ trait CantonConfig {
           initialProtocolVersion = participantParameters.initialProtocolVersion.unwrap,
         ),
         participantConfig.init.parameters.uniqueContractKeys,
-        participantConfig.init.parameters.unsafeEnableCausalityTracking,
         participantParameters.ledgerApiServerParameters,
         maxDbConnections = participantConfig.storage.maxConnectionsCanton(true, false, false).value,
         participantParameters.excludeInfrastructureTransactions,

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

@@ -540,6 +540,12 @@ trait RemoteParticipantReferenceCommon
   @Help.Group("Participant Pruning")
   def pruning: ParticipantPruningAdministrationGroup = pruning_
 
+  private lazy val repair_ =
+    new ParticipantRepairAdministration(consoleEnvironment, this, loggerFactory)
+
+  @Help.Summary("Commands to repair the participant contract state", FeatureFlag.Repair)
+  @Help.Group("Repair")
+  def repair: ParticipantRepairAdministration = repair_
 }
 
 class RemoteParticipantReference(environment: ConsoleEnvironment, override val name: String)
@@ -611,13 +617,13 @@ class LocalParticipantReference(
   def commitments: LocalCommitmentsAdministrationGroup = commitments_
 
   private lazy val repair_ =
-    new ParticipantRepairAdministration(consoleEnvironment, this, loggerFactory) {
+    new LocalParticipantRepairAdministration(consoleEnvironment, this, loggerFactory) {
       override protected def access[T](handler: ParticipantNode => T): T =
         LocalParticipantReference.this.access(handler)
     }
   @Help.Summary("Commands to repair the local participant contract state", FeatureFlag.Repair)
   @Help.Group("Repair")
-  def repair: ParticipantRepairAdministration = repair_
+  def repair: LocalParticipantRepairAdministration = repair_
 
   @Help.Summary("Inspect and manage parties")
   @Help.Group("Parties")

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

@@ -955,6 +955,7 @@ trait ParticipantAdministration extends FeatureFlagFilter {
         timeout: NonNegativeDuration = consoleEnvironment.commandTimeouts.bounded
     ): Unit = {
       val connected = domains.list_connected().map(_.domainId).toSet
+
       // ensure that the ledger api server has seen all packages
       try {
         AdminCommandRunner.retryUntilTrue(timeout) {

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

@@ -35,16 +35,145 @@ import io.grpc.StatusRuntimeException
 import java.time.Instant
 import scala.concurrent.{Await, Promise, TimeoutException}
 
-abstract class ParticipantRepairAdministration(
+class ParticipantRepairAdministration(
     val consoleEnvironment: ConsoleEnvironment,
     runner: AdminCommandRunner,
     val loggerFactory: NamedLoggerFactory,
 ) extends FeatureFlagFilter
     with NoTracing
     with Helpful {
 
-  protected def access[T](handler: ParticipantNode => T): T
+  @Help.Summary("Migrate contracts from one domain to another one.")
+  @Help.Description(
+    """This method can be used to migrate all the contracts associated with a domain to a new domain connection.
+         This method will register the new domain, connect to it and then re-associate all contracts on the source
+         domain to the target domain. Please note that this migration needs to be done by all participants
+         at the same time. The domain should only be used once all participants have finished their migration.
+
+         The arguments are:
+         source: the domain alias of the source domain
+         target: the configuration for the target domain
+         """
+  )
+  def migrate_domain(
+      source: DomainAlias,
+      target: DomainConnectionConfig,
+  ): Unit = {
+    consoleEnvironment.run {
+      runner.adminCommand(
+        ParticipantAdminCommands.ParticipantRepairManagement.MigrateDomain(source, target)
+      )
+    }
+  }
+
+  @Help.Summary("Download all contracts for the given set of parties to a file.")
+  @Help.Description(
+    """This command can be used to download the current active contract set of a given set of parties to a text file.
+        |This is mainly interesting for recovery and operational purposes.
+        |
+        |The file will contain base64 encoded strings, one line per contract. The lines are written
+        |sorted according to their domain and contract id. This allows to compare the contracts stored
+        |by two participants using standard file comparison tools.
+        |The domain-id is printed with the prefix domain-id before the block of contracts starts.
+        |
+        |This command may take a long time to complete and may require significant resources.
+        |It will first load the contract ids of the active contract set into memory and then subsequently
+        |load the contracts in batches and inspect their stakeholders. As this operation needs to traverse
+        |the entire datastore, it might take a long time to complete.
+        |
+        |The command will return a map of domainId -> number of active contracts stored
+        |
+        The arguments are:
+        - parties: identifying contracts having at least one stakeholder from the given set
+        - outputFile: the output file name where to store the data. Use .gz as a suffix to get a compressed file (recommended)
+        - filterDomainId: restrict the export to a given domain
+        - timestamp: optionally a timestamp for which we should take the state (useful to reconcile states of a domain)
+        - protocolVersion: optional the protocol version to use for the serialization. Defaults to the one of the domains.
+        - chunkSize: size of the byte chunks to stream back: default 1024 * 1024 * 2 = (2MB)
+        """
+  )
+  def download(
+      parties: Set[PartyId],
+      outputFile: String = ParticipantRepairAdministration.defaultFile,
+      filterDomainId: String = "",
+      timestamp: Option[Instant] = None,
+      protocolVersion: Option[ProtocolVersion] = None,
+      chunkSize: Option[PositiveInt] = None,
+  ): Unit = {
+    consoleEnvironment.run {
+      val target = File(outputFile)
+      val requestComplete = Promise[String]()
+      val observer = new GrpcByteChunksToFileObserver[AcsSnapshotChunk](
+        target,
+        requestComplete,
+      )
+      val timeout = consoleEnvironment.commandTimeouts.ledgerCommand
 
+      def call = consoleEnvironment.run {
+        runner.adminCommand(
+          ParticipantAdminCommands.ParticipantRepairManagement
+            .Download(
+              parties,
+              filterDomainId,
+              timestamp,
+              protocolVersion,
+              chunkSize,
+              observer,
+              target.toJava.getName.endsWith(".gz"),
+            )
+        )
+      }
+
+      try {
+        ResourceUtil.withResource(call) { _ =>
+          CommandSuccessful(
+            Await
+              .result(
+                requestComplete.future,
+                timeout.duration,
+              )
+              .discard
+          )
+        }
+      } catch {
+        case sre: StatusRuntimeException =>
+          GenericCommandError(
+            GrpcError("Generating acs snapshot file", "download_acs_snapshot", sre).toString
+          )
+        case _: TimeoutException =>
+          target.delete(swallowIOExceptions = true)
+          CommandErrors.ConsoleTimeout.Error(timeout.asJavaApproximation)
+      }
+    }
+  }
+
+  @Help.Summary("Import ACS snapshot", FeatureFlag.Preview)
+  @Help.Description("""Uploads a binary into the participant's ACS""")
+  def upload(
+      inputFile: String = ParticipantRepairAdministration.defaultFile
+  ): Unit = {
+    val file = File(inputFile)
+    consoleEnvironment.run {
+      runner.adminCommand(
+        ParticipantAdminCommands.ParticipantRepairManagement.Upload(
+          ByteString.copyFrom(file.loadBytes)
+        )
+      )
+    }
+  }
+}
+
+abstract class LocalParticipantRepairAdministration(
+    override val consoleEnvironment: ConsoleEnvironment,
+    runner: AdminCommandRunner,
+    override val loggerFactory: NamedLoggerFactory,
+) extends ParticipantRepairAdministration(
+      consoleEnvironment = consoleEnvironment,
+      runner = runner,
+      loggerFactory = loggerFactory,
+    ) {
+
+  protected def access[T](handler: ParticipantNode => T): T
   @Help.Summary("Add specified contracts to specific domain on local participant.")
   @Help.Description(
     """This is a last resort command to recover from data corruption, e.g. in scenarios in which participant
@@ -153,29 +282,6 @@ abstract class ParticipantRepairAdministration(
       )
     )
 
-  @Help.Summary("Migrate domain to a new version.")
-  @Help.Description(
-    """This method can be used to migrate all the contracts associated with a domain to a new domain connection.
-         This method will register the new domain, connect to it and then re-associate all contracts on the source
-         domain to the target domain. Please note that this migration needs to be done by all participants
-         at the same time. The domain should only be used once all participants have finished their migration.
-
-         The arguments are:
-         source: the domain alias of the source domain
-         target: the configuration for the target domain
-         """
-  )
-  def migrate_domain(
-      source: DomainAlias,
-      target: DomainConnectionConfig,
-  ): Unit = {
-    consoleEnvironment.run {
-      runner.adminCommand(
-        ParticipantAdminCommands.ParticipantRepairManagement.MigrateDomain(source, target)
-      )
-    }
-  }
-
   @Help.Summary("Mark sequenced events as ignored.")
   @Help.Description(
     """This is the last resort to ignore events that the participant is unable to process.
@@ -199,7 +305,9 @@ abstract class ParticipantRepairAdministration(
       force: Boolean = false,
   ): Unit =
     runRepairCommand(tc =>
-      access { _.sync.repairService.ignoreEvents(domainId, from, to, force)(tc) }
+      access {
+        _.sync.repairService.ignoreEvents(domainId, from, to, force)(tc)
+      }
     )
 
   @Help.Summary("Remove the ignored status from sequenced events.")
@@ -222,105 +330,10 @@ abstract class ParticipantRepairAdministration(
       force: Boolean = false,
   ): Unit =
     runRepairCommand(tc =>
-      access { _.sync.repairService.unignoreEvents(domainId, from, to, force)(tc) }
-    )
-
-  @Help.Summary("Download all contracts for the given set of parties to a file.")
-  @Help.Description(
-    """This command can be used to download the current active contract set of a given set of parties to a text file.
-        |This is mainly interesting for recovery and operational purposes.
-        |
-        |The file will contain base64 encoded strings, one line per contract. The lines are written
-        |sorted according to their domain and contract id. This allows to compare the contracts stored
-        |by two participants using standard file comparison tools.
-        |The domain-id is printed with the prefix domain-id before the block of contracts starts.
-        |
-        |This command may take a long time to complete and may require significant resources.
-        |It will first load the contract ids of the active contract set into memory and then subsequently
-        |load the contracts in batches and inspect their stakeholders. As this operation needs to traverse
-        |the entire datastore, it might take a long time to complete.
-        |
-        |The command will return a map of domainId -> number of active contracts stored
-        |
-        The arguments are:
-        - parties: identifying contracts having at least one stakeholder from the given set
-        - outputFile: the output file name where to store the data. Use .gz as a suffix to get a compressed file (recommended)
-        - filterDomainId: restrict the export to a given domain
-        - timestamp: optionally a timestamp for which we should take the state (useful to reconcile states of a domain)
-        - protocolVersion: optional the protocol version to use for the serialization. Defaults to the one of the domains.
-        - chunkSize: size of the byte chunks to stream back: default 1024 * 1024 * 2 = (2MB)
-        """
-  )
-  def download(
-      parties: Set[PartyId],
-      outputFile: String = ParticipantRepairAdministration.defaultFile,
-      filterDomainId: String = "",
-      timestamp: Option[Instant] = None,
-      protocolVersion: Option[ProtocolVersion] = None,
-      chunkSize: Option[PositiveInt] = None,
-  ): Unit = {
-    consoleEnvironment.run {
-      val target = File(outputFile)
-      val requestComplete = Promise[String]()
-      val observer = new GrpcByteChunksToFileObserver[AcsSnapshotChunk](
-        target,
-        requestComplete,
-      )
-      val timeout = consoleEnvironment.commandTimeouts.ledgerCommand
-
-      def call = consoleEnvironment.run {
-        runner.adminCommand(
-          ParticipantAdminCommands.ParticipantRepairManagement
-            .Download(
-              parties,
-              filterDomainId,
-              timestamp,
-              protocolVersion,
-              chunkSize,
-              observer,
-              target.toJava.getName.endsWith(".gz"),
-            )
-        )
-      }
-
-      try {
-        ResourceUtil.withResource(call) { _ =>
-          CommandSuccessful(
-            Await
-              .result(
-                requestComplete.future,
-                timeout.duration,
-              )
-              .discard
-          )
-        }
-      } catch {
-        case sre: StatusRuntimeException =>
-          GenericCommandError(
-            GrpcError("Generating acs snapshot file", "download_acs_snapshot", sre).toString
-          )
-        case _: TimeoutException =>
-          target.delete(swallowIOExceptions = true)
-          CommandErrors.ConsoleTimeout.Error(timeout.asJavaApproximation)
+      access {
+        _.sync.repairService.unignoreEvents(domainId, from, to, force)(tc)
       }
-    }
-  }
-
-  @Help.Summary("Import ACS snapshot", FeatureFlag.Preview)
-  @Help.Description("""Uploads a binary into the participant's ACS""")
-  def upload(
-      inputFile: String = ParticipantRepairAdministration.defaultFile
-  ): Unit = {
-    val file = File(inputFile)
-    consoleEnvironment.run {
-      runner.adminCommand(
-        ParticipantAdminCommands.ParticipantRepairManagement.Upload(
-          ByteString.copyFrom(file.loadBytes)
-        )
-      )
-    }
-  }
-
+    )
 }
 
 object ParticipantRepairAdministration {

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

@@ -148,12 +148,13 @@ class TopologyAdministrationGroupX(
 
       val thisNodeRootKey = Some(instance.id.uid.namespace.fingerprint)
 
+      val codes = Set(NamespaceDelegationX.code, OwnerToKeyMappingX.code)
       // provide the root namespace delegation and owner to key mapping
       val namespace = instance.topology.transactions
         .list(filterAuthorizedKey = thisNodeRootKey)
         .result
         .map(_.transaction)
-        .filter(_.transaction.mapping.code == NamespaceDelegationX.code)
+        .filter(x => codes.contains(x.transaction.mapping.code))
 
       // create and sign the unionspace for the domain
       val unionspaceTransaction = instance.topology.unionspaces.propose(

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

@@ -298,6 +298,7 @@ trait Environment extends NamedLogging with AutoCloseable with NoTracing {
   private def runningNodes: Seq[CantonNodeBootstrap[CantonNode]] = allNodes.flatMap(_.running)
 
   private def autoConnectLocalNodes(): Either[StartupError, Unit] = {
+    // TODO(#11255) extend this to x-nodes
     val activeDomains = domains.running
       .filter(_.isActive)
       .filter(_.config.topology.open)
@@ -416,7 +417,7 @@ trait Environment extends NamedLogging with AutoCloseable with NoTracing {
         }
       } yield ()
     }
-
+    // TODO(#11255) implement the same for participantsX
     participants.running.traverse_(reconnect)
   }
 

community/app/src/pack/examples/06-messaging/contact/daml.yaml

@@ -1,4 +1,4 @@
-sdk-version: 2.7.0-snapshot.20230417.11682.0.ac68bdf4
+sdk-version: 2.7.0-snapshot.20230424.11710.0.2f27f987
 sandbox-options: 
 - --wall-clock-time
 name: contact

community/app/src/pack/examples/06-messaging/message/daml.yaml

@@ -1,4 +1,4 @@
-sdk-version: 2.7.0-snapshot.20230417.11682.0.ac68bdf4
+sdk-version: 2.7.0-snapshot.20230424.11710.0.2f27f987
 sandbox-options: 
 - --wall-clock-time
 name: message