Update 2024-03-19.23 (#146)

Reference commit: 6ad9b204b1

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

community/admin-api/src/main/protobuf/com/digitalasset/canton/admin/domain/v30/sequencer_connection.proto

@@ -24,6 +24,17 @@ message SequencerConnection {
   }
 }
 
+enum SequencerConnectionValidation {
+  UNKNOWN = 0;
+  // Do not validate the sequencer connection
+  DISABLED = 1;
+  // Validate only the ones we could reach
+  ACTIVE = 2;
+  // Validate all the connections
+  ALL = 3;
+
+}
+
 message SequencerConnections {
   option (scalapb.message).companion_extends = "com.digitalasset.canton.version.StorageProtoVersion";
 

community/admin-api/src/main/protobuf/com/digitalasset/canton/admin/participant/v30/domain_connectivity.proto

@@ -66,15 +66,25 @@ message ReconnectDomainsResponse {}
 message RegisterDomainRequest {
   DomainConnectionConfig add = 1;
 
-  // If true, only performs the handshake but does not connect to the domain
+  // If true, only performs the handshake but does not establish an active connection to the domain
   // Note that in that case domain connection will need to be performed subsequently
+  // This will invoke the "handshake" with the sequencer_connect_service and persist the static
+  // domain parameters for the given domain connection.
+  // This is useful during major version upgrades where we need to locally initialize the domain
+  // connection, but don't want to start processing contracts before we have imported the ACS.
   bool handshake_only = 2;
+
+  // Determines how well the provided sequencer connections should be validated before they are
+  // persisted. The more paranoid the validation, the higher the chance of the command failing,
+  // as it will require the sequencer to be online and responsive.
+  com.digitalasset.canton.admin.domain.v30.SequencerConnectionValidation sequencer_connection_validation = 3;
 }
 
 message RegisterDomainResponse {}
 
 message ModifyDomainRequest {
   DomainConnectionConfig modify = 1;
+  com.digitalasset.canton.admin.domain.v30.SequencerConnectionValidation sequencer_connection_validation = 2;
 }
 
 message ModifyDomainResponse {}

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

@@ -12,13 +12,13 @@ import com.digitalasset.canton.admin.api.client.commands.GrpcAdminCommand.{
 import com.digitalasset.canton.admin.pruning.v30.LocatePruningTimestamp
 import com.digitalasset.canton.config.RequireTypes.PositiveInt
 import com.digitalasset.canton.data.CantonTimestamp
-import com.digitalasset.canton.domain.admin.v30
 import com.digitalasset.canton.domain.mediator.admin.gprc.{
   InitializeMediatorRequestX,
   InitializeMediatorResponseX,
 }
+import com.digitalasset.canton.mediator.admin.v30
 import com.digitalasset.canton.protocol.StaticDomainParameters
-import com.digitalasset.canton.sequencing.SequencerConnections
+import com.digitalasset.canton.sequencing.{SequencerConnectionValidation, SequencerConnections}
 import com.digitalasset.canton.topology.DomainId
 import io.grpc.ManagedChannel
 
@@ -47,6 +47,7 @@ object EnterpriseMediatorAdministrationCommands {
       domainId: DomainId,
       domainParameters: StaticDomainParameters,
       sequencerConnections: SequencerConnections,
+      validation: SequencerConnectionValidation,
   ) extends BaseMediatorXInitializationCommand[
         v30.InitializeMediatorRequest,
         v30.InitializeMediatorResponse,
@@ -58,6 +59,7 @@ object EnterpriseMediatorAdministrationCommands {
           domainId,
           domainParameters,
           sequencerConnections,
+          validation,
         ).toProtoV30
       )
 

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

@@ -12,14 +12,10 @@ import com.digitalasset.canton.admin.api.client.commands.GrpcAdminCommand.{
 import com.digitalasset.canton.admin.pruning.v30.LocatePruningTimestamp
 import com.digitalasset.canton.config.RequireTypes.PositiveInt
 import com.digitalasset.canton.data.CantonTimestamp
-import com.digitalasset.canton.domain.admin.v30
-import com.digitalasset.canton.domain.sequencing.admin.grpc.{
-  InitializeSequencerRequestX,
-  InitializeSequencerResponseX,
-}
+import com.digitalasset.canton.domain.sequencing.admin.grpc.InitializeSequencerResponse
 import com.digitalasset.canton.domain.sequencing.sequencer.SequencerSnapshot
-import com.digitalasset.canton.topology.Member
-import com.digitalasset.canton.topology.store.StoredTopologyTransactionsX.GenericStoredTopologyTransactionsX
+import com.digitalasset.canton.sequencer.admin.v30
+import com.digitalasset.canton.topology.{Member, SequencerId}
 import com.google.protobuf.ByteString
 import io.grpc.ManagedChannel
 
@@ -47,14 +43,11 @@ object EnterpriseSequencerAdminCommands {
       v30.SequencerPruningAdministrationServiceGrpc.stub(channel)
   }
 
-  final case class InitializeX(
-      topologySnapshot: GenericStoredTopologyTransactionsX,
-      domainParameters: com.digitalasset.canton.protocol.StaticDomainParameters,
-      sequencerSnapshot: Option[SequencerSnapshot],
-  ) extends GrpcAdminCommand[
-        v30.InitializeSequencerRequest,
-        v30.InitializeSequencerResponse,
-        InitializeSequencerResponseX,
+  final case class InitializeFromOnboardingState(onboardingState: ByteString)
+      extends GrpcAdminCommand[
+        v30.InitializeSequencerFromOnboardingStateRequest,
+        v30.InitializeSequencerFromOnboardingStateResponse,
+        InitializeSequencerResponse,
       ] {
     override type Svc = v30.SequencerInitializationServiceGrpc.SequencerInitializationServiceStub
 
@@ -65,32 +58,28 @@ object EnterpriseSequencerAdminCommands {
 
     override def submitRequest(
         service: v30.SequencerInitializationServiceGrpc.SequencerInitializationServiceStub,
-        request: v30.InitializeSequencerRequest,
-    ): Future[v30.InitializeSequencerResponse] =
-      service.initializeSequencer(request)
+        request: v30.InitializeSequencerFromOnboardingStateRequest,
+    ): Future[v30.InitializeSequencerFromOnboardingStateResponse] =
+      service.initializeSequencerFromOnboardingState(request)
 
-    override def createRequest(): Either[String, v30.InitializeSequencerRequest] =
+    override def createRequest()
+        : Either[String, v30.InitializeSequencerFromOnboardingStateRequest] =
       Right(
-        InitializeSequencerRequestX(
-          topologySnapshot,
-          domainParameters,
-          sequencerSnapshot,
-        ).toProtoV30
+        v30.InitializeSequencerFromOnboardingStateRequest(onboardingState)
       )
 
     override def handleResponse(
-        response: v30.InitializeSequencerResponse
-    ): Either[String, InitializeSequencerResponseX] =
-      InitializeSequencerResponseX.fromProtoV30(response).leftMap(_.toString)
+        response: v30.InitializeSequencerFromOnboardingStateResponse
+    ): Either[String, InitializeSequencerResponse] =
+      Right(InitializeSequencerResponse(response.replicated))
   }
-  final case class Initialize(
+  final case class InitializeFromGenesisState(
       topologySnapshot: ByteString,
       domainParameters: com.digitalasset.canton.protocol.StaticDomainParameters,
-      sequencerSnapshot: ByteString,
   ) extends GrpcAdminCommand[
-        v30.InitializeSequencerVersionedRequest,
-        v30.InitializeSequencerVersionedResponse,
-        InitializeSequencerResponseX,
+        v30.InitializeSequencerFromGenesisStateRequest,
+        v30.InitializeSequencerFromGenesisStateResponse,
+        InitializeSequencerResponse,
       ] {
     override type Svc = v30.SequencerInitializationServiceGrpc.SequencerInitializationServiceStub
 
@@ -101,23 +90,22 @@ object EnterpriseSequencerAdminCommands {
 
     override def submitRequest(
         service: v30.SequencerInitializationServiceGrpc.SequencerInitializationServiceStub,
-        request: v30.InitializeSequencerVersionedRequest,
-    ): Future[v30.InitializeSequencerVersionedResponse] =
-      service.initializeSequencerVersioned(request)
+        request: v30.InitializeSequencerFromGenesisStateRequest,
+    ): Future[v30.InitializeSequencerFromGenesisStateResponse] =
+      service.initializeSequencerFromGenesisState(request)
 
-    override def createRequest(): Either[String, v30.InitializeSequencerVersionedRequest] =
+    override def createRequest(): Either[String, v30.InitializeSequencerFromGenesisStateRequest] =
       Right(
-        v30.InitializeSequencerVersionedRequest(
+        v30.InitializeSequencerFromGenesisStateRequest(
           topologySnapshot = topologySnapshot,
           Some(domainParameters.toProtoV30),
-          sequencerSnapshot,
         )
       )
 
     override def handleResponse(
-        response: v30.InitializeSequencerVersionedResponse
-    ): Either[String, InitializeSequencerResponseX] =
-      Right(InitializeSequencerResponseX(response.replicated))
+        response: v30.InitializeSequencerFromGenesisStateResponse
+    ): Either[String, InitializeSequencerResponse] =
+      Right(InitializeSequencerResponse(response.replicated))
   }
 
   final case class Snapshot(timestamp: CantonTimestamp)
@@ -155,6 +143,85 @@ object EnterpriseSequencerAdminCommands {
     override def timeoutType: TimeoutType = DefaultUnboundedTimeout
   }
 
+  final case class OnboardingState(memberOrTimestamp: Either[SequencerId, CantonTimestamp])
+      extends BaseSequencerAdministrationCommand[
+        v30.OnboardingStateRequest,
+        v30.OnboardingStateResponse,
+        ByteString,
+      ] {
+    override def createRequest(): Either[String, v30.OnboardingStateRequest] = {
+      Right(
+        v30.OnboardingStateRequest(request =
+          memberOrTimestamp.fold[v30.OnboardingStateRequest.Request](
+            member => v30.OnboardingStateRequest.Request.SequencerId(member.toProtoPrimitive),
+            timestamp => v30.OnboardingStateRequest.Request.Timestamp(timestamp.toProtoTimestamp),
+          )
+        )
+      )
+    }
+
+    override def submitRequest(
+        service: v30.SequencerAdministrationServiceGrpc.SequencerAdministrationServiceStub,
+        request: v30.OnboardingStateRequest,
+    ): Future[v30.OnboardingStateResponse] = service.onboardingState(request)
+
+    override def handleResponse(
+        response: v30.OnboardingStateResponse
+    ): Either[String, ByteString] =
+      response.value match {
+        case v30.OnboardingStateResponse.Value
+              .Failure(v30.OnboardingStateResponse.Failure(reason)) =>
+          Left(reason)
+        case v30.OnboardingStateResponse.Value
+              .Success(
+                v30.OnboardingStateResponse.Success(onboardingState)
+              ) =>
+          Right(onboardingState)
+        case _ => Left("response is empty")
+      }
+
+    //  command will potentially take a long time
+    override def timeoutType: TimeoutType = DefaultUnboundedTimeout
+  }
+
+  final case class GenesisState(
+      timestamp: Option[CantonTimestamp]
+  ) extends BaseSequencerAdministrationCommand[
+        v30.GenesisStateRequest,
+        v30.GenesisStateResponse,
+        ByteString,
+      ] {
+    override def createRequest(): Either[String, v30.GenesisStateRequest] =
+      Right(
+        v30.GenesisStateRequest(
+          timestamp = timestamp.map(_.toProtoTimestamp)
+        )
+      )
+
+    override def submitRequest(
+        service: v30.SequencerAdministrationServiceGrpc.SequencerAdministrationServiceStub,
+        request: v30.GenesisStateRequest,
+    ): Future[v30.GenesisStateResponse] = service.genesisState(request)
+
+    override def handleResponse(
+        response: v30.GenesisStateResponse
+    ): Either[String, ByteString] =
+      response.value match {
+        case v30.GenesisStateResponse.Value
+              .Failure(v30.GenesisStateResponse.Failure(reason)) =>
+          Left(reason)
+        case v30.GenesisStateResponse.Value
+              .Success(
+                v30.GenesisStateResponse.Success(genesisState)
+              ) =>
+          Right(genesisState)
+        case _ => Left("response is empty")
+      }
+
+    //  command will potentially take a long time
+    override def timeoutType: TimeoutType = DefaultUnboundedTimeout
+  }
+
   final case class Prune(timestamp: CantonTimestamp)
       extends BaseSequencerPruningAdministrationCommand[
         v30.SequencerPruning.PruneRequest,

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

@@ -5,8 +5,8 @@ package com.digitalasset.canton.admin.api.client.commands
 
 import cats.implicits.toTraverseOps
 import cats.syntax.either.*
-import com.digitalasset.canton.domain.admin.v30
-import com.digitalasset.canton.sequencing.SequencerConnections
+import com.digitalasset.canton.mediator.admin.v30
+import com.digitalasset.canton.sequencing.{SequencerConnectionValidation, SequencerConnections}
 import io.grpc.ManagedChannel
 
 import scala.concurrent.Future
@@ -47,8 +47,10 @@ object EnterpriseSequencerConnectionAdminCommands {
     }
   }
 
-  final case class SetConnection(connections: SequencerConnections)
-      extends BaseSequencerConnectionAdminCommand[
+  final case class SetConnection(
+      connections: SequencerConnections,
+      validation: SequencerConnectionValidation,
+  ) extends BaseSequencerConnectionAdminCommand[
         v30.SetConnectionRequest,
         v30.SetConnectionResponse,
         Unit,
@@ -59,7 +61,10 @@ object EnterpriseSequencerConnectionAdminCommands {
     ): Future[v30.SetConnectionResponse] = service.setConnection(request)
 
     override def createRequest(): Either[String, v30.SetConnectionRequest] = Right(
-      v30.SetConnectionRequest(Some(connections.toProtoV30))
+      v30.SetConnectionRequest(
+        Some(connections.toProtoV30),
+        sequencerConnectionValidation = validation.toProtoV30,
+      )
     )
 
     override def handleResponse(response: v30.SetConnectionResponse): Either[String, Unit] =

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

@@ -40,6 +40,7 @@ import com.digitalasset.canton.participant.admin.grpc.{
 import com.digitalasset.canton.participant.domain.DomainConnectionConfig as CDomainConnectionConfig
 import com.digitalasset.canton.participant.sync.UpstreamOffsetConvert
 import com.digitalasset.canton.protocol.LfContractId
+import com.digitalasset.canton.sequencing.SequencerConnectionValidation
 import com.digitalasset.canton.serialization.ProtoConverter.InstantConverter
 import com.digitalasset.canton.topology.{DomainId, PartyId}
 import com.digitalasset.canton.tracing.TraceContext
@@ -736,14 +737,18 @@ object ParticipantAdminCommands {
       }
     }
 
-    final case class RegisterDomain(config: CDomainConnectionConfig, handshakeOnly: Boolean)
-        extends Base[RegisterDomainRequest, RegisterDomainResponse, Unit] {
+    final case class RegisterDomain(
+        config: CDomainConnectionConfig,
+        handshakeOnly: Boolean,
+        sequencerConnectionValidation: SequencerConnectionValidation,
+    ) extends Base[RegisterDomainRequest, RegisterDomainResponse, Unit] {
 
       override def createRequest(): Either[String, RegisterDomainRequest] =
         Right(
           RegisterDomainRequest(
             add = Some(config.toProtoV30),
             handshakeOnly = handshakeOnly,
+            sequencerConnectionValidation = sequencerConnectionValidation.toProtoV30,
           )
         )
 
@@ -761,11 +766,18 @@ object ParticipantAdminCommands {
 
     }
 
-    final case class ModifyDomainConnection(config: CDomainConnectionConfig)
-        extends Base[ModifyDomainRequest, ModifyDomainResponse, Unit] {
+    final case class ModifyDomainConnection(
+        config: CDomainConnectionConfig,
+        sequencerConnectionValidation: SequencerConnectionValidation,
+    ) extends Base[ModifyDomainRequest, ModifyDomainResponse, Unit] {
 
       override def createRequest(): Either[String, ModifyDomainRequest] =
-        Right(ModifyDomainRequest(modify = Some(config.toProtoV30)))
+        Right(
+          ModifyDomainRequest(
+            modify = Some(config.toProtoV30),
+            sequencerConnectionValidation = sequencerConnectionValidation.toProtoV30,
+          )
+        )
 
       override def submitRequest(
           service: DomainConnectivityServiceStub,

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

@@ -7,9 +7,9 @@ import cats.syntax.either.*
 import cats.syntax.traverse.*
 import com.digitalasset.canton.config.RequireTypes.{NonNegativeLong, PositiveInt}
 import com.digitalasset.canton.data.CantonTimestamp
-import com.digitalasset.canton.domain.admin
 import com.digitalasset.canton.domain.sequencing.sequencer.SequencerPruningStatus
 import com.digitalasset.canton.domain.sequencing.sequencer.traffic.SequencerTrafficStatus
+import com.digitalasset.canton.sequencer.admin
 import com.digitalasset.canton.topology.Member
 import io.grpc.ManagedChannel