[Testing] Speed up unit test suite: slow-first scheduling and long-pole parallelization

Makefile

@@ -10,6 +10,19 @@ VERSION := $(shell git describe --tags --abbrev=2 --match "v*" --match "secure-c
 # dynamically split up CI jobs into smaller jobs that can be run in parallel
 GO_TEST_PACKAGES := ./...
 
+# The slowest test packages. `go test` schedules package test binaries in argument order
+# (alphabetical for ./...), so these long, internally-sequential packages would otherwise
+# start last and form a straggler tail that dominates the wall time of a full-suite run.
+# Listing them first lets them run while the remaining packages fill the other slots.
+# Only prepended when running the full suite (not for CI's per-job package subsets).
+# Overlapping package patterns are deduplicated by the go tool, so nothing runs twice.
+# First line: the slowest packages (the wall-time critical path).
+# Second line: second-tier packages (~20-35s each) that would otherwise start only
+# after the first wave of short packages drains and form a tail at the end of the run.
+SLOW_TEST_PACKAGES := ./network/p2p/scoring/... ./ledger/complete/... ./network/p2p/connection/... ./network/p2p/inspector/validation/... \
+	./consensus/integration/... ./cmd/util/ledger/util/... ./engine/verification/test/... ./network/alsp/... ./network/test/... \
+	./network/p2p/test/... ./network/p2p/node/... ./engine/verification/assigner/blockconsumer/... ./module/dkg/... ./engine/access/state_stream/backend/...
+
 # Image tag: if image tag is not set, set it with version (or short commit if empty)
 ifeq (${IMAGE_TAG},)
 IMAGE_TAG := ${VERSION}
@@ -68,7 +81,7 @@ update-cadence-version:
 .PHONY: unittest-main
 unittest-main:
 	# test all packages
-	CGO_CFLAGS=$(CRYPTO_FLAG) go test $(if $(VERBOSE),-v,) -coverprofile=$(COVER_PROFILE) -covermode=atomic $(if $(RACE_DETECTOR),-race,) $(if $(JSON_OUTPUT),-json,) $(if $(NUM_RUNS),-count $(NUM_RUNS),) $(GO_TEST_PACKAGES)
+	CGO_CFLAGS=$(CRYPTO_FLAG) go test $(if $(VERBOSE),-v,) -coverprofile=$(COVER_PROFILE) -covermode=atomic $(if $(RACE_DETECTOR),-race,) $(if $(JSON_OUTPUT),-json,) $(if $(NUM_RUNS),-count $(NUM_RUNS),) $(if $(filter ./...,$(GO_TEST_PACKAGES)),$(SLOW_TEST_PACKAGES)) $(GO_TEST_PACKAGES)
 
 .PHONY: install-mock-generators
 install-mock-generators:

engine/verification/test/happypath_test.go

@@ -121,6 +121,10 @@ func TestVerificationHappyPath(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.msg, func(t *testing.T) {
+			// each subtest builds its own full node fixture (own hub, networks, mocks,
+			// and FVM instance), so they are independent and can run in parallel.
+			t.Parallel()
+
 			collector := &metrics.NoopCollector{}
 
 			vertestutils.NewVerificationHappyPathTest(t,

ledger/complete/compactor_test.go

@@ -58,6 +58,8 @@ func (co *CompactorObserver) OnComplete() {
 // TestCompactorCreation tests creation of WAL segments and checkpoints, and
 // checks if the rebuilt ledger state matches previous ledger state.
 func TestCompactorCreation(t *testing.T) {
+	t.Parallel()
+
 	const (
 		numInsPerStep      = 2
 		pathByteSize       = 32
@@ -287,6 +289,8 @@ func TestCompactorCreation(t *testing.T) {
 // TestCompactorSkipCheckpointing tests that only one
 // checkpointing is running at a time.
 func TestCompactorSkipCheckpointing(t *testing.T) {
+	t.Parallel()
+
 	const (
 		numInsPerStep      = 2
 		pathByteSize       = 32
@@ -410,6 +414,7 @@ func TestCompactorSkipCheckpointing(t *testing.T) {
 // (from segment 0, ignoring prior checkpoints) to the checkpoint number.
 // This verifies that checkpointed tries are snapshopt of segments and at segment boundary.
 func TestCompactorAccuracy(t *testing.T) {
+	t.Parallel()
 
 	const (
 		numInsPerStep      = 2
@@ -526,6 +531,7 @@ func TestCompactorAccuracy(t *testing.T) {
 // TestCompactorTriggeredByAdminTool tests that the compactor will listen to the signal from admin tool
 // to trigger checkpoint when current segment file is finished.
 func TestCompactorTriggeredByAdminTool(t *testing.T) {
+	t.Parallel()
 
 	const (
 		numInsPerStep      = 2 // the number of payloads in each trie update
@@ -628,6 +634,8 @@ func TestCompactorTriggeredByAdminTool(t *testing.T) {
 // When initial state is from ledger's forest (LRU cache), its
 // sequence is altered by reads when replaying segment records.
 func TestCompactorConcurrency(t *testing.T) {
+	t.Parallel()
+
 	const (
 		numInsPerStep      = 2
 		pathByteSize       = 32

ledger/complete/ledger_test.go

@@ -585,6 +585,8 @@ func Test_WAL(t *testing.T) {
 }
 
 func TestLedgerFunctionality(t *testing.T) {
+	t.Parallel()
+
 	const (
 		checkpointDistance = math.MaxInt // A large number to prevent checkpoint creation.
 		checkpointsToKeep  = 1

network/alsp/manager/manager_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/onflow/flow-go/network/internal/testutils"
 	mocknetwork "github.com/onflow/flow-go/network/mock"
 	"github.com/onflow/flow-go/network/p2p"
+	p2pbuilderconfig "github.com/onflow/flow-go/network/p2p/builder/config"
 	p2ptest "github.com/onflow/flow-go/network/p2p/test"
 	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/underlay"
@@ -299,6 +300,10 @@ func TestHandleReportedMisbehavior_And_DisallowListing_Integration(t *testing.T)
 // handling of repeated reported misbehavior and disallow listing.
 func TestHandleReportedMisbehavior_And_DisallowListing_RepeatOffender_Integration(t *testing.T) {
 	cfg := managerCfgFixture(t)
+	// compress the heartbeat interval: all assertions in this test are per-heartbeat
+	// (decay deltas, disallow-listing rounds), so a faster heartbeat preserves semantics
+	// while cutting the wall time roughly proportionally.
+	cfg.HeartBeatInterval = 100 * time.Millisecond
 	sporkId := unittest.IdentifierFixture()
 	fastDecay := false
 	fastDecayFunc := func(record *model.ProtocolSpamRecord) float64 {
@@ -327,7 +332,11 @@ func TestHandleReportedMisbehavior_And_DisallowListing_RepeatOffender_Integratio
 	}
 
 	ids, nodes := testutils.LibP2PNodeForNetworkFixture(t, sporkId, 3,
-		p2ptest.WithPeerManagerEnabled(p2ptest.PeerManagerConfigFixture(p2ptest.WithZeroJitterAndZeroBackoff(t)), nil))
+		p2ptest.WithPeerManagerEnabled(p2ptest.PeerManagerConfigFixture(p2ptest.WithZeroJitterAndZeroBackoff(t), func(cfg *p2pbuilderconfig.PeerManagerConfig) {
+			// compress the peer manager tick as well: pruning and reconnection waits are
+			// tick-quantized, and the test only cares about the sequence of events.
+			cfg.UpdateInterval = 100 * time.Millisecond
+		}), nil))
 	idProvider := unittest.NewUpdatableIDProvider(ids)
 	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], idProvider, sporkId, nodes[0], underlay.WithAlspConfig(cfg))
 
@@ -417,7 +426,7 @@ func TestHandleReportedMisbehavior_And_DisallowListing_RepeatOffender_Integratio
 		penalty1 := record.Penalty
 
 		// wait for one heartbeat to be processed: poll for the first penalty change instead of
-		// sleeping exactly one heartbeat interval. A fixed 1s sleep races the 1s heartbeat ticker
+		// sleeping exactly one heartbeat interval. A fixed sleep races the heartbeat ticker
 		// (which can be delayed under load) and may span 0 or 2 decays; polling at 10ms granularity
 		// reliably catches the state after exactly one decay.
 		require.Eventually(t, func() bool {
@@ -1422,6 +1431,8 @@ func TestHandleMisbehaviorReport_DuplicateReportsForSinglePeer_Concurrently(t *t
 // is decayed after a single heartbeat. The test guarantees waiting for at least one heartbeat by waiting for the first decay to happen.
 func TestDecayMisbehaviorPenalty_SingleHeartbeat(t *testing.T) {
 	cfg := managerCfgFixture(t)
+	// decay assertions are per-heartbeat; a faster heartbeat preserves semantics.
+	cfg.HeartBeatInterval = 100 * time.Millisecond
 	consumer := mocknetwork.NewDisallowListNotificationConsumer(t)
 
 	var cache alsp.SpamRecordCache
@@ -1525,6 +1536,8 @@ func TestDecayMisbehaviorPenalty_SingleHeartbeat(t *testing.T) {
 // The test ensures that the misbehavior penalty is decayed with a linear progression within multiple heartbeats.
 func TestDecayMisbehaviorPenalty_MultipleHeartbeats(t *testing.T) {
 	cfg := managerCfgFixture(t)
+	// decay assertions are per-heartbeat; a faster heartbeat preserves semantics.
+	cfg.HeartBeatInterval = 100 * time.Millisecond
 	consumer := mocknetwork.NewDisallowListNotificationConsumer(t)
 
 	var cache alsp.SpamRecordCache

network/p2p/connection/connection_gater_test.go

@@ -3,6 +3,7 @@ package connection_test
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -421,28 +422,46 @@ func ensureCommunicationSilenceAmongGroups(
 	// ensures no connection, unicast, or pubsub going to the disallow-listed nodes
 	p2ptest.EnsureNotConnectedBetweenGroups(t, ctx, groupANodes, groupBNodes)
 
+	// check both pubsub directions (A->B and B->A) concurrently on two distinct topics.
+	// Each direction is self-paced (1s subscription propagation + 5s never-receive window);
+	// checking them sequentially doubles that cost, while distinct topics keep the checks
+	// fully isolated from each other. Identical per-direction windows and assertions.
 	blockTopic := channels.TopicFromChannel(channels.PushBlocks, sporkId)
-	p2ptest.EnsureNoPubsubExchangeBetweenGroups(
-		t,
-		ctx,
-		groupANodes,
-		groupAIdentifiers,
-		groupBNodes,
-		groupBIdentifiers,
-		blockTopic,
-		1,
-		func() interface{} {
-			return (*messages.Proposal)(unittest.ProposalFixture())
-		})
+	receiptTopic := channels.TopicFromChannel(channels.PushReceipts, sporkId)
+	messageFactory := func() interface{} {
+		return (*messages.Proposal)(unittest.ProposalFixture())
+	}
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		p2ptest.EnsureNoPubsubMessageExchange(t, ctx, groupANodes, groupBNodes, groupBIdentifiers, blockTopic, 1, messageFactory)
+	}()
+	go func() {
+		defer wg.Done()
+		p2ptest.EnsureNoPubsubMessageExchange(t, ctx, groupBNodes, groupANodes, groupAIdentifiers, receiptTopic, 1, messageFactory)
+	}()
+	unittest.RequireReturnsBefore(t, wg.Wait, 12*time.Second, "timed out waiting for pubsub silence checks")
 	p2pfixtures.EnsureNoStreamCreationBetweenGroups(t, ctx, groupANodes, groupBNodes)
 }
 
 // ensureCommunicationOverAllProtocols ensures that all nodes are connected to each other, and they can exchange messages over the pubsub and unicast.
 func ensureCommunicationOverAllProtocols(t *testing.T, ctx context.Context, sporkId flow.Identifier, nodes []p2p.LibP2PNode, inbounds []chan string) {
 	blockTopic := channels.TopicFromChannel(channels.PushBlocks, sporkId)
 	p2ptest.TryConnectionAndEnsureConnected(t, ctx, nodes)
-	p2ptest.EnsurePubsubMessageExchange(t, ctx, nodes, blockTopic, 1, func() interface{} {
-		return (*messages.Proposal)(unittest.ProposalFixture())
-	})
-	p2pfixtures.EnsureMessageExchangeOverUnicast(t, ctx, nodes, inbounds, p2pfixtures.LongStringMessageFactoryFixture(t))
+	// the pubsub and unicast exchanges are independent of each other (different protocols);
+	// run them concurrently. Both return early on success.
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		p2ptest.EnsurePubsubMessageExchange(t, ctx, nodes, blockTopic, 1, func() interface{} {
+			return (*messages.Proposal)(unittest.ProposalFixture())
+		})
+	}()
+	go func() {
+		defer wg.Done()
+		p2pfixtures.EnsureMessageExchangeOverUnicast(t, ctx, nodes, inbounds, p2pfixtures.LongStringMessageFactoryFixture(t))
+	}()
+	unittest.RequireReturnsBefore(t, wg.Wait, 30*time.Second, "timed out waiting for protocol exchanges")
 }

network/p2p/inspector/validation/control_message_validation_inspector_test.go

@@ -1596,6 +1596,8 @@ func TestControlMessageValidationInspector_ActiveClustersChanged(t *testing.T) {
 // TestControlMessageValidationInspector_TruncationConfigToggle ensures that rpc's are not truncated when truncation is disabled through configs.
 func TestControlMessageValidationInspector_TruncationConfigToggle(t *testing.T) {
 	t.Run("should not perform truncation when disabled is set to true", func(t *testing.T) {
+		t.Parallel()
+
 		numOfMsgs := 5000
 		logCounter := atomic.NewInt64(0)
 		logger := hookedLogger(logCounter, zerolog.TraceLevel, validation.RPCTruncationDisabledWarning, worker.QueuedItemProcessedLog)
@@ -1640,6 +1642,8 @@ func TestControlMessageValidationInspector_TruncationConfigToggle(t *testing.T)
 	})
 
 	t.Run("should not perform truncation when disabled for each individual control message type directly", func(t *testing.T) {
+		t.Parallel()
+
 		numOfMsgs := 5000
 		expectedLogStrs := []string{
 			validation.GraftTruncationDisabledWarning,
@@ -1701,6 +1705,8 @@ func TestControlMessageValidationInspector_TruncationConfigToggle(t *testing.T)
 // TestControlMessageValidationInspector_InspectionConfigToggle ensures that rpc's are not inspected when inspection is disabled through configs.
 func TestControlMessageValidationInspector_InspectionConfigToggle(t *testing.T) {
 	t.Run("should not perform inspection when disabled is set to true", func(t *testing.T) {
+		t.Parallel()
+
 		numOfMsgs := 5000
 		logCounter := atomic.NewInt64(0)
 		logger := hookedLogger(logCounter, zerolog.TraceLevel, validation.RPCInspectionDisabledWarning)
@@ -1735,6 +1741,8 @@ func TestControlMessageValidationInspector_InspectionConfigToggle(t *testing.T)
 	})
 
 	t.Run("should not check identity when reject-unstaked-peers is false", func(t *testing.T) {
+		t.Parallel()
+
 		inspector, signalerCtx, cancel, consumer, rpcTracker, _, idProvider, _ := inspectorFixture(t, func(params *validation.InspectorParams) {
 			// disable inspector for all control message types
 			params.Config.InspectionProcess.Inspect.RejectUnstakedPeers = false
@@ -1758,6 +1766,8 @@ func TestControlMessageValidationInspector_InspectionConfigToggle(t *testing.T)
 	})
 
 	t.Run("should check identity when reject-unstaked-peers is true", func(t *testing.T) {
+		t.Parallel()
+
 		inspector, signalerCtx, cancel, consumer, rpcTracker, _, idProvider, _ := inspectorFixture(t, func(params *validation.InspectorParams) {
 			// disable inspector for all control message types
 			params.Config.InspectionProcess.Inspect.RejectUnstakedPeers = true
@@ -1785,6 +1795,8 @@ func TestControlMessageValidationInspector_InspectionConfigToggle(t *testing.T)
 	})
 
 	t.Run("should not perform inspection when disabled for each individual control message type directly", func(t *testing.T) {
+		t.Parallel()
+
 		numOfMsgs := 5000
 		expectedLogStrs := []string{
 			validation.GraftInspectionDisabledWarning,

network/p2p/scoring/app_score_test.go

@@ -29,6 +29,8 @@ import (
 // pushing access nodes to the edges of the network (i.e., the access nodes are not in the mesh of any honest nodes)
 // will not cause the network to partition, i.e., all honest nodes can still communicate with each other through GossipSub.
 func TestFullGossipSubConnectivity(t *testing.T) {
+	t.Parallel()
+
 	ctx, cancel := context.WithCancel(context.Background())
 	signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx)
 	sporkId := unittest.IdentifierFixture()
@@ -121,6 +123,8 @@ func TestFullGossipSubConnectivity(t *testing.T) {
 // when the network topology is a complete graph (i.e., full topology) and a malicious majority of access nodes are present.
 // The honest nodes (i.e., non-Access nodes) are enabled with peer scoring, then the honest nodes are enabled with peer scoring.
 func TestFullGossipSubConnectivityAmongHonestNodesWithMaliciousMajority(t *testing.T) {
+	t.Parallel()
+
 	// Note: if this test is ever flaky, this means a bug in our scoring system. Please escalate to the team instead of skipping.
 	ctx, cancel := context.WithCancel(context.Background())
 	signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx)