[Explicit disclosure] Update Read- and WriteService interfaces for explicit disclosure [DPP-1094] (#14341)

* Update Read- and WriteService with explicit disclosure payloads

changelog_begin
changelog_end

* Addressed review comments

ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiSubmissionService.scala

@@ -5,7 +5,6 @@ package com.daml.platform.apiserver.services
 
 import java.time.{Duration, Instant}
 import java.util.UUID
-
 import com.daml.api.util.TimeProvider
 import com.daml.error.ErrorCode.LoggedApiException
 import com.daml.error.definitions.{ErrorCause, LedgerApiErrors, RejectionGenerators}
@@ -17,7 +16,7 @@ import com.daml.ledger.configuration.Configuration
 import com.daml.ledger.participant.state.index.v2._
 import com.daml.ledger.participant.state.{v2 => state}
 import com.daml.lf.crypto
-import com.daml.lf.data.Ref
+import com.daml.lf.data.{ImmArray, Ref}
 import com.daml.lf.transaction.SubmittedTransaction
 import com.daml.logging.LoggingContext.withEnrichedLoggingContext
 import com.daml.logging.{ContextualizedLogger, LoggingContext}
@@ -260,6 +259,7 @@ private[apiserver] final class ApiSubmissionService private[services] (
         result.transaction,
         result.interpretationTimeNanos,
         result.globalKeyMapping,
+        ImmArray.empty,
       )
       .toScalaUnwrapped
   }

ledger/participant-integration-api/src/main/scala/platform/store/dao/JdbcLedgerDao.scala

@@ -535,6 +535,7 @@ private class JdbcLedgerDao(
               recordTime = recordTime,
               divulgedContracts = divulgedContracts.toList,
               blindingInfo = blindingInfo,
+              contractMetadata = Map.empty,
             )
           ),
         )

ledger/participant-integration-api/src/test/lib/scala/platform/indexer/ha/EndlessReadService.scala

@@ -100,6 +100,7 @@ case class EndlessReadService(
               recordTime = recordTime(i),
               divulgedContracts = List.empty,
               blindingInfo = None,
+              contractMetadata = Map.empty,
             )
           case i =>
             offset(i) -> Update.TransactionAccepted(
@@ -110,6 +111,7 @@ case class EndlessReadService(
               recordTime = recordTime(i),
               divulgedContracts = List.empty,
               blindingInfo = None,
+              contractMetadata = Map.empty,
             )
         }
         .via(killSwitch.flow)

ledger/participant-integration-api/src/test/suite/scala/platform/index/InMemoryStateUpdaterSpec.scala

@@ -134,6 +134,7 @@ object InMemoryStateUpdaterSpec {
     recordTime = Timestamp.Epoch,
     divulgedContracts = List.empty,
     blindingInfo = None,
+    contractMetadata = Map.empty,
   )
   private val update2 = offset(2L) -> Update.ConfigurationChanged(
     Timestamp.Epoch,
@@ -149,6 +150,7 @@ object InMemoryStateUpdaterSpec {
     recordTime = Timestamp.Epoch,
     divulgedContracts = List.empty,
     blindingInfo = None,
+    contractMetadata = Map.empty,
   )
 
   private val txLogUpdate1 = TransactionLogUpdate.Transaction(

ledger/participant-integration-api/src/test/suite/scala/platform/indexer/parallel/ParallelIndexerSubscriptionSpec.scala

@@ -206,6 +206,7 @@ class ParallelIndexerSubscriptionSpec extends AnyFlatSpec with Matchers {
       recordTime = someRecordTime,
       divulgedContracts = List.empty,
       blindingInfo = None,
+      contractMetadata = Map.empty,
     )
 
     val expected: Vector[DbDto.TransactionMetering] = Vector(

ledger/participant-integration-api/src/test/suite/scala/platform/store/backend/UpdateToDbDtoSpec.scala

@@ -291,6 +291,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -367,6 +368,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -453,6 +455,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -538,6 +541,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -649,6 +653,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -799,6 +804,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -861,6 +867,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -963,6 +970,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -1094,6 +1102,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
             divulgence = Map(createNode.coid -> Set(Ref.Party.assertFromString("divulgee"))),
           )
         ),
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -1200,6 +1209,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -1264,6 +1274,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
         recordTime = someRecordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
       val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
         someOffset
@@ -1392,6 +1403,7 @@ class UpdateToDbDtoSpec extends AnyWordSpec with Matchers {
             recordTime = someRecordTime,
             divulgedContracts = List.empty,
             blindingInfo = None,
+            contractMetadata = Map.empty,
           )
           val dtos = UpdateToDbDto(someParticipantId, valueSerialization, compressionStrategy)(
             someOffset

ledger/participant-integration-api/src/test/suite/scala/platform/store/backend/UpdateToMeteringDbDtoSpec.scala

@@ -64,6 +64,7 @@ class UpdateToMeteringDbDtoSpec extends AnyWordSpec {
       recordTime = someRecordTime,
       divulgedContracts = List.empty,
       blindingInfo = None,
+      Map.empty,
     )
 
     "extract transaction metering" in {

ledger/participant-state-metrics/src/main/scala/com/daml/ledger/participant/state/v2/metrics/TimedWriteService.scala

@@ -4,7 +4,6 @@
 package com.daml.ledger.participant.state.v2.metrics
 
 import java.util.concurrent.CompletionStage
-
 import com.daml.daml_lf_dev.DamlLf
 import com.daml.ledger.api.health.HealthStatus
 import com.daml.ledger.configuration.Configuration
@@ -16,8 +15,9 @@ import com.daml.ledger.participant.state.v2.{
   TransactionMeta,
   WriteService,
 }
-import com.daml.lf.data.{Ref, Time}
+import com.daml.lf.command.DisclosedContract
-import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction}
+import com.daml.lf.data.{ImmArray, Ref, Time}
+import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction, Versioned}
 import com.daml.lf.value.Value
 import com.daml.logging.LoggingContext
 import com.daml.metrics.{Metrics, Timed}
@@ -31,6 +31,7 @@ final class TimedWriteService(delegate: WriteService, metrics: Metrics) extends
       transaction: SubmittedTransaction,
       estimatedInterpretationCost: Long,
       globalKeyMapping: Map[GlobalKey, Option[Value.ContractId]],
+      explicitlyDisclosedContracts: ImmArray[Versioned[DisclosedContract]],
   )(implicit
       loggingContext: LoggingContext,
       telemetryContext: TelemetryContext,
@@ -44,6 +45,7 @@ final class TimedWriteService(delegate: WriteService, metrics: Metrics) extends
         transaction,
         estimatedInterpretationCost,
         globalKeyMapping,
+        ImmArray.empty,
       ),
     )
 

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

@@ -4,14 +4,14 @@
 package com.daml.ledger.participant.state.v2
 
 import java.time.Duration
-
 import com.daml.daml_lf_dev.DamlLf
 import com.daml.ledger.api.DeduplicationPeriod
 import com.daml.ledger.configuration.Configuration
 import com.daml.ledger.grpc.GrpcStatuses
-import com.daml.lf.data.Ref
+import com.daml.lf.data.{Bytes, Ref}
 import com.daml.lf.data.Time.Timestamp
 import com.daml.lf.transaction.{BlindingInfo, CommittedTransaction}
+import com.daml.lf.value.Value
 import com.daml.logging.entries.{LoggingEntry, LoggingValue, ToLoggingValue}
 import com.google.rpc.status.{Status => RpcStatus}
 
@@ -245,6 +245,12 @@ object Update {
     *                          determines how this transaction's recordTime relates to its
     *                          [[TransactionMeta.ledgerEffectiveTime]].
     * @param divulgedContracts List of divulged contracts. See [[DivulgedContract]] for details.
+    * @param contractMetadata  For each contract created in this transaction, this map may contain
+    *                          contract metadata assigned by the ledger implementation.
+    *                          This data is opaque and can only be used in [[com.daml.lf.command.DisclosedContract]]s
+    *                          when submitting transactions trough the [[WriteService]].
+    *                          If a contract created by this transaction is not element of this map,
+    *                          its metadata is equal to the empty byte array.
     */
   final case class TransactionAccepted(
       optCompletionInfo: Option[CompletionInfo],
@@ -254,6 +260,7 @@ object Update {
       recordTime: Timestamp,
       divulgedContracts: List[DivulgedContract],
       blindingInfo: Option[BlindingInfo],
+      contractMetadata: Map[Value.ContractId, Bytes],
   ) extends Update {
     override def description: String = s"Accept transaction $transactionId"
   }
@@ -268,6 +275,7 @@ object Update {
             recordTime,
             _,
             _,
+            _,
           ) =>
         LoggingValue.Nested.fromEntries(
           Logging.recordTime(recordTime),

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

@@ -4,9 +4,10 @@
 package com.daml.ledger.participant.state.v2
 
 import java.util.concurrent.CompletionStage
-
 import com.daml.ledger.api.health.ReportsHealth
-import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction}
+import com.daml.lf.command.DisclosedContract
+import com.daml.lf.data.ImmArray
+import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction, Versioned}
 import com.daml.lf.value.Value
 import com.daml.logging.LoggingContext
 import com.daml.telemetry.TelemetryContext
@@ -99,6 +100,7 @@ trait WriteService
     * @param globalKeyMapping            Input key mapping inferred by interpretation.
     *                                    The map should contain all contract keys that were used during interpretation.
     *                                    A value of None means no contract was found with this contract key.
+    * @param explicitlyDisclosedContracts  Explicitly disclosed contracts used during interpretation.
     */
   def submitTransaction(
       submitterInfo: SubmitterInfo,
@@ -106,6 +108,7 @@ trait WriteService
       transaction: SubmittedTransaction,
       estimatedInterpretationCost: Long,
       globalKeyMapping: Map[GlobalKey, Option[Value.ContractId]],
+      explicitlyDisclosedContracts: ImmArray[Versioned[DisclosedContract]],
   )(implicit
       loggingContext: LoggingContext,
       telemetryContext: TelemetryContext,

ledger/sandbox-on-x/src/main/scala/com/daml/ledger/sandbox/BridgeWriteService.scala

@@ -16,14 +16,15 @@ import com.daml.ledger.offset.Offset
 import com.daml.ledger.participant.state.v2._
 import com.daml.ledger.sandbox.bridge.{BridgeMetrics, LedgerBridge}
 import com.daml.ledger.sandbox.domain.{Rejection, Submission}
-import com.daml.lf.data.{Ref, Time}
+import com.daml.lf.command.DisclosedContract
-import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction}
+import com.daml.lf.data.{ImmArray, Ref, Time}
+import com.daml.lf.transaction.{GlobalKey, SubmittedTransaction, Versioned}
 import com.daml.logging.{ContextualizedLogger, LoggingContext}
 import com.daml.metrics.InstrumentedGraph
 import com.daml.telemetry.TelemetryContext
+
 import java.time.Duration
 import java.util.concurrent.{CompletableFuture, CompletionStage}
-
 import com.daml.lf.value.Value
 
 class BridgeWriteService(
@@ -49,6 +50,7 @@ class BridgeWriteService(
       transaction: SubmittedTransaction,
       estimatedInterpretationCost: Long,
       globalKeyMapping: Map[GlobalKey, Option[Value.ContractId]],
+      disclosedContracts: ImmArray[Versioned[DisclosedContract]],
   )(implicit
       loggingContext: LoggingContext,
       telemetryContext: TelemetryContext,

ledger/sandbox-on-x/src/main/scala/com/daml/ledger/sandbox/bridge/LedgerBridge.scala

@@ -146,6 +146,7 @@ object LedgerBridge {
       recordTime = currentTimestamp,
       divulgedContracts = Nil,
       blindingInfo = None,
+      contractMetadata = Map.empty,
     )
   }
 

ledger/sandbox-on-x/src/test/suite/scala/com/daml/ledger/sandbox/bridge/validate/ConflictCheckWithCommittedSpec.scala

@@ -10,8 +10,8 @@ import com.daml.ledger.configuration.{Configuration, LedgerTimeModel}
 import com.daml.ledger.offset.Offset
 import com.daml.ledger.participant.state.index.v2.{IndexService, MaximumLedgerTime}
 import com.daml.ledger.participant.state.v2.{SubmitterInfo, TransactionMeta}
-import com.daml.ledger.sandbox.bridge.validate.ConflictCheckWithCommittedSpec._
 import com.daml.ledger.sandbox.bridge.BridgeMetrics
+import com.daml.ledger.sandbox.bridge.validate.ConflictCheckWithCommittedSpec._
 import com.daml.ledger.sandbox.domain.Rejection._
 import com.daml.ledger.sandbox.domain.Submission
 import com.daml.lf.crypto.Hash
@@ -27,8 +27,8 @@ import org.mockito.{ArgumentMatchersSugar, MockitoSugar}
 import org.scalatest.FixtureContext
 import org.scalatest.flatspec.AsyncFlatSpec
 import org.scalatest.matchers.should.Matchers
-import java.time.Duration
 
+import java.time.Duration
 import scala.concurrent.Future
 
 class ConflictCheckWithCommittedSpec

ledger/sandbox-on-x/src/test/suite/scala/com/daml/ledger/sandbox/bridge/validate/SequenceSpec.scala

@@ -539,6 +539,7 @@ class SequenceSpec
         recordTime = recordTime,
         divulgedContracts = List.empty,
         blindingInfo = None,
+        contractMetadata = Map.empty,
       )
 
     def assertCommandRejected(