feat(p2p): request response skeleton

yarn-project/p2p/src/service/reqresp/handlers.ts

@@ -0,0 +1,7 @@
+export function pingHandler(msg: any) {
+  return Uint8Array.from(Buffer.from('pong'));
+}
+
+export function statusHandler(msg: any) {
+  return Uint8Array.from(Buffer.from('ok'));
+}

yarn-project/p2p/src/service/reqresp/index.ts

@@ -0,0 +1,4 @@
+/**
+ * Request Response protocol allows nodes to ask their peers for data
+ * that they missed via the traditional gossip protocol.
+ */

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

@@ -0,0 +1,9 @@
+export enum ReqRespType {
+  Status = 'status',
+  Ping = 'ping',
+  /** Ask peers for specific transactions */
+  TxsByHash = 'txs_by_hash',
+}
+
+export const PING_PROTOCOL: string = '/aztec/ping/0.1.0';
+export const STATUS_PROTOCOL: string = '/aztec/status/0.1.0';

yarn-project/p2p/src/service/reqresp/reqresp.test.ts

@@ -0,0 +1,155 @@
+import { sleep } from '@aztec/foundation/sleep';
+
+import { noise } from '@chainsafe/libp2p-noise';
+import { yamux } from '@chainsafe/libp2p-yamux';
+import { bootstrap } from '@libp2p/bootstrap';
+import { tcp } from '@libp2p/tcp';
+import { type Libp2p, type Libp2pOptions, createLibp2p } from 'libp2p';
+
+import { PING_PROTOCOL } from './interface.js';
+import { ReqResp } from './reqresp.js';
+
+/**
+ * Creates a libp2p node, pre configured.
+ * @param boostrapAddrs - an optional list of bootstrap addresses
+ * @returns Lip2p node
+ */
+async function createLibp2pNode(boostrapAddrs: string[] = []): Promise<Libp2p> {
+  const options: Libp2pOptions = {
+    addresses: {
+      listen: ['/ip4/0.0.0.0/tcp/0'],
+    },
+    connectionEncryption: [noise()],
+    streamMuxers: [yamux()],
+    transports: [tcp()],
+  };
+
+  if (boostrapAddrs.length > 0) {
+    options.peerDiscovery = [
+      bootstrap({
+        list: boostrapAddrs,
+      }),
+    ];
+  }
+
+  return await createLibp2p(options);
+}
+
+/**
+ * A p2p / req resp node pairing the req node will always contain the p2p node.
+ * they are provided as a pair to allow access the p2p node directly
+ */
+type ReqRespNode = {
+  p2p: Libp2p;
+  req: ReqResp;
+};
+
+/**
+ * @param numberOfNodes - the number of nodes to create
+ * @returns An array of the created nodes
+ */
+const createNodes = async (numberOfNodes: number): Promise<ReqRespNode[]> => {
+  return await Promise.all(Array.from({ length: numberOfNodes }, () => createReqResp()));
+};
+
+const startNodes = async (nodes: ReqRespNode[]): Promise<void> => {
+  for (const node of nodes) {
+    node.req.start();
+  }
+};
+
+const stopNodes = async (nodes: ReqRespNode[]): Promise<void> => {
+  for (const node of nodes) {
+    await node.req.stop();
+    await node.p2p.stop();
+  }
+};
+
+// Create a req resp node, exposing the underlying p2p node
+const createReqResp = async (): Promise<ReqRespNode> => {
+  const p2p = await createLibp2pNode();
+  const req = new ReqResp(p2p);
+  return {
+    p2p,
+    req,
+  };
+};
+
+// Given a node list; hand shake all of the nodes with each other
+const connectToPeers = async (nodes: ReqRespNode[]): Promise<void> => {
+  for (const node of nodes) {
+    for (const otherNode of nodes) {
+      if (node === otherNode) {
+        continue;
+      }
+      const addr = otherNode.p2p.getMultiaddrs()[0];
+      await node.p2p.dial(addr);
+    }
+  }
+};
+
+// The Req Resp protocol should allow nodes to dial specific peers
+// and ask for specific data that they missed via the traditional gossip protocol.
+describe('ReqResp', () => {
+  it('Should perform a ping request', async () => {
+    // Create two nodes
+    // They need to discover each other
+    const nodes = await createNodes(2);
+    const { req: pinger } = nodes[0];
+
+    startNodes(nodes);
+
+    // connect the nodes
+    await connectToPeers(nodes);
+
+    await sleep(500);
+
+    const res = await pinger.sendRequest(PING_PROTOCOL, Buffer.from('ping'));
+
+    await sleep(500);
+    expect(res?.toString('utf-8')).toEqual('pong');
+
+    await stopNodes(nodes);
+  });
+
+  it('Should handle gracefully if a peer connected peer is offline', async () => {
+    const nodes = await createNodes(2);
+
+    const { req: pinger } = nodes[0];
+    const { req: ponger } = nodes[1];
+    startNodes(nodes);
+
+    // connect the nodes
+    await connectToPeers(nodes);
+    await sleep(500);
+
+    ponger.stop();
+
+    // It should return undefined if it cannot dial the peer
+    const res = await pinger.sendRequest(PING_PROTOCOL, Buffer.from('ping'));
+
+    expect(res).toBeUndefined();
+
+    await stopNodes(nodes);
+  });
+
+  it('Should request from a later peer if other peers are offline', async () => {
+    const nodes = await createNodes(4);
+
+    await startNodes(nodes);
+    await sleep(500);
+    await connectToPeers(nodes);
+    await sleep(500);
+
+    // Stop the second middle two nodes
+    nodes[1].req.stop();
+    nodes[2].req.stop();
+
+    // send from the first node
+    const res = await nodes[0].req.sendRequest(PING_PROTOCOL, Buffer.from('ping'));
+
+    expect(res?.toString('utf-8')).toEqual('pong');
+
+    await stopNodes(nodes);
+  });
+});

yarn-project/p2p/src/service/reqresp/reqresp.ts

@@ -0,0 +1,143 @@
+// @attribution: lodestar impl for inspiration
+import { type Logger, createDebugLogger } from '@aztec/foundation/log';
+
+import { type IncomingStreamData, type PeerId } from '@libp2p/interface';
+import { pipe } from 'it-pipe';
+import { type Libp2p } from 'libp2p';
+import { type Uint8ArrayList } from 'uint8arraylist';
+
+import { pingHandler, statusHandler } from './handlers.js';
+import { PING_PROTOCOL, STATUS_PROTOCOL } from './interface.js';
+
+/**
+ * A mapping from a protocol to a handler function
+ */
+const REQ_RESP_PROTOCOLS = {
+  [PING_PROTOCOL]: pingHandler,
+  [STATUS_PROTOCOL]: statusHandler,
+};
+
+export class ReqResp {
+  protected readonly logger: Logger;
+
+  private abortController: AbortController = new AbortController();
+
+  constructor(protected readonly libp2p: Libp2p) {
+    this.logger = createDebugLogger('aztec:p2p:reqresp');
+  }
+
+  /**
+   * Start the reqresp service
+   */
+  async start() {
+    // Register all protocol handlers
+    for (const [protocol, _] of Object.entries(REQ_RESP_PROTOCOLS)) {
+      await this.libp2p.handle(protocol, streamHandler.bind(this));
+    }
+  }
+
+  /**
+   * Stop the reqresp service
+   */
+  async stop() {
+    // Unregister all handlers
+    for (const [protocol, _] of Object.keys(REQ_RESP_PROTOCOLS)) {
+      this.libp2p.unhandle(protocol);
+    }
+    await this.libp2p.stop();
+    this.abortController.abort();
+  }
+
+  /**
+   * Send a request to peers, returns the first response
+   *
+   * @param protocol - The protocol being requested
+   * @param payload - The payload to send
+   * @returns - The response from the peer, otherwise undefined
+   */
+  async sendRequest(protocol: string, payload: Buffer): Promise<Buffer | undefined> {
+    // Get active peers
+    const peers = this.libp2p.getPeers();
+
+    // Attempt to ask all of our peers
+    for (const peer of peers) {
+      const response = await this.sendRequestToPeer(peer, protocol, payload);
+
+      // If we get a response, return it, otherwise we iterate onto the next peer
+      if (response) {
+        return response;
+      }
+    }
+    return undefined;
+  }
+
+  /**
+   * Sends a request to a specific peer
+   *
+   * @param peerId - The peer to send the request to
+   * @param protocol - The protocol to use to request
+   * @param payload - The payload to send
+   * @returns If the request is successful, the response is returned, otherwise undefined
+   */
+  async sendRequestToPeer(peerId: PeerId, protocol: string, payload: Buffer): Promise<Buffer | undefined> {
+    try {
+      const stream = await this.libp2p.dialProtocol(peerId, protocol);
+
+      const result = await pipe(
+        // Send message in two chunks - protocol && payload
+        [Buffer.from(protocol), Buffer.from(payload)],
+        stream,
+        this.readMessage,
+      );
+      return result;
+    } catch (e) {
+      this.logger.warn(`Failed to send request to peer ${peerId.publicKey}`);
+      return undefined;
+    }
+  }
+
+  /**
+   * Read a message returned from a stream into a single buffer
+   */
+  private async readMessage(source: AsyncIterable<Uint8ArrayList>): Promise<Buffer> {
+    const chunks: Uint8Array[] = [];
+    for await (const chunk of source) {
+      chunks.push(chunk.subarray());
+    }
+    const messageData = chunks.concat();
+    return Buffer.concat(messageData);
+  }
+}
+
+/**
+ * Stream Handler
+ * Reads the incoming stream, determines the protocol, then triggers the appropriate handler
+ *
+ * @param param0 - The incoming stream data
+ */
+async function streamHandler({ stream }: IncomingStreamData) {
+  try {
+    await pipe(
+      stream,
+      async function* (source: any) {
+        let protocol: string | undefined = undefined;
+        for await (const chunk of source) {
+          // The first message should contain the protocol, subsequent messages should contain the payload
+
+          const msg = Buffer.from(chunk.subarray()).toString();
+          if (!protocol) {
+            protocol = msg.toString();
+          } else {
+            const handler: any = REQ_RESP_PROTOCOLS[protocol];
+            yield handler(msg);
+          }
+        }
+      },
+      stream,
+    );
+  } catch (e: any) {
+    console.log(e);
+  } finally {
+    await stream.close();
+  }
+}