[Testing] Fix data race in cluster-prefix RecordCache

consensus/hotstuff/votecollector/statemachine_test.go

@@ -484,6 +484,58 @@ func (s *StateMachineTestSuite) TestRegisterVoteConsumer() {
 	require.Equal(s.T(), expectedVotes, actualVotes)
 }
 
+// TestRegisterVoteConsumer_CachingToVerifyingTransition verifies that a vote consumer registered
+// while the collector is in the caching state continues to receive votes, in order of arrival,
+// across the transition to the verifying state. The proposer's vote is cached synchronously by
+// `ProcessBlock` (via `ensureVoteUnique`), so the consumer receives it at a deterministic
+// position: between the votes added before and after the transition.
+func (s *StateMachineTestSuite) TestRegisterVoteConsumer_CachingToVerifyingTransition() {
+	votes := 5
+	proposal := makeSignedProposalWithView(s.view)
+	block := proposal.Block
+	processor := s.prepareMockedProcessor(proposal, 1)
+	proposerVote, err := proposal.ProposerVote()
+	require.NoError(s.T(), err)
+
+	actualVotes := make([]*model.Vote, 0)
+	s.collector.RegisterVoteConsumer(func(vote *model.Vote) {
+		actualVotes = append(actualVotes, vote)
+	})
+
+	expectedVotes := make([]*model.Vote, 0)
+	// Add votes while the collector is in the caching state. Each vote emits `OnVoteProcessed`
+	// twice: once when cached, and once when replayed into the verifying processor during the
+	// transition below (consistent with `TestProcessBlock_ProcessingOfCachedVotes`).
+	for i := 0; i < votes; i++ {
+		vote := unittest.VoteForBlockFixture(block)
+		s.notifier.On("OnVoteProcessed", vote).Twice()
+		processor.On("Process", vote).Return(nil).Once()
+		require.NoError(s.T(), s.collector.AddVote(vote))
+		expectedVotes = append(expectedVotes, vote)
+	}
+
+	// `ProcessBlock` caches the proposer's vote (delivering it to the consumer) and transitions
+	// the collector to the verifying state
+	require.NoError(s.T(), s.collector.ProcessBlock(proposal))
+	require.Equal(s.T(), hotstuff.VoteCollectorStatusVerifying, s.collector.Status())
+	expectedVotes = append(expectedVotes, proposerVote)
+
+	// add more votes after the transition; each is processed exactly once
+	for i := 0; i < votes; i++ {
+		vote := unittest.VoteForBlockFixture(block)
+		s.notifier.On("OnVoteProcessed", vote).Once()
+		processor.On("Process", vote).Return(nil).Once()
+		require.NoError(s.T(), s.collector.AddVote(vote))
+		expectedVotes = append(expectedVotes, vote)
+	}
+
+	// The cache invokes consumers synchronously in the goroutine adding the vote, and all votes
+	// are added on this test's goroutine, so `actualVotes` is complete at this point. The
+	// asynchronous replay of cached votes into the verifying processor does not re-deliver
+	// votes to consumers.
+	require.Equal(s.T(), expectedVotes, actualVotes)
+}
+
 func makeSignedProposalWithView(view uint64) *model.SignedProposal {
 	return helper.MakeSignedProposal(helper.WithProposal(helper.MakeProposal(helper.WithBlock(helper.MakeBlock(helper.WithBlockView(view))))))
 }

network/p2p/builder/libp2pscaler_test.go

@@ -144,9 +144,22 @@ func TestApplyResourceLimitOverride(t *testing.T) {
 		Memory:              9870,
 	}
 
+	// Flow's override config treats every non-positive value as "not overridden". In particular,
+	// negative values must NOT be passed through to rcmgr, where they carry special meaning
+	// (-1 = unlimited, -2 = block-all).
+	transientOverride := p2pconfig.ResourceManagerOverrideLimit{
+		StreamsInbound:      -1, // should not be overridden.
+		StreamsOutbound:     -2, // should not be overridden.
+		ConnectionsInbound:  -1, // should not be overridden.
+		ConnectionsOutbound: -2, // should not be overridden.
+		FD:                  -1, // should not be overridden.
+		Memory:              -2, // should not be overridden.
+	}
+
 	partial := rcmgr.PartialLimitConfig{}
 	partial.System = ApplyResourceLimitOverride(unittest.Logger(), p2pconfig.ResourceScopeSystem, scaled.ToPartialLimitConfig().System, systemOverride)
 	partial.PeerDefault = ApplyResourceLimitOverride(unittest.Logger(), p2pconfig.ResourceScopePeer, scaled.ToPartialLimitConfig().PeerDefault, peerOverride)
+	partial.Transient = ApplyResourceLimitOverride(unittest.Logger(), p2pconfig.ResourceScopeTransient, scaled.ToPartialLimitConfig().Transient, transientOverride)
 
 	final := partial.Build(scaled).ToPartialLimitConfig()
 	require.Equal(t, 456, int(final.System.StreamsOutbound))                                           // should be overridden.
@@ -155,6 +168,48 @@ func TestApplyResourceLimitOverride(t *testing.T) {
 	require.Equal(t, 7890, int(final.System.Memory))                                                   // should be overridden.
 	require.Equal(t, scaled.ToPartialLimitConfig().System.StreamsInbound, final.System.StreamsInbound) // should NOT be overridden.
 	require.Equal(t, scaled.ToPartialLimitConfig().System.ConnsOutbound, final.System.ConnsOutbound)   // should NOT be overridden.
+
+	// none of the negative values may take effect - the entire transient scope keeps its defaults.
+	require.Equal(t, scaled.ToPartialLimitConfig().Transient, final.Transient)
+}
+
+// TestBuildLibp2pResourceManagerLimits_NonPositiveOverridesKeepDefaults tests that overrides with
+// a mix of positive, zero, and negative values are merged correctly end-to-end through
+// BuildLibp2pResourceManagerLimits: positive values override, non-positive values keep the
+// scaled defaults.
+func TestBuildLibp2pResourceManagerLimits_NonPositiveOverridesKeepDefaults(t *testing.T) {
+	cfg, err := config.DefaultConfig()
+	require.NoError(t, err)
+
+	// Build the baseline limits with an explicitly zeroed override (the shipped default config
+	// contains non-zero overrides, e.g. system streams-inbound), so that the baseline reflects
+	// the pure scaled defaults that non-positive override values must fall back to.
+	cfg.NetworkConfig.ResourceManager.Override = p2pconfig.ResourceManagerOverrideScope{}
+	defaultConcreteLimits, err := BuildLibp2pResourceManagerLimits(unittest.Logger(), &cfg.NetworkConfig.ResourceManager)
+	require.NoError(t, err)
+	defaultSystem := defaultConcreteLimits.ToPartialLimitConfig().System
+
+	streamsOutboundOverride := int(defaultSystem.StreamsOutbound) + 42
+	fdOverride := int(defaultSystem.FD) + 42
+	cfg.NetworkConfig.ResourceManager.Override.System = p2pconfig.ResourceManagerOverrideLimit{
+		StreamsInbound:      -1,                      // should not be overridden.
+		StreamsOutbound:     streamsOutboundOverride, // should be overridden.
+		ConnectionsInbound:  0,                       // should not be overridden.
+		ConnectionsOutbound: -2,                      // should not be overridden.
+		FD:                  fdOverride,              // should be overridden.
+		Memory:              0,                       // should not be overridden.
+	}
+
+	overriddenConcreteLimits, err := BuildLibp2pResourceManagerLimits(unittest.Logger(), &cfg.NetworkConfig.ResourceManager)
+	require.NoError(t, err)
+
+	actualSystem := overriddenConcreteLimits.ToPartialLimitConfig().System
+	require.Equal(t, streamsOutboundOverride, int(actualSystem.StreamsOutbound)) // should be overridden.
+	require.Equal(t, fdOverride, int(actualSystem.FD))                           // should be overridden.
+	require.Equal(t, defaultSystem.StreamsInbound, actualSystem.StreamsInbound)  // should NOT be overridden.
+	require.Equal(t, defaultSystem.ConnsInbound, actualSystem.ConnsInbound)      // should NOT be overridden.
+	require.Equal(t, defaultSystem.ConnsOutbound, actualSystem.ConnsOutbound)    // should NOT be overridden.
+	require.Equal(t, defaultSystem.Memory, actualSystem.Memory)                  // should NOT be overridden.
 }
 
 // TestBuildLibp2pResourceManagerLimits tests the BuildLibp2pResourceManagerLimits function.

network/p2p/inspector/internal/cache/cache.go

@@ -75,15 +75,21 @@ func NewRecordCache(config *RecordCacheConfig, recordFactory recordFactory) (*Re
 //   - exception only in cases of internal data inconsistency or bugs. No errors are expected.
 func (r *RecordCache) ReceivedClusterPrefixedMessage(pid peer.ID) (float64, error) {
 	var err error
+	var gauge float64
 	adjustFunc := func(record *ClusterPrefixedMessagesReceivedRecord) *ClusterPrefixedMessagesReceivedRecord {
 		record, err = r.decayAdjustment(record) // first decay the record
 		if err != nil {
 			return record
 		}
-		return r.incrementAdjustment(record) // then increment the record
+		record = r.incrementAdjustment(record) // then increment the record
+		// Capture the gauge value while the cache holds its lock. Records are mutated in place, so
+		// reading the gauge from the returned record pointer after AdjustWithInit releases the lock
+		// would race with concurrent adjustments of the same record.
+		gauge = record.Gauge
+		return record
 	}
 	nodeID := p2p.MakeId(pid)
-	adjustedRecord, adjusted := r.c.AdjustWithInit(nodeID, adjustFunc, func() *ClusterPrefixedMessagesReceivedRecord {
+	_, adjusted := r.c.AdjustWithInit(nodeID, adjustFunc, func() *ClusterPrefixedMessagesReceivedRecord {
 		return r.recordFactory(nodeID)
 	})
 
@@ -95,7 +101,7 @@ func (r *RecordCache) ReceivedClusterPrefixedMessage(pid peer.ID) (float64, erro
 		return 0, fmt.Errorf("adjustment failed for peer %s", pid)
 	}
 
-	return adjustedRecord.Gauge, nil
+	return gauge, nil
 }
 
 // GetWithInit returns the current number of cluster prefixed control messages received from a peer.
@@ -109,13 +115,21 @@ func (r *RecordCache) ReceivedClusterPrefixedMessage(pid peer.ID) (float64, erro
 // No errors are expected during normal operation.
 func (r *RecordCache) GetWithInit(pid peer.ID) (float64, bool, error) {
 	var err error
+	var gauge float64
 	adjustLogic := func(record *ClusterPrefixedMessagesReceivedRecord) *ClusterPrefixedMessagesReceivedRecord {
 		// perform decay on gauge value
 		record, err = r.decayAdjustment(record)
+		if err != nil {
+			return record
+		}
+		// Capture the gauge value while the cache holds its lock. Records are mutated in place, so
+		// reading the gauge from the returned record pointer after AdjustWithInit releases the lock
+		// would race with concurrent adjustments of the same record.
+		gauge = record.Gauge
 		return record
 	}
 	nodeID := p2p.MakeId(pid)
-	adjustedRecord, adjusted := r.c.AdjustWithInit(nodeID, adjustLogic, func() *ClusterPrefixedMessagesReceivedRecord {
+	_, adjusted := r.c.AdjustWithInit(nodeID, adjustLogic, func() *ClusterPrefixedMessagesReceivedRecord {
 		return r.recordFactory(nodeID)
 	})
 	if err != nil {
@@ -125,7 +139,7 @@ func (r *RecordCache) GetWithInit(pid peer.ID) (float64, bool, error) {
 		return 0, false, fmt.Errorf("decay adjustment failed for peer %s", pid)
 	}
 
-	return adjustedRecord.Gauge, true, nil
+	return gauge, true, nil
 }
 
 // Remove removes the record of the given peer id from the cache.

network/p2p/inspector/validation/control_message_validation_inspector_test.go

@@ -1461,6 +1461,55 @@ func TestNewControlMsgValidationInspector_validateClusterPrefixedTopic(t *testin
 		cancel()
 		unittest.RequireCloseBefore(t, inspector.Done(), 5*time.Second, "inspector did not stop")
 	})
+
+	t.Run("validateClusterPrefixedTopic with multiple workers disseminates between 1 and one-per-message notifications past the hard threshold", func(t *testing.T) {
+		// total number of inspected messages; with a hard threshold of 10, the last 5 exceed it
+		const total = 15
+		logCounter := atomic.NewInt64(0)
+		logger := hookedLogger(logCounter, zerolog.TraceLevel, worker.QueuedItemProcessedLog)
+		notificationCount := atomic.NewInt64(0)
+		inspector, signalerCtx, cancel, consumer, rpcTracker, sporkID, idProvider, topicProviderOracle := inspectorFixture(t, func(params *validation.InspectorParams) {
+			params.Config.ClusterPrefixedMessage.HardThreshold = 10
+			params.Config.GraftPrune.InvalidTopicIdThreshold = 0
+			// This test deliberately keeps the default (multi-)worker count and documents the
+			// concurrency-tolerant contract: the cluster-prefix tracker increment and the subsequent
+			// hard-threshold check are not atomic, so each of the `total` concurrent inspections may
+			// observe the tracker above the hard threshold and disseminate a notification. At least
+			// one notification is guaranteed (the globally last inspection observes all increments),
+			// and at most one notification per inspected message can be disseminated.
+			params.Logger = logger
+		})
+		clusterID := flow.ChainID(unittest.IdentifierFixture().String())
+		clusterPrefixedTopic := channels.Topic(fmt.Sprintf("%s/%s", channels.SyncCluster(clusterID), sporkID)).String()
+		topicProviderOracle.UpdateTopics([]string{clusterPrefixedTopic})
+		from := unittest.PeerIdFixture(t)
+		identity := unittest.IdentityFixture()
+		idProvider.On("ByPeerID", from).Return(identity, true).Times(total)
+		checkNotification := checkNotificationFunc(t, from, p2pmsg.CtrlMsgGraft, validation.IsInvalidTopicIDThresholdExceeded, p2p.CtrlMsgTopicTypeClusterPrefixed)
+		inspectMsgRpc := unittest.P2PRPCFixture(unittest.WithGrafts(unittest.P2PRPCGraftFixture(&clusterPrefixedTopic)))
+		inspector.ActiveClustersChanged(flow.ChainIDList{flow.ChainID(unittest.IdentifierFixture().String())})
+		rpcTracker.On("LastHighestIHaveRPCSize").Return(int64(100)).Maybe()
+		consumer.On("OnInvalidControlMessageNotification", mock.AnythingOfType("*p2p.InvCtrlMsgNotif")).Return(nil).Run(func(args mock.Arguments) {
+			notificationCount.Inc()
+			checkNotification(args)
+		})
+		inspector.Start(signalerCtx)
+		unittest.RequireComponentsReadyBefore(t, 1*time.Second, inspector)
+
+		for i := 0; i < total; i++ {
+			require.NoError(t, inspector.Inspect(from, inspectMsgRpc))
+		}
+		// wait until all inspections have been processed by the workers
+		require.Eventually(t, func() bool {
+			return logCounter.Load() == total
+		}, time.Second, 100*time.Millisecond)
+
+		count := notificationCount.Load()
+		require.GreaterOrEqual(t, count, int64(1), "at least one inspection past the hard threshold must disseminate a notification")
+		require.LessOrEqual(t, count, int64(total), "at most one notification per inspected message may be disseminated")
+		cancel()
+		unittest.RequireCloseBefore(t, inspector.Done(), 5*time.Second, "inspector did not stop")
+	})
 }
 
 // TestControlMessageValidationInspector_ActiveClustersChanged validates the expected update of the active cluster IDs list.