feat: integrate new proving broker

yarn-project/aztec-node/src/aztec-node/server.ts

@@ -164,6 +164,7 @@ export class AztecNodeService implements AztecNode {
     // now create the merkle trees and the world state synchronizer
     const worldStateSynchronizer = await createWorldStateSynchronizer(config, archiver, telemetry);
     const proofVerifier = config.realProofs ? await BBCircuitVerifier.new(config) : new TestCircuitVerifier();
+    log.info(`Aztec node accepting ${config.realProofs ? 'real' : 'test'} proofs`);
 
     // create the tx pool and the p2p client, which will need the l2 block source
     const p2pClient = await createP2PClient(config, archiver, proofVerifier, worldStateSynchronizer, telemetry);

yarn-project/aztec/src/cli/aztec_start_options.ts

@@ -1,16 +1,21 @@
 import { type ArchiverConfig, archiverConfigMappings } from '@aztec/archiver';
 import { sequencerClientConfigMappings } from '@aztec/aztec-node';
 import { botConfigMappings } from '@aztec/bot';
+import {
+  type ProverAgentConfig,
+  type ProverBrokerConfig,
+  proverAgentConfigMappings,
+  proverBrokerConfigMappings,
+} from '@aztec/circuit-types';
 import {
   type ConfigMapping,
   type EnvVar,
   booleanConfigHelper,
-  filterConfigMappings,
   isBooleanConfigValue,
+  omitConfigMappings,
 } from '@aztec/foundation/config';
 import { bootnodeConfigMappings, p2pConfigMappings } from '@aztec/p2p';
 import { proofVerifierConfigMappings } from '@aztec/proof-verifier';
-import { proverClientConfigMappings } from '@aztec/prover-client';
 import { proverNodeConfigMappings } from '@aztec/prover-node';
 import { allPxeConfigMappings } from '@aztec/pxe';
 import { telemetryClientConfigMappings } from '@aztec/telemetry-client/start';
@@ -239,15 +244,6 @@ export const aztecStartOptions: { [key: string]: AztecStartOption[] } = {
     },
     ...getOptions('sequencer', sequencerClientConfigMappings),
   ],
-  'PROVER AGENT': [
-    {
-      flag: '--prover',
-      description: 'Starts Aztec Prover Agent with options',
-      defaultValue: undefined,
-      envVar: undefined,
-    },
-    ...getOptions('prover', proverClientConfigMappings),
-  ],
   'PROVER NODE': [
     {
       flag: '--prover-node',
@@ -263,10 +259,36 @@ export const aztecStartOptions: { [key: string]: AztecStartOption[] } = {
     },
     ...getOptions(
       'proverNode',
+      omitConfigMappings(proverNodeConfigMappings, [
+        // filter out options passed separately
+        ...(Object.keys(archiverConfigMappings) as (keyof ArchiverConfig)[]),
+        ...(Object.keys(proverBrokerConfigMappings) as (keyof ProverBrokerConfig)[]),
+        ...(Object.keys(proverAgentConfigMappings) as (keyof ProverAgentConfig)[]),
+      ]),
+    ),
+  ],
+  'PROVER BROKER': [
+    {
+      flag: '--prover-broker',
+      description: 'Starts Aztec proving job broker',
+      defaultValue: undefined,
+      envVar: undefined,
+    },
+    ...getOptions(
+      'proverBroker',
       // filter out archiver options from prover node options as they're passed separately in --archiver
-      filterConfigMappings(proverNodeConfigMappings, Object.keys(archiverConfigMappings) as (keyof ArchiverConfig)[]),
+      proverBrokerConfigMappings,
     ),
   ],
+  'PROVER AGENT': [
+    {
+      flag: '--prover-agent',
+      description: 'Starts Aztec Prover Agent with options',
+      defaultValue: undefined,
+      envVar: undefined,
+    },
+    ...getOptions('proverAgent', proverAgentConfigMappings),
+  ],
   'P2P BOOTSTRAP': [
     {
       flag: '--p2p-bootstrap',

yarn-project/aztec/src/cli/cli.ts

@@ -96,9 +96,12 @@ export function injectAztecCommands(program: Command, userLog: LogFn, debugLogge
       } else if (options.p2pBootstrap) {
         const { startP2PBootstrap } = await import('./cmds/start_p2p_bootstrap.js');
         await startP2PBootstrap(options, userLog, debugLogger);
-      } else if (options.prover) {
-        const { startProverAgent } = await import('./cmds/start_prover_agent.js');
-        await startProverAgent(options, signalHandlers, services, userLog);
+      } else if (options.proverNodeAgent) {
+        const { startProverAgent: startProverNodeAgent } = await import('./cmds/start_prover_agent.js');
+        await startProverNodeAgent(options, signalHandlers, services, userLog);
+      } else if (options.proverNodeBroker) {
+        const { startProverBroker: startProverNodeBroker } = await import('./cmds/start_prover_broker.js');
+        await startProverNodeBroker(options, signalHandlers, services, userLog);
       } else if (options.txe) {
         const { startTXE } = await import('./cmds/start_txe.js');
         await startTXE(options, debugLogger);

yarn-project/aztec/src/cli/cmds/start_prover_agent.ts

@@ -1,51 +1,51 @@
-import { BBNativeRollupProver, TestCircuitProver } from '@aztec/bb-prover';
-import { ProverAgentApiSchema, type ServerCircuitProver } from '@aztec/circuit-types';
+import { type ProverAgentConfig, proverAgentConfigMappings } from '@aztec/circuit-types';
+import { times } from '@aztec/foundation/collection';
 import { type NamespacedApiHandlers } from '@aztec/foundation/json-rpc/server';
 import { type LogFn } from '@aztec/foundation/log';
-import { type ProverClientConfig, proverClientConfigMappings } from '@aztec/prover-client';
-import { ProverAgent, createProvingJobSourceClient } from '@aztec/prover-client/prover-agent';
-import {
-  type TelemetryClientConfig,
-  createAndStartTelemetryClient,
-  telemetryClientConfigMappings,
-} from '@aztec/telemetry-client/start';
+import { buildServerCircuitProver } from '@aztec/prover-client';
+import { InlineProofStore, ProvingAgent, createProvingJobConsumerClient } from '@aztec/prover-client/broker';
+import { getProverNodeAgentConfigFromEnv } from '@aztec/prover-node';
+import { createAndStartTelemetryClient, telemetryClientConfigMappings } from '@aztec/telemetry-client/start';
 
 import { extractRelevantOptions } from '../util.js';
 
 export async function startProverAgent(
   options: any,
   signalHandlers: (() => Promise<void>)[],
   services: NamespacedApiHandlers,
-  logger: LogFn,
+  userLog: LogFn,
 ) {
-  const proverConfig = extractRelevantOptions<ProverClientConfig>(options, proverClientConfigMappings, 'prover');
-  const proverJobSourceUrl = proverConfig.proverJobSourceUrl ?? proverConfig.nodeUrl;
-  if (!proverJobSourceUrl) {
-    throw new Error('Starting prover without PROVER_JOB_SOURCE_URL is not supported');
+  if (options.node || options.sequencer || options.pxe || options.p2pBootstrap || options.txe) {
+    userLog(`Starting a prover agent with --node, --sequencer, --pxe, --p2p-bootstrap, or --txe is not supported.`);
+    process.exit(1);
   }
 
-  logger(`Connecting to prover at ${proverJobSourceUrl}`);
-  const source = createProvingJobSourceClient(proverJobSourceUrl);
+  const config = {
+    ...getProverNodeAgentConfigFromEnv(), // get default config from env
+    ...extractRelevantOptions<ProverAgentConfig>(options, proverAgentConfigMappings, 'proverAgent'), // override with command line options
+  };
 
-  const telemetryConfig = extractRelevantOptions<TelemetryClientConfig>(options, telemetryClientConfigMappings, 'tel');
-  const telemetry = await createAndStartTelemetryClient(telemetryConfig);
+  if (config.realProofs && (!config.bbBinaryPath || config.acvmBinaryPath)) {
+    process.exit(1);
+  }
 
-  let circuitProver: ServerCircuitProver;
-  if (proverConfig.realProofs) {
-    if (!proverConfig.acvmBinaryPath || !proverConfig.bbBinaryPath) {
-      throw new Error('Cannot start prover without simulation or native prover options');
-    }
-    circuitProver = await BBNativeRollupProver.new(proverConfig, telemetry);
-  } else {
-    circuitProver = new TestCircuitProver(telemetry, undefined, proverConfig);
+  if (!config.proverBrokerUrl) {
+    process.exit(1);
   }
 
-  const { proverAgentConcurrency, proverAgentPollInterval } = proverConfig;
-  const agent = new ProverAgent(circuitProver, proverAgentConcurrency, proverAgentPollInterval);
-  agent.start(source);
+  const broker = createProvingJobConsumerClient(config.proverBrokerUrl);
+
+  const telemetry = await createAndStartTelemetryClient(
+    extractRelevantOptions(options, telemetryClientConfigMappings, 'tel'),
+  );
+  const prover = await buildServerCircuitProver(config, telemetry);
+  const proofStore = new InlineProofStore();
+  const agents = times(config.proverAgentCount, () => new ProvingAgent(broker, proofStore, prover));
 
-  logger(`Started prover agent with concurrency limit of ${proverAgentConcurrency}`);
+  await Promise.all(agents.map(agent => agent.start()));
 
-  services.prover = [agent, ProverAgentApiSchema];
-  signalHandlers.push(() => agent.stop());
+  signalHandlers.push(async () => {
+    await Promise.all(agents.map(agent => agent.stop()));
+    await telemetry.stop();
+  });
 }

yarn-project/aztec/src/cli/cmds/start_prover_broker.ts

@@ -0,0 +1,30 @@
+import { type ProverBrokerConfig, proverBrokerConfigMappings } from '@aztec/circuit-types';
+import { type NamespacedApiHandlers } from '@aztec/foundation/json-rpc/server';
+import { type LogFn } from '@aztec/foundation/log';
+import { ProvingJobBrokerSchema, createAndStartProvingBroker } from '@aztec/prover-client/broker';
+import { getProverNodeBrokerConfigFromEnv } from '@aztec/prover-node';
+
+import { extractRelevantOptions } from '../util.js';
+
+export async function startProverBroker(
+  options: any,
+  signalHandlers: (() => Promise<void>)[],
+  services: NamespacedApiHandlers,
+  userLog: LogFn,
+) {
+  if (options.node || options.sequencer || options.pxe || options.p2pBootstrap || options.txe) {
+    userLog(`Starting a prover broker with --node, --sequencer, --pxe, --p2p-bootstrap, or --txe is not supported.`);
+    process.exit(1);
+  }
+
+  const config: ProverBrokerConfig = {
+    ...getProverNodeBrokerConfigFromEnv(), // get default config from env
+    ...extractRelevantOptions<ProverBrokerConfig>(options, proverBrokerConfigMappings, 'proverBroker'), // override with command line options
+  };
+
+  const broker = await createAndStartProvingBroker(config);
+  services.proverNodeProducer = [broker, ProvingJobBrokerSchema];
+  signalHandlers.push(() => broker.stop());
+
+  await broker.start();
+}

yarn-project/aztec/src/cli/cmds/start_prover_node.ts

@@ -1,7 +1,8 @@
-import { ProverNodeApiSchema, ProvingJobSourceSchema, createAztecNodeClient } from '@aztec/circuit-types';
+import { ProverNodeApiSchema, createAztecNodeClient } from '@aztec/circuit-types';
 import { NULL_KEY } from '@aztec/ethereum';
 import { type NamespacedApiHandlers } from '@aztec/foundation/json-rpc/server';
 import { type LogFn } from '@aztec/foundation/log';
+import { ProvingJobConsumerSchema, createProvingJobBrokerClient } from '@aztec/prover-client/broker';
 import {
   type ProverNodeConfig,
   createProverNode,
@@ -37,10 +38,10 @@ export async function startProverNode(
 
   if (options.prover || options.proverAgentEnabled) {
     userLog(`Running prover node with local prover agent.`);
-    proverConfig.proverAgentEnabled = true;
+    proverConfig.proverAgentCount = 1;
   } else {
     userLog(`Running prover node without local prover agent. Connect one or more prover agents to this node.`);
-    proverConfig.proverAgentEnabled = false;
+    proverConfig.proverAgentCount = 0;
   }
 
   if (!proverConfig.publisherPrivateKey || proverConfig.publisherPrivateKey === NULL_KEY) {
@@ -67,12 +68,14 @@ export async function startProverNode(
   const telemetry = await createAndStartTelemetryClient(
     extractRelevantOptions(options, telemetryClientConfigMappings, 'tel'),
   );
-  const proverNode = await createProverNode(proverConfig, { telemetry });
+
+  const broker = proverConfig.proverBrokerUrl ? createProvingJobBrokerClient(proverConfig.proverBrokerUrl) : undefined;
+  const proverNode = await createProverNode(proverConfig, { telemetry, broker });
 
   services.proverNode = [proverNode, ProverNodeApiSchema];
 
-  if (!options.prover) {
-    services.provingJobSource = [proverNode.getProver().getProvingJobSource(), ProvingJobSourceSchema];
+  if (!proverConfig.proverBrokerUrl) {
+    services.provingJobSource = [proverNode.getProver().getProvingJobSource(), ProvingJobConsumerSchema];
   }
 
   signalHandlers.push(proverNode.stop.bind(proverNode));

yarn-project/bb-prover/src/config.ts

@@ -6,6 +6,8 @@ export interface BBConfig {
 }
 
 export interface ACVMConfig {
+  /** The path to the ACVM binary */
   acvmBinaryPath: string;
+  /** The working directory to use for simulation/proving */
   acvmWorkingDirectory: string;
 }

yarn-project/circuit-types/src/interfaces/index.ts

@@ -20,3 +20,4 @@ export * from './server_circuit_prover.js';
 export * from './service.js';
 export * from './sync-status.js';
 export * from './world_state.js';
+export * from './prover-broker.js';

yarn-project/circuit-types/src/interfaces/prover-agent.ts

@@ -1,7 +1,63 @@
+import { type ConfigMappingsType, booleanConfigHelper, numberConfigHelper } from '@aztec/foundation/config';
 import { type ApiSchemaFor } from '@aztec/foundation/schemas';
 
 import { z } from 'zod';
 
+import { ProvingRequestType } from './proving-job.js';
+
+export const ProverAgentConfig = z.object({
+  /** The number of prover agents to start */
+  proverAgentCount: z.number(),
+  /** The types of proofs the prover agent can generate */
+  proverAgentProofTypes: z.array(z.nativeEnum(ProvingRequestType)),
+  /** How often the prover agents poll for jobs */
+  proverAgentPollIntervalMs: z.number(),
+  /** The URL where this agent takes jobs from */
+  proverBrokerUrl: z.string().optional(),
+  /** Whether to construct real proofs */
+  realProofs: z.boolean(),
+  /** Artificial delay to introduce to all operations to the test prover. */
+  proverTestDelayMs: z.number(),
+});
+
+export type ProverAgentConfig = z.infer<typeof ProverAgentConfig>;
+
+export const proverAgentConfigMappings: ConfigMappingsType<ProverAgentConfig> = {
+  proverAgentCount: {
+    env: 'PROVER_AGENT_COUNT',
+    description: 'Whether this prover has a local prover agent',
+    ...numberConfigHelper(1),
+  },
+  proverAgentPollIntervalMs: {
+    env: 'PROVER_AGENT_POLL_INTERVAL_MS',
+    description: 'The interval agents poll for jobs at',
+    ...numberConfigHelper(100),
+  },
+  proverAgentProofTypes: {
+    env: 'PROVER_AGENT_PROOF_TYPES',
+    description: 'The types of proofs the prover agent can generate',
+    parseEnv: (val: string) =>
+      val
+        .split(',')
+        .map(v => ProvingRequestType[v as any])
+        .filter(v => typeof v === 'number'),
+  },
+  proverBrokerUrl: {
+    env: 'PROVER_BROKER_URL',
+    description: 'The URL where this agent takes jobs from',
+  },
+  realProofs: {
+    env: 'PROVER_REAL_PROOFS',
+    description: 'Whether to construct real proofs',
+    ...booleanConfigHelper(false),
+  },
+  proverTestDelayMs: {
+    env: 'PROVER_TEST_DELAY_MS',
+    description: 'Artificial delay to introduce to all operations to the test prover.',
+    ...numberConfigHelper(0),
+  },
+};
+
 export interface ProverAgentApi {
   setMaxConcurrency(maxConcurrency: number): Promise<void>;