[Networking] Optimizing GossipSub RPC Handling Memory Usage Through Asynchronous Subscription Updates

cmd/access/node_builder/access_node_builder.go

@@ -1563,6 +1563,7 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 			UpdateInterval:    builder.FlowConfig.NetworkConfig.PeerUpdateInterval,
 			ConnectorFactory:  connection.DefaultLibp2pBackoffConnectorFactory(),
 		},
+		&builder.FlowConfig.NetworkConfig.GossipSubConfig.SubscriptionProviderConfig,
 		&p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),

cmd/observer/node_builder/observer_builder.go

@@ -722,6 +722,7 @@ func (builder *ObserverServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		&builder.FlowConfig.NetworkConfig.ResourceManager,
 		&builder.FlowConfig.NetworkConfig.GossipSubConfig.GossipSubRPCInspectorsConfig,
 		p2pconfig.PeerManagerDisableConfig(), // disable peer manager for observer node.
+		&builder.FlowConfig.NetworkConfig.GossipSubConfig.SubscriptionProviderConfig,
 		&p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),

config/default-config.yml

@@ -134,6 +134,15 @@ network-config:
   gossipsub-rpc-sent-tracker-workers: 5
   # Peer scoring is the default value for enabling peer scoring
   gossipsub-peer-scoring-enabled: true
+  # The interval for updating the list of subscribed peers to all topics in gossipsub. This is used to keep track of subscriptions
+  # violations and penalize peers accordingly. Recommended value is in the order of a few minutes to avoid contentions; as the operation
+  # reads all topics and all peers subscribed to each topic.
+  gossipsub-subscription-provider-update-interval: 10m
+  # The size of cache for keeping the list of all peers subscribed to each topic (same as the local node). This cache is the local node's
+  # view of the network and is used to detect subscription violations and penalize peers accordingly. Recommended to be big enough to
+  # keep the entire network's size. Otherwise, the local node's view of the network will be incomplete due to cache eviction.
+  # Recommended size is 10x the number of peers in the network.
+  gossipsub-subscription-provider-cache-size: 10000
 
   # Gossipsub rpc inspectors configs
   # The size of the queue for notifications about invalid RPC messages

follower/follower_builder.go

@@ -600,6 +600,7 @@ func (builder *FollowerServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		&builder.FlowConfig.NetworkConfig.ResourceManager,
 		&builder.FlowConfig.NetworkConfig.GossipSubConfig.GossipSubRPCInspectorsConfig,
 		p2pconfig.PeerManagerDisableConfig(), // disable peer manager for follower
+		&builder.FlowConfig.NetworkConfig.GossipSubConfig.SubscriptionProviderConfig,
 		&p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),

insecure/Makefile

@@ -12,3 +12,18 @@ endif
 .PHONY: test
 test:
 	go test $(if $(VERBOSE),-v,) -coverprofile=$(COVER_PROFILE) $(RACE_FLAG) $(if $(JSON_OUTPUT),-json,) $(if $(NUM_RUNS),-count $(NUM_RUNS),) --tags relic ./...
+
+.PHONY: lint
+lint: tidy
+	# revive -config revive.toml -exclude storage/ledger/trie ./...
+	golangci-lint run -v --build-tags relic ./...
+
+# this ensures there is no unused dependency being added by accident
+.PHONY: tidy
+tidy:
+	go mod tidy -v
+	cd integration; go mod tidy -v
+	cd crypto; go mod tidy -v
+	cd cmd/testclient; go mod tidy -v
+	cd insecure; go mod tidy -v
+	git diff --exit-code

module/metrics/herocache.go

@@ -72,6 +72,10 @@ func NetworkReceiveCacheMetricsFactory(f HeroCacheMetricsFactory, networkType ne
 	return f(namespaceNetwork, r)
 }
 
+func NewSubscriptionRecordCacheMetricsFactory(f HeroCacheMetricsFactory) module.HeroCacheMetrics {
+	return f(namespaceNetwork, ResourceNetworkingSubscriptionRecordsCache)
+}
+
 // DisallowListCacheMetricsFactory is the factory method for creating a new HeroCacheCollector for the disallow list cache.
 // The disallow-list cache is used to keep track of peers that are disallow-listed and the reasons for it.
 // Args:

module/metrics/labels.go

@@ -85,6 +85,7 @@ const (
 	ResourceEpochCommit                                = "epoch_commit"
 	ResourceEpochStatus                                = "epoch_status"
 	ResourceNetworkingReceiveCache                     = "networking_received_message" // networking layer
+	ResourceNetworkingSubscriptionRecordsCache         = "subscription_records_cache"  // networking layer
 	ResourceNetworkingDnsIpCache                       = "networking_dns_ip_cache"     // networking layer
 	ResourceNetworkingDnsTxtCache                      = "networking_dns_txt_cache"    // networking layer
 	ResourceNetworkingDisallowListNotificationQueue    = "networking_disallow_list_notification_queue"

network/alsp/internal/cache.go

@@ -81,7 +81,6 @@ func (s *SpamRecordCache) Adjust(originId flow.Identifier, adjustFunc model.Reco
 	penalty, err := s.adjust(originId, adjustFunc)
 
 	switch {
-
 	case err == ErrSpamRecordNotFound:
 		// if the record does not exist, we initialize the record and try to adjust it again.
 		// Note: there is an edge case where the record is initialized by another goroutine between the two calls.

network/internal/p2pfixtures/fixtures.go

@@ -126,6 +126,7 @@ func CreateNode(t *testing.T, networkKey crypto.PrivateKey, sporkID flow.Identif
 		&defaultFlowConfig.NetworkConfig.ResourceManager,
 		&defaultFlowConfig.NetworkConfig.GossipSubRPCInspectorsConfig,
 		p2pconfig.PeerManagerDisableConfig(),
+		&defaultFlowConfig.NetworkConfig.GossipSubConfig.SubscriptionProviderConfig,
 		&p2p.DisallowListCacheConfig{
 			MaxSize: uint32(1000),
 			Metrics: metrics.NewNoopCollector(),

network/netconf/flags.go

@@ -60,6 +60,9 @@ const (
 	rpcSentTrackerNumOfWorkers   = "gossipsub-rpc-sent-tracker-workers"
 	scoreTracerInterval          = "gossipsub-score-tracer-interval"
 
+	gossipSubSubscriptionProviderUpdateInterval = "gossipsub-subscription-provider-update-interval"
+	gossipSubSubscriptionProviderCacheSize      = "gossipsub-subscription-provider-cache-size"
+
 	// gossipsub validation inspector
 	gossipSubRPCInspectorNotificationCacheSize                 = "gossipsub-rpc-inspector-notification-cache-size"
 	validationInspectorNumberOfWorkers                         = "gossipsub-rpc-validation-inspector-workers"
@@ -171,9 +174,13 @@ func AllFlagNames() []string {
 func InitializeNetworkFlags(flags *pflag.FlagSet, config *Config) {
 	flags.Bool(networkingConnectionPruning, config.NetworkConnectionPruning, "enabling connection trimming")
 	flags.Duration(dnsCacheTTL, config.DNSCacheTTL, "time-to-live for dns cache")
-	flags.StringSlice(preferredUnicastsProtocols, config.PreferredUnicastProtocols, "preferred unicast protocols in ascending order of preference")
+	flags.StringSlice(
+		preferredUnicastsProtocols, config.PreferredUnicastProtocols, "preferred unicast protocols in ascending order of preference")
 	flags.Uint32(receivedMessageCacheSize, config.NetworkReceivedMessageCacheSize, "incoming message cache size at networking layer")
-	flags.Uint32(disallowListNotificationCacheSize, config.DisallowListNotificationCacheSize, "cache size for notification events from disallow list")
+	flags.Uint32(
+		disallowListNotificationCacheSize,
+		config.DisallowListNotificationCacheSize,
+		"cache size for notification events from disallow list")
 	flags.Duration(peerUpdateInterval, config.PeerUpdateInterval, "how often to refresh the peer connections for the node")
 	flags.Duration(unicastMessageTimeout, config.UnicastMessageTimeout, "how long a unicast transmission can take to complete")
 	// unicast manager options
@@ -209,10 +216,22 @@ func InitializeNetworkFlags(flags *pflag.FlagSet, config *Config) {
 	flags.Duration(silencePeriod, config.ConnectionManagerConfig.SilencePeriod, "silence period for libp2p connection manager")
 	flags.Bool(peerScoring, config.GossipSubConfig.PeerScoring, "enabling peer scoring on pubsub network")
 	flags.Duration(localMeshLogInterval, config.GossipSubConfig.LocalMeshLogInterval, "logging interval for local mesh in gossipsub")
-	flags.Duration(scoreTracerInterval, config.GossipSubConfig.ScoreTracerInterval, "logging interval for peer score tracer in gossipsub, set to 0 to disable")
-	flags.Uint32(rpcSentTrackerCacheSize, config.GossipSubConfig.RPCSentTrackerCacheSize, "cache size of the rpc sent tracker used by the gossipsub mesh tracer.")
-	flags.Uint32(rpcSentTrackerQueueCacheSize, config.GossipSubConfig.RPCSentTrackerQueueCacheSize, "cache size of the rpc sent tracker worker queue.")
-	flags.Int(rpcSentTrackerNumOfWorkers, config.GossipSubConfig.RpcSentTrackerNumOfWorkers, "number of workers for the rpc sent tracker worker pool.")
+	flags.Duration(
+		scoreTracerInterval,
+		config.GossipSubConfig.ScoreTracerInterval,
+		"logging interval for peer score tracer in gossipsub, set to 0 to disable")
+	flags.Uint32(
+		rpcSentTrackerCacheSize,
+		config.GossipSubConfig.RPCSentTrackerCacheSize,
+		"cache size of the rpc sent tracker used by the gossipsub mesh tracer.")
+	flags.Uint32(
+		rpcSentTrackerQueueCacheSize,
+		config.GossipSubConfig.RPCSentTrackerQueueCacheSize,
+		"cache size of the rpc sent tracker worker queue.")
+	flags.Int(
+		rpcSentTrackerNumOfWorkers,
+		config.GossipSubConfig.RpcSentTrackerNumOfWorkers,
+		"number of workers for the rpc sent tracker worker pool.")
 	// gossipsub RPC control message validation limits used for validation configuration and rate limiting
 	flags.Int(validationInspectorNumberOfWorkers,
 		config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.NumberOfWorkers,
@@ -280,12 +299,15 @@ func InitializeNetworkFlags(flags *pflag.FlagSet, config *Config) {
 		config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IWantRPCInspectionConfig.DuplicateMsgIDThreshold,
 		"max allowed duplicate message IDs in a single iWant control message")
 
-	flags.Int(rpcMessageMaxSampleSize,
-		config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.RpcMessageMaxSampleSize,
-		"the max sample size used for RPC message validation. If the total number of RPC messages exceeds this value a sample will be taken but messages will not be truncated")
-	flags.Int(rpcMessageErrorThreshold,
-		config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.RpcMessageErrorThreshold,
-		"the threshold at which an error will be returned if the number of invalid RPC messages exceeds this value")
+	flags.Int(rpcMessageMaxSampleSize, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.RpcMessageMaxSampleSize, "the max sample size used for RPC message validation. If the total number of RPC messages exceeds this value a sample will be taken but messages will not be truncated")
+	flags.Int(rpcMessageErrorThreshold, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.RpcMessageErrorThreshold, "the threshold at which an error will be returned if the number of invalid RPC messages exceeds this value")
+	flags.Duration(
+		gossipSubSubscriptionProviderUpdateInterval, config.GossipSubConfig.SubscriptionProviderConfig.SubscriptionUpdateInterval,
+		"interval for updating the list of subscribed topics for all peers in the gossipsub, recommended value is a few minutes")
+	flags.Uint32(
+		gossipSubSubscriptionProviderCacheSize,
+		config.GossipSubConfig.SubscriptionProviderConfig.CacheSize,
+		"size of the cache that keeps the list of topics each peer has subscribed to, recommended size is 10x the number of authorized nodes")
 }
 
 // LoadLibP2PResourceManagerFlags loads all CLI flags for the libp2p resource manager configuration on the provided pflag set.
@@ -359,7 +381,8 @@ func SetAliases(conf *viper.Viper) error {
 	for _, flagName := range AllFlagNames() {
 		fullKey, ok := m[flagName]
 		if !ok {
-			return fmt.Errorf("invalid network configuration missing configuration key flag name %s check config file and cli flags", flagName)
+			return fmt.Errorf(
+				"invalid network configuration missing configuration key flag name %s check config file and cli flags", flagName)
 		}
 		conf.RegisterAlias(fullKey, flagName)
 	}

network/p2p/builder.go

@@ -110,6 +110,7 @@ type GossipSubRpcInspectorSuiteFactoryFunc func(
 	metrics.HeroCacheMetricsFactory,
 	flownet.NetworkingType,
 	module.IdentityProvider,
+	func() TopicProvider,
 ) (GossipSubInspectorSuite, error)
 
 // NodeBuilder is a builder pattern for creating a libp2p Node instance.

network/p2p/consumers.go

@@ -81,11 +81,4 @@ type GossipSubInspectorSuite interface {
 	// pattern where the consumer is notified when a new notification is published.
 	// A consumer is only notified once for each notification, and only receives notifications that were published after it was added.
 	AddInvalidControlMessageConsumer(GossipSubInvCtrlMsgNotifConsumer)
-
-	// SetTopicOracle sets the topic oracle of the gossipsub inspector suite.
-	// The topic oracle is used to determine the list of topics that the node is subscribed to.
-	// If an oracle is not set, the node will not be able to determine the list of topics that the node is subscribed to.
-	// This func is expected to be called once and will return an error on all subsequent calls.
-	// All errors returned from this func are considered irrecoverable.
-	SetTopicOracle(topicOracle func() []string) error
 }

network/p2p/inspector/internal/mockTopicProvider.go

@@ -0,0 +1,35 @@
+package internal
+
+import (
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+// MockUpdatableTopicProvider is a mock implementation of the TopicProvider interface.
+// TODO: this should be moved to a common package (e.g. network/p2p/test). Currently, it is not possible to do so because of a circular dependency.
+type MockUpdatableTopicProvider struct {
+	topics        []string
+	subscriptions map[string][]peer.ID
+}
+
+func NewMockUpdatableTopicProvider() *MockUpdatableTopicProvider {
+	return &MockUpdatableTopicProvider{
+		topics:        []string{},
+		subscriptions: map[string][]peer.ID{},
+	}
+}
+
+func (m *MockUpdatableTopicProvider) GetTopics() []string {
+	return m.topics
+}
+
+func (m *MockUpdatableTopicProvider) ListPeers(topic string) []peer.ID {
+	return m.subscriptions[topic]
+}
+
+func (m *MockUpdatableTopicProvider) UpdateTopics(topics []string) {
+	m.topics = topics
+}
+
+func (m *MockUpdatableTopicProvider) UpdateSubscriptions(topic string, peers []peer.ID) {
+	m.subscriptions[topic] = peers
+}

network/p2p/inspector/validation/control_message_validation_inspector.go

@@ -60,7 +60,7 @@ type ControlMsgValidationInspector struct {
 	// networkingType indicates public or private network, rpc publish messages are inspected for unstaked senders when running the private network.
 	networkingType network.NetworkingType
 	// topicOracle callback used to retrieve the current subscribed topics of the libp2p node.
-	topicOracle func() []string
+	topicOracle func() p2p.TopicProvider
 }
 
 type InspectorParams struct {
@@ -82,6 +82,9 @@ type InspectorParams struct {
 	RpcTracker p2p.RpcControlTracking `validate:"required"`
 	// NetworkingType the networking type of the node.
 	NetworkingType network.NetworkingType `validate:"required"`
+	// TopicOracle callback used to retrieve the current subscribed topics of the libp2p node.
+	// It is set as a callback to avoid circular dependencies between the topic oracle and the inspector.
+	TopicOracle func() p2p.TopicProvider `validate:"required"`
 }
 
 var _ component.Component = (*ControlMsgValidationInspector)(nil)
@@ -105,13 +108,18 @@ func NewControlMsgValidationInspector(params *InspectorParams) (*ControlMsgValid
 	inspectMsgQueueCacheCollector := metrics.GossipSubRPCInspectorQueueMetricFactory(params.HeroCacheMetricsFactory, params.NetworkingType)
 	clusterPrefixedCacheCollector := metrics.GossipSubRPCInspectorClusterPrefixedCacheMetricFactory(params.HeroCacheMetricsFactory, params.NetworkingType)
 
-	clusterPrefixedTracker, err := cache.NewClusterPrefixedMessagesReceivedTracker(params.Logger, params.Config.ClusterPrefixedControlMsgsReceivedCacheSize, clusterPrefixedCacheCollector, params.Config.ClusterPrefixedControlMsgsReceivedCacheDecay)
+	clusterPrefixedTracker, err := cache.NewClusterPrefixedMessagesReceivedTracker(params.Logger,
+		params.Config.ClusterPrefixedControlMsgsReceivedCacheSize,
+		clusterPrefixedCacheCollector,
+		params.Config.ClusterPrefixedControlMsgsReceivedCacheDecay)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create cluster prefix topics received tracker")
 	}
 
 	if params.Config.RpcMessageMaxSampleSize < params.Config.RpcMessageErrorThreshold {
-		return nil, fmt.Errorf("rpc message max sample size must be greater than or equal to rpc message error threshold, got %d and %d respectively", params.Config.RpcMessageMaxSampleSize, params.Config.RpcMessageErrorThreshold)
+		return nil, fmt.Errorf("rpc message max sample size must be greater than or equal to rpc message error threshold, got %d and %d respectively",
+			params.Config.RpcMessageMaxSampleSize,
+			params.Config.RpcMessageErrorThreshold)
 	}
 
 	c := &ControlMsgValidationInspector{
@@ -124,6 +132,7 @@ func NewControlMsgValidationInspector(params *InspectorParams) (*ControlMsgValid
 		idProvider:     params.IdProvider,
 		metrics:        params.InspectorMetrics,
 		networkingType: params.NetworkingType,
+		topicOracle:    params.TopicOracle,
 	}
 
 	store := queue.NewHeroStore(params.Config.CacheSize, params.Logger, inspectMsgQueueCacheCollector)
@@ -167,18 +176,6 @@ func (c *ControlMsgValidationInspector) ActiveClustersChanged(clusterIDList flow
 	c.tracker.StoreActiveClusterIds(clusterIDList)
 }
 
-// SetTopicOracle Sets the topic oracle. The topic oracle is used to determine the list of topics that the node is subscribed to.
-// If an oracle is not set, the node will not be able to determine the list of topics that the node is subscribed to.
-// This func is expected to be called once and will return an error on all subsequent calls.
-// All errors returned from this func are considered irrecoverable.
-func (c *ControlMsgValidationInspector) SetTopicOracle(topicOracle func() []string) error {
-	if c.topicOracle != nil {
-		return fmt.Errorf("topic oracle already set")
-	}
-	c.topicOracle = topicOracle
-	return nil
-}
-
 // Inspect is called by gossipsub upon reception of a rpc from a remote  node.
 // It creates a new InspectRPCRequest for the RPC to be inspected async by the worker pool.
 // Args:
@@ -469,7 +466,7 @@ func (c *ControlMsgValidationInspector) inspectRpcPublishMessages(from peer.ID,
 		messages[i], messages[j] = messages[j], messages[i]
 	})
 
-	subscribedTopics := c.topicOracle()
+	subscribedTopics := c.topicOracle().GetTopics()
 	hasSubscription := func(topic string) bool {
 		for _, subscribedTopic := range subscribedTopics {
 			if topic == subscribedTopic {