Skip to content

feat!: disable PXE concurrency #12637

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nventuro merged 6 commits into from
Mar 11, 2025
Merged

feat!: disable PXE concurrency #12637

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
f754381
Add job queue
nventuro Mar 10, 2025
c9a46f5
feat!: disable PXE concurrency
nventuro Mar 11, 2025
4c2f7c1
Restore e2e test
nventuro Mar 11, 2025
1039a9d
Remove extra whitespace
nventuro Mar 11, 2025
4643439
restore test
nventuro Mar 11, 2025
4bedf9e
Merge branch 'master' into nv/pxe-no-concurrency
nventuro Mar 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/docs/migration_notes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
287 changes: 170 additions & 117 deletions yarn-project/pxe/src/pxe_service/pxe_service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -103,6 +104,7 @@ export class PXEService implements PXE {
private proofCreator: PrivateKernelProver,
private protocolContractsProvider: ProtocolContractsProvider,
private log: Logger,
private jobQueue: SerialQueue,
) {}

/**
Expand Down Expand Up @@ -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,
Expand All @@ -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);
}
Expand Down Expand Up @@ -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[]> {
Expand Down Expand Up @@ -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,
Expand All @@ -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> {
Expand All @@ -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> {
Expand Down