Minor runtime fixes for --next (#3265)

* some runtime fixes

* fixes

* fixes

* fixes

* fixes

* fixes

beacon-chain/deprecated-sync/BUILD.bazel

@@ -55,6 +55,7 @@ go_test(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/internal:go_default_library",
+        "//beacon-chain/sync:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//proto/eth/v1alpha1:go_default_library",
         "//shared/bytesutil:go_default_library",

beacon-chain/deprecated-sync/service.go

@@ -13,12 +13,6 @@ import (
 
 var slog = logrus.WithField("prefix", "sync")
 
-// Checker defines a struct which can verify whether a node is currently
-// synchronizing a chain with the rest of peers in the network.
-type Checker interface {
-	Syncing() bool
-}
-
 // Service defines the main routines used in the sync service.
 type Service struct {
 	RegularSync     *RegularSync

beacon-chain/deprecated-sync/service_test.go

@@ -7,12 +7,13 @@ import (
 
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/beacon-chain/internal"
+	"github.com/prysmaticlabs/prysm/beacon-chain/sync"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 )
 
-var _ = Checker(&Service{})
+var _ = sync.Checker(&Service{})
 
 func NotSyncQuerierConfig() *QuerierConfig {
 	return &QuerierConfig{

beacon-chain/node/node.go

@@ -403,9 +403,19 @@ func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
 		return err
 	}
 
-	var syncService *rbcsync.Service
-	if err := b.services.FetchService(&syncService); err != nil {
-		return err
+	var syncChecker prysmsync.Checker
+	if featureconfig.FeatureConfig().UseNewSync {
+		var syncService *prysmsync.RegularSync
+		if err := b.services.FetchService(&syncService); err != nil {
+			return err
+		}
+		syncChecker = syncService
+	} else {
+		var syncService *rbcsync.Service
+		if err := b.services.FetchService(&syncService); err != nil {
+			return err
+		}
+		syncChecker = syncService
 	}
 
 	port := ctx.GlobalString(flags.RPCPort.Name)
@@ -420,7 +430,7 @@ func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
 		ChainService:     chainService,
 		OperationService: operationService,
 		POWChainService:  web3Service,
-		SyncService:      syncService,
+		SyncService:      syncChecker,
 	})
 
 	return b.services.RegisterService(rpcService)

beacon-chain/p2p/interfaces.go

@@ -21,6 +21,8 @@ type P2P interface {
 	HandshakeManager
 	Sender
 	DeprecatedSubscriber
+
+	Started() bool
 }
 
 // Broadcaster broadcasts messages to peers over the p2p pubsub protocol.

beacon-chain/p2p/service.go

@@ -52,7 +52,6 @@ func (s *Service) Start() {
 		log.Error("Attempted to start p2p service when it was already started")
 		return
 	}
-	s.started = true
 
 	ipAddr := ipAddr(s.cfg)
 	privKey, err := privKey(s.cfg)
@@ -69,22 +68,29 @@ func (s *Service) Start() {
 		return
 	}
 	s.host = h
-	listener, err := startDiscoveryV5(ipAddr, privKey, s.cfg)
-	if err != nil {
-		s.startupErr = err
-		return
-	}
-	s.dv5Listener = listener
+	if s.cfg.BootstrapNodeAddr != "" {
+		listener, err := startDiscoveryV5(ipAddr, privKey, s.cfg)
+		if err != nil {
+			log.WithError(err).Error("Failed to start discovery")
+			s.startupErr = err
+			return
+		}
+		s.dv5Listener = listener
 
-	go s.listenForNewNodes()
+		go s.listenForNewNodes()
+	}
 
 	// TODO(3147): Add gossip sub options
 	gs, err := pubsub.NewGossipSub(s.ctx, s.host)
 	if err != nil {
 		s.startupErr = err
+
+		log.WithError(err).Error("Failed to start pubsub")
 		return
 	}
 	s.pubsub = gs
+
+	s.started = true
 }
 
 // Stop the p2p service and terminate all peer connections.
@@ -103,6 +109,11 @@ func (s *Service) Status() error {
 	return nil
 }
 
+// Started returns true if the p2p service has successfully started.
+func (s *Service) Started() bool {
+	return s.started
+}
+
 // Encoding returns the configured networking encoding.
 func (s *Service) Encoding() encoder.NetworkEncoding {
 	encoding := s.cfg.Encoding

beacon-chain/p2p/testing/p2p.go

@@ -150,3 +150,8 @@ func (p *TestP2P) Subscribe(msg proto.Message, ch chan deprecatedp2p.Message) ev
 	// TODO(3147): remove this.
 	return nil
 }
+
+// Started always returns true.
+func (p *TestP2P) Started() bool {
+	return true
+}

beacon-chain/rpc/BUILD.bazel

@@ -24,9 +24,9 @@ go_library(
         "//beacon-chain/core/state/stateutils:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filters:go_default_library",
-        "//beacon-chain/deprecated-sync:go_default_library",
         "//beacon-chain/operations:go_default_library",
         "//beacon-chain/p2p:go_default_library",
+        "//beacon-chain/sync:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",
         "//proto/beacon/rpc/v1:go_default_library",
         "//proto/eth/v1alpha1:go_default_library",

beacon-chain/rpc/node_server.go

@@ -7,7 +7,7 @@ import (
 
 	ptypes "github.com/gogo/protobuf/types"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
-	sync "github.com/prysmaticlabs/prysm/beacon-chain/deprecated-sync"
+	"github.com/prysmaticlabs/prysm/beacon-chain/sync"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/shared/version"
 	"google.golang.org/grpc"

beacon-chain/rpc/node_server_test.go

@@ -25,6 +25,10 @@ func (m *mockSyncChecker) Syncing() bool {
 	return m.syncing
 }
 
+func (m *mockSyncChecker) Status() error {
+	return nil
+}
+
 func TestNodeServer_GetSyncStatus(t *testing.T) {
 	mSync := &mockSyncChecker{false}
 	ns := &NodeServer{

beacon-chain/rpc/service.go

@@ -16,9 +16,9 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
-	sync "github.com/prysmaticlabs/prysm/beacon-chain/deprecated-sync"
 	"github.com/prysmaticlabs/prysm/beacon-chain/operations"
 	"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
+	"github.com/prysmaticlabs/prysm/beacon-chain/sync"
 	pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
@@ -68,11 +68,6 @@ type powChainService interface {
 	ChainStartETH1Data() *ethpb.Eth1Data
 }
 
-type syncService interface {
-	Status() error
-	sync.Checker
-}
-
 // Service defining an RPC server for a beacon node.
 type Service struct {
 	ctx                 context.Context
@@ -81,7 +76,7 @@ type Service struct {
 	chainService        chainService
 	powChainService     powChainService
 	operationService    operationService
-	syncService         syncService
+	syncService         sync.Checker
 	port                string
 	listener            net.Listener
 	withCert            string
@@ -102,7 +97,7 @@ type Config struct {
 	ChainService     chainService
 	POWChainService  powChainService
 	OperationService operationService
-	SyncService      syncService
+	SyncService      sync.Checker
 	Broadcaster      p2p.Broadcaster
 }
 

beacon-chain/sync/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/sync",
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
+        "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/core/blocks:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filters:go_default_library",

beacon-chain/sync/service.go

@@ -2,7 +2,9 @@ package sync
 
 import (
 	"context"
+	"time"
 
+	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/beacon-chain/operations"
 	"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
@@ -34,13 +36,19 @@ type RegularSync struct {
 	ctx        context.Context
 	p2p        p2p.P2P
 	db         db.Database
+	chain      *blockchain.ChainService
 	operations *operations.Service
 }
 
 // Start the regular sync service by initializing all of the p2p sync handlers.
 func (r *RegularSync) Start() {
+	log.Info("Starting regular sync")
+	for !r.p2p.Started() {
+		time.Sleep(200 * time.Millisecond)
+	}
 	r.registerRPCHandlers()
 	r.registerSubscribers()
+	log.Info("Regular sync started")
 }
 
 // Stop the regular sync service.
@@ -52,3 +60,16 @@ func (r *RegularSync) Stop() error {
 func (r *RegularSync) Status() error {
 	return nil
 }
+
+// Syncing returns true if the node is currently syncing with the network.
+func (r *RegularSync) Syncing() bool {
+	// TODO(3147): Use real value.
+	return false
+}
+
+// Checker defines a struct which can verify whether a node is currently
+// synchronizing a chain with the rest of peers in the network.
+type Checker interface {
+	Syncing() bool
+	Status() error
+}

shared/deprecated-p2p/service.go

@@ -170,6 +170,11 @@ func NewServer(cfg *ServerConfig) (*Server, error) {
 	}, nil
 }
 
+// Started always returns true as this library starts in the constructor.
+func (s *Server) Started() bool {
+	return true
+}
+
 func checkAvailablePort(port int) bool {
 	ip, err := iputils.ExternalIPv4()
 	if err != nil {

shared/service_registry.go

@@ -41,7 +41,7 @@ func (s *ServiceRegistry) StartAll() {
 	log.Infof("Starting %d services: %v", len(s.serviceTypes), s.serviceTypes)
 	for _, kind := range s.serviceTypes {
 		log.Debugf("Starting service type %v", kind)
-		s.services[kind].Start()
+		go s.services[kind].Start()
 	}
 }