feat!: disable PXE concurrency (#12637)

Debugging #12391 led
me to discover that we cannot have concurrent simulations due to
contracts now being allowed to read and write to PXE's stores at
arbitrary moments. E.g.
#12391 was failing
CI due to multiple concurrent simulations deleting the same pending
partial note from a capsule array.

This PR disables that behavior by putting the problematic tasks in a
serial queue. Multiple tests still call PXE expecting concurrency
(typically via usage of `await Promise.all`), but I thought it made more
sense to disable the behavior this way and issue a warning (to unblock
#12391) and then
worry about removing attempts to achieve concurrent behavior.

I considered putting _all_ PXE functions in the serial queue, but
refrained from doing so to avoid introducing a larger than strictly
needed change. We may want to do this automatically via e.g.
monkey-patching to avoid accidentally forgetting a case.

Author: nventuro

docs/docs/migration_notes.md

@@ -8,6 +8,10 @@ Aztec is in full-speed development. Literally every version breaks compatibility
 
 ## TBD
 
+### [PXE] Concurrent contract function simulation disabled
+
+PXE is no longer be able to execute contract functions concurrently (e.g. by collecting calls to `simulateTx` and then using `await Promise.all`). They will instead be put in a job queue and executed sequentially in order of arrival.
+
 ### [aztec.js] Changes to `BatchCall` and `BaseContractInteraction`
 
 The constructor arguments of `BatchCall` have been updated to improve usability. Previously, it accepted an array of `FunctionCall`, requiring users to manually set additional data such as `authwit` and `capsules`. Now, `BatchCall` takes an array of `BaseContractInteraction`, which encapsulates all necessary information.

yarn-project/pxe/src/pxe_service/pxe_service.ts

@@ -1,6 +1,7 @@
 import { L1_TO_L2_MSG_TREE_HEIGHT } from '@aztec/constants';
 import { Fr, type Point } from '@aztec/foundation/fields';
 import { type Logger, createLogger } from '@aztec/foundation/log';
+import { SerialQueue } from '@aztec/foundation/queue';
 import { Timer } from '@aztec/foundation/timer';
 import type { SiblingPath } from '@aztec/foundation/trees';
 import { KeyStore } from '@aztec/key-store';
@@ -103,6 +104,7 @@ export class PXEService implements PXE {
     private proofCreator: PrivateKernelProver,
     private protocolContractsProvider: ProtocolContractsProvider,
     private log: Logger,
+    private jobQueue: SerialQueue,
   ) {}
 
   /**
@@ -160,6 +162,8 @@ export class PXEService implements PXE {
       log,
     );
     const simulator = new AcirSimulator(pxeOracleInterface, simulationProvider);
+    const jobQueue = new SerialQueue();
+
     const pxeService = new PXEService(
       node,
       synchronizer,
@@ -177,13 +181,34 @@ export class PXEService implements PXE {
       proofCreator,
       protocolContractsProvider,
       log,
+      jobQueue,
     );
+
+    pxeService.jobQueue.start();
+
     await pxeService.#registerProtocolContracts();
     const info = await pxeService.getNodeInfo();
     log.info(`Started PXE connected to chain ${info.l1ChainId} version ${info.protocolVersion}`);
     return pxeService;
   }
 
+  /**
+   * Enqueues a job for execution once no other jobs are running. Returns a promise that will resolve once the job is
+   * complete.
+   *
+   * Useful for tasks that cannot run concurrently, such as contract function simulation.
+   */
+  #putInJobQueue<T>(fn: () => Promise<T>): Promise<T> {
+    // TODO(#12636): relax the conditions under which we forbid concurrency.
+    if (this.jobQueue.length() != 0) {
+      this.log.warn(
+        `PXE is already processing ${this.jobQueue.length()} jobs, concurrent execution is not supported. Will run once those are complete.`,
+      );
+    }
+
+    return this.jobQueue.put(fn);
+  }
+
   isL1ToL2MessageSynced(l1ToL2Message: Fr): Promise<boolean> {
     return this.node.isL1ToL2MessageSynced(l1ToL2Message);
   }
@@ -364,35 +389,39 @@ export class PXEService implements PXE {
     );
   }
 
-  public async updateContract(contractAddress: AztecAddress, artifact: ContractArtifact): Promise<void> {
-    const currentInstance = await this.contractDataProvider.getContractInstance(contractAddress);
-    const contractClass = await getContractClassFromArtifact(artifact);
-    await this.synchronizer.sync();
+  public updateContract(contractAddress: AztecAddress, artifact: ContractArtifact): Promise<void> {
+    // We disable concurrently updating contracts to avoid concurrently syncing with the node, or changing a contract's
+    // class while we're simulating it.
+    return this.#putInJobQueue(async () => {
+      const currentInstance = await this.contractDataProvider.getContractInstance(contractAddress);
+      const contractClass = await getContractClassFromArtifact(artifact);
+      await this.synchronizer.sync();
 
-    const header = await this.syncDataProvider.getBlockHeader();
+      const header = await this.syncDataProvider.getBlockHeader();
 
-    const currentClassId = await readCurrentClassId(
-      contractAddress,
-      currentInstance,
-      this.node,
-      header.globalVariables.blockNumber.toNumber(),
-    );
-    if (!contractClass.id.equals(currentClassId)) {
-      throw new Error('Could not update contract to a class different from the current one.');
-    }
+      const currentClassId = await readCurrentClassId(
+        contractAddress,
+        currentInstance,
+        this.node,
+        header.globalVariables.blockNumber.toNumber(),
+      );
+      if (!contractClass.id.equals(currentClassId)) {
+        throw new Error('Could not update contract to a class different from the current one.');
+      }
 
-    await this.contractDataProvider.addContractArtifact(contractClass.id, artifact);
+      await this.contractDataProvider.addContractArtifact(contractClass.id, artifact);
 
-    const publicFunctionSignatures = artifact.functions
-      .filter(fn => fn.functionType === FunctionType.PUBLIC)
-      .map(fn => decodeFunctionSignature(fn.name, fn.parameters));
-    await this.node.registerContractFunctionSignatures(contractAddress, publicFunctionSignatures);
+      const publicFunctionSignatures = artifact.functions
+        .filter(fn => fn.functionType === FunctionType.PUBLIC)
+        .map(fn => decodeFunctionSignature(fn.name, fn.parameters));
+      await this.node.registerContractFunctionSignatures(contractAddress, publicFunctionSignatures);
 
-    // TODO(#10007): Node should get public contract class from the registration event, not from PXE registration
-    await this.node.addContractClass({ ...contractClass, privateFunctions: [], unconstrainedFunctions: [] });
-    currentInstance.currentContractClassId = contractClass.id;
-    await this.contractDataProvider.addContractInstance(currentInstance);
-    this.log.info(`Updated contract ${artifact.name} at ${contractAddress.toString()} to class ${contractClass.id}`);
+      // TODO(#10007): Node should get public contract class from the registration event, not from PXE registration
+      await this.node.addContractClass({ ...contractClass, privateFunctions: [], unconstrainedFunctions: [] });
+      currentInstance.currentContractClassId = contractClass.id;
+      await this.contractDataProvider.addContractInstance(currentInstance);
+      this.log.info(`Updated contract ${artifact.name} at ${contractAddress.toString()} to class ${contractClass.id}`);
+    });
   }
 
   public getContracts(): Promise<AztecAddress[]> {
@@ -456,24 +485,33 @@ export class PXEService implements PXE {
     return await this.node.getCurrentBaseFees();
   }
 
-  public async proveTx(
+  public proveTx(
     txRequest: TxExecutionRequest,
     privateExecutionResult: PrivateExecutionResult,
   ): Promise<TxProvingResult> {
-    try {
-      const { publicInputs, clientIvcProof } = await this.#prove(txRequest, this.proofCreator, privateExecutionResult, {
-        simulate: false,
-        skipFeeEnforcement: false,
-        profile: false,
-      });
-      return new TxProvingResult(privateExecutionResult, publicInputs, clientIvcProof!);
-    } catch (err: any) {
-      throw this.contextualizeError(err, inspect(txRequest), inspect(privateExecutionResult));
-    }
+    // We disable proving concurrently mostly out of caution, since it accesses some of our stores. Proving is so
+    // computationally demanding that it'd be rare for someone to try to do it concurrently regardless.
+    return this.#putInJobQueue(async () => {
+      try {
+        const { publicInputs, clientIvcProof } = await this.#prove(
+          txRequest,
+          this.proofCreator,
+          privateExecutionResult,
+          {
+            simulate: false,
+            skipFeeEnforcement: false,
+            profile: false,
+          },
+        );
+        return new TxProvingResult(privateExecutionResult, publicInputs, clientIvcProof!);
+      } catch (err: any) {
+        throw this.contextualizeError(err, inspect(txRequest), inspect(privateExecutionResult));
+      }
+    });
   }
 
   // TODO(#7456) Prevent msgSender being defined here for the first call
-  public async simulateTx(
+  public simulateTx(
     txRequest: TxExecutionRequest,
     simulatePublic: boolean,
     msgSender: AztecAddress | undefined = undefined,
@@ -482,74 +520,84 @@ export class PXEService implements PXE {
     profile: boolean = false,
     scopes?: AztecAddress[],
   ): Promise<TxSimulationResult> {
-    try {
-      const txInfo = {
-        origin: txRequest.origin,
-        functionSelector: txRequest.functionSelector,
-        simulatePublic,
-        msgSender,
-        chainId: txRequest.txContext.chainId,
-        version: txRequest.txContext.version,
-        authWitnesses: txRequest.authWitnesses.map(w => w.requestHash),
-      };
-      this.log.info(
-        `Simulating transaction execution request to ${txRequest.functionSelector} at ${txRequest.origin}`,
-        txInfo,
-      );
-      const timer = new Timer();
-      await this.synchronizer.sync();
-      const privateExecutionResult = await this.#executePrivate(txRequest, msgSender, scopes);
-
-      const { publicInputs, profileResult } = await this.#prove(txRequest, this.proofCreator, privateExecutionResult, {
-        simulate: !profile,
-        skipFeeEnforcement,
-        profile,
-      });
-
-      const privateSimulationResult = new PrivateSimulationResult(privateExecutionResult, publicInputs);
-      const simulatedTx = privateSimulationResult.toSimulatedTx();
-      let publicOutput: PublicSimulationOutput | undefined;
-      if (simulatePublic && publicInputs.forPublic) {
-        publicOutput = await this.#simulatePublicCalls(simulatedTx, skipFeeEnforcement);
-      }
+    // We disable concurrent simulations since those might execute oracles which read and write to the PXE stores (e.g.
+    // to the capsules), and we need to prevent concurrent runs from interfering with one another (e.g. attempting to
+    // delete the same read value, or reading values that another simulation is currently modifying).
+    return this.#putInJobQueue(async () => {
+      try {
+        const txInfo = {
+          origin: txRequest.origin,
+          functionSelector: txRequest.functionSelector,
+          simulatePublic,
+          msgSender,
+          chainId: txRequest.txContext.chainId,
+          version: txRequest.txContext.version,
+          authWitnesses: txRequest.authWitnesses.map(w => w.requestHash),
+        };
+        this.log.info(
+          `Simulating transaction execution request to ${txRequest.functionSelector} at ${txRequest.origin}`,
+          txInfo,
+        );
+        const timer = new Timer();
+        await this.synchronizer.sync();
+        const privateExecutionResult = await this.#executePrivate(txRequest, msgSender, scopes);
+
+        const { publicInputs, profileResult } = await this.#prove(
+          txRequest,
+          this.proofCreator,
+          privateExecutionResult,
+          {
+            simulate: !profile,
+            skipFeeEnforcement,
+            profile,
+          },
+        );
 
-      if (!skipTxValidation) {
-        const validationResult = await this.node.isValidTx(simulatedTx, { isSimulation: true, skipFeeEnforcement });
-        if (validationResult.result === 'invalid') {
-          throw new Error('The simulated transaction is unable to be added to state and is invalid.');
+        const privateSimulationResult = new PrivateSimulationResult(privateExecutionResult, publicInputs);
+        const simulatedTx = privateSimulationResult.toSimulatedTx();
+        let publicOutput: PublicSimulationOutput | undefined;
+        if (simulatePublic && publicInputs.forPublic) {
+          publicOutput = await this.#simulatePublicCalls(simulatedTx, skipFeeEnforcement);
         }
-      }
 
-      const txHash = await simulatedTx.getTxHash();
-      this.log.info(`Simulation completed for ${txHash.toString()} in ${timer.ms()}ms`, {
-        txHash,
-        ...txInfo,
-        ...(profileResult ? { gateCounts: profileResult.gateCounts } : {}),
-        ...(publicOutput
-          ? {
-              gasUsed: publicOutput.gasUsed,
-              revertCode: publicOutput.txEffect.revertCode.getCode(),
-              revertReason: publicOutput.revertReason,
-            }
-          : {}),
-      });
+        if (!skipTxValidation) {
+          const validationResult = await this.node.isValidTx(simulatedTx, { isSimulation: true, skipFeeEnforcement });
+          if (validationResult.result === 'invalid') {
+            throw new Error('The simulated transaction is unable to be added to state and is invalid.');
+          }
+        }
 
-      return TxSimulationResult.fromPrivateSimulationResultAndPublicOutput(
-        privateSimulationResult,
-        publicOutput,
-        profileResult,
-      );
-    } catch (err: any) {
-      throw this.contextualizeError(
-        err,
-        inspect(txRequest),
-        `simulatePublic=${simulatePublic}`,
-        `msgSender=${msgSender?.toString() ?? 'undefined'}`,
-        `skipTxValidation=${skipTxValidation}`,
-        `profile=${profile}`,
-        `scopes=${scopes?.map(s => s.toString()).join(', ') ?? 'undefined'}`,
-      );
-    }
+        const txHash = await simulatedTx.getTxHash();
+        this.log.info(`Simulation completed for ${txHash.toString()} in ${timer.ms()}ms`, {
+          txHash,
+          ...txInfo,
+          ...(profileResult ? { gateCounts: profileResult.gateCounts } : {}),
+          ...(publicOutput
+            ? {
+                gasUsed: publicOutput.gasUsed,
+                revertCode: publicOutput.txEffect.revertCode.getCode(),
+                revertReason: publicOutput.revertReason,
+              }
+            : {}),
+        });
+
+        return TxSimulationResult.fromPrivateSimulationResultAndPublicOutput(
+          privateSimulationResult,
+          publicOutput,
+          profileResult,
+        );
+      } catch (err: any) {
+        throw this.contextualizeError(
+          err,
+          inspect(txRequest),
+          `simulatePublic=${simulatePublic}`,
+          `msgSender=${msgSender?.toString() ?? 'undefined'}`,
+          `skipTxValidation=${skipTxValidation}`,
+          `profile=${profile}`,
+          `scopes=${scopes?.map(s => s.toString()).join(', ') ?? 'undefined'}`,
+        );
+      }
+    });
   }
 
   public async sendTx(tx: Tx): Promise<TxHash> {
@@ -565,29 +613,34 @@ export class PXEService implements PXE {
     return txHash;
   }
 
-  public async simulateUnconstrained(
+  public simulateUnconstrained(
     functionName: string,
     args: any[],
     to: AztecAddress,
     _from?: AztecAddress,
     scopes?: AztecAddress[],
   ): Promise<AbiDecoded> {
-    try {
-      await this.synchronizer.sync();
-      // TODO - Should check if `from` has the permission to call the view function.
-      const functionCall = await this.#getFunctionCall(functionName, args, to);
-      const executionResult = await this.#simulateUnconstrained(functionCall, scopes);
-
-      // TODO - Return typed result based on the function artifact.
-      return executionResult;
-    } catch (err: any) {
-      const stringifiedArgs = args.map(arg => arg.toString()).join(', ');
-      throw this.contextualizeError(
-        err,
-        `simulateUnconstrained ${to}:${functionName}(${stringifiedArgs})`,
-        `scopes=${scopes?.map(s => s.toString()).join(', ') ?? 'undefined'}`,
-      );
-    }
+    // We disable concurrent simulations since those might execute oracles which read and write to the PXE stores (e.g.
+    // to the capsules), and we need to prevent concurrent runs from interfering with one another (e.g. attempting to
+    // delete the same read value, or reading values that another simulation is currently modifying).
+    return this.#putInJobQueue(async () => {
+      try {
+        await this.synchronizer.sync();
+        // TODO - Should check if `from` has the permission to call the view function.
+        const functionCall = await this.#getFunctionCall(functionName, args, to);
+        const executionResult = await this.#simulateUnconstrained(functionCall, scopes);
+
+        // TODO - Return typed result based on the function artifact.
+        return executionResult;
+      } catch (err: any) {
+        const stringifiedArgs = args.map(arg => arg.toString()).join(', ');
+        throw this.contextualizeError(
+          err,
+          `simulateUnconstrained ${to}:${functionName}(${stringifiedArgs})`,
+          `scopes=${scopes?.map(s => s.toString()).join(', ') ?? 'undefined'}`,
+        );
+      }
+    });
   }
 
   public getTxReceipt(txHash: TxHash): Promise<TxReceipt> {