kvutils reports lookupByKey inconsistencies as Inconsistent (#7914)

CHANGELOG_BEGIN
kvutils reports LookupByKey node mismatches during validation as Inconsistent
instead of Disputed if they can be due to contention on the contract key
CHANGELOG_END

daml-lf/transaction/src/main/scala/com/digitalasset/daml/lf/transaction/Transaction.scala

@@ -761,7 +761,7 @@ object Transaction {
 
 }
 
-sealed abstract class ReplayMismatch[Nid, Cid] {
+sealed abstract class ReplayMismatch[Nid, Cid] extends Product with Serializable {
   def recordedTransaction: VersionedTransaction[Nid, Cid]
   def replayedTransaction: VersionedTransaction[Nid, Cid]
 

ledger/participant-state/kvutils/src/main/scala/com/daml/ledger/participant/state/kvutils/committer/TransactionCommitter.scala

@@ -17,7 +17,7 @@ import com.daml.lf.archive.Reader.ParseError
 import com.daml.lf.crypto
 import com.daml.lf.data.Ref.{PackageId, Party}
 import com.daml.lf.data.Time.Timestamp
-import com.daml.lf.engine.{Blinding, Engine}
+import com.daml.lf.engine.{Blinding, Engine, ReplayMismatch}
 import com.daml.lf.language.Ast
 import com.daml.lf.transaction.{
   BlindingInfo,
@@ -25,7 +25,9 @@ import com.daml.lf.transaction.{
   GlobalKeyWithMaintainers,
   Node,
   NodeId,
+  ReplayNodeMismatch,
   SubmittedTransaction,
+  VersionedTransaction,
   Transaction => Tx
 }
 import com.daml.lf.value.Value
@@ -238,11 +240,53 @@ private[kvutils] class TransactionCommitter(
             err =>
               reject[DamlTransactionEntrySummary](
                 commitContext.getRecordTime,
-                buildRejectionLogEntry(transactionEntry, RejectionReason.Disputed(err.msg))),
+                buildRejectionLogEntry(transactionEntry, rejectionReasonForValidationError(err))),
             _ => StepContinue[DamlTransactionEntrySummary](transactionEntry)
           )
       })
 
+  private[committer] def rejectionReasonForValidationError(
+      validationError: com.daml.lf.engine.Error): RejectionReason = {
+    def disputed: RejectionReason = RejectionReason.Disputed(validationError.msg)
+
+    def resultIsCreatedInTx(
+        tx: VersionedTransaction[NodeId, ContractId],
+        result: Option[Value.ContractId]): Boolean =
+      result.exists { contractId =>
+        tx.nodes.exists {
+          case (nodeId @ _, create: Node.NodeCreate[_, _]) => create.coid == contractId
+          case _ => false
+        }
+      }
+
+    validationError match {
+      case ReplayMismatch(
+          ReplayNodeMismatch(recordedTx, recordedNodeId, replayedTx, replayedNodeId)) =>
+        // If the problem is that a key lookup has changed and the results do not involve contracts created in this transaction,
+        // then it's a consistency problem.
+
+        (recordedTx.nodes(recordedNodeId), replayedTx.nodes(replayedNodeId)) match {
+          case (
+              Node.NodeLookupByKey(
+                recordedTemplateId,
+                recordedOptLocation @ _,
+                recordedKey,
+                recordedResult),
+              Node.NodeLookupByKey(
+                replayedTemplateId,
+                replayedOptLocation @ _,
+                replayedKey,
+                replayedResult))
+              if recordedTemplateId == replayedTemplateId && recordedKey == replayedKey
+                && !resultIsCreatedInTx(recordedTx, recordedResult)
+                && !resultIsCreatedInTx(replayedTx, replayedResult) =>
+            RejectionReason.Inconsistent(validationError.msg)
+          case _ => disputed
+        }
+      case _ => disputed
+    }
+  }
+
   /** Validate the submission's conformance to the DAML model */
   private[committer] def blind: Step =
     (commitContext, transactionEntry) => {
@@ -309,7 +353,7 @@ private[kvutils] class TransactionCommitter(
         recordTime,
         buildRejectionLogEntry(
           transactionEntry,
-          RejectionReason.Disputed("DuplicateKey: Contract Key not unique")))
+          RejectionReason.Inconsistent("DuplicateKey: Contract Key not unique")))
 
   }
 

ledger/participant-state/kvutils/src/test/suite/scala/com/daml/ledger/participant/state/kvutils/committer/TransactionCommitterSpec.scala

@@ -12,14 +12,24 @@ import com.daml.ledger.participant.state.kvutils.Conversions.buildTimestamp
 import com.daml.ledger.participant.state.kvutils.DamlKvutils._
 import com.daml.ledger.participant.state.kvutils.TestHelpers._
 import com.daml.ledger.participant.state.kvutils.committer.TransactionCommitter.DamlTransactionEntrySummary
-import com.daml.ledger.participant.state.v1.Configuration
+import com.daml.ledger.participant.state.v1.{Configuration, RejectionReason}
 import com.daml.lf.data.Time.Timestamp
-import com.daml.lf.engine.Engine
+import com.daml.lf.engine.{Engine, ReplayMismatch}
+import com.daml.lf.transaction
+import com.daml.lf.transaction.{
+  NodeId,
+  RecordedNodeMissing,
+  ReplayNodeMismatch,
+  ReplayedNodeMissing,
+  Transaction
+}
 import com.daml.lf.transaction.test.TransactionBuilder
+import com.daml.lf.value.Value
 import com.daml.metrics.Metrics
 import com.google.protobuf.ByteString
 import org.scalatest.mockito.MockitoSugar
 import org.scalatest.{Matchers, WordSpec}
+import org.scalatest.Inspectors.forEvery
 
 class TransactionCommitterSpec extends WordSpec with Matchers with MockitoSugar {
   private val metrics = new Metrics(new MetricRegistry)
@@ -264,6 +274,128 @@ class TransactionCommitterSpec extends WordSpec with Matchers with MockitoSugar
     }
   }
 
+  "rejectionReasonForValidationError" when {
+    val maintainer = "maintainer"
+    val dummyValue = TransactionBuilder.record("field" -> "value")
+
+    def create(contractId: String, key: String = "key"): TransactionBuilder.Create =
+      TransactionBuilder.create(
+        id = contractId,
+        template = "dummyPackage:DummyModule:DummyTemplate",
+        argument = dummyValue,
+        signatories = Seq(maintainer),
+        observers = Seq.empty,
+        key = Some(key)
+      )
+
+    def mkMismatch(
+        recorded: (Transaction.Transaction, NodeId),
+        replayed: (Transaction.Transaction, NodeId)): ReplayNodeMismatch[NodeId, Value.ContractId] =
+      ReplayNodeMismatch(recorded._1, recorded._2, replayed._1, replayed._2)
+    def mkRecordedMissing(
+        recorded: Transaction.Transaction,
+        replayed: (Transaction.Transaction, NodeId))
+      : RecordedNodeMissing[NodeId, Value.ContractId] =
+      RecordedNodeMissing(recorded, replayed._1, replayed._2)
+    def mkReplayedMissing(
+        recorded: (Transaction.Transaction, NodeId),
+        replayed: Transaction.Transaction): ReplayedNodeMissing[NodeId, Value.ContractId] =
+      ReplayedNodeMissing(recorded._1, recorded._2, replayed)
+
+    def checkRejectionReason(mkReason: String => RejectionReason)(
+        mismatch: transaction.ReplayMismatch[NodeId, Value.ContractId]) = {
+      val replayMismatch = ReplayMismatch(mismatch)
+      instance.rejectionReasonForValidationError(replayMismatch) shouldBe mkReason(
+        replayMismatch.msg)
+    }
+
+    val createInput = create("#inputContractId")
+    val create1 = create("#someContractId")
+    val create2 = create("#otherContractId")
+
+    val exercise = TransactionBuilder.exercise(
+      contract = createInput,
+      choice = "DummyChoice",
+      consuming = false,
+      actingParties = Set(maintainer),
+      argument = dummyValue,
+      byKey = false
+    )
+    val otherKeyCreate = create("#contractWithOtherKey", "otherKey")
+
+    val lookupNodes @ Seq(lookup1, lookup2, lookupNone, lookupOther @ _) =
+      Seq(create1 -> true, create2 -> true, create1 -> false, otherKeyCreate -> true) map {
+        case (create, found) => TransactionBuilder.lookupByKey(create, found)
+      }
+    val Seq(tx1, tx2, txNone, txOther) = lookupNodes map { node =>
+      val builder = TransactionBuilder()
+      val rootId = builder.add(exercise)
+      val lookupId = builder.add(node, rootId)
+      builder.build() -> lookupId
+    }
+
+    "there is a mismatch in lookupByKey nodes" should {
+      "report an inconsistency if the contracts are not created in the same transaction" in {
+        val inconsistentLookups = Seq(
+          mkMismatch(tx1, tx2),
+          mkMismatch(tx1, txNone),
+          mkMismatch(txNone, tx2),
+        )
+        forEvery(inconsistentLookups)(checkRejectionReason(RejectionReason.Inconsistent))
+      }
+
+      "report Disputed if one of contracts is created in the same transaction" in {
+        val Seq(txC1, txC2, txCNone) = Seq(lookup1, lookup2, lookupNone) map { node =>
+          val builder = TransactionBuilder()
+          val rootId = builder.add(exercise)
+          builder.add(create1, rootId)
+          val lookupId = builder.add(node, rootId)
+          builder.build() -> lookupId
+        }
+        val Seq(tx1C, txNoneC) = Seq(lookup1, lookupNone) map { node =>
+          val builder = TransactionBuilder()
+          val rootId = builder.add(exercise)
+          val lookupId = builder.add(node, rootId)
+          builder.add(create1)
+          builder.build() -> lookupId
+        }
+        val recordedKeyInconsistent = Seq(
+          mkMismatch(txC2, txC1),
+          mkMismatch(txCNone, txC1),
+          mkMismatch(txC1, txCNone),
+          mkMismatch(tx1C, txNoneC),
+        )
+        forEvery(recordedKeyInconsistent)(checkRejectionReason(RejectionReason.Disputed))
+      }
+
+      "report Disputed if the keys are different" in {
+        checkRejectionReason(RejectionReason.Disputed)(mkMismatch(txOther, tx1))
+      }
+    }
+
+    "the mismatch is not between two lookup nodes" should {
+      "report Disputed" in {
+        val txExerciseOnly = {
+          val builder = TransactionBuilder()
+          builder.add(exercise)
+          builder.build()
+        }
+        val txCreate = {
+          val builder = TransactionBuilder()
+          val rootId = builder.add(exercise)
+          val createId = builder.add(create1, rootId)
+          builder.build() -> createId
+        }
+        val miscMismatches = Seq(
+          mkMismatch(txCreate, tx1),
+          mkRecordedMissing(txExerciseOnly, tx2),
+          mkReplayedMissing(tx1, txExerciseOnly),
+        )
+        forEvery(miscMismatches)(checkRejectionReason(RejectionReason.Disputed))
+      }
+    }
+  }
+
   private def createTransactionCommitter(): TransactionCommitter =
     new TransactionCommitter(theDefaultConfig, mock[Engine], metrics, inStaticTimeMode = false)