Merge pull request #1191 from onflow/smnzhu/staked-sync-provider

add sync provider to staked AN

Author: synzhu

cmd/access/node_builder/access_node_builder.go

@@ -601,12 +601,11 @@ func (builder *FlowAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 func (builder *FlowAccessNodeBuilder) initNetwork(nodeID module.Local,
 	networkMetrics module.NetworkMetrics,
 	middleware network.Middleware,
-	topology network.Topology) (*p2p.Network, error) {
+	topology network.Topology,
+) (*p2p.Network, error) {
 
 	codec := cborcodec.NewCodec()
 
-	subscriptionManager := p2p.NewChannelSubscriptionManager(middleware)
-
 	// creates network instance
 	net, err := p2p.NewNetwork(
 		builder.Logger,
@@ -615,7 +614,7 @@ func (builder *FlowAccessNodeBuilder) initNetwork(nodeID module.Local,
 		builder.Middleware,
 		p2p.DefaultCacheSize,
 		topology,
-		subscriptionManager,
+		p2p.NewChannelSubscriptionManager(middleware),
 		networkMetrics,
 		builder.IdentityProvider,
 	)

cmd/access/node_builder/staked_access_node_builder.go

@@ -8,11 +8,16 @@ import (
 
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/engine"
 	pingeng "github.com/onflow/flow-go/engine/access/ping"
+	"github.com/onflow/flow-go/engine/common/splitter"
+	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/id"
+	"github.com/onflow/flow-go/module/metrics/unstaked"
+	"github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/network/topology"
 	"github.com/onflow/flow-go/state/protocol/events/gadgets"
@@ -86,23 +91,61 @@ func (builder *StakedAccessNodeBuilder) Initialize() cmd.NodeBuilder {
 }
 
 func (anb *StakedAccessNodeBuilder) Build() AccessNodeBuilder {
-	anb.FlowAccessNodeBuilder.
-		Build().
-		Component("ping engine", func(builder cmd.NodeBuilder, node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			ping, err := pingeng.New(
-				node.Logger,
-				node.State,
-				node.Me,
-				anb.PingMetrics,
-				anb.pingEnabled,
-				node.Middleware,
-				anb.nodeInfoFile,
-			)
-			if err != nil {
-				return nil, fmt.Errorf("could not create ping engine: %w", err)
-			}
-			return ping, nil
-		})
+	anb.FlowAccessNodeBuilder.Build()
+
+	if anb.supportsUnstakedFollower {
+		var unstakedNetworkConduit network.Conduit
+		var proxyEngine *splitter.Engine
+
+		anb.
+			Component("unstaked sync request proxy", func(builder cmd.NodeBuilder, node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+				proxyEngine = splitter.New(node.Logger, engine.UnstakedSyncCommittee)
+
+				// register the proxy engine with the unstaked network
+				var err error
+				unstakedNetworkConduit, err = node.Network.Register(engine.UnstakedSyncCommittee, proxyEngine)
+				if err != nil {
+					return nil, fmt.Errorf("could not register unstaked sync request proxy: %w", err)
+				}
+
+				return proxyEngine, nil
+			}).
+			Component("unstaked sync request handler", func(builder cmd.NodeBuilder, node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+				syncRequestHandler := synceng.NewRequestHandlerEngine(
+					node.Logger.With().Bool("unstaked", true).Logger(),
+					unstaked.NewUnstakedEngineCollector(node.Metrics.Engine),
+					unstakedNetworkConduit,
+					node.Me,
+					node.Storage.Blocks,
+					anb.SyncCore,
+					anb.FinalizedHeader,
+					// don't queue missing heights from unstaked nodes
+					// since we are always more up-to-date than them
+					false,
+				)
+
+				// register the sync request handler with the proxy engine
+				proxyEngine.RegisterEngine(syncRequestHandler)
+
+				return syncRequestHandler, nil
+			})
+	}
+
+	anb.Component("ping engine", func(builder cmd.NodeBuilder, node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+		ping, err := pingeng.New(
+			node.Logger,
+			node.State,
+			node.Me,
+			anb.PingMetrics,
+			anb.pingEnabled,
+			node.Middleware,
+			anb.nodeInfoFile,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("could not create ping engine: %w", err)
+		}
+		return ping, nil
+	})
 
 	return anb
 }
@@ -156,7 +199,7 @@ func (builder *StakedAccessNodeBuilder) initLibP2PFactory(ctx context.Context,
 		myAddr = builder.BaseConfig.BindAddr
 	}
 
-	connManager := p2p.NewConnManager(builder.Logger, builder.Metrics.Network)
+	connManager := p2p.NewConnManager(builder.Logger, builder.Metrics.Network, p2p.TrackUnstakedConnections(builder.IdentityProvider))
 
 	return func() (*p2p.Node, error) {
 		libp2pNode, err := p2p.NewDefaultLibP2PNodeBuilder(nodeID, myAddr, networkKey).

cmd/access/node_builder/unstaked_access_node_builder.go

@@ -8,11 +8,13 @@ import (
 
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/id"
 	"github.com/onflow/flow-go/module/local"
 	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/network/converter"
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/state/protocol/events/gadgets"
 )
@@ -225,7 +227,7 @@ func (anb *UnstakedAccessNodeBuilder) enqueueUnstakedNetworkInit(ctx context.Con
 		network, err := anb.initNetwork(anb.Me, unstakedNetworkMetrics, anb.Middleware, nil)
 		anb.MustNot(err)
 
-		anb.Network = network
+		anb.Network = converter.NewNetwork(network, engine.SyncCommittee, engine.UnstakedSyncCommittee)
 
 		anb.Logger.Info().Msgf("network will run on address: %s", anb.BindAddr)
 

cmd/node_builder.go

@@ -132,7 +132,7 @@ type NodeConfig struct {
 	ProtocolEvents    *events.Distributor
 	State             protocol.State
 	Middleware        network.Middleware
-	Network           p2p.ReadyDoneAwareNetwork
+	Network           module.ReadyDoneAwareNetwork
 	MsgValidators     []network.MessageValidator
 	FvmOptions        []fvm.Option
 	StakingKey        crypto.PrivateKey

cmd/scaffold.go

@@ -164,7 +164,8 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit(ctx context.Context) {
 			fnb.RootBlock.ID().String(),
 			p2p.DefaultMaxPubSubMsgSize,
 			fnb.Metrics.Network,
-			pingProvider)
+			pingProvider,
+		)
 		if err != nil {
 			return nil, fmt.Errorf("could not generate libp2p node factory: %w", err)
 		}

engine/channels.go

@@ -133,6 +133,9 @@ const (
 	ProvideChunks            = RequestChunks
 	ProvideReceiptsByBlockID = RequestReceiptsByBlockID
 	ProvideApprovalsByChunk  = RequestApprovalsByChunk
+
+	// Unstaked network channels
+	UnstakedSyncCommittee = network.Channel("unstaked-sync-committee")
 )
 
 // initializeChannelRoleMap initializes an instance of channelRoleMap and populates it with the channels and their

engine/common/splitter/engine.go

@@ -1,7 +1,6 @@
 package splitter
 
 import (
-	"errors"
 	"fmt"
 	"sync"
 
@@ -40,17 +39,11 @@ func New(
 // RegisterEngine registers a new engine with the splitter. Events
 // that are received by the splitter after the engine has registered
 // will be passed down to it.
-func (e *Engine) RegisterEngine(engine module.Engine) error {
+func (e *Engine) RegisterEngine(engine module.Engine) {
 	e.enginesMu.Lock()
 	defer e.enginesMu.Unlock()
 
-	if _, ok := e.engines[engine]; ok {
-		return errors.New("engine already registered with splitter")
-	}
-
 	e.engines[engine] = struct{}{}
-
-	return nil
 }
 
 // UnregisterEngine unregisters an engine with the splitter. After

engine/common/splitter/engine_test.go

@@ -48,10 +48,8 @@ func (suite *Suite) TestDownstreamEngineFailure() {
 	engine1 := new(mockmodule.Engine)
 	engine2 := new(mockmodule.Engine)
 
-	err := suite.engine.RegisterEngine(engine1)
-	suite.Assert().Nil(err)
-	err = suite.engine.RegisterEngine(engine2)
-	suite.Assert().Nil(err)
+	suite.engine.RegisterEngine(engine1)
+	suite.engine.RegisterEngine(engine2)
 
 	processError := errors.New("Process Error!")
 
@@ -60,7 +58,7 @@ func (suite *Suite) TestDownstreamEngineFailure() {
 	engine1.On("Process", suite.channel, id, event).Return(processError).Once()
 	engine2.On("Process", suite.channel, id, event).Return(nil).Once()
 
-	err = suite.engine.Process(suite.channel, id, event)
+	err := suite.engine.Process(suite.channel, id, event)
 	merr, ok := err.(*multierror.Error)
 	suite.Assert().True(ok)
 	suite.Assert().Len(merr.Errors, 1)
@@ -100,26 +98,14 @@ func (suite *Suite) TestProcessUnknownChannel() {
 
 	engine := new(mockmodule.Engine)
 
-	err := suite.engine.RegisterEngine(engine)
-	suite.Assert().Nil(err)
+	suite.engine.RegisterEngine(engine)
 
-	err = suite.engine.Process(unknownChannel, id, event)
+	err := suite.engine.Process(unknownChannel, id, event)
 	suite.Assert().Error(err)
 
 	engine.AssertNumberOfCalls(suite.T(), "Process", 0)
 }
 
-// TestDuplicateRegistrations tests that an engine cannot register for the same channel twice.
-func (suite *Suite) TestDuplicateRegistrations() {
-	engine := new(mockmodule.Engine)
-
-	err := suite.engine.RegisterEngine(engine)
-	suite.Assert().Nil(err)
-
-	err = suite.engine.RegisterEngine(engine)
-	suite.Assert().Error(err)
-}
-
 // TestConcurrentEvents tests that sending multiple messages concurrently, results in each engine
 // receiving every message.
 func (suite *Suite) TestConcurrentEvents() {
@@ -131,8 +117,7 @@ func (suite *Suite) TestConcurrentEvents() {
 
 	for i := 0; i < numEngines; i++ {
 		engine := new(mockmodule.Engine)
-		err := suite.engine.RegisterEngine(engine)
-		suite.Assert().Nil(err)
+		suite.engine.RegisterEngine(engine)
 		engines[i] = engine
 	}
 

engine/common/splitter/network/network.go

@@ -11,7 +11,6 @@ import (
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/lifecycle"
 	"github.com/onflow/flow-go/network"
-	"github.com/onflow/flow-go/network/p2p"
 )
 
 // Network is the splitter network. It is a wrapper around the default network implementation
@@ -21,7 +20,7 @@ import (
 // splitter engine. As a result, multiple engines can register with the splitter network on
 // the same channel and will each receive all events on that channel.
 type Network struct {
-	p2p.ReadyDoneAwareNetwork
+	module.ReadyDoneAwareNetwork
 	mu        sync.RWMutex
 	log       zerolog.Logger
 	splitters map[network.Channel]*splitterEngine.Engine // stores splitters for each channel
@@ -31,7 +30,7 @@ type Network struct {
 
 // NewNetwork returns a new splitter network.
 func NewNetwork(
-	net p2p.ReadyDoneAwareNetwork,
+	net module.ReadyDoneAwareNetwork,
 	log zerolog.Logger,
 ) *Network {
 	return &Network{
@@ -76,19 +75,10 @@ func (n *Network) Register(channel network.Channel, e network.Engine) (network.C
 	}
 
 	// register engine with splitter
-	err := splitter.RegisterEngine(engine)
-
-	if err != nil {
-		// remove the splitter engine if this was the first time the given channel was registered
-		if !channelRegistered {
-			delete(n.splitters, channel)
-		}
-
-		return nil, fmt.Errorf("failed to register engine with splitter: %w", err)
-	}
+	splitter.RegisterEngine(engine)
 
 	if !channelRegistered {
-		conduit, err = n.ReadyDoneAwareNetwork.Register(channel, splitter)
+		conduit, err := n.ReadyDoneAwareNetwork.Register(channel, splitter)
 
 		if err != nil {
 			// undo previous steps

engine/common/splitter/network/network_test.go

@@ -32,7 +32,7 @@ type Suite struct {
 }
 
 func (suite *Suite) SetupTest() {
-	net := new(mocknetwork.ReadyDoneAwareNetwork)
+	net := new(mockmodule.ReadyDoneAwareNetwork)
 	suite.con = new(mocknetwork.Conduit)
 	suite.engines = make(map[network.Channel]module.Engine)