[main] Update 2025-11-04.22 (#399)

Reference commit: 98c1480bfc

Author: canton-machine

UNRELEASED.md

@@ -4,6 +4,7 @@ Canton CANTON_VERSION has been released on RELEASE_DATE.
 
 ## Summary
 
+
 _Write summary of release_
 
 ## What’s New
@@ -13,14 +14,19 @@ Template for a bigger topic
 
 #### Background
 #### Specific Changes
+
 #### Impact and Migration
 
 ### Minor Improvements
+- Ledger JSON Api changes:
+  - extra fields in JSON objects are no longer tolerated,
+  - All JSON values are optional by default upon decoding (this is not reflected in the openapi spec yet, but written comments should reflect the optionality),
 - ApiRequestLogger now also used by Ledger JSON Api. Changes:
     - Redundant Request TID removed from logs.
     - Additional CLI options added: `--log-access`, `--log-access-errors`...
     - Additional config options added: `debugInProcessRequests`, `prefixGrpcAddresses`
 - ParticipantRepairService.ExportAcsOld and ImportAcsOld are deprecated. Instead use ParticipantRepairService.ExportAcs and ImportAcs respectively as a direct replacement. For party replication use PartyManagementService.ExportPartyAcs and ImportPartyAcs instead.
+- Removed `packageDependencyCache` from `caching` configuration.
 
 ### Preview Features
 - preview feature
@@ -45,6 +51,19 @@ Template for a bigger topic
 
 #### Recommendation
 
+## Other changes
+
+### Changes from NonNegativeLong to Long
+Some console commands using a NonNegativeLong for the offset are changed to accept a Long instead.
+Similarly, some console commands returning an offset now return a Long instead of a NonNegativeLong.
+It brings consistency and allows to pass the output of `participant.ledger_api.state.end()`.
+
+Impacted commands:
+- `participant.repair.export_acs`
+- `participant.parties.find_party_max_activation_offset`
+- `participant.parties.find_party_max_deactivation_offset`
+- `participant.parties.find_highest_offset_by_timestamp`
+
 ## Compatibility
 
 The following Canton protocol and Ethereum sequencer contract versions are supported:

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

@@ -98,7 +98,7 @@ message RegisterSynchronizerRequest {
 
   // 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.
+  // as it will require the sequencers to be online and responsive.
   com.digitalasset.canton.admin.sequencer.v30.SequencerConnectionValidation sequencer_connection_validation = 3;
 }
 
@@ -141,7 +141,7 @@ message ConnectSynchronizerRequest {
 
   // 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.
+  // as it will require the sequencers to be online and responsive.
   com.digitalasset.canton.admin.sequencer.v30.SequencerConnectionValidation sequencer_connection_validation = 2;
 }
 

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

@@ -39,7 +39,7 @@ class GrpcCtlRunner(
     def service(channel: ManagedChannel): command.Svc = {
       val baseService = command
         .createServiceInternal(channel)
-        .withInterceptors(TraceContextGrpc.clientInterceptor)
+        .withInterceptors(TraceContextGrpc.clientInterceptor())
       token.toList.foldLeft(baseService)(AuthCallCredentials.authorizingStub)
     }
     val client = GrpcClient.create(managedChannel, service)

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

@@ -561,7 +561,7 @@ object ParticipantAdminCommands {
     ) extends GrpcAdminCommand[
           v30.GetHighestOffsetByTimestampRequest,
           v30.GetHighestOffsetByTimestampResponse,
-          NonNegativeLong,
+          Long,
         ] {
       override type Svc = PartyManagementServiceStub
 
@@ -586,15 +586,14 @@ object ParticipantAdminCommands {
 
       override protected def handleResponse(
           response: v30.GetHighestOffsetByTimestampResponse
-      ): Either[String, NonNegativeLong] =
-        NonNegativeLong.create(response.ledgerOffset).leftMap(_.toString)
+      ): Either[String, Long] = Right(response.ledgerOffset)
     }
 
     final case class ExportPartyAcs(
         party: PartyId,
         synchronizerId: SynchronizerId,
         targetParticipantId: ParticipantId,
-        beginOffsetExclusive: NonNegativeLong,
+        beginOffsetExclusive: Long,
         waitForActivationTimeout: Option[config.NonNegativeFiniteDuration],
         observer: StreamObserver[v30.ExportPartyAcsResponse],
     ) extends GrpcAdminCommand[
@@ -614,7 +613,7 @@ object ParticipantAdminCommands {
             party.toProtoPrimitive,
             synchronizerId.toProtoPrimitive,
             targetParticipantId.uid.toProtoPrimitive,
-            beginOffsetExclusive.unwrap,
+            beginOffsetExclusive,
             waitForActivationTimeout.map(_.toProtoPrimitive),
           )
         )
@@ -836,7 +835,7 @@ object ParticipantAdminCommands {
     final case class ExportAcs(
         parties: Set[PartyId],
         filterSynchronizerId: Option[SynchronizerId],
-        offset: NonNegativeLong,
+        offset: Long,
         observer: StreamObserver[v30.ExportAcsResponse],
         contractSynchronizerRenames: Map[SynchronizerId, SynchronizerId],
         excludedStakeholders: Set[PartyId],
@@ -856,7 +855,7 @@ object ParticipantAdminCommands {
           v30.ExportAcsRequest(
             parties.map(_.toProtoPrimitive).toSeq,
             filterSynchronizerId.map(_.toProtoPrimitive).getOrElse(""),
-            offset.unwrap,
+            offset,
             contractSynchronizerRenames.map { case (source, targetSynchronizerId) =>
               val target = v30.ExportAcsTargetSynchronizer(
                 targetSynchronizerId.toProtoPrimitive
@@ -1746,6 +1745,8 @@ object ParticipantAdminCommands {
               } yield synchronizerId -> receivedCmts
             )
             .map(_.toMap)
+
+      override def timeoutType: TimeoutType = DefaultUnboundedTimeout
     }
 
     final case class TimeRange(startExclusive: CantonTimestamp, endInclusive: CantonTimestamp)
@@ -1868,6 +1869,8 @@ object ParticipantAdminCommands {
               } yield synchronizerId -> sentCmts
             )
             .map(_.toMap)
+
+      override def timeoutType: TimeoutType = DefaultUnboundedTimeout
     }
 
     final case class SentAcsCmt(
@@ -2146,6 +2149,8 @@ object ParticipantAdminCommands {
           )
         }
       }.sequence
+
+      override def timeoutType: TimeoutType = ServerEnforcedTimeout
     }
   }
 

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

@@ -524,7 +524,7 @@ final case class CantonConfig(
         asyncWriter = sequencerNodeConfig.parameters.asyncWriter.toParameters,
         unsafeEnableOnlinePartyReplication =
           sequencerNodeConfig.parameters.unsafeEnableOnlinePartyReplication,
-        streamLimits = sequencerNodeConfig.publicApi.stream,
+        requestLimits = sequencerNodeConfig.publicApi.limits,
       )
     }
 
@@ -671,6 +671,7 @@ private[canton] object CantonNodeParameterConverter {
       exitOnFatalFailures = parent.parameters.exitOnFatalFailures,
       watchdog = node.parameters.watchdog,
       startupMemoryCheckConfig = parent.parameters.startupMemoryCheckConfig,
+      dispatchQueueBackpressureLimit = node.topology.dispatchQueueBackpressureLimit,
     )
 
   def protocol(parent: CantonConfig, config: ProtocolConfig): CantonNodeParameters.Protocol =
@@ -1354,8 +1355,9 @@ object CantonConfig {
     lazy implicit final val startupMemoryCheckConfigReader: ConfigReader[StartupMemoryCheckConfig] =
       deriveReader[StartupMemoryCheckConfig]
 
-    lazy implicit final val streamLimitConfigReader: ConfigReader[StreamLimitConfig] =
-      deriveReader[StreamLimitConfig]
+    lazy implicit final val activeRequestLimitsConfigReader
+        : ConfigReader[ActiveRequestLimitsConfig] =
+      deriveReader[ActiveRequestLimitsConfig]
 
     implicit val participantReplicationConfigReader: ConfigReader[ReplicationConfig] =
       deriveReader[ReplicationConfig]
@@ -2022,8 +2024,9 @@ object CantonConfig {
     lazy implicit final val startupMemoryCheckConfigWriter: ConfigWriter[StartupMemoryCheckConfig] =
       deriveWriter[StartupMemoryCheckConfig]
 
-    lazy implicit final val streamLimitConfigWriter: ConfigWriter[StreamLimitConfig] =
-      deriveWriter[StreamLimitConfig]
+    lazy implicit final val activeRequestLimitsConfigWriter
+        : ConfigWriter[ActiveRequestLimitsConfig] =
+      deriveWriter[ActiveRequestLimitsConfig]
 
     implicit val participantReplicationConfigWriter: ConfigWriter[ReplicationConfig] =
       deriveWriter[ReplicationConfig]

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

@@ -8,7 +8,6 @@ import cats.syntax.either.*
 import cats.syntax.foldable.*
 import com.digitalasset.canton.admin.api.client.commands.ParticipantAdminCommands
 import com.digitalasset.canton.admin.participant.v30.{ExportAcsOldResponse, ExportAcsResponse}
-import com.digitalasset.canton.config.RequireTypes.NonNegativeLong
 import com.digitalasset.canton.config.{ConsoleCommandTimeout, NonNegativeDuration}
 import com.digitalasset.canton.console.{
   AdminCommandRunner,
@@ -303,7 +302,7 @@ class ParticipantRepairAdministration(
   )
   def export_acs(
       parties: Set[PartyId],
-      ledgerOffset: NonNegativeLong,
+      ledgerOffset: Long,
       exportFilePath: String = "canton-acs-export.gz",
       excludedStakeholders: Set[PartyId] = Set.empty,
       synchronizerId: Option[SynchronizerId] = None,

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

@@ -934,7 +934,7 @@ class ParticipantPartiesAdministrationGroup(
       endOffsetInclusive: Option[Long] = None,
       completeAfter: PositiveInt = PositiveInt.MaxValue,
       timeout: NonNegativeDuration = timeouts.bounded,
-  ): NonNegativeLong = {
+  ): Long = {
     val filter = TopologyTxFiltering.getTopologyFilter(
       partyId,
       participantId,
@@ -999,7 +999,7 @@ class ParticipantPartiesAdministrationGroup(
       endOffsetInclusive: Option[Long] = None,
       completeAfter: PositiveInt = PositiveInt.MaxValue,
       timeout: NonNegativeDuration = timeouts.bounded,
-  ): NonNegativeLong = {
+  ): Long = {
     val filter = TopologyTxFiltering.getTopologyFilter(
       partyId,
       participantId,
@@ -1025,7 +1025,7 @@ class ParticipantPartiesAdministrationGroup(
       completeAfter: PositiveInt,
       timeout: NonNegativeDuration,
       filter: UpdateWrapper => Boolean,
-  ): NonNegativeLong = {
+  ): Long = {
     val topologyTransactions: Seq[com.daml.ledger.api.v2.topology_transaction.TopologyTransaction] =
       reference.ledger_api.updates
         .topology_transactions(
@@ -1040,7 +1040,6 @@ class ParticipantPartiesAdministrationGroup(
 
     topologyTransactions
       .map(_.offset)
-      .map(NonNegativeLong.tryCreate)
       .lastOption
       .getOrElse(
         consoleEnvironment.raiseError(
@@ -1080,7 +1079,7 @@ class ParticipantPartiesAdministrationGroup(
       synchronizerId: SynchronizerId,
       timestamp: Instant,
       force: Boolean = false,
-  ): NonNegativeLong = consoleEnvironment.run {
+  ): Long = consoleEnvironment.run {
     reference.adminCommand(
       ParticipantAdminCommands.PartyManagement
         .GetHighestOffsetByTimestamp(synchronizerId, timestamp, force)
@@ -1143,7 +1142,7 @@ class ParticipantPartiesAdministrationGroup(
             party,
             synchronizerId,
             targetParticipantId,
-            NonNegativeLong.tryCreate(beginOffsetExclusive),
+            beginOffsetExclusive,
             waitForActivationTimeout,
             responseObserver,
           )

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

@@ -3387,7 +3387,8 @@ class TopologyAdministrationGroup(
 
   object synchronizer_upgrade extends Helpful {
 
-    @Help.Summary("Inspect synchronizer migration announcements")
+    // TODO(#28972) Remove preview flag once LSU is stable
+    @Help.Summary("Inspect synchronizer migration announcements", FeatureFlag.Preview)
     @Help.Group("Synchronizer Migration Announcement")
     object announcement extends Helpful {
       def list(
@@ -3446,28 +3447,29 @@ class TopologyAdministrationGroup(
           synchronize: Option[config.NonNegativeDuration] = Some(
             consoleEnvironment.commandTimeouts.unbounded
           ),
-      ): SignedTopologyTransaction[TopologyChangeOp, SynchronizerUpgradeAnnouncement] = {
-
-        val mapping = SynchronizerUpgradeAnnouncement(
-          successorPhysicalSynchronizerId,
-          upgradeTime,
-        )
+      ): SignedTopologyTransaction[TopologyChangeOp, SynchronizerUpgradeAnnouncement] =
+        // TODO(#28972) Remove preview flag once LSU is stable
+        check(FeatureFlag.Preview) {
+          val mapping = SynchronizerUpgradeAnnouncement(
+            successorPhysicalSynchronizerId,
+            upgradeTime,
+          )
 
-        consoleEnvironment.run {
-          adminCommand(
-            TopologyAdminCommands.Write.Propose(
-              mapping = mapping,
-              signedBy = signedBy.toList,
-              serial = serial,
-              change = TopologyChangeOp.Replace,
-              mustFullyAuthorize = mustFullyAuthorize,
-              forceChanges = ForceFlags.none,
-              store = store.getOrElse(successorPhysicalSynchronizerId.logical),
-              waitToBecomeEffective = synchronize,
+          consoleEnvironment.run {
+            adminCommand(
+              TopologyAdminCommands.Write.Propose(
+                mapping = mapping,
+                signedBy = signedBy.toList,
+                serial = serial,
+                change = TopologyChangeOp.Replace,
+                mustFullyAuthorize = mustFullyAuthorize,
+                forceChanges = ForceFlags.none,
+                store = store.getOrElse(successorPhysicalSynchronizerId.logical),
+                waitToBecomeEffective = synchronize,
+              )
             )
-          )
+          }
         }
-      }
 
       @Help.Summary("Propose the revocation of a synchronizer migration announcement")
       @Help.Description(
@@ -3502,27 +3504,29 @@ class TopologyAdministrationGroup(
           synchronize: Option[config.NonNegativeDuration] = Some(
             consoleEnvironment.commandTimeouts.unbounded
           ),
-      ): SignedTopologyTransaction[TopologyChangeOp, SynchronizerUpgradeAnnouncement] = {
-        val mapping = SynchronizerUpgradeAnnouncement(
-          successorPhysicalSynchronizerId,
-          upgradeTime,
-        )
+      ): SignedTopologyTransaction[TopologyChangeOp, SynchronizerUpgradeAnnouncement] =
+        // TODO(#28972) Remove preview flag once LSU is stable
+        check(FeatureFlag.Preview) {
+          val mapping = SynchronizerUpgradeAnnouncement(
+            successorPhysicalSynchronizerId,
+            upgradeTime,
+          )
 
-        consoleEnvironment.run {
-          adminCommand(
-            TopologyAdminCommands.Write.Propose(
-              mapping = mapping,
-              signedBy = signedBy.toList,
-              serial = serial,
-              change = TopologyChangeOp.Remove,
-              mustFullyAuthorize = mustFullyAuthorize,
-              forceChanges = ForceFlags.none,
-              store = store.getOrElse(successorPhysicalSynchronizerId.logical),
-              waitToBecomeEffective = synchronize,
+          consoleEnvironment.run {
+            adminCommand(
+              TopologyAdminCommands.Write.Propose(
+                mapping = mapping,
+                signedBy = signedBy.toList,
+                serial = serial,
+                change = TopologyChangeOp.Remove,
+                mustFullyAuthorize = mustFullyAuthorize,
+                forceChanges = ForceFlags.none,
+                store = store.getOrElse(successorPhysicalSynchronizerId.logical),
+                waitToBecomeEffective = synchronize,
+              )
             )
-          )
+          }
         }
-      }
     }
 
     object sequencer_successors extends Helpful {