WIP

Author: infrmtcs

.github/workflows/juno-test.yml

@@ -18,6 +18,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, ubuntu-arm64-4-core]
+        iteration: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
     runs-on: ${{ matrix.os }}
     env:
       VM_DEBUG: true

consensus/p2p/buffered/proto_broadcaster.go

@@ -42,6 +42,7 @@ func (b ProtoBroadcaster[M]) Broadcast(ctx context.Context, msg M) {
 func (b ProtoBroadcaster[M]) Loop(ctx context.Context, topic *pubsub.Topic) {
 	readinessOpt := pubsub.WithReadiness(pubsub.MinTopicSize(1))
 	var rebroadcasted rebroadcastMessages
+
 	for {
 		select {
 		case <-ctx.Done():

consensus/p2p/p2p.go

@@ -144,30 +144,37 @@ func (p *p2p[V, H, A]) Run(ctx context.Context) error {
 		return fmt.Errorf("unable to create gossipsub with error: %w", err)
 	}
 
-	topics := make(map[topicName]*libp2p.Topic)
+	topics := make([]*libp2p.Topic, 0, len(p.topicAttachment))
+	relayCancels := make([]func(), 0, len(p.topicAttachment))
 	defer func() {
+		for _, cancel := range relayCancels {
+			cancel()
+		}
 		for _, topic := range topics {
 			topic.Close()
 		}
 	}()
 
 	wg := conc.NewWaitGroup()
+	defer wg.Wait()
 
-	for topicName := range p.topicAttachment {
-		if topics[topicName], err = gossipSub.Join(string(topicName)); err != nil {
+	for topicName, services := range p.topicAttachment {
+		topic, relayCancel, err := pubsub.JoinTopic(gossipSub, string(topicName))
+		if err != nil {
 			return fmt.Errorf("unable to join topic %s with error: %w", topicName, err)
 		}
-	}
 
-	for topicName, services := range p.topicAttachment {
+		topics = append(topics, topic)
+		relayCancels = append(relayCancels, relayCancel)
+
 		for _, service := range services {
 			wg.Go(func() {
-				service.Loop(ctx, topics[topicName])
+				service.Loop(ctx, topic)
 			})
 		}
 	}
 
-	wg.Wait()
+	<-ctx.Done()
 	return nil
 }
 

mempool/p2p/p2p.go

@@ -61,10 +61,11 @@ func (p *P2P) Run(ctx context.Context) error {
 		return fmt.Errorf("unable to create gossipsub with error: %w", err)
 	}
 
-	topic, err := gossipSub.Join(transactionTopicName)
+	topic, relayCancel, err := pubsub.JoinTopic(gossipSub, transactionTopicName)
 	if err != nil {
 		return fmt.Errorf("unable to join topic %s with error: %w", transactionTopicName, err)
 	}
+	defer relayCancel()
 	defer topic.Close()
 
 	wg := conc.NewWaitGroup()

mempool/p2p/p2p_broadcasters_listeners_test.go

@@ -22,7 +22,7 @@ import (
 const (
 	logLevel  = zapcore.DebugLevel
 	nodeCount = 3
-	txCount   = 300
+	txCount   = 10
 	maxWait   = 5 * time.Second
 )
 

p2p/pubsub/pubsub.go

@@ -70,6 +70,21 @@ func Run(
 	)
 }
 
+func JoinTopic(pubSub *pubsub.PubSub, topicName string) (*pubsub.Topic, func(), error) {
+	topic, err := pubSub.Join(topicName)
+	if err != nil {
+		return nil, nil, fmt.Errorf("unable to join topic %s with error: %w", topicName, err)
+	}
+
+	// Make sure that the host starts connecting to other nodes
+	relayCancel, err := topic.Relay()
+	if err != nil {
+		return nil, nil, fmt.Errorf("unable to relay topic %s with error: %w", topicName, err)
+	}
+
+	return topic, relayCancel, nil
+}
+
 func ExtractPeers(peers string) ([]peer.AddrInfo, error) {
 	if peers == "" {
 		return nil, nil

p2p/pubsub/testutils/pubsub.go

@@ -64,15 +64,12 @@ func BuildNetworks(
 	return nodes
 }
 
-func (n Nodes) JoinTopic(t *testing.T, chainID, protocolID protocol.ID, topic string) []*libp2p.Topic {
+func (n Nodes) JoinTopic(t *testing.T, chainID, protocolID protocol.ID, topicName string) []*libp2p.Topic {
 	return iter.Map(n, func(node *Node) *libp2p.Topic {
-		pubsub, err := pubsub.Run(t.Context(), chainID, protocolID, node.Host, config.DefaultBufferSizes.PubSubQueueSize, node.GetBootstrapPeers)
+		pubSub, err := pubsub.Run(t.Context(), chainID, protocolID, node.Host, config.DefaultBufferSizes.PubSubQueueSize, node.GetBootstrapPeers)
 		require.NoError(t, err)
 
-		topic, err := pubsub.Join(topic)
-		require.NoError(t, err)
-
-		relayCancel, err := topic.Relay()
+		topic, relayCancel, err := pubsub.JoinTopic(pubSub, topicName)
 		require.NoError(t, err)
 		t.Cleanup(relayCancel)