Subscription filter

cmd/access/node_builder/staked_access_node_builder.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/libp2p/go-libp2p-core/host"
 	dht "github.com/libp2p/go-libp2p-kad-dht"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/crypto"
@@ -100,11 +102,11 @@ func (anb *StakedAccessNodeBuilder) Build() AccessNodeBuilder {
 
 		anb.
 			Component("unstaked sync request proxy", func(builder cmd.NodeBuilder, node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-				proxyEngine = splitter.New(node.Logger, engine.UnstakedSyncCommittee)
+				proxyEngine = splitter.New(node.Logger, engine.PublicSyncCommittee)
 
 				// register the proxy engine with the unstaked network
 				var err error
-				unstakedNetworkConduit, err = node.Network.Register(engine.UnstakedSyncCommittee, proxyEngine)
+				unstakedNetworkConduit, err = node.Network.Register(engine.PublicSyncCommittee, proxyEngine)
 				if err != nil {
 					return nil, fmt.Errorf("could not register unstaked sync request proxy: %w", err)
 				}
@@ -214,13 +216,19 @@ func (builder *StakedAccessNodeBuilder) initLibP2PFactory(ctx context.Context,
 	}
 
 	return func() (*p2p.Node, error) {
+		psOpts := p2p.DefaultPubsubOptions(p2p.DefaultMaxPubSubMsgSize)
+		psOpts = append(psOpts, func(_ context.Context, h host.Host) (pubsub.Option, error) {
+			return pubsub.WithSubscriptionFilter(p2p.NewRoleBasedFilter(
+				h.ID(), builder.RootBlock.ID(), builder.RootChainID, builder.IdentityProvider,
+			)), nil
+		})
 		libp2pNode, err := p2p.NewDefaultLibP2PNodeBuilder(nodeID, myAddr, networkKey).
-			SetRootBlockID(builder.RootBlock.ID().String()).
+			SetRootBlockID(builder.RootBlock.ID()).
 			// no connection gater
 			SetConnectionManager(connManager).
 			// act as a DHT server
 			SetDHTOptions(dhtOptions...).
-			SetPubsubOptions(p2p.DefaultPubsubOptions(p2p.DefaultMaxPubSubMsgSize)...).
+			SetPubsubOptions(psOpts...).
 			SetLogger(builder.Logger).
 			SetResolver(resolver).
 			Build(ctx)
@@ -247,7 +255,7 @@ func (builder *StakedAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 		factoryFunc,
 		nodeID,
 		networkMetrics,
-		builder.RootBlock.ID().String(),
+		builder.RootBlock.ID(),
 		p2p.DefaultUnicastTimeout,
 		false, // no connection gating to allow unstaked nodes to connect
 		builder.IDTranslator,

cmd/access/node_builder/unstaked_access_node_builder.go

@@ -162,7 +162,7 @@ func (builder *UnstakedAccessNodeBuilder) initLibP2PFactory(ctx context.Context,
 
 	return func() (*p2p.Node, error) {
 		libp2pNode, err := p2p.NewDefaultLibP2PNodeBuilder(nodeID, builder.BaseConfig.BindAddr, networkKey).
-			SetRootBlockID(builder.RootBlock.ID().String()).
+			SetRootBlockID(builder.RootBlock.ID()).
 			SetConnectionManager(connManager).
 			// unlike the staked side of the network where currently all the node addresses are known upfront,
 			// for the unstaked side of the network, the  nodes need to discover each other using DHT Discovery.
@@ -249,7 +249,7 @@ func (anb *UnstakedAccessNodeBuilder) enqueueUnstakedNetworkInit(ctx context.Con
 			return nil, err
 		}
 
-		anb.Network = converter.NewNetwork(network, engine.SyncCommittee, engine.UnstakedSyncCommittee)
+		anb.Network = converter.NewNetwork(network, engine.SyncCommittee, engine.PublicSyncCommittee)
 
 		anb.Logger.Info().Msgf("network will run on address: %s", anb.BindAddr)
 
@@ -284,7 +284,7 @@ func (anb *UnstakedAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 		factoryFunc,
 		nodeID,
 		networkMetrics,
-		anb.RootBlock.ID().String(),
+		anb.RootBlock.ID(),
 		p2p.DefaultUnicastTimeout,
 		false, // no connection gating for the unstaked nodes
 		anb.IDTranslator,

cmd/scaffold.go

@@ -178,7 +178,9 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit(ctx context.Context) {
 			fnb.Me.NodeID(),
 			myAddr,
 			fnb.NetworkKey,
-			fnb.RootBlock.ID().String(),
+			fnb.RootBlock.ID(),
+			fnb.RootChainID,
+			fnb.IdentityProvider,
 			p2p.DefaultMaxPubSubMsgSize,
 			fnb.Metrics.Network,
 			pingProvider,
@@ -204,7 +206,7 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit(ctx context.Context) {
 			libP2PNodeFactory,
 			fnb.Me.NodeID(),
 			fnb.Metrics.Network,
-			fnb.RootBlock.ID().String(),
+			fnb.RootBlock.ID(),
 			fnb.BaseConfig.UnicastMessageTimeout,
 			true,
 			fnb.IDTranslator,

engine/channels.go

@@ -21,20 +21,22 @@ var channelRoleMap map[network.Channel]flow.RoleList
 
 // RolesByChannel returns list of flow roles involved in the channel.
 func RolesByChannel(channel network.Channel) (flow.RoleList, bool) {
-	if clusterChannel, isCluster := ClusterChannel(channel); isCluster {
-		// replaces channel with the stripped-off prefix
-		channel = clusterChannel
+	if _, isCluster := ClusterChannel(channel); isCluster {
+		return ClusterChannelRoles(), true
 	}
 	roles, ok := channelRoleMap[channel]
 	return roles, ok
 }
 
-// Exists returns true if channel exists in channelRoleMap.
-// At the current state, any developer-defined channel should be added
-// to channelRoleMap as a constant channel type manually.
+// Exists returns true if the channel exists.
 func Exists(channel network.Channel) bool {
-	_, exists := RolesByChannel(channel)
-	return exists || UnstakedChannels().Contains(channel)
+	if _, ok := RolesByChannel(channel); !ok {
+		if _, isClusterChannel := ClusterChannel(channel); !isClusterChannel {
+			return PublicChannels().Contains(channel)
+		}
+	}
+
+	return true
 }
 
 // ChannelsByRole returns a list of all channels the role subscribes to.
@@ -63,10 +65,9 @@ func UniqueChannels(channels network.ChannelList) network.ChannelList {
 	// has already been added to uniques.
 	// We use identifier of RoleList to determine its uniqueness.
 	for _, channel := range channels {
-		id := channelRoleMap[channel].ID()
-
 		// non-cluster channel deduplicated based identifier of role list
 		if _, cluster := ClusterChannel(channel); !cluster {
+			id := channelRoleMap[channel].ID()
 			if _, ok := added[id]; ok {
 				// a channel with same RoleList already added, hence skips
 				continue
@@ -90,10 +91,10 @@ func Channels() network.ChannelList {
 	return channels
 }
 
-// UnstakedChannels returns all channels that unstaked nodes can send messages on.
-func UnstakedChannels() network.ChannelList {
+// PublicChannels returns all channels that on the public network.
+func PublicChannels() network.ChannelList {
 	return network.ChannelList{
-		UnstakedSyncCommittee,
+		PublicSyncCommittee,
 	}
 }
 
@@ -141,8 +142,8 @@ const (
 	ProvideReceiptsByBlockID = RequestReceiptsByBlockID
 	ProvideApprovalsByChunk  = RequestApprovalsByChunk
 
-	// Unstaked network channels
-	UnstakedSyncCommittee = network.Channel("unstaked-sync-committee")
+	// Public network channels
+	PublicSyncCommittee = network.Channel("public-sync-committee")
 )
 
 // initializeChannelRoleMap initializes an instance of channelRoleMap and populates it with the channels and their
@@ -161,7 +162,7 @@ func initializeChannelRoleMap() {
 	channelRoleMap[ConsensusCommittee] = flow.RoleList{flow.RoleConsensus}
 
 	// Channels for protocols actively synchronizing state across nodes
-	channelRoleMap[SyncCommittee] = flow.RoleList{flow.RoleConsensus}
+	channelRoleMap[SyncCommittee] = flow.Roles()
 	channelRoleMap[SyncExecution] = flow.RoleList{flow.RoleExecution}
 
 	// Channels for DKG communication
@@ -177,7 +178,7 @@ func initializeChannelRoleMap() {
 	channelRoleMap[PushApprovals] = flow.RoleList{flow.RoleConsensus, flow.RoleVerification}
 
 	// Channels for actively requesting missing entities
-	channelRoleMap[RequestCollections] = flow.RoleList{flow.RoleCollection, flow.RoleExecution}
+	channelRoleMap[RequestCollections] = flow.RoleList{flow.RoleCollection, flow.RoleExecution, flow.RoleAccess}
 	channelRoleMap[RequestChunks] = flow.RoleList{flow.RoleExecution, flow.RoleVerification}
 	channelRoleMap[RequestReceiptsByBlockID] = flow.RoleList{flow.RoleConsensus, flow.RoleExecution}
 	channelRoleMap[RequestApprovalsByChunk] = flow.RoleList{flow.RoleConsensus, flow.RoleVerification}
@@ -190,13 +191,15 @@ func initializeChannelRoleMap() {
 		flow.RoleAccess}
 	channelRoleMap[ReceiveApprovals] = flow.RoleList{flow.RoleConsensus, flow.RoleVerification}
 
-	channelRoleMap[ProvideCollections] = flow.RoleList{flow.RoleCollection, flow.RoleExecution}
+	channelRoleMap[ProvideCollections] = flow.RoleList{flow.RoleCollection, flow.RoleExecution, flow.RoleAccess}
 	channelRoleMap[ProvideChunks] = flow.RoleList{flow.RoleExecution, flow.RoleVerification}
 	channelRoleMap[ProvideReceiptsByBlockID] = flow.RoleList{flow.RoleConsensus, flow.RoleExecution}
 	channelRoleMap[ProvideApprovalsByChunk] = flow.RoleList{flow.RoleConsensus, flow.RoleVerification}
+}
 
-	channelRoleMap[syncClusterPrefix] = flow.RoleList{flow.RoleCollection}
-	channelRoleMap[consensusClusterPrefix] = flow.RoleList{flow.RoleCollection}
+// ClusterChannelRoles returns the list of roles that are involved in cluster-based channels.
+func ClusterChannelRoles() flow.RoleList {
+	return flow.RoleList{flow.RoleCollection}
 }
 
 // ClusterChannel returns true if channel is cluster-based.
@@ -217,13 +220,13 @@ func ClusterChannel(channel network.Channel) (network.Channel, bool) {
 // TopicFromChannel returns the unique LibP2P topic form the channel.
 // The channel is made up of name string suffixed with root block id.
 // The root block id is used to prevent cross talks between nodes on different sporks.
-func TopicFromChannel(channel network.Channel, rootBlockID string) network.Topic {
+func TopicFromChannel(channel network.Channel, rootBlockID flow.Identifier) network.Topic {
 	// skip root block suffix, if this is a cluster specific channel. A cluster specific channel is inherently
 	// unique for each epoch
-	if strings.HasPrefix(channel.String(), syncClusterPrefix.String()) || strings.HasPrefix(string(channel), consensusClusterPrefix.String()) {
+	if _, isClusterChannel := ClusterChannel(channel); isClusterChannel {
 		return network.Topic(channel)
 	}
-	return network.Topic(fmt.Sprintf("%s/%s", string(channel), rootBlockID))
+	return network.Topic(fmt.Sprintf("%s/%s", string(channel), rootBlockID.String()))
 }
 
 // ChannelConsensusCluster returns a dynamic cluster consensus channel based on

engine/channels_test.go

@@ -59,14 +59,15 @@ func TestGetChannelByRole(t *testing.T) {
 	// - TestMetric
 	// the roles list should contain collection and consensus roles
 	topics := ChannelsByRole(flow.RoleVerification)
-	assert.Len(t, topics, 7)
+	assert.Len(t, topics, 8)
 	assert.Contains(t, topics, PushBlocks)
 	assert.Contains(t, topics, PushReceipts)
 	assert.Contains(t, topics, PushApprovals)
 	assert.Contains(t, topics, ProvideApprovalsByChunk)
 	assert.Contains(t, topics, RequestChunks)
 	assert.Contains(t, topics, TestMetrics)
 	assert.Contains(t, topics, TestNetwork)
+	assert.Contains(t, topics, SyncCommittee)
 }
 
 // TestIsClusterChannel verifies the correctness of ClusterChannel method
@@ -121,10 +122,13 @@ func TestUniqueChannels_Uniqueness(t *testing.T) {
 // We use the identifier of RoleList to determine their uniqueness.
 func TestUniqueChannels_ClusterChannels(t *testing.T) {
 	channels := ChannelsByRole(flow.RoleCollection)
+	consensusCluster := ChannelConsensusCluster(flow.Emulator)
+	syncCluster := ChannelSyncCluster(flow.Emulator)
+	channels = append(channels, consensusCluster, syncCluster)
 	uniques := UniqueChannels(channels)
 	// collection role has two cluster and one non-cluster channels all with the same RoleList.
 	// Hence all of them should be returned as unique channels.
-	require.Contains(t, uniques, syncClusterPrefix)      // cluster channel
-	require.Contains(t, uniques, consensusClusterPrefix) // cluster channel
-	require.Contains(t, uniques, PushTransactions)       // non-cluster channel
+	require.Contains(t, uniques, syncCluster)      // cluster channel
+	require.Contains(t, uniques, consensusCluster) // cluster channel
+	require.Contains(t, uniques, PushTransactions) // non-cluster channel
 }

model/flow/role.go

@@ -71,7 +71,7 @@ func (r *Role) UnmarshalText(text []byte) error {
 	return err
 }
 
-func Roles() []Role {
+func Roles() RoleList {
 	return []Role{RoleCollection, RoleConsensus, RoleExecution, RoleVerification, RoleAccess}
 }
 

network/p2p/libp2pNode.go

@@ -31,6 +31,7 @@ import (
 	fcrypto "github.com/onflow/flow-go/crypto"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/id"
 	flownet "github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/network/message"
 	"github.com/onflow/flow-go/network/p2p/dns"
@@ -65,7 +66,9 @@ func DefaultLibP2PNodeFactory(ctx context.Context,
 	me flow.Identifier,
 	address string,
 	flowKey fcrypto.PrivateKey,
-	rootBlockID string,
+	rootBlockID flow.Identifier,
+	chainID flow.ChainID,
+	idProvider id.IdentityProvider,
 	maxPubSubMsgSize int,
 	metrics module.NetworkMetrics,
 	pingInfoProvider PingInfoProvider,
@@ -80,12 +83,21 @@ func DefaultLibP2PNodeFactory(ctx context.Context,
 		return nil, fmt.Errorf("could not create dns resolver: %w", err)
 	}
 
+	psOpts := DefaultPubsubOptions(maxPubSubMsgSize)
+	if chainID == flow.Mainnet {
+		psOpts = append(psOpts, func(_ context.Context, h host.Host) (pubsub.Option, error) {
+			return pubsub.WithSubscriptionFilter(NewRoleBasedFilter(
+				h.ID(), rootBlockID, chainID, idProvider,
+			)), nil
+		})
+	}
+
 	return func() (*Node, error) {
 		return NewDefaultLibP2PNodeBuilder(me, address, flowKey).
 			SetRootBlockID(rootBlockID).
 			SetConnectionGater(connGater).
 			SetConnectionManager(connManager).
-			SetPubsubOptions(DefaultPubsubOptions(maxPubSubMsgSize)...).
+			SetPubsubOptions(psOpts...).
 			SetPingInfoProvider(pingInfoProvider).
 			SetLogger(log).
 			SetResolver(resolver).
@@ -94,7 +106,7 @@ func DefaultLibP2PNodeFactory(ctx context.Context,
 }
 
 type NodeBuilder interface {
-	SetRootBlockID(string) NodeBuilder
+	SetRootBlockID(flow.Identifier) NodeBuilder
 	SetConnectionManager(TagLessConnManager) NodeBuilder
 	SetConnectionGater(*ConnGater) NodeBuilder
 	SetPubsubOptions(...PubsubOption) NodeBuilder
@@ -108,7 +120,7 @@ type NodeBuilder interface {
 
 type DefaultLibP2PNodeBuilder struct {
 	id               flow.Identifier
-	rootBlockID      string
+	rootBlockID      flow.Identifier
 	logger           zerolog.Logger
 	connGater        *ConnGater
 	connMngr         TagLessConnManager
@@ -144,7 +156,7 @@ func (builder *DefaultLibP2PNodeBuilder) SetTopicValidation(enabled bool) NodeBu
 	return builder
 }
 
-func (builder *DefaultLibP2PNodeBuilder) SetRootBlockID(rootBlockId string) NodeBuilder {
+func (builder *DefaultLibP2PNodeBuilder) SetRootBlockID(rootBlockId flow.Identifier) NodeBuilder {
 	builder.rootBlockID = rootBlockId
 	return builder
 }
@@ -196,7 +208,7 @@ func (builder *DefaultLibP2PNodeBuilder) Build(ctx context.Context) (*Node, erro
 		return nil, errors.New("unable to create libp2p pubsub: factory function not provided")
 	}
 
-	if builder.rootBlockID == "" {
+	if builder.rootBlockID == flow.ZeroID {
 		return nil, errors.New("root block ID must be provided")
 	}
 	node.flowLibP2PProtocolID = generateFlowProtocolID(builder.rootBlockID)
@@ -213,11 +225,6 @@ func (builder *DefaultLibP2PNodeBuilder) Build(ctx context.Context) (*Node, erro
 		node.connMgr = builder.connMngr
 	}
 
-	if builder.rootBlockID == "" {
-		return nil, errors.New("root block ID must be provided")
-	}
-	node.flowLibP2PProtocolID = generateFlowProtocolID(builder.rootBlockID)
-
 	if builder.pingInfoProvider != nil {
 		opts = append(opts, libp2p.Ping(true))
 	}
@@ -729,7 +736,6 @@ func DefaultPubsubOptions(maxPubSubMsgSize int) []PubsubOption {
 		pubSubOptionFunc(pubsub.WithStrictSignatureVerification(true)),
 		// set max message size limit for 1-k PubSub messaging
 		pubSubOptionFunc(pubsub.WithMaxMessageSize(maxPubSubMsgSize)),
-		// no discovery
 	}
 }
 

network/p2p/libp2pNode_test.go

@@ -41,7 +41,7 @@ const tickForAssertEventually = 100 * time.Millisecond
 // "0.0.0.0:<selected-port-by-os>
 const defaultAddress = "0.0.0.0:0"
 
-var rootBlockID = unittest.IdentifierFixture().String()
+var rootBlockID = unittest.IdentifierFixture()
 
 type LibP2PNodeTestSuite struct {
 	suite.Suite
@@ -685,7 +685,7 @@ func (suite *LibP2PNodeTestSuite) NodesFixture(count int, handler func(t *testin
 
 // NodeFixture creates a single LibP2PNodes with the given key, root block id, and callback function for stream handling.
 // It returns the nodes and their identities.
-func NodeFixture(t *testing.T, log zerolog.Logger, key fcrypto.PrivateKey, rootID string, handler func(t *testing.T) network.StreamHandler, allowList bool, address string) (*Node, flow.Identity) {
+func NodeFixture(t *testing.T, log zerolog.Logger, key fcrypto.PrivateKey, rootID flow.Identifier, handler func(t *testing.T) network.StreamHandler, allowList bool, address string) (*Node, flow.Identity) {
 
 	identity := unittest.IdentityFixture(unittest.WithNetworkingKey(key.PublicKey()), unittest.WithAddress(address))
 

network/p2p/libp2pUtils.go

@@ -162,12 +162,12 @@ func IPPortFromMultiAddress(addrs ...multiaddr.Multiaddr) (string, string, error
 	return "", "", fmt.Errorf("ip address or hostname not found")
 }
 
-func generateFlowProtocolID(rootBlockID string) protocol.ID {
-	return protocol.ID(FlowLibP2POneToOneProtocolIDPrefix + rootBlockID)
+func generateFlowProtocolID(rootBlockID flow.Identifier) protocol.ID {
+	return protocol.ID(FlowLibP2POneToOneProtocolIDPrefix + rootBlockID.String())
 }
 
-func generatePingProtcolID(rootBlockID string) protocol.ID {
-	return protocol.ID(FlowLibP2PPingProtocolPrefix + rootBlockID)
+func generatePingProtcolID(rootBlockID flow.Identifier) protocol.ID {
+	return protocol.ID(FlowLibP2PPingProtocolPrefix + rootBlockID.String())
 }
 
 // PeerAddressInfo generates the libp2p peer.AddrInfo for the given Flow.Identity.