fix(p2p): wait for GossipSub mesh formation before sending txs in e2e tests

yarn-project/end-to-end/src/e2e_p2p/p2p_network.ts

@@ -25,6 +25,7 @@ import type { BootstrapNode } from '@aztec/p2p/bootstrap';
 import { createBootstrapNodeFromPrivateKey, getBootstrapNodeEnr } from '@aztec/p2p/test-helpers';
 import { tryStop } from '@aztec/stdlib/interfaces/server';
 import { SlashFactoryContract } from '@aztec/stdlib/l1-contracts';
+import { TopicType } from '@aztec/stdlib/p2p';
 import type { PublicDataTreeLeaf } from '@aztec/stdlib/trees';
 import { ZkPassportProofParams } from '@aztec/stdlib/zkpassport';
 import { getGenesisValues } from '@aztec/world-state/testing';
@@ -431,6 +432,27 @@ export class P2PNetworkTest {
     );
 
     this.logger.warn('All nodes connected to P2P mesh');
+
+    // Wait for GossipSub mesh to form for the tx topic.
+    // We only require at least 1 mesh peer per node because GossipSub
+    // stops grafting once it reaches Dlo peers and won't fill the mesh to all available peers.
+    this.logger.warn('Waiting for GossipSub mesh to form for tx topic...');
+    await Promise.all(
+      nodes.map(async (node, index) => {
+        const p2p = node.getP2P();
+        await retryUntil(
+          async () => {
+            const meshPeers = await p2p.getGossipMeshPeerCount(TopicType.tx);
+            this.logger.debug(`Node ${index} has ${meshPeers} gossip mesh peers for tx topic`);
+            return meshPeers >= 1 ? true : undefined;
+          },
+          `Node ${index} to have gossip mesh peers for tx topic`,
+          timeoutSeconds,
+          checkIntervalSeconds,
+        );
+      }),
+    );
+    this.logger.warn('All nodes have gossip mesh peers for tx topic');
   }
 
   async teardown() {

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

@@ -1,7 +1,13 @@
 import type { SlotNumber } from '@aztec/foundation/branded-types';
 import type { EthAddress, L2BlockId } from '@aztec/stdlib/block';
 import type { ITxProvider, P2PApiFull } from '@aztec/stdlib/interfaces/server';
-import type { BlockProposal, CheckpointAttestation, CheckpointProposal, P2PClientType } from '@aztec/stdlib/p2p';
+import type {
+  BlockProposal,
+  CheckpointAttestation,
+  CheckpointProposal,
+  P2PClientType,
+  TopicType,
+} from '@aztec/stdlib/p2p';
 import type { BlockHeader, Tx, TxHash } from '@aztec/stdlib/tx';
 
 import type { PeerId } from '@libp2p/interface';
@@ -237,4 +243,7 @@ export type P2P<T extends P2PClientType = P2PClientType.Full> = P2PApiFull<T> &
 
   /** If node running this P2P stack is validator, passes in validator address to P2P layer */
   registerThisValidatorAddresses(address: EthAddress[]): void;
+
+  /** Returns the number of peers in the GossipSub mesh for a given topic type. */
+  getGossipMeshPeerCount(topicType: TopicType): Promise<number>;
 };

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

@@ -25,6 +25,7 @@ import {
   CheckpointAttestation,
   type CheckpointProposal,
   type P2PClientType,
+  type TopicType,
 } from '@aztec/stdlib/p2p';
 import type { BlockHeader, Tx, TxHash } from '@aztec/stdlib/tx';
 import { Attributes, type TelemetryClient, WithTracer, getTelemetryClient, trackSpan } from '@aztec/telemetry-client';
@@ -167,6 +168,10 @@ export class P2PClient<T extends P2PClientType = P2PClientType.Full>
     return Promise.resolve(this.p2pService.getPeers(includePending));
   }
 
+  public getGossipMeshPeerCount(topicType: TopicType): Promise<number> {
+    return Promise.resolve(this.p2pService.getGossipMeshPeerCount(topicType));
+  }
+
   public getL2BlockHash(number: BlockNumber): Promise<string | undefined> {
     return this.l2Tips.getL2BlockHash(number);
   }

yarn-project/p2p/src/services/dummy_service.ts

@@ -1,6 +1,6 @@
 import type { EthAddress } from '@aztec/foundation/eth-address';
 import type { PeerInfo } from '@aztec/stdlib/interfaces/server';
-import type { Gossipable, PeerErrorSeverity } from '@aztec/stdlib/p2p';
+import type { Gossipable, PeerErrorSeverity, TopicType } from '@aztec/stdlib/p2p';
 import { Tx, TxHash } from '@aztec/stdlib/tx';
 
 import type { PeerId } from '@libp2p/interface';
@@ -44,6 +44,10 @@ export class DummyP2PService implements P2PService {
     return [];
   }
 
+  getGossipMeshPeerCount(_topicType: TopicType): number {
+    return 0;
+  }
+
   /**
    * Starts the dummy implementation.
    * @returns A resolved promise.

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

@@ -614,6 +614,10 @@ export class LibP2PService<T extends P2PClientType = P2PClientType.Full> extends
     return this.peerManager.getPeers(includePending);
   }
 
+  public getGossipMeshPeerCount(topicType: TopicType): number {
+    return this.node.services.pubsub.getMeshPeers(this.topicStrings[topicType]).length;
+  }
+
   private handleGossipSubEvent(e: CustomEvent<GossipsubMessage>) {
     this.logger.trace(`Received PUBSUB message.`);
 

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

@@ -1,7 +1,13 @@
 import type { SlotNumber } from '@aztec/foundation/branded-types';
 import type { EthAddress } from '@aztec/foundation/eth-address';
 import type { PeerInfo } from '@aztec/stdlib/interfaces/server';
-import type { BlockProposal, CheckpointAttestation, CheckpointProposalCore, Gossipable } from '@aztec/stdlib/p2p';
+import type {
+  BlockProposal,
+  CheckpointAttestation,
+  CheckpointProposalCore,
+  Gossipable,
+  TopicType,
+} from '@aztec/stdlib/p2p';
 import type { Tx } from '@aztec/stdlib/tx';
 
 import type { PeerId } from '@libp2p/interface';
@@ -130,6 +136,9 @@ export interface P2PService {
 
   getPeers(includePending?: boolean): PeerInfo[];
 
+  /** Returns the number of peers in the GossipSub mesh for a given topic type. */
+  getGossipMeshPeerCount(topicType: TopicType): number;
+
   validate(txs: Tx[]): Promise<void>;
 
   addReqRespSubProtocol(

yarn-project/p2p/src/test-helpers/mock-pubsub.ts

@@ -270,6 +270,16 @@ class MockGossipSubService extends TypedEventEmitter<GossipsubEvents> implements
       { msgId, propagationSource, acceptance },
     );
   }
+
+  getMeshPeers(topic?: TopicStr): PeerIdStr[] {
+    if (topic && !this.subscribedTopics.has(topic)) {
+      return [];
+    }
+    return this.network
+      .getPeers()
+      .filter(peer => !this.peerId.equals(peer))
+      .map(peer => peer.toString());
+  }
 }
 
 /**

yarn-project/p2p/src/util.ts

@@ -23,7 +23,13 @@ export interface PubSubLibp2p extends Pick<Libp2p, 'status' | 'start' | 'stop' |
   services: {
     pubsub: Pick<
       GossipSub,
-      'addEventListener' | 'removeEventListener' | 'publish' | 'subscribe' | 'reportMessageValidationResult' | 'direct'
+      | 'addEventListener'
+      | 'removeEventListener'
+      | 'publish'
+      | 'subscribe'
+      | 'reportMessageValidationResult'
+      | 'direct'
+      | 'getMeshPeers'
     > & { score: Pick<GossipSub['score'], 'score'> };
   };
 }

yarn-project/txe/src/state_machine/dummy_p2p_client.ts

@@ -17,7 +17,7 @@ import type {
 } from '@aztec/p2p';
 import type { EthAddress, L2BlockStreamEvent, L2Tips } from '@aztec/stdlib/block';
 import type { ITxProvider, PeerInfo } from '@aztec/stdlib/interfaces/server';
-import type { BlockProposal, CheckpointAttestation, CheckpointProposal } from '@aztec/stdlib/p2p';
+import type { BlockProposal, CheckpointAttestation, CheckpointProposal, TopicType } from '@aztec/stdlib/p2p';
 import type { BlockHeader, Tx, TxHash } from '@aztec/stdlib/tx';
 
 export class DummyP2P implements P2P {
@@ -41,6 +41,10 @@ export class DummyP2P implements P2P {
     throw new Error('DummyP2P does not implement "getPeers"');
   }
 
+  public getGossipMeshPeerCount(_topicType: TopicType): Promise<number> {
+    return Promise.resolve(0);
+  }
+
   public broadcastProposal(_proposal: BlockProposal): Promise<void> {
     throw new Error('DummyP2P does not implement "broadcastProposal"');
   }