Enable multi-party submissions [KVL-708] (#8266)

* Remove single-party check

CHANGELOG_BEGIN
- [Ledger API] The ledger API now supports multi-party submissions.
  In order to use multi-party submissions, use the new act_as and
  read_as fields in submission requests.
CHANGELOG_END

* Remove usage of temporary SubmitterInfo methods

* Fix validator tests

* Fix auth test

* Add multi-party tests to the ledger API

* Fix compile errors

* Improve tests

* Remove temporary single-party method from SubmitterInfo

* Remove temporary single-party method from SubmitterInfo companion object

* Remove temporary single-party method from TxEntry

* Run multi-party submission ITs for ledger-on-memory and ledger-on-sql

* Minor improvement

Co-authored-by: Kamil Bozek <kamil.bozek@digitalasset.com>

ledger/daml-on-sql/BUILD.bazel

@@ -105,6 +105,20 @@ conformance_test(
     ],
 )
 
+conformance_test(
+    name = "conformance-test-multi-party-submissions",
+    server = ":daml-on-sql-ephemeral-postgresql",
+    server_args = [
+        "--ledgerid=conformance-test",
+        "--port=6865",
+        "--eager-package-loading",
+    ],
+    test_tool_args = [
+        "--verbose",
+        "--include=MultiPartySubmissionIT",
+    ],
+)
+
 genrule(
     name = "docs",
     srcs = [

ledger/ledger-api-common/src/main/scala/com/digitalasset/ledger/api/validation/CommandsValidator.scala

@@ -205,20 +205,11 @@ object CommandsValidator {
     def actAsMustNotBeEmpty(effectiveActAs: Set[Ref.Party]) =
       Either.cond(effectiveActAs.nonEmpty, (), missingField("party or act_as"))
 
-    // Temporary check to reject all multi-party submissions until they are implemented
-    // Note: when removing this method, also remove the call to `actAs.head` in validateCommands()
-    def requireSingleSubmitter(actAs: Set[Ref.Party], readAs: Set[Ref.Party]) =
-      if (actAs.size > 1 || readAs.nonEmpty)
-        Left(unimplemented("Multi-party submissions are not supported"))
-      else
-        Right(())
-
     val submitters = effectiveSubmitters(commands)
     for {
       actAs <- requireParties(submitters.actAs)
       readAs <- requireParties(submitters.readAs)
       _ <- actAsMustNotBeEmpty(actAs)
-      _ <- requireSingleSubmitter(actAs, readAs)
     } yield Submitters(actAs, readAs)
   }
 }

ledger/ledger-api-common/src/test/suite/scala/com/digitalasset/ledger/api/validation/SubmitRequestValidatorTest.scala

@@ -19,7 +19,8 @@ import com.daml.ledger.api.v1.value.Value.Sum
 import com.daml.ledger.api.v1.value.{List => ApiList, Map => ApiMap, Optional => ApiOptional, _}
 import com.google.protobuf.duration.Duration
 import com.google.protobuf.empty.Empty
-import io.grpc.Status.Code.{INVALID_ARGUMENT, UNAVAILABLE, UNIMPLEMENTED}
+import io.grpc.Status.Code.{INVALID_ARGUMENT, UNAVAILABLE}
+import org.scalatest.EitherValues._
 import org.scalatest.wordspec.AnyWordSpec
 import org.scalatest.prop.TableDrivenPropertyChecks
 import scalaz.syntax.tag._
@@ -190,30 +191,26 @@ class SubmitRequestValidatorTest
         )
       }
 
-      "not allow multiple submitters" in {
-        requestMustFailWith(
-          commandsValidator
-            .validateCommands(
-              api.commands.withParty("alice").addActAs("bob"),
-              internal.ledgerTime,
-              internal.submittedAt,
-              Some(internal.maxDeduplicationTime)),
-          UNIMPLEMENTED,
-          "Multi-party submissions are not supported",
-        )
-      }
-
-      "not allow readAs parties" in {
-        requestMustFailWith(
-          commandsValidator
-            .validateCommands(
-              api.commands.withParty("charlie").addReadAs("bob"),
-              internal.ledgerTime,
-              internal.submittedAt,
-              Some(internal.maxDeduplicationTime)),
-          UNIMPLEMENTED,
-          "Multi-party submissions are not supported",
-        )
+      "correctly read and deduplicate multiple submitters" in {
+        val result = commandsValidator
+          .validateCommands(
+            api.commands
+              .withParty("alice")
+              .addActAs("bob")
+              .addReadAs("alice")
+              .addReadAs("charlie")
+              .addReadAs("bob"),
+            internal.ledgerTime,
+            internal.submittedAt,
+            Some(internal.maxDeduplicationTime),
+          )
+        inside(result.right.value) {
+          case cmd: ApiCommands =>
+            // actAs parties are gathered from "party" and "readAs" fields
+            cmd.actAs shouldEqual Set("alice", "bob")
+            // readAs should exclude all parties that are already actAs parties
+            cmd.readAs shouldEqual Set("charlie")
+        }
       }
 
       "tolerate a single submitter specified in the actAs fields" in {
@@ -222,7 +219,8 @@ class SubmitRequestValidatorTest
             api.commands.withParty("").addActAs(api.submitter),
             internal.ledgerTime,
             internal.submittedAt,
-            Some(internal.maxDeduplicationTime)) shouldEqual Right(internal.emptyCommands)
+            Some(internal.maxDeduplicationTime),
+          ) shouldEqual Right(internal.emptyCommands)
       }
 
       "tolerate a single submitter specified in party, actAs, and readAs fields" in {
@@ -231,7 +229,7 @@ class SubmitRequestValidatorTest
             api.commands.withParty(api.submitter).addActAs(api.submitter).addReadAs(api.submitter),
             internal.ledgerTime,
             internal.submittedAt,
-            Some(internal.maxDeduplicationTime)
+            Some(internal.maxDeduplicationTime),
           ) shouldEqual Right(internal.emptyCommands)
       }
 

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

@@ -37,6 +37,7 @@ object Tests {
   val optional: Vector[LedgerTestSuite] =
     Vector(
       new ParticipantPruningIT,
+      new MultiPartySubmissionIT,
     )
 
   val retired: Vector[LedgerTestSuite] =

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

@@ -478,6 +478,15 @@ private[testtool] final class ParticipantTestContext private[participant] (
       .map(extractContracts)
       .map(_.head)
 
+  def create[T](
+      actAs: List[Party],
+      readAs: List[Party],
+      template: Template[T],
+  ): Future[Primitive.ContractId[T]] =
+    submitAndWaitForTransaction(submitAndWaitRequest(actAs, readAs, template.create.command))
+      .map(extractContracts)
+      .map(_.head)
+
   def createAndGetTransactionId[T](
       party: Party,
       template: Template[T],
@@ -494,6 +503,13 @@ private[testtool] final class ParticipantTestContext private[participant] (
   ): Future[TransactionTree] =
     submitAndWaitForTransactionTree(submitAndWaitRequest(party, exercise(party).command))
 
+  def exercise[T](
+      actAs: List[Party],
+      readAs: List[Party],
+      exercise: => Primitive.Update[T],
+  ): Future[TransactionTree] =
+    submitAndWaitForTransactionTree(submitAndWaitRequest(actAs, readAs, exercise.command))
+
   def exerciseForFlatTransaction[T](
       party: Party,
       exercise: Party => Primitive.Update[T],
@@ -538,6 +554,20 @@ private[testtool] final class ParticipantTestContext private[participant] (
       ),
     )
 
+  def submitRequest(actAs: List[Party], readAs: List[Party], commands: Command*): SubmitRequest =
+    new SubmitRequest(
+      Some(
+        new Commands(
+          ledgerId = ledgerId,
+          applicationId = applicationId,
+          commandId = nextCommandId(),
+          actAs = Party.unsubst(actAs),
+          readAs = Party.unsubst(readAs),
+          commands = commands,
+        ),
+      ),
+    )
+
   def submitRequest(party: Party, commands: Command*): SubmitRequest =
     new SubmitRequest(
       Some(
@@ -551,6 +581,23 @@ private[testtool] final class ParticipantTestContext private[participant] (
       ),
     )
 
+  def submitAndWaitRequest(
+      actAs: List[Party],
+      readAs: List[Party],
+      commands: Command*): SubmitAndWaitRequest =
+    new SubmitAndWaitRequest(
+      Some(
+        new Commands(
+          ledgerId = ledgerId,
+          applicationId = applicationId,
+          commandId = nextCommandId(),
+          actAs = Party.unsubst(actAs),
+          readAs = Party.unsubst(readAs),
+          commands = commands,
+        ),
+      ),
+    )
+
   def submitAndWaitRequest(party: Party, commands: Command*): SubmitAndWaitRequest =
     new SubmitAndWaitRequest(
       Some(

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

@@ -0,0 +1,422 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.testtool.tests
+
+import java.util.UUID
+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
+import com.daml.ledger.api.testtool.infrastructure.participant.ParticipantTestContext
+import com.daml.ledger.client.binding.Primitive
+import com.daml.ledger.client.binding.Primitive.{Party, List => PList}
+import com.daml.ledger.test.model.Test._
+import io.grpc.Status
+
+import scala.concurrent.{ExecutionContext, Future}
+
+final class MultiPartySubmissionIT extends LedgerTestSuite {
+
+  test(
+    "MPSSubmit",
+    "Submit creates a multi-party contract",
+    allocate(Parties(2)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob)) =>
+      // Create a contract for (Alice, Bob)
+      val request = ledger.submitRequest(
+        actAs = List(alice, bob),
+        readAs = List.empty,
+        commands = MultiPartyContract(PList(alice, bob), "").create.command,
+      )
+
+      for {
+        _ <- ledger.submit(request)
+        completions <- ledger.firstCompletions(bob)
+      } yield {
+        assert(completions.length == 1)
+        assert(completions.head.commandId == request.commands.get.commandId)
+      }
+  })
+
+  test(
+    "MPSCreateSuccess",
+    "Create succeeds with sufficient authorization",
+    allocate(Parties(2)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob)) =>
+      for {
+        // Create a contract for (Alice, Bob)
+        _ <- ledger.create(
+          actAs = List(alice, bob),
+          readAs = List.empty,
+          template = MultiPartyContract(PList(alice, bob), ""),
+        )
+      } yield ()
+  })
+
+  test(
+    "MPSCreateInsufficientAuthorization",
+    "Create fails with insufficient authorization",
+    allocate(Parties(3)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie)) =>
+      for {
+        // Create a contract for (Alice, Bob, Charlie), but only submit as (Alice, Bob).
+        // Should fail because required authorizer Charlie is missing from submitters.
+        failure <- ledger
+          .create(
+            actAs = List(alice, bob),
+            readAs = List.empty,
+            template = MultiPartyContract(PList(alice, bob, charlie), ""),
+          )
+          .failed
+      } yield {
+        assertGrpcError(
+          failure,
+          Status.Code.INVALID_ARGUMENT,
+          None,
+        )
+      }
+  })
+
+  test(
+    "MPSAddSignatoriesSuccess",
+    "Exercise AddSignatories succeeds with sufficient authorization",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create a contract for (Alice, Bob)
+        (contract, _) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Exercise a choice to add (Charlie, David)
+        // Requires authorization from all four parties
+        _ <- ledger.exercise(
+          actAs = List(alice, bob, charlie, david),
+          readAs = List.empty,
+          exercise =
+            contract.exerciseMPAddSignatories(unusedActor, PList(alice, bob, charlie, david)),
+        )
+      } yield ()
+  })
+
+  test(
+    "MPSAddSignatoriesInsufficientAuthorization",
+    "Exercise AddSignatories fails with insufficient authorization",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create a contract for (Alice, Bob)
+        (contract, _) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Exercise a choice to add (Charlie, David) to the list of signatories
+        // Should fail as it's missing authorization from one of the original signatories (Alice)
+        failure <- ledger
+          .exercise(
+            actAs = List(bob, charlie, david),
+            readAs = List.empty,
+            exercise =
+              contract.exerciseMPAddSignatories(unusedActor, PList(alice, bob, charlie, david)),
+          )
+          .failed
+      } yield {
+        assertGrpcError(
+          failure,
+          Status.Code.INVALID_ARGUMENT,
+          None,
+        )
+      }
+  })
+
+  test(
+    "MPSFetchOtherSuccess",
+    "Exercise FetchOther succeeds with sufficient authorization and read delegation",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create contract A for (Alice, Bob)
+        (contractA, _) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Create contract B for (Alice, Bob, Charlie, David)
+        (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+        // Fetch contract A through contract B as (Charlie, David)
+        _ <- ledger.exercise(
+          actAs = List(charlie, david),
+          readAs = List(alice),
+          exercise = contractB.exerciseMPFetchOther(unusedActor, contractA, PList(charlie, david)),
+        )
+      } yield ()
+  })
+
+  test(
+    "MPSFetchOtherInsufficientAuthorization",
+    "Exercise FetchOther fails with insufficient authorization",
+    allocate(Parties(4)),
+  )(
+    implicit ec => {
+      case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+        for {
+          // Create contract A for (Alice, Bob)
+          (contractA, _) <- createMultiPartyContract(ledger, List(alice, bob))
+
+          // Create contract B for (Charlie, David)
+          (contractB, _) <- createMultiPartyContract(ledger, List(charlie, david))
+
+          // Fetch contract A through contract B as (Charlie, David)
+          // Should fail with an authorization error
+          failure <- ledger
+            .exercise(
+              actAs = List(charlie, david),
+              readAs = List(bob, alice),
+              exercise =
+                contractB.exerciseMPFetchOther(unusedActor, contractA, PList(charlie, david)),
+            )
+            .failed
+        } yield {
+          assertGrpcError(
+            failure,
+            Status.Code.INVALID_ARGUMENT,
+            Some(
+              Pattern.compile("of the fetched contract to be an authorizer, but authorizers were")),
+          )
+        }
+    })
+
+  test(
+    "MPSFetchOtherInvisible",
+    "Exercise FetchOther fails because the contract isn't visible",
+    allocate(Parties(4)),
+  )(
+    implicit ec => {
+      case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+        for {
+          // Create contract A for (Alice, Bob)
+          (contractA, _) <- createMultiPartyContract(ledger, List(alice, bob))
+
+          // Create contract B for (Alice, Bob, Charlie, David)
+          (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+          // Fetch contract A through contract B as (Charlie, David)
+          // Should fail with an interpretation error because the fetched contract isn't visible to any submitter
+          failure <- ledger
+            .exercise(
+              actAs = List(charlie, david),
+              readAs = List.empty,
+              exercise =
+                contractB.exerciseMPFetchOther(unusedActor, contractA, PList(charlie, david)),
+            )
+            .failed
+        } yield {
+          assertGrpcError(
+            failure,
+            Status.Code.INVALID_ARGUMENT,
+            Some(Pattern.compile("dependency error: couldn't find contract")),
+          )
+        }
+    })
+
+  test(
+    "MPSFetchOtherByKeyOtherSuccess",
+    "Exercise FetchOtherByKey succeeds with sufficient authorization and read delegation",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create contract A for (Alice, Bob)
+        (_, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Create contract B for (Alice, Bob, Charlie, David)
+        (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+        // Fetch contract A through contract B as (Charlie, David)
+        _ <- ledger.exercise(
+          actAs = List(charlie, david),
+          readAs = List(alice),
+          exercise = contractB.exerciseMPFetchOtherByKey(unusedActor, keyA, PList(charlie, david)),
+        )
+      } yield ()
+  })
+
+  test(
+    "MPSFetchOtherByKeyInsufficientAuthorization",
+    "Exercise FetchOtherByKey fails with insufficient authorization",
+    allocate(Parties(4)),
+  )(
+    implicit ec => {
+      case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+        for {
+          // Create contract A for (Alice, Bob)
+          (_, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+          // Create contract B for (Charlie, David)
+          (contractB, _) <- createMultiPartyContract(ledger, List(charlie, david))
+
+          // Fetch contract A through contract B as (Charlie, David)
+          // Should fail with an authorization error
+          failure <- ledger
+            .exercise(
+              actAs = List(charlie, david),
+              readAs = List(bob, alice),
+              exercise =
+                contractB.exerciseMPFetchOtherByKey(unusedActor, keyA, PList(charlie, david)),
+            )
+            .failed
+        } yield {
+          assertGrpcError(
+            failure,
+            Status.Code.INVALID_ARGUMENT,
+            Some(
+              Pattern.compile("of the fetched contract to be an authorizer, but authorizers were")),
+          )
+        }
+    })
+
+  test(
+    "MPSFetchOtherByKeyInvisible",
+    "Exercise FetchOtherByKey fails because the contract isn't visible",
+    allocate(Parties(4)),
+  )(
+    implicit ec => {
+      case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+        for {
+          // Create contract A for (Alice, Bob)
+          (_, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+          // Create contract B for (Alice, Bob, Charlie, David)
+          (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+          // Fetch contract A through contract B as (Charlie, David)
+          // Should fail with an interpretation error because the fetched contract isn't visible to any submitter
+          failure <- ledger
+            .exercise(
+              actAs = List(charlie, david),
+              readAs = List.empty,
+              exercise =
+                contractB.exerciseMPFetchOtherByKey(unusedActor, keyA, PList(charlie, david)),
+            )
+            .failed
+        } yield {
+          assertGrpcError(
+            failure,
+            Status.Code.INVALID_ARGUMENT,
+            Some(Pattern.compile("dependency error: couldn't find key")),
+          )
+        }
+    })
+
+  test(
+    "MPSLookupOtherByKeyOtherSuccess",
+    "Exercise LookupOtherByKey succeeds with sufficient authorization and read delegation",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create contract A for (Alice, Bob)
+        (contractA, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Create contract B for (Alice, Bob, Charlie, David)
+        (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+        // Fetch contract A through contract B as (Charlie, David)
+        _ <- ledger.exercise(
+          actAs = List(charlie, david),
+          readAs = List(alice),
+          exercise = contractB
+            .exerciseMPLookupOtherByKey(unusedActor, keyA, PList(charlie, david), Some(contractA)),
+        )
+      } yield ()
+  })
+
+  test(
+    "MPSLookupOtherByKeyInsufficientAuthorization",
+    "Exercise LookupOtherByKey fails with insufficient authorization",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create contract A for (Alice, Bob)
+        (contractA, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Create contract B for (Charlie, David)
+        (contractB, _) <- createMultiPartyContract(ledger, List(charlie, david))
+
+        // Fetch contract A through contract B as (Charlie, David)
+        // Should fail with an authorization error
+        failure <- ledger
+          .exercise(
+            actAs = List(charlie, david),
+            readAs = List(bob, alice),
+            exercise = contractB.exerciseMPLookupOtherByKey(
+              unusedActor,
+              keyA,
+              PList(charlie, david),
+              Some(contractA)),
+          )
+          .failed
+      } yield {
+        assertGrpcError(
+          failure,
+          Status.Code.INVALID_ARGUMENT,
+          Some(Pattern.compile("requires authorizers (.*) for lookup by key, but it only has")),
+        )
+      }
+  })
+
+  test(
+    "MPSLookupOtherByKeyInvisible",
+    "Exercise LookupOtherByKey fails because the contract isn't visible",
+    allocate(Parties(4)),
+  )(implicit ec => {
+    case Participants(Participant(ledger, alice, bob, charlie, david)) =>
+      for {
+        // Create contract A for (Alice, Bob)
+        (contractA, keyA) <- createMultiPartyContract(ledger, List(alice, bob))
+
+        // Create contract B for (Alice, Bob, Charlie, David)
+        (contractB, _) <- createMultiPartyContract(ledger, List(alice, bob, charlie, david))
+
+        // Fetch contract A through contract B as (Charlie, David)
+        // Should fail with an interpretation error because the fetched contract isn't visible to any submitter
+        failure <- ledger
+          .exercise(
+            actAs = List(charlie, david),
+            readAs = List.empty,
+            exercise = contractB.exerciseMPLookupOtherByKey(
+              unusedActor,
+              keyA,
+              PList(charlie, david),
+              Some(contractA)),
+          )
+          .failed
+      } yield {
+        assertGrpcError(
+          failure,
+          Status.Code.INVALID_ARGUMENT,
+          Some(Pattern.compile("User abort: LookupOtherByKey value matches")),
+        )
+      }
+  })
+
+  private[this] def createMultiPartyContract(
+      ledger: ParticipantTestContext,
+      submitters: List[Party],
+      value: String = UUID.randomUUID().toString,
+  )(implicit ec: ExecutionContext)
+    : Future[(Primitive.ContractId[MultiPartyContract], MultiPartyContract)] =
+    ledger
+      .create(
+        actAs = submitters,
+        readAs = List.empty,
+        template = MultiPartyContract(submitters, value),
+      )
+      .map(cid => cid -> MultiPartyContract(submitters, value))
+
+  // The "actor" argument in the generated methods to exercise choices is not used
+  private[this] val unusedActor: Party = Party("")
+}

ledger/ledger-on-memory/BUILD.bazel

@@ -230,6 +230,21 @@ conformance_test(
     ],
 )
 
+conformance_test(
+    name = "conformance-test-multi-party-submission",
+    ports = [6865],
+    server = ":app",
+    server_args = [
+        "--contract-id-seeding=testing-weak",
+        "--participant participant-id=example,port=6865",
+        "--batching enable=true,max-batch-size-bytes=262144",
+    ],
+    test_tool_args = [
+        "--verbose",
+        "--include=MultiPartySubmissionIT",
+    ],
+)
+
 conformance_test(
     name = "benchmark-performance-envelope",
     ports = [6865],

ledger/ledger-on-sql/BUILD.bazel

@@ -268,6 +268,20 @@ da_scala_test_suite(
                 "--include=ParticipantPruningIT",
             ],
         ),
+        conformance_test(
+            name = "conformance-test-multi-party-submission-{}".format(db["name"]),
+            ports = [6865],
+            server = ":conformance-test-{}-bin".format(db["name"]),
+            server_args = [
+                "--contract-id-seeding=testing-weak",
+                "--participant participant-id=conformance-test,port=6865",
+            ] + db.get("conformance_test_server_args", []),
+            tags = db.get("conformance_test_tags", []),
+            test_tool_args = db.get("conformance_test_tool_args", []) + [
+                "--verbose",
+                "--include=MultiPartySubmissionIT",
+            ],
+        ),
         conformance_test(
             name = "benchmark-performance-envelope-{}".format(db["name"]),
             ports = [6865],

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

@@ -50,7 +50,7 @@ class KeyValueSubmission(metrics: Metrics) {
       DamlSubmission.newBuilder
         .addInputDamlState(commandDedupKey(encodedSubInfo))
         .addInputDamlState(configurationStateKey)
-        .addInputDamlState(partyStateKey(submitterInfo.singleSubmitterOrThrow()))
+        .addAllInputDamlState(submitterInfo.actAs.map(partyStateKey).asJava)
         .addAllInputDamlState(inputDamlStateFromTx.asJava)
         .setTransactionEntry(
           DamlTransactionEntry.newBuilder

ledger/participant-state/kvutils/src/test/lib/scala/com/daml/ledger/participant/state/kvutils/KVTest.scala

@@ -407,8 +407,8 @@ object KVTest {
       deduplicationTime: Duration,
       recordTime: Timestamp,
   ): SubmitterInfo =
-    SubmitterInfo.withSingleSubmitter(
-      submitter = submitter,
+    SubmitterInfo(
+      actAs = List(submitter),
       applicationId = Ref.LedgerString.assertFromString("test"),
       commandId = commandId,
       deduplicateUntil = recordTime.addMicros(deduplicationTime.toNanos / 1000).toInstant,

ledger/participant-state/kvutils/src/test/lib/scala/com/daml/ledger/participant/state/kvutils/ParticipantStateIntegrationSpecBase.scala

@@ -683,8 +683,8 @@ abstract class ParticipantStateIntegrationSpecBase(implementationName: String)(
   }
 
   private def submitterInfo(party: Ref.Party, commandId: String = "X") =
-    SubmitterInfo.withSingleSubmitter(
-      submitter = party,
+    SubmitterInfo(
+      actAs = List(party),
       applicationId = Ref.LedgerString.assertFromString("tests"),
       commandId = Ref.LedgerString.assertFromString(commandId),
       deduplicateUntil = inTheFuture(10.seconds).toInstant,

ledger/participant-state/kvutils/src/test/suite/scala/com/daml/ledger/participant/state/kvutils/api/KeyValueParticipantStateWriterSpec.scala

@@ -146,8 +146,8 @@ object KeyValueParticipantStateWriterSpec {
   }
 
   private def submitterInfo(recordTime: Timestamp, party: Ref.Party, commandId: String) =
-    SubmitterInfo.withSingleSubmitter(
-      submitter = party,
+    SubmitterInfo(
+      actAs = List(party),
       applicationId = Ref.LedgerString.assertFromString("tests"),
       commandId = Ref.LedgerString.assertFromString(commandId),
       deduplicateUntil = recordTime.addMicros(Duration.ofDays(1).toNanos / 1000).toInstant,

ledger/participant-state/kvutils/tools/engine-benchmark/src/benchmark/scala/ledger/participant/state/kvutils/tools/engine/benchmark/Replay.scala

@@ -41,15 +41,7 @@ final case class TxEntry(
     ledgerTime: Time.Timestamp,
     submissionTime: Time.Timestamp,
     submissionSeed: crypto.Hash,
-) {
-  // Note: this method will be removed when the entire kvutils code base
-  // supports multi-party submissions
-  def singleSubmitterOrThrow(): Ref.Party =
-    if (submitters.length == 1)
-      submitters.head
-    else
-      sys.error("Multi-party submissions are not supported")
-}
+)
 
 final case class BenchmarkState(
     name: String,

ledger/participant-state/src/main/scala/com/daml/ledger/participant/state/v1/SubmitterInfo.scala

@@ -35,29 +35,4 @@ final case class SubmitterInfo(
     applicationId: ApplicationId,
     commandId: CommandId,
     deduplicateUntil: Instant,
-) {
-  // Note: this function is only available temporarily until the entire DAML code base
-  // supports multi-party submissions. Use at your own risk.
-  def singleSubmitterOrThrow(): Party = {
-    if (actAs.length == 1)
-      actAs.head
-    else
-      throw new RuntimeException("SubmitterInfo contains more than one acting party")
-  }
-}
-
-object SubmitterInfo {
-  // Note: this function is only available temporarily until the entire DAML code base
-  // supports multi-party submissions. Use at your own risk.
-  def withSingleSubmitter(
-      submitter: Party,
-      applicationId: ApplicationId,
-      commandId: CommandId,
-      deduplicateUntil: Instant,
-  ): SubmitterInfo = SubmitterInfo.apply(
-    actAs = List(submitter),
-    applicationId = applicationId,
-    commandId = commandId,
-    deduplicateUntil = deduplicateUntil,
-  )
-}
+)

ledger/sandbox-classic/src/main/scala/platform/sandbox/stores/LedgerBackedWriteService.scala

@@ -24,7 +24,6 @@ import com.daml.lf.data.Time
 import com.daml.logging.LoggingContext
 import com.daml.logging.LoggingContext.withEnrichedLoggingContext
 import com.daml.platform.sandbox.stores.ledger.{Ledger, PartyIdGenerator}
-
 import io.grpc.Status
 
 import scala.compat.java8.FutureConverters
@@ -42,7 +41,7 @@ private[stores] final class LedgerBackedWriteService(ledger: Ledger, timeProvide
       estimatedInterpretationCost: Long,
   ): CompletionStage[SubmissionResult] =
     withEnrichedLoggingContext(
-      "submitter" -> submitterInfo.singleSubmitterOrThrow(),
+      "actAs" -> submitterInfo.actAs.mkString(","),
       "applicationId" -> submitterInfo.applicationId,
       "commandId" -> submitterInfo.commandId,
       "deduplicateUntil" -> submitterInfo.deduplicateUntil.toString,

ledger/sandbox-classic/src/test/lib/scala/platform/sandbox/auth/MultiPartyServiceCallAuthTests.scala

@@ -126,40 +126,35 @@ trait MultiPartyServiceCallAuthTests extends ServiceCallAuthTests {
   }
 
   it should "allow multi-party calls authorized to exactly the required parties" in {
-    // Note: use expectSuccess() once multi-party submissions are enabled
-    expectUnimplemented(
+    expectSuccess(
       serviceCallFor(
         TokenParties(actAs, readAs),
         RequestSubmitters("", actAs, readAs),
       ))
   }
   it should "allow multi-party calls authorized to a superset of the required parties" in {
-    // Note: use expectSuccess() once multi-party submissions are enabled
-    expectUnimplemented(
+    expectSuccess(
       serviceCallFor(
         TokenParties(randomActAs ++ actAs, randomReadAs ++ readAs),
         RequestSubmitters("", actAs, readAs),
       ))
   }
   it should "allow multi-party calls with all parties authorized in read-write mode" in {
-    // Note: use expectSuccess() once multi-party submissions are enabled
-    expectUnimplemented(
+    expectSuccess(
       serviceCallFor(
         TokenParties(actAs ++ readAs, List.empty),
         RequestSubmitters("", actAs, readAs),
       ))
   }
   it should "allow multi-party calls with actAs parties spread across party and actAs fields" in {
-    // Note: use expectSuccess() once multi-party submissions are enabled
-    expectUnimplemented(
+    expectSuccess(
       serviceCallFor(
         TokenParties(actAs, readAs),
         RequestSubmitters(actAs.head, actAs.tail, readAs),
       ))
   }
   it should "allow multi-party calls with actAs parties duplicated in the readAs field" in {
-    // Note: use expectSuccess() once multi-party submissions are enabled
-    expectUnimplemented(
+    expectSuccess(
       serviceCallFor(
         TokenParties(actAs, readAs),
         RequestSubmitters("", actAs, actAs ++ readAs),

ledger/sandbox-classic/src/test/suite/scala/platform/sandbox/stores/ledger/TransactionTimeModelComplianceIT.scala

@@ -86,8 +86,8 @@ class TransactionTimeModelComplianceIT
   private[this] def publishTxAt(ledger: Ledger, ledgerTime: Instant, commandId: String) = {
     val dummyTransaction = TransactionBuilder.EmptySubmitted
 
-    val submitterInfo = SubmitterInfo.withSingleSubmitter(
-      submitter = Ref.Party.assertFromString("submitter"),
+    val submitterInfo = SubmitterInfo(
+      actAs = List(Ref.Party.assertFromString("submitter")),
       applicationId = Ref.LedgerString.assertFromString("appId"),
       commandId = Ref.LedgerString.assertFromString(commandId + UUID.randomUUID().toString),
       deduplicateUntil = Instant.EPOCH
@@ -114,7 +114,7 @@ class TransactionTimeModelComplianceIT
           Some(offset),
           None,
           com.daml.ledger.api.domain.ApplicationId(submitterInfo.applicationId),
-          Set(submitterInfo.singleSubmitterOrThrow())
+          submitterInfo.actAs.toSet,
         )
         .filter(_._2.completions.head.commandId == submitterInfo.commandId)
         .runWith(Sink.head)

ledger/test-common/src/main/daml/model/Test.daml

@@ -613,3 +613,49 @@ template Asset
       AssetTransfer: ContractId Asset
         with newOwner: Party
         do create this with owner = newOwner
+
+template MultiPartyContract
+  with
+    parties: [Party]
+    value: Text
+  where
+    signatory parties
+
+    key this: MultiPartyContract
+    maintainer key.parties
+
+    choice MPAddSignatories : ContractId MultiPartyContract
+      with
+        newParties: [Party]
+      controller parties ++ newParties
+      do
+        create this with
+          parties = parties ++ newParties
+
+    nonconsuming choice MPFetchOther : ()
+      with
+        cid: ContractId MultiPartyContract
+        actors: [Party]
+      controller actors
+      do
+        actualContract <- fetch cid
+        return ()
+
+    nonconsuming choice MPFetchOtherByKey : ()
+      with
+        keyToFetch: MultiPartyContract
+        actors: [Party]
+      controller actors
+      do
+        (actualCid, actualContract) <- fetchByKey @MultiPartyContract keyToFetch
+        return ()
+
+    nonconsuming choice MPLookupOtherByKey : ()
+      with
+        keyToFetch: MultiPartyContract
+        actors: [Party]
+        expectedCid: Optional (ContractId MultiPartyContract)
+      controller actors
+      do
+        actualCid <- lookupByKey @MultiPartyContract keyToFetch
+        assertMsg "LookupOtherByKey value matches" (expectedCid == actualCid)