refactor: Separate Sync P2P server from P2P service

Author: infrmtcs

node/node.go

@@ -396,7 +396,6 @@ func New(cfg *Config, version string, logLevel *utils.LogLevel) (*Node, error) {
 			} else if p2pService != nil {
 				// regular p2p node
 				p2pService.WithListener(makeSyncMetrics(&sync.NoopSynchronizer{}, chain))
-				p2pService.WithGossipTracer()
 			}
 		}
 		earlyServices = append(earlyServices, makeMetrics(cfg.MetricsHost, cfg.MetricsPort))

p2p/p2p.go

@@ -12,21 +12,21 @@ import (
 	"github.com/NethermindEth/juno/blockchain"
 	"github.com/NethermindEth/juno/db"
 	"github.com/NethermindEth/juno/p2p/dht"
-	p2pPeers "github.com/NethermindEth/juno/p2p/peers"
+	"github.com/NethermindEth/juno/p2p/server"
 	"github.com/NethermindEth/juno/p2p/starknetp2p"
 	p2pSync "github.com/NethermindEth/juno/p2p/sync"
 	junoSync "github.com/NethermindEth/juno/sync"
 	"github.com/NethermindEth/juno/utils"
 	"github.com/libp2p/go-libp2p"
 	libp2pdht "github.com/libp2p/go-libp2p-kad-dht"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/crypto/pb"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/multiformats/go-multiaddr"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -40,14 +40,12 @@ type Service struct {
 	host host.Host
 
 	network *utils.Network
-	handler *p2pPeers.Handler
+	server  *server.Server
 	log     utils.SimpleLogger
 
-	dht    *libp2pdht.IpfsDHT
-	pubsub *pubsub.PubSub
+	dht *libp2pdht.IpfsDHT
 
 	synchroniser *p2pSync.Service
-	gossipTracer *gossipTracer
 
 	feederNode bool
 	database   db.KeyValueStore
@@ -146,14 +144,15 @@ func NewWithHost(p2phost host.Host, peers string, feederNode bool, bc *blockchai
 
 	blockFetcher := p2pSync.NewBlockFetcher(bc, p2phost, snNetwork, log)
 	synchroniser := p2pSync.New(bc, log, &blockFetcher)
+	server := server.New(p2phost, bc, log)
 	s := &Service{
 		synchroniser: synchroniser,
 		log:          log,
 		host:         p2phost,
 		network:      snNetwork,
 		dht:          p2pdht,
 		feederNode:   feederNode,
-		handler:      p2pPeers.NewHandler(bc, log),
+		server:       &server,
 		database:     database,
 	}
 	return s, nil
@@ -208,16 +207,6 @@ func (s *Service) Run(ctx context.Context) error {
 		return err
 	}
 
-	var options []pubsub.Option
-	if s.gossipTracer != nil {
-		options = append(options, pubsub.WithRawTracer(s.gossipTracer))
-	}
-
-	s.pubsub, err = pubsub.NewGossipSub(ctx, s.host, options...)
-	if err != nil {
-		return err
-	}
-
 	defer s.callAndLogErr(s.dht.Close, "Failed stopping DHT")
 
 	listenAddrs, err := s.ListenAddrs()
@@ -228,10 +217,21 @@ func (s *Service) Run(ctx context.Context) error {
 		s.log.Infow("Listening on", "addr", addr)
 	}
 
-	s.setProtocolHandlers()
+	g := errgroup.Group{}
 
 	if !s.feederNode {
-		s.synchroniser.Run(ctx)
+		g.Go(func() error {
+			s.synchroniser.Run(ctx)
+			return nil
+		})
+	}
+
+	g.Go(func() error {
+		return s.server.Run(ctx)
+	})
+
+	if err := g.Wait(); err != nil {
+		return err
 	}
 
 	<-ctx.Done()
@@ -244,14 +244,6 @@ func (s *Service) Run(ctx context.Context) error {
 	return nil
 }
 
-func (s *Service) setProtocolHandlers() {
-	s.SetProtocolHandler(starknetp2p.HeadersSyncSubProtocol, s.handler.HeadersHandler)
-	s.SetProtocolHandler(starknetp2p.EventsSyncSubProtocol, s.handler.EventsHandler)
-	s.SetProtocolHandler(starknetp2p.TransactionsSyncSubProtocol, s.handler.TransactionsHandler)
-	s.SetProtocolHandler(starknetp2p.ClassesSyncSubProtocol, s.handler.ClassesHandler)
-	s.SetProtocolHandler(starknetp2p.StateDiffSyncSubProtocol, s.handler.StateDiffHandler)
-}
-
 func (s *Service) callAndLogErr(f func() error, msg string) {
 	err := f()
 	if err != nil {
@@ -299,21 +291,10 @@ func (s *Service) NewStream(ctx context.Context, pids ...protocol.ID) (network.S
 	}
 }
 
-func (s *Service) SetProtocolHandler(
-	syncSubProtocol starknetp2p.SyncSubProtocol,
-	handler func(network.Stream),
-) {
-	s.host.SetStreamHandler(starknetp2p.Sync(s.network, syncSubProtocol), handler)
-}
-
 func (s *Service) WithListener(l junoSync.EventListener) {
 	s.synchroniser.WithListener(l)
 }
 
-func (s *Service) WithGossipTracer() {
-	s.gossipTracer = NewGossipTracer(s.host)
-}
-
 // persistPeers stores the given peers in the peers database
 func (s *Service) persistPeers() error {
 	txn := s.database.NewBatch()

p2p/peers/iterator.go renamed to p2p/server/iterator.go

@@ -1,4 +1,4 @@
-package peers
+package server
 
 import (
 	"errors"

p2p/peers/iterator_test.go renamed to p2p/server/iterator_test.go

@@ -1,4 +1,4 @@
-package peers
+package server
 
 import (
 	"testing"