[Testing] Fix flaky tests exposed by parallel test scheduling

engine/access/state_stream/backend/backend_executiondata_test.go

@@ -524,7 +524,7 @@ func (s *BackendExecutionDataSuite) subscribe(subscribeFunc func(ctx context.Con
 
 					assert.Equal(s.T(), b.Height, resp.Height)
 					assert.Equal(s.T(), execData.BlockExecutionData, resp.ExecutionData)
-				}, time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
+				}, 10*time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
 			}
 
 			// make sure there are no new messages waiting. the channel should be opened with nothing waiting

engine/access/state_stream/backend/handler_test.go

@@ -111,7 +111,7 @@ func (s *HandlerTestSuite) TestHeartbeatResponse() {
 				require.NoError(s.T(), err)
 				require.Equal(s.T(), b.ID(), blockID)
 				require.Equal(s.T(), b.Height, resp.BlockHeight)
-			}, time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
+			}, 10*time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
 		}
 	})
 
@@ -146,7 +146,7 @@ func (s *HandlerTestSuite) TestHeartbeatResponse() {
 				require.NoError(s.T(), err)
 				require.Equal(s.T(), b.ID(), blockID)
 				require.Equal(s.T(), b.Height, resp.BlockHeight)
-			}, time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
+			}, 10*time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
 		}
 	})
 
@@ -192,7 +192,7 @@ func (s *HandlerTestSuite) TestHeartbeatResponse() {
 				require.Equal(s.T(), b.Height, resp.BlockHeight)
 				require.Equal(s.T(), b.ID(), blockID)
 				require.Empty(s.T(), resp.Events)
-			}, time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
+			}, 10*time.Second, fmt.Sprintf("timed out waiting for exec data for block %d %v", b.Height, b.ID()))
 		}
 	})
 }

engine/verification/fetcher/chunkconsumer/consumer_test.go

@@ -56,8 +56,9 @@ func TestProduceConsume(t *testing.T) {
 			<-consumer.Done()
 
 			// expect the mock engine receive only the first 3 calls (since it is blocked on those, hence no
-			// new job is fetched to process).
-			require.Equal(t, locators[:3], called)
+			// new job is fetched to process). The 3 concurrent workers append in nondeterministic
+			// order, so assert the multiset rather than the exact sequence.
+			require.ElementsMatch(t, locators[:3], called)
 		})
 	})
 
@@ -91,8 +92,10 @@ func TestProduceConsume(t *testing.T) {
 
 			finishAll.Wait() // wait until all 10 jobs are processed and notified
 			<-consumer.Done()
-			// expect the mock engine receives all 10 calls
-			require.Equal(t, locators, called)
+			// expect the mock engine receives all 10 calls.
+			// the consumer processes jobs with 3 concurrent workers, so the receive order is
+			// not deterministic; assert the multiset rather than the exact sequence.
+			require.ElementsMatch(t, locators, called)
 		})
 	})
 

ledger/complete/wal/checkpoint_v6_test.go

@@ -96,11 +96,22 @@ func randPathPayload() (ledger.Path, ledger.Payload) {
 	return path, *payload
 }
 
-func randNPathPayloads(n int) ([]ledger.Path, []ledger.Payload) {
+// randNPathPayloadsUnique returns n random path/payload pairs whose payload keys are unique
+// with respect to each other and to the keys already present in `used` (which it updates).
+// Tests compare the payloads of the final trie by payload key, so a duplicate key with a
+// different value makes the comparison nondeterministic: with 1-byte random keys this
+// happens with probability ~1/256 per colliding pair (observed as a rare flake).
+func randNPathPayloadsUnique(n int, used map[string]struct{}) ([]ledger.Path, []ledger.Payload) {
 	paths := make([]ledger.Path, n)
 	payloads := make([]ledger.Payload, n)
 	for i := 0; i < n; i++ {
 		path, payload := randPathPayload()
+		key := hex.EncodeToString(payload.EncodedKey())
+		for _, dup := used[key]; dup; _, dup = used[key] {
+			path, payload = randPathPayload()
+			key = hex.EncodeToString(payload.EncodedKey())
+		}
+		used[key] = struct{}{}
 		paths[i] = path
 		payloads[i] = payload
 	}
@@ -110,23 +121,24 @@ func randNPathPayloads(n int) ([]ledger.Path, []ledger.Payload) {
 func createMultipleRandomTries(t *testing.T) []*trie.MTrie {
 	tries := make([]*trie.MTrie, 0)
 	activeTrie := trie.NewEmptyMTrie()
+	usedKeys := make(map[string]struct{})
 
 	var err error
 	// add tries with no shared paths
 	for i := 0; i < 100; i++ {
-		paths, payloads := randNPathPayloads(100)
+		paths, payloads := randNPathPayloadsUnique(100, usedKeys)
 		activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, paths, payloads, false)
 		require.NoError(t, err, "update registers")
 		tries = append(tries, activeTrie)
 	}
 
 	// add trie with some shared path
-	sharedPaths, payloads1 := randNPathPayloads(100)
+	sharedPaths, payloads1 := randNPathPayloadsUnique(100, usedKeys)
 	activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, sharedPaths, payloads1, false)
 	require.NoError(t, err, "update registers")
 	tries = append(tries, activeTrie)
 
-	_, payloads2 := randNPathPayloads(100)
+	_, payloads2 := randNPathPayloadsUnique(100, usedKeys)
 	activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, sharedPaths, payloads2, false)
 	require.NoError(t, err, "update registers")
 	tries = append(tries, activeTrie)
@@ -156,23 +168,24 @@ func isTrieDeepEnough(trie *trie.MTrie) bool {
 func createMultipleRandomTriesMini(t *testing.T) ([]*trie.MTrie, *trie.MTrie) {
 	tries := make([]*trie.MTrie, 0)
 	activeTrie := trie.NewEmptyMTrie()
+	usedKeys := make(map[string]struct{})
 
 	var err error
 	// add tries with no shared paths
 	for i := 0; i < 5; i++ {
-		paths, payloads := randNPathPayloads(20)
+		paths, payloads := randNPathPayloadsUnique(20, usedKeys)
 		activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, paths, payloads, false)
 		require.NoError(t, err, "update registers")
 		tries = append(tries, activeTrie)
 	}
 
 	// add trie with some shared path
-	sharedPaths, payloads1 := randNPathPayloads(10)
+	sharedPaths, payloads1 := randNPathPayloadsUnique(10, usedKeys)
 	activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, sharedPaths, payloads1, false)
 	require.NoError(t, err, "update registers")
 	tries = append(tries, activeTrie)
 
-	_, payloads2 := randNPathPayloads(10)
+	_, payloads2 := randNPathPayloadsUnique(10, usedKeys)
 	activeTrie, _, err = trie.NewTrieWithUpdatedRegisters(activeTrie, sharedPaths, payloads2, false)
 	require.NoError(t, err, "update registers")
 	tries = append(tries, activeTrie)

network/alsp/manager/manager_test.go

@@ -416,12 +416,16 @@ func TestHandleReportedMisbehavior_And_DisallowListing_RepeatOffender_Integratio
 
 		penalty1 := record.Penalty
 
-		// wait for one heartbeat to be processed.
-		time.Sleep(1 * time.Second)
-
-		record, ok = victimSpamRecordCache.Get(ids[spammerIndex].NodeID)
-		require.True(t, ok)
-		require.NotNil(t, record)
+		// 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
+		// (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 {
+			record, ok = victimSpamRecordCache.Get(ids[spammerIndex].NodeID)
+			require.True(t, ok)
+			require.NotNil(t, record)
+			return record.Penalty != penalty1
+		}, 10*time.Second, 10*time.Millisecond, "penalty did not decay after one heartbeat")
 
 		// check the penalty of the spammer node, which should be below the disallow-listing threshold.
 		// i.e. spammer penalty should be more negative than the disallow-listing threshold, hence disallow-listed.