refactor(p2p): remove old reqresp mode

Author: fcarreiro

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

@@ -20,7 +20,7 @@ import type { TxPoolV2 } from '../mem_pools/tx_pool_v2/interfaces.js';
 import { AztecKVTxPoolV2 } from '../mem_pools/tx_pool_v2/tx_pool_v2.js';
 import {
   createCheckAllowedSetupCalls,
-  createTxValidatorForReqResponseReceivedTxs,
+  createTxValidatorForOnDemandReceivedTxs,
   createTxValidatorForTransactionsEnteringPendingTxPool,
   getDefaultAllowedSetupFunctions,
 } from '../msg_validators/index.js';
@@ -153,7 +153,7 @@ export async function createP2PClient(
     telemetry,
   );
 
-  const txValidatorForTxCollection = createTxValidatorForReqResponseReceivedTxs(proofVerifier, config);
+  const txValidatorForTxCollection = createTxValidatorForOnDemandReceivedTxs(proofVerifier, config);
   const nodeSources = [
     ...createNodeRpcTxSources(config.txCollectionNodeRpcUrls, txValidatorForTxCollection, config),
     ...(deps.rpcTxProviders ?? []).map(

yarn-project/p2p/src/client/test/tx_proposal_collector/p2p_client.proposal_tx_collector.bench.test.ts

@@ -6,7 +6,6 @@ import { mkdir, writeFile } from 'fs/promises';
 import path from 'path';
 
 import {
-  type CollectorType,
   type DistributionPattern,
   WorkerClientManager,
   testChainConfig,
@@ -15,8 +14,6 @@ import {
 const TEST_TIMEOUT_MS = 600_000; // 10 minutes
 jest.setTimeout(TEST_TIMEOUT_MS);
 
-const COLLECTOR_TYPES: CollectorType[] = ['batch-requester', 'send-batch-request'];
-
 const PEERS_PER_RUN = 30;
 const TIMEOUT_MS = 30_000;
 
@@ -39,7 +36,6 @@ interface BenchmarkCase extends ScenarioBase {
 interface BenchmarkResult {
   missingTxCount: number;
   distribution: DistributionPattern;
-  collector: CollectorType;
   durationMs: number;
   fetchedCount: number;
   success: boolean;
@@ -128,7 +124,7 @@ describe('ProposalTxCollector Benchmarks', () => {
   });
 
   describe.each(CASES)('$name (missing=$missingTxCount)', benchCase => {
-    it.each(COLLECTOR_TYPES)('collector: %s', async collectorType => {
+    it('runs batch tx requester benchmark', async () => {
       if (!workerManager) {
         throw new Error('Worker manager not initialized');
       }
@@ -137,15 +133,12 @@ describe('ProposalTxCollector Benchmarks', () => {
       const pinnedPeerIndex = distribution === 'pinned-only' ? benchCase.pinnedPeerIndex : undefined;
       const seed = blockNumber * 1_000_000;
 
-      logger.info(
-        `Case=${benchCase.name}, missing=${missingTxCount}, collector=${collectorType}, peers=${peers}, timeoutMs=${timeoutMs}`,
-      );
+      logger.info(`Case=${benchCase.name}, missing=${missingTxCount}, peers=${peers}, timeoutMs=${timeoutMs}`);
 
       try {
         const result = await workerManager.runReqRespBenchmark({
           txCount: missingTxCount,
           distribution,
-          collectorType,
           timeoutMs,
           pinnedPeerIndex,
           blockNumber,
@@ -155,22 +148,18 @@ describe('ProposalTxCollector Benchmarks', () => {
         results.push({
           missingTxCount,
           distribution,
-          collector: collectorType,
           durationMs: result.durationMs,
           fetchedCount: result.fetchedCount,
           success: result.fetchedCount === missingTxCount,
         });
 
-        logger.info(
-          `${collectorType}: fetched ${result.fetchedCount}/${missingTxCount} in ${result.durationMs.toFixed(0)}ms`,
-        );
+        logger.info(`fetched ${result.fetchedCount}/${missingTxCount} in ${result.durationMs.toFixed(0)}ms`);
       } catch (err: any) {
-        logger.error(`${collectorType} failed: ${err?.message ?? String(err)}`);
+        logger.error(`Benchmark failed: ${err?.message ?? String(err)}`);
 
         results.push({
           missingTxCount,
           distribution,
-          collector: collectorType,
           durationMs: timeoutMs,
           fetchedCount: 0,
           success: false,
@@ -195,64 +184,23 @@ function toPrettyString(benchResults: BenchmarkResult[]): string {
   lines.push('ProposalTxCollector Benchmark Results');
   lines.push('='.repeat(80));
   lines.push('');
-  lines.push('| Collector           | Distribution | Missing | Duration (ms) | Fetched | Success |');
-  lines.push('|---------------------|--------------|---------|---------------|---------|---------|');
+  lines.push('| Distribution | Missing | Duration (ms) | Fetched | Success |');
+  lines.push('|--------------|---------|---------------|---------|---------|');
 
   const sorted = [...benchResults].sort((a, b) => {
     if (a.distribution !== b.distribution) {
       return a.distribution.localeCompare(b.distribution);
     }
-    if (a.missingTxCount !== b.missingTxCount) {
-      return a.missingTxCount - b.missingTxCount;
-    }
-    return a.collector.localeCompare(b.collector);
+    return a.missingTxCount - b.missingTxCount;
   });
 
   for (const r of sorted) {
     lines.push(
-      `| ${r.collector.padEnd(19)} | ${r.distribution.padEnd(12)} | ${String(r.missingTxCount).padStart(7)} | ` +
+      `| ${r.distribution.padEnd(12)} | ${String(r.missingTxCount).padStart(7)} | ` +
         `${r.durationMs.toFixed(0).padStart(13)} | ${String(r.fetchedCount).padStart(7)} | ${r.success ? '  Yes  ' : '  No   '} |`,
     );
   }
 
-  lines.push('');
-  lines.push('## Comparison Summary');
-  lines.push('');
-
-  const keys = [...new Set(sorted.map(r => `${r.distribution}:${r.missingTxCount}`))];
-
-  for (const key of keys) {
-    const [distRaw, missingRaw] = key.split(':');
-    const dist = distRaw as DistributionPattern;
-    const missing = Number(missingRaw);
-
-    const batch = sorted.find(
-      r => r.distribution === dist && r.missingTxCount === missing && r.collector === 'batch-requester',
-    );
-    const send = sorted.find(
-      r => r.distribution === dist && r.missingTxCount === missing && r.collector === 'send-batch-request',
-    );
-
-    if (!batch || !send) {
-      continue;
-    }
-
-    if (!batch.success || !send.success) {
-      lines.push(
-        `- ${dist} (missing=${missing}): cannot compare reliably (success: batch=${batch.success}, send=${send.success})`,
-      );
-      continue;
-    }
-
-    const faster = batch.durationMs <= send.durationMs ? 'BatchTxRequester' : 'SendBatchRequest';
-    const slower = faster === 'BatchTxRequester' ? 'SendBatchRequest' : 'BatchTxRequester';
-
-    const delta = Math.abs(send.durationMs - batch.durationMs);
-    const pct = (delta / Math.max(batch.durationMs, send.durationMs)) * 100;
-
-    lines.push(`- ${dist} (missing=${missing}): ${faster} is ${pct.toFixed(1)}% faster than ${slower}`);
-  }
-
   lines.push('');
   lines.push('='.repeat(80));
   lines.push('');
@@ -264,7 +212,7 @@ function toBenchmarkJSON(benchResults: BenchmarkResult[], indent = 2): string {
   const metrics: JsonBenchmarkResult[] = [];
 
   for (const result of benchResults) {
-    const baseName = `ProposalTxCollector/${result.collector}/${result.distribution}/missing_${result.missingTxCount}`;
+    const baseName = `ProposalTxCollector/${result.distribution}/missing_${result.missingTxCount}`;
     metrics.push(
       {
         name: `${baseName}/duration`,

yarn-project/p2p/src/client/test/tx_proposal_collector/proposal_tx_collector_worker.ts

@@ -17,7 +17,7 @@ import type { PeerId } from '@libp2p/interface';
 import { peerIdFromString } from '@libp2p/peer-id';
 
 import type { P2PConfig } from '../../../config.js';
-import { BatchTxRequesterCollector, SendBatchRequestCollector } from '../../../services/index.js';
+import { BatchTxRequester } from '../../../services/reqresp/batch-tx-requester/batch_tx_requester.js';
 import type { IBatchRequestTxValidator } from '../../../services/reqresp/batch-tx-requester/tx_validator.js';
 import { RateLimitStatus } from '../../../services/reqresp/rate-limiter/rate_limiter.js';
 import { RequestTracker } from '../../../services/tx_collection/request_tracker.js';
@@ -170,7 +170,7 @@ function installUnlimitedRateLimits() {
 }
 
 async function runCollector(cmd: Extract<WorkerCommand, { type: 'RUN_COLLECTOR' }>) {
-  const { collectorType, txHashes, blockProposal, pinnedPeerId, peerIds, timeoutMs } = cmd;
+  const { txHashes, blockProposal, pinnedPeerId, peerIds, timeoutMs } = cmd;
   const reqResp = (ensureClient() as any).p2pService.reqresp as any;
   const peerList = peerIds.map(peerId => peerIdFromString(peerId));
 
@@ -211,37 +211,24 @@ async function runCollector(cmd: Extract<WorkerCommand, { type: 'RUN_COLLECTOR'
   };
 
   try {
-    if (collectorType === 'batch-requester') {
-      const collector = new BatchTxRequesterCollector(p2pService, logger, new DateProvider(), noopTxValidator);
-      const fetched = await executeTimeout(
-        (_signal: AbortSignal) =>
-          collector.collectTxs(
-            RequestTracker.create(parsedTxHashes, new Date(Date.now() + internalTimeoutMs)),
-            parsedProposal,
-            pinnedPeer,
-          ),
-        timeoutMs,
-        () => new Error(`Collector timed out after ${timeoutMs}ms`),
-      );
-      fetchedCount = fetched.length;
-    } else {
-      const collector = new SendBatchRequestCollector(
-        p2pService,
-        BENCHMARK_CONSTANTS.FIXED_MAX_PEERS,
-        BENCHMARK_CONSTANTS.FIXED_MAX_RETRY_ATTEMPTS,
-      );
-      const fetched = await executeTimeout(
-        (_signal: AbortSignal) =>
-          collector.collectTxs(
-            RequestTracker.create(parsedTxHashes, new Date(Date.now() + internalTimeoutMs)),
-            parsedProposal,
-            pinnedPeer,
-          ),
-        timeoutMs,
-        () => new Error(`Collector timed out after ${timeoutMs}ms`),
-      );
-      fetchedCount = fetched.length;
-    }
+    const fetched = await executeTimeout(
+      (_signal: AbortSignal) => {
+        const tracker = RequestTracker.create(parsedTxHashes, new Date(Date.now() + internalTimeoutMs));
+        const batchRequester = new BatchTxRequester(
+          tracker,
+          parsedProposal,
+          pinnedPeer,
+          p2pService,
+          logger,
+          new DateProvider(),
+          { txValidator: noopTxValidator },
+        );
+        return BatchTxRequester.collectAllTxs(batchRequester.run());
+      },
+      timeoutMs,
+      () => new Error(`Collector timed out after ${timeoutMs}ms`),
+    );
+    fetchedCount = fetched.length;
   } catch (err: any) {
     logger.warn(`Collector error: ${err?.message ?? String(err)}`);
   }

yarn-project/p2p/src/client/test/tx_proposal_collector/proposal_tx_collector_worker_protocol.ts

@@ -5,16 +5,13 @@ import type { P2PConfig } from '../../../config.js';
 
 export type SerializedP2PConfig = Omit<P2PConfig, 'peerIdPrivateKey'> & { peerIdPrivateKey?: string };
 
-export type CollectorType = 'batch-requester' | 'send-batch-request';
-
 export type WorkerCommand =
   | { type: 'START'; requestId: string; clientIndex: number; config: SerializedP2PConfig }
   | { type: 'SET_TXS'; requestId: string; txs: string[]; mode?: 'replace' | 'append' }
   | { type: 'SET_BLOCK_PROPOSAL'; requestId: string; blockProposal: string }
   | {
       type: 'RUN_COLLECTOR';
       requestId: string;
-      collectorType: CollectorType;
       txHashes: string[];
       blockProposal: string;
       pinnedPeerId?: string;

yarn-project/p2p/src/config.test.ts

@@ -79,11 +79,6 @@ describe('config', () => {
     expect(allowList).toEqual([instanceFunction]);
   });
 
-  it('defaults missing txs collector type to new', () => {
-    const config = getP2PDefaultConfig();
-    expect(config.txCollectionMissingTxsCollectorType).toBe('new');
-  });
-
   it('defaults public IP service URLs', () => {
     const config = getP2PDefaultConfig();
     expect(config.publicIpServices).toEqual([

yarn-project/p2p/src/msg_validators/tx_validator/factory.test.ts

@@ -23,7 +23,7 @@ import {
   createTxValidatorForAcceptingTxsOverRPC,
   createTxValidatorForBlockBuilding,
   createTxValidatorForBlockProposalReceivedTxs,
-  createTxValidatorForReqResponseReceivedTxs,
+  createTxValidatorForOnDemandReceivedTxs,
   createTxValidatorForTransactionsEnteringPendingTxPool,
 } from './factory.js';
 import { GasLimitsValidator, GasTxValidator, MaxFeePerGasValidator } from './gas_validator.js';
@@ -160,9 +160,9 @@ describe('Validator factory functions', () => {
     });
   });
 
-  describe('createTxValidatorForReqResponseReceivedTxs', () => {
+  describe('createTxValidatorForOnDemandReceivedTxs', () => {
     it('contains well-formedness validators only', () => {
-      const validator = createTxValidatorForReqResponseReceivedTxs(proofVerifier, {
+      const validator = createTxValidatorForOnDemandReceivedTxs(proofVerifier, {
         l1ChainId: 1,
         rollupVersion: 2,
       });

yarn-project/p2p/src/msg_validators/tx_validator/factory.ts

@@ -234,7 +234,7 @@ function createTxValidatorForMinimumTxIntegrityChecks(
  * are invalid against current state. State-dependent checks happen when the tx
  * enters the pending pool or during block building.
  */
-export function createTxValidatorForReqResponseReceivedTxs(
+export function createTxValidatorForOnDemandReceivedTxs(
   verifier: ClientProtocolCircuitVerifier,
   {
     l1ChainId,

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

@@ -74,7 +74,7 @@ import {
   createFirstStageTxValidationsForGossipedTransactions,
   createSecondStageTxValidationsForGossipedTransactions,
   createTxValidatorForBlockProposalReceivedTxs,
-  createTxValidatorForReqResponseReceivedTxs,
+  createTxValidatorForOnDemandReceivedTxs,
 } from '../../msg_validators/tx_validator/factory.js';
 import { GossipSubEvent } from '../../types/index.js';
 import { type PubSubLibp2p, convertToMultiaddr } from '../../util.js';
@@ -1680,7 +1680,7 @@ export class LibP2PService extends WithTracer implements P2PService {
   }
 
   protected createRequestedTxValidator(): TxValidator {
-    return createTxValidatorForReqResponseReceivedTxs(this.proofVerifier, {
+    return createTxValidatorForOnDemandReceivedTxs(this.proofVerifier, {
       l1ChainId: this.config.l1ChainId,
       rollupVersion: this.config.rollupVersion,
     });

yarn-project/p2p/src/services/reqresp/batch-tx-requester/interface.ts

@@ -32,7 +32,7 @@ export interface ITxMetadataCollection {
  */
 export interface BatchTxRequesterLibP2PService {
   /** ReqResp interface for sending requests to peers */
-  reqResp: Pick<ReqRespInterface, 'sendBatchRequest' | 'sendRequestToPeer'>;
+  reqResp: Pick<ReqRespInterface, 'sendRequestToPeer'>;
   /** Connection sampler for getting peer lists */
   connectionSampler: Pick<ConnectionSampler, 'getPeerListSortedByConnectionCountAsc'>;
   /** Configuration needed for transaction validation */

yarn-project/p2p/src/services/reqresp/batch-tx-requester/tx_validator.ts

@@ -1,7 +1,7 @@
 import type { ClientProtocolCircuitVerifier } from '@aztec/stdlib/interfaces/server';
 import { Tx, type TxValidationResult, type TxValidator } from '@aztec/stdlib/tx';
 
-import { createTxValidatorForReqResponseReceivedTxs } from '../../../msg_validators/index.js';
+import { createTxValidatorForOnDemandReceivedTxs } from '../../../msg_validators/index.js';
 
 export interface BatchRequestTxValidatorConfig {
   l1ChainId: number;
@@ -29,7 +29,7 @@ export class BatchRequestTxValidator implements IBatchRequestTxValidator {
   }
 
   static createRequestedTxValidator(config: BatchRequestTxValidatorConfig): TxValidator {
-    return createTxValidatorForReqResponseReceivedTxs(config.proofVerifier, {
+    return createTxValidatorForOnDemandReceivedTxs(config.proofVerifier, {
       l1ChainId: config.l1ChainId,
       rollupVersion: config.rollupVersion,
     });