refactor: parallelize per-scope contract syncs (backport #22525) (#22561)

## Summary
Backport of #22525
to v4-next.

Cherry-pick applied cleanly with no conflicts. No additional changes
needed.

## Original PR
Parallelizes per-scope contract syncs in the PXE ContractSyncService,
capping concurrency at 5 via a Semaphore. This addresses a performance
regression from moving to sequential per-scope-and-contract syncing.
Also adds a concurrent WASM execution test.

ClaudeBox log: https://claudebox.work/s/ea5bb0472c2216a5?run=1

Author: mverzilli

yarn-project/pxe/src/contract_function_simulator/oracle/utility_execution.test.ts

@@ -227,6 +227,96 @@ describe('Utility Execution test suite', () => {
     expect(offchainEffects).toEqual([]);
   }, 30_000);
 
+  it('WASM simulator supports N-way concurrent utility execution', async () => {
+    // Fires N concurrent runUtility calls through the shared WASMSimulator instance. Each call runs its own ACIR
+    // execution with its own oracle, but they all funnel through the same `executeUserCircuit` entry point and
+    // therefore the same Wasm module. If the underlying noir-acvm_js binding cannot handle overlapping executions,
+    // concurrent calls can corrupt each other's witness maps: we'd expect crashes, rejections, or wrong sums.
+    //
+    // This guards the parallelization in ContractSyncService (MAX_CONCURRENT_SCOPE_SYNCS = 5), which is the first
+    // code path to run multiple utility ACIR calls simultaneously. N is set to 10 so we test twice the production
+    // cap to catch issues that would only surface at higher concurrency.
+    const N = 10;
+    const artifact = {
+      ...StatefulTestContractArtifact.functions.find(f => f.name === 'summed_values')!,
+      contractName: StatefulTestContractArtifact.name,
+    };
+
+    const notes: Note[] = [...Array(5).fill(buildNote(1n)), ...Array(2).fill(buildNote(2n))];
+    const expectedSum = new Fr(9);
+
+    const instanceFields = {
+      version: 1 as const,
+      salt: Fr.random(),
+      deployer: await AztecAddress.random(),
+      currentContractClassId: new Fr(42),
+      originalContractClassId: new Fr(42),
+      initializationHash: Fr.random(),
+      publicKeys: await PublicKeys.random(),
+    };
+    const contractAddress = await computeContractAddressFromInstance(instanceFields);
+
+    aztecNode.getPublicStorageAt.mockResolvedValue(Fr.ZERO);
+    aztecNode.findLeavesIndexes.mockResolvedValue([
+      { data: 1n, l2BlockNumber: BlockNumber(1), l2BlockHash: BlockHash.random() },
+    ]);
+    contractStore.getFunctionArtifact.mockResolvedValue(artifact);
+    contractStore.getContractInstance.mockResolvedValue({
+      ...instanceFields,
+      address: contractAddress,
+    } as ContractInstanceWithAddress);
+    contractStore.getFunctionArtifactWithDebugMetadata.mockImplementation(async (address, selector) => {
+      const artifact = await contractStore.getFunctionArtifact(address, selector);
+      if (!artifact) {
+        throw new Error(`Function not found: ${selector.toString()} in contract ${address}`);
+      }
+      return { ...artifact, debug: undefined };
+    });
+    noteStore.getNotes.mockResolvedValue(
+      notes.map(
+        note =>
+          new NoteDao(
+            note,
+            contractAddress,
+            owner,
+            Fr.random(),
+            Fr.random(),
+            Fr.random(),
+            Fr.random(),
+            Fr.random(),
+            TxHash.random(),
+            BlockNumber(42),
+            BlockHash.random().toString(),
+            0,
+            0,
+          ),
+      ),
+    );
+    capsuleStore.getCapsule.mockImplementation((_, __) => Promise.resolve(null));
+
+    const execRequest = FunctionCall.from({
+      name: artifact.name,
+      to: contractAddress,
+      selector: FunctionSelector.empty(),
+      type: FunctionType.UTILITY,
+      hideMsgSender: false,
+      isStatic: false,
+      args: encodeArguments(artifact, [owner]),
+      returnTypes: artifact.returnTypes,
+    });
+
+    const results = await Promise.all(
+      Array.from({ length: N }, (_, i) =>
+        acirSimulator.runUtility(execRequest, [], anchorBlockHeader, [], `reentrance-job-${i}`),
+      ),
+    );
+
+    expect(results).toHaveLength(N);
+    for (const { result } of results) {
+      expect(result).toEqual([expectedSum]);
+    }
+  }, 60_000);
+
   describe('UtilityExecutionOracle', () => {
     let contractAddress: AztecAddress;
     let utilityExecutionOracle: UtilityExecutionOracle;

yarn-project/pxe/src/contract_sync/contract_sync_service.ts

@@ -1,4 +1,5 @@
 import type { Logger } from '@aztec/foundation/log';
+import { Semaphore } from '@aztec/foundation/queue';
 import type { FunctionCall, FunctionSelector } from '@aztec/stdlib/abi';
 import type { AztecAddress } from '@aztec/stdlib/aztec-address';
 import type { AztecNode } from '@aztec/stdlib/interfaces/client';
@@ -9,6 +10,9 @@ import type { ContractStore } from '../storage/contract_store/contract_store.js'
 import type { NoteStore } from '../storage/note_store/note_store.js';
 import { syncState, verifyCurrentClassId } from './helpers.js';
 
+/** Maximum number of scope syncs running concurrently across the PXE. */
+const MAX_CONCURRENT_SCOPE_SYNCS = 5;
+
 /**
  * Service for syncing the private state of contracts and verifying that the PXE holds the current class artifact.
  * It uses a cache to avoid redundant sync operations - the cache is wiped when the anchor block changes.
@@ -26,6 +30,11 @@ export class ContractSyncService implements StagedStore {
   // Per-job excluded contract addresses - these contracts should not be synced.
   private excludedFromSync: Map<string, Set<string>> = new Map();
 
+  // Bounds the number of scope syncs running concurrently. Scopes beyond this limit queue here. Sized to trade off
+  // parallelism on non-ACIR work (node RPC, note store reads) against memory pressure from concurrent circuit
+  // execution.
+  #syncSlot = new Semaphore(MAX_CONCURRENT_SCOPE_SYNCS);
+
   constructor(
     private aztecNode: AztecNode,
     private contractStore: ContractStore,
@@ -59,15 +68,22 @@ export class ContractSyncService implements StagedStore {
       return;
     }
 
-    this.#startSyncIfNeeded(contractAddress, scopes, scopesToSync =>
-      this.#syncContract(
-        contractAddress,
-        functionToInvokeAfterSync,
-        utilityExecutor,
-        anchorBlockHeader,
-        jobId,
-        scopesToSync,
-      ),
+    this.#startSyncIfNeeded(
+      contractAddress,
+      scopes,
+      () => verifyCurrentClassId(contractAddress, this.aztecNode, this.contractStore, anchorBlockHeader),
+      scope =>
+        syncState(
+          contractAddress,
+          this.contractStore,
+          functionToInvokeAfterSync,
+          utilityExecutor,
+          this.noteStore,
+          this.aztecNode,
+          anchorBlockHeader,
+          jobId,
+          scope,
+        ),
     );
 
     await this.#awaitSync(contractAddress, scopes);
@@ -81,39 +97,6 @@ export class ContractSyncService implements StagedStore {
     scopes.forEach(scope => this.syncedContracts.delete(toKey(contractAddress, scope)));
   }
 
-  async #syncContract(
-    contractAddress: AztecAddress,
-    functionToInvokeAfterSync: FunctionSelector | null,
-    utilityExecutor: (call: FunctionCall, scopes: AztecAddress[]) => Promise<any>,
-    anchorBlockHeader: BlockHeader,
-    jobId: string,
-    scopes: AztecAddress[],
-  ): Promise<void> {
-    this.log.debug(`Syncing contract ${contractAddress}`);
-
-    await Promise.all([
-      // Call sync_state sequentially for each scope address — each invocation synchronizes one account's private
-      // state using scoped capsule arrays.
-      (async () => {
-        for (const scope of scopes) {
-          await syncState(
-            contractAddress,
-            this.contractStore,
-            functionToInvokeAfterSync,
-            utilityExecutor,
-            this.noteStore,
-            this.aztecNode,
-            anchorBlockHeader,
-            jobId,
-            scope,
-          );
-        }
-      })(),
-      verifyCurrentClassId(contractAddress, this.aztecNode, this.contractStore, anchorBlockHeader),
-    ]);
-    this.log.debug(`Contract ${contractAddress} synced`);
-  }
-
   /** Clears sync cache. Called by BlockSynchronizer when anchor block changes. */
   wipe(): void {
     this.log.debug(`Wiping contract sync cache (${this.syncedContracts.size} entries)`);
@@ -138,22 +121,45 @@ export class ContractSyncService implements StagedStore {
     return !!this.excludedFromSync.get(jobId)?.has(contractAddress.toString());
   }
 
-  /** If there are unsynced scopes, starts sync and stores the promise in cache with error cleanup. */
+  /**
+   * If there are unsynced scopes, starts one sync per scope (bounded by #syncSlot) and stores each promise in the
+   * cache with per-scope error cleanup. The verifyFn runs once for the whole fan-out and is awaited by every new
+   * scope's promise, matching the pre-parallelization invariant that a cache-miss batch re-verifies the class id.
+   */
   #startSyncIfNeeded(
     contractAddress: AztecAddress,
     scopes: AztecAddress[],
-    syncFn: (scopesToSync: AztecAddress[]) => Promise<void>,
+    verifyFn: () => Promise<void>,
+    syncScopeFn: (scope: AztecAddress) => Promise<void>,
   ): void {
     const scopesToSync = scopes.filter(scope => !this.syncedContracts.has(toKey(contractAddress, scope)));
-    const keys = scopesToSync.map(scope => toKey(contractAddress, scope));
-    if (keys.length === 0) {
+    if (scopesToSync.length === 0) {
       return;
     }
-    const promise = syncFn(scopesToSync).catch(err => {
-      keys.forEach(key => this.syncedContracts.delete(key));
-      throw err;
-    });
-    keys.forEach(key => this.syncedContracts.set(key, promise));
+
+    this.log.debug(`Syncing contract ${contractAddress} for ${scopesToSync.length} scope(s)`);
+    const verifyPromise = verifyFn();
+
+    for (const scope of scopesToSync) {
+      const key = toKey(contractAddress, scope);
+      const promise = Promise.all([verifyPromise, this.#runBounded(() => syncScopeFn(scope))])
+        .then(() => {})
+        .catch(err => {
+          this.syncedContracts.delete(key);
+          throw err;
+        });
+      this.syncedContracts.set(key, promise);
+    }
+  }
+
+  /** Runs fn while holding a slot in #syncSlot, bounding total concurrent scope syncs. */
+  async #runBounded<T>(fn: () => Promise<T>): Promise<T> {
+    await this.#syncSlot.acquire();
+    try {
+      return await fn();
+    } finally {
+      this.#syncSlot.release();
+    }
   }
 
   /** Collects all relevant scope promises (including in-flight ones from concurrent calls) and awaits them. */