Add ledger API tests about command deduplication information in completions [KVL-1057] (#10748)

* Add Ledger API Test Tool tests about command deduplication information present in completions

CHANGELOG_BEGIN
CHANGELOG_END

* Address review comments

* Fix offset reference and move new tests to a separate, optional suite

* Cover rejections as well

* Test both the command and command submission services

* Run new test suite (only) on sandbox-classic append-only and daml-on-sql

* Use the append-only schema with daml-on-sql for CommandDeduplicationInfoIT

* Fix tests except offset, successful completions only.

* Remove completion offset test as it's not supported by most ledgers

* Remove support for multiple submissions as rejection completions are not being tested for the moment

* Consolidate test cases for faster run

* Avoid forbidden characters in short identifiers

* Clarify assert

* Remove wrong test about deduplication time being preserved in completion

* Adhere to the Scala style guide

* Eliminate some code duplication

* fmt

* Make 3e404be compile

* Never assume a specific deduplication period format in completions

* Code cleanup

* Pass party to completionStreamRequest to avoid ILLEGAL_ARG failures

* Enable in append-only mode for all ledgers that support it

* Clarify that the new tests are append-only-only

* Update ledger/ledger-on-sql/BUILD.bazel

Fix suite name

* Update ledger/sandbox-classic/BUILD.bazel

Fix suite name

* fmt

ledger/daml-on-sql/BUILD.bazel

@@ -166,6 +166,22 @@ conformance_test(
     ],
 )
 
+conformance_test(
+    name = "conformance-test-append-only-command-completion-dedup-info",
+    server = ":daml-on-sql-ephemeral-postgresql",
+    server_args = [
+        "--ledgerid=conformance-test",
+        "--port=6865",
+        "--enable-append-only-schema",
+    ],
+    test_tool_args = [
+        "--verbose",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
 genrule(
     name = "docs",
     srcs = [

ledger/ledger-api-test-tool/BUILD.bazel

@@ -87,6 +87,7 @@ da_scala_binary(
                 "//ledger/test-common:test-common-%s" % lf_version,
                 "//ledger/test-common:model-tests-%s.scala" % lf_version,
                 "//ledger/test-common:dar-files-%s-lib" % lf_version,
+                "//ledger-api/grpc-definitions:ledger_api_proto_scala",
                 "//libs-scala/build-info",
                 "//libs-scala/grpc-utils",
                 "//libs-scala/resources",
@@ -139,6 +140,7 @@ da_scala_binary(
             deps = [
                 "//daml-lf/language",
                 "//ledger/test-common:dar-files-%s-lib" % lf_version,
+                "//ledger-api/grpc-definitions:ledger_api_proto_scala",
                 ":ledger-api-test-tool-%s-lib" % lf_version,
                 ":ledger-api-test-tool-%s-test-suites" % lf_version,
             ],

ledger/ledger-api-test-tool/src/main/scala/com/daml/ledger/api/testtool/infrastructure/LedgerTestSuite.scala

@@ -10,8 +10,6 @@ import scala.collection.mutable.ListBuffer
 import scala.concurrent.{ExecutionContext, Future}
 
 private[testtool] abstract class LedgerTestSuite {
-  val name: String = getClass.getSimpleName
-
   private val testCaseBuffer: ListBuffer[LedgerTestCase] = ListBuffer()
 
   final lazy val tests: Vector[LedgerTestCase] = testCaseBuffer.toVector
@@ -38,4 +36,6 @@ private[testtool] abstract class LedgerTestSuite {
       )
     )
   }
+
+  private[testtool] def name: String = getClass.getSimpleName
 }

ledger/ledger-api-test-tool/src/main/scala/com/daml/ledger/api/testtool/suites/AppendOnlyCompletionDeduplicationInfoIT.scala

@@ -0,0 +1,192 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.testtool.suites
+
+import com.daml.ledger.api.SubmissionIdGenerator
+import com.daml.ledger.api.testtool.infrastructure.Allocation._
+import com.daml.ledger.api.testtool.infrastructure.LedgerTestSuite
+import com.daml.ledger.api.testtool.infrastructure.participant.ParticipantTestContext
+import com.daml.ledger.api.testtool.suites.AppendOnlyCompletionDeduplicationInfoIT._
+import com.daml.ledger.api.v1.command_service.SubmitAndWaitRequest
+import com.daml.ledger.api.v1.command_submission_service.SubmitRequest
+import com.daml.ledger.api.v1.commands.Command
+import com.daml.ledger.api.v1.completion.Completion
+import com.daml.ledger.api.v1.ledger_offset.LedgerOffset
+import com.daml.ledger.client.binding
+import com.daml.ledger.client.binding.Primitive
+import com.daml.ledger.test.model.Test.Dummy
+import com.daml.lf.data.Ref
+import com.daml.lf.data.Ref.SubmissionId
+import com.daml.platform.testing.WithTimeout
+import io.grpc.Status
+
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{ExecutionContext, Future}
+
+final class AppendOnlyCompletionDeduplicationInfoIT[ServiceRequest](
+    service: Service[ServiceRequest]
+) extends LedgerTestSuite {
+
+  private val serviceName: String = service.productPrefix
+
+  override private[testtool] def name = super.name + serviceName
+
+  test(
+    shortIdentifier = s"CCDIIncludeDedupInfo$serviceName",
+    description = s"Deduplication information is preserved in completions ($serviceName)",
+    allocate(SingleParty),
+  )(implicit ec => { case Participants(Participant(ledger, party)) =>
+    val requestWithoutSubmissionId = service.buildRequest(ledger, party)
+    val requestWithSubmissionId = service.buildRequest(ledger, party, Some(RandomSubmissionId))
+    for {
+      optNoDeduplicationSubmittedCompletion <- service.submitRequest(
+        ledger,
+        party,
+        requestWithoutSubmissionId,
+      )
+      optSubmissionIdSubmittedCompletion <- service
+        .submitRequest(ledger, party, requestWithSubmissionId)
+    } yield {
+      assertApplicationIdIsPreserved(ledger.applicationId, optNoDeduplicationSubmittedCompletion)
+      assertSubmissionIdIsGenerated(optNoDeduplicationSubmittedCompletion)
+      assertDeduplicationPeriodIsReported(optNoDeduplicationSubmittedCompletion)
+      assertSubmissionIdIsPreserved(optSubmissionIdSubmittedCompletion, RandomSubmissionId)
+    }
+  })
+}
+
+private[testtool] object AppendOnlyCompletionDeduplicationInfoIT {
+
+  private[testtool] sealed trait Service[ProtoRequestType] extends Serializable with Product {
+    def buildRequest(
+        ledger: ParticipantTestContext,
+        party: Primitive.Party,
+        optSubmissionId: Option[Ref.SubmissionId] = None,
+    ): ProtoRequestType
+
+    def submitRequest(
+        ledger: ParticipantTestContext,
+        party: Primitive.Party,
+        request: ProtoRequestType,
+    )(implicit ec: ExecutionContext): Future[Option[Completion]]
+  }
+
+  case object CommandService extends Service[SubmitAndWaitRequest] {
+    override def buildRequest(
+        ledger: ParticipantTestContext,
+        party: binding.Primitive.Party,
+        optSubmissionId: Option[SubmissionId],
+    ): SubmitAndWaitRequest = {
+      val request = ledger.submitAndWaitRequest(party, simpleCreate(party))
+      optSubmissionId
+        .map { submissionId =>
+          request.update(_.commands.submissionId := submissionId)
+        }
+        .getOrElse(request)
+    }
+
+    override def submitRequest(
+        ledger: ParticipantTestContext,
+        party: binding.Primitive.Party,
+        request: SubmitAndWaitRequest,
+    )(implicit ec: ExecutionContext): Future[Option[Completion]] =
+      for {
+        offset <- ledger.currentEnd()
+        _ <- ledger.submitAndWait(request)
+        completion <- singleCompletionAfterOffset(ledger, party, offset)
+      } yield completion
+  }
+
+  case object CommandSubmissionService extends Service[SubmitRequest] {
+    override def buildRequest(
+        ledger: ParticipantTestContext,
+        party: binding.Primitive.Party,
+        optSubmissionId: Option[SubmissionId],
+    ): SubmitRequest = {
+      val request = ledger.submitRequest(party, simpleCreate(party))
+      optSubmissionId
+        .map { submissionId =>
+          request.update(_.commands.submissionId := submissionId)
+        }
+        .getOrElse(request)
+    }
+
+    override def submitRequest(
+        ledger: ParticipantTestContext,
+        party: binding.Primitive.Party,
+        request: SubmitRequest,
+    )(implicit ec: ExecutionContext): Future[Option[Completion]] =
+      for {
+        offset <- ledger.currentEnd()
+        _ <- ledger.submit(request)
+        completion <- singleCompletionAfterOffset(ledger, party, offset)
+      } yield completion
+  }
+
+  private def singleCompletionAfterOffset(
+      ledger: ParticipantTestContext,
+      party: binding.Primitive.Party,
+      offset: LedgerOffset,
+  ): Future[Option[Completion]] =
+    WithTimeout(5.seconds)(
+      ledger.findCompletion(ledger.completionStreamRequest(offset)(party))(_ => true)
+    )
+
+  private def assertSubmissionIdIsPreserved(
+      optCompletion: Option[Completion],
+      requestedSubmissionId: Ref.SubmissionId,
+  ): Unit = {
+    val submissionIdCompletion = assertDefined(optCompletion)
+    val actualSubmissionId = submissionIdCompletion.submissionId
+    assert(submissionIdCompletion.status.forall(_.code == Status.Code.OK.value()))
+    assert(
+      actualSubmissionId == requestedSubmissionId,
+      "Wrong submission ID in completion, " +
+        s"expected: $requestedSubmissionId, actual: $actualSubmissionId",
+    )
+  }
+
+  private def assertDeduplicationPeriodIsReported(
+      optCompletion: Option[Completion]
+  ): Unit = {
+    val completion = assertDefined(optCompletion)
+    assert(completion.status.forall(_.code == Status.Code.OK.value()))
+    assert(completion.deduplicationPeriod.isDefined, "The deduplication period was not reported")
+  }
+
+  private def assertSubmissionIdIsGenerated(optCompletion: Option[Completion]): Unit = {
+    val completion = assertDefined(optCompletion)
+    assert(completion.status.forall(_.code == Status.Code.OK.value()))
+    assert(
+      Ref.SubmissionId.fromString(completion.submissionId).isRight,
+      "Missing or invalid submission ID in completion",
+    )
+  }
+
+  private def assertApplicationIdIsPreserved(
+      requestedApplicationId: String,
+      optCompletion: Option[Completion],
+  ): Unit = {
+    val expectedApplicationId = requestedApplicationId
+    assertDefined(optCompletion)
+    val applicationIdCompletion = optCompletion.get
+    assert(applicationIdCompletion.status.forall(_.code == Status.Code.OK.value()))
+    val actualApplicationId = applicationIdCompletion.applicationId
+    assert(
+      Ref.ApplicationId.fromString(actualApplicationId).contains(expectedApplicationId),
+      "Wrong application ID in completion, " +
+        s"expected: $expectedApplicationId, actual: $actualApplicationId",
+    )
+  }
+
+  private def assertDefined(optCompletion: Option[Completion]): Completion = {
+    assert(optCompletion.isDefined, "No completion has been produced")
+    optCompletion.get
+  }
+
+  private def simpleCreate(party: Primitive.Party): Command = Dummy(party).create.command
+
+  private val RandomSubmissionId =
+    Ref.SubmissionId.assertFromString(SubmissionIdGenerator.Random.generate())
+}

ledger/ledger-api-test-tool/src/main/scala/com/daml/ledger/api/testtool/suites/CommandSubmissionCompletionIT.scala

@@ -3,8 +3,6 @@
 
 package com.daml.ledger.api.testtool.suites
 
-import java.util.regex.Pattern
-
 import com.daml.ledger.api.testtool.infrastructure.Allocation._
 import com.daml.ledger.api.testtool.infrastructure.Assertions._
 import com.daml.ledger.api.testtool.infrastructure.LedgerTestSuite
@@ -13,6 +11,7 @@ import com.daml.ledger.test.model.Test.Dummy._
 import com.daml.platform.testing.{TimeoutException, WithTimeout}
 import io.grpc.Status
 
+import java.util.regex.Pattern
 import scala.concurrent.duration.DurationInt
 
 final class CommandSubmissionCompletionIT extends LedgerTestSuite {
@@ -163,5 +162,4 @@ final class CommandSubmissionCompletionIT extends LedgerTestSuite {
       // Nothing to do, if the two completions are found the test is passed
     }
   })
-
 }

ledger/ledger-api-test-tool/src/main/scala/com/daml/ledger/api/testtool/tests/Tests.scala

@@ -3,13 +3,16 @@
 
 package com.daml.ledger.api.testtool.tests
 
-import java.nio.file.Path
-
 import com.daml.ledger.api.testtool.infrastructure.{BenchmarkReporter, Envelope, LedgerTestSuite}
+import com.daml.ledger.api.testtool.suites.AppendOnlyCompletionDeduplicationInfoIT.{
+  CommandService,
+  CommandSubmissionService,
+}
 import com.daml.ledger.api.testtool.suites._
-import com.daml.lf.language.LanguageVersion
 import com.daml.ledger.test.TestDar
+import com.daml.lf.language.LanguageVersion
 
+import java.nio.file.Path
 import scala.collection.SortedSet
 import scala.concurrent.duration.FiniteDuration
 
@@ -64,6 +67,8 @@ object Tests {
   ): Vector[LedgerTestSuite] =
     Vector(
       new CommandDeduplicationOffsetIT,
+      new AppendOnlyCompletionDeduplicationInfoIT(CommandService),
+      new AppendOnlyCompletionDeduplicationInfoIT(CommandSubmissionService),
       new KVCommandDeduplicationIT(timeoutScaleFactor, ledgerClockGranularity),
       new ContractIdIT,
       new MultiPartySubmissionIT,
@@ -93,5 +98,4 @@ object Tests {
 
   private[testtool] val PerformanceTestsKeys: SortedSet[String] =
     SortedSet(Envelope.All.map(_.name): _*)
-
 }

ledger/ledger-on-memory/BUILD.bazel

@@ -150,6 +150,24 @@ conformance_test(
     ],
 )
 
+conformance_test(
+    name = "conformance-test-append-only-command-completion-dedup-info",
+    ports = [6865],
+    server = ":app",
+    server_args = [
+        "--contract-id-seeding=testing-weak",
+        "--index-append-only-schema",
+        "--mutable-contract-state-cache",
+        "--participant=participant-id=example,port=6865",
+    ],
+    test_tool_args = [
+        "--verbose",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
 conformance_test(
     name = "conformance-test-min-1.14",
     lf_versions = ["1.14"],

ledger/ledger-on-sql/BUILD.bazel

@@ -396,6 +396,64 @@ conformance_test(
     ],
 )
 
+conformance_test(
+    name = "conformance-test-append-only-postgres-command-completion-dedup-info",
+    ports = [6865],
+    server = ":conformance-test-postgresql-bin",
+    server_args = [
+        "--contract-id-seeding=testing-weak",
+        "--participant participant-id=conformance-test,port=6865,contract-state-cache-max-size=1,contract-key-state-cache-max-size=1",
+        "--index-append-only-schema",
+        "--mutable-contract-state-cache",
+    ],
+    tags = [],
+    test_tool_args = [
+        "--verbose",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
+conformance_test(
+    name = "conformance-test-append-only-h2-command-completion-dedup-info",
+    ports = [6865],
+    server = ":conformance-test-h2-memory-bin",
+    server_args = [
+        "--contract-id-seeding=testing-weak",
+        "--participant participant-id=conformance-test,port=6865,contract-state-cache-max-size=1,contract-key-state-cache-max-size=1",
+        "--index-append-only-schema",
+        "--mutable-contract-state-cache",
+        "--jdbc-url=jdbc:h2:mem:daml-on-sql-conformance-test",
+    ],
+    tags = [],
+    test_tool_args = [
+        "--verbose",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
+conformance_test(
+    name = "conformance-test-append-only-oracle-command-completion-dedup-info",
+    ports = [6865],
+    server = ":conformance-test-oracle-bin",
+    server_args = [
+        "--contract-id-seeding=testing-weak",
+        "--participant participant-id=conformance-test,port=6865,contract-state-cache-max-size=1,contract-key-state-cache-max-size=1",
+        "--index-append-only-schema",
+        "--mutable-contract-state-cache",
+    ],
+    tags = [] if oracle_testing else ["manual"],
+    test_tool_args = [
+        "--verbose",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
 conformance_test(
     name = "conformance-test-oracle",
     ports = [6865],

ledger/sandbox-classic/BUILD.bazel

@@ -437,3 +437,19 @@ server_conformance_test(
         "--include=ContractIdIT:Accept",
     ],
 )
+
+server_conformance_test(
+    name = "conformance-test-append-only-command-completion-dedup-info",
+    server_args = [
+        "--wall-clock-time",
+        "--contract-id-seeding=testing-weak",
+        "--enable-append-only-schema",
+    ],
+    servers = ONLY_POSTGRES_SERVER,
+    test_tool_args = [
+        "--open-world",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)

ledger/sandbox/BUILD.bazel

@@ -339,6 +339,20 @@ server_conformance_test(
     ],
 )
 
+server_conformance_test(
+    name = "next-conformance-test-append-only-command-completion-dedup-info",
+    server_args = [
+        "--enable-append-only-schema",
+    ],
+    servers = {"postgresql": NEXT_SERVERS["postgresql"]},
+    test_tool_args = [
+        "--open-world",
+        "--include=" +
+        "AppendOnlyCompletionDeduplicationInfoITCommandService" +
+        ",AppendOnlyCompletionDeduplicationInfoITCommandSubmissionService",
+    ],
+)
+
 server_conformance_test(
     name = "next-conformance-test-command-deduplication",
     servers = NEXT_SERVERS,