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/buffered_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"maps"
-	"sync"
 	"testing"
 	"time"
 
@@ -26,9 +25,10 @@ const (
 	protocolID    = "test-buffered-topic-subscription-protocol"
 	topicName     = "test-buffered-topic-subscription-topic"
 	nodeCount     = 20
-	messageCount  = 100
-	logLevel      = zapcore.ErrorLevel
+	messageCount  = 50
+	logLevel      = zapcore.InfoLevel
 	retryInterval = 1 * time.Second
+	maxWait       = 5 * time.Second
 )
 
 type TestMessage = consensus.ConsensusStreamId
@@ -57,22 +57,29 @@ func TestBufferedTopicSubscriptionAndProtoBroadcaster(t *testing.T) {
 			}
 		}
 
-		iterator := iter.Iterator[*pubsub.Topic]{MaxGoroutines: len(topics)}
-		wg := sync.WaitGroup{}
-		wg.Add(len(messages))
+		iterator := iter.Iterator[*pubsub.Topic]{MaxGoroutines: nodeCount}
+		finished := make(chan struct{}, nodeCount)
+		liveness := make(chan struct{}, 1)
 
 		go func() {
 			iterator.ForEachIdx(topics, func(i int, destination **pubsub.Topic) {
 				logger := &utils.ZapLogger{SugaredLogger: logger.Named(fmt.Sprintf("destination-%d", i))}
 				pending := maps.Clone(allMessages)
 				subscription := buffered.NewTopicSubscription(logger, nodeCount*messageCount, func(ctx context.Context, msg *pubsub.Message) {
-					if len(pending) == 0 {
+					msgStr := string(msg.Message.Data)
+					if _, ok := pending[msgStr]; !ok {
 						return
 					}
 
-					delete(pending, string(msg.Message.Data))
+					select {
+					case liveness <- struct{}{}:
+					default:
+					}
+
+					delete(pending, msgStr)
+
 					if len(pending) == 0 {
-						wg.Done()
+						finished <- struct{}{}
 						logger.Info("all messages received")
 					}
 					logger.Debugw("received", "message", string(msg.Message.Data), "pending", len(pending))
@@ -83,7 +90,6 @@ func TestBufferedTopicSubscriptionAndProtoBroadcaster(t *testing.T) {
 		}()
 
 		go func() {
-			time.Sleep(1 * time.Second)
 			iterator.ForEachIdx(topics, func(i int, source **pubsub.Topic) {
 				logger := &utils.ZapLogger{SugaredLogger: logger.Named(fmt.Sprintf("source-%d", i))}
 				var rebroadcastStrategy buffered.RebroadcastStrategy[*TestMessage]
@@ -101,7 +107,9 @@ func TestBufferedTopicSubscriptionAndProtoBroadcaster(t *testing.T) {
 			})
 		}()
 
-		wg.Wait()
+		for range nodeCount {
+			wait(t, liveness, finished)
+		}
 	})
 
 	t.Run("canceled context", func(t *testing.T) {
@@ -128,6 +136,21 @@ func TestBufferedTopicSubscriptionAndProtoBroadcaster(t *testing.T) {
 	})
 }
 
+func wait(t *testing.T, liveness, finished chan struct{}) {
+	t.Helper()
+	for {
+		select {
+		case <-finished:
+			return
+		case <-liveness:
+			continue
+		case <-time.After(maxWait):
+			require.FailNow(t, "liveness check failed")
+			return
+		}
+	}
+}
+
 func getTestMessage(node, messageIndex int) *TestMessage {
 	return &TestMessage{
 		Nonce: uint64(node*messageCount + messageIndex),

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

@@ -25,12 +25,10 @@ import (
 type topicName string
 
 const (
-	protocolPrefix                = "starknet"
 	chainID                       = "1" // TODO: Make this configurable
 	consensusProtocolID           = "consensus"
 	proposalTopicName   topicName = "consensus_proposals"
 	voteTopicName       topicName = "consensus_votes"
-	gossipSubHistory              = 60
 )
 
 type P2P[V types.Hashable[H], H types.Hash, A types.Addr] interface {
@@ -122,14 +120,6 @@ func New(
 	}
 }
 
-func (p *p2p[V, H, A]) getGossipSubOptions() libp2p.Option {
-	params := libp2p.DefaultGossipSubParams()
-	params.HistoryLength = gossipSubHistory
-	params.HistoryGossip = gossipSubHistory
-
-	return libp2p.WithGossipSubParams(params)
-}
-
 func (p *p2p[V, H, A]) Run(ctx context.Context) error {
 	gossipSub, err := pubsub.Run(
 		ctx,
@@ -138,36 +128,42 @@ func (p *p2p[V, H, A]) Run(ctx context.Context) error {
 		p.host,
 		p.pubSubQueueSize,
 		p.bootstrapPeersFn,
-		p.getGossipSubOptions(),
 	)
 	if err != nil {
 		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

@@ -20,9 +20,9 @@ import (
 )
 
 const (
-	logLevel  = zapcore.DebugLevel
-	nodeCount = 3
-	txCount   = 300
+	logLevel  = zapcore.InfoLevel
+	nodeCount = 20
+	txCount   = 50
 	maxWait   = 5 * time.Second
 )
 
@@ -81,6 +81,7 @@ func TestMempoolBroadcastersAndListeners(t *testing.T) {
 					delete(pending, transaction.Transaction.Hash().String())
 					logger.Debugw("pending", "count", len(pending))
 					if len(pending) == 0 {
+						logger.Infow("all transactions received")
 						return
 					}
 				case <-time.After(maxWait):
@@ -96,9 +97,7 @@ func TestMempoolBroadcastersAndListeners(t *testing.T) {
 
 func getRandomTransactions(t *testing.T) []mempool.BroadcastedTransaction {
 	transactions := make([]mempool.BroadcastedTransaction, txCount)
-	transactions[0], _ = testutils.TransactionBuilder.GetTestDeclareTransaction(t)
-	transactions[1], _ = testutils.TransactionBuilder.GetTestDeployAccountTransaction(t)
-	for i := 2; i < txCount; i++ {
+	for i := range txCount {
 		transactions[i], _ = testutils.TransactionBuilder.GetTestInvokeTransaction(t)
 	}
 	return transactions

p2p/pubsub/pubsub.go

@@ -16,7 +16,8 @@ import (
 )
 
 const (
-	protocolPrefix = "starknet"
+	protocolPrefix   = "starknet"
+	gossipSubHistory = 60
 )
 
 func GetHost(hostPrivateKey crypto.PrivKey, hostAddress string) (host.Host, error) {
@@ -44,7 +45,6 @@ func Run(
 	host host.Host,
 	pubSubQueueSize int,
 	bootstrapPeersFn func() []peer.AddrInfo,
-	opts ...pubsub.Option,
 ) (*pubsub.PubSub, error) {
 	dht, err := dht.New(
 		ctx,
@@ -59,17 +59,35 @@ func Run(
 		return nil, fmt.Errorf("unable to create dht with error: %w", err)
 	}
 
+	params := pubsub.DefaultGossipSubParams()
+	params.HistoryLength = gossipSubHistory
+	params.HistoryGossip = gossipSubHistory
+
 	return pubsub.NewGossipSub(
 		ctx,
 		host,
-		append(opts,
-			pubsub.WithPeerOutboundQueueSize(pubSubQueueSize),
-			pubsub.WithValidateQueueSize(pubSubQueueSize),
-			pubsub.WithDiscovery(routing.NewRoutingDiscovery(dht)),
-		)...,
+		pubsub.WithPeerOutboundQueueSize(pubSubQueueSize),
+		pubsub.WithValidateQueueSize(pubSubQueueSize),
+		pubsub.WithDiscovery(routing.NewRoutingDiscovery(dht)),
+		pubsub.WithGossipSubParams(params),
 	)
 }
 
+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)