Bump Canton fork to 2025-06-23 (#1192)

* Undo our changes

Signed-off-by: Moritz Kiefer <[email protected]>

* Bump Canton commit

Signed-off-by: Moritz Kiefer <[email protected]>

* Reapply our changes

Signed-off-by: Moritz Kiefer <[email protected]>

* Resolve conflicts

[ci]

Signed-off-by: Moritz Kiefer <[email protected]>

* bump commit

[ci]

Signed-off-by: Moritz Kiefer <[email protected]>

* add headers

[ci]

Signed-off-by: Moritz Kiefer <[email protected]>

---------

Signed-off-by: Moritz Kiefer <[email protected]>
Co-authored-by: Moritz Kiefer <[email protected]>

Author: moritzkiefer-da

MAINTENANCE.md

@@ -39,7 +39,7 @@ Initial setup:
 1. Check out the [Canton **Open Source** repo](https://github.com/digital-asset/canton)
 2. Define the environment variable used in the commands below using `export PATH_TO_CANTON_OSS=<your-canton-oss-repo-path>`. This can be added to your private env vars.
 
-Current Canton commit: `36118b5abce6d21ab6cd8d47daeff8a1c584132b`
+Current Canton commit: `a0f2007fa5c9f9ecf1d58a4be564ceaa3158ebb3`
 
 1. Checkout the **current Canton commit listed above** in the Canton open source repo from above, so we can diff our current fork against this checkout.
 2. Change to your checkout of the Splice repo and execute the following steps:

canton/UNRELEASED.md

@@ -9,6 +9,18 @@ schedule, i.e. if you add an entry effective at or after the first
 header, prepend the new date header that corresponds to the
 Wednesday after your change.
 
+## Until 2025-06-25 (Exclusive)
+- Adds new gRPC endpoint `GetHighestOffsetByTimestamp` (and console command `find_highest_offset_by_timestamp`) that
+  for a given timestamp, finds the highest ledger offset among all events that have record time <= timestamp. This is a
+  backward-compatible change, because it's an addition only. It's useful for party replication / major upgrade.
+
+## Until 2025-06-11 (Exclusive)
+- Dead parameter `canton.participants.<participant>.http-ledger-api.allow-insecure-tokens` has been removed.
+
+## Until 2025-06-04 (Exclusive)
+- **Breaking** The console command `connect_local_bft` takes now a list of `SequencerReference` instead of a `NonEmpty[Map[SequencerAlias, SequencerReference]]`
+- Console command - A new console command `connect_bft` has been added to connect by url to Decentralized Sequencers
+-
 ## Until 2025-05-14 (Exclusive)
 - JSON - changes in openapi (`Any` renamed as `ProtoAny`, `Event1` renamed to `TopologyEvent` and fixed, fixed `Field`, `FieldMask`,`JsReassignmentEvent` mappings.
 - JSON API - fixed openapi documentation for maps: (`eventsById`,`filtersByParty`).

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

@@ -28,11 +28,19 @@ service PartyManagementService {
   // successfully.
   rpc GetAddPartyStatus(GetAddPartyStatusRequest) returns (GetAddPartyStatusResponse);
 
-  // Export the ACS for the given parties from the participant
+  // Export the ACS for the given parties from the participant.
   rpc ExportAcs(ExportAcsRequest) returns (stream ExportAcsResponse);
 
-  // Export the ACS for the given parties at a timestamp (that is the effective time of a topology transaction)
+  // Export the ACS for the given parties at a timestamp (that is the effective time of a topology transaction).
   rpc ExportAcsAtTimestamp(ExportAcsAtTimestampRequest) returns (stream ExportAcsAtTimestampResponse);
+
+  // For a given timestamp, find the highest ledger offset among all events that have record time <= timestamp.
+  //
+  // Returns a ledger offset, or an error otherwise. Depending on the error cause, a retry may make sense.
+  // Retryable errors are defined as: OUT_OF_RANGE/INVALID_TIMESTAMP_PARTY_MANAGEMENT_ERROR.
+  // Further, a returned offset is guaranteed to be "clean", meaning all events have been processed fully and
+  // published to the Ledger API DB until the requested timestamp.
+  rpc GetHighestOffsetByTimestamp(GetHighestOffsetByTimestampRequest) returns (GetHighestOffsetByTimestampResponse);
 }
 
 message AddPartyAsyncRequest {
@@ -213,3 +221,29 @@ message ExportAcsAtTimestampResponse {
   // Required
   bytes chunk = 1;
 }
+
+// Requests the highest ledger offset among all events belonging to the synchronizer (`synchronizer_id`)
+// that have a record time before or at the given `timestamp`.
+//
+// This endpoint features a `force` message field. This is intended for disaster recovery scenarios only.
+message GetHighestOffsetByTimestampRequest {
+  // The identifier of the synchronizer.
+  // Required
+  string synchronizer_id = 1;
+
+  // The requested timestamp for which a ledger offset should be found.
+  // Required
+  google.protobuf.Timestamp timestamp = 2;
+
+  // If true, the ledger end offset is requested instead of the highest ledger offset among events
+  // with a record time at or before the requested `timestamp`.
+  // Required, defaults to false.
+  bool force = 3;
+}
+
+message GetHighestOffsetByTimestampResponse {
+  // The highest ledger offset among events that have their record time before or at the requested timestamp.
+  // An error when no such offset (yet) exists.
+  // Required
+  int64 ledger_offset = 1;
+}

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

@@ -537,6 +537,42 @@ object ParticipantAdminCommands {
       ): Either[String, AddPartyStatus] = AddPartyStatus.fromProtoV30(response).leftMap(_.toString)
     }
 
+    final case class GetHighestOffsetByTimestamp(
+        synchronizerId: SynchronizerId,
+        timestamp: Instant,
+        force: Boolean,
+    ) extends GrpcAdminCommand[
+          v30.GetHighestOffsetByTimestampRequest,
+          v30.GetHighestOffsetByTimestampResponse,
+          NonNegativeLong,
+        ] {
+      override type Svc = PartyManagementServiceStub
+
+      override def createService(channel: ManagedChannel): PartyManagementServiceStub =
+        v30.PartyManagementServiceGrpc.stub(channel)
+
+      override protected def createRequest()
+          : Either[String, v30.GetHighestOffsetByTimestampRequest] =
+        Right(
+          v30.GetHighestOffsetByTimestampRequest(
+            synchronizerId.toProtoPrimitive,
+            Some(Timestamp(timestamp)),
+            force,
+          )
+        )
+
+      override protected def submitRequest(
+          service: PartyManagementServiceStub,
+          request: v30.GetHighestOffsetByTimestampRequest,
+      ): Future[v30.GetHighestOffsetByTimestampResponse] =
+        service.getHighestOffsetByTimestamp(request)
+
+      override protected def handleResponse(
+          response: v30.GetHighestOffsetByTimestampResponse
+      ): Either[String, NonNegativeLong] =
+        NonNegativeLong.create(response.ledgerOffset).leftMap(_.toString)
+    }
+
     final case class ExportAcs(
         parties: Set[PartyId],
         filterSynchronizerId: Option[SynchronizerId],

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

@@ -20,10 +20,7 @@ import com.digitalasset.canton.topology.*
 import com.digitalasset.canton.topology.admin.grpc.{BaseQuery, TopologyStoreId}
 import com.digitalasset.canton.topology.admin.v30
 import com.digitalasset.canton.topology.admin.v30.*
-import com.digitalasset.canton.topology.admin.v30.AuthorizeRequest.Type.{
-  Proposal,
-  TransactionHashBytes,
-}
+import com.digitalasset.canton.topology.admin.v30.AuthorizeRequest.Type.{Proposal, TransactionHash}
 import com.digitalasset.canton.topology.admin.v30.IdentityInitializationServiceGrpc.IdentityInitializationServiceStub
 import com.digitalasset.canton.topology.admin.v30.TopologyAggregationServiceGrpc.TopologyAggregationServiceStub
 import com.digitalasset.canton.topology.admin.v30.TopologyManagerReadServiceGrpc.TopologyManagerReadServiceStub
@@ -915,7 +912,7 @@ object TopologyAdminCommands {
     }
 
     final case class Authorize[M <: TopologyMapping: ClassTag](
-        transactionHash: ByteString,
+        transactionHash: String,
         mustFullyAuthorize: Boolean,
         signedBy: Seq[Fingerprint],
         store: TopologyStoreId,
@@ -928,7 +925,7 @@ object TopologyAdminCommands {
 
       override protected def createRequest(): Either[String, AuthorizeRequest] = Right(
         AuthorizeRequest(
-          TransactionHashBytes(transactionHash),
+          TransactionHash(transactionHash),
           mustFullyAuthorize = mustFullyAuthorize,
           forceChanges = Seq.empty,
           signedBy = signedBy.map(_.toProtoPrimitive),

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

@@ -77,6 +77,7 @@ object CommunityConfigValidations extends ConfigValidations with NamedLogging {
       eitherUserListsOrPrivilegedTokensOnParticipants,
       sessionSigningKeysOnlyWithKms,
       distinctScopesAndAudiencesOnAuthServices,
+      engineAdditionalConsistencyChecksParticipants,
     )
 
   /** Group node configs by db access to find matching db storage configs. Overcomplicated types
@@ -250,6 +251,19 @@ object CommunityConfigValidations extends ConfigValidations with NamedLogging {
     toValidated(errors)
   }
 
+  private def engineAdditionalConsistencyChecksParticipants(
+      config: CantonConfig
+  ): Validated[NonEmpty[Seq[String]], Unit] = {
+    val errors = config.participants.toSeq.mapFilter { case (name, participantConfig) =>
+      Option.when(
+        participantConfig.parameters.engine.enableAdditionalConsistencyChecks && !config.parameters.nonStandardConfig
+      )(
+        s"Enabling additional consistency checks on the Daml Engine for participant ${name.unwrap} requires to explicitly set canton.parameters.non-standard-config = true"
+      )
+    }
+    toValidated(errors)
+  }
+
   private def sessionSigningKeysOnlyWithKms(
       config: CantonConfig
   ): Validated[NonEmpty[Seq[String]], Unit] = {

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

@@ -90,43 +90,37 @@ object AmmoniteCacheLock {
   ): Either[Throwable, Option[AmmoniteCacheLock]] = blocking(synchronized {
     try {
       val myLockFile = path / "lock"
-      if (myLockFile.toIO.exists()) {
-        Right(None)
-      } else {
-        logger.debug(s"Attempting to obtain lock $myLockFile")
-        val out = new RandomAccessFile(myLockFile.toIO, "rw")
-        Option(out.getChannel.tryLock()) match {
-          case None =>
-            logger.debug(s"Failed to acquire lock for $myLockFile")
-            out.close()
-            Right(None)
-          case Some(lock) =>
-            myLockFile.toIO.deleteOnExit()
-            Right(Some(new AmmoniteCacheLock {
-              override def release(): Unit =
-                try {
-                  logger.debug(s"Releasing lock $myLockFile...")
-                  lock.release()
-                  out.close()
-                  if (!Files.deleteIfExists(myLockFile.toNIO)) { // throws when the file cannot be deleted
-                    logger.warn(s"Failed to delete lock file $myLockFile because it did not exist")
-                  }
-                } catch {
-                  case NonFatal(e) =>
-                    logger.error(s"Releasing ammonite cache lock $lockFile failed", e)
+      logger.debug(s"Attempting to obtain lock $myLockFile")
+      val out = new RandomAccessFile(myLockFile.toIO, "rw")
+      Option(out.getChannel.tryLock()) match {
+        case None =>
+          logger.debug(s"Failed to acquire lock for $myLockFile")
+          out.close()
+          Right(None)
+        case Some(lock) =>
+          Right(Some(new AmmoniteCacheLock {
+            override def release(): Unit =
+              try {
+                logger.debug(s"Releasing lock $myLockFile...")
+                lock.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")
                 }
+              } catch {
+                case NonFatal(e) =>
+                  logger.error(s"Releasing ammonite cache lock $lockFile failed", e)
+              }
 
-              override val storage: Storage = new Storage.Folder(path, isRepl = isRepl)
+            override val storage: Storage = new Storage.Folder(path, isRepl = isRepl)
 
-              override def toString: String = s"file cache at $path"
+            override def toString: String = s"file cache at $path"
 
-              override def lockFile: Option[File] = Some(myLockFile.toIO)
-            }))
-
-        }
+            override def lockFile: Option[File] = Some(myLockFile.toIO)
+          }))
       }
     } catch {
-      case e: OverlappingFileLockException => Right(None)
+      case _: OverlappingFileLockException => Right(None)
       case NonFatal(e) => Left(e)
     }
   })

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

@@ -31,7 +31,7 @@ import com.digitalasset.canton.time.SimClock
 import com.digitalasset.canton.topology.admin.grpc.TopologyStoreId
 import com.digitalasset.canton.topology.{ParticipantId, PartyId, SynchronizerId}
 import com.digitalasset.canton.tracing.{NoTracing, TraceContext}
-import com.digitalasset.canton.{LfPartyId, SynchronizerAlias}
+import com.digitalasset.canton.{LfPartyId, SequencerAlias, SynchronizerAlias}
 import com.digitalasset.daml.lf.data.Ref.PackageId
 import com.typesafe.scalalogging.Logger
 import io.opentelemetry.api.trace.Tracer
@@ -570,6 +570,9 @@ object ConsoleEnvironment {
     implicit def toGrpcSequencerConnection(connection: String): SequencerConnection =
       GrpcSequencerConnection.tryCreate(connection)
 
+    implicit def toSequencerAlias(alias: String): SequencerAlias =
+      SequencerAlias.tryCreate(alias)
+
     implicit def toSequencerConnections(connection: String): SequencerConnections =
       SequencerConnections.single(GrpcSequencerConnection.tryCreate(connection))
 

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

@@ -81,6 +81,7 @@ object ConsoleEnvironmentBinding {
          |import com.digitalasset.canton.sequencing.SequencerConnection
          |import com.digitalasset.canton.sequencing.SequencerConnections
          |import com.digitalasset.canton.sequencing.SequencerConnectionValidation._
+         |import com.digitalasset.canton.sequencing.SubmissionRequestAmplification
          |import com.digitalasset.canton.sequencing.GrpcSequencerConnection
          |""".stripMargin