Always use the latest ledger config (#5669)

CHANGELOG_BEGIN
- [Sandbox] The ledger API server will now always use the most recent ledger configuration.
  Until a ledger configuration is read from the ledger, command submissions will fail with the UNAVAILABLE error.
CHANGELOG_END

In kvutils, the first ledger configuration change needs
to have a generation one higher than the one returned
by getLedgerInitialConditions().

Remove initial config writing from sandbox as it's now written by the ledger API server

ledger/ledger-api-common/src/main/scala/com/digitalasset/ledger/api/validation/CommandsValidator.scala

@@ -35,7 +35,7 @@ final class CommandsValidator(ledgerId: LedgerId) {
       commands: ProtoCommands,
       currentLedgerTime: Instant,
       currentUtcTime: Instant,
-      maxDeduplicationTime: Duration): Either[StatusRuntimeException, domain.Commands] =
+      maxDeduplicationTime: Option[Duration]): Either[StatusRuntimeException, domain.Commands] =
     for {
       cmdLegerId <- requireLedgerString(commands.ledgerId, "ledger_id")
       ledgerId <- matchLedgerId(ledgerId)(LedgerId(cmdLegerId))

ledger/ledger-api-common/src/main/scala/com/digitalasset/ledger/api/validation/SubmitAndWaitRequestValidator.scala

@@ -17,7 +17,8 @@ class SubmitAndWaitRequestValidator(commandsValidator: CommandsValidator) {
       req: SubmitAndWaitRequest,
       currentLedgerTime: Instant,
       currentUtcTime: Instant,
-      maxDeduplicationTime: Duration): Either[StatusRuntimeException, submission.SubmitRequest] =
+      maxDeduplicationTime: Option[Duration])
+    : Either[StatusRuntimeException, submission.SubmitRequest] =
     for {
       commands <- requirePresence(req.commands, "commands")
       validatedCommands <- commandsValidator.validateCommands(

ledger/ledger-api-common/src/main/scala/com/digitalasset/ledger/api/validation/SubmitRequestValidator.scala

@@ -17,7 +17,8 @@ class SubmitRequestValidator(commandsValidator: CommandsValidator) {
       req: SubmitRequest,
       currentLedgerTime: Instant,
       currentUtcTime: Instant,
-      maxDeduplicationTime: Duration): Either[StatusRuntimeException, submission.SubmitRequest] =
+      maxDeduplicationTime: Option[Duration])
+    : Either[StatusRuntimeException, submission.SubmitRequest] =
     for {
       commands <- requirePresence(req.commands, "commands")
       validatedCommands <- commandsValidator.validateCommands(

ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/services/grpc/GrpcCommandService.scala

@@ -23,7 +23,7 @@ class GrpcCommandService(
     val ledgerId: LedgerId,
     currentLedgerTime: () => Instant,
     currentUtcTime: () => Instant,
-    maxDeduplicationTime: () => Duration
+    maxDeduplicationTime: () => Option[Duration]
 ) extends CommandService
     with GrpcApiService
     with ProxyCloseable {

ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/services/grpc/GrpcCommandSubmissionService.scala

@@ -30,7 +30,7 @@ class GrpcCommandSubmissionService(
     ledgerId: LedgerId,
     currentLedgerTime: () => Instant,
     currentUtcTime: () => Instant,
-    maxDeduplicationTime: () => Duration,
+    maxDeduplicationTime: () => Option[Duration],
     metrics: Metrics,
 ) extends ApiCommandSubmissionService
     with ProxyCloseable

ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/validation/ErrorFactories.scala

@@ -43,6 +43,9 @@ trait ErrorFactories {
   def unauthenticated(): StatusRuntimeException =
     grpcError(Status.UNAUTHENTICATED)
 
+  def missingLedgerConfig(): StatusRuntimeException =
+    grpcError(Status.UNAVAILABLE.withDescription("The ledger configuration is not available."))
+
   def resourceExhausted(description: String): StatusRuntimeException =
     grpcError(Status.RESOURCE_EXHAUSTED.withDescription(description))
 

ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/validation/FieldValidations.scala

@@ -92,21 +92,24 @@ trait FieldValidations {
 
   def validateDeduplicationTime(
       durationO: Option[com.google.protobuf.duration.Duration],
-      maxDeduplicationTime: Duration,
-      fieldName: String): Either[StatusRuntimeException, Duration] = durationO match {
-    case None =>
-      Right(maxDeduplicationTime)
-    case Some(duration) =>
-      val result = Duration.ofSeconds(duration.seconds, duration.nanos.toLong)
-      if (result.isNegative)
-        Left(invalidField(fieldName, "Duration must be positive"))
-      else if (result.compareTo(maxDeduplicationTime) > 0)
-        Left(invalidField(
-          fieldName,
-          s"The given deduplication time of $result exceeds the maximum deduplication time of $maxDeduplicationTime"))
-      else
-        Right(result)
-  }
+      maxDeduplicationTimeO: Option[Duration],
+      fieldName: String): Either[StatusRuntimeException, Duration] =
+    maxDeduplicationTimeO.fold[Either[StatusRuntimeException, Duration]](
+      Left(missingLedgerConfig()))(maxDeduplicationTime =>
+      durationO match {
+        case None =>
+          Right(maxDeduplicationTime)
+        case Some(duration) =>
+          val result = Duration.ofSeconds(duration.seconds, duration.nanos.toLong)
+          if (result.isNegative)
+            Left(invalidField(fieldName, "Duration must be positive"))
+          else if (result.compareTo(maxDeduplicationTime) > 0)
+            Left(invalidField(
+              fieldName,
+              s"The given deduplication time of $result exceeds the maximum deduplication time of $maxDeduplicationTime"))
+          else
+            Right(result)
+    })
 
   def validateIdentifier(identifier: Identifier): Either[StatusRuntimeException, Ref.Identifier] =
     for {

ledger/ledger-api-common/src/test/suite/scala/com/digitalasset/ledger/api/validation/SubmitRequestValidatorTest.scala

@@ -19,7 +19,7 @@ import com.daml.ledger.api.v1.value.Value.Sum
 import com.daml.ledger.api.v1.value.{List => ApiList, Map => ApiMap, Optional => ApiOptional, _}
 import com.google.protobuf.duration.Duration
 import com.google.protobuf.empty.Empty
-import io.grpc.Status.Code.INVALID_ARGUMENT
+import io.grpc.Status.Code.{INVALID_ARGUMENT, UNAVAILABLE}
 import org.scalatest.WordSpec
 import org.scalatest.prop.TableDrivenPropertyChecks
 import scalaz.syntax.tag._
@@ -124,7 +124,7 @@ class SubmitRequestValidatorTest
             api.commands.withCommands(Seq.empty),
             internal.ledgerTime,
             internal.submittedAt,
-            internal.maxDeduplicationTime),
+            Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           "Missing field: commands"
         )
@@ -137,7 +137,7 @@ class SubmitRequestValidatorTest
               api.commands.withLedgerId(""),
               internal.ledgerTime,
               internal.submittedAt,
-              internal.maxDeduplicationTime),
+              Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           "Missing field: ledger_id"
         )
@@ -148,7 +148,7 @@ class SubmitRequestValidatorTest
           api.commands.withWorkflowId(""),
           internal.ledgerTime,
           internal.submittedAt,
-          internal.maxDeduplicationTime) shouldEqual Right(
+          Some(internal.maxDeduplicationTime)) shouldEqual Right(
           internal.emptyCommands.copy(
             workflowId = None,
             commands = internal.emptyCommands.commands.copy(commandsReference = "")))
@@ -160,7 +160,7 @@ class SubmitRequestValidatorTest
             api.commands.withApplicationId(""),
             internal.ledgerTime,
             internal.submittedAt,
-            internal.maxDeduplicationTime),
+            Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           "Missing field: application_id"
         )
@@ -172,7 +172,7 @@ class SubmitRequestValidatorTest
             api.commands.withCommandId(""),
             internal.ledgerTime,
             internal.submittedAt,
-            internal.maxDeduplicationTime),
+            Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           "Missing field: command_id"
         )
@@ -185,7 +185,7 @@ class SubmitRequestValidatorTest
               api.commands.withParty(""),
               internal.ledgerTime,
               internal.submittedAt,
-              internal.maxDeduplicationTime),
+              Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           """Missing field: party"""
         )
@@ -198,7 +198,7 @@ class SubmitRequestValidatorTest
             minLedgerTimeAbs = Some(TimestampConversion.fromInstant(minLedgerTimeAbs))),
           internal.ledgerTime,
           internal.submittedAt,
-          internal.maxDeduplicationTime
+          Some(internal.maxDeduplicationTime)
         ) shouldEqual Right(withLedgerTime(internal.emptyCommands, minLedgerTimeAbs))
       }
 
@@ -209,7 +209,7 @@ class SubmitRequestValidatorTest
             minLedgerTimeRel = Some(DurationConversion.toProto(internal.timeDelta))),
           internal.ledgerTime,
           internal.submittedAt,
-          internal.maxDeduplicationTime
+          Some(internal.maxDeduplicationTime)
         ) shouldEqual Right(withLedgerTime(internal.emptyCommands, minLedgerTimeAbs))
       }
 
@@ -219,7 +219,7 @@ class SubmitRequestValidatorTest
             api.commands.copy(deduplicationTime = Some(Duration.of(-1, 0))),
             internal.ledgerTime,
             internal.submittedAt,
-            internal.maxDeduplicationTime),
+            Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           "Invalid field deduplication_time: Duration must be positive"
         )
@@ -232,7 +232,7 @@ class SubmitRequestValidatorTest
             api.commands.copy(deduplicationTime = Some(Duration.of(manySeconds, 0))),
             internal.ledgerTime,
             internal.submittedAt,
-            internal.maxDeduplicationTime),
+            Some(internal.maxDeduplicationTime)),
           INVALID_ARGUMENT,
           s"Invalid field deduplication_time: The given deduplication time of ${java.time.Duration
             .ofSeconds(manySeconds)} exceeds the maximum deduplication time of ${internal.maxDeduplicationTime}"
@@ -244,11 +244,20 @@ class SubmitRequestValidatorTest
           api.commands.copy(deduplicationTime = None),
           internal.ledgerTime,
           internal.submittedAt,
-          internal.maxDeduplicationTime) shouldEqual Right(
+          Some(internal.maxDeduplicationTime)) shouldEqual Right(
           internal.emptyCommands.copy(
             deduplicateUntil = internal.submittedAt.plus(internal.maxDeduplicationTime)))
       }
 
+      "not allow missing ledger configuration" in {
+        requestMustFailWith(
+          commandsValidator
+            .validateCommands(api.commands, internal.ledgerTime, internal.submittedAt, None),
+          UNAVAILABLE,
+          "The ledger configuration is not available."
+        )
+      }
+
     }
 
     "validating contractId values" should {

ledger/ledger-on-memory/src/app/scala/com/daml/ledger/on/memory/Main.scala

@@ -18,6 +18,7 @@ import com.daml.logging.LoggingContext
 import com.daml.metrics.Metrics
 import com.daml.platform.akkastreams.dispatcher.Dispatcher
 import com.daml.platform.apiserver.ApiServerConfig
+import com.daml.platform.configuration.LedgerConfiguration
 import com.daml.resources.{ProgramResource, ResourceOwner}
 import scopt.OptionParser
 
@@ -66,6 +67,9 @@ object Main {
         engine = engine,
       )
 
+    override def ledgerConfig(config: Config[ExtraConfig]): LedgerConfiguration =
+      LedgerConfiguration.defaultLocalLedger
+
     override val defaultExtraConfig: ExtraConfig = ExtraConfig.default
 
     override final def extraConfigParser(parser: OptionParser[Config[ExtraConfig]]): Unit = {

ledger/ledger-on-sql/src/app/scala/com/daml/ledger/on/sql/SqlLedgerFactory.scala

@@ -3,6 +3,8 @@
 
 package com.daml.ledger.on.sql
 
+import java.time.Duration
+
 import akka.stream.Materializer
 import com.daml.ledger.participant.state.kvutils.api.KeyValueParticipantState
 import com.daml.ledger.participant.state.kvutils.app.{
@@ -16,6 +18,7 @@ import com.daml.ledger.participant.state.v1.SeedService
 import com.daml.lf.engine.Engine
 import com.daml.logging.LoggingContext
 import com.daml.metrics.Metrics
+import com.daml.platform.configuration.LedgerConfiguration
 import com.daml.resources.{Resource, ResourceOwner}
 import scopt.OptionParser
 
@@ -26,6 +29,9 @@ object SqlLedgerFactory extends LedgerFactory[ReadWriteService, ExtraConfig] {
     jdbcUrl = None,
   )
 
+  override def ledgerConfig(config: Config[ExtraConfig]): LedgerConfiguration =
+    super.ledgerConfig(config).copy(initialConfigurationSubmitDelay = Duration.ZERO)
+
   override def extraConfigParser(parser: OptionParser[Config[ExtraConfig]]): Unit = {
     parser
       .opt[String]("jdbc-url")

ledger/participant-state-index/src/main/scala/com/daml/ledger/participant/state/index/v2/IndexConfigManagementService.scala

@@ -23,7 +23,7 @@ trait IndexConfigManagementService {
   def lookupConfiguration(): Future[Option[(LedgerOffset.Absolute, Configuration)]]
 
   /** Retrieve configuration entries. */
-  def configurationEntries(
-      startExclusive: Option[LedgerOffset.Absolute]): Source[ConfigurationEntry, NotUsed]
+  def configurationEntries(startExclusive: Option[LedgerOffset.Absolute])
+    : Source[(LedgerOffset.Absolute, ConfigurationEntry), NotUsed]
 
 }

ledger/participant-state/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/LedgerFactory.scala

@@ -3,8 +3,6 @@
 
 package com.daml.ledger.participant.state.kvutils.app
 
-import java.time.Duration
-
 import akka.stream.Materializer
 import com.codahale.metrics.{MetricRegistry, SharedMetricRegistries}
 import com.daml.ledger.api.auth.{AuthService, AuthServiceWildcard}
@@ -17,8 +15,7 @@ import com.daml.platform.apiserver.{ApiServerConfig, TimeServiceBackend}
 import com.daml.platform.configuration.{
   CommandConfiguration,
   LedgerConfiguration,
-  PartyConfiguration,
-  SubmissionConfiguration
+  PartyConfiguration
 }
 import com.daml.platform.indexer.{IndexerConfig, IndexerStartupMode}
 import com.daml.resources.ResourceOwner
@@ -73,13 +70,8 @@ trait ConfigProvider[ExtraConfig] {
   def partyConfig(config: Config[ExtraConfig]): PartyConfiguration =
     PartyConfiguration.default
 
-  def submissionConfig(config: Config[ExtraConfig]): SubmissionConfiguration =
-    SubmissionConfiguration.default
-
   def ledgerConfig(config: Config[ExtraConfig]): LedgerConfiguration =
-    LedgerConfiguration.default.copy(
-      initialConfigurationSubmitDelay = Duration.ofSeconds(5)
-    )
+    LedgerConfiguration.defaultRemote
 
   def timeServiceBackend(config: Config[ExtraConfig]): Option[TimeServiceBackend] = None
 

ledger/participant-state/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/Runner.scala

@@ -81,7 +81,6 @@ final class Runner[T <: ReadWriteService, Extra](
                 config = factory.apiServerConfig(participantConfig, config),
                 commandConfig = factory.commandConfig(participantConfig, config),
                 partyConfig = factory.partyConfig(config),
-                submissionConfig = factory.submissionConfig(config),
                 ledgerConfig = factory.ledgerConfig(config),
                 readService = readService,
                 writeService = writeService,