fix: Enable block proposal mempool in p2p prover clients (#19190)

This allows prover clients to track block proposals as they are
broadcasted, which they rely on to mark txs as non evictable. Note that
prover clients were already doing this, but they were unable to protect
against an evil proposer blasting multiple proposals per block.

Author: spalladino

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

@@ -26,7 +26,7 @@ import { configureP2PClientAddresses, createLibP2PPeerIdFromPrivateKey, getPeerI
 export type P2PClientDeps<T extends P2PClientType> = {
   txPool?: TxPool;
   store?: AztecAsyncKVStore;
-  attestationPool?: T extends P2PClientType.Full ? AttestationPool : undefined;
+  attestationPool?: AttestationPool;
   logger?: Logger;
   txCollectionNodeSources?: TxSource[];
   p2pServiceFactory?: (...args: Parameters<(typeof LibP2PService)['new']>) => Promise<LibP2PService<T>>;
@@ -73,19 +73,14 @@ export async function createP2PClient<T extends P2PClientType>(
   );
   const l1Constants = await archiver.getL1Constants();
 
-  const mempools: MemPools<T> = {
+  const mempools: MemPools = {
     txPool:
       deps.txPool ??
       new AztecKVTxPool(store, archive, worldStateSynchronizer, telemetry, {
         maxTxPoolSize: config.maxTxPoolSize,
         archivedTxLimit: config.archivedTxLimit,
       }),
-    attestationPool:
-      clientType === P2PClientType.Full
-        ? ((deps.attestationPool ?? new KvAttestationPool(attestationStore, telemetry)) as T extends P2PClientType.Full
-            ? AttestationPool
-            : undefined)
-        : undefined,
+    attestationPool: deps.attestationPool ?? new KvAttestationPool(attestationStore, telemetry),
   };
 
   const p2pService = await createP2PService<T>(
@@ -147,7 +142,7 @@ async function createP2PService<T extends P2PClientType>(
   epochCache: EpochCacheInterface,
   store: AztecAsyncKVStore,
   peerStore: AztecLMDBStoreV2,
-  mempools: MemPools<T>,
+  mempools: MemPools,
   p2pServiceFactory: P2PClientDeps<T>['p2pServiceFactory'],
   packageVersion: string,
   logger: Logger,

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

@@ -69,7 +69,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
   private synchedLatestSlot: AztecAsyncSingleton<bigint>;
 
   private txPool: TxPool;
-  private attestationPool: T extends P2PClientType.Full ? AttestationPool : undefined;
+  private attestationPool: AttestationPool;
 
   private config: P2PConfig;
 
@@ -91,7 +91,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
     _clientType: T,
     private store: AztecAsyncKVStore,
     private l2BlockSource: L2BlockSource & ContractDataSource,
-    mempools: MemPools<T>,
+    mempools: MemPools,
     private p2pService: P2PService,
     private txCollection: TxCollection,
     config: Partial<P2PConfig> = {},
@@ -103,7 +103,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
 
     this.config = { ...getP2PDefaultConfig(), ...config };
     this.txPool = mempools.txPool;
-    this.attestationPool = mempools.attestationPool!;
+    this.attestationPool = mempools.attestationPool;
 
     this.txProvider = new TxProvider(
       this.txCollection,
@@ -282,10 +282,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
     const syncedProvenBlock = (await this.getSyncedProvenBlockNum()) + 1;
     const syncedFinalizedBlock = (await this.getSyncedFinalizedBlockNum()) + 1;
 
-    if (
-      (await this.txPool.isEmpty()) &&
-      (this.attestationPool === undefined || (await this.attestationPool?.isEmpty()))
-    ) {
+    if ((await this.txPool.isEmpty()) && (await this.attestationPool.isEmpty())) {
       // if mempools are empty, we don't care about syncing prior blocks
       this.initBlockStream(BlockNumber(this.latestBlockNumberAtStart));
       this.setCurrentState(P2PClientState.RUNNING);
@@ -389,19 +386,17 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
   }
 
   public async getAttestationsForSlot(slot: SlotNumber, proposalId?: string): Promise<BlockAttestation[]> {
-    return (
-      (await (proposalId
-        ? this.attestationPool?.getAttestationsForSlotAndProposal(slot, proposalId)
-        : this.attestationPool?.getAttestationsForSlot(slot))) ?? []
-    );
+    return await (proposalId
+      ? this.attestationPool.getAttestationsForSlotAndProposal(slot, proposalId)
+      : this.attestationPool.getAttestationsForSlot(slot));
   }
 
   public addAttestations(attestations: BlockAttestation[]): Promise<void> {
-    return this.attestationPool?.addAttestations(attestations) ?? Promise.resolve();
+    return this.attestationPool.addAttestations(attestations);
   }
 
   public deleteAttestation(attestation: BlockAttestation): Promise<void> {
-    return this.attestationPool?.deleteAttestations([attestation]) ?? Promise.resolve();
+    return this.attestationPool.deleteAttestations([attestation]);
   }
 
   // REVIEW: https://github.com/AztecProtocol/aztec-packages/issues/7963
@@ -782,7 +777,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
     await this.txPool.deleteTxs(txHashes, { permanently: true });
     await this.txPool.cleanupDeletedMinedTxs(lastBlockNum);
 
-    await this.attestationPool?.deleteAttestationsOlderThan(lastBlockSlot);
+    await this.attestationPool.deleteAttestationsOlderThan(lastBlockSlot);
 
     await this.synchedFinalizedBlockNumber.set(lastBlockNum);
     this.log.debug(`Synched to finalized block ${lastBlockNum} at slot ${lastBlockSlot}`);

yarn-project/p2p/src/mem_pools/interface.ts

@@ -1,12 +1,10 @@
-import type { P2PClientType } from '@aztec/stdlib/p2p';
-
 import type { AttestationPool } from './attestation_pool/attestation_pool.js';
 import type { TxPool } from './tx_pool/tx_pool.js';
 
 /**
  * A interface the combines all mempools
  */
-export type MemPools<T extends P2PClientType = P2PClientType.Full> = {
+export type MemPools = {
   txPool: TxPool;
-  attestationPool?: T extends P2PClientType.Full ? AttestationPool : undefined;
+  attestationPool: AttestationPool;
 };

yarn-project/p2p/src/services/libp2p/libp2p_service.test.ts

@@ -231,7 +231,7 @@ describe('LibP2PService', () => {
     function setProposalTxHashes(
       svc: {
         mempools: {
-          attestationPool?: {
+          attestationPool: {
             getBlockProposal: (id: string) => Promise<{ txHashes: { toString(): string }[] } | undefined>;
           };
         };

yarn-project/p2p/src/services/libp2p/libp2p_service.ts

@@ -153,7 +153,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
     private peerDiscoveryService: PeerDiscoveryService,
     private reqresp: ReqRespInterface,
     private peerManager: PeerManagerInterface,
-    protected mempools: MemPools<T>,
+    protected mempools: MemPools,
     private archiver: L2BlockSource & ContractDataSource,
     private epochCache: EpochCacheInterface,
     private proofVerifier: ClientProtocolCircuitVerifier,
@@ -185,7 +185,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
 
     // Use FishermanAttestationValidator in fisherman mode to validate attestation payloads against proposals
     this.attestationValidator = config.fishermanMode
-      ? new FishermanAttestationValidator(epochCache, mempools.attestationPool!, telemetry)
+      ? new FishermanAttestationValidator(epochCache, mempools.attestationPool, telemetry)
       : new AttestationValidator(epochCache);
     this.blockProposalValidator = new BlockProposalValidator(epochCache, { txsPermitted: !config.disableTransactions });
 
@@ -215,7 +215,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
     config: P2PConfig,
     peerId: PeerId,
     deps: {
-      mempools: MemPools<T>;
+      mempools: MemPools;
       l2BlockSource: L2BlockSource & ContractDataSource;
       epochCache: EpochCacheInterface;
       proofVerifier: ClientProtocolCircuitVerifier;
@@ -486,8 +486,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
       [ReqRespSubProtocol.BLOCK]: blockHandler.bind(this),
     };
 
-    // Only handle block transactions request if attestation pool is available to the client
-    if (this.mempools.attestationPool && !this.config.disableTransactions) {
+    if (!this.config.disableTransactions) {
       const blockTxsHandler = reqRespBlockTxsHandler(this.mempools.attestationPool, this.mempools.txPool);
       requestResponseHandlers[ReqRespSubProtocol.BLOCK_TXS] = blockTxsHandler.bind(this);
     }
@@ -809,7 +808,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
   private async processAttestationFromPeer(payloadData: Buffer, msgId: string, source: PeerId): Promise<void> {
     const validationFunc: () => Promise<ReceivedMessageValidationResult<BlockAttestation>> = async () => {
       const attestation = BlockAttestation.fromBuffer(payloadData);
-      const pool = this.mempools.attestationPool!;
+      const pool = this.mempools.attestationPool;
       const isValid = await this.validateAttestation(source, attestation);
       const exists = isValid && (await pool.hasAttestation(attestation));
 
@@ -866,7 +865,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
       },
     );
 
-    await this.mempools.attestationPool!.addAttestations([attestation]);
+    await this.mempools.attestationPool.addAttestations([attestation]);
   }
 
   private async processBlockFromPeer(payloadData: Buffer, msgId: string, source: PeerId): Promise<void> {
@@ -875,10 +874,8 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
       const isValid = await this.validateBlockProposal(source, block);
       const pool = this.mempools.attestationPool;
 
-      // Note that we dont have an attestation pool if we're a prover node, but we still
-      // subscribe to block proposal topics in order to prevent their txs from being cleared.
-      const exists = isValid && pool ? await pool.hasBlockProposal(block) : false;
-      const canAdd = isValid && pool ? await pool.canAddProposal(block) : true; // If pool DNE, set canAdd to true to avoid peer penalization (the block is not added to the pool)
+      const exists = isValid && (await pool.hasBlockProposal(block));
+      const canAdd = isValid && (await pool.canAddProposal(block));
 
       this.logger.trace(`Validate propagated block proposal`, {
         isValid,
@@ -934,14 +931,12 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
       archive: block.archive.toString(),
       source: sender.toString(),
     });
-    const attestationsForPreviousSlot = await this.mempools.attestationPool?.getAttestationsForSlot(previousSlot);
-    if (attestationsForPreviousSlot !== undefined) {
-      this.logger.verbose(`Received ${attestationsForPreviousSlot.length} attestations for slot ${previousSlot}`);
-    }
+    const attestationsForPreviousSlot = await this.mempools.attestationPool.getAttestationsForSlot(previousSlot);
+    this.logger.verbose(`Received ${attestationsForPreviousSlot.length} attestations for slot ${previousSlot}`);
 
     // Attempt to add proposal, then mark the txs in this proposal as non-evictable
     try {
-      await this.mempools.attestationPool?.addBlockProposal(block);
+      await this.mempools.attestationPool.addBlockProposal(block);
     } catch (err: unknown) {
       // Drop proposals if we hit per-slot cap in the attestation pool; rethrow unknown errors
       if (err instanceof ProposalSlotCapExceededError) {
@@ -1047,7 +1042,7 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
       }
 
       // Given proposal (should have locally), ensure returned txs are valid subset and match request indices
-      const proposal = await this.mempools.attestationPool?.getBlockProposal(request.blockHash.toString());
+      const proposal = await this.mempools.attestationPool.getBlockProposal(request.blockHash.toString());
       if (proposal) {
         // Build intersected indices
         const intersectIdx = request.txIndices.getTrueIndices().filter(i => response.txIndices.isSet(i));

yarn-project/p2p/src/services/reqresp/protocols/tx.ts

@@ -1,5 +1,4 @@
 import { chunk } from '@aztec/foundation/collection';
-import type { P2PClientType } from '@aztec/stdlib/p2p';
 import { TxArray, TxHash, TxHashArray } from '@aztec/stdlib/tx';
 
 import type { PeerId } from '@libp2p/interface';
@@ -16,7 +15,7 @@ import { ReqRespStatus, ReqRespStatusError } from '../status.js';
  * @param mempools - the mempools
  * @returns the Tx request handler
  */
-export function reqRespTxHandler<T extends P2PClientType>(mempools: MemPools<T>): ReqRespSubProtocolHandler {
+export function reqRespTxHandler(mempools: MemPools): ReqRespSubProtocolHandler {
   /**
    * Handler for tx requests
    * @param msg - the tx request message

yarn-project/p2p/src/test-helpers/mock-pubsub.ts

@@ -39,7 +39,7 @@ export function getMockPubSubP2PServiceFactory<T extends P2PClientType>(
     peerId: PeerId,
     deps: {
       packageVersion: string;
-      mempools: MemPools<T>;
+      mempools: MemPools;
       l2BlockSource: L2BlockSource & ContractDataSource;
       epochCache: EpochCacheInterface;
       proofVerifier: ClientProtocolCircuitVerifier;

yarn-project/p2p/src/test-helpers/reqresp-nodes.ts

@@ -112,7 +112,7 @@ export async function createTestLibP2PService<T extends P2PClientType>(
   archiver: L2BlockSource & ContractDataSource,
   worldStateSynchronizer: WorldStateSynchronizer,
   epochCache: EpochCache,
-  mempools: MemPools<T>,
+  mempools: MemPools,
   telemetry: TelemetryClient,
   port: number = 0,
   peerId?: PeerId,

yarn-project/p2p/src/testbench/p2p_client_testbench_worker.ts

@@ -126,7 +126,7 @@ class TestLibP2PService<T extends P2PClientType = P2PClientType.Full> extends Li
     peerDiscoveryService: PeerDiscoveryService,
     reqresp: ReqResp,
     peerManager: PeerManager,
-    mempools: MemPools<T>,
+    mempools: MemPools,
     archiver: L2BlockSource & ContractDataSource,
     epochCache: EpochCacheInterface,
     proofVerifier: ClientProtocolCircuitVerifier,