Added skeleton implementation

Author: mralj

yarn-project/p2p/src/services/reqresp/connection-sampler/connection_sampler.ts

@@ -129,6 +129,22 @@ export class ConnectionSampler {
     return { peer: lastPeer, sampledPeers };
   }
 
+  /*
+   * Returns all peers sorted by connection count ascending,
+   * meaning that the peers with the least number of active connections are earlier in an array
+   *
+   * @param: excluding - peers to exclude
+   * @return: list of peer ids
+   * */
+  public getPeerListSortedByConnectionCountAsc(excluding?: Set<string>): PeerId[] {
+    return this.libp2p
+      .getPeers()
+      .filter(id => !excluding?.has(id.toString()))
+      .map(id => ({ id, count: this.activeConnectionsCount.get(id) ?? 0 }))
+      .sort((a, b) => a.count - b.count)
+      .map(p => p.id);
+  }
+
   /**
    * Samples a batch of unique peers from the libp2p node, prioritizing peers without active connections
    *

yarn-project/p2p/src/services/reqresp/protocols/block_txs/bitvector.ts

@@ -62,6 +62,13 @@ export class BitVector {
     return Array.from({ length: this.length }, (_, i) => i).filter(i => this.isSet(i));
   }
 
+  /**
+   * Returns true if no indices are set to true
+   * */
+  isEmpty(): boolean {
+    return this.getTrueIndices().length === 0;
+  }
+
   /**
    * Serializes the BitVector object into a Buffer
    *

yarn-project/p2p/src/services/reqresp/protocols/block_txs/block_txs_reqresp.ts

@@ -1,6 +1,7 @@
 import { Fr } from '@aztec/foundation/fields';
 import { BufferReader, serializeToBuffer } from '@aztec/foundation/serialize';
-import { TxArray } from '@aztec/stdlib/tx';
+import type { BlockProposal } from '@aztec/stdlib/p2p';
+import { TxArray, TxHash } from '@aztec/stdlib/tx';
 
 import { BitVector } from './bitvector.js';
 
@@ -15,6 +16,36 @@ export class BlockTxsRequest {
     readonly txIndices: BitVector,
   ) {}
 
+  /**
+   * Crates new BlockTxsRequest given proposal and missing tx hashes
+   *
+   * @param: blockProposal - The block proposal for which we are making request
+   * @param: missingTxHashes - Tx hashes from the proposal we are missing
+   *
+   * @returns undefined if there were no missingTxHashes matching BlockProposal hashes, otherwise
+   * returns new BlockTxsRequest*/
+  static fromBlockProposalAndMissingTxs(
+    blockProposal: BlockProposal,
+    missingTxHashes: TxHash[],
+  ): BlockTxsRequest | undefined {
+    if (missingTxHashes.length === 0) {
+      return undefined; // No missing txs to request
+    }
+
+    const missingHashesSet = new Set(missingTxHashes);
+    const missingIndices = blockProposal.txHashes
+      .map((hash, idx) => (missingHashesSet.has(hash) ? idx : -1))
+      .filter(i => i != -1);
+
+    if (missingIndices.length === 0) {
+      return undefined; // No indices found for missing tx hashes
+    }
+
+    const requestBitVector = BitVector.init(blockProposal.txHashes.length, missingIndices);
+
+    return new BlockTxsRequest(blockProposal.archive, requestBitVector);
+  }
+
   /**
    * Deserializes the BlockTxRequest object from a Buffer
    * @param buffer - Buffer or BufferReader object to deserialize

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

@@ -0,0 +1,228 @@
+import { chunk } from '@aztec/foundation/collection';
+import { createLogger } from '@aztec/foundation/log';
+import { promiseWithResolvers } from '@aztec/foundation/promise';
+import type { BlockProposal } from '@aztec/stdlib/p2p';
+import type { TxHash } from '@aztec/stdlib/tx';
+
+import type { PeerId } from '@libp2p/interface';
+import { peerIdFromString } from '@libp2p/peer-id';
+
+import type { ConnectionSampler } from './connection-sampler/connection_sampler.js';
+import { type ReqRespInterface, ReqRespSubProtocol } from './interface.js';
+import { BlockTxsRequest, BlockTxsResponse } from './protocols/index.js';
+import { ReqRespStatus } from './status.js';
+
+const TX_BATCH_SIZE = 8;
+const PEERS_TO_QUERY_IN_PARALLEL = 10;
+
+class MissingTxMetadata {
+  constructor(
+    public readonly txHash: TxHash,
+    public fetched = false,
+    public inFlight = false,
+    public requestedTimes = 0,
+    public readonly peers = new Set<string>(),
+  ) {}
+}
+
+export class BatchTxRequester {
+  private readonly peers: PeerId[];
+  private readonly smartPeers = new Set<string>();
+  private readonly peersToTxMap = new Map<string, Array<TxHash>>();
+
+  private readonly txsMetadata;
+
+  private startSmartRequester: ((v: void) => void) | undefined = undefined;
+
+  constructor(
+    private readonly blockProposal: BlockProposal,
+    private readonly missingTxs: TxHash[],
+    private readonly pinnedPeer: PeerId | undefined,
+    private readonly reqresp: ReqRespInterface,
+    private readonly connectionSampler: ConnectionSampler,
+    private logger = createLogger('p2p:reqresp_batch'),
+  ) {
+    this.txsMetadata = new Map(this.missingTxs.map(txHash => [txHash.toString(), new MissingTxMetadata(txHash)]));
+    this.peers = this.connectionSampler.getPeerListSortedByConnectionCountAsc();
+    if (this.pinnedPeer) {
+      this.smartPeers.add(this.pinnedPeer.toString());
+    }
+  }
+
+  public async run() {
+    if (this.missingTxs.length === 0) {
+      this.logger.debug('No missing txs to request');
+      return;
+    }
+
+    const { promise, resolve } = promiseWithResolvers<void>();
+    this.startSmartRequester = resolve;
+
+    Promise.allSettled([this.smartRequester(promise), this.dumbRequester()]);
+  }
+
+  private async smartRequester(start: Promise<void>) {
+    const nextPeerIndex = this.makeRoundRobinIndexer(() => this.smartPeers.size);
+    // if we don't have a pinned peer we wait to start smart requester
+    // otherwise we start immediately with the pinned peer
+    if (!this.pinnedPeer) {
+      await start;
+    }
+
+    const nextPeer = () => peerIdFromString(Array.from(this.smartPeers)[nextPeerIndex()]);
+    const makeRequest = (pid: PeerId) => {
+      //TODO: for this peer we have to make batch on the fly based on which txs peer has
+      const txsPeerHas = this.peersToTxMap.get(pid.toString());
+      const peerHasTxs = txsPeerHas && txsPeerHas.length > 0;
+      if (!peerHasTxs) {
+        return undefined;
+      }
+
+      //TODO: make this smarter, we should only request txs that we don't have
+      // and we should request txs that have been requested the least times
+      const txsToRequest = txsPeerHas.slice(0, TX_BATCH_SIZE);
+
+      return BlockTxsRequest.fromBlockProposalAndMissingTxs(this.blockProposal, txsToRequest);
+    };
+
+    // Kick off workers
+    const workers = Array.from({ length: Math.min(PEERS_TO_QUERY_IN_PARALLEL, this.smartPeers.size) }, () =>
+      this.workerLoop(nextPeer, makeRequest),
+    );
+    await Promise.allSettled(workers);
+  }
+
+  private async dumbRequester() {
+    const nextPeerIndex = this.makeRoundRobinIndexer(() => peers.length);
+    const nextBatchIndex = this.makeRoundRobinIndexer(() => txChunks.length);
+
+    const peers = [...this.peers];
+    const txChunks = chunk<TxHash>(this.missingTxs, TX_BATCH_SIZE);
+
+    //TODO: batches should be adaptive
+    const makeRequest = (_pid: PeerId) => {
+      return BlockTxsRequest.fromBlockProposalAndMissingTxs(this.blockProposal, txChunks[nextBatchIndex()]);
+    };
+
+    const nextPeer = () => peers.filter(pid => !this.smartPeers.has(pid.toString()))[nextPeerIndex()];
+
+    const workers = Array.from({ length: Math.min(PEERS_TO_QUERY_IN_PARALLEL, peers.length) }, () =>
+      this.workerLoop(nextPeer, makeRequest),
+    );
+    await Promise.allSettled(workers);
+  }
+
+  private async workerLoop(pickNextPeer: () => PeerId, request: (pid: PeerId) => BlockTxsRequest | undefined) {
+    while (!this.shouldStop()) {
+      const peerId = pickNextPeer();
+      //
+      // TODO: think about this a bit more what should we do in this case?
+      const nextBatchTxRequest = request(peerId);
+      if (!nextBatchTxRequest) {
+        this.logger.warn('No txs matched');
+        continue;
+      }
+
+      await this.requestTxBatch(peerId, nextBatchTxRequest);
+    }
+  }
+
+  private async requestTxBatch(peerId: PeerId, request: BlockTxsRequest): Promise<BlockTxsResponse | undefined> {
+    try {
+      const response = await this.reqresp.sendRequestToPeer(peerId, ReqRespSubProtocol.BLOCK_TXS, request.toBuffer());
+      if (response.status !== ReqRespStatus.SUCCESS) {
+        return;
+      }
+
+      const block_response = BlockTxsResponse.fromBuffer(response.data);
+      this.handleSuccessResponseFromPeer(peerId, block_response);
+    } catch (err: any) {
+      this.logger.error(`Failed to deserialize response from peer ${peerId.toString()}: ${err.message}`, {
+        peerId,
+        error: err,
+      });
+
+      this.handleFailResponseFromPeer(peerId);
+    }
+  }
+
+  //TODO: 1 mark missing transactions as fetched
+  //TODO: 2 mark peer having this transactions
+  //TODO: 3 stream responses either via async generator or callbacks
+  private handleSuccessResponseFromPeer(peerId: PeerId, response: BlockTxsResponse) {
+    this.logger.debug(`Received txs: ${response.txs.length} from peer ${peerId.toString()} `);
+    if (!this.isBlockResponseValid(response)) {
+      return;
+    }
+
+    //TODO: yield txs
+    for (const tx of response.txs) {
+      const key = tx.txHash.toString();
+      let meta = this.txsMetadata.get(key);
+
+      if (!meta) {
+        meta = new MissingTxMetadata(tx.txHash, true);
+        this.txsMetadata.set(key, meta);
+      } else {
+        meta.fetched = true; // mutate in place; no need to re-set
+      }
+    }
+
+    const peerIdStr = peerId.toString();
+    this.smartPeers.add(peerIdStr);
+    const txsPeerHas = this.extractHashesPeerHasFromResponse(response);
+    // NOTE: it's ok to override this and not make it union with previous data
+    // because the newer request contains most up to date info
+    this.peersToTxMap.set(peerIdStr, txsPeerHas);
+
+    //TODO: maybe wait for at least couple of peers so that we don't spam single one?
+    if (this.startSmartRequester !== undefined) {
+      this.startSmartRequester();
+      // We use "undefined" here as marker that startSmartRequester has been called
+      this.startSmartRequester = undefined;
+    }
+  }
+
+  private isBlockResponseValid(response: BlockTxsResponse): boolean {
+    //TODO: maybe ban peer if this does not match?
+    const blockIdsMatch = this.blockProposal.archive === response.blockHash;
+    const peerHasSomeTxsFromProposal = !response.txIndices.isEmpty();
+    return blockIdsMatch && peerHasSomeTxsFromProposal;
+  }
+
+  //TODO:
+  private handleFailResponseFromPeer(peerId: PeerId) {}
+
+  private extractHashesPeerHasFromResponse(response: BlockTxsResponse): Array<TxHash> {
+    const hashes: TxHash[] = [];
+    const indicesOfHashesPeerHas = new Set(response.txIndices.getTrueIndices());
+    this.blockProposal.txHashes.forEach((hash, idx) => {
+      if (indicesOfHashesPeerHas.has(idx)) {
+        hashes.push(hash);
+      }
+    });
+
+    return hashes;
+  }
+
+  private makeRoundRobinIndexer(size: () => number, start = 0) {
+    let i = start;
+    return () => {
+      const current = i;
+      i = (current + 1) % size();
+      return current;
+    };
+  }
+
+  private fetchedAllTxs() {
+    return this.txsMetadata.values().every(tx => tx.fetched);
+  }
+
+  //TODO: stop on:
+  //1. abort signal
+  //2. deadline
+  //3. received all
+  private shouldStop() {
+    return this.fetchedAllTxs() || this.txsMetadata.size === 0;
+  }
+}