feat(p2p): deduplicate on-demand TX validation

.test_patterns.yml

@@ -142,6 +142,12 @@ tests:
     owners:
       - *palla
 
+  # Flakes under pipelining.
+  - regex: "src/e2e_expiration_timestamp.test.ts"
+    error_regex: "✕ invalidates the transaction"
+    owners:
+      - *palla
+
   # yarn-project tests
   # Attempt to catch all kv-store browser test failures (consider them quarantined for now)
   - regex: "yarn-project/kv-store"

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

@@ -277,6 +277,7 @@ export type EnvVar =
   | 'TX_COLLECTION_FILE_STORE_FAST_WORKER_COUNT'
   | 'TX_COLLECTION_FILE_STORE_FAST_BACKOFF_BASE_MS'
   | 'TX_COLLECTION_FILE_STORE_FAST_BACKOFF_MAX_MS'
+  | 'TX_VALIDATION_CACHE_SIZE'
   | 'TX_FILE_STORE_URL'
   | 'TX_FILE_STORE_UPLOAD_CONCURRENCY'
   | 'TX_FILE_STORE_MAX_QUEUE_SIZE'

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

@@ -27,6 +27,7 @@ import {
 import { DummyP2PService } from '../services/dummy_service.js';
 import { LibP2PService } from '../services/index.js';
 import { createFileStoreTxSources } from '../services/tx_collection/file_store_tx_source.js';
+import { SharedTxValidationCache } from '../services/tx_collection/shared_tx_validation_cache.js';
 import { TxCollection } from '../services/tx_collection/tx_collection.js';
 import { NodeRpcTxSource, type TxSource, createNodeRpcTxSources } from '../services/tx_collection/tx_source.js';
 import { TxFileStore } from '../services/tx_file_store/tx_file_store.js';
@@ -154,10 +155,15 @@ export async function createP2PClient(
   );
 
   const txValidatorForTxCollection = createTxValidatorForOnDemandReceivedTxs(proofVerifier, config);
+  const txValidationCache = new SharedTxValidationCache(
+    txValidatorForTxCollection,
+    config.txValidationCacheSize,
+    logger.createChild('shared-tx-validation-cache'),
+  );
   const nodeSources = [
-    ...createNodeRpcTxSources(config.txCollectionNodeRpcUrls, txValidatorForTxCollection, config),
+    ...createNodeRpcTxSources(config.txCollectionNodeRpcUrls, txValidationCache, config),
     ...(deps.rpcTxProviders ?? []).map(
-      (node, i) => new NodeRpcTxSource(node, txValidatorForTxCollection, `node-rpc-provider-${i}`),
+      (node, i) => new NodeRpcTxSource(node, txValidationCache, `node-rpc-provider-${i}`),
     ),
     ...(deps.txCollectionNodeSources ?? []),
   ];
@@ -170,7 +176,7 @@ export async function createP2PClient(
   const fileStoreSources = await createFileStoreTxSources(
     config.txCollectionFileStoreUrls,
     txFileStoreBasePath,
-    txValidatorForTxCollection,
+    txValidationCache,
     logger.createChild('file-store-tx-source'),
     telemetry,
   );
@@ -187,6 +193,7 @@ export async function createP2PClient(
     mempools.txPool,
     config,
     fileStoreSources,
+    txValidationCache,
     dateProvider,
     telemetry,
     logger.createChild('tx-collection'),

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

@@ -9,7 +9,7 @@ import { sleep } from '@aztec/foundation/sleep';
 import { emptyChainConfig } from '@aztec/stdlib/config';
 import type { WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
 import { makeBlockHeader, makeBlockProposal, mockTx } from '@aztec/stdlib/testing';
-import { Tx, TxHash, type TxValidator } from '@aztec/stdlib/tx';
+import { Tx, TxHash } from '@aztec/stdlib/tx';
 
 import { describe, expect, it, jest } from '@jest/globals';
 import { type MockProxy, mock } from 'jest-mock-extended';
@@ -22,6 +22,7 @@ import { BatchTxRequester } from '../../services/reqresp/batch-tx-requester/batc
 import type { BatchTxRequesterLibP2PService } from '../../services/reqresp/batch-tx-requester/interface.js';
 import type { ConnectionSampler } from '../../services/reqresp/connection-sampler/connection_sampler.js';
 import { RequestTracker } from '../../services/tx_collection/request_tracker.js';
+import type { ISharedTxValidationCache } from '../../services/tx_collection/shared_tx_validation_cache.js';
 import { generatePeerIdPrivateKeys } from '../../test-helpers/generate-peer-id-private-keys.js';
 import { getPorts } from '../../test-helpers/get-ports.js';
 import { makeEnrs } from '../../test-helpers/make-enrs.js';
@@ -38,7 +39,7 @@ describe('p2p client integration batch txs', () => {
 
   let mockP2PService: MockProxy<BatchTxRequesterLibP2PService>;
   let connectionSampler: MockProxy<ConnectionSampler>;
-  let txValidator: TxValidator;
+  let validationCache: MockProxy<ISharedTxValidationCache>;
 
   let logger: Logger;
   let p2pBaseConfig: P2PConfig;
@@ -56,11 +57,11 @@ describe('p2p client integration batch txs', () => {
       connectionSampler,
       validateRequestedBlockTxsConsistency: () => Promise.resolve(true),
     });
-    txValidator = {
-      validateTx: () => Promise.resolve({ result: 'valid' }),
-    };
-
     logger = createLogger('p2p:test:integration:batch');
+    validationCache = mock<ISharedTxValidationCache>();
+    validationCache.submitBatch.mockImplementation((txs: Tx[]) =>
+      Promise.resolve(txs.map(() => ({ status: 'accepted' as const }))),
+    );
     p2pBaseConfig = { ...emptyChainConfig, ...getP2PDefaultConfig() };
 
     //@ts-expect-error - we want to mock the getEpochAndSlotInNextL1Slot method, mocking ts is enough
@@ -236,9 +237,10 @@ describe('p2p client integration batch txs', () => {
       blockProposal,
       undefined, // no pinned peer
       mockP2PService,
+      validationCache,
       logger,
       undefined,
-      { smartParallelWorkerCount: 10, dumbParallelWorkerCount: 10, txValidator },
+      { smartParallelWorkerCount: 10, dumbParallelWorkerCount: 10 },
     );
 
     const fetchedTxs = await BatchTxRequester.collectAllTxs(requester.run());