[Networking] Refinement of message scope; encapsulating the publish logic in the libp2p node

cmd/access/node_builder/access_node_builder.go

@@ -802,6 +802,7 @@ func (builder *FlowAccessNodeBuilder) initNetwork(nodeID module.Local,
 		IdentityProvider:    builder.IdentityProvider,
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             builder.SporkID,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  builder.Logger,
 			SpamRecordCacheSize:     builder.FlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,
@@ -1404,7 +1405,7 @@ func (builder *FlowAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 		Libp2pNode:            libp2pNode,
 		FlowId:                nodeID,
 		BitSwapMetrics:        builder.Metrics.Bitswap,
-		RootBlockID:           builder.SporkID,
+		SporkId:               builder.SporkID,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		IdTranslator:          builder.IDTranslator,
 		Codec:                 builder.CodecFactory(),

cmd/observer/node_builder/observer_builder.go

@@ -493,6 +493,7 @@ func (builder *ObserverServiceBuilder) initNetwork(nodeID module.Local,
 		IdentityProvider:    builder.IdentityProvider,
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             builder.SporkID,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  builder.Logger,
 			SpamRecordCacheSize:     builder.FlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,
@@ -1031,7 +1032,7 @@ func (builder *ObserverServiceBuilder) initMiddleware(nodeID flow.Identifier,
 		Libp2pNode:            libp2pNode,
 		FlowId:                nodeID,
 		BitSwapMetrics:        builder.Metrics.Bitswap,
-		RootBlockID:           builder.SporkID,
+		SporkId:               builder.SporkID,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		IdTranslator:          builder.IDTranslator,
 		Codec:                 builder.CodecFactory(),

cmd/scaffold.go

@@ -440,7 +440,7 @@ func (fnb *FlowNodeBuilder) InitFlowNetworkWithConduitFactory(
 		Libp2pNode:            fnb.LibP2PNode,
 		FlowId:                fnb.Me.NodeID(),
 		BitSwapMetrics:        fnb.Metrics.Bitswap,
-		RootBlockID:           fnb.SporkID,
+		SporkId:               fnb.SporkID,
 		UnicastMessageTimeout: fnb.FlowConfig.NetworkConfig.UnicastMessageTimeout,
 		IdTranslator:          fnb.IDTranslator,
 		Codec:                 fnb.CodecFactory(),
@@ -465,6 +465,7 @@ func (fnb *FlowNodeBuilder) InitFlowNetworkWithConduitFactory(
 		Logger:              fnb.Logger,
 		Codec:               fnb.CodecFactory(),
 		Me:                  fnb.Me,
+		SporkId:             fnb.SporkID,
 		MiddlewareFactory:   func() (network.Middleware, error) { return fnb.Middleware, nil },
 		Topology:            topology.NewFullyConnectedTopology(),
 		SubscriptionManager: subscriptionManager,

follower/follower_builder.go

@@ -380,6 +380,7 @@ func (builder *FollowerServiceBuilder) initNetwork(nodeID module.Local,
 		IdentityProvider:    builder.IdentityProvider,
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             builder.SporkID,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  builder.Logger,
 			SpamRecordCacheSize:     builder.FlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,
@@ -758,7 +759,7 @@ func (builder *FollowerServiceBuilder) initMiddleware(nodeID flow.Identifier,
 		Libp2pNode:            libp2pNode,
 		FlowId:                nodeID,
 		BitSwapMetrics:        builder.Metrics.Bitswap,
-		RootBlockID:           builder.SporkID,
+		SporkId:               builder.SporkID,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		IdTranslator:          builder.IDTranslator,
 		Codec:                 builder.CodecFactory(),

insecure/fixtures.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/model/libp2p/message"
 	"github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/network/channels"
+	flownetmsg "github.com/onflow/flow-go/network/message"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -40,7 +41,7 @@ func EgressMessageFixture(t *testing.T, codec network.Codec, protocol Protocol,
 	// encodes event to create payload
 	payload, err := codec.Encode(content)
 	require.NoError(t, err)
-	eventIDHash, err := network.EventId(channel, payload)
+	eventIDHash, err := flownetmsg.EventId(channel, payload)
 	require.NoError(t, err)
 
 	eventID := flow.HashToID(eventIDHash)

insecure/integration/functional/test/gossipsub/rpc_inspector/validation_inspector_test.go

@@ -1228,9 +1228,18 @@ func TestGossipSubSpamMitigationIntegration(t *testing.T) {
 
 	// now we expect the detection and mitigation to kick in and the victim node to disconnect from the spammer node.
 	// so the spammer and victim nodes should not be able to exchange messages on the topic.
-	p2ptest.EnsureNoPubsubExchangeBetweenGroups(t, ctx, []p2p.LibP2PNode{victimNode}, []p2p.LibP2PNode{spammer.SpammerNode}, blockTopic, 1, func() interface{} {
-		return unittest.ProposalFixture()
-	})
+	p2ptest.EnsureNoPubsubExchangeBetweenGroups(
+		t,
+		ctx,
+		[]p2p.LibP2PNode{victimNode},
+		flow.IdentifierList{victimId.NodeID},
+		[]p2p.LibP2PNode{spammer.SpammerNode},
+		flow.IdentifierList{spammer.SpammerId.NodeID},
+		blockTopic,
+		1,
+		func() interface{} {
+			return unittest.ProposalFixture()
+		})
 }
 
 // mockDistributorReadyDoneAware mocks the Ready and Done methods of the distributor to return a channel that is already closed,

insecure/integration/functional/test/gossipsub/scoring/ihave_spam_test.go

@@ -302,9 +302,18 @@ func TestGossipSubIHaveBrokenPromises_Above_Threshold(t *testing.T) {
 	require.Lessf(t, spammerScore, scoring.DefaultGraylistThreshold, "sanity check failed, the score of the spammer node must be less than graylist threshold: %f, actual: %f", scoring.DefaultGraylistThreshold, spammerScore)
 
 	// since the spammer score is below the gossip, graylist and publish thresholds, it should not be able to exchange messages with victim anymore.
-	p2ptest.EnsureNoPubsubExchangeBetweenGroups(t, ctx, []p2p.LibP2PNode{spammer.SpammerNode}, []p2p.LibP2PNode{victimNode}, blockTopic, 1, func() interface{} {
-		return unittest.ProposalFixture()
-	})
+	p2ptest.EnsureNoPubsubExchangeBetweenGroups(
+		t,
+		ctx,
+		[]p2p.LibP2PNode{spammer.SpammerNode},
+		flow.IdentifierList{spammer.SpammerId.NodeID},
+		[]p2p.LibP2PNode{victimNode},
+		flow.IdentifierList{victimIdentity.NodeID},
+		blockTopic,
+		1,
+		func() interface{} {
+			return unittest.ProposalFixture()
+		})
 }
 
 // spamIHaveBrokenPromises is a test utility function that is exclusive for the TestGossipSubIHaveBrokenPromises tests.

insecure/integration/functional/test/gossipsub/scoring/scoring_test.go

@@ -173,9 +173,18 @@ func testGossipSubInvalidMessageDeliveryScoring(t *testing.T, spamMsgFactory fun
 	// ensure that the topic snapshot of the spammer contains a record of at least (60%) of the spam messages sent. The 60% is to account for the messages that were delivered before the score was updated, after the spammer is PRUNED, as well as to account for decay.
 	require.True(t, blkTopicSnapshot.InvalidMessageDeliveries > 0.6*float64(totalSpamMessages), "invalid message deliveries must be greater than %f. invalid message deliveries: %f", 0.9*float64(totalSpamMessages), blkTopicSnapshot.InvalidMessageDeliveries)
 
-	p2ptest.EnsureNoPubsubExchangeBetweenGroups(t, ctx, []p2p.LibP2PNode{victimNode}, []p2p.LibP2PNode{spammer.SpammerNode}, blockTopic, 1, func() interface{} {
-		return unittest.ProposalFixture()
-	})
+	p2ptest.EnsureNoPubsubExchangeBetweenGroups(
+		t,
+		ctx,
+		[]p2p.LibP2PNode{victimNode},
+		flow.IdentifierList{victimIdentity.NodeID},
+		[]p2p.LibP2PNode{spammer.SpammerNode},
+		flow.IdentifierList{spammer.SpammerId.NodeID},
+		blockTopic,
+		1,
+		func() interface{} {
+			return unittest.ProposalFixture()
+		})
 }
 
 // TestGossipSubMeshDeliveryScoring_UnderDelivery_SingleTopic tests that when a peer is under-performing in a topic mesh, its score is (slightly) penalized.
@@ -474,7 +483,7 @@ func TestGossipSubMeshDeliveryScoring_Replay_Will_Not_Counted(t *testing.T) {
 		proposalList[i] = unittest.ProposalFixture()
 	}
 	i := -1
-	p2ptest.EnsurePubsubMessageExchangeFromNode(t, ctx, replayingNode, thisNode, blockTopic, len(proposalList), func() interface{} {
+	p2ptest.EnsurePubsubMessageExchangeFromNode(t, ctx, replayingNode, thisNode, thisId.NodeID, blockTopic, len(proposalList), func() interface{} {
 		i += 1
 		return proposalList[i]
 	})
@@ -500,10 +509,18 @@ func TestGossipSubMeshDeliveryScoring_Replay_Will_Not_Counted(t *testing.T) {
 
 	// now the replaying node acts maliciously and just replays the same messages again.
 	i = -1
-	p2ptest.EnsureNoPubsubMessageExchange(t, ctx, []p2p.LibP2PNode{replayingNode}, []p2p.LibP2PNode{thisNode}, blockTopic, len(proposalList), func() interface{} {
-		i += 1
-		return proposalList[i]
-	})
+	p2ptest.EnsureNoPubsubMessageExchange(
+		t,
+		ctx,
+		[]p2p.LibP2PNode{replayingNode},
+		[]p2p.LibP2PNode{thisNode},
+		flow.IdentifierList{thisId.NodeID},
+		blockTopic,
+		len(proposalList),
+		func() interface{} {
+			i += 1
+			return proposalList[i]
+		})
 
 	// since the last decay interval, the replaying node has not delivered anything new, so its score should be penalized for under-performing.
 	require.Eventually(t, func() bool {

insecure/orchestrator/network.go

@@ -13,6 +13,7 @@ import (
 	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/network/channels"
+	flownetmsg "github.com/onflow/flow-go/network/message"
 	"github.com/onflow/flow-go/utils/logging"
 )
 
@@ -159,7 +160,7 @@ func (on *Network) processEgressMessage(message *insecure.EgressMessage) error {
 
 	channel := channels.Channel(message.ChannelID)
 
-	egressEventIDHash, err := network.EventId(channel, message.Payload)
+	egressEventIDHash, err := flownetmsg.EventId(channel, message.Payload)
 	if err != nil {
 		return fmt.Errorf("could not create egress event ID: %w", err)
 	}
@@ -205,7 +206,7 @@ func (on *Network) processIngressMessage(message *insecure.IngressMessage) error
 	defer on.orchestratorMutex.Unlock()
 
 	channel := channels.Channel(message.ChannelID)
-	ingressEventIDHash, err := network.EventId(channel, message.Payload)
+	ingressEventIDHash, err := flownetmsg.EventId(channel, message.Payload)
 	if err != nil {
 		return fmt.Errorf("could not create ingress event ID: %w", err)
 	}

network/alsp/manager/manager_test.go

@@ -63,7 +63,7 @@ func TestNetworkPassesReportedMisbehavior(t *testing.T) {
 		testutils.MiddlewareConfigFixture(t, sporkId),
 		mocknetwork.NewViolationsConsumer(t))
 
-	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0])
+	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, sporkId, mws[0])
 	net, err := p2p.NewNetwork(networkCfg, p2p.WithAlspManager(misbehaviorReportManger))
 	require.NoError(t, err)
 
@@ -126,7 +126,7 @@ func TestHandleReportedMisbehavior_Cache_Integration(t *testing.T) {
 		nodes,
 		testutils.MiddlewareConfigFixture(t, sporkId),
 		mocknetwork.NewViolationsConsumer(t))
-	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
+	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, sporkId, mws[0], p2p.WithAlspConfig(cfg))
 	net, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
 
@@ -230,7 +230,7 @@ func TestHandleReportedMisbehavior_And_DisallowListing_Integration(t *testing.T)
 		nodes,
 		testutils.MiddlewareConfigFixture(t, sporkId),
 		mocknetwork.NewViolationsConsumer(t))
-	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
+	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, sporkId, mws[0], p2p.WithAlspConfig(cfg))
 	victimNetwork, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
 
@@ -278,7 +278,7 @@ func TestHandleReportedMisbehavior_And_DisallowListing_Integration(t *testing.T)
 	unittest.RequireReturnsBefore(t, wg.Wait, 100*time.Millisecond, "not all misbehavior reports have been processed")
 
 	// ensures that the spammer is disallow-listed by the victim
-	p2ptest.RequireEventuallyNotConnected(t, []p2p.LibP2PNode{nodes[victimIndex]}, []p2p.LibP2PNode{nodes[spammerIndex]}, 100*time.Millisecond, 3*time.Second)
+	p2ptest.RequireEventuallyNotConnected(t, []p2p.LibP2PNode{nodes[victimIndex]}, []p2p.LibP2PNode{nodes[spammerIndex]}, 100*time.Millisecond, 5*time.Second)
 
 	// despite disallow-listing spammer, ensure that (victim and honest) and (honest and spammer) are still connected.
 	p2ptest.RequireConnectedEventually(t, []p2p.LibP2PNode{nodes[spammerIndex], nodes[honestIndex]}, 1*time.Millisecond, 100*time.Millisecond)
@@ -323,7 +323,7 @@ func TestHandleReportedMisbehavior_And_DisallowListing_RepeatOffender_Integratio
 	ids, nodes := testutils.LibP2PNodeForMiddlewareFixture(t, sporkId, 3,
 		p2ptest.WithPeerManagerEnabled(p2ptest.PeerManagerConfigFixture(p2ptest.WithZeroJitterAndZeroBackoff(t)), nil))
 	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t, sporkId), mocknetwork.NewViolationsConsumer(t))
-	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
+	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, sporkId, mws[0], p2p.WithAlspConfig(cfg))
 
 	victimNetwork, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
@@ -482,6 +482,7 @@ func TestHandleReportedMisbehavior_And_SlashingViolationsConsumer_Integration(t
 		t,
 		*ids[0],
 		ids,
+		sporkId,
 		mws[0],
 		p2p.WithAlspConfig(managerCfgFixture(t)))
 	victimNetwork, err := p2p.NewNetwork(networkCfg)
@@ -575,7 +576,7 @@ func TestMisbehaviorReportMetrics(t *testing.T) {
 	sporkId := unittest.IdentifierFixture()
 	ids, nodes := testutils.LibP2PNodeForMiddlewareFixture(t, sporkId, 1)
 	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t, sporkId), mocknetwork.NewViolationsConsumer(t))
-	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
+	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, sporkId, mws[0], p2p.WithAlspConfig(cfg))
 	net, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
 

network/cache/rcvcache_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/onflow/flow-go/network"
+	"github.com/onflow/flow-go/network/message"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -43,28 +43,28 @@ func (r *ReceiveCacheTestSuite) SetupTest() {
 
 // TestSingleElementAdd adds a single element to the cache and verifies its existence.
 func (r *ReceiveCacheTestSuite) TestSingleElementAdd() {
-	eventID, err := network.EventId(channels.Channel("0"), []byte("event-1"))
+	eventID, err := message.EventId(channels.Channel("0"), []byte("event-1"))
 	require.NoError(r.T(), err)
 
 	assert.True(r.Suite.T(), r.c.Add(eventID))
 	assert.False(r.Suite.T(), r.c.Add(eventID))
 
 	// same channel but different event should be treated as unseen
-	eventID2, err := network.EventId(channels.Channel("0"), []byte("event-2"))
+	eventID2, err := message.EventId(channels.Channel("0"), []byte("event-2"))
 	require.NoError(r.T(), err)
 	assert.True(r.Suite.T(), r.c.Add(eventID2))
 	assert.False(r.Suite.T(), r.c.Add(eventID2))
 
 	// same event but different channels should be treated as unseen
-	eventID3, err := network.EventId(channels.Channel("1"), []byte("event-2"))
+	eventID3, err := message.EventId(channels.Channel("1"), []byte("event-2"))
 	require.NoError(r.T(), err)
 	assert.True(r.Suite.T(), r.c.Add(eventID3))
 	assert.False(r.Suite.T(), r.c.Add(eventID3))
 }
 
 // TestNoneExistence evaluates the correctness of cache operation against non-existing element
 func (r *ReceiveCacheTestSuite) TestNoneExistence() {
-	eventID, err := network.EventId(channels.Channel("1"), []byte("non-existing event"))
+	eventID, err := message.EventId(channels.Channel("1"), []byte("non-existing event"))
 	require.NoError(r.T(), err)
 
 	// adding new event to cache should return true
@@ -76,7 +76,7 @@ func (r *ReceiveCacheTestSuite) TestMultipleElementAdd() {
 	// creates and populates slice of 10 events
 	eventIDs := make([]hash.Hash, 0)
 	for i := 0; i < r.size; i++ {
-		eventID, err := network.EventId(channels.Channel("1"), []byte(fmt.Sprintf("event-%d", i)))
+		eventID, err := message.EventId(channels.Channel("1"), []byte(fmt.Sprintf("event-%d", i)))
 		require.NoError(r.T(), err)
 
 		eventIDs = append(eventIDs, eventID)
@@ -114,7 +114,7 @@ func (r *ReceiveCacheTestSuite) TestLRU() {
 	eventIDs := make([]hash.Hash, 0)
 	total := r.size + 1
 	for i := 0; i < total; i++ {
-		eventID, err := network.EventId(channels.Channel("1"), []byte(fmt.Sprintf("event-%d", i)))
+		eventID, err := message.EventId(channels.Channel("1"), []byte(fmt.Sprintf("event-%d", i)))
 		require.NoError(r.T(), err)
 
 		eventIDs = append(eventIDs, eventID)

network/internal/testutils/testUtil.go

@@ -154,7 +154,7 @@ func MiddlewareConfigFixture(t *testing.T, sporkId flow.Identifier) *middleware.
 	return &middleware.Config{
 		Logger:                unittest.Logger(),
 		BitSwapMetrics:        metrics.NewNoopCollector(),
-		RootBlockID:           sporkId,
+		SporkId:               sporkId,
 		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
 		Codec:                 unittest.NetworkCodec(),
 	}
@@ -191,6 +191,7 @@ func MiddlewareFixtures(t *testing.T, identities flow.IdentityList, libP2PNodes
 
 // NetworksFixture generates the network for the given middlewares
 func NetworksFixture(t *testing.T,
+	sporkId flow.Identifier,
 	ids flow.IdentityList,
 	mws []network.Middleware) []network.Network {
 
@@ -199,7 +200,7 @@ func NetworksFixture(t *testing.T,
 
 	for i := 0; i < count; i++ {
 
-		params := NetworkConfigFixture(t, *ids[i], ids, mws[i])
+		params := NetworkConfigFixture(t, *ids[i], ids, sporkId, mws[i])
 		net, err := p2p.NewNetwork(params)
 		require.NoError(t, err)
 
@@ -213,6 +214,7 @@ func NetworkConfigFixture(
 	t *testing.T,
 	myId flow.Identity,
 	allIds flow.IdentityList,
+	sporkId flow.Identifier,
 	mw network.Middleware,
 	opts ...p2p.NetworkConfigOption) *p2p.NetworkConfig {
 
@@ -240,6 +242,7 @@ func NetworkConfigFixture(
 		IdentityProvider:    id.NewFixedIdentityProvider(allIds),
 		ReceiveCache:        receiveCache,
 		ConduitFactory:      conduit.NewDefaultConduitFactory(),
+		SporkId:             sporkId,
 		AlspCfg: &alspmgr.MisbehaviorReportManagerConfig{
 			Logger:                  unittest.Logger(),
 			SpamRecordCacheSize:     defaultFlowConfig.NetworkConfig.AlspConfig.SpamRecordCacheSize,