Split scenario & ledger execution (#10039)

* Split scenario & ledger execution

This PR by no means aims to solve everything we can do here. It is
rather the minimal change that I could get to work that provides us
with one Speedy machine for scenario execution (which is still an
on-ledger machine to avoid having to change the callsites to much in
this PR) and one speedy machine per submission.

There is tons of cleanup we can do afterwards but this should
hopefully set the right foundations.

changelog_begin
changelog_end

compiler/scenario-service/server/src/main/scala/com/digitalasset/daml/lf/scenario/Context.scala

@@ -174,7 +174,7 @@ class Context(val contextId: Context.ContextId, languageVersion: LanguageVersion
     buildMachine(
       Identifier(PackageId.assertFromString(pkgId), QualifiedName.assertFromString(name))
     ).map { machine =>
-      ScenarioRunner(machine).run() match {
+      ScenarioRunner(machine, txSeeding).run() match {
         case Right((diff @ _, steps @ _, ledger, value)) =>
           (ledger, machine, Right(value))
         case Left((err, ledger)) =>

daml-lf/engine/src/main/scala/com/digitalasset/daml/lf/engine/Engine.scala

@@ -358,13 +358,8 @@ class Engine(val config: EngineConfig = new EngineConfig(LanguageVersion.StableV
             },
           )
 
-        case _: SResultScenarioCommit =>
-          return ResultError(Error.Interpretation.Generic("unexpected ScenarioCommit"))
-
-        case _: SResultScenarioInsertMustFail =>
-          return ResultError(Error.Interpretation.Generic("unexpected ScenarioInsertMustFail"))
-        case _: SResultScenarioMustFail =>
-          return ResultError(Error.Interpretation.Generic("unexpected ScenarioMustFail"))
+        case _: SResultScenarioSubmit =>
+          return ResultError(Error.Interpretation.Generic("unexpected SResultScenarioSubmit"))
         case _: SResultScenarioPassTime =>
           return ResultError(Error.Interpretation.Generic("unexpected ScenarioPassTime"))
         case _: SResultScenarioGetParty =>

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/Anf.scala

@@ -317,19 +317,6 @@ private[lf] object Anf {
       case SELet1General(rhs, body) =>
         Bounce(() => transformLet1(depth, env, rhs, body, k, transform))
 
-      case SECatchSubmitMustFail(body0) =>
-        Bounce(() =>
-          flattenExp(depth, env, body0) { body =>
-            Bounce(() =>
-              transform(
-                depth,
-                SECatchSubmitMustFail(body.wrapped),
-                k,
-              )
-            )
-          }
-        )
-
       case SELocation(loc, body) => {
         Bounce(() =>
           transformExp(depth, env, body, k) { (depth, body, txK) =>

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/Compiler.scala

@@ -806,9 +806,9 @@ private[lf] final class Compiler(
       case ScenarioBlock(bindings, body) =>
         compileBlock(bindings, body)
       case ScenarioCommit(partyE, updateE, _retType @ _) =>
-        compileCommit(partyE, updateE, optLoc)
+        compileCommit(partyE, updateE, optLoc, mustFail = false)
       case ScenarioMustFailAt(partyE, updateE, _retType @ _) =>
-        compileMustFail(partyE, updateE, optLoc)
+        compileCommit(partyE, updateE, optLoc, mustFail = true)
       case ScenarioGetTime =>
         SEGetTime
       case ScenarioGetParty(e) =>
@@ -820,41 +820,19 @@ private[lf] final class Compiler(
     }
 
   @inline
-  private[this] def compileCommit(partyE: Expr, updateE: Expr, optLoc: Option[Location]): SExpr =
+  private[this] def compileCommit(
+      partyE: Expr,
+      updateE: Expr,
+      optLoc: Option[Location],
+      mustFail: Boolean,
+  ): SExpr =
     // let party = <partyE>
     //     update = <updateE>
-    // in \token ->
-    //   let _ = $beginCommit party token
-    //       r = update token
-    //   in $endCommit(mustFail = false) r token
+    // in $submit(mustFail)(party, update)
     withEnv { _ =>
-      let(compile(partyE)) { partyPos =>
-        let(compile(updateE)) { updatePos =>
-          labeledUnaryFunction(Profile.SubmitLabel) { tokenPos =>
-            let(SBSBeginCommit(optLoc)(svar(partyPos), svar(tokenPos))) { _ =>
-              let(app(svar(updatePos), svar(tokenPos))) { resultPos =>
-                SBSEndCommit(mustFail = false)(svar(resultPos), svar(tokenPos))
-              }
-            }
-          }
-        }
-      }
-    }
-
-  @inline
-  private[this] def compileMustFail(party: Expr, update: Expr, optLoc: Option[Location]): SExpr =
-    // \token ->
-    //   let _ = $beginCommit [party] <token>
-    //       <r> = $catch ([update] <token>) true false
-    //   in $endCommit(mustFail = true) <r> <token>
-    withEnv { _ =>
-      labeledUnaryFunction(Profile.SubmitMustFailLabel) { tokenPos =>
-        let(SBSBeginCommit(optLoc)(compile(party), svar(tokenPos))) { _ =>
-          let(
-            SECatchSubmitMustFail(app(compile(update), svar(tokenPos)))
-          ) { resultPos =>
-            SBSEndCommit(mustFail = true)(svar(resultPos), svar(tokenPos))
-          }
+      let(compile(partyE)) { partyLoc =>
+        let(compile(updateE)) { updateLoc =>
+          SBSSubmit(optLoc, mustFail)(svar(partyLoc), svar(updateLoc))
         }
       }
     }
@@ -1134,11 +1112,6 @@ private[lf] final class Compiler(
           closureConvert(shift(remaps, bounds.length), body),
         )
 
-      case SECatchSubmitMustFail(body) =>
-        SECatchSubmitMustFail(
-          closureConvert(remaps, body)
-        )
-
       case SETryCatch(body, handler) =>
         SETryCatch(
           closureConvert(remaps, body),
@@ -1213,8 +1186,6 @@ private[lf] final class Compiler(
           bounds.zipWithIndex.foldLeft(go(body, bound + bounds.length, free)) {
             case (acc, (expr, idx)) => go(expr, bound + idx, acc)
           }
-        case SECatchSubmitMustFail(body) =>
-          go(body, bound, free)
         case SELabelClosure(_, expr) =>
           go(expr, bound, free)
         case SETryCatch(body, handler) =>
@@ -1303,8 +1274,6 @@ private[lf] final class Compiler(
         case _: SELet1General => goLets(maxS)(expr)
         case _: SELet1Builtin => goLets(maxS)(expr)
         case _: SELet1BuiltinArithmetic => goLets(maxS)(expr)
-        case SECatchSubmitMustFail(body) =>
-          go(body)
         case SELocation(_, body) =>
           go(body)
         case SELabelClosure(_, expr) =>

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/Pretty.scala

@@ -566,9 +566,6 @@ private[lf] object Pretty {
             text("-> ")
           prettySExpr(index + n)(body).tightBracketBy(prefix, char(')'))
 
-        case SECatchSubmitMustFail(body) =>
-          text("catch-submit-must-fail") + char('(') + prettySExpr(index)(body) + text(")")
-
         case SELocation(loc @ _, body) =>
           prettySExpr(index)(body)
 

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/SBuiltin.scala

@@ -1023,7 +1023,7 @@ private[lf] object SBuiltin {
               coid,
               templateId,
               onLedger.committers,
-              cbMissing = _ => machine.tryHandleSubmitMustFail(),
+              cbMissing = _ => false,
               cbPresent = { case V.ContractInst(actualTmplId, V.VersionedValue(_, arg), _) =>
                 // Note that we cannot throw in this continuation -- instead
                 // set the control appropriately which will crash the machine
@@ -1162,8 +1162,7 @@ private[lf] object SBuiltin {
 
   private[this] object KeyOperation {
     final class Fetch(override val templateId: TypeConName) extends KeyOperation {
-      override def handleInputKeyNotFound(machine: Machine): Boolean =
-        machine.tryHandleSubmitMustFail()
+      override def handleInputKeyNotFound(machine: Machine): Boolean = false
       override def handleInputKeyFound(machine: Machine, cid: V.ContractId): Unit =
         machine.ctrl = importCid(cid)
       override def handleInactiveKey(machine: Machine, gkey: GlobalKey): Unit =
@@ -1269,8 +1268,7 @@ private[lf] object SBuiltin {
                           keys = onLedger.ptx.keys.updated(gkey, KeyInactive)
                         )
                         operation.handleInputKeyNotFound(machine)
-                      case SKeyLookupResult.NotVisible =>
-                        machine.tryHandleSubmitMustFail()
+                      case SKeyLookupResult.NotVisible => false
                     }
                   },
                 )
@@ -1303,92 +1301,22 @@ private[lf] object SBuiltin {
     }
   }
 
-  /** $beginCommit :: Party -> Token -> () */
-  final case class SBSBeginCommit(optLocation: Option[Location]) extends OnLedgerBuiltin(2) {
-    override protected def execute(
-        args: util.ArrayList[SValue],
-        machine: Machine,
-        onLedger: OnLedger,
-    ): Unit = {
-      checkToken(args, 1)
-      onLedger.cachedContracts = Map.empty
-      onLedger.globalDiscriminators = Set.empty
-      onLedger.committers = extractParties(args.get(0))
-      onLedger.commitLocation = optLocation
-      machine.returnValue = SV.Unit
-    }
-  }
-
-  /** $endCommit[mustFail?] :: result -> Token -> () */
-  final case class SBSEndCommit(mustFail: Boolean) extends OnLedgerBuiltin(2) {
-    override protected def execute(
-        args: util.ArrayList[SValue],
-        machine: Machine,
-        onLedger: OnLedger,
-    ): Unit = {
-      checkToken(args, 1)
-      if (mustFail) executeMustFail(args, machine, onLedger)
-      else executeCommit(args, machine, onLedger)
-    }
-
-    private[this] def executeMustFail(
-        args: util.ArrayList[SValue],
-        machine: Machine,
-        onLedger: OnLedger,
-    ): Unit = {
-      // A mustFail commit evaluated the update with
-      // a catch. The second argument is a boolean
-      // that marks whether an exception was thrown
-      // or not.
-      val committerOld = onLedger.committers
-      val ptxOld = onLedger.ptx
-      val commitLocationOld = onLedger.commitLocation
-
-      if (getSBool(args, 0)) {
-        // update expression threw an exception. we're
-        // now done.
-        machine.clearCommit
-        machine.returnValue = SV.Unit
-        throw SpeedyHungry(SResultScenarioInsertMustFail(committerOld, commitLocationOld))
-      } else {
-        ptxOld.finish match {
-          case PartialTransaction.CompleteTransaction(tx) =>
-            // Transaction finished successfully. It might still
-            // fail when committed, so tell the scenario runner to
-            // do that.
-            machine.returnValue = SV.Unit
-            throw SpeedyHungry(
-              SResultScenarioMustFail(tx, committerOld, _ => machine.clearCommit)
-            )
-          case PartialTransaction.IncompleteTransaction(_) =>
+  final case class SBSSubmit(optLocation: Option[Location], mustFail: Boolean) extends SBuiltin(3) {
+    override private[speedy] def execute(args: util.ArrayList[SValue], machine: Machine): Unit = {
+      checkToken(args, 2)
+      throw SpeedyHungry(
+        SResultScenarioSubmit(
+          committers = extractParties(args.get(0)),
+          commands = args.get(1),
+          location = optLocation,
+          mustFail = mustFail,
+          callback = newValue => {
             machine.clearCommit
-            machine.returnValue = SV.Unit
-        }
-      }
+            machine.returnValue = newValue
+          },
+        )
+      )
     }
-
-    private[this] def executeCommit(
-        args: util.ArrayList[SValue],
-        machine: Machine,
-        onLedger: OnLedger,
-    ): Unit =
-      onLedger.ptx.finish match {
-        case PartialTransaction.CompleteTransaction(tx) =>
-          throw SpeedyHungry(
-            SResultScenarioCommit(
-              value = args.get(0),
-              tx = tx,
-              committers = onLedger.committers,
-              callback = newValue => {
-                machine.clearCommit
-                machine.returnValue = newValue
-              },
-            )
-          )
-        case PartialTransaction.IncompleteTransaction(ptx) =>
-          checkAborted(ptx)
-          crash("PartialTransaction.finish failed, but transaction was not aborted")
-      }
   }
 
   /** $pure :: a -> Token -> a */

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/SExpr.scala

@@ -358,19 +358,6 @@ object SExpr {
     }
   }
 
-  /** catch-submit-must-fail. This is used internally solely for the purpose of implementing
-    * mustFailAt. If the evaluation of 'body' causes an exception of type 'DamlException'
-    * (see SError), then 'True' is returned. If the evaluation is successful, then 'False'
-    * is returned.  This is on purpose very limited, with no mechanism to inspect the
-    * exception, nor a way to access the value returned from 'body'.
-    */
-  final case class SECatchSubmitMustFail(body: SExpr) extends SExpr {
-    def execute(machine: Machine): Unit = {
-      machine.pushKont(KCatchSubmitMustFail(machine))
-      machine.ctrl = body
-    }
-  }
-
   /** This is used only during profiling. When a package is compiled with
     * profiling enabled, the right hand sides of top-level and let bindings,
     * lambdas and some builtins are wrapped into [[SELabelClosure]]. During

daml-lf/interpreter/src/main/scala/com/digitalasset/daml/lf/speedy/SResult.scala

@@ -7,7 +7,7 @@ import com.daml.lf.CompiledPackages
 import com.daml.lf.value.Value.{ContractId, ContractInst}
 import com.daml.lf.data.Ref._
 import com.daml.lf.data.Time
-import com.daml.lf.transaction.{GlobalKeyWithMaintainers, SubmittedTransaction}
+import com.daml.lf.transaction.GlobalKeyWithMaintainers
 import com.daml.lf.speedy.SError._
 import com.daml.lf.value.Value
 
@@ -35,6 +35,7 @@ object SResult {
       committers: Set[Party],
       // Callback to signal that the contract was not present
       // or visible. Returns true if this was recoverable.
+      // TODO (MK) Drop now that tryHandleSubmitMustFail is dead.
       cbMissing: Unit => Boolean,
       cbPresent: ContractInst[Value.VersionedValue[ContractId]] => Unit,
   ) extends SResult
@@ -48,32 +49,14 @@ object SResult {
       callback: CompiledPackages => Unit,
   ) extends SResult
 
-  /** Commit the partial transaction to the (scenario) ledger.
-    * Machine expects the value back with the contract ids rewritten
-    * to be absolute.
-    */
-  final case class SResultScenarioCommit(
-      value: SValue,
-      tx: SubmittedTransaction,
+  final case class SResultScenarioSubmit(
       committers: Set[Party],
+      commands: SValue,
+      location: Option[Location],
+      mustFail: Boolean,
       callback: SValue => Unit,
   ) extends SResult
 
-  final case class SResultScenarioInsertMustFail(
-      committers: Set[Party],
-      optLocation: Option[Location],
-  ) extends SResult
-
-  /** A "must fail" update resulted in a partial transaction, try and
-    * commit this transaction with the expectation that it fails.
-    * The callback signals success and clears the partial transaction.
-    */
-  final case class SResultScenarioMustFail(
-      ptx: SubmittedTransaction,
-      committers: Set[Party],
-      callback: Unit => Unit,
-  ) extends SResult
-
   /** Pass the ledger time and return back the new ledger time. */
   final case class SResultScenarioPassTime(
       relTime: Long,