[Testing] Add regression tests for network data race fixes

network/alsp/internal/cache_test.go

@@ -813,3 +813,56 @@ func TestSpamRecordCache_ConcurrentIdentitiesAndOperations(t *testing.T) {
 
 	unittest.RequireReturnsBefore(t, wg.Wait, 1*time.Second, "timed out waiting for goroutines to finish")
 }
+
+// TestSpamRecordCache_ConcurrentGetWhileAdjustingSameRecord is a regression test for two data
+// races on the same record (run with -race):
+//  1. Get copied the record's fields after the backend lock was released, racing the in-place
+//     mutation performed by a concurrent adjustment of the same record.
+//  2. AdjustWithInit read the adjusted record's penalty after the lock was released, so the
+//     returned penalty could reflect a different (later) adjustment.
+func TestSpamRecordCache_ConcurrentGetWhileAdjustingSameRecord(t *testing.T) {
+	cache := internal.NewSpamRecordCache(100, zerolog.Nop(), metrics.NewNoopCollector(), model.SpamRecordFactory())
+	originID := unittest.IdentifierFixture()
+
+	const adjustments = 500
+	const penaltyDelta = -1.0
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	// writer: repeatedly adjusts the same record, and verifies the returned penalty is the one
+	// produced by its own adjustment (captured inside the adjust function, under the lock).
+	go func() {
+		defer wg.Done()
+		for i := 0; i < adjustments; i++ {
+			var want float64
+			penalty, err := cache.AdjustWithInit(originID, func(record *model.ProtocolSpamRecord) (*model.ProtocolSpamRecord, error) {
+				record.Penalty += penaltyDelta
+				want = record.Penalty
+				return record, nil
+			})
+			require.NoError(t, err)
+			require.Equal(t, want, penalty, "returned penalty must match the penalty produced by this adjustment")
+		}
+	}()
+
+	// reader: repeatedly reads the same record while it is being adjusted.
+	go func() {
+		defer wg.Done()
+		for i := 0; i < adjustments; i++ {
+			record, ok := cache.Get(originID)
+			if !ok {
+				continue // record not initialized yet
+			}
+			// the penalty only ever decreases in steps of penaltyDelta
+			require.LessOrEqual(t, record.Penalty, float64(0))
+			require.GreaterOrEqual(t, record.Penalty, float64(adjustments)*penaltyDelta)
+		}
+	}()
+
+	unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent adjustments and reads did not finish on time")
+
+	record, ok := cache.Get(originID)
+	require.True(t, ok)
+	require.Equal(t, float64(adjustments)*penaltyDelta, record.Penalty)
+}

network/p2p/inspector/validation/control_message_validation_inspector_test.go

@@ -299,6 +299,48 @@ func TestControlMessageValidationInspector_truncateRPC(t *testing.T) {
 	})
 }
 
+// TestControlMessageInspection_DoesNotMutateRpc is a regression test verifying that the
+// asynchronous inspection does not mutate the inspected RPC: the RPC object remains in use by
+// the libp2p pubsub layer while the inspector's worker pool processes it, so any in-place
+// modification (e.g. shuffling the publish message list for sampling) is a data race.
+func TestControlMessageInspection_DoesNotMutateRpc(t *testing.T) {
+	logCounter := atomic.NewInt64(0)
+	logger := hookedLogger(logCounter, zerolog.TraceLevel, worker.QueuedItemProcessedLog)
+	inspector, signalerCtx, cancel, consumer, rpcTracker, sporkID, idProvider, topicProviderOracle := inspectorFixture(t, func(params *validation.InspectorParams) {
+		params.Logger = logger
+	})
+	defer consumer.AssertNotCalled(t, "OnInvalidControlMessageNotification")
+
+	topic := fmt.Sprintf("%s/%s", channels.TestNetworkChannel, sporkID)
+	topicProviderOracle.UpdateTopics([]string{topic})
+	inspector.Start(signalerCtx)
+	unittest.RequireComponentsReadyBefore(t, 1*time.Second, inspector)
+
+	from := unittest.PeerIdFixture(t)
+	// enough messages that an accidental in-place shuffle is detected with near certainty
+	pubsubMsgs := unittest.GossipSubMessageFixtures(20, topic, unittest.WithFrom(from))
+	rpc := unittest.P2PRPCFixture(unittest.WithPubsubMessages(pubsubMsgs...))
+	rpcTracker.On("LastHighestIHaveRPCSize").Return(int64(100)).Maybe()
+	idProvider.On("ByPeerID", from).Return(unittest.IdentityFixture(), true)
+
+	// snapshot the publish message list before inspection
+	originalPublish := append([]*pubsub_pb.Message{}, rpc.GetPublish()...)
+
+	require.NoError(t, inspector.Inspect(from, rpc))
+	require.Eventually(t, func() bool {
+		return logCounter.Load() == 1
+	}, 30*time.Second, 100*time.Millisecond)
+
+	// the RPC's publish message list must be exactly as it was before the inspection
+	require.Equal(t, len(originalPublish), len(rpc.GetPublish()))
+	for i, msg := range rpc.GetPublish() {
+		require.Same(t, originalPublish[i], msg, "publish message %d was moved or replaced by the inspection", i)
+	}
+
+	cancel()
+	unittest.RequireCloseBefore(t, inspector.Done(), 5*time.Second, "inspector did not stop")
+}
+
 // TestControlMessageInspection_ValidRpc ensures inspector does not disseminate invalid control message notifications for a valid RPC.
 func TestControlMessageInspection_ValidRpc(t *testing.T) {
 	logCounter := atomic.NewInt64(0)

network/p2p/scoring/internal/appSpecificScoreCache_test.go

@@ -164,3 +164,45 @@ func TestAppSpecificScoreCache_Eviction(t *testing.T) {
 	_, _, found := cache.Get(peerIds[0])
 	require.False(t, found, "score should not be in cache")
 }
+
+// TestAppSpecificScoreCache_ConcurrentGetWhileUpdatingSamePeer is a regression test for a data
+// race on a single peer's record (run with -race): Get read the record's fields after the
+// backend lock was released, racing the in-place mutation performed by a concurrent update of
+// the same record. The cache now replaces records copy-on-write, keeping returned records
+// immutable.
+func TestAppSpecificScoreCache_ConcurrentGetWhileUpdatingSamePeer(t *testing.T) {
+	cache := internal.NewAppSpecificScoreCache(10, unittest.Logger(), metrics.NewNoopCollector())
+
+	peerID := unittest.PeerIdFixture(t)
+	const updates = 500
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	// writer: repeatedly updates the same peer's score.
+	go func() {
+		defer wg.Done()
+		for i := 0; i < updates; i++ {
+			require.NoError(t, cache.AdjustWithInit(peerID, float64(i), time.Now()))
+		}
+	}()
+
+	// reader: repeatedly reads the same peer's score while it is being updated.
+	go func() {
+		defer wg.Done()
+		for i := 0; i < updates; i++ {
+			score, _, found := cache.Get(peerID)
+			if !found {
+				continue // not initialized yet
+			}
+			require.GreaterOrEqual(t, score, float64(0))
+			require.Less(t, score, float64(updates))
+		}
+	}()
+
+	unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent updates and reads did not finish on time")
+
+	score, _, found := cache.Get(peerID)
+	require.True(t, found)
+	require.Equal(t, float64(updates-1), score)
+}

network/p2p/scoring/registry_internal_test.go

@@ -0,0 +1,39 @@
+package scoring
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+// TestAfterSilencePeriod verifies the state machine of the scoring registry's startup silence
+// period, in particular that the silence period does not end before the registry has started.
+// Regression: the start time used to be a plain time.Time (a data race with the startup worker),
+// and its zero value made time.Since(zero) exceed any configured duration, spuriously ending the
+// silence period if the score function was queried before startup.
+func TestAfterSilencePeriod(t *testing.T) {
+	reg := &GossipSubAppSpecificScoreRegistry{
+		silencePeriodDuration:  time.Hour,
+		silencePeriodStartTime: atomic.NewPointer[time.Time](nil),
+		silencePeriodElapsed:   atomic.NewBool(false),
+	}
+
+	// before startup (start time not set), the silence period has not even begun
+	require.False(t, reg.afterSilencePeriod())
+
+	// silence period started, but not yet over
+	now := time.Now()
+	reg.silencePeriodStartTime.Store(&now)
+	require.False(t, reg.afterSilencePeriod())
+
+	// silence period over
+	past := time.Now().Add(-2 * time.Hour)
+	reg.silencePeriodStartTime.Store(&past)
+	require.True(t, reg.afterSilencePeriod())
+
+	// once elapsed, the silence period stays elapsed
+	require.True(t, reg.silencePeriodElapsed.Load())
+	require.True(t, reg.afterSilencePeriod())
+}

network/slashing/consumer_test.go

@@ -0,0 +1,96 @@
+package slashing_test
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/model/flow"
+	mockmodule "github.com/onflow/flow-go/module/mock"
+	"github.com/onflow/flow-go/network"
+	"github.com/onflow/flow-go/network/channels"
+	"github.com/onflow/flow-go/network/message"
+	mocknetwork "github.com/onflow/flow-go/network/mock"
+	"github.com/onflow/flow-go/network/slashing"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// newConsumerFixture returns a slashing violations consumer whose metrics and misbehavior
+// report consumer tolerate any number of calls.
+func newConsumerFixture(t *testing.T) *slashing.Consumer {
+	metrics := mockmodule.NewNetworkSecurityMetrics(t)
+	metrics.On("OnUnauthorizedMessage", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return().Maybe()
+	metrics.On("OnViolationReportSkipped").Return().Maybe()
+
+	misbehaviorReportConsumer := mocknetwork.NewMisbehaviorReportConsumer(t)
+	misbehaviorReportConsumer.On("ReportMisbehaviorOnChannel", mock.Anything, mock.Anything).Return().Maybe()
+
+	return slashing.NewSlashingViolationsConsumer(unittest.Logger(), metrics, misbehaviorReportConsumer)
+}
+
+// violationFixture returns a violation without an identity (a violation from an unknown peer)
+// and without a message type (a violation raised before the message could be decoded).
+func violationFixture() *network.Violation {
+	return &network.Violation{
+		Identity: nil,
+		PeerID:   "peer-id",
+		OriginID: flow.ZeroID,
+		MsgType:  "", // simulates a violation raised before the message type is known
+		Channel:  channels.TestNetworkChannel,
+		Protocol: message.ProtocolTypeUnicast,
+		Err:      errors.New("unauthorized"),
+	}
+}
+
+// TestConsumer_DoesNotMutateViolation is a regression test verifying that the consumer never
+// mutates the violation passed to it: callers may share one violation object across goroutines
+// (or reuse it for multiple notifications), so an in-place default (e.g. setting MsgType to
+// "unknown") would be a data race and would leak into subsequent uses.
+func TestConsumer_DoesNotMutateViolation(t *testing.T) {
+	consumer := newConsumerFixture(t)
+
+	violation := violationFixture()
+	consumer.OnUnauthorizedSenderError(violation)
+
+	require.Empty(t, violation.MsgType, "consumer must not mutate the violation's MsgType")
+	require.Nil(t, violation.Identity, "consumer must not mutate the violation's Identity")
+}
+
+// TestConsumer_ConcurrentNotifications is a regression test verifying that a single violation
+// object can be reported concurrently through all consumer entry points without a data race
+// (run with -race). Before the fix, logOffense wrote a default MsgType into the shared
+// violation, racing the reads of concurrent notifications.
+func TestConsumer_ConcurrentNotifications(t *testing.T) {
+	consumer := newConsumerFixture(t)
+
+	violation := violationFixture()
+
+	notify := []func(*network.Violation){
+		consumer.OnUnauthorizedSenderError,
+		consumer.OnUnknownMsgTypeError,
+		consumer.OnInvalidMsgError,
+		consumer.OnSenderEjectedError,
+		consumer.OnUnauthorizedUnicastOnChannel,
+		consumer.OnUnauthorizedPublishOnChannel,
+	}
+
+	workers := 4
+	iterations := 50
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for range workers {
+		go func() {
+			defer wg.Done()
+			for i := 0; i < iterations; i++ {
+				notify[i%len(notify)](violation)
+			}
+		}()
+	}
+	unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent notifications did not finish on time")
+
+	require.Empty(t, violation.MsgType, "consumer must not mutate the violation's MsgType")
+}