fix(p2p): de-duplicate and cap tx req/resp requests

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

@@ -1,8 +1,12 @@
 import { MAX_TX_SIZE_KB } from '@aztec-labs/stdlib/p2p';
 import { TxHash, TxHashArray } from '@aztec-labs/stdlib/tx';
 import { describe, expect, it } from '@jest/globals';
+import type { PeerId } from '@libp2p/interface';
+import { mock, mockDeep } from 'jest-mock-extended';
 
-import { calculateTxResponseSize } from './tx.js';
+import type { MemPools } from '../../../mem_pools/interface.js';
+import { ReqRespStatus, ReqRespStatusError } from '../status.js';
+import { calculateTxResponseSize, reqRespTxHandler } from './tx.js';
 
 describe('calculateTxResponseSize', () => {
   it('should return correct size for a single tx hash', () => {
@@ -28,7 +32,7 @@ describe('calculateTxResponseSize', () => {
 
   it('should fall back to single tx size for a raw TxHash buffer (not TxHashArray)', () => {
     // A raw TxHash (32 bytes) is not a valid TxHashArray serialization.
-    // TxHashArray.fromBuffer silently returns empty array on parse failure.
+    // TxHashArray.fromBuffer returns an empty array (no error) on a parse failure.
     const rawHash = TxHash.random().toBuffer();
 
     expect(calculateTxResponseSize(rawHash)).toBe(MAX_TX_SIZE_KB + 1);
@@ -49,3 +53,60 @@ describe('calculateTxResponseSize', () => {
     expect(calculateTxResponseSize(buffer)).toBe(MAX_TX_SIZE_KB + 1);
   });
 });
+
+describe('reqRespTxHandler', () => {
+  const peerId = mock<PeerId>();
+
+  const makeMempools = (getTxByHash: (h: TxHash) => Promise<undefined>): MemPools => {
+    const mempools = mockDeep<MemPools>();
+    mempools.txPool.getTxByHash.mockImplementation(getTxByHash);
+    return mempools;
+  };
+
+  it('serves a repeated hash only once (de-duplicates before pool reads)', async () => {
+    const h = TxHash.random();
+    const lookups: string[] = [];
+    const mempools = makeMempools((x: TxHash) => {
+      lookups.push(x.toString());
+      return Promise.resolve(undefined);
+    });
+    const request = new TxHashArray(h, h, h, h, h).toBuffer();
+
+    await reqRespTxHandler(mempools)(peerId, request);
+
+    // One repeated hash must cause exactly one pool read, not one per copy.
+    expect(lookups).toEqual([h.toString()]);
+  });
+
+  it('still serves each distinct hash in a normal batch', async () => {
+    const hashes = Array.from({ length: 3 }, () => TxHash.random());
+    const lookups: string[] = [];
+    const mempools = makeMempools((x: TxHash) => {
+      lookups.push(x.toString());
+      return Promise.resolve(undefined);
+    });
+    const request = new TxHashArray(...hashes).toBuffer();
+
+    await reqRespTxHandler(mempools)(peerId, request);
+
+    expect(lookups.sort()).toEqual(hashes.map(h => h.toString()).sort());
+  });
+
+  it('rejects a request with too many hashes', async () => {
+    const lookups: string[] = [];
+    const mempools = makeMempools((x: TxHash) => {
+      lookups.push(x.toString());
+      return Promise.resolve(undefined);
+    });
+    const request = new TxHashArray(...Array.from({ length: 101 }, () => TxHash.random())).toBuffer();
+
+    const err = await reqRespTxHandler(mempools)(peerId, request).then(
+      () => undefined,
+      e => e,
+    );
+    expect(err).toBeInstanceOf(ReqRespStatusError);
+    expect((err as ReqRespStatusError).status).toBe(ReqRespStatus.BADLY_FORMED_REQUEST);
+    // Over-cap request is rejected before any pool read.
+    expect(lookups).toHaveLength(0);
+  });
+});

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

@@ -4,9 +4,17 @@ import { TxArray, TxHash, TxHashArray } from '@aztec-labs/stdlib/tx';
 import type { PeerId } from '@libp2p/interface';
 
 import type { MemPools } from '../../../mem_pools/interface.js';
+import { DEFAULT_MAX_RESPONSE_SIZE_KB } from '../../encoding.js';
 import type { ReqRespSubProtocolHandler } from '../interface.js';
 import { ReqRespStatus, ReqRespStatusError } from '../status.js';
 
+// Bound the request so the response the responder builds cannot exceed the reqresp
+// transport's max response size: each hash yields up to MAX_TX_SIZE_KB, so cap the
+// count at that budget. Honest requesters chunk at 8 (see chunkTxHashesRequest), well
+// under this; a peer naming more is rejected before the pool lookup instead of forcing
+// the node to read and serialize a response larger than the transport will carry.
+const MAX_TX_HASHES_PER_REQUEST = Math.floor(DEFAULT_MAX_RESPONSE_SIZE_KB / MAX_TX_SIZE_KB);
+
 /**
  * We want to keep the logic of the req resp handler in this file, but we do not have a reference to the mempools here
  * so we need to pass it in as a parameter.
@@ -30,9 +38,23 @@ export function reqRespTxHandler(mempools: MemPools): ReqRespSubProtocolHandler
       throw new ReqRespStatusError(ReqRespStatus.BADLY_FORMED_REQUEST, { cause: err });
     }
 
+    if (txHashes.length > MAX_TX_HASHES_PER_REQUEST) {
+      throw new ReqRespStatusError(ReqRespStatus.BADLY_FORMED_REQUEST);
+    }
+
+    // De-duplicate before serving: without this a peer can repeat one hash many
+    // times and make the node re-read and re-serialize the same tx per copy,
+    // turning a small request into a much larger response.
+    const uniqueByHash = new Map<string, TxHash>();
+    for (const txHash of txHashes) {
+      uniqueByHash.set(txHash.toString(), txHash);
+    }
+
     try {
       const txs = new TxArray(
-        ...(await Promise.all(txHashes.map(txHash => mempools.txPool.getTxByHash(txHash)))).filter(t => !!t),
+        ...(await Promise.all([...uniqueByHash.values()].map(txHash => mempools.txPool.getTxByHash(txHash)))).filter(
+          t => !!t,
+        ),
       );
       return txs.toBuffer();
     } catch (err: any) {