feat!: Use combined p2p and http prover coordination

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

@@ -12,7 +12,7 @@ import {
   getProverNodeConfigFromEnv,
   proverNodeConfigMappings,
 } from '@aztec/prover-node';
-import { P2PApiSchema, ProverNodeApiSchema, type ProvingJobBroker } from '@aztec/stdlib/interfaces/server';
+import { ProverNodeApiSchema, type ProvingJobBroker } from '@aztec/stdlib/interfaces/server';
 import { initTelemetryClient, makeTracedFetch, telemetryClientConfigMappings } from '@aztec/telemetry-client';
 import { getGenesisValues } from '@aztec/world-state/testing';
 
@@ -105,14 +105,9 @@ export async function startProverNode(
 
   await preloadCrsDataForVerifying(proverConfig, userLog);
 
-  const proverNode = await createProverNode(proverConfig, { telemetry, broker }, { prefilledPublicData });
+  const proverNode = await createProverNode(proverConfig, { telemetry, broker }, { prefilledPublicData }, services);
   services.proverNode = [proverNode, ProverNodeApiSchema];
 
-  const p2p = proverNode.getP2P();
-  if (p2p) {
-    services.p2p = [proverNode.getP2P(), P2PApiSchema];
-  }
-
   if (!proverConfig.proverBrokerUrl) {
     services.provingJobSource = [proverNode.getProver().getProvingJobSource(), ProvingJobConsumerSchema];
   }

yarn-project/end-to-end/src/e2e_prover/e2e_prover_test.ts

@@ -282,7 +282,7 @@ export class FullProverTest {
     this.logger.verbose('Starting prover node');
     const proverConfig: ProverNodeConfig = {
       ...this.context.aztecNodeConfig,
-      proverCoordinationNodeUrl: undefined,
+      proverCoordinationNodeUrls: [],
       dataDirectory: undefined,
       proverId: this.proverAddress.toField(),
       realProofs: this.realProofs,

yarn-project/end-to-end/src/e2e_snapshot_sync.test.ts

@@ -68,7 +68,7 @@ describe('e2e_snapshot_sync', () => {
         ...context.config,
         dataDirectory: join(context.config.dataDirectory!, randomBytes(8).toString('hex')),
         p2pEnabled: true, // So we don't need prover coordination
-        proverCoordinationNodeUrl: undefined,
+        proverCoordinationNodeUrls: [],
         proverNodeMaxPendingJobs: 10,
         proverNodeMaxParallelBlocksPerEpoch: 32,
         proverNodePollingIntervalMs: 200,

yarn-project/end-to-end/src/fixtures/utils.ts

@@ -801,7 +801,7 @@ export async function createAndSyncProverNode(
   // Prover node config is for simulated proofs
   const proverConfig: ProverNodeConfig = {
     ...aztecNodeConfig,
-    proverCoordinationNodeUrl: undefined,
+    proverCoordinationNodeUrls: [],
     dataDirectory: undefined,
     realProofs: false,
     proverAgentCount: 2,

yarn-project/foundation/src/config/env_var.ts

@@ -132,7 +132,7 @@ export type EnvVar =
   | 'PROVER_BROKER_BATCH_INTERVAL_MS'
   | 'PROVER_BROKER_BATCH_SIZE'
   | 'PROVER_BROKER_MAX_EPOCHS_TO_KEEP_RESULTS_FOR'
-  | 'PROVER_COORDINATION_NODE_URL'
+  | 'PROVER_COORDINATION_NODE_URLS'
   | 'PROVER_FAILED_PROOF_STORE'
   | 'PROVER_ID'
   | 'PROVER_NODE_POLLING_INTERVAL_MS'

yarn-project/p2p/src/client/p2p_client.ts

@@ -10,7 +10,7 @@ import type {
   PublishedL2Block,
 } from '@aztec/stdlib/block';
 import type { ContractDataSource } from '@aztec/stdlib/contract';
-import type { P2PApi, PeerInfo, ProverCoordination } from '@aztec/stdlib/interfaces/server';
+import type { P2PApi, PeerInfo } from '@aztec/stdlib/interfaces/server';
 import { BlockAttestation, type BlockProposal, ConsensusPayload, type P2PClientType } from '@aztec/stdlib/p2p';
 import type { Tx, TxHash } from '@aztec/stdlib/tx';
 import {
@@ -58,118 +58,144 @@ export interface P2PSyncState {
 /**
  * Interface of a P2P client.
  **/
-export type P2P<T extends P2PClientType = P2PClientType.Full> = ProverCoordination &
-  P2PApi<T> & {
-    /**
-     * Broadcasts a block proposal to other peers.
-     *
-     * @param proposal - the block proposal
-     */
-    broadcastProposal(proposal: BlockProposal): void;
-
-    /**
-     * Registers a callback from the validator client that determines how to behave when
-     * foreign block proposals are received
-     *
-     * @param handler - A function taking a received block proposal and producing an attestation
-     */
-    // REVIEW: https://github.com/AztecProtocol/aztec-packages/issues/7963
-    // ^ This pattern is not my favorite (md)
-    registerBlockProposalHandler(handler: (block: BlockProposal) => Promise<BlockAttestation | undefined>): void;
-
-    /**
-     * Request a list of transactions from another peer by their tx hashes.
-     * @param txHashes - Hashes of the txs to query.
-     * @returns A list of transactions or undefined if the transactions are not found.
-     */
-    requestTxs(txHashes: TxHash[]): Promise<(Tx | undefined)[]>;
-
-    /**
-     * Request a transaction from another peer by its tx hash.
-     * @param txHash - Hash of the tx to query.
-     */
-    requestTxByHash(txHash: TxHash): Promise<Tx | undefined>;
-
-    /**
-     * Verifies the 'tx' and, if valid, adds it to local tx pool and forwards it to other peers.
-     * @param tx - The transaction.
-     **/
-    sendTx(tx: Tx): Promise<void>;
-
-    /**
-     * Deletes 'txs' from the pool, given hashes.
-     * NOT used if we use sendTx as reconcileTxPool will handle this.
-     * @param txHashes - Hashes to check.
-     **/
-    deleteTxs(txHashes: TxHash[]): Promise<void>;
-
-    /**
-     * Returns a transaction in the transaction pool by its hash.
-     * @param txHash  - Hash of tx to return.
-     * @returns A single tx or undefined.
-     */
-    getTxByHashFromPool(txHash: TxHash): Promise<Tx | undefined>;
-
-    /**
-     * Returns a transaction in the transaction pool by its hash, requesting it from the network if it is not found.
-     * @param txHash  - Hash of tx to return.
-     * @returns A single tx or undefined.
-     */
-    getTxByHash(txHash: TxHash): Promise<Tx | undefined>;
-
-    /**
-     * Returns an archived transaction from the transaction pool by its hash.
-     * @param txHash  - Hash of tx to return.
-     * @returns A single tx or undefined.
-     */
-    getArchivedTxByHash(txHash: TxHash): Promise<Tx | undefined>;
-
-    /**
-     * Returns whether the given tx hash is flagged as pending or mined.
-     * @param txHash - Hash of the tx to query.
-     * @returns Pending or mined depending on its status, or undefined if not found.
-     */
-    getTxStatus(txHash: TxHash): Promise<'pending' | 'mined' | undefined>;
-
-    /** Returns an iterator over pending txs on the mempool. */
-    iteratePendingTxs(): AsyncIterableIterator<Tx>;
-
-    /** Returns the number of pending txs in the mempool. */
-    getPendingTxCount(): Promise<number>;
-
-    /**
-     * Starts the p2p client.
-     * @returns A promise signalling the completion of the block sync.
-     */
-    start(): Promise<void>;
-
-    /**
-     * Stops the p2p client.
-     * @returns A promise signalling the completion of the stop process.
-     */
-    stop(): Promise<void>;
-
-    /**
-     * Indicates if the p2p client is ready for transaction submission.
-     * @returns A boolean flag indicating readiness.
-     */
-    isReady(): boolean;
-
-    /**
-     * Returns the current status of the p2p client.
-     */
-    getStatus(): Promise<P2PSyncState>;
-
-    /**
-     * Returns the ENR of this node, if any.
-     */
-    getEnr(): ENR | undefined;
-
-    /** Identifies a p2p client. */
-    isP2PClient(): true;
-
-    updateP2PConfig(config: Partial<P2PConfig>): Promise<void>;
-  };
+export type P2P<T extends P2PClientType = P2PClientType.Full> = P2PApi<T> & {
+  /**
+   * Broadcasts a block proposal to other peers.
+   *
+   * @param proposal - the block proposal
+   */
+  broadcastProposal(proposal: BlockProposal): void;
+
+  /**
+   * Registers a callback from the validator client that determines how to behave when
+   * foreign block proposals are received
+   *
+   * @param handler - A function taking a received block proposal and producing an attestation
+   */
+  // REVIEW: https://github.com/AztecProtocol/aztec-packages/issues/7963
+  // ^ This pattern is not my favorite (md)
+  registerBlockProposalHandler(handler: (block: BlockProposal) => Promise<BlockAttestation | undefined>): void;
+
+  /**
+   * Request a list of transactions from another peer by their tx hashes.
+   * @param txHashes - Hashes of the txs to query.
+   * @returns A list of transactions or undefined if the transactions are not found.
+   */
+  requestTxs(txHashes: TxHash[]): Promise<(Tx | undefined)[]>;
+
+  /**
+   * Request a transaction from another peer by its tx hash.
+   * @param txHash - Hash of the tx to query.
+   */
+  requestTxByHash(txHash: TxHash): Promise<Tx | undefined>;
+
+  /**
+   * Verifies the 'tx' and, if valid, adds it to local tx pool and forwards it to other peers.
+   * @param tx - The transaction.
+   **/
+  sendTx(tx: Tx): Promise<void>;
+
+  /**
+   * Adds transactions to the pool. Does not send to peers or validate the tx.
+   * @param txs - The transactions.
+   **/
+  addTxs(txs: Tx[]): Promise<void>;
+
+  /**
+   * Deletes 'txs' from the pool, given hashes.
+   * NOT used if we use sendTx as reconcileTxPool will handle this.
+   * @param txHashes - Hashes to check.
+   **/
+  deleteTxs(txHashes: TxHash[]): Promise<void>;
+
+  /**
+   * Returns a transaction in the transaction pool by its hash.
+   * @param txHash  - Hash of tx to return.
+   * @returns A single tx or undefined.
+   */
+  getTxByHashFromPool(txHash: TxHash): Promise<Tx | undefined>;
+
+  /**
+   * Returns transactions in the transaction pool by hash.
+   * @param txHashes  - Hashes of txs to return.
+   * @returns An array of txs or undefined.
+   */
+  getTxsByHashFromPool(txHashes: TxHash[]): Promise<(Tx | undefined)[]>;
+
+  /**
+   * Checks if transactions exist in the pool and returns the hashes of those that don't
+   * @param txHashes - The hashes of the transactions to check for
+   * @returns The transaction hashes if not found
+   */
+  getUnavailableTxsFromPool(txHashes: TxHash[]): Promise<TxHash[]>;
+
+  /**
+   * Returns a transaction in the transaction pool by its hash, requesting it from the network if it is not found.
+   * @param txHash  - Hash of tx to return.
+   * @returns A single tx or undefined.
+   */
+  getTxByHash(txHash: TxHash): Promise<Tx | undefined>;
+
+  /**
+   * Returns transactions in the transaction pool by hash, requesting from the network if not found.
+   * @param txHashes  - Hashes of tx to return.
+   * @returns An array of tx or undefined.
+   */
+  getTxsByHash(txHashes: TxHash[]): Promise<(Tx | undefined)[]>;
+
+  /**
+   * Returns an archived transaction from the transaction pool by its hash.
+   * @param txHash  - Hash of tx to return.
+   * @returns A single tx or undefined.
+   */
+  getArchivedTxByHash(txHash: TxHash): Promise<Tx | undefined>;
+
+  /**
+   * Returns whether the given tx hash is flagged as pending or mined.
+   * @param txHash - Hash of the tx to query.
+   * @returns Pending or mined depending on its status, or undefined if not found.
+   */
+  getTxStatus(txHash: TxHash): Promise<'pending' | 'mined' | undefined>;
+
+  /** Returns an iterator over pending txs on the mempool. */
+  iteratePendingTxs(): AsyncIterableIterator<Tx>;
+
+  /** Returns the number of pending txs in the mempool. */
+  getPendingTxCount(): Promise<number>;
+
+  /**
+   * Starts the p2p client.
+   * @returns A promise signalling the completion of the block sync.
+   */
+  start(): Promise<void>;
+
+  /**
+   * Stops the p2p client.
+   * @returns A promise signalling the completion of the stop process.
+   */
+  stop(): Promise<void>;
+
+  /**
+   * Indicates if the p2p client is ready for transaction submission.
+   * @returns A boolean flag indicating readiness.
+   */
+  isReady(): boolean;
+
+  /**
+   * Returns the current status of the p2p client.
+   */
+  getStatus(): Promise<P2PSyncState>;
+
+  /**
+   * Returns the ENR of this node, if any.
+   */
+  getEnr(): ENR | undefined;
+
+  /** Identifies a p2p client. */
+  isP2PClient(): true;
+
+  updateP2PConfig(config: Partial<P2PConfig>): Promise<void>;
+};
 
 /**
  * The P2P client implementation.
@@ -520,6 +546,19 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
     return this.txPool.getTxByHash(txHash);
   }
 
+  /**
+   * Returns transactions in the transaction pool by hash.
+   * @param txHashes - Hashes of the transactions to look for.
+   * @returns The txs found, not necessarily on the same order as the hashes.
+   */
+  getTxsByHashFromPool(txHashes: TxHash[]): Promise<(Tx | undefined)[]> {
+    return this.txPool.getTxsByHash(txHashes);
+  }
+
+  getUnavailableTxsFromPool(txHashes: TxHash[]): Promise<TxHash[]> {
+    return this.txPool.getUnavailableTxs(txHashes);
+  }
+
   /**
    * Returns a transaction in the transaction pool by its hash.
    * If the transaction is not in the pool, it will be requested from the network.
@@ -571,11 +610,19 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
    * @returns Empty promise.
    **/
   public async sendTx(tx: Tx): Promise<void> {
-    this.#assertIsReady();
-    await this.txPool.addTxs([tx]);
+    await this.addTxs([tx]);
     this.p2pService.propagate(tx);
   }
 
+  /**
+   * Adds transactions to the pool. Does not send to peers or validate the txs.
+   * @param txs - The transactions.
+   **/
+  public async addTxs(txs: Tx[]): Promise<void> {
+    this.#assertIsReady();
+    await this.txPool.addTxs(txs);
+  }
+
   /**
    * Returns whether the given tx hash is flagged as pending or mined.
    * @param txHash - Hash of the tx to query.

yarn-project/p2p/src/mem_pools/tx_pool/aztec_kv_tx_pool.ts

@@ -218,6 +218,29 @@ export class AztecKVTxPool implements TxPool {
     return undefined;
   }
 
+  async getTxsByHash(txHashes: TxHash[]): Promise<(Tx | undefined)[]> {
+    const txs = await Promise.all(txHashes.map(txHash => this.#txs.getAsync(txHash.toString())));
+    return txs.map((buffer, index) => {
+      if (buffer) {
+        const tx = Tx.fromBuffer(buffer);
+        tx.setTxHash(txHashes[index]);
+        return tx;
+      }
+      return undefined;
+    });
+  }
+
+  async getUnavailableTxs(txHashes: TxHash[]): Promise<TxHash[]> {
+    const results = await Promise.all(txHashes.map(txHash => this.#txs.hasAsync(txHash.toString())));
+    return txHashes
+      .map((txHash, index) => {
+        if (!results[index]) {
+          return txHash;
+        }
+      })
+      .filter(txHash => txHash !== undefined) as TxHash[];
+  }
+
   /**
    * Checks if an archived tx exists and returns it.
    * @param txHash - The tx hash.