KV: do not use "Inputs" part of InputsAndEffects

daml-lf/transaction-test-lib/src/main/scala/lf/transaction/test/TransactionBuilder.scala

@@ -60,7 +60,7 @@ final class TransactionBuilder(
   def build(): Tx.Transaction = ids.synchronized {
     import TransactionVersion.Ordering
     val finalNodes = nodes.transform {
-      case (nid, rb: Node.NodeRollback[NodeId]) =>
+      case (nid, rb: TxRollBack) =>
         rb.copy(children = children(nid).toImmArray)
       case (nid, exe: TxExercise) =>
         exe.copy(children = children(nid).toImmArray)
@@ -226,6 +226,7 @@ object TransactionBuilder {
   type TxExercise = Node.NodeExercises[NodeId, ContractId]
   type TxRollback = Node.NodeRollback[NodeId]
   type TxKeyWithMaintainers = Node.KeyWithMaintainers[TxValue]
+  type TxRollBack = Node.NodeRollback[NodeId]
 
   private val Create = Node.NodeCreate
   private val Exercise = Node.NodeExercises

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

@@ -384,6 +384,27 @@ sealed abstract class HasTxNodes[Nid, +Cid] {
       case (acc, _) => acc
     } -- localContracts.keySet
 
+  /** Return all the contract keys referenced by this transaction.
+    * This includes the keys created, exercised, fetched, or looked up, even those
+    * that refer transient contracts or that appear under a rollback node.
+    */
+  final def contractKeys(implicit
+      ev: HasTxNodes[Nid, Cid] <:< HasTxNodes[_, Value.ContractId]
+  ): Set[GlobalKey] = {
+    ev(this).fold(Set.empty[GlobalKey]) {
+      case (acc, (_, node: Node.NodeCreate[Value.ContractId])) =>
+        node.key.fold(acc)(key => acc + GlobalKey.assertBuild(node.templateId, key.key))
+      case (acc, (_, node: Node.NodeExercises[_, Value.ContractId])) =>
+        node.key.fold(acc)(key => acc + GlobalKey.assertBuild(node.templateId, key.key))
+      case (acc, (_, node: Node.NodeFetch[Value.ContractId])) =>
+        node.key.fold(acc)(key => acc + GlobalKey.assertBuild(node.templateId, key.key))
+      case (acc, (_, node: Node.NodeLookupByKey[Value.ContractId])) =>
+        acc + GlobalKey.assertBuild(node.templateId, node.key.key)
+      case (acc, (_, _: Node.NodeRollback[_])) =>
+        acc
+    }
+  }
+
   // This method visits to all nodes of the transaction in execution order.
   // Exercise/rollback nodes are visited twice: when execution reaches them and when execution leaves their body.
   // On the first visit of an execution/rollback node, the caller can prevent traversal of the children

daml-lf/transaction/src/test/scala/com/digitalasset/daml/lf/transaction/TransactionSpec.scala

@@ -12,6 +12,7 @@ import com.daml.lf.transaction.GenTransaction.{
   OrphanedNode,
 }
 import com.daml.lf.transaction.Node.{GenNode, NodeCreate, NodeExercises, NodeRollback}
+import com.daml.lf.transaction.test.TransactionBuilder
 import com.daml.lf.value.Value.ContractId
 import com.daml.lf.value.{Value => V}
 import com.daml.lf.value.test.ValueGenerators.danglingRefGenNode
@@ -25,6 +26,7 @@ import scala.language.implicitConversions
 import scala.util.Random
 
 class TransactionSpec extends AnyFreeSpec with Matchers with ScalaCheckDrivenPropertyChecks {
+
   import TransactionSpec._
 
   "isWellFormed" - {
@@ -194,10 +196,12 @@ class TransactionSpec extends AnyFreeSpec with Matchers with ScalaCheckDrivenPro
 
     "fail if version is different" in {
       val versions = TransactionVersion.All
+
       def diffVersion(v: TransactionVersion) = {
         val randomVersion = versions(Random.nextInt(versions.length - 1))
         if (randomVersion != v) randomVersion else versions.last
       }
+
       forAll(genEmptyNode, minSuccessful(10)) { n =>
         val m = n.updateVersion(diffVersion(n.version))
         isReplayedBy(n, m) shouldBe Symbol("left")
@@ -268,6 +272,108 @@ class TransactionSpec extends AnyFreeSpec with Matchers with ScalaCheckDrivenPro
 
     }
   }
+
+  "contractKeys" - {
+    "return all the contract keys" in {
+      // TODO: https://github.com/digital-asset/daml/issues/8020
+      // change VDev to  TransactionVersion.StableVersions.max once exception are released
+      val builder = TransactionBuilder(TransactionVersion.VDev)
+      val parties = List("Alice")
+
+      def create(s: String) = builder
+        .create(
+          id = s"#$s",
+          template = s"-pkg-:Mod:$s",
+          argument = V.ValueUnit,
+          signatories = parties,
+          observers = parties,
+          key = Some(V.ValueText(s)),
+        )
+
+      def exe(s: String, consuming: Boolean, byKey: Boolean) =
+        builder
+          .exercise(
+            contract = create(s),
+            choice = s"Choice$s",
+            actingParties = parties.toSet,
+            consuming = consuming,
+            argument = V.ValueUnit,
+            byKey = byKey,
+          )
+
+      def fetch(s: String, byKey: Boolean) =
+        builder.fetch(contract = create(s), byKey = byKey)
+
+      def lookup(s: String, found: Boolean) =
+        builder.lookupByKey(contract = create(s), found = found)
+
+      val root1 =
+        builder.create(
+          "#root",
+          template = "-pkg-:Mod:Root",
+          argument = V.ValueUnit,
+          signatories = parties,
+          observers = parties,
+          key = None,
+        )
+
+      val root2 = builder.exercise(
+        root1,
+        "ExerciseRoot",
+        actingParties = parties.toSet,
+        consuming = true,
+        argument = V.ValueUnit,
+      )
+
+      builder.add(root1)
+      val exeId = builder.add(root2)
+      builder.add(create("Create"))
+      builder.add(exe("NonConsumingExerciseById", false, false), exeId)
+      builder.add(exe("ConsumingExerciseById", true, false), exeId)
+      builder.add(exe("NonConsumingExerciseByKey", false, true), exeId)
+      builder.add(exe("NonConsumingExerciseByKey", true, true), exeId)
+      builder.add(fetch("FetchById", false), exeId)
+      builder.add(fetch("FetchByKey", true), exeId)
+      builder.add(lookup("SuccessfulLookup", true), exeId)
+      builder.add(lookup("UnsuccessfulLookup", true), exeId)
+      val rollbackId = builder.add(Node.NodeRollback(ImmArray.empty, root2.version))
+      builder.add(create("RolledBackCreate"))
+      builder.add(exe("RolledBackNonConsumingExerciseById", false, false), rollbackId)
+      builder.add(exe("RolledBackConsumingExerciseById", true, false), rollbackId)
+      builder.add(exe("RolledBackNonConsumingExerciseByKey", false, true), rollbackId)
+      builder.add(exe("RolledBackNonConsumingExerciseByKey", true, true), rollbackId)
+      builder.add(fetch("RolledBackFetchById", false), rollbackId)
+      builder.add(fetch("RolledBackFetchByKey", true), rollbackId)
+      builder.add(lookup("RolledBackSuccessfulLookup", true), rollbackId)
+      builder.add(lookup("RolledBackUnsuccessfulLookup", true), rollbackId)
+
+      val expectedResults =
+        Iterator(
+          "Create",
+          "NonConsumingExerciseById",
+          "ConsumingExerciseById",
+          "NonConsumingExerciseByKey",
+          "NonConsumingExerciseByKey",
+          "FetchById",
+          "FetchByKey",
+          "SuccessfulLookup",
+          "UnsuccessfulLookup",
+          "RolledBackCreate",
+          "RolledBackNonConsumingExerciseById",
+          "RolledBackConsumingExerciseById",
+          "RolledBackNonConsumingExerciseByKey",
+          "RolledBackNonConsumingExerciseByKey",
+          "RolledBackFetchById",
+          "RolledBackFetchByKey",
+          "RolledBackSuccessfulLookup",
+          "RolledBackUnsuccessfulLookup",
+        ).map(s =>
+          GlobalKey.assertBuild(Ref.Identifier.assertFromString(s"-pkg-:Mod:$s"), V.ValueText(s))
+        ).toSet
+
+      builder.build().contractKeys shouldBe expectedResults
+    }
+  }
 }
 
 object TransactionSpec {

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

@@ -56,19 +56,21 @@ private[state] object Conversions {
       .build
   }
 
+  def encodeContractKey(tmplId: Identifier, key: Value[ContractId]): DamlContractKey =
+    encodeGlobalKey(
+      GlobalKey
+        .build(tmplId, key)
+        .fold(msg => throw Err.InvalidSubmission(msg), identity)
+    )
+
   def decodeIdentifier(protoIdent: ValueOuterClass.Identifier): Identifier =
-    ValueCoder
-      .decodeIdentifier(protoIdent)
-      .getOrElse(
-        throw Err
-          .DecodeError("Identifier", s"Cannot decode identifier: $protoIdent")
-      )
+    assertDecode("Identifier", ValueCoder.decodeIdentifier(protoIdent))
 
-  def globalKeyToStateKey(key: GlobalKey): DamlStateKey = {
-    DamlStateKey.newBuilder
-      .setContractKey(encodeGlobalKey(key))
-      .build
-  }
+  def globalKeyToStateKey(key: GlobalKey): DamlStateKey =
+    DamlStateKey.newBuilder.setContractKey(encodeGlobalKey(key)).build
+
+  def contractKeyToStateKey(templateId: Identifier, key: Value[ContractId]): DamlStateKey =
+    DamlStateKey.newBuilder.setContractKey(encodeContractKey(templateId, key)).build
 
   def commandDedupKey(subInfo: DamlSubmitterInfo): DamlStateKey = {
     val sortedUniqueSubmitters =

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

@@ -8,9 +8,11 @@ import com.daml.ledger.participant.state.kvutils.Conversions._
 import com.daml.ledger.participant.state.kvutils.DamlKvutils._
 import com.daml.ledger.participant.state.v1._
 import com.daml.lf.data.Time.Timestamp
+import com.daml.lf.value.Value.ContractId
 import com.daml.metrics.Metrics
 import com.google.protobuf.ByteString
 
+import scala.collection.compat.immutable.LazyList
 import scala.jdk.CollectionConverters._
 
 /** Methods to produce the [[DamlSubmission]] message.
@@ -37,21 +39,36 @@ class KeyValueSubmission(metrics: Metrics) {
       effects.createdContracts.map(_._1) ++ effects.consumedContracts
     }
 
+  private def submissionParties(
+      submitterInfo: SubmitterInfo,
+      tx: SubmittedTransaction,
+  ): Set[Party] =
+    tx.informees ++ submitterInfo.actAs
+
   /** Prepare a transaction submission. */
   def transactionToSubmission(
       submitterInfo: SubmitterInfo,
       meta: TransactionMeta,
       tx: SubmittedTransaction,
   ): DamlSubmission =
     metrics.daml.kvutils.submission.conversion.transactionToSubmission.time { () =>
-      val inputDamlStateFromTx = InputsAndEffects.computeInputs(tx, meta)
       val encodedSubInfo = buildSubmitterInfo(submitterInfo)
+      val packageIdStates = meta.optUsedPackages
+        .getOrElse(
+          throw new InternalError("Transaction was not annotated with used packages")
+        )
+        .map(Conversions.packageStateKey)
+      val partyStates = submissionParties(submitterInfo, tx).toList.map(Conversions.partyStateKey)
+      val contractIdStates = tx.inputContracts[ContractId].map(Conversions.contractIdToStateKey)
+      val contractKeyStates = tx.contractKeys.map(Conversions.globalKeyToStateKey)
 
       DamlSubmission.newBuilder
         .addInputDamlState(commandDedupKey(encodedSubInfo))
         .addInputDamlState(configurationStateKey)
-        .addAllInputDamlState(submitterInfo.actAs.map(partyStateKey).asJava)
-        .addAllInputDamlState(inputDamlStateFromTx.asJava)
+        .addAllInputDamlState(packageIdStates.asJava)
+        .addAllInputDamlState(partyStates.asJava)
+        .addAllInputDamlState(contractIdStates.asJava)
+        .addAllInputDamlState(contractKeyStates.asJava)
         .setTransactionEntry(
           DamlTransactionEntry.newBuilder
             .setTransaction(Conversions.encodeTransaction(tx))