chore(p2p): reorganise reqresp handlers (#11327)

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

@@ -32,7 +32,7 @@ import { type AttestationPool } from '../mem_pools/attestation_pool/attestation_
 import { type EpochProofQuotePool } from '../mem_pools/epoch_proof_quote_pool/epoch_proof_quote_pool.js';
 import { type MemPools } from '../mem_pools/interface.js';
 import { type TxPool } from '../mem_pools/tx_pool/index.js';
-import { TX_REQ_PROTOCOL } from '../services/reqresp/interface.js';
+import { ReqRespSubProtocol } from '../services/reqresp/interface.js';
 import type { P2PService } from '../services/service.js';
 
 /**
@@ -459,7 +459,7 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
    * @returns A promise that resolves to a transaction or undefined.
    */
   public async requestTxByHash(txHash: TxHash): Promise<Tx | undefined> {
-    const tx = await this.p2pService.sendRequest(TX_REQ_PROTOCOL, txHash);
+    const tx = await this.p2pService.sendRequest(ReqRespSubProtocol.TX, txHash);
 
     if (tx) {
       this.log.debug(`Received tx ${txHash.toString()} from peer`);

yarn-project/p2p/src/mocks/index.ts

@@ -28,15 +28,14 @@ import { DiscV5Service } from '../services/discv5/discV5_service.js';
 import { LibP2PService } from '../services/libp2p/libp2p_service.js';
 import { type PeerManager } from '../services/peer_manager.js';
 import { type P2PReqRespConfig } from '../services/reqresp/config.js';
-import { pingHandler, statusHandler } from '../services/reqresp/handlers.js';
 import {
-  PING_PROTOCOL,
+  ReqRespSubProtocol,
   type ReqRespSubProtocolHandlers,
   type ReqRespSubProtocolValidators,
-  STATUS_PROTOCOL,
-  TX_REQ_PROTOCOL,
   noopValidator,
 } from '../services/reqresp/interface.js';
+import { pingHandler } from '../services/reqresp/protocols/ping.js';
+import { statusHandler } from '../services/reqresp/protocols/status.js';
 import { ReqResp } from '../services/reqresp/reqresp.js';
 import { type PubSubLibp2p } from '../util.js';
 
@@ -151,17 +150,17 @@ export type ReqRespNode = {
 
 // Mock sub protocol handlers
 export const MOCK_SUB_PROTOCOL_HANDLERS: ReqRespSubProtocolHandlers = {
-  [PING_PROTOCOL]: pingHandler,
-  [STATUS_PROTOCOL]: statusHandler,
-  [TX_REQ_PROTOCOL]: (_msg: any) => Promise.resolve(Buffer.from('tx')),
+  [ReqRespSubProtocol.PING]: pingHandler,
+  [ReqRespSubProtocol.STATUS]: statusHandler,
+  [ReqRespSubProtocol.TX]: (_msg: any) => Promise.resolve(Buffer.from('tx')),
 };
 
 // By default, all requests are valid
 // If you want to test an invalid response, you can override the validator
 export const MOCK_SUB_PROTOCOL_VALIDATORS: ReqRespSubProtocolValidators = {
-  [PING_PROTOCOL]: noopValidator,
-  [STATUS_PROTOCOL]: noopValidator,
-  [TX_REQ_PROTOCOL]: noopValidator,
+  [ReqRespSubProtocol.PING]: noopValidator,
+  [ReqRespSubProtocol.STATUS]: noopValidator,
+  [ReqRespSubProtocol.TX]: noopValidator,
 };
 
 /**

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

@@ -12,7 +12,7 @@ import {
   type RawGossipMessage,
   TopicTypeMap,
   Tx,
-  TxHash,
+  type TxHash,
   type TxValidationResult,
   type WorldStateSynchronizer,
   getTopicTypeForClientType,
@@ -57,17 +57,15 @@ import { type PubSubLibp2p, convertToMultiaddr } from '../../util.js';
 import { AztecDatastore } from '../data_store.js';
 import { SnappyTransform, fastMsgIdFn, getMsgIdFn, msgIdToStrFn } from '../encoding.js';
 import { PeerManager } from '../peer_manager.js';
-import { pingHandler, statusHandler } from '../reqresp/handlers.js';
 import {
   DEFAULT_SUB_PROTOCOL_HANDLERS,
   DEFAULT_SUB_PROTOCOL_VALIDATORS,
-  PING_PROTOCOL,
-  type ReqRespSubProtocol,
+  ReqRespSubProtocol,
   type ReqRespSubProtocolHandlers,
-  STATUS_PROTOCOL,
   type SubProtocolMap,
-  TX_REQ_PROTOCOL,
 } from '../reqresp/interface.js';
+import { pingHandler, statusHandler } from '../reqresp/protocols/index.js';
+import { reqRespTxHandler } from '../reqresp/protocols/tx.js';
 import { ReqResp } from '../reqresp/reqresp.js';
 import type { P2PService, PeerDiscoveryService } from '../service.js';
 import { GossipSubEvent } from '../types.js';
@@ -199,7 +197,7 @@ export class LibP2PService<T extends P2PClientType> extends WithTracer implement
     // Define the sub protocol validators - This is done within this start() method to gain a callback to the existing validateTx function
     const reqrespSubProtocolValidators = {
       ...DEFAULT_SUB_PROTOCOL_VALIDATORS,
-      [TX_REQ_PROTOCOL]: this.validateRequestedTx.bind(this),
+      [ReqRespSubProtocol.TX]: this.validateRequestedTx.bind(this),
     };
     await this.reqresp.start(this.requestResponseHandlers, reqrespSubProtocolValidators);
     this.logger.info(`Started P2P service`, {
@@ -337,22 +335,12 @@ export class LibP2PService<T extends P2PClientType> extends WithTracer implement
     });
 
     // Create request response protocol handlers
-    /**
-     * Handler for tx requests
-     * @param msg - the tx request message
-     * @returns the tx response message
-     */
-    const txHandler = (msg: Buffer): Promise<Buffer> => {
-      const txHash = TxHash.fromBuffer(msg);
-      const foundTx = mempools.txPool.getTxByHash(txHash);
-      const buf = foundTx ? foundTx.toBuffer() : Buffer.alloc(0);
-      return Promise.resolve(buf);
-    };
+    const txHandler = reqRespTxHandler(mempools);
 
     const requestResponseHandlers = {
-      [PING_PROTOCOL]: pingHandler,
-      [STATUS_PROTOCOL]: statusHandler,
-      [TX_REQ_PROTOCOL]: txHandler,
+      [ReqRespSubProtocol.PING]: pingHandler,
+      [ReqRespSubProtocol.STATUS]: statusHandler,
+      [ReqRespSubProtocol.TX]: txHandler,
     };
 
     return new LibP2PService(
@@ -548,7 +536,7 @@ export class LibP2PService<T extends P2PClientType> extends WithTracer implement
    * In order to perform this check, the tx proof must be verified.
    *
    * Note: This function is called from within `ReqResp.sendRequest` as part of the
-   * TX_REQ_PROTOCOL subprotocol validation.
+   * ReqRespSubProtocol.TX subprotocol validation.
    *
    * @param requestedTxHash - The hash of the tx that was requested.
    * @param responseTx - The tx that was received as a response to the request.

yarn-project/p2p/src/services/reqresp/interface.ts

@@ -5,12 +5,15 @@ import { type PeerId } from '@libp2p/interface';
 /*
  * Request Response Sub Protocols
  */
-export const PING_PROTOCOL = '/aztec/req/ping/0.1.0';
-export const STATUS_PROTOCOL = '/aztec/req/status/0.1.0';
-export const TX_REQ_PROTOCOL = '/aztec/req/tx/0.1.0';
-
-// Sum type for sub protocols
-export type ReqRespSubProtocol = typeof PING_PROTOCOL | typeof STATUS_PROTOCOL | typeof TX_REQ_PROTOCOL;
+const PING_PROTOCOL = '/aztec/req/ping/0.1.0';
+const STATUS_PROTOCOL = '/aztec/req/status/0.1.0';
+const TX_PROTOCOL = '/aztec/req/tx/0.1.0';
+
+export enum ReqRespSubProtocol {
+  PING = PING_PROTOCOL,
+  STATUS = STATUS_PROTOCOL,
+  TX = TX_PROTOCOL,
+}
 
 /**
  * A handler for a sub protocol
@@ -66,9 +69,9 @@ export type ReqRespSubProtocolValidators = {
 };
 
 export const DEFAULT_SUB_PROTOCOL_VALIDATORS: ReqRespSubProtocolValidators = {
-  [PING_PROTOCOL]: noopValidator,
-  [STATUS_PROTOCOL]: noopValidator,
-  [TX_REQ_PROTOCOL]: noopValidator,
+  [ReqRespSubProtocol.PING]: noopValidator,
+  [ReqRespSubProtocol.STATUS]: noopValidator,
+  [ReqRespSubProtocol.TX]: noopValidator,
 };
 
 /**
@@ -91,9 +94,9 @@ const defaultHandler = (_msg: any): Promise<Buffer> => {
  * Default sub protocol handlers - this SHOULD be overwritten by the service,
  */
 export const DEFAULT_SUB_PROTOCOL_HANDLERS: ReqRespSubProtocolHandlers = {
-  [PING_PROTOCOL]: defaultHandler,
-  [STATUS_PROTOCOL]: defaultHandler,
-  [TX_REQ_PROTOCOL]: defaultHandler,
+  [ReqRespSubProtocol.PING]: defaultHandler,
+  [ReqRespSubProtocol.STATUS]: defaultHandler,
+  [ReqRespSubProtocol.TX]: defaultHandler,
 };
 
 /**
@@ -135,15 +138,15 @@ export class RequestableBuffer {
  * as a type rather than an object
  */
 export const subProtocolMap: SubProtocolMap = {
-  [PING_PROTOCOL]: {
+  [ReqRespSubProtocol.PING]: {
     request: RequestableBuffer,
     response: RequestableBuffer,
   },
-  [STATUS_PROTOCOL]: {
+  [ReqRespSubProtocol.STATUS]: {
     request: RequestableBuffer,
     response: RequestableBuffer,
   },
-  [TX_REQ_PROTOCOL]: {
+  [ReqRespSubProtocol.TX]: {
     request: TxHash,
     response: Tx,
   },

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

@@ -0,0 +1,6 @@
+/**
+ * Request Response protocol handlers
+ */
+export * from './ping.js';
+export * from './status.js';
+export * from './tx.js';

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

@@ -0,0 +1,8 @@
+/**
+ * Handles the ping request.
+ * @param _msg - The ping request message.
+ * @returns A resolved promise with the pong response.
+ */
+export function pingHandler(_msg: any): Promise<Buffer> {
+  return Promise.resolve(Buffer.from('pong'));
+}

yarn-project/p2p/src/services/reqresp/handlers.ts → yarn-project/p2p/src/services/reqresp/protocols/status.ts

@@ -1,12 +1,3 @@
-/**
- * Handles the ping request.
- * @param _msg - The ping request message.
- * @returns A resolved promise with the pong response.
- */
-export function pingHandler(_msg: any): Promise<Buffer> {
-  return Promise.resolve(Buffer.from('pong'));
-}
-
 /**
  * Handles the status request.
  * @param _msg - The status request message.

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

@@ -0,0 +1,26 @@
+import { type P2PClientType } from '@aztec/circuit-types';
+import { TxHash } from '@aztec/circuit-types/tx_hash';
+
+import { type MemPools } from '../../../mem_pools/interface.js';
+
+/**
+ * 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.
+ *
+ * Handler for tx requests
+ * @param mempools - the mempools
+ * @returns the tx response message
+ */
+export function reqRespTxHandler<T extends P2PClientType>(mempools: MemPools<T>): (msg: Buffer) => Promise<Buffer> {
+  /**
+   * Handler for tx requests
+   * @param msg - the tx request message
+   * @returns the tx response message
+   */
+  return (msg: Buffer) => {
+    const txHash = TxHash.fromBuffer(msg);
+    const foundTx = mempools.txPool.getTxByHash(txHash);
+    const buf = foundTx ? foundTx.toBuffer() : Buffer.alloc(0);
+    return Promise.resolve(buf);
+  };
+}