Race condition test suite for the ledger-api-test-tool [DPP-274] (#9138)

* Early draft of the race condition ITs

* Archival vs Successful lookup by key test

* More descriptive failure messages

* Unsuccessful lookup vs non-transient creation test

* Double-archival test

* Fixed a test case name

* Archival vs Creation order test

* Reduced number of test templates

* Improved race test template naming

* Helper object with transaction and template utils

* Simplified transaction util

* Fixed wrong choice name

* Removed redundant println

* Formatted code changes

* Minor change

* CHANGELOG_BEGIN
- Integration Kit - added a test suite for race condition to the ledger-api-test-tool
CHANGELOG_END

* Removed unnecessary sorting of transactions

* Added explanatory comments to test cases

* Mechanism for running ledger-api-test-tool test cases multiple times

* Running each race condition test case 5 times

* Fixed WWArchiveVsNonTransientCreate test case

* Fixed flakiness of RWArchiveVsNonConsumingChoice

* Disabled RaceConditionIT in Canton tests

* Formatted code changes

* Moved RaceConditionIT to conformance tests with unique contract keys mode on for Canton

* Nicer delay mechanism

* Improved WWArchiveVsNonTransientCreate to take contention into account

* Fixed RWTransientCreateVsNonTransientCreate conditions for Canton

* Increased the delay before reading the transaction trees stream to 1 second

* Fixed incorrect conformance tests definition for RaceConditionIT

* Running race condition tests sequentially to avoid timeouts

* Simplified race condition test case definition

* Return sum of durations for repeating test cases in the ledger-api-test-tool

* Reverted previous change with computing sum of durations

* Exclude RaceConditionIT from sandbox-on-x conformance tests

* Print the number of a test run only for cases when the number of repetitions is > 1

* Fixed RWArchiveVsFetch scenario

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

@@ -78,7 +78,8 @@ conformance_test(
         "--exclude=ContractKeysIT,ContractKeysIT:CKFetchOrLookup,ContractKeysIT:CKNoFetchUndisclosed,ContractKeysIT:CKMaintainerScoped" +
         ",ParticipantPruningIT" +  # see "conformance-test-participant-pruning" below
         ",ConfigManagementServiceIT,LedgerConfigurationServiceIT" +  # dynamic config management not supported by Canton
-        ",ClosedWorldIT",  # Canton currently fails this test with a different error (missing namespace in "unallocated" party id)
+        ",ClosedWorldIT" +  # Canton currently fails this test with a different error (missing namespace in "unallocated" party id)
+        ",RaceConditionIT",
     ],
 ) if not is_windows else None
 
@@ -110,7 +111,8 @@ conformance_test(
     test_tool_args = [
         "--verbose",
         "--concurrent-test-runs=4",  # lowered from default #procs to reduce flakes - details in https://github.com/digital-asset/daml/issues/7316
-        "--include=ContractKeysIT",
+        "--include=ContractKeysIT" +
+        ",RaceConditionIT",
     ],
 ) if not is_windows else None
 

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

@@ -22,6 +22,7 @@ sealed class LedgerTestCase(
     val description: String,
     val timeoutScale: Double,
     val runConcurrently: Boolean,
+    val repeated: Int = 1,
     participants: ParticipantAllocation,
     runTestCase: ExecutionContext => Participants => Future[Unit],
 ) {

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

@@ -58,13 +58,27 @@ final class LedgerTestCasesRunner(
 
   private def start(test: LedgerTestCase, session: LedgerSession)(implicit
       executionContext: ExecutionContext
+  ): Future[Duration] = {
+    def logAndStart(repetition: Int): Future[Duration] = {
+      if (test.repeated > 1)
+        logger.info(s"Starting '${test.description}'. Run: $repetition out of ${test.repeated}")
+      startSingle(test, session, repetition)
+    }
+
+    (2 to test.repeated).foldLeft(logAndStart(1)) { (result, repetition) =>
+      result.flatMap(_ => logAndStart(repetition))
+    }
+  }
+
+  private def startSingle(test: LedgerTestCase, session: LedgerSession, repetition: Int)(implicit
+      executionContext: ExecutionContext
   ): Future[Duration] = {
     val execution = Promise[Duration]()
     val scaledTimeout = DefaultTimeout * timeoutScaleFactor * test.timeoutScale
 
     val startedTest =
       session
-        .createTestContext(test.shortIdentifier, identifierSuffix)
+        .createTestContext(s"${test.shortIdentifier}_$repetition", identifierSuffix)
         .flatMap { context =>
           val start = System.nanoTime()
           val result = test(context).map(_ => Duration.fromNanos(System.nanoTime() - start))

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

@@ -22,6 +22,7 @@ private[testtool] abstract class LedgerTestSuite {
       participants: ParticipantAllocation,
       timeoutScale: Double = 1.0,
       runConcurrently: Boolean = true,
+      repeated: Int = 1,
   )(testCase: ExecutionContext => Participants => Future[Unit]): Unit = {
     val shortIdentifierRef = Ref.LedgerString.assertFromString(shortIdentifier)
     testCaseBuffer.append(
@@ -31,6 +32,7 @@ private[testtool] abstract class LedgerTestSuite {
         description,
         timeoutScale,
         runConcurrently,
+        repeated,
         participants,
         testCase,
       )

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

@@ -0,0 +1,392 @@
+// 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.testtool.infrastructure.Allocation._
+import com.daml.ledger.api.testtool.infrastructure.Assertions._
+import com.daml.ledger.api.testtool.infrastructure.LedgerTestSuite
+import com.daml.ledger.api.testtool.infrastructure.participant.ParticipantTestContext
+import com.daml.ledger.api.v1.transaction.{TransactionTree, TreeEvent}
+import com.daml.ledger.api.v1.value.RecordField
+import com.daml.ledger.client.binding.Primitive
+import com.daml.ledger.test.semantic.RaceTests._
+import com.daml.lf.data.{Bytes, Ref}
+import com.daml.timer.Delayed
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
+import scala.util.Success
+
+final class RaceConditionIT extends LedgerTestSuite {
+
+  private val DefaultRepetitionsNumber: Int = 5
+  private val WaitBeforeGettingTransactions = 1.second
+
+  raceConditionTest(
+    "WWDoubleNonTransientCreate",
+    "Cannot concurrently create multiple non-transient contracts with the same key",
+  ) { implicit ec => ledger => alice =>
+    val Attempts = 5
+    val ExpectedNumberOfSuccessfulCreations = 1
+    Future
+      .traverse(1 to Attempts) { _ =>
+        ledger.create(alice, ContractWithKey(alice)).transform(Success(_))
+      }
+      .map { results =>
+        assertLength(
+          "Successful contract creations",
+          ExpectedNumberOfSuccessfulCreations,
+          results.filter(_.isSuccess),
+        )
+        ()
+      }
+  }
+
+  raceConditionTest(
+    "WWDoubleArchive",
+    "Cannot archive the same contract multiple times",
+  ) { implicit ec => ledger => alice =>
+    val Attempts = 5
+    val ExpectedNumberOfSuccessfulArchivals = 1
+    for {
+      contract <- ledger.create(alice, ContractWithKey(alice))
+      _ <- Future.traverse(1 to Attempts) { _ =>
+        ledger.exercise(alice, contract.exerciseContractWithKey_Archive).transform(Success(_))
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+      assertLength(
+        "Successful contract archivals",
+        ExpectedNumberOfSuccessfulArchivals,
+        transactions.filter(isArchival),
+      )
+      ()
+    }
+  }
+
+  raceConditionTest(
+    "WWArchiveVsNonTransientCreate",
+    "Cannot create a contract with a key if that key is still used by another contract",
+  ) { implicit ec => ledger => alice =>
+    /*
+    This test case is intended to catch a race condition ending up in two consecutive successful contract
+    create or archive commands. E.g.:
+    [create]  <wait>  [archive]-race-[create]
+    In case of a bug causing the second [create] to see a partial result of [archive] command we could end up
+    with two consecutive successful contract creations.
+     */
+    for {
+      contract <- ledger.create(alice, ContractWithKey(alice))
+      _ <- Delayed.by(1.second)(())
+      createFuture = ledger.create(alice, ContractWithKey(alice)).transform(Success(_))
+      exerciseFuture = ledger
+        .exercise(alice, contract.exerciseContractWithKey_Archive)
+        .transform(Success(_))
+      _ <- createFuture
+      _ <- exerciseFuture
+      _ <- ledger.create(
+        alice,
+        DummyContract(alice),
+      ) // Create a dummy contract to ensure that we're not stuck with previous commands
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      assert(
+        isCreateNonTransient(transactions.head),
+        "The first transaction is expected to be a contract creation",
+      )
+      assert(
+        transactions.exists(isCreateDummyContract),
+        "A dummy contract creation is missing. A possible reason might be reading the transactions stream in the test case before submitted commands had chance to be processed.",
+      )
+
+      val (_, valid) =
+        transactions.filterNot(isCreateDummyContract).tail.foldLeft((transactions.head, true)) {
+          case ((previousTx, isValidSoFar), currentTx) =>
+            if (isValidSoFar) {
+              val valid = (isArchival(previousTx) && isCreateNonTransient(
+                currentTx
+              )) || (isCreateNonTransient(previousTx) && isArchival(currentTx))
+              (currentTx, valid)
+            } else {
+              (previousTx, isValidSoFar)
+            }
+        }
+
+      if (!valid)
+        fail(
+          s"""Invalid transaction sequence: ${transactions.map(printTransaction).mkString("\n")}"""
+        )
+    }
+  }
+
+  raceConditionTest(
+    "RWTransientCreateVsNonTransientCreate",
+    "Cannot create a transient contract and a non-transient contract with the same key",
+  ) { implicit ec => ledger => alice =>
+    val Attempts = 100
+    for {
+      wrapper <- ledger.create(alice, CreateWrapper(alice))
+      _ <- Future.traverse(1 to Attempts) { attempt =>
+        if (attempt == Attempts) {
+          ledger.create(alice, ContractWithKey(alice)).map(_ => ()).transform(Success(_))
+        } else {
+          ledger
+            .exercise(alice, wrapper.exerciseCreateWrapper_CreateTransient)
+            .map(_ => ())
+            .transform(Success(_))
+        }
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      // We deliberately allow situations where no non-transient contract is created and verify the transactions
+      // order when such contract is actually created.
+      transactions.find(isCreateNonTransient).foreach { nonTransientCreateTransaction =>
+        transactions
+          .filter(isTransientCreate)
+          .foreach(assertTransactionOrder(_, nonTransientCreateTransaction))
+      }
+    }
+  }
+
+  raceConditionTest(
+    "RWArchiveVsNonConsumingChoice",
+    "Cannot exercise a choice after a contract archival",
+  ) { implicit ec => ledger => alice =>
+    val ArchiveAt = 5
+    val Attempts = 10
+    for {
+      contract <- ledger.create(alice, ContractWithKey(alice))
+      _ <- Future.traverse(1 to Attempts) { attempt =>
+        if (attempt == ArchiveAt) {
+          ledger.exercise(alice, contract.exerciseContractWithKey_Archive).transform(Success(_))
+        } else {
+          ledger.exercise(alice, contract.exerciseContractWithKey_Exercise).transform(Success(_))
+        }
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      val archivalTransaction = transactions
+        .find(isArchival)
+        .getOrElse(fail("No archival transaction found"))
+
+      transactions
+        .filter(isNonConsumingExercise)
+        .foreach(assertTransactionOrder(_, archivalTransaction))
+
+    }
+  }
+
+  raceConditionTest(
+    "RWArchiveVsFetch",
+    "Cannot fetch an archived contract",
+  ) { implicit ec => ledger => alice =>
+    val Attempts = 10
+    for {
+      contract <- ledger.create(alice, ContractWithKey(alice))
+      fetchConract <- ledger.create(alice, FetchWrapper(alice, contract))
+      _ <- Future.traverse(1 to Attempts) { attempt =>
+        if (attempt == Attempts) {
+          ledger.exercise(alice, contract.exerciseContractWithKey_Archive).transform(Success(_))
+        } else {
+          ledger.exercise(alice, fetchConract.exerciseFetchWrapper_Fetch).transform(Success(_))
+        }
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      val archivalTransaction = transactions
+        .find(isArchival)
+        .getOrElse(fail("No archival transaction found"))
+
+      transactions
+        .filter(isFetch)
+        .foreach(assertTransactionOrder(_, archivalTransaction))
+    }
+  }
+
+  raceConditionTest(
+    "RWArchiveVsLookupByKey",
+    "Cannot successfully lookup by key an archived contract",
+  ) { implicit ec => ledger => alice =>
+    val ArchiveAt = 90
+    val Attempts = 100
+    for {
+      contract <- ledger.create(alice, ContractWithKey(alice))
+      looker <- ledger.create(alice, LookupWrapper(alice))
+      _ <- Future.traverse(1 to Attempts) { attempt =>
+        if (attempt == ArchiveAt) {
+          ledger.exercise(alice, contract.exerciseContractWithKey_Archive).transform(Success(_))
+        } else {
+          ledger.exercise(alice, looker.exerciseLookupWrapper_Lookup).transform(Success(_))
+        }
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      val archivalTransaction =
+        transactions.find(isArchival).getOrElse(fail("No archival transaction found"))
+
+      transactions
+        .filter(isSuccessfulContractLookup(success = true))
+        .foreach(assertTransactionOrder(_, archivalTransaction))
+    }
+  }
+
+  raceConditionTest(
+    "RWArchiveVsFailedLookupByKey",
+    "Lookup by key cannot fail after a contract creation",
+  ) { implicit ec => ledger => alice =>
+    val CreateAt = 90
+    val Attempts = 100
+    for {
+      looker <- ledger.create(alice, LookupWrapper(alice))
+      _ <- Future.traverse(1 to Attempts) { attempt =>
+        if (attempt == CreateAt) {
+          ledger.create(alice, ContractWithKey(alice)).transform(Success(_))
+        } else {
+          ledger.exercise(alice, looker.exerciseLookupWrapper_Lookup).transform(Success(_))
+        }
+      }
+      transactions <- transactions(ledger, alice)
+    } yield {
+      import TransactionUtil._
+
+      val createNonTransientTransaction = transactions
+        .find(isCreateNonTransient)
+        .getOrElse(fail("No create-non-transient transaction found"))
+
+      transactions
+        .filter(isSuccessfulContractLookup(success = false))
+        .foreach(assertTransactionOrder(_, createNonTransientTransaction))
+    }
+  }
+
+  private def raceConditionTest(
+      shortIdentifier: String,
+      description: String,
+  )(testCase: ExecutionContext => ParticipantTestContext => Primitive.Party => Future[Unit]): Unit =
+    test(
+      shortIdentifier = shortIdentifier,
+      description = description,
+      participants = allocate(SingleParty),
+      repeated = DefaultRepetitionsNumber,
+      runConcurrently = false,
+    )(implicit ec => { case Participants(Participant(ledger, party)) =>
+      testCase(ec)(ledger)(party)
+    })
+
+  private object TransactionUtil {
+    private implicit class TransactionTreeTestOps(tx: TransactionTree) {
+      def hasEventsNumber(expectedNumberOfEvents: Int): Boolean =
+        tx.eventsById.size == expectedNumberOfEvents
+
+      def containsEvent(condition: TreeEvent => Boolean): Boolean =
+        tx.eventsById.values.toList.exists(condition)
+    }
+
+    private def isCreated(templateName: String)(event: TreeEvent): Boolean =
+      event.kind.isCreated && event.getCreated.templateId.exists(_.entityName == templateName)
+
+    private def isExerciseEvent(choiceName: String)(event: TreeEvent): Boolean =
+      event.kind.isExercised && event.getExercised.choice == choiceName
+
+    def isCreateDummyContract(tx: TransactionTree): Boolean =
+      tx.containsEvent(isCreated(RaceTests.DummyContract.TemplateName))
+
+    def isCreateNonTransient(tx: TransactionTree): Boolean =
+      tx.hasEventsNumber(1) &&
+        tx.containsEvent(isCreated(RaceTests.ContractWithKey.TemplateName))
+
+    def isTransientCreate(tx: TransactionTree): Boolean =
+      tx.containsEvent(isExerciseEvent(RaceTests.CreateWrapper.ChoiceCreateTransient)) &&
+        tx.containsEvent(isCreated(RaceTests.ContractWithKey.TemplateName))
+
+    def isArchival(tx: TransactionTree): Boolean =
+      tx.hasEventsNumber(1) &&
+        tx.containsEvent(isExerciseEvent(RaceTests.ContractWithKey.ChoiceArchive))
+
+    def isNonConsumingExercise(tx: TransactionTree): Boolean =
+      tx.hasEventsNumber(1) &&
+        tx.containsEvent(isExerciseEvent(RaceTests.ContractWithKey.ChoiceExercise))
+
+    def isFetch(tx: TransactionTree): Boolean =
+      tx.hasEventsNumber(1) &&
+        tx.containsEvent(isExerciseEvent(RaceTests.FetchWrapper.ChoiceFetch))
+
+    private def isFoundContractField(found: Boolean)(field: RecordField) = {
+      field.label == "found" && field.value.exists(_.getBool == found)
+    }
+
+    def isSuccessfulContractLookup(success: Boolean)(tx: TransactionTree): Boolean =
+      tx.containsEvent { event =>
+        isCreated(RaceTests.LookupResult.TemplateName)(event) &&
+        event.getCreated.getCreateArguments.fields.exists(isFoundContractField(found = success))
+      }
+
+  }
+
+  private def transactions(ledger: ParticipantTestContext, party: Primitive.Party)(implicit
+      ec: ExecutionContext
+  ) =
+    Delayed.by(WaitBeforeGettingTransactions)(()).flatMap(_ => ledger.transactionTrees(party))
+
+  private def assertTransactionOrder(
+      expectedFirst: TransactionTree,
+      expectedSecond: TransactionTree,
+  ): Unit = {
+    if (offsetLessThan(expectedFirst.offset, expectedSecond.offset)) ()
+    else fail(s"""Offset ${expectedFirst.offset} is not before ${expectedSecond.offset}
+         |
+         |Expected first: ${printTransaction(expectedFirst)}
+         |Expected second: ${printTransaction(expectedSecond)}
+         |""".stripMargin)
+  }
+
+  private def printTransaction(transactionTree: TransactionTree): String = {
+    s"""Offset: ${transactionTree.offset}, number of events: ${transactionTree.eventsById.size}
+       |${transactionTree.eventsById.values.map(e => s" -> $e").mkString("\n")}
+       |""".stripMargin
+  }
+
+  private def offsetLessThan(a: String, b: String): Boolean =
+    Bytes.ordering.lt(offsetBytes(a), offsetBytes(b))
+
+  private def offsetBytes(offset: String): Bytes = {
+    Bytes.fromHexString(Ref.HexString.assertFromString(offset))
+  }
+
+  private object RaceTests {
+    object ContractWithKey {
+      val TemplateName = "ContractWithKey"
+      val ChoiceArchive = "ContractWithKey_Archive"
+      val ChoiceExercise = "ContractWithKey_Exercise"
+    }
+
+    object DummyContract {
+      val TemplateName = "DummyContract"
+    }
+
+    object FetchWrapper {
+      val ChoiceFetch = "FetchWrapper_Fetch"
+    }
+
+    object LookupResult {
+      val TemplateName = "LookupResult"
+    }
+
+    object CreateWrapper {
+      val ChoiceCreateTransient = "CreateWrapper_CreateTransient"
+    }
+  }
+
+}

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

@@ -32,6 +32,7 @@ object Tests {
       new PackageManagementServiceIT,
       new PackageServiceIT,
       new PartyManagementServiceIT,
+      new RaceConditionIT,
       new SemanticTests,
       new TransactionServiceIT,
       new WitnessesIT,

ledger/sandbox-on-x/BUILD.bazel

@@ -141,6 +141,7 @@ conformance_test(
         "--exclude=CommandDeduplicationIT",
         "--exclude=ContractKeysIT",
         "--exclude=SemanticTests",
+        "--exclude=RaceConditionIT",
     ],
 )
 

ledger/test-common/src/main/daml/semantic/RaceTests.daml

@@ -0,0 +1,71 @@
+-- Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+-- SPDX-License-Identifier: Apache-2.0
+
+module RaceTests where
+
+type RaceKey = Text
+
+template DummyContract with
+    owner : Party
+  where
+    signatory owner
+
+template ContractWithKey with
+    owner : Party
+  where
+    signatory owner
+    key owner: Party
+    maintainer key
+
+    controller owner can
+      ContractWithKey_Archive : ()
+        do
+          return ()
+
+    controller owner can
+      nonconsuming ContractWithKey_Exercise : ()
+        do
+          return ()
+
+template FetchWrapper with
+    fetcher : Party
+    contractId : ContractId ContractWithKey
+  where 
+    signatory fetcher
+    controller fetcher can
+      nonconsuming FetchWrapper_Fetch: ContractWithKey
+        do fetch contractId
+
+template LookupResult with
+    owner : Party
+    found : Bool
+  where
+    signatory owner
+
+foundContract (result : Optional (ContractId ContractWithKey)) : Bool =
+  case result of
+    Some val -> True
+    None -> False
+
+template LookupWrapper with
+    owner : Party
+  where
+    signatory owner
+
+    controller owner can
+      nonconsuming LookupWrapper_Lookup : ()
+        do
+          optionalContractId <- lookupByKey @ContractWithKey owner
+          create LookupResult with owner = owner, found = foundContract(optionalContractId)
+          pure ()
+
+template CreateWrapper with
+    owner : Party
+  where
+    signatory owner
+    controller owner can
+      nonconsuming CreateWrapper_CreateTransient : ()
+        do
+          contract <- create ContractWithKey with owner
+          _ <- exercise contract ContractWithKey_Archive
+          return ()