fix: Ignore duplicate p2p messages after validation

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

@@ -21,6 +21,15 @@ export interface AttestationPool {
    */
   getBlockProposal(id: string): Promise<BlockProposal | undefined>;
 
+  /**
+   * Check if a block proposal exists in the pool
+   *
+   * @param idOrProposal - The ID of the block proposal or the block proposal itself to check. The ID is proposal.payload.archive
+   *
+   * @return True if the block proposal exists, false otherwise.
+   */
+  hasBlockProposal(idOrProposal: string | BlockProposal): Promise<boolean>;
+
   /**
    * AddAttestations
    *
@@ -84,6 +93,14 @@ export interface AttestationPool {
    */
   getAttestationsForSlotAndProposal(slot: bigint, proposalId: string): Promise<BlockAttestation[]>;
 
+  /**
+   * Check if a specific attestation exists in the pool
+   *
+   * @param attestation - The attestation to check
+   * @return True if the attestation exists, false otherwise
+   */
+  hasAttestation(attestation: BlockAttestation): Promise<boolean>;
+
   /** Returns whether the pool is empty. */
   isEmpty(): Promise<boolean>;
 }

yarn-project/p2p/src/mem_pools/attestation_pool/attestation_pool_test_suite.ts

@@ -72,6 +72,11 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
     expect(retrievedAttestations.length).toBe(attestations.length);
     compareAttestations(retrievedAttestations, attestations);
 
+    // Check hasAttestation for added attestations
+    for (const attestation of attestations) {
+      expect(await ap.hasAttestation(attestation)).toBe(true);
+    }
+
     const retrievedAttestationsForSlot = await ap.getAttestationsForSlot(BigInt(slotNumber));
     expect(retrievedAttestationsForSlot.length).toBe(attestations.length);
     compareAttestations(retrievedAttestationsForSlot, attestations);
@@ -85,6 +90,7 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
     );
     expect(retrievedAttestationsAfterAdd.length).toBe(attestations.length + 1);
     compareAttestations(retrievedAttestationsAfterAdd, [...attestations, newAttestation]);
+    expect(await ap.hasAttestation(newAttestation)).toBe(true);
     const retrievedAttestationsForSlotAfterAdd = await ap.getAttestationsForSlot(BigInt(slotNumber));
     expect(retrievedAttestationsForSlotAfterAdd.length).toBe(attestations.length + 1);
     compareAttestations(retrievedAttestationsForSlotAfterAdd, [...attestations, newAttestation]);
@@ -97,6 +103,11 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
       archive.toString(),
     );
     expect(retreivedAttestationsAfterDelete.length).toBe(0);
+    // Check hasAttestation after deletion
+    for (const attestation of attestations) {
+      expect(await ap.hasAttestation(attestation)).toBe(false);
+    }
+    expect(await ap.hasAttestation(newAttestation)).toBe(false);
   });
 
   it('should handle duplicate proposals in a slot', async () => {
@@ -170,10 +181,20 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
     expect(retreivedAttestations.length).toBe(NUMBER_OF_SIGNERS_PER_TEST);
     compareAttestations(retreivedAttestations, attestations);
 
+    // Check hasAttestation before deletion
+    for (const attestation of attestations) {
+      expect(await ap.hasAttestation(attestation)).toBe(true);
+    }
+
     await ap.deleteAttestations(attestations);
 
     const gottenAfterDelete = await ap.getAttestationsForSlotAndProposal(BigInt(slotNumber), proposalId);
     expect(gottenAfterDelete.length).toBe(0);
+
+    // Check hasAttestation after deletion
+    for (const attestation of attestations) {
+      expect(await ap.hasAttestation(attestation)).toBe(false);
+    }
   });
 
   it('should blanket delete attestations per slot', async () => {
@@ -265,12 +286,19 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
 
       expect(retrievedProposal).toBeDefined();
       expect(retrievedProposal!).toEqual(proposal);
+
+      // Check hasBlockProposal with both id and object
+      expect(await ap.hasBlockProposal(proposalId)).toBe(true);
+      expect(await ap.hasBlockProposal(proposal)).toBe(true);
     });
 
     it('should return undefined for non-existent block proposal', async () => {
       const nonExistentId = Fr.random().toString();
       const retrievedProposal = await ap.getBlockProposal(nonExistentId);
       expect(retrievedProposal).toBeUndefined();
+
+      // Check hasBlockProposal returns false for non-existent proposal
+      expect(await ap.hasBlockProposal(nonExistentId)).toBe(false);
     });
 
     it('should update block proposal if added twice with same id', async () => {
@@ -323,13 +351,15 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
       // Verify proposal exists
       let retrievedProposal = await ap.getBlockProposal(proposalId);
       expect(retrievedProposal).toBeDefined();
+      expect(await ap.hasBlockProposal(proposalId)).toBe(true);
 
       // Delete attestations for slot and proposal
       await ap.deleteAttestationsForSlotAndProposal(BigInt(slotNumber), proposalId);
 
       // Proposal should be deleted
       retrievedProposal = await ap.getBlockProposal(proposalId);
       expect(retrievedProposal).toBeUndefined();
+      expect(await ap.hasBlockProposal(proposalId)).toBe(false);
     });
 
     it('should delete block proposal when deleting attestations for slot', async () => {
@@ -344,13 +374,15 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
       // Verify proposal exists
       let retrievedProposal = await ap.getBlockProposal(proposalId);
       expect(retrievedProposal).toBeDefined();
+      expect(await ap.hasBlockProposal(proposal)).toBe(true);
 
       // Delete attestations for slot
       await ap.deleteAttestationsForSlot(BigInt(slotNumber));
 
       // Proposal should be deleted
       retrievedProposal = await ap.getBlockProposal(proposalId);
       expect(retrievedProposal).toBeUndefined();
+      expect(await ap.hasBlockProposal(proposal)).toBe(false);
     });
 
     it('should be able to fetch both block proposal and attestations', async () => {
@@ -372,8 +404,13 @@ export function describeAttestationPool(getAttestationPool: () => AttestationPoo
 
       expect(retrievedProposal).toBeDefined();
       expect(retrievedProposal).toEqual(proposal);
+      expect(await ap.hasBlockProposal(proposalId)).toBe(true);
 
       compareAttestations(retrievedAttestations, attestations);
+      // Check hasAttestation for all attestations
+      for (const attestation of attestations) {
+        expect(await ap.hasAttestation(attestation)).toBe(true);
+      }
     });
   });
 }

yarn-project/p2p/src/mem_pools/attestation_pool/kv_attestation_pool.ts

@@ -213,6 +213,22 @@ export class KvAttestationPool implements AttestationPool {
     });
   }
 
+  public async hasAttestation(attestation: BlockAttestation): Promise<boolean> {
+    const slotNumber = attestation.payload.header.slotNumber;
+    const proposalId = attestation.archive;
+    const sender = attestation.getSender();
+
+    // Attestations with invalid signatures are never in the pool
+    if (!sender) {
+      return false;
+    }
+
+    const address = sender.toString();
+    const key = this.getAttestationKey(slotNumber, proposalId, address);
+
+    return await this.attestations.hasAsync(key);
+  }
+
   public async getBlockProposal(id: string): Promise<BlockProposal | undefined> {
     const buffer = await this.proposals.getAsync(id);
     try {
@@ -226,6 +242,11 @@ export class KvAttestationPool implements AttestationPool {
     return Promise.resolve(undefined);
   }
 
+  public async hasBlockProposal(idOrProposal: string | BlockProposal): Promise<boolean> {
+    const id = typeof idOrProposal === 'string' ? idOrProposal : idOrProposal.payload.archive.toString();
+    return await this.proposals.hasAsync(id);
+  }
+
   public async addBlockProposal(blockProposal: BlockProposal): Promise<void> {
     await this.store.transactionAsync(async () => {
       await this.proposalsForSlot.set(blockProposal.slotNumber.toString(), blockProposal.archive.toString());

yarn-project/p2p/src/mem_pools/attestation_pool/memory_attestation_pool.ts

@@ -173,6 +173,29 @@ export class InMemoryAttestationPool implements AttestationPool {
     return Promise.resolve();
   }
 
+  public hasAttestation(attestation: BlockAttestation): Promise<boolean> {
+    const slotNumber = attestation.payload.header.slotNumber;
+    const proposalId = attestation.archive.toString();
+    const sender = attestation.getSender();
+
+    // Attestations with invalid signatures are never in the pool
+    if (!sender) {
+      return Promise.resolve(false);
+    }
+
+    const slotAttestationMap = this.attestations.get(slotNumber.toBigInt());
+    if (!slotAttestationMap) {
+      return Promise.resolve(false);
+    }
+
+    const proposalAttestationMap = slotAttestationMap.get(proposalId);
+    if (!proposalAttestationMap) {
+      return Promise.resolve(false);
+    }
+
+    return Promise.resolve(proposalAttestationMap.has(sender.toString()));
+  }
+
   public addBlockProposal(blockProposal: BlockProposal): Promise<void> {
     // We initialize slot-proposal mapping if it does not exist
     // This is important to ensure we can delete this proposal if there were not attestations for it
@@ -186,6 +209,11 @@ export class InMemoryAttestationPool implements AttestationPool {
   public getBlockProposal(id: string): Promise<BlockProposal | undefined> {
     return Promise.resolve(this.proposals.get(id));
   }
+
+  public hasBlockProposal(idOrProposal: string | BlockProposal): Promise<boolean> {
+    const id = typeof idOrProposal === 'string' ? idOrProposal : idOrProposal.payload.archive.toString();
+    return Promise.resolve(this.proposals.has(id));
+  }
 }
 
 /**

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

@@ -273,6 +273,11 @@ export class AztecKVTxPool extends (EventEmitter as new () => TypedEventEmitter<
     return await Promise.all(txHashes.map(txHash => this.#txs.hasAsync(txHash.toString())));
   }
 
+  async hasTx(txHash: TxHash): Promise<boolean> {
+    const result = await this.hasTxs([txHash]);
+    return result[0];
+  }
+
   /**
    * Checks if an archived tx exists and returns it.
    * @param txHash - The tx hash.

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

@@ -151,6 +151,11 @@ export class InMemoryTxPool extends (EventEmitter as new () => TypedEventEmitter
     return Promise.resolve(txHashes.map(txHash => this.txs.has(txHash.toBigInt())));
   }
 
+  async hasTx(txHash: TxHash): Promise<boolean> {
+    const result = await this.hasTxs([txHash]);
+    return result[0];
+  }
+
   public getArchivedTxByHash(): Promise<Tx | undefined> {
     return Promise.resolve(undefined);
   }

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

@@ -43,6 +43,13 @@ export interface TxPool extends TypedEventEmitter<TxPoolEvents> {
    */
   hasTxs(txHashes: TxHash[]): Promise<boolean[]>;
 
+  /**
+   * Checks if a transaction exists in the pool
+   * @param txHash - The hash of the transaction to check for
+   * @returns True if the transaction exists, false otherwise
+   */
+  hasTx(txHash: TxHash): Promise<boolean>;
+
   /**
    * Checks if an archived transaction exists in the pool and returns it.
    * @param txHash - The hash of the transaction, used as an ID.